LangGraph Streaming

Real-time updates and progress tracking for LangGraph workflows.

5 Stream Modes

Available modes

for mode, chunk in graph.stream(inputs, stream_mode=["values", "updates", "messages", "custom", "debug"]): print(f"[{mode}] {chunk}")

Mode Purpose Use Case

values Full state after each step Debugging, state inspection

updates State deltas after each step Efficient UI updates

messages LLM tokens + metadata Chat interfaces, typing indicators

custom User-defined events Progress bars, status updates

debug Maximum information Development, troubleshooting

Custom Events with StreamWriter

from langgraph.config import get_stream_writer

def node_with_progress(state: State): """Emit custom progress events.""" writer = get_stream_writer()

for i, item in enumerate(state["items"]):
    writer({
        "type": "progress",
        "current": i + 1,
        "total": len(state["items"]),
        "status": f"Processing {item}"
    })
    result = process(item)

writer({"type": "complete", "message": "All items processed"})
return {"results": results}

Consume custom events

for mode, chunk in graph.stream(inputs, stream_mode=["updates", "custom"]): if mode == "custom": if chunk.get("type") == "progress": print(f"Progress: {chunk['current']}/{chunk['total']}") elif mode == "updates": print(f"State updated: {list(chunk.keys())}")

LLM Token Streaming

Stream tokens from LLM calls

for message_chunk, metadata in graph.stream( {"topic": "AI safety"}, stream_mode="messages" ): if message_chunk.content: print(message_chunk.content, end="", flush=True)

Filter by node

for msg, meta in graph.stream(inputs, stream_mode="messages"): if meta["langgraph_node"] == "writer_agent": print(msg.content, end="")

Filter by tags

model = init_chat_model("claude-sonnet-4-20250514", tags=["main_response"])

for msg, meta in graph.stream(inputs, stream_mode="messages"): if "main_response" in meta.get("tags", []): print(msg.content, end="")

Subgraph Streaming

Enable subgraph visibility

for namespace, chunk in graph.stream( inputs, subgraphs=True, stream_mode="updates" ): # namespace shows graph hierarchy: (), ("child",), ("child", "grandchild") print(f"[{'/'.join(namespace) or 'root'}] {chunk}")

Multiple Modes Simultaneously

Combine modes for comprehensive feedback

async for mode, chunk in graph.astream( inputs, stream_mode=["updates", "custom", "messages"] ): match mode: case "updates": update_ui_state(chunk) case "custom": show_progress(chunk) case "messages": append_to_chat(chunk)

Non-LangChain LLM Streaming

def call_custom_llm(state: State): """Stream from arbitrary LLM APIs.""" writer = get_stream_writer()

for chunk in your_streaming_client.generate(state["prompt"]):
    writer({"type": "llm_token", "content": chunk.text})

return {"response": full_response}

FastAPI SSE Integration

from fastapi import FastAPI from fastapi.responses import StreamingResponse import json

app = FastAPI()

@app.post("/stream") async def stream_workflow(request: WorkflowRequest): async def event_generator(): async for mode, chunk in graph.astream( request.inputs, stream_mode=["updates", "custom"] ): yield f"data: {json.dumps({'mode': mode, 'data': chunk})}\n\n"

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

Python < 3.11 Async

Manual config propagation required

async def call_model(state: State, config: RunnableConfig): response = await model.ainvoke(state["messages"], config) return {"messages": [response]}

Explicit writer injection

async def node_with_custom_stream(state: State, writer: StreamWriter): writer({"status": "processing"}) result = await process_async(state) return {"result": result}

Key Decisions

Decision Recommendation

Mode selection Use ["updates", "custom"] for most UIs

Token streaming Use messages mode with node filtering

Progress tracking Use custom mode with get_stream_writer()

Subgraph visibility Enable subgraphs=True for complex workflows

Common Mistakes

• Forgetting stream_mode parameter (defaults to values only)

• Not handling async properly in Python < 3.11

• Missing flush=True on print for real-time display

• Not filtering messages by node/tags (noisy output)

Evaluations

See references/evaluations.md for test cases.

Related Skills

• langgraph-subgraphs

• Stream updates from nested graphs

• langgraph-human-in-loop

• Stream status while awaiting human

• langgraph-supervisor

• Stream agent progress in supervisor workflows

• langgraph-parallel

• Stream from parallel execution branches

• langgraph-tools

• Stream tool execution progress

• api-design-framework

• SSE endpoint design patterns

Capability Details

stream-modes

Keywords: stream mode, values, updates, messages, custom, debug Solves:

• Configure streaming output format

• Choose appropriate mode for use case

• Combine multiple stream modes

custom-events

Keywords: custom event, progress, status, stream writer, get_stream_writer Solves:

• Emit custom progress events

• Track workflow status

• Implement progress bars

token-streaming

Keywords: token, LLM stream, chat, typing indicator, messages mode Solves:

• Stream LLM tokens in real-time

• Build chat interfaces

• Show typing indicators

subgraph-streaming

Keywords: subgraph, nested, hierarchy, namespace Solves:

• Stream from nested graphs

• Track subgraph progress

• Debug complex workflows