Streaming - LangGraph

Overview

LangGraph supports multiple streaming modes to provide real-time feedback, build responsive UIs, and monitor graph execution. Instead of waiting for the entire graph to complete, you can process results as they become available.

Stream Modes

LangGraph offers 7 different streaming modes, each serving different use cases:

from langgraph.graph import StateGraph

graph = builder.compile()

# Single mode
for event in graph.stream(input_data, stream_mode="values"):
    print(event)

# Multiple modes simultaneously
for event in graph.stream(input_data, stream_mode=["values", "updates"]):
    print(event)

Mode Overview

values

Complete state after each step

updates

Individual node outputs

messages

LLM token streaming

custom

User-defined events

checkpoints

State snapshots

tasks

Task execution events

debug

Debugging information

Values Mode

Emits the complete state after each step:

from typing_extensions import TypedDict

class State(TypedDict):
    messages: list[str]
    count: int

for state in graph.stream({"messages": [], "count": 0}, stream_mode="values"):
    print(state)
    # {'messages': ['Hello'], 'count': 1}
    # {'messages': ['Hello', 'Response'], 'count': 2}
    # {'messages': ['Hello', 'Response', 'Done'], 'count': 3}

Use cases:

Display complete state in UI
Monitor full state changes
Simple progress tracking

values is the default stream mode. It includes the initial state before any nodes execute.

Updates Mode

Emits individual node outputs as they complete:

for event in graph.stream({"count": 0}, stream_mode="updates"):
    print(event)
    # {'node1': {'count': 1}}
    # {'node2': {'count': 2}}
    # {'node3': {'count': 3}}

Each event is a dictionary with:

Key: Node name
Value: Node’s output (state update)

Use cases:

Track which nodes executed
Show per-node progress
Collect individual results

Parallel Node Updates

# If nodes run in parallel, you get separate events
for event in graph.stream(input_data, stream_mode="updates"):
    print(event)
    # {'fetch_user': {'user': {...}}}
    # {'fetch_orders': {'orders': [...]}}
    # {'combine': {'result': {...}}}

Messages Mode

Stream LLM tokens in real-time:

from langgraph.graph import MessagesState

class State(MessagesState):
    pass

def llm_node(state: State):
    # LLM calls are automatically streamed
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

for event in graph.stream(
    {"messages": [("user", "Tell me a story")]},
    stream_mode="messages"
):
    # Event format: (message_chunk, metadata)
    chunk, metadata = event
    print(chunk.content, end="", flush=True)
    
    # Metadata includes:
    # - langgraph_node: which node emitted this
    # - langgraph_step: execution step
    # - langgraph_path: node path in nested graphs

Message Metadata

for event in graph.stream(input_data, stream_mode="messages"):
    chunk, metadata = event
    
    print(f"Node: {metadata['langgraph_node']}")
    print(f"Step: {metadata['langgraph_step']}")
    print(f"Content: {chunk.content}")
    
    # For tool calls
    if hasattr(chunk, "tool_calls"):
        print(f"Tool calls: {chunk.tool_calls}")

Use cases:

Real-time chat interfaces
Streaming chatbots
Progressive text generation

Custom Mode

Emit custom events from within nodes:

from langgraph.types import StreamWriter

def my_node(state: State, writer: StreamWriter) -> dict:
    """Node that emits custom events."""
    
    # Emit progress updates
    writer("Starting processing...")
    
    for i, item in enumerate(state["items"]):
        result = process_item(item)
        
        # Emit custom event
        writer({
            "type": "progress",
            "item_index": i,
            "result": result,
            "percentage": (i + 1) / len(state["items"]) * 100
        })
    
    writer("Processing complete!")
    
    return {"status": "done"}

# Stream custom events
for event in graph.stream(input_data, stream_mode="custom"):
    print(event)
    # "Starting processing..."
    # {'type': 'progress', 'item_index': 0, ...}
    # {'type': 'progress', 'item_index': 1, ...}
    # "Processing complete!"

