Intercepción de llamadas al modelo de LangChain -- @wrap_model_call
@wrap_model_call es el gancho más potente del middleware.
@wrap_model_call no se limita a observar como before/after, sino que puedecontrolar completamente el proceso de ejecución del modelo— reintentos, degradación, caché, e incluso omitir el modelo y usar directamente una respuesta predefinida.
Entendiendo el callback del handler
El núcleo de @wrap_model_call es unafunción callback del handler.Al llamar a handler(request) se ejecuta realmente el modelo; si no se llama, se omite el modelo.
Ejemplo
# Estructura básica de wrap_model_call
# request: contiene toda la información como modelo, mensajes, herramientas, etc.
# handler: un objeto invocable; solo al ejecutarlo se llama realmente al modelo
@wrap_model_call
def my_middleware(request, handler):
# Antes de la llamada al modelo se puede hacer cualquier cosa
print("El modelo está a punto de ser llamado...")
# Solo al llamar a handler(request) se ejecuta realmente el modelo
response = handler(request)
# Después de la llamada al modelo se puede hacer cualquier cosa
print("Llamada al modelo completada")
return response
# request: contiene toda la información como modelo, mensajes, herramientas, etc.
# handler: un objeto invocable; solo al ejecutarlo se llama realmente al modelo
@wrap_model_call
def my_middleware(request, handler):
# Antes de la llamada al modelo se puede hacer cualquier cosa
print("El modelo está a punto de ser llamado...")
# Solo al llamar a handler(request) se ejecuta realmente el modelo
response = handler(request)
# Después de la llamada al modelo se puede hacer cualquier cosa
print("Llamada al modelo completada")
return response
Escenario 1: Mecanismo de reintentos
Este es el escenario más común: las llamadas al modelo pueden fallar por problemas de red; los reintentos automáticos pueden mejorar la fiabilidad:
Ejemplo
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@wrap_model_call
def retry_on_error(request, handler):
"""Reintenta automáticamente cuando falla la llamada al modelo, hasta 3 veces"""
max_retries = 3
last_error = None
for attempt in range(max_retries):
try:
result = handler(request)
if attempt > 0:
print(f" [Reintento exitoso] Intento {attempt + 1}")
return result
except Exception as e:
last_error = e
if attempt < max_retries - 1:
wait_time = (attempt + 1) * 2 # Espera incremental: 2s, 4s
print(f" [Reintento] El intento {attempt + 1} falló, reintentando en {wait_time} segundos...")
import time
time.sleep(wait_time)
# Todos los reintentos han fallado
raise last_error
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[retry_on_error],
system_prompt="Eres el asistente de EJEMPLO.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Presenta EJEMPLO")]
})
print(f"\nRespuesta: {result['messages'][-1].content[:100]}...")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@wrap_model_call
def retry_on_error(request, handler):
"""Reintenta automáticamente cuando falla la llamada al modelo, hasta 3 veces"""
max_retries = 3
last_error = None
for attempt in range(max_retries):
try:
result = handler(request)
if attempt > 0:
print(f" [Reintento exitoso] Intento {attempt + 1}")
return result
except Exception as e:
last_error = e
if attempt < max_retries - 1:
wait_time = (attempt + 1) * 2 # Espera incremental: 2s, 4s
print(f" [Reintento] El intento {attempt + 1} falló, reintentando en {wait_time} segundos...")
import time
time.sleep(wait_time)
# Todos los reintentos han fallado
raise last_error
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[retry_on_error],
system_prompt="Eres el asistente de EJEMPLO.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Presenta EJEMPLO")]
})
print(f"\nRespuesta: {result['messages'][-1].content[:100]}...")
Escenario 2: Degradación del modelo / Conmutación por error
Cuando el modelo principal no está disponible, se cambia automáticamente al modelo de respaldo:
Ejemplo
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage
# Modelo principal y modelo de respaldo
primary_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
fallback_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0,
max_tokens=100) # El modelo degradado limita los tokens
@wrap_model_call
def fallback_on_error(request, handler):
"""Cambia automáticamente al modelo de respaldo cuando falla el modelo principal"""
try:
# Intentar con el modelo principal
return handler(request)
except Exception as e:
print(f"[Degradación] El modelo principal falló: {e}, cambiando al modelo de respaldo...")
# Sobrescribir model en request para cambiar al modelo de respaldo
request = request.override(model=fallback_model)
try:
return handler(request)
except Exception as e2:
print(f"[Degradación] El modelo de respaldo también falló: {e2}")
# Ambos modelos fallaron, devolver un mensaje amigable
return AIMessage(
content="Lo sentimos, el servicio no está disponible temporalmente, por favor inténtelo de nuevo más tarde."
)
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage
# Modelo principal y modelo de respaldo
primary_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
fallback_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0,
max_tokens=100) # El modelo degradado limita los tokens
@wrap_model_call
def fallback_on_error(request, handler):
"""Cambia automáticamente al modelo de respaldo cuando falla el modelo principal"""
try:
# Intentar con el modelo principal
return handler(request)
except Exception as e:
print(f"[Degradación] El modelo principal falló: {e}, cambiando al modelo de respaldo...")
# Sobrescribir model en request para cambiar al modelo de respaldo
request = request.override(model=fallback_model)
try:
return handler(request)
except Exception as e2:
print(f"[Degradación] El modelo de respaldo también falló: {e2}")
# Ambos modelos fallaron, devolver un mensaje amigable
return AIMessage(
content="Lo sentimos, el servicio no está disponible temporalmente, por favor inténtelo de nuevo más tarde."
)
request.override() es un método inmutable: devuelve una copia nueva de request, sin modificar el objeto original. Esto garantiza que cada llamada sea independiente y segura.
Escenario 3: Almacenamiento en caché de respuestas del modelo
Para consultas repetidas, se pueden almacenar en caché las respuestas del modelo para reducir el costo de las llamadas a la API:
Ejemplo
from langchain.agents.middleware import wrap_model_call
from langchain.messages import AIMessage
# Caché simple en memoria
cache = {}
@wrap_model_call
def cache_responses(request, handler):
"""Almacena en caché las respuestas del modelo; no repite llamadas para la misma pregunta"""
# Usar el contenido del último mensaje del usuario como clave de caché
messages = request.messages
if not messages:
return handler(request)
# Generar la clave de caché
last_content = str(messages[-1].content) if hasattr(messages[-1], 'content') else ""
cache_key = last_content[:200] # Truncar el contenido demasiado largo
# Verificar la caché
if cache_key in cache:
print(f"[Acierto de caché] Devolver directamente el resultado en caché")
cached = cache[cache_key]
return AIMessage(content=f"{cached}\n\n*(de la caché)*")
# Si no hay acierto de caché, llamar al modelo
result = handler(request)
# El content de AIMessage puede estar mezclado en una lista; tomar directamente el primero
if hasattr(result, 'content'):
cache[cache_key] = result.content
print(f"[Sin acierto de caché] Guardado en la caché, actualmente {len(cache)} entradas")
elif hasattr(result, 'model_response'):
# Caso de ExtendedModelResponse
pass
return result
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[cache_responses],
system_prompt="Eres el asistente de EJEMPLO, responde de manera concisa.",
)
# Primera consulta (sin acierto de caché)
result = agent.invoke({
"messages": [{"role": "user", "content": "¿Qué es Python?"}]
})
print(f"Primera: {result['messages'][-1].content[:80]}...\n")
# Segunda consulta con la misma pregunta (acierto de caché)
result = agent.invoke({
"messages": [{"role": "user", "content": "¿Qué es Python?"}]
})
print(f"Segunda: {result['messages'][-1].content[:80]}...)
from langchain.messages import AIMessage
# Caché simple en memoria
cache = {}
@wrap_model_call
def cache_responses(request, handler):
"""Almacena en caché las respuestas del modelo; no repite llamadas para la misma pregunta"""
# Usar el contenido del último mensaje del usuario como clave de caché
messages = request.messages
if not messages:
return handler(request)
# Generar la clave de caché
last_content = str(messages[-1].content) if hasattr(messages[-1], 'content') else ""
cache_key = last_content[:200] # Truncar el contenido demasiado largo
# Verificar la caché
if cache_key in cache:
print(f"[Acierto de caché] Devolver directamente el resultado en caché")
cached = cache[cache_key]
return AIMessage(content=f"{cached}\n\n*(de la caché)*")
# Si no hay acierto de caché, llamar al modelo
result = handler(request)
# El content de AIMessage puede estar mezclado en una lista; tomar directamente el primero
if hasattr(result, 'content'):
cache[cache_key] = result.content
print(f"[Sin acierto de caché] Guardado en la caché, actualmente {len(cache)} entradas")
elif hasattr(result, 'model_response'):
# Caso de ExtendedModelResponse
pass
return result
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[cache_responses],
system_prompt="Eres el asistente de EJEMPLO, responde de manera concisa.",
)
# Primera consulta (sin acierto de caché)
result = agent.invoke({
"messages": [{"role": "user", "content": "¿Qué es Python?"}]
})
print(f"Primera: {result['messages'][-1].content[:80]}...\n")
# Segunda consulta con la misma pregunta (acierto de caché)
result = agent.invoke({
"messages": [{"role": "user", "content": "¿Qué es Python?"}]
})
print(f"Segunda: {result['messages'][-1].content[:80]}...)
Resultado de la ejecución:
[缓存未命中] 已存入缓存，当前 1 条 第一次: Python 是一种高级编程语言，以简洁易读的语法著称... [缓存命中] 直接返回缓存结果 第二次: Python 是一种高级编程语言，以简洁易读的语法著称... *(来自缓存)*
Escenario 4: Modificar request — inyección dinámica de mensajes del sistema
Ejemplo
from datetime import datetime
from langchain.agents.middleware import wrap_model_call
from langchain.messages import SystemMessage
@wrap_model_call
def inject_time_context(request, handler):
"""Inyectar la información de la hora actual antes de cada llamada al modelo"""
now = datetime.now()
time_context = (
f"Hora actual: {now.strftime('%Y/%m/%d %H:%M')}."
f"Hoy es {['Lunes', 'Martes', 'Miércoles', 'Jueves', 'Viernes', 'Sábado', 'Domingo'][now.weekday()]}."
)
# Añadir la información de la hora al system_message existente
if request.system_message:
new_content = f"{request.system_message.content}\n\n{time_context}"
else:
new_content = time_context
# Crear una nueva request con override
new_request = request.override(
system_message=SystemMessage(content=new_content)
)
return handler(new_request)
from langchain.agents.middleware import wrap_model_call
from langchain.messages import SystemMessage
@wrap_model_call
def inject_time_context(request, handler):
"""Inyectar la información de la hora actual antes de cada llamada al modelo"""
now = datetime.now()
time_context = (
f"Hora actual: {now.strftime('%Y/%m/%d %H:%M')}."
f"Hoy es {['Lunes', 'Martes', 'Miércoles', 'Jueves', 'Viernes', 'Sábado', 'Domingo'][now.weekday()]}."
)
# Añadir la información de la hora al system_message existente
if request.system_message:
new_content = f"{request.system_message.content}\n\n{time_context}"
else:
new_content = time_context
# Crear una nueva request con override
new_request = request.override(
system_message=SystemMessage(content=new_content)
)
return handler(new_request)
Escenario 5: Combinación de múltiples wrap_model_call
Múltiples middlewares wrap_model_call se combinan automáticamente en orden: el primero es el más externo:
Ejemplo
from langchain.agents.middleware import wrap_model_call
@wrap_model_call
def outer_middleware(request, handler):
"""Middleware más externo"""
print("[Externo] Inicio")
result = handler(request) # Aquí se entra en inner_middleware
print("[Externo] Fin")
return result
@wrap_model_call
def inner_middleware(request, handler):
"""Middleware interno"""
print(" [Interno] Inicio")
result = handler(request) # Aquí es donde realmente se llama al modelo
print(" [Interno] Fin")
return result
# Orden de ejecución:
# [Externo] Inicio
# [Interno] Inicio
# → Llamada real al modelo
# [Interno] Fin
# [Externo] Fin
@wrap_model_call
def outer_middleware(request, handler):
"""Middleware más externo"""
print("[Externo] Inicio")
result = handler(request) # Aquí se entra en inner_middleware
print("[Externo] Fin")
return result
@wrap_model_call
def inner_middleware(request, handler):
"""Middleware interno"""
print(" [Interno] Inicio")
result = handler(request) # Aquí es donde realmente se llama al modelo
print(" [Interno] Fin")
return result
# Orden de ejecución:
# [Externo] Inicio
# [Interno] Inicio
# → Llamada real al modelo
# [Interno] Fin
# [Externo] Fin
Otras extensionesMúltiples wrap_model_call se envuelven en capas como una cebolla. La capa más externa se ejecuta primero y retorna al final. Esto te permite combinar varias funciones independientes — por ejemplo, la capa externa hace la comprobación de caché y la interna hace los reintentos, sin interferir entre sí.