Streaming de Agentes

Descripción de la cápsula

En las cápsulas anteriores aprendiste a crear agentes con create_agent, configurar system prompts, y manejar estado y memoria. Pero hay un problema práctico: cuando llamas agent.invoke(), tu programa se bloquea hasta que el agente completa todas sus iteraciones internas — tool calls, análisis de resultados, más tool calls, y finalmente la respuesta. Si el agente necesita 3 iteraciones de tools, puedes esperar 10-15 segundos viendo... nada.

Con streaming, ves cada paso del agente en tiempo real: qué tool está llamando, qué retornó, qué está pensando el modelo. Esto no solo mejora la experiencia del usuario — te da visibilidad sobre el proceso de razonamiento del agente, lo cual es crítico para debugging y para construir interfaces profesionales.

agent.stream() es la interfaz de streaming de agentes en LangChain. Te permite elegir qué nivel de detalle quieres con stream_mode: el estado completo después de cada paso, solo los cambios, o incluso los tokens individuales del modelo.

agent.stream() — la interfaz de streaming

En lugar de agent.invoke() que retorna el resultado final, agent.stream() retorna un iterador que emite eventos a medida que el agente progresa.

Ejemplo básico: ver cada paso del agente

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información sobre un tema."""
    data = {
        "langchain": "LangChain es un framework para aplicaciones con LLMs.",
        "python": "Python es un lenguaje de programación de alto nivel.",
    }
    for key, value in data.items():
        if key in query.lower():
            return value
    return f"No se encontró información sobre: {query}"

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

for step in agent.stream({"messages": [("user", "¿Qué es LangChain?")]}):
    print(step)
    print("---")
# Output esperado:
# {'agent': {'messages': [AIMessage(content='', tool_calls=[{'name': 'search', 'args': {'query': 'langchain'}, 'id': 'call_abc123'}])]}}
# ---
# {'tools': {'messages': [ToolMessage(content='LangChain es un framework para aplicaciones con LLMs.', tool_call_id='call_abc123')]}}
# ---
# {'agent': {'messages': [AIMessage(content='LangChain es un framework diseñado para crear aplicaciones que utilizan modelos de lenguaje (LLMs).')]}}
# ---

Cada step es un diccionario donde la clave es el nombre del nodo que generó el evento ("agent" para el modelo, "tools" para la ejecución de tools) y el valor contiene los mensajes producidos.

stream_mode: controlando el nivel de detalle

El parámetro stream_mode define qué información emite el stream. Hay tres modos principales:

stream_mode="updates" (por defecto)

Emite solo los cambios producidos por cada nodo. Es el modo más práctico para la mayoría de casos.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

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

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

for step in agent.stream(
    {"messages": [("user", "¿Qué es LangChain?")]},
    stream_mode="updates"
):
    for node_name, update in step.items():
        print(f"[{node_name}]")
        if "messages" in update:
            for msg in update["messages"]:
                if isinstance(msg, AIMessage) and msg.tool_calls:
                    for tc in msg.tool_calls:
                        print(f"  → Llamando: {tc['name']}({tc['args']})")
                elif isinstance(msg, AIMessage):
                    print(f"  Respuesta: {msg.content[:100]}")
                elif isinstance(msg, ToolMessage):
                    print(f"  ← Resultado: {msg.content[:80]}")
# Output esperado:
# [agent]
#   → Llamando: search({'query': 'LangChain'})
# [tools]
#   ← Resultado: Resultado: LangChain es un concepto importante en AI.
# [agent]
#   Respuesta: LangChain es un concepto importante en el ámbito de la inteligencia artificial...

stream_mode="values"

Emite el estado completo del agente después de cada paso. Incluye todo el historial de mensajes acumulado.

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información sobre un tema."""
    return f"Resultado: {query} es un framework para LLMs."

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

for step in agent.stream(
    {"messages": [("user", "¿Qué es LangChain?")]},
    stream_mode="values"
):
    messages = step["messages"]
    print(f"Total de mensajes: {len(messages)}")
    last = messages[-1]
    print(f"  Último: {type(last).__name__} → {str(last.content)[:80]}")
    print("---")
# Output esperado:
# Total de mensajes: 1
#   Último: HumanMessage → ¿Qué es LangChain?
# ---
# Total de mensajes: 2
#   Último: AIMessage →
# ---
# Total de mensajes: 3
#   Último: ToolMessage → Resultado: LangChain es un framework para LLMs.
# ---
# Total de mensajes: 4
#   Último: AIMessage → LangChain es un framework diseñado para trabajar con modelos de len
# ---

