LangGraph Streaming: Returning Intermediate Steps to Users

Why your LangGraph agent feels slow even when it isn't

You built an agent. It calls a search tool, reasons over the results, maybe calls a second tool, and then writes a final answer. Locally, in a notebook, it takes four seconds end to end and you don't think twice about it. Then you wire it into a product, put it behind a chat UI, and four seconds turns into a wall of silence followed by a wall of text. Users refresh the page. Users assume it's broken. Users leave.

The problem isn't latency. Four seconds is fine for a multi-step agent doing real work. The problem is that the user has no idea anything is happening. A single invoke() call is a black box — the graph runs node by node, tools fire, the LLM reasons, and none of it reaches the frontend until the whole run finishes and returns one final state.

LangGraph is built around a state machine that already knows exactly when each node starts, finishes, and what it produced. That means the information you need to build a good loading experience already exists inside the graph — you just aren't asking for it. This is what .stream() and its more granular sibling .astream_events() are for: they let you tap into the graph's execution as it happens, node by node, token by token, and forward that to the user instead of making them wait for a single opaque response.

This article walks through the practical mechanics of streaming intermediate steps from a LangGraph agent — what "intermediate steps" actually means in graph terms, the different streaming modes, how to combine them with tool-calling agents, and how to wire the output into a frontend without rebuilding your whole app around websockets.

What "intermediate steps" means in a graph

In a linear chain, intermediate steps are vague — there's really just "before" and "after." In LangGraph, they're precise, because your agent is a graph of nodes and edges, and the graph runtime executes one node at a time (or a batch of parallel nodes at a super-step). Every node transition is an event you can subscribe to.

Concretely, for a typical tool-calling agent built with create_react_agent or a hand-rolled StateGraph, the intermediate steps are:

• The agent node deciding whether to call a tool or answer directly
• The tool node executing a tool call and returning its result
• The agent node running again with the tool result appended to state
• This loop repeating until the model returns a final answer without further tool calls

Each of those is a distinct state update. If you stream at the node level, you get one message per step: "agent decided to call search_docs", "tool returned 3 results", "agent is now writing the final answer." That's already enough to build a genuinely useful "thinking" indicator — far better than a static spinner, and far cheaper to build than a custom event bus.

If you want more granularity than "which node just ran," you go one level down to LLM token streaming, which is where .astream_events() comes in. We'll cover both.

The three streaming modes in `.stream()`

LangGraph's .stream() method (and its async counterpart .astream()) accepts a stream_mode argument that controls what gets yielded at each step. The three you'll use constantly are values, updates, and messages.

from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# Assume `graph` is a compiled StateGraph with agent + tool nodes
config = {"configurable": {"thread_id": "session-42"}}

# stream_mode="updates" yields only the delta each node produced
for chunk in graph.stream(
    {"messages": [("user", "What's the weather in Bangalore and should I carry an umbrella?")]},
    config=config,
    stream_mode="updates",
):
    for node_name, node_output in chunk.items():
        print(f"[{node_name}] produced:", node_output)

stream_mode="updates" is the one you want for intermediate-step reporting. Each yielded chunk is a dict keyed by node name, containing only what that node changed in state — not the full accumulated state. That distinction matters at scale: if your state carries a long message history, stream_mode="values" (which returns the *entire* state after every step) gets expensive fast, while updates stays small and cheap regardless of how long the conversation gets.

stream_mode="values" is still useful when you want the full picture after each step — for example, if you're logging complete state snapshots for debugging or building a "replay" feature. But for a live UI, prefer updates.

# stream_mode="values" yields the full state snapshot after each step
for full_state in graph.stream(
    {"messages": [("user", "Summarize the Q3 report and flag any risks.")]},
    config=config,
    stream_mode="values",
):
    last_message = full_state["messages"][-1]
    print(f"Current state has {len(full_state['messages'])} messages")
    print("Latest:", last_message.content if hasattr(last_message, "content") else last_message)

