LangGraph Module 7: Streaming Output and Debugging

Step 1: Stream state updates

Instead of invoke, call stream with a stream_mode. The updates mode yields one item per step, shaped as a dictionary from node name to that node's update. The values mode yields the full state after each step.

inputs = {"messages": [{"role": "user", "content": "What is 6 * 7?"}]}

for chunk in agent.stream(inputs, stream_mode="updates"):
    print(chunk)   # {'llm_call': {...}} then {'tool_node': {...}} ...

for state in agent.stream(inputs, stream_mode="values"):
    print(len(state["messages"]), "messages so far")

Step 2: Stream model tokens

The messages mode yields tuples of a message chunk and metadata, as the model generates. The metadata includes the node name, so you can show tokens only from the node you care about.

for chunk, metadata in agent.stream(inputs, stream_mode="messages"):
    if metadata["langgraph_node"] == "llm_call":
        print(chunk.text, end="", flush=True)

Step 3: Emit custom progress events

Inside a node, get a stream writer and send any data you like. It appears when you stream with the custom mode, which is handy for long-running steps that should report progress.

from typing_extensions import TypedDict
from langgraph.config import get_stream_writer
from langgraph.graph import StateGraph, START, END

class JobState(TypedDict):
    result: str

def slow_node(state: JobState) -> dict:
    writer = get_stream_writer()
    writer({"progress": "Fetching data..."})
    # ... do the real work here ...
    writer({"progress": "Done"})
    return {"result": "finished"}

builder = StateGraph(JobState)
builder.add_node("slow_node", slow_node)
builder.add_edge(START, "slow_node")
builder.add_edge("slow_node", END)
job_graph = builder.compile()

for event in job_graph.stream({"result": ""}, stream_mode="custom"):
    print(event)   # {'progress': 'Fetching data...'} then {'progress': 'Done'}

Step 4: Combine modes

Pass a list of modes to receive several kinds of output. Each item is then a tuple of the mode name and its data. The documentation also describes newer event-streaming APIs in recent releases, which it recommends for new applications, so check it after you are comfortable with these modes. On Python versions older than 3.11, async code needs to pass the configuration explicitly.

for mode, data in agent.stream(inputs, stream_mode=["updates", "messages"]):
    print(mode)

Step 5: Debug with traces and state inspection

For deeper debugging, enable LangSmith tracing with the LANGSMITH_TRACING and LANGSMITH_API_KEY environment variables, which records every node, model call, and tool call. With a checkpointer, you can also read the saved state and history after a failure to see exactly how a run got where it did.