Chapter 14: Streaming and Real-Time Updates
As you build Multi-Agent Systems, your execution times will inevitably increase. A simple conversational loop might take 2 seconds. A multi-agent research, writing, and review workflow might take 45 seconds or more.
If your backend application uses app.invoke(), the user will sit staring at a frozen loading spinner for 45 seconds. In modern UX, this is entirely unacceptable. Users expect to see the AI "typing" in real-time, and they expect to see transparent progress bars detailing exactly what the agents are currently doing.
In this chapter, we will master Streaming. We will learn how to stream state updates as nodes finish, how to stream raw LLM tokens as they are generated, how to dispatch custom events from deep inside tools, and how to architect responsive UIs.
14.1 The Architecture of a Streaming Backend
Before writing code, we must understand the architecture.
When you use .invoke(), your FastAPI or Express backend receives an HTTP Request, waits for the LangGraph cycle to finish, and returns an HTTP Response.
When you use .stream(), you cannot use a standard HTTP Request/Response cycle. You must use technologies designed for persistent streams:
- Server-Sent Events (SSE): A unidirectional HTTP connection where the server pushes chunks of text to the client. Perfect for LLM streaming.
- WebSockets: A bidirectional persistent connection. Excellent for complex, highly interactive agent platforms.
LangGraph operates as a Python Generator. It yields data. Your backend framework (like FastAPI) takes those yielded chunks and forwards them over SSE or WebSockets to the React/Next.js frontend.
14.2 .invoke() vs .stream()
Until now, we have exclusively used .invoke(). This is a blocking, synchronous call that waits for the entire graph to hit the END node before returning the final state dictionary.
LangGraph provides .stream() (and its asynchronous sibling .astream()).
.stream() yields chunks of data as the graph executes.
Streaming State Updates (Node Boundaries)
By default, when you call app.stream(), LangGraph will yield the State updates generated by each individual node the exact moment that node finishes executing.
# Instead of invoke, we iterate over the stream generator
for output in app.stream({"messages": [HumanMessage(content="Search for Apple stock")]}):
# The output is a dictionary where the KEY is the name of the node that just finished,
# and the VALUE is the state update dictionary that node returned.
for node_name, state_update in output.items():
print(f"\n--- Node '{node_name}' Finished! ---")
# If the node appended a message, print it!
if "messages" in state_update:
latest_msg = state_update["messages"][-1]
print(f"Message: {latest_msg.content}")
Execution Trace:
--- Node 'agent_node' Finished! ---
Message: (Tool Call Request for search_web)
--- Node 'tool_node' Finished! ---
Message: Apple stock is currently $185.
--- Node 'agent_node' Finished! ---
Message: Apple stock is trading at $185 today.
This is incredibly powerful for Progress Reporting UIs. If your web app receives these chunks, it can show the user an expanding log:
- Supervisor evaluating...
- Research Agent searching the web...
- Writer Agent generating final answer...
14.3 Streaming LLM Tokens (The Typing Effect)
Streaming state updates is great for progress bars, but it still waits for the LLM node to completely finish before yielding the AIMessage.
If you want the "typing effect" (where words appear character-by-character on the user's screen), you must stream the raw LLM tokens.
LangGraph supports a special streaming mode called "messages".
# Stream mode 'messages' intercepts the raw tokens from the LLM!
for chunk, metadata in app.stream(
{"messages": [HumanMessage(content="Write a poem about LangGraph")]},
stream_mode="messages"
):
# chunk is an AIMessageChunk object containing just a few letters/words
# print with end="" to append to the same line!
print(chunk.content, end="", flush=True)
Execution Trace:
LangGraph is great,
It manages state,
With edges and nodes,
It carries the loads...
(The words appear instantly as the LLM generates them!)
14.4 The V2 Async Events API (astream_events)
What if you want BOTH? What if you want to stream the node boundaries (to update the UI progress bar) AND stream the LLM tokens (for the typing effect)?
Using stream_mode="messages" only gives you messages. Using default stream() only gives you node boundaries.
To get absolute, granular control over everything happening inside your graph, you use the advanced astream_events() API.
import asyncio
async def stream_everything():
# We must pass a version flag (v2 is the current standard)
events = app.astream_events(
{"messages": [HumanMessage(content="Hello")]},
version="v2"
)
async for event in events:
event_type = event["event"]
# Catch LLM tokens
if event_type == "on_chat_model_stream":
content = event["data"]["chunk"].content
if content:
print(content, end="", flush=True)
# Catch Tool starts
elif event_type == "on_tool_start":
print(f"\n[UI Alert] Starting tool: {event['name']}...")
# Catch Node finishes
elif event_type == "on_chain_end" and event.get("name") == "supervisor_node":
print("\n[UI Alert] Supervisor finished evaluating.")
# Run the async function
asyncio.run(stream_everything())
astream_events emits a massive firehose of data. Every LLM call, every tool invocation, every subgraph transition generates an event. You simply filter for the event_type you care about and forward it to your frontend.
14.5 Yielding Custom Progress from Tools
In advanced enterprise apps, you might have a Tool that takes 30 seconds to run. For example, a process_pdf tool that chunks 100 pages.
If the user sits for 30 seconds with a UI saying "Starting tool: process_pdf", they will think the app crashed.
You can dispatch custom progress events from deep inside your standard Python functions, and catch them in the astream_events firehose!
from langchain_core.callbacks.manager import adispatch_custom_event
@tool
async def process_pdf(file_path: str):
"""Processes a massive PDF document."""
for i in range(1, 101): # 100 pages
await asyncio.sleep(0.1) # Simulate heavy parsing
# Dispatch a real-time event!
await adispatch_custom_event(
"pdf_progress",
{"page": i, "total": 100}
)
return "PDF processing complete."
In your backend streaming loop:
# Catch the custom event!
elif event_type == "on_custom_event" and event["name"] == "pdf_progress":
page = event["data"]["page"]
total = event["data"]["total"]
print(f"[UI Progress Bar] Parsing PDF: {page}/{total}")
This level of transparency is what separates amateur AI scripts from enterprise platforms.
14.6 Handling Interruptions and Disconnects
When building streaming backends, you must account for network realities. Users will close their laptops. Cell connections will drop.
If a user closes their browser while a 45-second LangGraph execution is streaming over Server-Sent Events, the connection breaks.
By default, Python will raise a CancelledError and the graph execution will violently abort. If you are using Checkpointing (Chapter 10), this means the checkpoint might save in a corrupted, half-finished state.
When building your FastAPI or Express handlers, you must catch disconnects gracefully. If the user disconnects, you can choose to either safely abort the graph, or push the job to a background Celery worker (Chapter 21) so the agent finishes the work regardless of the user's browser state.
14.7 Summary
By replacing .invoke() with .stream() and astream_events(), you completely transform the user experience of your LangGraph application.
You can provide radical transparency into the Agent's reasoning process (streaming node updates and custom tool events) and provide instant gratification (streaming LLM tokens).
In the next chapter, we will return to the concept of State, and explore how to build Long-Term Memory that persists across completely different threads, days, and years.