TopolearnUsers expect to see progress and tokens live; agents can take many seconds per run.
graph.stream(input, stream_mode=...) yields output while the graph runs instead of waiting for the end. The mode picks what you get:
"updates": after each node, what it changed, as{node_name: update}. Good for progress logs."values": the full state after each step."messages": LLM output token by token, as(message_chunk, metadata)pairs, from any LangChain model called inside the graph. Metadata includes which node produced it, so you can show only the final answer node."custom": anything you emit yourself. Inside a node,writer = get_stream_writer()thenwriter({"progress": 50})."checkpoints","tasks","debug": detailed runtime events (most need a checkpointer).
Pass a list to get several modes at once. With the default (v1) format, a list gives (mode, data) tuples. LangGraph 1.1 added version="v2": every chunk is then a dict with type, ns (which subgraph it came from) and data, the same shape whatever options you use. Prefer v2 in new code.
astream is the async version for web servers.
LangGraph 1.2 also introduced event streaming (graph.stream_events(..., version="v3"), beta), which returns a run object with typed views such as run.messages and run.output. The docs now recommend it for application code; it is built on the same modes, so what you learn here carries over.
Example
from langgraph.config import get_stream_writer
from langgraph.graph import StateGraph, START, END, MessagesState
def research(state: MessagesState) -> dict:
writer = get_stream_writer()
for pct in (25, 50, 100):
writer({"progress": pct}) # -> "custom" mode
return {"messages": [{"role": "assistant", "content": "Found 3 sources"}]}
b = StateGraph(MessagesState)
b.add_node(research)
b.add_edge(START, "research")
b.add_edge("research", END)
graph = b.compile()
for part in graph.stream({"messages": [{"role": "user", "content": "bees"}]},
stream_mode=["custom", "updates"], version="v2"):
if part["type"] == "custom":
print("progress", part["data"]["progress"])
else:
print("update from", list(part["data"]))
# progress 25 / progress 50 / progress 100 / update from ['research']
# Tokens: for msg, meta in graph.stream(inp, stream_mode="messages"): print(msg.content, end="")Remember
- updates = per-node changes; values = full state; messages = LLM tokens
- custom mode carries your own events via get_stream_writer()
- version='v2' gives uniform {type, ns, data} chunks
- stream_events(version='v3') is the newer (beta) event API