Streaming in LangGraph

• If we’re building a responsive app for the users, real-time updates are key to keeping them engaged.
• Common use-cases are:
  • Workflow progress (eg. get state updates after each graph node is executed).
  • LLM tokens as they’re generated.
  • Custom updates (e.g., “Fetched 10/100 records”)

In LangGraph

• .stream and .astream are sync and async methods for streaming back outputs from a graph run.

• There are several different modes you can specify when calling these methods (eg., ‘graph.stream(input, stream_mode=“values”)):

• Most common modes are:

  • stream_mode = “values”: This streams the full value of the state after each step of the graph.
  • stream_mode = “updates”: This streams the updates to the state after each step of the graph.

• In prod apps, we usually want to stream more than the state.

• In particular, with LLM calls it is common to stream the tokens as they are generated.

• We can do this using the .astream_events method, which streams back events as they happen inside nodes.

• Each event is a dict with a few keys:

  • event: This is the type of event that is being emitted.
  • name: This is the name of the event.
  • data: This is the data associated with the event.
  • metadata: Which contains langgraph_node, the node emitting the event.

Stream Modes in Detail

1. “values” Mode

• Returns the complete state after each node execution
• Useful for tracking overall progress and state changes
• Example output: Full state object with all accumulated data

2. “updates” Mode

• Returns only the changes/updates made by each node
• More efficient for large states as it only sends deltas
• Example output: Only the fields that were modified by the current node

3. “debug” Mode

• Provides detailed debugging information
• Includes node execution details, timing, and intermediate results
• Useful during development and troubleshooting

Event Streaming with .astream_events

Event Types

• on_chat_model_start: When an LLM call begins
• on_chat_model_stream: Individual tokens as they’re generated
• on_chat_model_end: When an LLM call completes
• on_tool_start: When a tool execution begins
• on_tool_end: When a tool execution completes
• on_chain_start: When a chain/node starts
• on_chain_end: When a chain/node ends

Event Structure

{
    "event": "on_chat_model_stream",
    "name": "ChatOpenAI",
    "data": {"chunk": "Hello"},
    "metadata": {
        "langgraph_node": "generate_response",
        "langgraph_step": 3
    }
}

Practical Example: Streaming LLM Tokens

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, List
import asyncio
 
class State(TypedDict):
    messages: List[HumanMessage | AIMessage]
 
def chat_node(state: State):
    llm = ChatOpenAI(model="gpt-4o", streaming=True)
    response = llm.invoke(state["messages"])
    return {"messages": [response]}
 
# Build graph
workflow = StateGraph(State)
workflow.add_node("chat", chat_node)
workflow.set_entry_point("chat")
workflow.add_edge("chat", END)
graph = workflow.compile()
 
# Stream tokens as they're generated
async def stream_tokens():
    async for event in graph.astream_events(
        {"messages": [HumanMessage(content="Tell me a joke")]},
        version="v1"
    ):
        if event["event"] == "on_chat_model_stream":
            token = event["data"]["chunk"].content
            if token:
                print(token, end="", flush=True)
 
# Run the streaming example
asyncio.run(stream_tokens())

Streaming State Updates

# Stream complete state after each node
for chunk in graph.stream(
    {"messages": [HumanMessage(content="Hello")]},
    stream_mode="values"
):
    print(f"State after node: {chunk}")
 
# Stream only updates from each node
for chunk in graph.stream(
    {"messages": [HumanMessage(content="Hello")]},
    stream_mode="updates"
):
    print(f"Updates from node: {chunk}")

Custom Progress Updates

from langgraph.graph import StateGraph, END
from typing import TypedDict
import time
 
class ProcessingState(TypedDict):
    data: List[str]
    processed_count: int
    total_count: int
 
def processing_node(state: ProcessingState):
    # Simulate processing with custom updates
    total = state["total_count"]
    processed = 0
    
    for item in state["data"]:
        # Process item
        time.sleep(0.1)  # Simulate work
        processed += 1
        
        # Emit custom progress event
        yield {
            "processed_count": processed,
            "progress_message": f"Processed {processed}/{total} items"
        }
    
    return {"processed_count": processed}
 
# Use with streaming to get real-time progress
for update in graph.stream(initial_state, stream_mode="updates"):
    if "progress_message" in update:
        print(update["progress_message"])

Best Practices for Streaming

1. Choose the Right Stream Mode

