Streaming

You send a message. A spinner turns for thirty seconds. Has it crashed? Should you refresh? Agents are slow, so people need to see what's going on. Showing progress makes the very same wait feel much shorter.

• A spinner for thirty seconds
• Has it crashed?
• Progress makes waits feel shorter

Streaming updates

Instead of invoke, we call stream, and it hands back an update as each node finishes. Each of these nodes takes a moment. Run it. The updates appear one by one, as they finish. In your interface, those become messages such as, checking the order, then, reading the policy. It's a small change, and it transforms how an agent feels.

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

class State(TypedDict):
    steps: int

def slow(name):
    def node(state):
        time.sleep(0.2)
        return {"steps": state["steps"] + 1}
    return node

g = StateGraph(State)
g.add_node("read_order", slow("a"))
g.add_node("check_policy", slow("b"))
g.add_node("draft_reply", slow("c"))
g.add_edge(START, "read_order")
g.add_edge("read_order", "check_policy")
g.add_edge("check_policy", "draft_reply")
g.add_edge("draft_reply", END)
app = g.compile()

for update in app.stream({"steps": 0}, stream_mode="updates"):
    print("Finished:", list(update)[0], update[list(update)[0]])

Finished: read_order {'steps': 1}
Finished: check_policy {'steps': 2}
Finished: draft_reply {'steps': 3}

Stream modes

You can stream different things. Updates shows what each node changed. Values shows the whole state after each step. Messages streams the model's words, token by token, for that typing effect. And custom sends your own progress messages from inside a tool. Most products combine updates and messages.

Summary