Streaming Responses in Production with LangGraph: Complete Guide
Building AI applications that feel responsive is one of the hardest challenges in production. Users don't want to stare at a loading spinner for thirty seconds while your agent thinks through a complex task. They want to see progress in real time — tokens appearing, intermediate steps completing, state updates flowing. LangGraph, built on top of LangChain, provides a robust streaming API that makes this possible, but using it correctly in production requires understanding its event model, choosing the right streaming modes, and handling edge cases like backpressure, errors, and partial failures.
This guide walks through everything you need to know to ship streaming responses with LangGraph in a production environment. We'll cover the streaming modes available, how to integrate them with web frameworks, how to handle errors gracefully, and the architectural patterns that scale.
What Streaming Means in LangGraph
LangGraph models your application as a stateful graph: nodes represent computation steps, edges represent transitions, and a shared state object flows through the graph. When you invoke a graph, LangGraph executes nodes in order, updating state at each step. Streaming in this context means emitting events as the graph runs rather than waiting for the entire execution to finish.
There are several distinct things you might want to stream:
- Token streaming — individual tokens from LLM calls as they're generated
- Node updates — state changes after each node completes
- Custom events — arbitrary events you emit from within nodes using
get_stream_writer() - Debug events — detailed execution traces useful during development
Each of these corresponds to a streaming mode in LangGraph's API, and you can subscribe to multiple modes simultaneously.
Why Streaming Matters in Production
Streaming is not just a nice-to-have UX feature. In production systems it serves several critical purposes. First, it dramatically improves perceived latency. A response that streams its first token in 200ms feels instant even if the full response takes ten seconds, whereas a buffered response that appears all at once after ten seconds feels broken. Second, streaming provides observability into long-running agent workflows. When an agent executes a multi-step plan, streaming node updates lets you show the user what's happening — "Searching the web...", "Reading document 3 of 7...", "Composing answer..." — which builds trust and sets expectations. Third, streaming enables early termination. If the user can see the output forming and realizes it's going in the wrong direction, they can cancel mid-stream, saving compute costs.
There's also a reliability angle. Long-running synchronous requests are fragile. Load balancers, proxies, and CDNs all have timeout defaults that can kill a request that takes more than 30 or 60 seconds. Streaming keeps the connection alive with a steady flow of data, sidestepping these limits entirely.
Setting Up a Streaming-Ready Graph
Before diving into streaming mechanics, let's build a graph that's worth streaming. We'll create a simple research assistant that searches the web, summarizes findings, and writes a final answer. This graph has multiple nodes and an LLM call, giving us interesting things to stream.
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.config import get_stream_writer
import operator
class ResearchState(TypedDict):
query: str
search_results: list[str]
summary: str
answer: str
llm = ChatOpenAI(model="gpt-4o", streaming=True)
def search_node(state: ResearchState) -> dict:
writer = get_stream_writer()
writer({"step": "search", "status": "started", "query": state["query"]})
# Simulated search — replace with Tavily, SerpAPI, etc.
results = [
f"Result 1 for: {state['query']}",
f"Result 2 for: {state['query']}",
f"Result 3 for: {state['query']}",
]
writer({"step": "search", "status": "completed", "count": len(results)})
return {"search_results": results}
def summarize_node(state: ResearchState) -> dict:
writer = get_stream_writer()
writer({"step": "summarize", "status": "started"})
combined = "\n".join(state["search_results"])
response = llm.invoke([
SystemMessage(content="Summarize the following search results concisely."),
HumanMessage(content=combined),
])
writer({"step": "summarize", "status": "completed"})
return {"summary": response.content}
def answer_node(state: ResearchState) -> dict:
response = llm.invoke([
SystemMessage(content="Answer the user's query using the summary. Be thorough."),
HumanMessage(content=f"Query: {state['query']}\nSummary: {state['summary']}"),
])
return {"answer": response.content}
graph_builder = StateGraph(ResearchState)
graph_builder.add_node("search", search_node)
graph_builder.add_node("summarize", summarize_node)
graph_builder.add_node("answer", answer_node)
graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "summarize")
graph_builder.add_edge("summarize", "answer")
graph_builder.add_edge("answer", END)
graph = graph_builder.compile()
Notice two things in this setup. First, we pass streaming=True to the ChatOpenAI constructor. This ensures the model emits tokens as they arrive rather than buffering. Second, we use get_stream_writer() inside nodes to emit custom events. This function retrieves a writer from LangGraph's runtime context, and anything passed to it becomes a stream event that subscribers can consume.
The Streaming API Explained
LangGraph's compiled graphs expose a stream() method that returns an iterator of events. The method accepts a stream_mode parameter that controls what gets emitted. You can pass a single mode string or a list of modes to subscribe to multiple event types at once.
The available stream modes are:
"values"— emits the full state after each node execution"updates"— emits only the delta (the return value of each node)"messages"— emits LLM tokens along with metadata about which message and model produced them"custom"— emits events fromget_stream_writer()calls"debug"— emits detailed execution events including task scheduling and state diffs
Here's a basic example using a single mode:
for event in graph.stream(
{"query": "What are the latest developments in fusion energy?"},
stream_mode="updates"
):
print(event)
This prints a dictionary for each node execution, keyed by node name, containing the state update that node returned. For multi-mode streaming, the structure changes — each event becomes a tuple of (mode, payload):
for mode, payload in graph.stream(
{"query": "What are the latest developments in fusion energy?"},
stream_mode=["updates", "messages", "custom"]
):
if mode == "updates":
print(f"[UPDATE] {payload}")
elif mode == "messages":
chunk, metadata = payload
print(chunk.content, end="", flush=True)
elif mode == "custom":
print(f"[CUSTOM] {payload}")
The "messages" mode deserves special attention. Its payload is a tuple of (message_chunk, metadata) where message_chunk is a AIMessageChunk containing one or more tokens and metadata includes langgraph_node (which node produced this token) and langgraph_step (the step number). This metadata is essential for routing tokens to the right place in your UI when multiple LLM calls happen across different nodes.
Integrating with FastAPI and Server-Sent Events
In a web application, you need to bridge LangGraph's synchronous iterator to an HTTP streaming response. Server-Sent Events (SSE) are the standard choice for this because they work over plain HTTP, are automatically reconnected by browsers, and have a simple text-based format. FastAPI supports SSE cleanly through its StreamingResponse class.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import json
app = FastAPI()
class QueryRequest(BaseModel):
query: str
def format_sse(data: dict, event: str = "message") -> str:
"""Format a dictionary as a Server-Sent Event string."""
return f"event: {event}\ndata: {json.dumps(data)}\n\n"
async def event_stream(query: str):
"""Async generator that yields SSE-formatted events from the graph."""
for mode, payload in graph.stream(
{"query": query},
stream_mode=["updates", "messages", "custom"],
):
if mode == "messages":
chunk, metadata = payload
if chunk.content:
yield format_sse({
"type": "token",
"content": chunk.content,
"node": metadata.get("langgraph_node"),
}, event="token")
elif mode == "custom":
yield format_sse({
"type": "step",
"data": payload,
}, event="step")
elif mode == "updates":
yield format_sse({
"type": "update",
"data": payload,
}, event="update")
yield format_sse({"type": "done"}, event="done")
@app.post("/chat")
async def chat(request: QueryRequest):
return StreamingResponse(
event_stream(request.query),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # Disable Nginx buffering
},
)
The X-Accel-Buffering: no header is critical if you're behind Nginx. By default, Nginx buffers responses, which defeats the purpose of streaming. This header tells Nginx to pass data through immediately. If you're using a CDN like Cloudflare, check its streaming settings too — some configurations buffer at the edge.
Consuming the Stream on the Frontend
The browser side uses the EventSource API for SSE, but since EventSource only supports GET requests, you'll typically use the fetch API with a ReadableStream for POST-based streaming. Here's a client-side implementation:
async function streamChat(query) {
const response = await fetch("/chat", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ query }),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
// SSE events are separated by double newlines
const events = buffer.split("\n\n");
buffer = events.pop(); // Keep incomplete event in buffer
for (const eventStr of events) {
const lines = eventStr.split("\n");
const eventType = lines
.find((l) => l.startsWith("event: "))
?.replace("event: ", "");
const dataStr = lines
.find((l) => l.startsWith("data: "))
?.replace("data: ", "");
if (!dataStr) continue;
const data = JSON.parse(dataStr);
switch (eventType) {
case "token":
appendToUI(data.content);
break;
case "step":
updateStatusIndicator(data.data);
break;
case "update":
updateStateDisplay(data.data);
break;
case "done":
finishResponse();
break;
}
}
}
}
The key detail here is buffering. Network chunks don't respect SSE event boundaries — a single read() might return half an event or two and a half events. The code above maintains a buffer and splits on double newlines, keeping any trailing incomplete data for the next iteration. This is the most common source of bugs in SSE client implementations.
Handling Errors During Streaming
Error handling in streaming systems is tricky because you've already sent a 200 OK response and potentially emitted tokens before the error occurs. You can't change the HTTP status code mid-stream. The standard approach is to emit an error event and close the connection gracefully.
async def event_stream(query: str):
try:
for mode, payload in graph.stream(
{"query": query},
stream_mode=["updates", "messages", "custom"],
):
if mode == "messages":
chunk, metadata = payload
if chunk.content:
yield format_sse({
"type": "token",
"content": chunk.content,
}, event="token")
elif mode == "custom":
yield format_sse({"type": "step", "data": payload}, event="step")
yield format_sse({"type": "done"}, event="done")
except Exception as e:
yield format_sse({
"type": "error",
"message": str(e),
"error_class": type(e).__name__,
}, event="error")
On the client side, handle the error event by displaying an appropriate message. If you've already shown partial tokens, you might append an error indicator rather than replacing the partial output, so the user can see what was generated before the failure.
For production, you should also consider retry logic. If a specific node fails — say, a web search API times out — you might want to retry that node rather than failing the entire stream. LangGraph supports this through checkpointing and time travel, but for simpler cases, wrapping individual node logic in retry decorators works well:
import asyncio
from functools import wraps
def with_retry(max_attempts=3, delay=1.0):
def decorator(func):
@wraps(func)
def wrapper(state):
for attempt in range(max_attempts):
try:
return func(state)
except Exception as e:
if attempt == max_attempts - 1:
raise
writer = get_stream_writer()
writer({
"step": func.__name__,
"status": "retrying",
"attempt": attempt + 1,
"error": str(e),
})
import time
time.sleep(delay * (attempt + 1))
return wrapper
return decorator
@with_retry(max_attempts=3, delay=2.0)
def search_node(state: ResearchState) -> dict:
# Your search logic here
...
Working with Checkpointing and Streaming
In production, you'll often want to persist graph state so conversations can be resumed and so you can recover from crashes. LangGraph's checkpointing system integrates with streaming. When you compile a graph with a checkpointer, each node update is persisted before the next node runs, and streamed events still flow normally.
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool
# For development:
checkpointer = InMemorySaver()
# For production:
pool = ConnectionPool(
conninfo="postgresql://user:pass@localhost:5432/langgraph",
max_size=20,
)
checkpointer = PostgresSaver(pool)
checkpointer.setup() # Creates tables if they don't exist
graph = graph_builder.compile(checkpointer=checkpointer)
When invoking a graph with a checkpointer, you pass a thread_id in the configuration. This ID scopes the conversation — all state updates for the same thread ID are stored together and can be resumed:
config = {"configurable": {"thread_id": "user-123-session-456"}}
for mode, payload in graph.stream(
{"query": "Tell me about fusion energy"},
config=config,
stream_mode=["updates", "messages"],
):
# Handle events as before
pass
# Later, the user asks a follow-up. State is automatically loaded.
for mode, payload in graph.stream(
{"query": "How does that compare to fission?"},
config=config,
stream_mode=["updates", "messages"],
):
# The graph has access to previous state
pass
This pattern is essential for chat applications. Each conversation gets a unique thread ID, and the checkpointer handles persistence transparently while streaming continues to work exactly the same way.
Streaming with Human-in-the-Loop Interruptions
LangGraph supports interrupting graph execution at specific nodes to wait for human input. This is powerful for approval workflows — for example, pausing before a node that sends an email or makes a purchase. Streaming works alongside interrupts, but you need to handle the interrupted state correctly.
from langgraph.types import interrupt, Command
def answer_node(state: ResearchState) -> dict:
# Generate a draft answer
response = llm.invoke([
SystemMessage(content="Draft an answer to the query."),
HumanMessage(content=state["query"]),
])
# Interrupt and wait for human approval
approval = interrupt({
"draft": response.content,
"message": "Please review and approve this answer before sending.",
})
if approval.get("approved"):
return {"answer": response.content}
else:
return {"answer": approval.get("feedback", "Answer rejected by reviewer.")}
graph = graph_builder.compile(
checkpointer=checkpointer,
interrupt_before=["answer"],
)
config = {"configurable": {"thread_id": "session-1"}}
# First stream — runs until the interrupt
for event in graph.stream(
{"query": "Draft a response to our customer"},
config=config,
stream_mode="updates",
):
print(event)
# The graph is now paused. Get the current state:
state = graph.get_state(config)
print(f"Paused at: {state.next}")
# Resume with human input
for event in graph.stream(
Command(resume={"approved": True}),
config=config,
stream_mode="updates",
):
print(event)
From the frontend, you'd stream events until you receive a custom event indicating an interrupt, then display the approval UI. When the user approves or rejects, you send a resume request to the backend, which calls graph.stream() with the Command(resume=...) input and continues streaming.
Performance and Backpressure
Streaming introduces a backpressure concern that doesn't exist with buffered responses. If your LLM generates tokens faster than the client can consume them — perhaps the client is on a slow connection or the browser is busy — those tokens accumulate in memory. In Python, the asyncio event loop handles this reasonably well because yield in an async generator naturally applies backpressure: the producer waits when the consumer isn't ready.
However, if you're doing CPU-intensive work between receiving a token and yielding it (for example, running it through a content filter), you can create a bottleneck. Keep per-token processing lightweight. If you need heavy processing, batch it:
async def event_stream(query: str):
token_buffer = []
flush_interval = 0.05 # 50ms
async def flush_buffer():
if token_buffer:
combined = "".join(token_buffer)
token_buffer.clear()
yield format_sse({"type": "token", "content": combined}, event="token")
last_flush = asyncio.get_event_loop().time()
for mode, payload in graph.stream(
{"query": query},
stream_mode=["messages"],
):
if mode == "messages":
chunk, _ = payload
if chunk.content:
token_buffer.append(chunk.content)
now = asyncio.get_event_loop().time()
if now - last_flush >= flush_interval:
last_flush = now
if token_buffer:
yield format_sse({
"type": "token",
"content": "".join(token_buffer),
}, event="token")
token_buffer.clear()
# Flush remaining tokens
if token_buffer:
yield format_sse({
"type": "token",
"content": "".join(token_buffer),
}, event="token")
yield format_sse({"type": "done"}, event="done")
This batching approach reduces the number of SSE events sent over the wire while still maintaining a responsive feel. A 50ms flush interval is imperceptible to users but can significantly reduce overhead on high-token-throughput responses.
Best Practices for Production Streaming
- Always set
streaming=Trueon your LLM — without it, the"messages"stream mode will only emit the complete message at the end of each LLM call, defeating the purpose of token streaming. - Use multiple stream modes together — combining
"messages"for tokens and"custom"for progress events gives you both real-time output and structural visibility into what the agent is doing. - Include metadata in custom events — emit node names, step numbers, timestamps, and status indicators so the frontend can render a meaningful progress UI without guessing.
- Handle SSE parsing correctly on the client — always buffer partial events and split on double newlines. Never assume a single
read()returns a complete event. - Disable proxy buffering — set
X-Accel-Buffering: nofor Nginx, configure Cloudflare for streaming, and test through your full proxy stack, not just localhost. - Use persistent checkpointers in production —
InMemorySaverloses all state on restart. UsePostgresSaverorRedisSaverfor durability. - Implement graceful error events — since you can't change the HTTP status code mid-stream, emit structured error events and teach your client to handle them.
- Set connection timeouts generously — configure your ASGI server (Uvicorn, Gunicorn) and reverse proxy with timeouts that account for long-running agent tasks. A 300-second timeout is reasonable for complex workflows.
- Monitor stream completion rates — track how many streams complete versus how many are cancelled or error out. A high cancellation rate might indicate your responses are too slow or going in the wrong direction.
- Test with slow clients — use Chrome DevTools network throttling to simulate slow connections and verify your backpressure handling works correctly.
Deployment Architecture
For production deployment, a typical architecture looks like this: your FastAPI application runs behind Uvicorn workers managed by Gunicorn, which sits behind Nginx. Nginx handles TLS termination and forwards to the ASGI server with streaming-compatible settings. The LangGraph checkpointer points to a managed PostgreSQL instance. For multi-instance deployments, all FastAPI workers share the same PostgreSQL checkpointer, so a stream can be resumed on a different worker if needed.
# gunicorn.conf.py
bind = "0.0.0.0:8000"
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
timeout = 300 # 5 minutes for long agent runs
keepalive = 10
graceful_timeout = 30 # Allow in-flight streams to finish on shutdown
max_requests = 1000 # Recycle workers to prevent memory leaks
max_requests_jitter = 100
# nginx.conf (relevant section)
location /chat {
proxy_pass http://127.0.0.1:8000;
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 300s;
chunked_transfer_encoding on;
}
The proxy_buffering off directive is the Nginx equivalent of the X-Accel-Buffering: no header — it ensures data passes through immediately rather than being buffered. Setting both is belt-and-suspenders.
Conclusion
Streaming transforms AI applications from batch-processing tools into interactive experiences. LangGraph's streaming API gives you fine-grained control over what events you emit — from individual tokens to node-level state updates to custom progress events — and its integration with checkpointing, interrupts, and human-in-the-loop patterns means you can build sophisticated agent workflows without sacrificing real-time feedback. The key to success in production is treating the stream as a first-class concern: handle partial events on the client, disable buffering at every proxy layer, emit structured error events, and use persistent checkpointers so state survives restarts. With these patterns in place, your LangGraph applications will feel fast, transparent, and reliable — exactly what users expect from production AI systems.