streaming-execution-in-langgraph

Streaming Execution in LangGraph

Purpose

Streaming execution enables real-time monitoring of LangGraph workflows by yielding incremental state updates as nodes execute, rather than waiting for complete workflow termination.

When to Use

• Monitor long-running workflows in real-time
• Provide progressive feedback to users
• Debug workflow execution step-by-step
• Build responsive UI applications
• Stream partial results as they're available
• Display LLM tokens as they're generated

Core Pattern

# Instead of app.invoke() which blocks until complete
for event in app.stream(initial_state):
    # Process each state update as it occurs
    print(event)

Event Structure

Each event is a dictionary mapping node names to their state updates:

for event in app.stream(initial_state):
    for node_name, state_update in event.items():
        print(f"Node '{node_name}' updated: {state_update}")

Basic Example

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class State(TypedDict):
    messages: list
    step: str

def greet_func(state: State) -> dict:
    return {"messages": state["messages"] + ["Hello!"], "step": "process"}

def process_func(state: State) -> dict:
    return {"messages": state["messages"] + ["Processed."], "step": "done"}

workflow = StateGraph(State)
workflow.add_node("greet", greet_func)
workflow.add_node("process", process_func)
workflow.add_edge(START, "greet")
workflow.add_edge("greet", "process")
workflow.add_edge("process", END)

app = workflow.compile()

# Stream execution
for event in app.stream({"messages": [], "step": ""}):
    print(f"Event: {event}")

Output:

Event: {'greet': {'messages': ['Hello!'], 'step': 'process'}}
Event: {'process': {'messages': ['Hello!', 'Processed.'], 'step': 'done'}}

Note: This output format is specific to stream_mode="updates" (the default). Use stream_mode="values" to receive complete state snapshots instead.