StreamWriter is automatically injected when requested as a parameter. It’s a no-op when not using stream_mode="custom".

Use cases:

Fine-grained progress tracking
Custom metrics/telemetry
Application-specific events

Checkpoints Mode

Emits state snapshots when checkpoints are created:

for event in graph.stream(input_data, stream_mode="checkpoints"):
    print(event)
    # StateSnapshot(
    #     values={'messages': [...], 'count': 1},
    #     next=('node2',),
    #     config={'configurable': {'thread_id': '1'}},
    #     metadata={'step': 0, 'source': 'loop'},
    #     created_at='2024-01-15T10:30:00Z',
    #     tasks=(...),
    # )

Each checkpoint includes:

values: Current state
next: Upcoming nodes
config: Runtime configuration
metadata: Step info, source
created_at: Timestamp
tasks: Pending tasks

Use cases:

Monitor checkpointing
Display execution timeline
Debug state persistence

Requires a checkpointer to be configured.

Tasks Mode

Emits events for task lifecycle:

for event in graph.stream(input_data, stream_mode="tasks"):
    print(event)
    # ('task', {
    #     'id': 'abc-123',
    #     'name': 'process_data',
    #     'path': ('process_data',),
    #     'result': None,  # None while running
    #     'error': None,
    # })
    # 
    # ('task_result', {
    #     'id': 'abc-123',
    #     'name': 'process_data',
    #     'path': ('process_data',),
    #     'result': {'count': 1},
    #     'error': None,
    # })

Event types:

task: Task started
task_result: Task completed (with result or error)

Use cases:

Monitor task execution
Track task duration
Debug failures

Debug Mode

Combines checkpoints and tasks for comprehensive debugging:

for event in graph.stream(input_data, stream_mode="debug"):
    event_type, event_data = event
    
    if event_type == "checkpoint":
        print(f"Checkpoint at step {event_data.metadata['step']}")
    elif event_type == "task":
        print(f"Task {event_data['name']} started")
    elif event_type == "task_result":
        print(f"Task {event_data['name']} completed")

Use cases:

Development debugging
Troubleshooting execution
Performance analysis

Multiple Stream Modes

Combine modes for richer output:

for event in graph.stream(
    input_data,
    stream_mode=["values", "updates", "messages"]
):
    event_type = list(event.keys())[0]
    event_data = event[event_type]
    
    if event_type == "values":
        print(f"State: {event_data}")
    elif event_type == "updates":
        print(f"Update: {event_data}")
    elif event_type == "messages":
        chunk, metadata = event_data
        print(f"Token: {chunk.content}")

Each event is a dict with a single key indicating the mode:

{"values": <state>}
{"updates": <update>}
{"messages": (<chunk>, <metadata>)}

Async Streaming

All streaming modes support async iteration:

async for event in graph.astream(
    input_data,
    stream_mode="updates"
):
    await process_event(event)

Use async streaming for:

Async I/O operations
Concurrent event processing
WebSocket connections
Server-sent events (SSE)

Streaming with Subgraphs

Control subgraph streaming:

# Stream events from subgraphs too
for event in graph.stream(
    input_data,
    stream_mode="updates",
    subgraphs=True  # Include subgraph events
):
    print(event)
    # Events include namespace information
    # {'parent_node:subgraph_node': {'count': 1}}

Building a Streaming UI

Real-Time Chat Interface

from langgraph.graph import MessagesState

class ChatState(MessagesState):
    pass

def display_stream(user_message: str):
    config = {"configurable": {"thread_id": "chat-1"}}
    
    # Stream LLM tokens
    print("Assistant: ", end="")
    for event in graph.stream(
        {"messages": [("user", user_message)]},
        config=config,
        stream_mode="messages"
    ):
        chunk, metadata = event
        print(chunk.content, end="", flush=True)
    print()  # Newline

display_stream("What's the weather?")
# Assistant: I don't have access to real-time weather...

