LangChain Salida en streaming

La salida en streaming hace que las respuestas de la IA se muestren palabra por palabra, como si se estuviera escribiendo, lo que mejora enormemente la experiencia del usuario. El Agente de LangChain incluye soporte completo para salida en streaming.

¿Por qué se necesita salida en streaming?

Si se usa invoke(), el usuario tiene que esperar a que el Agente complete todos los pasos (múltiples llamadas al modelo + ejecución de herramientas) antes de ver el resultado. Para tareas complejas, esto puede tardar más de diez segundos o incluso más.

La salida en streaming resuelve este problema: cada Token generado se devuelve de inmediato, y el usuario puede ver el progreso en tiempo real.

MétodoExperiencia de usuarioCasos de uso
invoke()Esperar → ver el resultado completo de una vezScripts, API, procesamiento por lotes
stream()Ver cada Token en tiempo realInterfaz de chat, visualización en tiempo real

stream_mode="messages" — Streaming token por token

Este es el modo de streaming con la granularidad más fina, cada chunk corresponde a un Token:

Ejemplo

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

model = init_chat_model("deepseek:deepseek-v4-flash")
agent = create_agent(
    model=model,
    system_prompt="Eres el asistente de Ejemplo.",
)

# stream_mode="messages" devuelve tokens uno a uno
print("Salida en streaming en tiempo real:")
for msg_chunk, metadata in agent.stream(
    {"messages": [HumanMessage(content="Presenta Ejemplo en una frase")]},
    stream_mode="messages",
):
    # msg_chunk es un AIMessageChunk
    # Cada chunk contiene solo una pequeña parte del contenido
    if msg_chunk.content:
        print(msg_chunk.content, end="", flush=True)
print()  # Salto de línea final

Resultado de la ejecución:

Salida en streaming en vivo:
Ejemplo es una plataforma gratuita de aprendizaje en línea para principiantes de programación, con tutoriales técnicos completos y ejemplos prácticos.

Entendiendo los metadatos

Los metadatos contienen información sobre el origen de este chunk:

Ejemplo

print("Ver información de los metadatos:\n")
for msg_chunk, metadata in agent.stream(
    {"messages": [HumanMessage(content="Hola, preséntate")]},
    stream_mode="messages",
):
    if msg_chunk.content and len(msg_chunk.content) > 5:
        print(f"Contenido: {msg_chunk.content}")
        print(f"Nodo de origen: {metadata.get('langgraph_node')}")
        print(f"Tipo de mensaje: {type(msg_chunk).__name__}")
        break  # Solo mirar el primer chunk significativo

Resultado de la ejecución:

Consultar metadatos:

Contenido: ¡Hola! Soy un asistente de IA
Nodo de origen: model
Tipo de mensaje: AIMessageChunk

stream_mode="updates" — Ver el proceso de ejecución del Agente paso a paso

Este modo es muy útil al construir interfaces que necesitan mostrar el "proceso de razonamiento":

Ejemplo

from langchain.tools import tool


@tool
def search_course(keyword: str) -> str:
    """Buscar cursos en Ejemplo"""
    courses = {
        "python": "Tutorial básico de Python3 (30 capítulos, 20 horas)",
        "html": "Tutorial básico de HTML (25 capítulos, 15 horas)",
    }
    return courses.get(keyword.lower(), "No se encontraron cursos relacionados")


agent = create_agent(
    model=init_chat_model("deepseek:deepseek-v4-flash", temperature=0),
    tools=[search_course],
    system_prompt="Eres el asesor de cursos de Ejemplo.",
)