You can also request multiple modes at once by passing a list, which is handy when you want node-level updates and message-level streaming in the same loop:

for stream_mode, chunk in graph.stream(
    {"messages": [("user", "Check the inventory API and tell me if SKU-4471 is in stock.")]},
    config=config,
    stream_mode=["updates", "messages"],
):
    if stream_mode == "updates":
        print("STEP:", chunk)
    elif stream_mode == "messages":
        message_chunk, metadata = chunk
        print("TOKEN:", message_chunk.content, "from node:", metadata.get("langgraph_node"))

That last mode, "messages", streams individual LLM token chunks tagged with metadata about which node emitted them — which is exactly what you need to show token-by-token generation *and* know whether those tokens came from an intermediate reasoning step or the final answer.

Streaming tool calls as they happen

The most common request from users building agent UIs is: "show me when the agent is calling a tool, and show me what it got back." With stream_mode="updates", this is almost free, because the tool node's output is a discrete chunk.

Here's a fuller example with a real tool-calling ReAct-style graph:

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

@tool
def get_stock_price(ticker: str) -> str:
    """Look up the current stock price for a given ticker symbol."""
    # In production this would call a real market data API
    fake_prices = {"AAPL": 227.50, "MSFT": 421.30, "NVDA": 138.20}
    price = fake_prices.get(ticker.upper())
    if price is None:
        return f"No price data found for {ticker}"
    return f"{ticker.upper()} is trading at ${price}"

@tool
def compare_to_target(current_price: float, target_price: float) -> str:
    """Compare a current price to a target and say whether it's above or below."""
    if current_price >= target_price:
        return f"${current_price} is above the target of ${target_price}"
    return f"${current_price} is below the target of ${target_price}"

llm = ChatAnthropic(model="claude-sonnet-4-5")
agent = create_react_agent(llm, tools=[get_stock_price, compare_to_target])

config = {"configurable": {"thread_id": "stock-session-1"}}

for chunk in agent.stream(
    {"messages": [("user", "Is NVDA trading above $130 right now?")]},
    config=config,
    stream_mode="updates",
):
    for node_name, output in chunk.items():
        if node_name == "tools":
            for msg in output["messages"]:
                print(f"TOOL RESULT: {msg.content}")
        elif node_name == "agent":
            last = output["messages"][-1]
            if getattr(last, "tool_calls", None):
                for call in last.tool_calls:
                    print(f"CALLING TOOL: {call['name']} with args {call['args']}")
            elif last.content:
                print(f"FINAL ANSWER: {last.content}")

Running this against a question like "Is NVDA trading above $130 right now?" produces a clean sequence you can forward straight to a UI: agent decides to call get_stock_price, tool returns the price, agent decides to call compare_to_target, tool returns the comparison, agent writes the final sentence. Each of those four events can become a line in a "thinking" panel that updates live instead of a blank screen followed by one paragraph.

Going deeper with `.astream_events()`

.stream() with stream_mode="updates" gets you node-level visibility, which covers most product needs. But sometimes you want finer-grained events — for example, distinguishing "the LLM started generating," "a specific tool started executing," and "a specific tool finished," each as separate, individually timestamped events, along with run IDs you can correlate across a distributed system. That's what .astream_events() is for.

.astream_events() is async-only and emits a unified event stream across every runnable in your graph — chat models, tools, retrievers, chains — using the same event schema LangChain uses everywhere else (on_chain_start, on_llm_start, on_tool_start, on_tool_end, and so on).

import asyncio