Progress Bar with Custom Events

from tqdm import tqdm
from langgraph.types import StreamWriter

def batch_processor(
    state: State,
    writer: StreamWriter
) -> dict:
    items = state["items"]
    
    for i, item in enumerate(items):
        result = process(item)
        writer({
            "processed": i + 1,
            "total": len(items)
        })
    
    return {"status": "complete"}

# Display progress
with tqdm(total=100) as pbar:
    for event in graph.stream(input_data, stream_mode="custom"):
        if isinstance(event, dict) and "processed" in event:
            pbar.update(1)

FastAPI SSE Endpoint

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

app = FastAPI()

@app.post("/stream")
async def stream_graph(request: dict):
    async def event_generator():
        async for event in graph.astream(
            request["input"],
            stream_mode="updates"
        ):
            yield f"data: {json.dumps(event)}\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

Stream Configuration

Early Emission

Force eager event emission:

graph = builder.compile(
    stream_eager=True  # Emit events as soon as possible
)

By default, events are batched. stream_eager=True reduces latency.

Filtering Stream Channels

Limit which state keys are streamed:

from langgraph.pregel import Pregel

# Using low-level Pregel API
app = Pregel(
    nodes={...},
    channels={...},
    output_channels=["messages", "status"],
    stream_channels=["messages"],  # Only stream 'messages'
)

Best Practices

Choosing Stream Modes

Use values for state monitoring and simple UIs
Use updates to track individual node execution
Use messages for chat interfaces with LLMs
Use custom for application-specific events
Use debug during development
Combine modes when you need multiple perspectives

Performance

Use async streaming for I/O-bound applications
Enable stream_eager for lower latency
Limit state size to reduce serialization overhead
Filter stream channels to reduce bandwidth
Batch custom events when possible

UI Integration

Buffer tokens before displaying (avoid flickering)
Show loading indicators between node executions
Handle reconnection for long-running streams
Display node names from updates mode
Use custom events for progress bars

Troubleshooting

No events emitted

Verify correct stream mode
Check if graph has any nodes
Ensure nodes return state updates
For messages mode, confirm LLM is used
For custom mode, verify StreamWriter calls

Delayed events

Enable stream_eager=True
Check for buffering in transport layer
Verify async streaming is used correctly
Review node execution time

Missing tokens in messages mode

Ensure LLM supports streaming
Check that LLM is configured for streaming
Verify message format is correct
Review LangChain callback configuration

Next Steps

Human-in-the-Loop

Combine streaming with interrupts for human oversight

Checkpointing

Use checkpoint streaming for state monitoring

Get Started

Core Concepts

Guides

Tutorials

CLI

​Overview

​Stream Modes

​Mode Overview

values

updates

messages

custom

checkpoints

tasks

debug

​Values Mode

​Updates Mode

​Parallel Node Updates

​Messages Mode

​Message Metadata

​Custom Mode

​Checkpoints Mode

​Tasks Mode

​Debug Mode

​Multiple Stream Modes

​Async Streaming

​Streaming with Subgraphs

​Building a Streaming UI

​Real-Time Chat Interface

​Progress Bar with Custom Events

​FastAPI SSE Endpoint

​Stream Configuration

​Early Emission

​Filtering Stream Channels

​Best Practices

​Troubleshooting

​Next Steps

Human-in-the-Loop

Checkpointing

Build docs developers (and LLMs) love

Overview

Stream Modes

Mode Overview

Values Mode

Updates Mode

Parallel Node Updates

Messages Mode

Message Metadata

Custom Mode

Checkpoints Mode

Tasks Mode

Debug Mode

Multiple Stream Modes

Async Streaming

Streaming with Subgraphs

Building a Streaming UI

Real-Time Chat Interface

Progress Bar with Custom Events

FastAPI SSE Endpoint

Stream Configuration

Early Emission

Filtering Stream Channels

Best Practices

Troubleshooting

Next Steps