Con "values", cada paso incluye todos los mensajes desde el inicio. Útil cuando necesitas el contexto completo en cada iteración.

stream_mode="messages"

Emite tokens individuales del modelo, dándote el máximo nivel de granularidad. Cada evento es una tupla (message_chunk, metadata).

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información sobre un tema."""
    return f"LangChain es un framework de código abierto para aplicaciones con LLMs."

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

for chunk, metadata in agent.stream(
    {"messages": [("user", "¿Qué es LangChain?")]},
    stream_mode="messages"
):
    if chunk.content:
        print(chunk.content, end="", flush=True)
print()
# Output esperado:
# LangChain es un framework de código abierto diseñado para crear aplicaciones que utilizan modelos de lenguaje (LLMs). Permite integrar herramientas, memoria y cadenas de procesamiento para construir sistemas inteligentes.

Con "messages", puedes mostrar la respuesta del agente token por token, exactamente como ChatGPT o Claude muestran sus respuestas.

Procesando eventos del stream: modelo vs tools

Cuando usas stream_mode="updates", necesitas distinguir entre eventos del modelo y eventos de tools para mostrar información relevante al usuario.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Obtiene el clima actual de una ciudad."""
    weathers = {
        "Madrid": "Soleado, 25°C",
        "París": "Nublado, 16°C",
        "Tokio": "Lluvioso, 19°C",
    }
    return weathers.get(city, f"Clima no disponible para {city}")

@tool
def get_population(city: str) -> str:
    """Obtiene la población de una ciudad."""
    populations = {
        "Madrid": "3.3 millones",
        "París": "2.1 millones",
        "Tokio": "13.9 millones",
    }
    return populations.get(city, f"Población no disponible para {city}")

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [get_weather, get_population])

step_count = 0
for step in agent.stream(
    {"messages": [("user", "¿Clima y población de Madrid?")]},
    stream_mode="updates"
):
    step_count += 1
    for node_name, update in step.items():
        if node_name == "agent":
            for msg in update["messages"]:
                if isinstance(msg, AIMessage) and msg.tool_calls:
                    print(f"[Paso {step_count}] Agente decide llamar tools:")
                    for tc in msg.tool_calls:
                        print(f"  → {tc['name']}({tc['args']})")
                elif isinstance(msg, AIMessage) and msg.content:
                    print(f"[Paso {step_count}] Agente responde:")
                    print(f"  {msg.content}")
        elif node_name == "tools":
            for msg in update["messages"]:
                if isinstance(msg, ToolMessage):
                    print(f"[Paso {step_count}] Tool retornó: {msg.content}")
# Output esperado:
# [Paso 1] Agente decide llamar tools:
#   → get_weather({'city': 'Madrid'})
#   → get_population({'city': 'Madrid'})
# [Paso 2] Tool retornó: Soleado, 25°C
# [Paso 2] Tool retornó: 3.3 millones
# [Paso 3] Agente responde:
#   Madrid tiene un clima soleado de 25°C y una población de 3.3 millones.

Streaming de tokens individuales del modelo

El modo "messages" te da acceso a cada token que genera el modelo. Esto es esencial para interfaces tipo chat.

Filtrando solo tokens de la respuesta final

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessageChunk

@tool
def search(query: str) -> str:
    """Busca información sobre un tema."""
    return f"{query}: es un framework popular para aplicaciones de IA."

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

for chunk, metadata in agent.stream(
    {"messages": [("user", "¿Qué es LangChain?")]},
    stream_mode="messages"
):
    if isinstance(chunk, AIMessageChunk):
        if chunk.tool_call_chunks:
            pass
        elif chunk.content:
            print(chunk.content, end="", flush=True)
print()
# Output esperado:
# LangChain es un framework popular para construir aplicaciones de inteligencia artificial que integran modelos de lenguaje con herramientas externas.

Usando metadata para identificar el nodo origen

La metadata incluye información sobre qué nodo del grafo generó el evento, lo que permite filtrar con precisión.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessageChunk

@tool
def calculate(expression: str) -> str:
    """Evalúa una expresión matemática."""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [calculate])

