@wrap_model_call: Interceptar Llamadas al Modelo

Descripción de la cápsula

@wrap_model_call es el middleware más potente del sistema de middleware de LangChain. Mientras que @before_model y @after_model solo observan — ven lo que entra y lo que sale del modelo, sin poder cambiar nada —, @wrap_model_call puede modificar tanto el request antes de enviarlo como la respuesta después de recibirla. Envuelve la llamada completa al modelo, dándote control total sobre qué recibe y qué devuelve.

En la cápsula anterior aprendiste a usar @before_model y @after_model para logging y monitoreo. Eran como cámaras de seguridad: ven todo pero no intervienen. @wrap_model_call es como un guardia de seguridad: ve todo, puede detener, modificar, o redirigir lo que pasa. Al terminar esta cápsula, sabrás interceptar llamadas al modelo para inyectar system messages, filtrar conversaciones largas, agregar metadata a respuestas, y aplicar lógica condicional basada en el contenido o el estado del agente.

El patrón handler

@wrap_model_call recibe tres parámetros:

1. messages — La lista de mensajes que se enviarán al modelo (system, human, AI, tool messages).
2. config — La configuración de ejecución (recursion_limit, configurable, metadata).
3. call_next — La función que hace la llamada real al modelo. Siempre debes llamarla.

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Obtiene el clima actual de una ciudad."""
    return f"El clima en {city} es soleado, 22°C"

def wrap_model_call(messages, config, call_next):
    """Intercepta la llamada al modelo — ve todo, puede cambiar todo."""
    print(f"[WRAP] Interceptando {len(messages)} mensajes")

    response = call_next(messages, config)

    print(f"[WRAP] Respuesta recibida: {response.content[:80]}")
    return response

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[get_weather],
    wrap_model_call=wrap_model_call,
)

result = agent.invoke({"messages": [("user", "¿Qué clima hace en Madrid?")]})
print(result["messages"][-1].content)

# Output esperado:
# [WRAP] Interceptando 1 mensajes
# [WRAP] Respuesta recibida: ...
# [WRAP] Interceptando 3 mensajes
# [WRAP] Respuesta recibida: El clima en Madrid es soleado...
# El clima en Madrid es soleado, con una temperatura de 22°C.

El middleware se ejecutó dos veces: una cuando el modelo decidió llamar a get_weather, y otra cuando generó la respuesta final. Esto es el loop ReAct en acción — cada paso por el model node pasa por tu middleware.

Si no llamas call_next, el modelo nunca se ejecuta y el agente lanza un error. call_next es el puente hacia el modelo — sin él, no hay respuesta.

El concepto de ModelRequest

Cuando @wrap_model_call intercepta una llamada, messages es el paquete completo que se enviará al modelo: system message (si configuraste prompt), historial de conversación, y último mensaje del usuario o resultado de tool.

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def calculator(expression: str) -> str:
    """Calcula una expresión matemática."""
    return str(eval(expression))

call_number = 0

def wrap_model_call(messages, config, call_next):
    """Inspecciona qué recibe el modelo en cada llamada."""
    global call_number
    call_number += 1

    print(f"\n=== Llamada #{call_number} al modelo ===")
    for i, msg in enumerate(messages):
        msg_type = type(msg).__name__
        if hasattr(msg, "tool_calls") and msg.tool_calls:
            calls = [(tc["name"], tc["args"]) for tc in msg.tool_calls]
            print(f"  [{i}] {msg_type}: tool_calls={calls}")
        else:
            content = msg.content[:60] if msg.content else "(sin contenido)"
            print(f"  [{i}] {msg_type}: {content}")

    return call_next(messages, config)

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[calculator],
    prompt="Eres un asistente matemático. Muestra los cálculos paso a paso.",
    wrap_model_call=wrap_model_call,
)

result = agent.invoke({"messages": [("user", "¿Cuánto es 15 * 8?")]})
print(f"\nRespuesta: {result['messages'][-1].content}")

# Output esperado:
# === Llamada #1 al modelo ===
#   [0] SystemMessage: Eres un asistente matemático. Muestra los cálculos paso
#   [1] HumanMessage: ¿Cuánto es 15 * 8?
#
# === Llamada #2 al modelo ===
#   [0] SystemMessage: Eres un asistente matemático. Muestra los cálculos paso
#   [1] HumanMessage: ¿Cuánto es 15 * 8?
#   [2] AIMessage: tool_calls=[('calculator', {'expression': '15 * 8'})]
#   [3] ToolMessage: 120
#
# Respuesta: 15 × 8 = 120.

En la llamada #1, el modelo recibe el system prompt + la pregunta. En la #2, recibe todo eso más el tool call que hizo y el resultado.

Modificar el request: inyectar system messages

El caso de uso más común — agregar un system message dinámico en cada llamada:

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage
from datetime import datetime

@tool
def search(query: str) -> str:
    """Busca información en internet."""
    return f"Resultado: LangChain v1.2 incluye create_agent con middleware."

def wrap_model_call(messages, config, call_next):
    """Inyecta un system message con la fecha actual."""
    date_msg = SystemMessage(
        content=f"Fecha actual: {datetime.now().strftime('%Y-%m-%d %H:%M')}. "
        f"Siempre menciona la fecha cuando des información temporal."
    )
    modified_messages = [date_msg] + list(messages)
    return call_next(modified_messages, config)

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[search],
    wrap_model_call=wrap_model_call,
)

result = agent.invoke({"messages": [("user", "¿Qué hay de nuevo en LangChain?")]})
print(result["messages"][-1].content)

# Output esperado:
# A fecha de 2026-02-28, LangChain v1.2 incluye create_agent con soporte para middleware.

A diferencia de prompt (que configuras al crear el agente), esta inyección ocurre en cada llamada al modelo y puede cambiar dinámicamente.

Modificar el request: filtrar mensajes sensibles

Puedes redactar información antes de enviarla al modelo:

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
import re

@tool
def process_order(order_id: str) -> str:
    """Procesa un pedido por su ID."""
    return f"Pedido {order_id} procesado correctamente."

def wrap_model_call(messages, config, call_next):
    """Redacta números de tarjeta de crédito."""
    redacted = []
    for msg in messages:
        if isinstance(msg, HumanMessage):
            cleaned = re.sub(
                r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',
                '[TARJETA REDACTADA]',
                msg.content
            )
            redacted.append(HumanMessage(content=cleaned))
        else:
            redacted.append(msg)
    return call_next(redacted, config)

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[process_order],
    wrap_model_call=wrap_model_call,
)

result = agent.invoke({
    "messages": [("user", "Procesa mi pedido ORD-123. Mi tarjeta es 4532 1234 5678 9012")]
})
print(result["messages"][-1].content)

# Output esperado:
# El pedido ORD-123 ha sido procesado correctamente.

El modelo nunca vio el número de tarjeta — seguridad a nivel de infraestructura, no dependes del prompt.

Leer estado dentro del middleware

Si tu agente usa state_schema custom, puedes acceder al estado a través de config:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage, SystemMessage
import operator

class SupportState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    user_tier: str

@tool
def search_kb(query: str) -> str:
    """Busca en la base de conocimiento."""
    return f"Artículo: '{query}' — Reinicia el dispositivo y verifica la conexión."

def wrap_model_call(messages, config, call_next):
    """Adapta el comportamiento según el tier del usuario."""
    user_tier = config.get("configurable", {}).get("user_tier", "free")

    tier_instruction = (
        "Este usuario es premium. Sé extra detallado y ofrece seguimiento."
        if user_tier == "premium"
        else "Usuario gratuito. Sé conciso. Sugiere upgrade para soporte avanzado."
    )
    tier_msg = SystemMessage(content=tier_instruction)
    return call_next([tier_msg] + list(messages), config)

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[search_kb],
    state_schema=SupportState,
    wrap_model_call=wrap_model_call,
)

result = agent.invoke(
    {"messages": [("user", "Mi internet no funciona")], "user_tier": "premium"},
    config={"configurable": {"user_tier": "premium"}},
)
print(result["messages"][-1].content)

# Output esperado:
# Lamento que estés teniendo problemas con tu conexión. Como usuario premium,
# te ofrezco los siguientes pasos detallados...

Modificar respuestas del modelo

@wrap_model_call también puede modificar lo que el modelo devuelve:

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage

@tool
def get_price(product: str) -> str:
    """Obtiene el precio de un producto."""
    prices = {"laptop": "999", "mouse": "29", "teclado": "79"}
    return prices.get(product.lower(), "Producto no encontrado")

def wrap_model_call(messages, config, call_next):
    """Agrega un disclaimer a la respuesta final."""
    response = call_next(messages, config)

    if response.content and not response.tool_calls:
        response = AIMessage(
            content=response.content + "\n\n_Respuesta generada por AI. Verifica antes de actuar._",
            tool_calls=response.tool_calls,
            response_metadata=response.response_metadata,
        )

    return response

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[get_price],
    wrap_model_call=wrap_model_call,
)

result = agent.invoke({"messages": [("user", "¿Cuánto cuesta un mouse?")]})
print(result["messages"][-1].content)

# Output esperado:
# El mouse cuesta $29.
#
# _Respuesta generada por AI. Verifica antes de actuar._

Solo agrega el disclaimer en respuestas finales (sin tool_calls), no en las intermedias donde el modelo decide qué tool llamar.

Lógica condicional: comportamiento dinámico

Aplica lógica diferente según el contenido de los mensajes:

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage, HumanMessage

@tool
def search(query: str) -> str:
    """Busca información en internet."""
    return f"Resultado: {query} — información relevante encontrada."

def wrap_model_call(messages, config, call_next):
    """Aplica instrucciones según el tipo de pregunta."""
    last_human = None
    for msg in reversed(messages):
        if isinstance(msg, HumanMessage):
            last_human = msg.content.lower()
            break

    if last_human and any(w in last_human for w in ["código", "programa", "script"]):
        extra = SystemMessage(
            content="El usuario pide código. Proporciona código completo y funcional."
        )
        messages = [extra] + list(messages)
    elif last_human and any(w in last_human for w in ["explica", "qué es", "cómo funciona"]):
        extra = SystemMessage(
            content="El usuario pide explicación. Usa analogías simples."
        )
        messages = [extra] + list(messages)

    return call_next(messages, config)

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[search],
    wrap_model_call=wrap_model_call,
)

result = agent.invoke({"messages": [("user", "Explica qué es un decorador en Python")]})
print(result["messages"][-1].content[:200])

# Output esperado:
# Un decorador en Python es como una envoltura de regalo. Imagina que tienes una
# función que hace algo, y quieres agregarle funcionalidad extra sin cambiarla...

Comparación: @before/@after vs @wrap_model_call

Regla general: usa @before_model / @after_model cuando solo necesites observar. Usa @wrap_model_call cuando necesites intervenir.

Troubleshooting

Problema 1: "AttributeError: 'NoneType' object has no attribute..."

Síntoma: Error al ejecutar el agente con wrap_model_call. Causa: Tu función no retorna nada (olvidaste return call_next(...) o return response). Solución: Asegúrate de retornar siempre el resultado:

def wrap_model_call(messages, config, call_next):
    response = call_next(messages, config)
    return response  # ← nunca olvides el return

Problema 2: El middleware se ejecuta más veces de lo esperado

Síntoma: Los prints del middleware aparecen 3, 4 o más veces. Causa: Se ejecuta en cada paso del loop ReAct por el model node. Si el agente hace 3 rondas de tools, el middleware se ejecuta 4 veces. Solución: Es el comportamiento esperado. Para actuar solo en la respuesta final:

def wrap_model_call(messages, config, call_next):
    response = call_next(messages, config)
    if not response.tool_calls:
        print("[WRAP] Esta es la respuesta final")
    return response

Problema 3: Error al modificar la respuesta — tool_calls se pierden

Síntoma: El agente falla después de modificar el AIMessage. Causa: No incluiste los tool_calls del original al crear un nuevo AIMessage. Solución: Preserva siempre los tool_calls:

from langchain_core.messages import AIMessage

def wrap_model_call(messages, config, call_next):
    response = call_next(messages, config)
    return AIMessage(
        content=response.content + " [modificado]",
        tool_calls=response.tool_calls,
        response_metadata=response.response_metadata,
    )

Ejercicios

Ejercicio 1: Middleware de logging básico (Fácil)

Crea un @wrap_model_call que imprima cuántos mensajes recibe el modelo y si la respuesta tiene tool calls o texto. Prueba con un agente que tenga una tool de calculadora.

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def calculator(expression: str) -> str:
    """Calcula una expresión matemática."""
    return str(eval(expression))

def wrap_model_call(messages, config, call_next):
    print(f"[LOG] Entrada: {len(messages)} mensajes")
    response = call_next(messages, config)
    has_tools = bool(response.tool_calls)
    print(f"[LOG] Salida: {'tool_calls' if has_tools else 'texto'}")
    return response

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[calculator],
    wrap_model_call=wrap_model_call,
)

result = agent.invoke({"messages": [("user", "¿Cuánto es 42 * 58?")]})
print(f"\nRespuesta: {result['messages'][-1].content}")

# Output esperado:
# [LOG] Entrada: 1 mensajes
# [LOG] Salida: tool_calls
# [LOG] Entrada: 3 mensajes
# [LOG] Salida: texto
#
# Respuesta: 42 × 58 = 2,436.

Ejercicio 2: Inyectar idioma forzado (Fácil)

Crea un middleware que fuerce al modelo a responder siempre en español, sin importar el idioma de la pregunta. Inyecta un system message con esta instrucción.

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage

@tool
def search(query: str) -> str:
    """Busca información en internet."""
    return f"Result for '{query}': Python was created by Guido van Rossum in 1991."

def wrap_model_call(messages, config, call_next):
    language_msg = SystemMessage(
        content="REGLA ABSOLUTA: Siempre responde en español, "
        "sin importar el idioma de la pregunta o de los resultados de tools."
    )
    return call_next([language_msg] + list(messages), config)

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[search],
    wrap_model_call=wrap_model_call,
)

result = agent.invoke({"messages": [("user", "Who created Python?")]})
print(result["messages"][-1].content)

# Output esperado:
# Python fue creado por Guido van Rossum en 1991.

Ejercicio 3: Cost tracker acumulativo (Medio)

Crea un middleware que rastree tokens y costo estimado. Al final, imprime un resumen con llamadas totales, tokens, y costo en USD.

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Busca información en internet."""
    return f"Resultado: {query} es un tema amplio."

