LangGraph — Módulo 8: streaming y observabilidad

Cierre del bloque intermedio del Curso completo de LangGraph. Tu agente ya piensa, actúa, recuerda y pide permiso. Pero desde fuera sigue siendo una caja negra: llamas invoke(), esperas quince segundos mirando un spinner, y aparece un resultado. Este módulo es sobre abrir esa caja — para el usuario (streaming) y para ti (observabilidad).

La tesis del módulo

Un agente que tarda 15 segundos en silencio se percibe como roto; uno que tarda 15 segundos mostrando lo que hace se percibe como trabajando. El streaming no es una optimización de rendimiento: es de percepción. Y su gemelo, la observabilidad, es lo que te permite responder “¿por qué hizo eso?” cuando falle.

stream(): el reemplazo de invoke()

La API es la misma que ya conoces, pero devuelve un iterador en vez de un resultado:

for chunk in agente.stream(entrada, config):
    print(chunk)

Lo interesante está en qué emite, y eso lo controla stream_mode.

Los cinco modos

Modo	Qué emite	Para qué
"updates"	Solo lo que cambió cada nodo	Trazar el flujo (default)
"values"	El estado completo tras cada paso	Ver el estado acumulado
"messages"	Tokens del LLM a medida que llegan	El efecto “máquina de escribir”
"custom"	Lo que tú emitas desde un nodo	Progreso de negocio
"debug"	Todo: checkpoints, tareas, tiempos	Depurar en profundidad

     QUÉ EMITE CADA stream_mode

   ┌──────────┐
   │          │  updates   ──► {"asistente": {...}}   delta
   │          │  values    ──► estado completo        foto
   │  grafo   │  messages  ──► token, token, token    UI
   │          │  custom    ──► {"paso": "3 de 40"}    progreso
   │          │  debug     ──► checkpoints, tareas    depurar
   └──────────┘

updates — qué hizo cada nodo

for chunk in agente.stream(entrada, config, stream_mode="updates"):
    for nodo, actualizacion in chunk.items():
        print(f"[{nodo}] escribió: {list(actualizacion.keys())}")

# [asistente] escribió: ['messages']
# [tools]     escribió: ['messages']
# [asistente] escribió: ['messages']

Este es tu mejor amigo para entender el recorrido de un agente. En tres líneas ves el ciclo ReAct completo ocurriendo.

values — el estado completo

for estado in agente.stream(entrada, config, stream_mode="values"):
    estado["messages"][-1].pretty_print()

Cada emisión es el estado entero. Cómodo, pero pesado si el estado es grande: recibes el historial completo en cada paso.

La diferencia en una línea: updates te da el delta; values te da la foto.

messages — el efecto máquina de escribir

Este es el que el usuario ve. Emite tuplas (fragmento_de_mensaje, metadata) a medida que el modelo genera:

for fragmento, metadata in agente.stream(entrada, config, stream_mode="messages"):
    if fragmento.content:
        print(fragmento.content, end="", flush=True)

El metadata trae información del contexto —entre otras cosas, langgraph_node— y es lo que te permite filtrar. En un agente ReAct, el modelo se invoca varias veces; normalmente solo quieres mostrar al usuario la respuesta final, no el razonamiento intermedio:

for fragmento, metadata in agente.stream(entrada, config, stream_mode="messages"):
    if metadata.get("langgraph_node") == "asistente" and fragmento.content:
        print(fragmento.content, end="", flush=True)

custom — tu propio progreso

Los modos anteriores emiten lo que hace el grafo. Pero a veces quieres emitir algo tuyo: “consultando 3 de 40 documentos”. Para eso está get_stream_writer():

from langgraph.config import get_stream_writer

def buscar_documentos(state: State) -> dict:
    writer = get_stream_writer()
    encontrados = []
    for i, fuente in enumerate(state["fuentes"], start=1):
        writer({"paso": f"Revisando fuente {i} de {len(state['fuentes'])}"})
        encontrados.append(consultar(fuente))
    return {"documentos": encontrados}

for evento in agente.stream(entrada, config, stream_mode="custom"):
    print(evento["paso"])