print("Tokens de la respuesta final:")
for chunk, metadata in agent.stream(
    {"messages": [("user", "¿Cuánto es 15 * 37 + 42?")]},
    stream_mode="messages"
):
    node = metadata.get("langgraph_node", "")
    if isinstance(chunk, AIMessageChunk) and chunk.content and node == "agent":
        print(chunk.content, end="", flush=True)
print()
# Output esperado:
# Tokens de la respuesta final:
# El resultado de 15 × 37 + 42 es **597**.

Mostrando progreso de tool calls al usuario

En una interfaz real, quieres mostrar al usuario qué está haciendo el agente: "Buscando información... Listo!" en lugar de silencio total.

from dotenv import load_dotenv
load_dotenv()

import time
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def search_web(query: str) -> str:
    """Busca información en la web."""
    time.sleep(0.5)
    return f"Encontrado: {query} tiene múltiples aplicaciones en producción."

@tool
def get_price(product: str) -> str:
    """Obtiene el precio de un producto."""
    time.sleep(0.3)
    prices = {"langchain": "Open source (gratis)", "langsmith": "Desde $39/mes"}
    return prices.get(product.lower(), f"Precio no disponible para {product}")

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search_web, get_price])

active_tools = {}

for step in agent.stream(
    {"messages": [("user", "Busca qué es LangChain y cuánto cuesta LangSmith")]},
    stream_mode="updates"
):
    for node_name, update in step.items():
        for msg in update.get("messages", []):
            if isinstance(msg, AIMessage) and msg.tool_calls:
                for tc in msg.tool_calls:
                    active_tools[tc["id"]] = tc["name"]
                    print(f"⏳ {tc['name']}({tc['args']})...")
            elif isinstance(msg, ToolMessage):
                tool_name = active_tools.get(msg.tool_call_id, "tool")
                print(f"✅ {tool_name} completado → {msg.content[:60]}")
            elif isinstance(msg, AIMessage) and msg.content:
                print(f"\n📝 Respuesta final:\n{msg.content}")
# Output esperado:
# ⏳ search_web({'query': 'LangChain'})...
# ⏳ get_price({'product': 'LangSmith'})...
# ✅ search_web completado → Encontrado: LangChain tiene múltiples aplicaciones en pro
# ✅ get_price completado → Desde $39/mes
#
# 📝 Respuesta final:
# LangChain es un framework con múltiples aplicaciones en producción. LangSmith tiene un precio desde $39/mes.

Streaming en contexto async (astream)

Para aplicaciones web o APIs, necesitas la versión async del streaming. agent.astream() funciona igual que agent.stream() pero dentro de un contexto async.

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def search(query: str) -> str:
    """Busca información sobre un tema."""
    return f"Resultado para '{query}': información relevante encontrada."

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

async def stream_agent():
    async for step in agent.astream(
        {"messages": [("user", "¿Qué es RAG?")]},
        stream_mode="updates"
    ):
        for node_name, update in step.items():
            for msg in update.get("messages", []):
                if isinstance(msg, AIMessage) and msg.tool_calls:
                    for tc in msg.tool_calls:
                        print(f"[async] Tool: {tc['name']}")
                elif isinstance(msg, AIMessage) and msg.content:
                    print(f"[async] Respuesta: {msg.content[:100]}")
                elif isinstance(msg, ToolMessage):
                    print(f"[async] Resultado: {msg.content[:60]}")

asyncio.run(stream_agent())
# Output esperado:
# [async] Tool: search
# [async] Resultado: Resultado para 'RAG': información relevante encontrada.
# [async] Respuesta: RAG (Retrieval-Augmented Generation) es una técnica que combina la recuperación de...

Streaming de tokens async

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessageChunk

@tool
def search(query: str) -> str:
    """Busca información sobre un tema."""
    return f"{query}: técnica de IA que combina búsqueda con generación."

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

async def stream_tokens():
    async for chunk, metadata in agent.astream(
        {"messages": [("user", "Explica RAG brevemente")]},
        stream_mode="messages"
    ):
        if isinstance(chunk, AIMessageChunk) and chunk.content:
            if not chunk.tool_call_chunks:
                print(chunk.content, end="", flush=True)
    print()

asyncio.run(stream_tokens())
# Output esperado:
# RAG (Retrieval-Augmented Generation) es una técnica de inteligencia artificial que combina la búsqueda de información en bases de datos con la generación de texto mediante modelos de lenguaje.