costs = {"calls": 0, "input_tokens": 0, "output_tokens": 0}

def wrap_model_call(messages, config, call_next):
    costs["calls"] += 1
    response = call_next(messages, config)

    usage = response.response_metadata.get("token_usage", {})
    costs["input_tokens"] += usage.get("prompt_tokens", 0)
    costs["output_tokens"] += usage.get("completion_tokens", 0)
    return response

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[search],
    wrap_model_call=wrap_model_call,
)

result = agent.invoke({"messages": [("user", "¿Qué es Python?")]})
print(f"Respuesta: {result['messages'][-1].content}")

total = (costs["input_tokens"] / 1000) * 0.00015 + (costs["output_tokens"] / 1000) * 0.0006
print(f"\nLlamadas: {costs['calls']}")
print(f"Tokens: {costs['input_tokens']} in / {costs['output_tokens']} out")
print(f"Costo: ${total:.6f} USD")

# Output esperado:
# Respuesta: Python es un lenguaje de programación...
#
# Llamadas: 2
# Tokens: ~170 in / ~50 out
# Costo: $0.000056 USD

Ejercicio 4: Content filter con whitelist (Avanzado)

Crea un middleware que permita solo ciertos temas. Si el usuario pregunta fuera de la whitelist, retorna un mensaje de rechazo sin llamar al modelo (sin gastar tokens).

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, HumanMessage

