LangChain Intercepción de llamadas a herramientas -- @wrap_tool_call
@wrap_tool_call te permite implementar capacidades de control similares a @wrap_model_call a nivel de ejecución de herramientas: reintentos, caché, reescritura de parámetros y postprocesamiento de resultados.
Estructura básica
@wrap_tool_call tiene una estructura similar a @wrap_model_call: recibe dos parámetros, request y handler:
Ejemplo
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def my_tool_wrapper(request, handler):
# request.tool_call: contiene el nombre y los parámetros de la herramienta
# request.tool: el objeto de la herramienta en sí
# request.state: estado actual del Agent
# request.runtime: contexto de ejecución
# solo se ejecuta realmente la herramienta al llamar a handler(request)
result = handler(request)
# result es ToolMessage o Command
return result
@wrap_tool_call
def my_tool_wrapper(request, handler):
# request.tool_call: contiene el nombre y los parámetros de la herramienta
# request.tool: el objeto de la herramienta en sí
# request.state: estado actual del Agent
# request.runtime: contexto de ejecución
# solo se ejecuta realmente la herramienta al llamar a handler(request)
result = handler(request)
# result es ToolMessage o Command
return result
Escenario 1: Reintento de llamadas a herramientas
La ejecución de herramientas puede fallar debido a la inestabilidad de servicios externos; los reintentos automáticos pueden mejorar la confiabilidad:
Ejemplo
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@wrap_tool_call
def retry_tool_on_error(request, handler):
"""Reintento automático cuando falla la llamada a la herramienta"""
max_retries = 3
last_result = None
for attempt in range(max_retries):
try:
result = handler(request)
# verificar si es un resultado de error
if hasattr(result, 'status') and result.status == "error":
if attempt < max_retries - 1:
print(f" [Reintento] La herramienta devolvió un error, reintento número {attempt + 1}...")
continue
if attempt > 0:
print(f" [Reintento exitoso] Intento número {attempt + 1}")
return result
except Exception as e:
if attempt < max_retries - 1:
import time
time.sleep((attempt + 1) * 2)
print(f" [Reintento] Excepción {e}, reintento número {attempt + 1}...")
else:
raise
return last_result
# simular una herramienta que puede fallar
call_count = 0
@tool
def fetch_course_data(course_id: str) -> str:
"""Obtener datos de cursos desde el Tutorial de EJEMPLO.
Args:
course_id: ID del curso
"""
global call_count
call_count += 1
# simular fallo en los dos primeros intentos y éxito en el tercero
if call_count < 3:
raise Exception(f"Error de red: no se pudo conectar al servicio de cursos (intento número {call_count})")
return f"Curso {course_id}: Tutorial básico de Python3, 30 capítulos, gratuito"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[fetch_course_data],
middleware=[retry_tool_on_error],
system_prompt="Eres el asistente de cursos de EJEMPLO.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Ayúdame a consultar la información del curso python-001")]
})
print(f"\nRespuesta final: {result['messages'][-1].content}")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@wrap_tool_call
def retry_tool_on_error(request, handler):
"""Reintento automático cuando falla la llamada a la herramienta"""
max_retries = 3
last_result = None
for attempt in range(max_retries):
try:
result = handler(request)
# verificar si es un resultado de error
if hasattr(result, 'status') and result.status == "error":
if attempt < max_retries - 1:
print(f" [Reintento] La herramienta devolvió un error, reintento número {attempt + 1}...")
continue
if attempt > 0:
print(f" [Reintento exitoso] Intento número {attempt + 1}")
return result
except Exception as e:
if attempt < max_retries - 1:
import time
time.sleep((attempt + 1) * 2)
print(f" [Reintento] Excepción {e}, reintento número {attempt + 1}...")
else:
raise
return last_result
# simular una herramienta que puede fallar
call_count = 0
@tool
def fetch_course_data(course_id: str) -> str:
"""Obtener datos de cursos desde el Tutorial de EJEMPLO.
Args:
course_id: ID del curso
"""
global call_count
call_count += 1
# simular fallo en los dos primeros intentos y éxito en el tercero
if call_count < 3:
raise Exception(f"Error de red: no se pudo conectar al servicio de cursos (intento número {call_count})")
return f"Curso {course_id}: Tutorial básico de Python3, 30 capítulos, gratuito"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[fetch_course_data],
middleware=[retry_tool_on_error],
system_prompt="Eres el asistente de cursos de EJEMPLO.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Ayúdame a consultar la información del curso python-001")]
})
print(f"\nRespuesta final: {result['messages'][-1].content}")
Resultado de la ejecución:
[重试] 异常 网络错误：无法连接到课程服务（第 1 次尝试），1 次重试... [重试] 异常 网络错误：无法连接到课程服务（第 2 次尝试），2 次重试... [重试成功] 第 3 次尝试 Respuesta final: El curso python-001 es el tutorial básico de Python3, 30 capítulos, gratuito.
Escenario 2: Modificación de los parámetros de la herramienta
Modificar parámetros dinámicamente antes de la ejecución de la herramienta permite transformar parámetros sin modificar el código de la herramienta:
Ejemplo
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def normalize_city_name(request, handler):
"""Normalizar automáticamente los nombres de ciudades (convertir caracteres de ancho completo a ancho medio, eliminar espacios redundantes, etc.)"""
tool_call = request.tool_call
# procesar solo llamadas a herramientas que contengan el parámetro city
if "city" in tool_call.get("args", {}):
city = tool_call["args"]["city"]
# normalizar el nombre de la ciudad: quitar espacios, unificar mayúsculas/minúsculas
normalized = city.strip().replace(" ", "") # eliminar espacios de ancho completo
# reemplazar el parámetro
new_args = {**tool_call["args"], "city": normalized}
new_tool_call = {**tool_call, "args": new_args}
request = request.override(tool_call=new_tool_call)
return handler(request)
@wrap_tool_call
def normalize_city_name(request, handler):
"""Normalizar automáticamente los nombres de ciudades (convertir caracteres de ancho completo a ancho medio, eliminar espacios redundantes, etc.)"""
tool_call = request.tool_call
# procesar solo llamadas a herramientas que contengan el parámetro city
if "city" in tool_call.get("args", {}):
city = tool_call["args"]["city"]
# normalizar el nombre de la ciudad: quitar espacios, unificar mayúsculas/minúsculas
normalized = city.strip().replace(" ", "") # eliminar espacios de ancho completo
# reemplazar el parámetro
new_args = {**tool_call["args"], "city": normalized}
new_tool_call = {**tool_call, "args": new_args}
request = request.override(tool_call=new_tool_call)
return handler(request)
Escenario 3: Caché de resultados de herramientas
Para llamadas a herramientas repetidas (misma herramienta + mismos parámetros), se puede almacenar en caché el resultado:
Ejemplo
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
tool_cache = {}
@wrap_tool_call
def cache_tool_results(request, handler):
"""Almacenar en caché el resultado de la ejecución de la herramienta"""
# generar la clave de caché: nombre de la herramienta + parámetros
tool_name = request.tool_call.get("name", "unknown")
tool_args = str(request.tool_call.get("args", {}))
cache_key = f"{tool_name}:{tool_args}"
# verificar la caché
if cache_key in tool_cache:
print(f"[Acierto de caché de herramienta] {tool_name}")
cached_content = tool_cache[cache_key]
return ToolMessage(
content=cached_content,
tool_call_id=request.tool_call.get("id", ""),
name=tool_name,
)
# ejecutar la herramienta
result = handler(request)
# guardar en caché
if hasattr(result, 'content'):
tool_cache[cache_key] = result.content
print(f"[Escritura en caché de herramienta] {tool_name}, actualmente {len(tool_cache)} entradas")
return result
from langchain.messages import ToolMessage
tool_cache = {}
@wrap_tool_call
def cache_tool_results(request, handler):
"""Almacenar en caché el resultado de la ejecución de la herramienta"""
# generar la clave de caché: nombre de la herramienta + parámetros
tool_name = request.tool_call.get("name", "unknown")
tool_args = str(request.tool_call.get("args", {}))
cache_key = f"{tool_name}:{tool_args}"
# verificar la caché
if cache_key in tool_cache:
print(f"[Acierto de caché de herramienta] {tool_name}")
cached_content = tool_cache[cache_key]
return ToolMessage(
content=cached_content,
tool_call_id=request.tool_call.get("id", ""),
name=tool_name,
)
# ejecutar la herramienta
result = handler(request)
# guardar en caché
if hasattr(result, 'content'):
tool_cache[cache_key] = result.content
print(f"[Escritura en caché de herramienta] {tool_name}, actualmente {len(tool_cache)} entradas")
return result
Escenario 4: Registro y monitoreo de llamadas a herramientas
Registrar información detallada de todas las llamadas a herramientas:
Ejemplo
import time
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def monitor_tool_performance(request, handler):
"""Monitorear las métricas de rendimiento de las llamadas a herramientas"""
tool_name = request.tool_call.get("name", "unknown")
tool_args = request.tool_call.get("args", {})
# registrar la hora de inicio
start_time = time.time()
try:
result = handler(request)
elapsed = time.time() - start_time
# registrar llamadas exitosas
print(f"[Monitoreo] {tool_name}({tool_args}) exitoso, tiempo transcurrido {elapsed:.2f}s")
return result
except Exception as e:
elapsed = time.time() - start_time
# registrar llamadas fallidas
print(f"[Monitoreo] {tool_name}({tool_args}) falló, tiempo transcurrido {elapsed:.2f}s, error: {e}")
raise
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def monitor_tool_performance(request, handler):
"""Monitorear las métricas de rendimiento de las llamadas a herramientas"""
tool_name = request.tool_call.get("name", "unknown")
tool_args = request.tool_call.get("args", {})
# registrar la hora de inicio
start_time = time.time()
try:
result = handler(request)
elapsed = time.time() - start_time
# registrar llamadas exitosas
print(f"[Monitoreo] {tool_name}({tool_args}) exitoso, tiempo transcurrido {elapsed:.2f}s")
return result
except Exception as e:
elapsed = time.time() - start_time
# registrar llamadas fallidas
print(f"[Monitoreo] {tool_name}({tool_args}) falló, tiempo transcurrido {elapsed:.2f}s, error: {e}")
raise
Escenario 5: Decidir el flujo posterior según los resultados
Puedes decidir si continuar el bucle del Agent según el resultado de la ejecución de la herramienta:
Ejemplo
from langchain.agents.middleware import wrap_tool_call
from langgraph.types import Command
@wrap_tool_call
def check_empty_result(request, handler):
"""Si la herramienta devuelve un resultado vacío, finalizar el Agent directamente, sin desperdiciar llamadas al modelo"""
result = handler(request)
# verificar si se devolvió un resultado vacío
if hasattr(result, 'content') and (
"No encontrado" in str(result.content)
or "Sin resultados" in str(result.content)
or "No" in str(result.content)
):
# devolver directamente un Command para actualizar el estado
# agregar un mensaje de AI explicando la situación
from langchain.messages import AIMessage
return Command(update={
"messages": [
AIMessage(content="Lo siento, no se encontró información relevante. Prueba con otra palabra clave.")
]
})
return result
from langgraph.types import Command
@wrap_tool_call
def check_empty_result(request, handler):
"""Si la herramienta devuelve un resultado vacío, finalizar el Agent directamente, sin desperdiciar llamadas al modelo"""
result = handler(request)
# verificar si se devolvió un resultado vacío
if hasattr(result, 'content') and (
"No encontrado" in str(result.content)
or "Sin resultados" in str(result.content)
or "No" in str(result.content)
):
# devolver directamente un Command para actualizar el estado
# agregar un mensaje de AI explicando la situación
from langchain.messages import AIMessage
return Command(update={
"messages": [
AIMessage(content="Lo siento, no se encontró información relevante. Prueba con otra palabra clave.")
]
})
return result
Cuando wrap_tool_call devuelve un Command, puedes modificar el estado del Agent mediante el parámetro update. Usar Command permite agregar directamente un mensaje de AI a la lista de mensajes, y luego el bucle del Agent termina de forma natural.
@wrap_model_call vs @wrap_tool_call
| Dimensión | @wrap_model_call | @wrap_tool_call |
|---|---|---|
| Objetivo de intercepción | Llamada al modelo | Ejecución de herramienta |
| Contenido de request | model、messages、tools、system_prompt | tool_call、tool、state、runtime |
| Tipo de retorno | ModelResponse o AIMessage | ToolMessage o Command |
| Escenarios aplicables | Reintentos de modelo, degradación, caché, modificación de prompt | Reintentos de herramienta, caché, reescritura de parámetros, procesamiento de resultados |