Construyendo output amigable para UI

En una aplicación real, necesitas transformar los eventos del stream en un formato que tu frontend pueda consumir.

Patrón: colectar eventos estructurados

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def search(query: str) -> str:
    """Busca información sobre un tema."""
    return f"Resultado: {query} es una tecnología moderna de IA."

@tool
def summarize(text: str) -> str:
    """Resume un texto."""
    return f"Resumen: {text[:50]}..."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search, summarize])

ui_events = []

for step in agent.stream(
    {"messages": [("user", "Busca qué es LangChain y resúmelo")]},
    stream_mode="updates"
):
    for node_name, update in step.items():
        for msg in update.get("messages", []):
            if isinstance(msg, AIMessage) and msg.tool_calls:
                for tc in msg.tool_calls:
                    ui_events.append({
                        "type": "tool_start",
                        "tool": tc["name"],
                        "args": tc["args"],
                    })
            elif isinstance(msg, ToolMessage):
                ui_events.append({
                    "type": "tool_result",
                    "content": msg.content,
                    "tool_call_id": msg.tool_call_id,
                })
            elif isinstance(msg, AIMessage) and msg.content:
                ui_events.append({
                    "type": "response",
                    "content": msg.content,
                })

print("Eventos para UI:")
for event in ui_events:
    print(f"  {event['type']}: ", end="")
    if event["type"] == "tool_start":
        print(f"{event['tool']}({event['args']})")
    elif event["type"] == "tool_result":
        print(f"{event['content'][:60]}")
    elif event["type"] == "response":
        print(f"{event['content'][:80]}")
# Output esperado:
# Eventos para UI:
#   tool_start: search({'query': 'LangChain'})
#   tool_result: Resultado: LangChain es una tecnología moderna de IA.
#   response: LangChain es una tecnología moderna de inteligencia artificial que permite const

Estos ui_events pueden enviarse a un frontend via WebSocket o Server-Sent Events para renderizar una interfaz de chat con indicadores de progreso.

Comparación: stream_mode "values" vs "updates" vs "messages"

Cuándo usar cada uno

• ✅ "updates" — Mayoría de aplicaciones. Muestra qué tools se llamaron y sus resultados
• ✅ "messages" — Interfaces de chat tipo ChatGPT donde la respuesta aparece token por token
• ✅ "values" — Debugging o cuando necesitas acceso al estado completo en cada paso
• ⚠️ No mezcles modos en una misma llamada a stream() — elige uno

Combinando modos: updates + tokens manuales

Si necesitas tanto progreso de tools como streaming de tokens, puedes hacer dos pasadas o usar "updates" y detectar la respuesta final para hacer un segundo stream:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage, AIMessageChunk

@tool
def search(query: str) -> str:
    """Busca información sobre un tema."""
    return f"{query}: framework de IA para producción."

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

for step in agent.stream(
    {"messages": [("user", "¿Qué es LangChain?")]},
    stream_mode="updates"
):
    for node_name, update in step.items():
        for msg in update.get("messages", []):
            if isinstance(msg, AIMessage) and msg.tool_calls:
                for tc in msg.tool_calls:
                    print(f"🔧 Llamando {tc['name']}...")
            elif isinstance(msg, ToolMessage):
                print(f"✅ Resultado recibido")
            elif isinstance(msg, AIMessage) and msg.content:
                print(f"💬 {msg.content}")
# Output esperado:
# 🔧 Llamando search...
# ✅ Resultado recibido
# 💬 LangChain es un framework de inteligencia artificial diseñado para aplicaciones en producción...

Conexión con el proyecto

En el Proyecto del módulo (Cápsula 08), usarás streaming para construir un agente de investigación donde el usuario ve en tiempo real: qué fuentes está consultando el agente, qué datos recolectó, y la respuesta final apareciendo token por token. El patrón de ui_events te servirá como base para integrar el agente en cualquier interfaz web.

Troubleshooting

Problema 1: El stream no emite nada

Causa: El agente no está configurado correctamente o la pregunta no activa ninguna tool. Solución: Verifica que el agente tiene tools y que la pregunta las active:

agent = create_agent(model, [search])
for step in agent.stream({"messages": [("user", "Busca info sobre Python")]}):
    print(step)

Problema 2: stream_mode="messages" no muestra tokens