@tool
def search(query: str) -> str:
    """Busca información en internet."""
    return f"Resultado: {query} — información encontrada."

ALLOWED = ["python", "langchain", "programación", "inteligencia artificial"]

def wrap_model_call(messages, config, call_next):
    last_human = None
    for msg in reversed(messages):
        if isinstance(msg, HumanMessage):
            last_human = msg.content.lower()
            break

    if last_human and not any(topic in last_human for topic in ALLOWED):
        return AIMessage(
            content="Solo puedo ayudarte con temas de programación, "
            "Python, LangChain e IA. ¿Tienes alguna pregunta sobre estos temas?"
        )

    return call_next(messages, config)

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[search],
    wrap_model_call=wrap_model_call,
)

for q in ["¿Qué es Python?", "¿Mejor receta de pasta?", "¿Cómo funciona LangChain?"]:
    result = agent.invoke({"messages": [("user", q)]})
    print(f"P: {q}\nR: {result['messages'][-1].content[:100]}\n")

# Output esperado:
# P: ¿Qué es Python?
# R: Python es un lenguaje de programación...
#
# P: ¿Mejor receta de pasta?
# R: Solo puedo ayudarte con temas de programación, Python, LangChain e IA...
#
# P: ¿Cómo funciona LangChain?
# R: LangChain es un framework para construir aplicaciones...

