Streaming de Tokens
Domina los modos de streaming de LangGraph — values, updates y streaming personalizado. Aprende a transmitir tokens LLM desde nodos del grafo en tiempo real.
Streaming de Tokens
Streaming lets you observe graph execution in real-time. LangGraph supports multiple streaming modes that give you fine-grained visibility into both state changes and LLM token output.
Visión General de Modos de Streaming
LangGraph provides three streaming modes:
| Modo | Descripción | Eventos |
|---|---|---|
"values" | Full state after each node | {"key": value, ...} |
"updates" | Only the changes from each node | {"node": {"key": new_value}} |
| Custom | Stream specific data from within nodes | Any data you yield |
Modo Values
Emits the complete state after each node execution:
config = {"configurable": {"thread_id": "stream-test"}}
for event in app.stream(
{"messages": [HumanMessage("Hello")]},
config,
stream_mode="values"
):
print(event)
# First event: initial state
# Second event: state after node 1
# Third event: state after node 2
# etc.Output:
{'messages': [HumanMessage('Hello')], 'output': ''}
{'messages': [HumanMessage('Hello'), AIMessage('Hi!')], 'output': 'Hi!'}
[!NOTA] "values" mode is the most comprehensive — you see the full state at every step. It's ideal for debugging and understanding state evolution.
Modo Updates
Emits only the changes from each node, keyed by node name:
for event in app.stream(
{"messages": [HumanMessage("Hello")]},
config,
stream_mode="updates"
):
for node_name, update in event.items():
if node_name != "__end__":
print(f"[{node_name}] → {update}")Output:
[node_a] → {'output': 'Processing...'}
[node_b] → {'output': 'Done!', 'messages': [AIMessage('Done!')]}
[!CONSEJO] "updates" mode is more efficient than "values" mode. Use it in production where you only care about what changed, not the full state.
Streaming de Tokens LLM
For real-time token-by-token output from an LLM inside a node, use LangGraph's callbacks or LangChain's .astream_events():
Using astream_events (Legacy)
from langchain_core.messages import HumanMessage
# In your node function, use astream_events
async def streaming_node(state: State) -> dict:
full = ""
async for event in llm.astream_events(
state["messages"],
version="v2"
):
if event["event"] == "on_chat_model_stream":
chunk = event["data"]["chunk"].content
full += chunk
# Send to client via WebSocket, etc.
return {"output": full}Using .astream() (Modorn)
async def streaming_node(state: State) -> dict:
full = ""
async for chunk in llm.astream(state["messages"]):
content = chunk.content
full += content
# Stream to client in real-time
print(content, end="", flush=True)
return {"output": full}[!IMPORTANTE]
Token streaming requires async (async for). You can use async def nodes in LangGraph, but the graph must be invoked with ainvoke() or astream().
Streaming Personalizado con Callbacks
Define custom callbacks to capture streaming data:
from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict, List
class TokenCollector(BaseCallbackHandler):
def __init__(self):
self.tokens: List[str] = []
def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
self.tokens.append(token)
def node_with_callback(state: State) -> dict:
collector = TokenCollector()
llm = ChatOpenAI(
model="gpt-4o-mini",
callbacks=[collector]
)
response = llm.invoke(state["messages"])
return {
"output": response.content,
"token_count": len(collector.tokens)
}Streaming de Nodos Específicos
Filter streaming to focus on specific nodes:
# Stream only updates from specific nodes
for event in app.stream(inputs, config, stream_mode="updates"):
for node_name, update in event.items():
if node_name == "chat_node":
# Process chat node updates
print(update)
elif node_name == "tool_node":
# Process tool node updates
print(update)Ejecución Asíncrona del Grafo con Streaming
import asyncio
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class StreamState(TypedDict):
messages: list
output: str
async def async_node(state: StreamState) -> dict:
full = ""
async for chunk in llm.astream(state["messages"]):
full += chunk.content
return {"output": full}
builder = StateGraph(StreamState)
builder.add_node("chat", async_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
app = builder.compile()
async def main():
async for event in app.astream(
{"messages": [HumanMessage("Tell me a story")], "output": ""},
stream_mode="updates"
):
print(event)
asyncio.run(main())Streaming con el Sistema de Eventos de LangGraph
LangGraph provides its own event system for custom streaming:
from langgraph.graph import StateGraph
from typing import Any, Dict, Iterator
def node_with_yield(state: State) -> Iterator[Dict[str, Any]]:
"""Use yield to stream intermediate results."""
yield {"status": "started"}
# Simulate work
import time
for i in range(3):
time.sleep(0.5)
yield {"progress": i * 33}
yield {"status": "complete", "result": "Done!"}
# This node can return or yield updates
builder.add_node("streaming_node", node_with_yield)Comparación de Modos de Streaming
| Modo | Token Level | State Level | Latency | Use Case |
|---|---|---|---|---|
"values" | No | Full state | After each node | Debugging, full visibility |
"updates" | No | Partial changes | After each node | Production, minimal data |
| Callback | Yes | No | Real-time | Token-by-token display |
| astream | Yes | No | Real-time | Async token streaming |
Ejemplo Completo de Streaming
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from typing_extensions import TypedDict, Annotated
from typing import List, Any
llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
class ChatState(TypedDict):
messages: Annotated[List[Any], add_messages]
def chat_node(state: ChatState) -> dict:
response = llm.invoke(state["messages"])
return {"messages": [response]}
builder = StateGraph(ChatState)
builder.add_node("chat", chat_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
app = builder.compile(checkpointer=MemorySaver())
# Stream values (full state after each node)
for event in app.stream(
{"messages": [HumanMessage("Tell me a story")]},
{"configurable": {"thread_id": "stream-demo"}},
stream_mode="values"
):
if "messages" in event and event["messages"]:
print(event["messages"][-1].content if hasattr(event["messages"][-1], "content") else event["messages"][-1])Streaming Asíncrono a Nivel de Token
import asyncio
async def main():
builder = StateGraph(ChatState)
builder.add_node("chat", chat_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
app = builder.compile()
async for event in app.astream(
{"messages": [HumanMessage("Write a poem")]},
stream_mode="updates"
):
for node, update in event.items():
if node == "chat" and "messages" in update:
msg = update["messages"][-1]
if hasattr(msg, "content"):
print(msg.content, end="", flush=True)
asyncio.run(main())[!ÉXITO] Async streaming provides real-time token output, making your agent feel responsive and interactive to end users.
Preguntas Prácticas
Which streaming mode emits the full state after each node?
Which streaming mode emits only the changes keyed by node name?
How do you stream tokens from an LLM inside a node?
What LangChain component captures LLM tokens during generation?
What method do you use for async graph execution with streaming?
Which streaming mode is most efficient for production use?
What is the event type for LLM token chunks in astream_events?
How do you enable streaming on a ChatOpenAI model?
What keyword allows a node function to emit multiple events?
What is the benefit of token-level streaming for end users?
[!ÉXITO]
Conclusiones Clave
- "values" mode: full state after each node (debugging)
- "updates" mode: only changes, keyed by node name (production)
- Token streaming via llm.stream() or llm.astream() with streaming=True
- BaseCallbackHandler on_llm_new_token captures individual tokens
- Async execution with astream() for non-blocking streaming
- yield in nodes allows multiple emissions per node
- Token streaming makes agents feel responsive to end users
- Choose the streaming mode based on your visibility vs efficiency needs