Causa: Estás filtrando incorrectamente los chunks o el modelo retorna todo el contenido en un solo chunk. Solución: Imprime todos los chunks sin filtrar para diagnosticar:

for chunk, metadata in agent.stream(
    {"messages": [("user", "Hola")]},
    stream_mode="messages"
):
    print(f"type={type(chunk).__name__}, content='{chunk.content}', tool_chunks={bool(chunk.tool_call_chunks) if hasattr(chunk, 'tool_call_chunks') else 'N/A'}")

Problema 3: Confusión entre tool calls y respuesta final en updates

Causa: Un AIMessage puede tener tool_calls (el agente quiere llamar tools) o content (respuesta final), o ambos. Solución: Verifica tool_calls primero:

if isinstance(msg, AIMessage):
    if msg.tool_calls:
        pass  # El agente quiere llamar tools
    elif msg.content:
        pass  # Respuesta final del agente

Problema 4: astream no funciona fuera de async

Causa: astream requiere un contexto async (async def + await). Solución: Usa asyncio.run() o ejecuta dentro de un framework async:

import asyncio

async def main():
    async for step in agent.astream({"messages": [("user", "Hola")]}):
        print(step)

asyncio.run(main())

Problema 5: Eventos duplicados en el stream

Causa: Estás iterando sobre step.items() sin filtrar por nodo, y el mismo mensaje aparece en múltiples contextos. Solución: Filtra explícitamente por node_name:

for step in agent.stream(input_data, stream_mode="updates"):
    for node_name, update in step.items():
        if node_name == "agent":
            pass  # Solo eventos del modelo

Ejercicios

Ejercicio 1: Stream básico con updates (Fácil)