async def run_and_stream_events():
    config = {"configurable": {"thread_id": "events-session-1"}}
    async for event in agent.astream_events(
        {"messages": [("user", "Is NVDA trading above $130? Also check MSFT.")]},
        config=config,
        version="v2",
    ):
        kind = event["event"]

        if kind == "on_tool_start":
            print(f"[TOOL START] {event['name']} args={event['data'].get('input')}")

        elif kind == "on_tool_end":
            print(f"[TOOL END] {event['name']} -> {event['data'].get('output')}")

        elif kind == "on_chat_model_stream":
            token = event["data"]["chunk"].content
            if token:
                print(token, end="", flush=True)

        elif kind == "on_chain_end" and event["name"] == "agent":
            print("\n[AGENT NODE COMPLETE]")

asyncio.run(run_and_stream_events())

The version="v2" argument matters — it locks in the newer, more consistent event schema. Each event carries a run_id, a name, tags, and metadata, which is what makes this API well suited to production tracing: you can filter by tag to show only your top-level agent's tool calls while silently ignoring internal retriever calls, or you can log every event to a trace store like LangSmith and just forward a filtered subset to the user-facing stream.

The tradeoff is verbosity. .astream_events() emits far more events than .stream() — every sub-runnable inside every node fires its own start/end pair. In practice, most teams filter aggressively:

async def stream_for_ui():
    config = {"configurable": {"thread_id": "ui-session-1"}}
    async for event in agent.astream_events(
        {"messages": [("user", "What's the capital of the country with the largest population?")]},
        config=config,
        version="v2",
    ):
        kind = event["event"]
        node = event.get("metadata", {}).get("langgraph_node")

        # Only forward events from nodes we actually want to expose
        if node not in ("agent", "tools"):
            continue

        if kind == "on_tool_start":
            yield {"type": "tool_call", "name": event["name"], "input": event["data"].get("input")}
        elif kind == "on_tool_end":
            yield {"type": "tool_result", "name": event["name"], "output": event["data"].get("output")}
        elif kind == "on_chat_model_stream":
            token = event["data"]["chunk"].content
            if token:
                yield {"type": "token", "content": token}

That node not in ("agent", "tools") filter is doing real work — without it, you'd also see events from the retry logic, the message trimming step, or any internal chains LangChain wraps around the model call. Filtering by langgraph_node in the metadata keeps your user-facing stream focused on things a human actually cares about.

Wiring streamed output into an API response

Streaming inside a Python loop is only half the job — you need to get those chunks to a browser. The most common pattern is Server-Sent Events (SSE) over a FastAPI endpoint, since SSE is simpler than websockets for one-directional server-to-client streaming and every modern browser supports it natively via EventSource.

