Streaming responses in LangGraph reduces perceived latency by delivering tokens to users as they generate rather than waiting for the full response. This guide walks through implementing streaming end to end — from graph configuration to frontend integration — with runnable code at each step.
Step 1: Understand LangGraph streaming modes
LangGraph exposes four streaming modes through the stream() method. Each serves a different purpose:
- messages: Streams individual messages as they complete. Best for chat interfaces where you want to show assistant messages incrementally.
- values: Streams the full state after each node executes. Useful for debugging or when downstream nodes depend on complete intermediate results.
- updates: Streams only the state changes (deltas) from each node. Lower bandwidth than values, good for complex state.
- custom: Lets you define exactly what to stream via a custom stream writer. Maximum control for specialized UX.
For most chat applications, messages mode with stream_mode="messages" gives the best perceived latency because tokens appear as the model generates them.
# Basic streaming invocation
from langgraph.graph import StateGraph, MessagesState
from langgraph.checkpoint.memory import MemorySaver
graph = StateGraph(MessagesState)
# ... add nodes and edges ...
compiled = graph.compile(checkpointer=MemorySaver())
# Stream messages as they arrive
for chunk in compiled.stream(
{"messages": [("user", "Hello")]},
config={"configurable": {"thread_id": "1"}},
stream_mode="messages"
):
print(chunk)Step 2: Configure your graph for token-level streaming
Token-level streaming requires the underlying model to support streaming and your nodes to yield tokens rather than returning complete responses. Most LangChain chat model wrappers support this natively when you call astream() or stream().
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from langchain_core.messages import AIMessageChunk
llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
async def call_model(state: MessagesState):
"""Node that streams tokens from the LLM."""
response = await llm.ainvoke(state["messages"])
return {"messages": [response]}
# For true token streaming, use astream in a generator node
async def stream_model(state: MessagesState):
"""Node that yields tokens one at a time."""
async for chunk in llm.astream(state["messages"]):
# Each chunk is an AIMessageChunk with content delta
yield {"messages": [chunk]}
graph = StateGraph(MessagesState)
graph.add_node("model", stream_model)
graph.set_entry_point("model")
graph.set_finish_point("model")
compiled = graph.compile(checkpointer=MemorySaver())The key difference: ainvoke() returns a complete AIMessage, while astream() yields AIMessageChunk objects with incremental content deltas. LangGraph’s messages stream mode automatically reassembles these chunks for you.
Step 3: Implement the streaming endpoint
Build a FastAPI endpoint that streams Server-Sent Events (SSE) to the client. This pattern works with any frontend that consumes SSE or can be adapted for WebSockets.
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import json
import asyncio
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
@app.post("/stream")
async def stream_endpoint(request: Request):
body = await request.json()
user_input = body.get("message", "")
thread_id = body.get("thread_id", "default")
async def event_generator():
config = {"configurable": {"thread_id": thread_id}}
input_data = {"messages": [("user", user_input)]}
try:
async for chunk in compiled.astream(
input_data,
config=config,
stream_mode="messages"
):
# chunk is a tuple: (message_chunk, metadata)
message_chunk, metadata = chunk
if isinstance(message_chunk, AIMessageChunk):
content = message_chunk.content
if content: # Filter empty chunks
event_data = {
"type": "token",
"content": content,
"thread_id": thread_id
}
yield f"data: {json.dumps(event_data)}\n\n"
except Exception as e:
error_data = {"type": "error", "message": str(e)}
yield f"data: {json.dumps(error_data)}\n\n"
# Signal completion
yield f"data: {json.dumps({'type': 'done', 'thread_id': thread_id})}\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # Disable nginx buffering
}
)Step 4: Handle tool calls in streaming mode
Tool calls complicate streaming because the model must emit a complete tool call before execution, then stream the tool result, then stream the final response. LangGraph handles this automatically when you use ToolNode and stream in messages mode.
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode
@tool
def get_weather(location: str) -> str:
"""Get current weather for a location."""
# Simulated API call
return f"Weather in {location}: 72°F, sunny"
tools = [get_weather]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)
async def call_model_with_tools(state: MessagesState):
response = await llm_with_tools.ainvoke(state["messages"])
return {"messages": [response]}
def should_continue(state: MessagesState):
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return "__end__"
graph = StateGraph(MessagesState)
graph.add_node("model", call_model_with_tools)
graph.add_node("tools", tool_node)
graph.set_entry_point("model")
graph.add_conditional_edges("model", should_continue)
graph.add_edge("tools", "model")
compiled = graph.compile(checkpointer=MemorySaver())When streaming this graph, you’ll see three phases in the output:
- Assistant message chunks containing the tool call (empty content,
tool_callspopulated) - Tool result message (from
ToolNode) - Final assistant message chunks with the natural language response
The frontend should handle each phase appropriately — showing a “thinking” indicator during tool execution, then streaming the final answer.
Step 5: Build a React frontend consumer
A minimal React hook that consumes the SSE stream and updates state incrementally:
// useStreamingChat.ts
import { useState, useCallback, useRef } from "react";
interface Message {
role: "user" | "assistant";
content: string;
}
interface StreamEvent {
type: "token" | "done" | "error";
content?: string;
thread_id: string;
message?: string;
}
export function useStreamingChat(threadId: string) {
const [messages, setMessages] = useState<Message[]>([]);
const [isStreaming, setIsStreaming] = useState(false);
const abortControllerRef = useRef<AbortController | null>(null);
const assistantMessageRef = useRef<string>("");
const sendMessage = useCallback(async (content: string) => {
if (isStreaming) return;
setIsStreaming(true);
setMessages(prev => [...prev, { role: "user", content }]);
assistantMessageRef.current = "";
// Add placeholder assistant message
const assistantIndex = messages.length;
setMessages(prev => [...prev, { role: "assistant", content: "" }]);
abortControllerRef.current = new AbortController();
try {
const response = await fetch("/stream", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ message: content, thread_id: threadId }),
signal: abortControllerRef.current.signal,
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
const reader = response.body?.getReader();
const decoder = new TextDecoder();
if (!reader) throw new Error("No response body");
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value);
const lines = chunk.split("\n\n");
for (const line of lines) {
if (!line.startsWith("data: ")) continue;
try {
const event: StreamEvent = JSON.parse(line.slice(6));
switch (event.type) {
case "token":
assistantMessageRef.current += event.content || "";
setMessages(prev => {
const next = [...prev];
next[assistantIndex] = {
...next[assistantIndex],
content: assistantMessageRef.current
};
return next;
});
break;
case "done":
setIsStreaming(false);
break;
case "error":
console.error("Stream error:", event.message);
setIsStreaming(false);
break;
}
} catch (e) {
console.warn("Failed to parse SSE event:", line);
}
}
}
} catch (e) {
if (e instanceof Error && e.name !== "AbortError") {
console.error("Stream failed:", e);
}
setIsStreaming(false);
}
}, [threadId, isStreaming, messages.length]);
const stopStreaming = useCallback(() => {
abortControllerRef.current?.abort();
setIsStreaming(false);
}, []);
return { messages, isStreaming, sendMessage, stopStreaming };
}Usage in a component:
// ChatInterface.tsx
import { useStreamingChat } from "./useStreamingChat";
export function ChatInterface() {
const { messages, isStreaming, sendMessage, stopStreaming } =
useStreamingChat("thread-123");
const [input, setInput] = useState("");
const handleSubmit = (e: React.FormEvent) => {
e.preventDefault();
if (input.trim()) sendMessage(input.trim());
setInput("");
};
return (
<div>
<div className="messages">
{messages.map((msg, i) => (
<div key={i} className={msg.role}>
<strong>{msg.role}:</strong> {msg.content}
</div>
))}
</div>
<form onSubmit={handleSubmit}>
<input
value={input}
onChange={e => setInput(e.target.value)}
disabled={isStreaming}
placeholder={isStreaming ? "Streaming..." : "Type a message..."}
/>
<button type="submit" disabled={isStreaming || !input.trim()}>
Send
</button>
{isStreaming && (
<button type="button" onClick={stopStreaming}>Stop</button>
)}
</form>
</div>
);
}Step 6: Add checkpointing for conversation persistence
Streaming works with LangGraph’s checkpointing to maintain conversation history across requests. The MemorySaver checkpointer stores state in memory — swap it for PostgresSaver or SqliteSaver in production.
# For production: persistent checkpointer
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.sqlite import SqliteSaver
import os
# SQLite for development
sqlite_checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
# PostgreSQL for production
# postgres_checkpointer = PostgresSaver.from_conn_string(
# os.environ["DATABASE_URL"]
# )
compiled = graph.compile(checkpointer=sqlite_checkpointer)With checkpointing enabled, the thread_id in your streaming config maintains conversation context automatically. Each new request with the same thread_id continues from the last checkpoint.
Step 7: Optimize for first-token latency
Perceived latency depends heavily on time-to-first-token (TTFT). Several techniques reduce TTFT in LangGraph:
Warm the model connection at startup:
# app/lifespan.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# Warm up the LLM connection on startup
await llm.ainvoke("warmup")
yield
app = FastAPI(lifespan=lifespan)Use a smaller model for routing or classification nodes, reserve the large model for final generation:
from langchain_openai import ChatOpenAI
router_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
generator_llm = ChatOpenAI(model="gpt-4o", temperature=0.7, streaming=True)
async def route(state: MessagesState):
# Fast classification with small model
result = await router_llm.ainvoke([
("system", "Classify intent: chat, tool_use, or handoff"),
*state["messages"]
])
return {"intent": result.content}
async def generate(state: MessagesState):
# High-quality generation with streaming
async for chunk in generator_llm.astream(state["messages"]):
yield {"messages": [chunk]}Enable provider-level caching where supported. Some providers (including those accessible via n4n.ai) honor cache-control headers for prompt prefixes, reducing repeat computation on shared context.
Step 8: Handle backpressure and client disconnects
Long streams can accumulate memory if the client disconnects or reads slowly. Implement proper cleanup:
async def event_generator():
config = {"configurable": {"thread_id": thread_id}}
input_data = {"messages": [("user", user_input)]}
stream = compiled.astream(input_data, config=config, stream_mode="messages")
try:
async for chunk in stream:
message_chunk, metadata = chunk
if isinstance(message_chunk, AIMessageChunk) and message_chunk.content:
yield f"data: {json.dumps({'type': 'token', 'content': message_chunk.content})}\n\n"
# Check if client still connected (FastAPI specific)
if await request.is_disconnected():
break
except asyncio.CancelledError:
# Client disconnected - clean up
pass
finally:
# Ensure stream is closed
await stream.aclose()Step 9: Verify the implementation
Test each layer independently before integrating:
1. Unit test the graph streaming:
# test_streaming.py
import pytest
from langgraph.graph import MessagesState
@pytest.mark.asyncio
async def test_graph_streams_tokens():
input_data = {"messages": [("user", "Say hello")]}
config = {"configurable": {"thread_id": "test-1"}}
tokens = []
async for chunk in compiled.astream(input_data, config=config, stream_mode="messages"):
message_chunk, _ = chunk
if hasattr(message_chunk, 'content') and message_chunk.content:
tokens.append(message_chunk.content)
assert len(tokens) > 1 # Multiple chunks received
full_response = "".join(tokens)
assert "hello" in full_response.lower()2. Integration test the SSE endpoint:
# Test with curl
curl -N -X POST http://localhost:8000/stream \
-H "Content-Type: application/json" \
-d '{"message": "Count to 5", "thread_id": "test-2"}'Expected output shows incremental tokens:
data: {"type": "token", "content": "One", "thread_id": "test-2"}
data: {"type": "token", "content": ", ", "thread_id": "test-2"}
data: {"type": "token", "content": "two", "thread_id": "test-2"}
data: {"type": "done", "thread_id": "test-2"}3. Load test for concurrent streams:
# Install hey: go install github.com/rakyll/hey@latest
hey -n 100 -c 10 -m POST \
-H "Content-Type: application/json" \
-d '{"message": "Hello", "thread_id": "load-test"}' \
http://localhost:8000/streamMonitor memory usage and TTFT percentiles. A healthy streaming endpoint maintains steady memory and sub-500ms TTFT under load.
4. Frontend verification checklist:
- Tokens appear incrementally in the UI (not all at once)
- Tool calls show a loading state, then results, then final answer
- Stop button cancels the stream and aborts the request
- Conversation history persists across page reloads (same
thread_id) - Error events display gracefully without breaking the chat
- Multiple concurrent tabs with different
thread_ids work independently
Step 10: Monitor streaming performance in production
Add structured logging to track streaming metrics:
import time
import structlog
logger = structlog.get_logger()
async def event_generator():
config = {"configurable": {"thread_id": thread_id}}
input_data = {"messages": [("user", user_input)]}
start_time = time.perf_counter()
first_token_time = None
token_count = 0
try:
async for chunk in compiled.astream(input_data, config=config, stream_mode="messages"):
message_chunk, metadata = chunk
if isinstance(message_chunk, AIMessageChunk) and message_chunk.content:
if first_token_time is None:
first_token_time = time.perf_counter()
ttft_ms = (first_token_time - start_time) * 1000
logger.info("stream_first_token",
thread_id=thread_id,
ttft_ms=ttft_ms)
token_count += 1
yield f"data: {json.dumps({'type': 'token', 'content': message_chunk.content})}\n\n"
total_time = (time.perf_counter() - start_time) * 1000
logger.info("stream_complete",
thread_id=thread_id,
total_ms=total_time,
token_count=token_count,
tokens_per_second=token_count / (total_time / 1000) if total_time > 0 else 0)
except Exception as e:
logger.error("stream_error", thread_id=thread_id, error=str(e))
raiseTrack these metrics in your observability stack:
- TTFT (Time to First Token): Target < 500ms for good UX
- Tokens per second: Sustained throughput indicator
- Stream duration: End-to-end latency
- Error rate: Failed streams per 1k requests
- Client disconnect rate: Abandoned streams (may indicate TTFT issues)
LangGraph streaming responses reduce latency perceptually by delivering incremental updates rather than blocking on full generation. The messages stream mode with astream() gives you token-level granularity with minimal code changes. Pair it with SSE on the backend and a reactive frontend, and users see responses appear word-by-word — the standard for modern LLM applications.