# Usar el modo updates para ver cada paso
print("=== Proceso de ejecución del Agente ==="\n")
for chunk in agent.stream(
    {"messages": [HumanMessage(content="Ayúdame a buscar el curso de Python")]},
    stream_mode="updates",
):
    for node_name, update in chunk.items():
        print(f"[{node_name}]", end=" ")
        if "messages" in update:
            for msg in update["messages"]:
                if msg.type == "ai":
                    if hasattr(msg, 'tool_calls') and msg.tool_calls:
                        calls = [tc['name'] for tc in msg.tool_calls]
                        print(f"Llamadas solicitadas: {calls}")
                    elif msg.content:
                        print(f"Respuesta: {msg.content[:80]}")
                elif msg.type == "tool":
                    print(f"Devolución de herramienta [{msg.name}]: {msg.content}")

Resultado de la ejecución:

=== Agent 执行过程 ===

[model] 请求调用: ['search_course']
[tools] 工具返回 [search_course]: Python3 基础教程（30章，20小时）
[model] Respuesta: Ejemplo tiene el tutorial básico de Python3, 30 capítulos, unas 20 horas de estudio, perfecto para principiantes de Python.

stream_mode="custom" — Enviar eventos personalizados

A través de runtime.stream_writer() del Middleware, puedes enviar eventos personalizados al flujo:

Ejemplo

from langchain.agents.middleware import before_model, after_model


@before_model
def notify_before(state, runtime):
    """Enviar evento personalizado antes de la llamada al modelo"""
    runtime.stream_writer({
        "type": "status",
        "message": "Pensando...",
    })
    return None


@after_model
def notify_after(state, runtime):
    """Enviar evento personalizado después de la llamada al modelo"""
    last_msg = state["messages"][-1] if state.get("messages") else None
    has_tools = hasattr(last_msg, 'tool_calls') and last_msg.tool_calls

    if has_tools:
        tool_names = [tc['name'] for tc in last_msg.tool_calls]
        runtime.stream_writer({
            "type": "status",
            "message": f"Llamando a la herramienta: {', '.join(tool_names)}...",
        })
    else:
        runtime.stream_writer({
            "type": "status",
            "message": "Respuesta completada",
        })
    return None


agent = create_agent(
    model=init_chat_model("deepseek:deepseek-v4-flash", temperature=0),
    tools=[search_course],
    middleware=[notify_before, notify_after],
    system_prompt="Eres el asesor de cursos de Ejemplo.",
)

# Usar stream_mode=["updates", "custom"] para recibir ambos tipos de eventos a la vez
print("=== Salida en streaming mixta ==="\n")
for mode, chunk in agent.stream(
    {"messages": [HumanMessage(content="Busca el curso de Python")]},
    stream_mode=["updates", "custom"],
):
    if mode == "custom":
        print(f"[Evento personalizado] Estado: {chunk['message']}")
    elif mode == "updates":
        for node_name, update in chunk.items():
            if "messages" in update:
                for msg in update["messages"]:
                    if msg.type == "ai" and msg.content:
                        print(f"[Respuesta] {msg.content}")

Resultado de la ejecución:

=== 混合流式输出 ===

[evento personalizado] Estado: pensando...
[evento personalizado] Estado: llamando a la herramienta: search_course...
[respuesta] Ejemplo tiene el tutorial básico de Python3, 30 capítulos, unas 20 horas de estudio.
[evento personalizado] Estado: pensando...
[evento personalizado] Estado: respuesta completada

Los modos de stream_mode se pueden combinar, por ejemplo stream_mode=["updates", "custom", "messages"]. Sin embargo, demasiados modos aumentan la cantidad de eventos en el flujo; se recomienda elegir según la necesidad.

Salida en streaming asíncrona

En servicios web, usar streaming asíncrono evita bloquear el bucle de eventos:

Ejemplo

import asyncio


async def stream_agent():
    """Ejecutar el Agente con streaming asíncrono"""
    agent = create_agent(
        model=init_chat_model("deepseek:deepseek-v4-flash"),
        system_prompt="Eres el asistente de Ejemplo.",
    )

    full_response = ""
    async for msg_chunk, metadata in agent.astream(
        {"messages": [HumanMessage(content="Presenta Ejemplo en una frase")]},
        stream_mode="messages",
    ):
        if msg_chunk.content:
            full_response += msg_chunk.content
            print(msg_chunk.content, end="", flush=True)

    print(f"\n\nLongitud de la respuesta completa: {len(full_response)} caracteres")


# Ejecutar la función asíncrona
asyncio.run(stream_agent())

Ejemplo de integración con FastAPI

Ejemplo

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

app = FastAPI()
agent = create_agent(
    model=init_chat_model("deepseek:deepseek-v4-flash"),
    system_prompt="Eres el asistente de Ejemplo.",
)


@app.get("/chat")
async def chat(message: str):
    """Interfaz de chat que devuelve una respuesta SSE en streaming"""
    async def generate():
        async for msg_chunk, metadata in agent.astream(
            {"messages": [HumanMessage(content=message)]},
            stream_mode="messages",
        ):
            if msg_chunk.content:
                # Formato SSE: data: xxx\n\n
                yield f"data: {msg_chunk.content}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
    )

# Iniciar: uvicorn main:app --reload

En entornos de producción, se recomienda crear la instancia del Agente como un singleton global para evitar crearla de nuevo en cada solicitud. El costo de creación del Agente es bajo (principalmente compilar el grafo), pero reutilizar la instancia es más eficiente.

Tabla de referencia rápida de stream_mode

ModoGranularidadObjeto de iteraciónUsos típicos
messagesNivel de Token(AIMessageChunk, metadata)Efecto de escritura, chat en tiempo real
updatesNivel de nodo{node_name: state_update}Mostrar el proceso de razonamiento
valuesNivel de nodo (completo)Estado completoInstantánea de estado, depuración
customPersonalizadoCualquier dictNotificaciones de progreso, envío de estado
debugDetalladoInformación de depuraciónResolución de problemas en la etapa de desarrollo
Otras extensiones