import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def event_generator(user_message: str, thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    async for event in agent.astream_events(
        {"messages": [("user", user_message)]},
        config=config,
        version="v2",
    ):
        kind = event["event"]
        node = event.get("metadata", {}).get("langgraph_node")
        if node not in ("agent", "tools"):
            continue

        payload = None
        if kind == "on_tool_start":
            payload = {"type": "tool_call", "name": event["name"]}
        elif kind == "on_tool_end":
            payload = {"type": "tool_result", "name": event["name"], "output": str(event["data"].get("output"))}
        elif kind == "on_chat_model_stream":
            token = event["data"]["chunk"].content
            if token:
                payload = {"type": "token", "content": token}

        if payload:
            yield f"data: {json.dumps(payload)}\n\n"

    yield "data: [DONE]\n\n"

@app.post("/chat/stream")
async def chat_stream(user_message: str, thread_id: str):
    return StreamingResponse(
        event_generator(user_message, thread_id),
        media_type="text/event-stream",
    )

On the frontend, an EventSource (or a fetch call reading the response body as a stream, if you need POST support) parses each data: line as JSON and updates the UI incrementally — appending tokens to the current answer, and rendering tool calls as collapsible "step" cards above it. This is the same pattern behind the "thinking" traces you see in most production agent products: it's not a special hidden API, it's .astream_events() filtered down and piped through SSE.

Handling checkpointing and resumability alongside streaming

One detail that trips people up: streaming and persistence are separate concerns, but they interact. If your graph uses a checkpointer (for example MemorySaver or a Postgres-backed checkpointer) so conversations survive across requests, every .stream() call still writes to the checkpoint at each super-step, regardless of streaming mode. That means if a client disconnects mid-stream, the graph's state up to the last completed node is safely persisted — you haven't lost the tool call that already ran, only the delivery of that update to the now-disconnected client.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent = create_react_agent(llm, tools=[get_stock_price, compare_to_target], checkpointer=checkpointer)

config = {"configurable": {"thread_id": "resumable-session-9"}}

# First call streams normally; if the client drops here, state is still checkpointed
for chunk in agent.stream(
    {"messages": [("user", "Check AAPL price and compare it to $200")]},
    config=config,
    stream_mode="updates",
):
    print(chunk)

# A later call with the same thread_id can pick up the conversation,
# and re-streaming from here only processes new steps, not a re-run of old ones
for chunk in agent.stream(
    {"messages": [("user", "Now do the same for MSFT")]},
    config=config,
    stream_mode="updates",
):
    print(chunk)

This is worth calling out explicitly in your API design: reconnect logic on the frontend should re-open a stream against the same thread_id, not restart the whole conversation. Because LangGraph checkpoints state after every super-step, a dropped connection is a delivery problem, not a data-loss problem — treat it that way in your retry code instead of re-sending the entire prior conversation as a new request.

Common mistakes when streaming intermediate steps

A few patterns show up repeatedly when teams first wire this up, and each one quietly breaks the experience it was meant to improve.

• Streaming raw tool output straight to the user. Not every tool result is meant for human eyes — a tool that returns a 40KB JSON blob from an internal API shouldn't be rendered verbatim as a "step." Summarize or truncate tool outputs before you forward them, and reserve full payloads for your own logs.
• Forgetting to filter `.astream_events()` by node. Without a langgraph_node filter, you'll surface internal LangChain plumbing events to the user — retry wrappers, output parsers, trimming steps — none of which mean anything to someone watching a chat window.
• Mixing `stream_mode="values"` into a live UI loop. Because values returns the entire accumulated state every time, a long conversation means every chunk gets bigger, and your frontend re-renders the whole message list on every token. Use updates for incremental UI work and reserve values for logging or debugging.
• Not handling the final answer as its own event type. If you only stream tool calls and tokens, you can lose track of *when* the final answer actually starts versus an intermediate reasoning message. Tag your token events with the node they came from (as shown above) so the frontend can distinguish "still working" tokens from "here's your answer" tokens.
• Blocking on `.invoke()` in a background job and only streaming the wrapper. If your endpoint calls agent.invoke() and then artificially drips out the final string to simulate streaming, you've reintroduced the exact wait time you were trying to eliminate — the graph already finished running before anything reached the client. Stream from the actual .stream() or .astream_events() call, not from a fake typewriter effect on top of a blocking call.

Putting it together

The core idea is simple even though the API surface has a few knobs: LangGraph already knows, step by step, what your agent is doing, because it's a graph and graphs execute node by node. .stream(stream_mode="updates") gives you cheap, node-level visibility that's enough for most "show me what the agent is doing" UI needs. .astream_events() gives you finer-grained, LLM-token-level and tool-level events with enough metadata to build serious tracing and filtering on top. Neither requires you to change how you build the graph itself — you're just choosing how to observe an execution that was always this granular under the hood.

Start with stream_mode="updates" for tool-call visibility, add stream_mode="messages" or .astream_events() when you need token-by-token text, and always filter by node before you show anything to a user. Combine that with a checkpointer so reconnects are cheap, and you've solved the actual problem: not making your agent faster, but making its existing speed feel transparent.

If you want to go from these snippets to a production-grade agent — proper state design, multi-agent graphs, human-in-the-loop interrupts, and deploying a streaming API that holds up under real traffic — that's exactly what we cover, end to end, in the LangGraph Tutorial course on teachyou.ai.