Streaming en agentes

Stream progreso del agente

Cuando un agente ejecuta una tarea compleja que involucra múltiples herramientas, es fundamental proporcionar feedback en tiempo real al usuario sobre lo que está ocurriendo. El sistema de streaming de LangChain permite visualizar cada paso del agente a medida que se produce, mejorando significativamente la experiencia de usuario en aplicaciones interactivas.

El modo de streaming updates emite eventos después de cada paso del agente. Si el agente llama a una herramienta, veremos tres actualizaciones principales: la decisión del modelo de llamar a la herramienta, el resultado de la ejecución de la herramienta, y la respuesta final del modelo.

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

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

agent = create_agent(
    model="gpt-5.1",
    tools=[obtener_clima]
)

# Streaming con modo "updates" para ver el progreso paso a paso
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "¿Qué tiempo hace en Barcelona?"}]},
    stream_mode="updates"
):
    for paso, datos in chunk.items():
        print(f"Paso: {paso}")
        print(f"Contenido: {datos['messages'][-1].content_blocks}")
        print()

Este código produce una salida similar a:

Paso: model
Contenido: [{'type': 'tool_call', 'name': 'obtener_clima', 'args': {'ciudad': 'Barcelona'}, 'id': 'call_abc123'}]

Paso: tools
Contenido: [{'type': 'text', 'text': 'El clima en Barcelona es soleado, 24°C'}]

Paso: model
Contenido: [{'type': 'text', 'text': 'El clima en Barcelona es soleado con una temperatura de 24°C.'}]

Cada paso representa un nodo en el grafo de ejecución del agente. El nodo model corresponde a las decisiones del LLM, mientras que el nodo tools muestra los resultados de las herramientas ejecutadas.

Stream LLM tokens

Para aplicaciones donde necesitamos mostrar la respuesta del modelo token a token mientras se genera, utilizamos el modo messages. Este modo es especialmente útil para crear interfaces de chat donde el texto aparece progresivamente:

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

@tool
def buscar_informacion(tema: str) -> str:
    """Busca información sobre un tema."""
    return f"Información relevante sobre {tema}: datos actualizados disponibles."

agent = create_agent(
    model="gpt-5.1",
    tools=[buscar_informacion]
)

for token, metadata in agent.stream(
    {"messages": [{"role": "user", "content": "Busca información sobre energías renovables"}]},
    stream_mode="messages"
):
    print(f"nodo: {metadata['langgraph_node']}")
    print(f"contenido: {token.content_blocks}")

Este código produce una salida donde cada token incluye su nodo de origen y el contenido:

nodo: model
contenido: [{'type': 'tool_call_chunk', 'name': 'buscar_informacion', 'args': '', 'index': 0}]

nodo: model
contenido: [{'type': 'tool_call_chunk', 'args': '{"tema":', 'index': 0}]

nodo: tools
contenido: [{'type': 'text', 'text': 'Información relevante sobre energías renovables: datos actualizados disponibles.'}]

nodo: model
contenido: [{'type': 'text', 'text': 'He encontrado'}]

El parámetro metadata contiene información adicional sobre el contexto del streaming, incluyendo el nodo del grafo que emite cada token (langgraph_node). Esto permite filtrar o procesar los tokens según su origen, diferenciando entre tokens del modelo y resultados de herramientas.

Stream updates

El modo custom permite emitir actualizaciones personalizadas desde dentro de las herramientas mientras se ejecutan. Esta funcionalidad resulta muy útil para informar al usuario sobre el progreso de operaciones largas o procesos en múltiples etapas.

Para emitir actualizaciones personalizadas, utilizamos la función get_stream_writer del módulo langgraph.config:

from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.config import get_stream_writer

@tool
def analizar_documento(documento: str) -> str:
    """Analiza un documento y extrae información relevante."""
    writer = get_stream_writer()
    
    # Emitimos actualizaciones de progreso
    writer("Iniciando análisis del documento...")
    writer("Extrayendo texto principal...")
    writer("Identificando conceptos clave...")
    writer("Análisis completado")
    
    return f"Documento '{documento}' analizado correctamente. Contiene 3 secciones principales."

agent = create_agent(
    model="gpt-5.1",
    tools=[analizar_documento]
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analiza el documento informe_ventas.pdf"}]},
    stream_mode="custom"
):
    print(f"Actualización: {chunk}")

La salida muestra cada mensaje emitido por el writer:

Actualización: Iniciando análisis del documento...
Actualización: Extrayendo texto principal...
Actualización: Identificando conceptos clave...
Actualización: Análisis completado

Combinación de modos de streaming

En aplicaciones de producción, frecuentemente necesitamos combinar varios modos de streaming para obtener tanto el progreso del agente como las actualizaciones personalizadas. Esto se logra pasando una lista de modos:

from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.config import get_stream_writer

@tool
def procesar_datos(origen: str) -> str:
    """Procesa datos de una fuente específica."""
    writer = get_stream_writer()
    writer(f"Conectando con {origen}...")
    writer("Procesando registros...")
    return f"Procesados 150 registros de {origen}"

agent = create_agent(
    model="gpt-5.1",
    tools=[procesar_datos]
)

# Combinamos modos "updates" y "custom"
for modo, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Procesa los datos del servidor principal"}]},
    stream_mode=["updates", "custom"]
):
    if modo == "custom":
        print(f"[Progreso] {chunk}")
    elif modo == "updates":
        for paso, datos in chunk.items():
            print(f"[{paso}] Paso completado")

Esta combinación permite construir interfaces que muestran simultáneamente el estado general del agente y los mensajes de progreso específicos de cada herramienta, proporcionando una experiencia completa al usuario sobre lo que está ocurriendo durante la ejecución.