LangGraph streaming output is the difference between an agent that feels responsive and one that hangs for ten seconds before dumping a wall of text. This guide shows how to build a LangGraph agent that emits tokens incrementally and how to consume that stream from an async Python client and a browser. We’ll use LangGraph’s astream with stream_mode="messages", which is the only mode that surfaces raw LLM token chunks from the underlying model provider.
Step 1: Install dependencies and configure the model endpoint
Install the packages you need. LangGraph ships separately from LangChain core, so pin versions to avoid surprise breakages.
pip install langgraph==0.2.20 langchain-openai==0.1.20 fastapi==0.111.0 uvicorn==0.30.0 httpx==0.27.0Point the chat model at an OpenAI-compatible endpoint. If you want automatic provider fallback when a model is rate-limited, route through a gateway like n4n.ai: its single OpenAI-compatible endpoint covers 240+ models and forwards cache-control hints without changing your LangGraph code.
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0,
streaming=True,
base_url="https://api.n4n.ai/v1", # OpenAI-compatible
api_key="YOUR_KEY",
)Set streaming=True. Without it, LangGraph streaming output will never yield token chunks—the SDK buffers the full completion and returns it as one message.
Step 2: Define the agent state and graph
Use MessagesState for a chat-style agent. It is a typed dict with a messages key and a built-in reducer that appends new messages. A single node that calls the LLM is enough to demonstrate token streaming; you can add tool nodes later.
from langgraph.graph import StateGraph, MessagesState, START, END
def call_model(state: MessagesState):
response = llm.invoke(state["messages"])
return {"messages": [response]}
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)
graph = builder.compile()The node itself is written synchronously, but LangGraph’s runtime wraps the LLM call in an async streaming context when you use astream. Do not block inside the node with heavy CPU work or you will stall the event loop.
Step 3: Stream tokens with astream and stream_mode=“messages”
LangGraph exposes three stream modes: values, updates, and messages. values yields the entire state after each step. updates yields only the delta from the last step. Neither emits partial tokens. Only messages taps the LLM’s native token stream.
async def stream_agent_response(user_input: str):
initial_state = {"messages": [("user", user_input)]}
async for chunk, metadata in graph.astream(
initial_state,
stream_mode="messages"
):
# chunk is a BaseMessageChunk (typically AIMessageChunk)
if isinstance(chunk.content, str) and chunk.content:
yield chunk.content
elif isinstance(chunk.content, list):
# multimodal chunks arrive as lists of dicts
for part in chunk.content:
if isinstance(part, dict) and part.get("type") == "text":
yield part["text"]The metadata dict contains langgraph_node and langgraph_step. Ignore it for a minimal demo. The generator yields strings as they arrive from the provider. That generator is the core of your LangGraph streaming output pipeline.
Step 4: Wrap the generator in a FastAPI endpoint
Expose the stream over HTTP so frontend clients can consume it. Use StreamingResponse with media_type="text/plain" for a raw token feed. If you target Server-Sent Events, use text/event-stream and format each yield as data: {token}\n\n.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.post("/agent/stream")
async def agent_stream(payload: dict):
user_input = payload.get("input", "")
return StreamingResponse(
stream_agent_response(user_input),
media_type="text/plain"
)Run the server with uvicorn main:app --port 8000 --reload. The endpoint returns a chunked transfer-encoded body that grows token by token. If you put a proxy in front, disable response buffering or the tokens will be coalesced.
Step 5: Consume the stream from an async Python client
Use httpx.AsyncClient with stream=True. Read the response with aiter_text() and print with flush=True so fragments appear immediately.
import httpx
async def main():
async with httpx.AsyncClient() as client:
async with client.stream(
"POST",
"http://localhost:8000/agent/stream",
json={"input": "Explain vector databases in three sentences."}
) as response:
async for text in response.aiter_text():
print(text, end="", flush=True)
if __name__ == "__main__":
import asyncio
asyncio.run(main())If words appear one fragment at a time, your LangGraph streaming output is working end to end. Add a try/except around the stream to handle disconnected clients gracefully.
Step 6: Consume from the browser with fetch
For a web UI, use the native fetch streaming reader. The code below works in any modern browser without libraries.
const response = await fetch("/agent/stream", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ input: "Summarize the Rust ownership model." })
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const token = decoder.decode(value, { stream: true });
document.getElementById("output").textContent += token;
}Attach an AbortController if you need to cancel mid-stream when the user clicks stop. The browser will not buffer unless you explicitly await the full body.
Step 7: Verify success with curl and timestamps
Do not trust your eyes alone. Measure inter-token latency to confirm the stream is incremental.
curl -N -X POST http://localhost:8000/agent/stream \
-H "Content-Type: application/json" \
-d '{"input":"Write a haiku about TCP."}' \
| while read -r line; do echo "$(date +%s.%N) $line"; doneYou should see multiple lines with timestamps separated by milliseconds, not one block after a two-second pause. That is the definitive proof of LangGraph streaming output. If you see a single delayed dump, check streaming=True and your proxy buffer settings.
Step 8: Handle tool calls and multi-node graphs
Real agents loop between an LLM node and a tool node. Token streaming still works, but only the LLM invocation streams; tool results arrive as a single updates chunk after execution.
def call_model(state):
return {"messages": [llm.invoke(state["messages"])]}
def call_tools(state):
# execute tools, return ToolMessage list
...With stream_mode="messages", you will see tokens during call_model, then a non-token message when call_tools runs. Design your UI to render both: append text for token chunks, render tool status separately.
Step 9: Avoid common pitfalls
- Forgot
streaming=Trueon the LLM. LangGraph cannot invent tokens the provider withholds. - Used
stream_mode="values". You get the full state after each node, not tokens. - Ran
graph.stream(sync) inside an async server. Useastream; mixing blocking I/O kills throughput. - Buffered in middleware. Nginx and some WSGI servers buffer by default. Send
X-Accel-Buffering: noor use a raw ASGI server. - Yielded non-string chunks. Always coerce
chunk.contentto string before yielding, or the response encoder will raise.
Step 10: Production hardening
Add request timeouts, max token limits, and authentication. If you use n4n.ai as the model gateway, per-token usage metering is automatic and client routing directives are honored, so you can shift models per request without code changes. That is the entire LangGraph streaming output setup—from graph compilation to a verified token stream in a real client.