Esto es lo que convierte un spinner genérico en una barra de progreso que explica qué está pasando. En la práctica es el modo que más diferencia hace en la experiencia percibida.

Combinar modos

Puedes pedir varios a la vez. Cuando lo haces, cada emisión llega como una tupla (modo, dato):

for modo, dato in agente.stream(entrada, config, stream_mode=["updates", "custom"]):
    if modo == "custom":
        mostrar_progreso(dato)
    elif modo == "updates":
        registrar(dato)

Típico en una app real: messages para el texto en pantalla y custom para los indicadores de estado.

Streaming asíncrono

Si tu backend es async (FastAPI, por ejemplo), la versión es astream y funciona igual:

async for fragmento, metadata in agente.astream(entrada, config, stream_mode="messages"):
    if fragmento.content:
        yield fragmento.content

Ese yield conecta directamente con una respuesta SSE (Server-Sent Events) hacia el navegador. Es, literalmente, cómo se construye la interfaz de chat que ves en cualquier producto de IA.

Observabilidad: la otra mitad

El streaming es para el usuario. La observabilidad es para ti: poder reconstruir qué pasó en una ejecución que salió mal — qué prompt recibió el modelo, qué tools llamó, con qué argumentos, cuánto tardó, cuánto costó.

Lo que ya tienes gratis

Del Módulo 6: si compilaste con checkpointer, el historial completo está guardado.

for snap in agente.get_state_history(config):
    print(snap.metadata["step"], snap.next, snap.metadata.get("writes"))

Eso reconstruye la ejecución paso a paso, después del hecho. Para muchos bugs es suficiente.

LangSmith: trazas con un par de variables

Para producción quieres algo visual y buscable. LangSmith se activa por entorno, sin tocar el código:

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="tu-api-key"
export LANGSMITH_PROJECT="agente-bodega"

A partir de ahí, cada ejecución queda registrada como un árbol: nodo por nodo, con los prompts exactos, las tool calls, los tokens y la latencia de cada paso. Cuando un usuario reporta “me respondió cualquier cosa”, abres la traza y ves el prompt real que recibió el modelo. Casi siempre la respuesta está ahí.

No es la única opción —hay integraciones con OpenTelemetry y otras plataformas— pero es la de menor fricción con LangGraph.

Errores comunes

• Usar values cuando querías updates. Recibes el estado completo en cada paso y te preguntas por qué todo va lento.
• No filtrar en messages. El usuario ve el razonamiento interno y las llamadas a tools mezcladas con la respuesta. Filtra por langgraph_node.
• Olvidar flush=True al imprimir en consola: el buffer se traga el efecto de streaming.
• Llamar get_stream_writer() fuera de un nodo. Necesita el contexto de ejecución del grafo.
• Loguear el estado completo en producción. Los messages traen datos del usuario. Registra metadata, no contenido, salvo que hayas resuelto el tema de privacidad.
• Confiar en los print() como observabilidad. Sirven en tu máquina; en producción necesitas trazas persistentes.

Cierre del bloque intermedio

Con los módulos 5 a 8 ya tienes todo lo necesario para un agente de verdad: usa herramientas, recuerda, pide permiso antes de actuar y muestra lo que está haciendo mientras lo hace. Eso ya es un producto.

Lo que viene en el bloque avanzado es escala: varios agentes coordinados, memoria que persiste entre conversaciones, y las piezas que hacen que esto aguante tráfico real.

La frase para llevarse

“El streaming es para que el usuario confíe en tu agente. La observabilidad es para que tú puedas defenderlo cuando falle. Las dos son la misma pregunta: ¿qué está pasando ahí dentro?”

Qué viene

Empieza el bloque avanzado (Módulo 9): sistemas multi-agente. Cuando un solo agente con veinte tools se vuelve inmanejable, la respuesta es dividirlo en especialistas con un supervisor que coordina — y en LangGraph eso son subgrafos.

Material relacionado

• Glosario de LangGraph
• Módulo 7: human-in-the-loop
• Módulo 6: persistencia y memoria