• Use "values" for complete state tracking
• Use "updates" for efficient delta updates
• Use "debug" during development

2. Handle Token Streaming Efficiently

• Buffer tokens for better UX (avoid character-by-character display)
• Implement proper error handling for streaming failures
• Use async/await for non-blocking operations

3. Implement Progress Indicators

• Show progress bars for long-running operations
• Provide meaningful status messages
• Allow users to cancel long operations

4. Error Handling in Streams

async def safe_streaming():
    try:
        async for event in graph.astream_events(input_data, version="v1"):
            # Process event
            handle_event(event)
    except Exception as e:
        print(f"Streaming error: {e}")
        # Implement fallback behavior

Performance Considerations

Buffering Strategies

• Character-level: Real-time but can be overwhelming
• Word-level: Good balance of responsiveness and readability
• Sentence-level: Smoother reading experience
• Chunk-based: Efficient for large outputs

Network Considerations

• Implement connection retry logic
• Handle network interruptions gracefully
• Consider compression for large state objects
• Use WebSockets for real-time web applications

Memory Management

• Clean up completed streams
• Implement backpressure for slow consumers
• Monitor memory usage with large state objects

Integration with Web Applications

FastAPI Example

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json
 
app = FastAPI()
 
@app.post("/chat/stream")
async def stream_chat(message: str):
    async def generate():
        async for event in graph.astream_events(
            {"messages": [HumanMessage(content=message)]},
            version="v1"
        ):
            if event["event"] == "on_chat_model_stream":
                token = event["data"]["chunk"].content
                if token:
                    yield f"data: {json.dumps({'token': token})}\n\n"
    
    return StreamingResponse(generate(), media_type="text/plain")

WebSocket Example

from fastapi import WebSocket
import asyncio
 
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await websocket.accept()
    
    while True:
        message = await websocket.receive_text()
        
        async for event in graph.astream_events(
            {"messages": [HumanMessage(content=message)]},
            version="v1"
        ):
            if event["event"] == "on_chat_model_stream":
                token = event["data"]["chunk"].content
                if token:
                    await websocket.send_text(json.dumps({
                        "type": "token",
                        "content": token
                    }))

Debugging Streaming Issues

Common Problems

1. Missing tokens: Check if streaming is enabled on the LLM
2. Memory leaks: Ensure proper cleanup of stream objects
3. Performance issues: Monitor event processing time
4. Connection drops: Implement reconnection logic

Debugging Tools

# Enable detailed logging
import logging
logging.basicConfig(level=logging.DEBUG)
 
# Monitor streaming performance
import time
 
async def debug_streaming():
    start_time = time.time()
    token_count = 0
    
    async for event in graph.astream_events(input_data, version="v1"):
        if event["event"] == "on_chat_model_stream":
            token_count += 1
            elapsed = time.time() - start_time
            print(f"Token {token_count} at {elapsed:.2f}s")

Advanced Streaming Patterns

Conditional Streaming

async def conditional_streaming():
    async for event in graph.astream_events(input_data, version="v1"):
        # Only stream certain types of events
        if event["event"] in ["on_chat_model_stream", "on_tool_end"]:
            process_event(event)

Multi-Stream Coordination

import asyncio
 
async def coordinate_multiple_streams():
    # Run multiple streaming operations concurrently
    tasks = [
        graph1.astream_events(input1, version="v1"),
        graph2.astream_events(input2, version="v1")
    ]
    
    async for completed_task in asyncio.as_completed(tasks):
        async for event in completed_task:
            handle_multi_stream_event(event)

Stream Filtering and Transformation

class StreamProcessor:
    def __init__(self, graph):
        self.graph = graph
        self.filters = []
        self.transformers = []
    
    def add_filter(self, filter_func):
        self.filters.append(filter_func)
    
    def add_transformer(self, transform_func):
        self.transformers.append(transform_func)
    
    async def process_stream(self, input_data):
        async for event in self.graph.astream_events(input_data, version="v1"):
            # Apply filters
            if all(f(event) for f in self.filters):
                # Apply transformations
                for transformer in self.transformers:
                    event = transformer(event)
                yield event
 
# Usage
processor = StreamProcessor(graph)
processor.add_filter(lambda e: e["event"] == "on_chat_model_stream")
processor.add_transformer(lambda e: {**e, "timestamp": time.time()})
 
async for processed_event in processor.process_stream(input_data):
    print(processed_event)