Ejercicio 5: Middleware compuesto — logging + cost tracking (Challenge)

Combina logging de entrada/salida y tracking de costos en un solo @wrap_model_call. Imprime un reporte al final.

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Busca información en internet."""
    return f"Resultado para '{query}': información pública."

@tool
def calculator(expression: str) -> str:
    """Calcula una expresión matemática."""
    return str(eval(expression))

metrics = {"calls": 0, "input_tokens": 0, "output_tokens": 0}

def wrap_model_call(messages, config, call_next):
    metrics["calls"] += 1
    call_num = metrics["calls"]
    print(f"[#{call_num}] ENTRADA: {len(messages)} mensajes")

    response = call_next(messages, config)

    usage = response.response_metadata.get("token_usage", {})
    metrics["input_tokens"] += usage.get("prompt_tokens", 0)
    metrics["output_tokens"] += usage.get("completion_tokens", 0)

    output_type = "tool_calls" if response.tool_calls else "texto"
    print(f"[#{call_num}] SALIDA: {output_type}")
    return response

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[search, calculator],
    wrap_model_call=wrap_model_call,
)

result = agent.invoke({"messages": [("user", "¿Qué es Python y cuánto es 100 * 3.14?")]})
print(f"\nRespuesta: {result['messages'][-1].content}")

total_cost = (
    (metrics["input_tokens"] / 1000) * 0.00015
    + (metrics["output_tokens"] / 1000) * 0.0006
)
print(f"\n=== REPORTE ===")
print(f"Llamadas: {metrics['calls']}")
print(f"Tokens: {metrics['input_tokens']} in / {metrics['output_tokens']} out")
print(f"Costo: ${total_cost:.6f} USD")

# Output esperado:
# [#1] ENTRADA: 1 mensajes
# [#1] SALIDA: tool_calls
# [#2] ENTRADA: 4 mensajes
# [#2] SALIDA: texto
#
# Respuesta: Python es un lenguaje... 100 × 3.14 = 314.
#
# === REPORTE ===
# Llamadas: 2
# Tokens: ~200 in / ~60 out
# Costo: $0.000066 USD

Resumen

En esta cápsula aprendiste:

• @wrap_model_call recibe (messages, config, call_next) y envuelve la llamada completa al modelo
• A diferencia de @before_model / @after_model, puede modificar tanto el request como la respuesta
• Siempre debes llamar call_next para que el modelo se ejecute (a menos que quieras bloquearlo intencionalmente)
• El middleware se ejecuta en cada paso del loop ReAct — una vez por cada paso del model node
• Puedes inyectar system messages dinámicos que cambian según el contexto
• Puedes filtrar mensajes antes de enviarlos (redactar información sensible)
• Puedes modificar respuestas después de recibirlas (disclaimers, metadata)
• Lógica condicional: analizar el contenido para aplicar comportamiento diferente
• Casos de uso clave: cost tracking, content filtering, request transformation, response enrichment
• Orden de ejecución con otros middleware: @before_model → @wrap_model_call → @after_model

Próxima cápsula: @wrap_tool_call — aprenderás a interceptar la ejecución de tools para agregar retry logic, custom error handling, y timing.

Recursos adicionales

1. create_agent API Reference — Documentación completa de parámetros de middleware
2. LangGraph Agents — Middleware — Guía oficial de middleware en agentes
3. Custom Agent Middleware — How-to de middleware custom
4. Token Usage Tracking — Rastreo de uso de tokens con metadata
5. Content Moderation with LangChain — Técnicas de moderación de contenido

Módulo 4 — LangChain & LangGraph: From Chains to Agents