Crea un agente con una tool get_weather y haz stream con stream_mode="updates". Imprime cada paso identificando si es una decisión del agente o un resultado de tool.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Obtiene el clima actual de una ciudad."""
    weathers = {"Madrid": "Soleado, 24°C", "Lima": "Nublado, 19°C"}
    return weathers.get(city, f"Clima no disponible para {city}")

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

for step in agent.stream(
    {"messages": [("user", "¿Qué clima hace en Madrid?")]},
    stream_mode="updates"
):
    for node_name, update in step.items():
        for msg in update.get("messages", []):
            if isinstance(msg, AIMessage) and msg.tool_calls:
                print(f"[AGENTE] Decide llamar:")
                for tc in msg.tool_calls:
                    print(f"  → {tc['name']}({tc['args']})")
            elif isinstance(msg, ToolMessage):
                print(f"[TOOL] Resultado: {msg.content}")
            elif isinstance(msg, AIMessage) and msg.content:
                print(f"[AGENTE] Respuesta: {msg.content}")
# Output esperado:
# [AGENTE] Decide llamar:
#   → get_weather({'city': 'Madrid'})
# [TOOL] Resultado: Soleado, 24°C
# [AGENTE] Respuesta: El clima en Madrid es soleado con una temperatura de 24°C.

Ejercicio 2: Streaming token por token (Fácil)

Usa stream_mode="messages" para mostrar la respuesta final del agente token por token, ignorando los chunks de tool calls.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessageChunk

@tool
def get_info(topic: str) -> str:
    """Obtiene información sobre un tema."""
    return f"{topic} es una herramienta fundamental en el desarrollo moderno de software."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [get_info])

token_count = 0
for chunk, metadata in agent.stream(
    {"messages": [("user", "¿Qué es Docker?")]},
    stream_mode="messages"
):
    if isinstance(chunk, AIMessageChunk) and chunk.content:
        if not chunk.tool_call_chunks:
            print(chunk.content, end="", flush=True)
            token_count += 1
print(f"\n\n(Total: {token_count} chunks recibidos)")
# Output esperado:
# Docker es una herramienta fundamental en el desarrollo moderno de software que permite crear, desplegar y ejecutar aplicaciones en contenedores...
#
# (Total: ~30 chunks recibidos)

Ejercicio 3: Indicador de progreso para tools (Medio)

Crea un agente con dos tools y muestra un indicador de progreso estilo "⏳ Buscando... ✅ Listo" para cada tool call.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def search_docs(query: str) -> str:
    """Busca en la documentación."""
    return f"Documentación encontrada: {query} tiene 3 métodos principales."

@tool
def search_examples(query: str) -> str:
    """Busca ejemplos de código."""
    return f"Ejemplo encontrado: usar {query} con async/await."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search_docs, search_examples])

pending_tools = {}

for step in agent.stream(
    {"messages": [("user", "Busca documentación y ejemplos de LangChain agents")]},
    stream_mode="updates"
):
    for node_name, update in step.items():
        for msg in update.get("messages", []):
            if isinstance(msg, AIMessage) and msg.tool_calls:
                for tc in msg.tool_calls:
                    pending_tools[tc["id"]] = tc["name"]
                    print(f"⏳ {tc['name']}('{tc['args'].get('query', '')}')...")
            elif isinstance(msg, ToolMessage):
                tool_name = pending_tools.pop(msg.tool_call_id, "unknown")
                print(f"✅ {tool_name} → {msg.content[:50]}...")
            elif isinstance(msg, AIMessage) and msg.content:
                print(f"\n📋 Respuesta:\n{msg.content}")
# Output esperado:
# ⏳ search_docs('LangChain agents')...
# ⏳ search_examples('LangChain agents')...
# ✅ search_docs → Documentación encontrada: LangChain agents tien...
# ✅ search_examples → Ejemplo encontrado: usar LangChain agents con...
#
# 📋 Respuesta:
# Aquí tienes lo que encontré sobre LangChain agents...

Ejercicio 4: Comparar output de los tres stream_modes (Medio)

Ejecuta la misma pregunta con "values", "updates" y "messages". Para cada modo, cuenta cuántos eventos emite y muestra el tipo de cada evento.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessageChunk

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultado: {query} es un concepto de IA."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search])
input_data = {"messages": [("user", "¿Qué es RAG?")]}

print("=== stream_mode='values' ===")
count_values = 0
for step in agent.stream(input_data, stream_mode="values"):
    count_values += 1
    msgs = step["messages"]
    last = msgs[-1]
    print(f"  Paso {count_values}: {len(msgs)} mensajes, último={type(last).__name__}")
print(f"  Total: {count_values} eventos\n")

print("=== stream_mode='updates' ===")
count_updates = 0
for step in agent.stream(input_data, stream_mode="updates"):
    count_updates += 1
    for node, update in step.items():
        msg_types = [type(m).__name__ for m in update.get("messages", [])]
        print(f"  Paso {count_updates}: nodo={node}, tipos={msg_types}")
print(f"  Total: {count_updates} eventos\n")

print("=== stream_mode='messages' ===")
count_messages = 0
for chunk, metadata in agent.stream(input_data, stream_mode="messages"):
    count_messages += 1
print(f"  Total: {count_messages} chunks")
# Output esperado:
# === stream_mode='values' ===
#   Paso 1: 1 mensajes, último=HumanMessage
#   Paso 2: 2 mensajes, último=AIMessage
#   Paso 3: 3 mensajes, último=ToolMessage
#   Paso 4: 4 mensajes, último=AIMessage
#   Total: 4 eventos
#
# === stream_mode='updates' ===
#   Paso 1: nodo=agent, tipos=['AIMessage']
#   Paso 2: nodo=tools, tipos=['ToolMessage']
#   Paso 3: nodo=agent, tipos=['AIMessage']
#   Total: 3 eventos
#
# === stream_mode='messages' ===
#   Total: ~35 chunks

Ejercicio 5: Agente async con astream (Medio)

Reescribe un agente síncrono para usar astream con stream_mode="updates". Muestra el progreso de las tools y la respuesta final.

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def fetch_data(source: str) -> str:
    """Obtiene datos de una fuente."""
    sources = {
        "wikipedia": "RAG combina recuperación de documentos con generación de texto.",
        "arxiv": "RAG fue propuesto por Lewis et al. en 2020.",
    }
    return sources.get(source.lower(), f"Fuente '{source}' no disponible.")

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [fetch_data])

async def run_async_agent():
    print("Iniciando agente async...\n")
    async for step in agent.astream(
        {"messages": [("user", "Busca información sobre RAG en Wikipedia y arXiv")]},
        stream_mode="updates"
    ):
        for node_name, update in step.items():
            for msg in update.get("messages", []):
                if isinstance(msg, AIMessage) and msg.tool_calls:
                    for tc in msg.tool_calls:
                        print(f"⏳ [{node_name}] {tc['name']}({tc['args']})")
                elif isinstance(msg, ToolMessage):
                    print(f"✅ [{node_name}] {msg.content[:60]}")
                elif isinstance(msg, AIMessage) and msg.content:
                    print(f"\n💬 [{node_name}] {msg.content}")

asyncio.run(run_async_agent())
# Output esperado:
# Iniciando agente async...
#
# ⏳ [agent] fetch_data({'source': 'wikipedia'})
# ⏳ [agent] fetch_data({'source': 'arxiv'})
# ✅ [tools] RAG combina recuperación de documentos con generación de texto
# ✅ [tools] RAG fue propuesto por Lewis et al. en 2020.
#
# 💬 [agent] RAG (Retrieval-Augmented Generation) combina la recuperación de documentos con la generación de texto. Fue propuesto por Lewis et al. en 2020.

Ejercicio 6: Colector de eventos para frontend (Difícil)

Crea una función collect_ui_events(agent, question) que retorne una lista de eventos estructurados (tool_start, tool_end, token, done) listos para enviar a un frontend.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

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

@tool
def translate(text: str, lang: str) -> str:
    """Traduce texto a otro idioma."""
    return f"[{lang}] {text}"

def collect_ui_events(agent, question: str) -> list[dict]:
    """Colecta eventos del agente en formato listo para UI."""
    events = []
    pending = {}

    for step in agent.stream(
        {"messages": [("user", question)]},
        stream_mode="updates"
    ):
        for node_name, update in step.items():
            for msg in update.get("messages", []):
                if isinstance(msg, AIMessage) and msg.tool_calls:
                    for tc in msg.tool_calls:
                        pending[tc["id"]] = tc["name"]
                        events.append({
                            "type": "tool_start",
                            "tool": tc["name"],
                            "args": tc["args"],
                            "id": tc["id"],
                        })
                elif isinstance(msg, ToolMessage):
                    tool_name = pending.pop(msg.tool_call_id, "unknown")
                    events.append({
                        "type": "tool_end",
                        "tool": tool_name,
                        "result": msg.content,
                        "id": msg.tool_call_id,
                    })
                elif isinstance(msg, AIMessage) and msg.content:
                    events.append({
                        "type": "response",
                        "content": msg.content,
                    })

    events.append({"type": "done"})
    return events

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search, translate])

events = collect_ui_events(agent, "Busca qué es LangChain y tradúcelo al inglés")

for e in events:
    if e["type"] == "tool_start":
        print(f"🟡 START: {e['tool']}({e['args']})")
    elif e["type"] == "tool_end":
        print(f"🟢 END: {e['tool']} → {e['result'][:50]}")
    elif e["type"] == "response":
        print(f"💬 RESPONSE: {e['content'][:80]}")
    elif e["type"] == "done":
        print(f"⚫ DONE")

print(f"\nTotal eventos: {len(events)}")
# Output esperado:
# 🟡 START: search({'query': 'LangChain'})
# 🟢 END: search → Resultado para LangChain: información relevante enc
# 🟡 START: translate({'text': '...', 'lang': 'english'})
# 🟢 END: translate → [english] ...
# 💬 RESPONSE: LangChain es un framework... En inglés: ...
# ⚫ DONE
#
# Total eventos: 5

Resumen

En esta cápsula aprendiste:

• agent.stream() es la interfaz para ver el proceso del agente en tiempo real, paso a paso
• stream_mode="updates" emite solo los cambios por nodo — ideal para mostrar progreso de tools
• stream_mode="values" emite el estado completo acumulado — útil para debugging y logs
• stream_mode="messages" emite tokens individuales — perfecto para interfaces de chat en tiempo real
• Para procesar eventos, distingues entre AIMessage con tool_calls (decisiones del agente) y ToolMessage (resultados de tools)
• astream es la versión async para aplicaciones web y APIs
• El patrón profesional colecta eventos estructurados (tool_start, tool_end, response) listos para consumir desde un frontend

Próxima cápsula: Structured Output en Agentes — cómo forzar al agente a retornar datos estructurados (Pydantic) en lugar de texto libre.

Recursos adicionales

1. How to stream agent data to the client — Guía oficial de streaming de agentes
2. How to stream from a LangGraph agent — Streaming en LangGraph
3. Streaming Conceptual Guide — Guía conceptual de streaming
4. create_agent API Reference — Referencia de create_agent
5. Server-Sent Events with LangChain — SSE para frontends
6. AsyncIO Documentation — Para streaming async

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