Node-by-node output (stream_mode="updates") lets you observe every internal dynamic of an "Editor Rejection" in real-time.
Welcome back to our LangGraph Multi-Agent Masterclass. I am your instructor.
In previous lessons, we built a pretty solid team for our "AI Content Agency": The Planner breaks down tasks, the Researcher gathers information, the Writer works hard on drafting, and the Editor strictly controls quality.
However, recently some students complained in the group: "Teacher, my multi-agent system runs too much like a black box! Especially when the Editor is unsatisfied with the Writer's article and sends it back for a rewrite, the entire Graph might loop in the background for several minutes. During this time, the frontend page is completely unresponsive. My boss stares at the screen and asks if it crashed, and I can only awkwardly wipe my sweat."
This is a very classic advanced engineering problem: Observability and User Experience (UX).
In traditional single LLM calls, we can use token-level streaming (the typewriter effect) to ease user anxiety. But in a multi-agent workflow (Graph), the real "big moves" happen during the transitions between nodes. Today, we are going to tear open this black box and use LangGraph's powerful stream_mode="updates" feature to expose every step of the Graph's execution to the light of day!
🎯 Learning Objectives
After taking this lesson, you will no longer be a novice who can only wait for the program to finish running, but an architect who can precisely control the execution rhythm of the Graph. Specific takeaways include:
- Shatter Black-Box Anxiety: Deeply understand the concept of "state slicing" during multi-agent execution.
- Master Core APIs: Thoroughly grasp the underlying differences between
stream_mode="updates"andvalues. - Agency Workflow in Practice: Accurately capture every node state update during the love-hate relationship (rejection loop) between the Writer and Editor, providing a real-time data source for your frontend.
- Elevate Architectural Taste: Learn how to elegantly parse LangGraph's Generator outputs.
📖 Under the Hood
Before diving into the code, let's talk about the underlying philosophy.
In LangGraph, when we call graph.invoke(), the system holds its breath until the entire graph reaches the END node before spitting out the final State to you. It's like ordering a complex dish like "Buddha Jumps Over the Wall" at a restaurant; you wait in the dining room for two hours, and finally, the waiter brings out the finished pot.
But if you call graph.stream(), it's like sitting at the bar of an open kitchen.
LangGraph provides several different stream_modes, with the two core ones being:
stream_mode="values": Every time the state updates, it pushes the complete current state to you. (Equivalent to taking a panoramic photo of the entire kitchen after every step and sending it to you).stream_mode="updates": Every time a node finishes executing, it only pushes the incremental modifications made by that node to the state. (Equivalent to a chef shouting: "Prep station finished, added a plate of chopped green onions!").
In multi-agent collaboration, the updates mode is our most commonly used weapon. Because it clearly tells us: "Who just did the work? And what work did they do?"
Let's use a Mermaid diagram to see how the data flows in our AI Content Agency when stream_mode="updates" is introduced:
sequenceDiagram
participant User as User (Client)
participant Graph as LangGraph Engine
participant Writer as Node: Writer
participant Editor as Node: Editor
User->>Graph: Initiate task stream(..., stream_mode="updates")
Graph->>Writer: 1. Execute writing
Writer-->>Graph: Return update: {draft: "初稿v1", revision_count: 1}
Graph-->>User: ⚡ yield {"Writer": {draft: "初稿v1", revision_count: 1}}
Graph->>Editor: 2. Review draft
Editor-->>Graph: Return update: {feedback: "缺乏深度", status: "REJECTED"}
Graph-->>User: ⚡ yield {"Editor": {feedback: "缺乏深度", status: "REJECTED"}}
Note over Graph, Editor: Conditional Edge: Rejected, route back to Writer
Graph->>Writer: 3. Rewrite based on feedback
Writer-->>Graph: Return update: {draft: "二稿v2", revision_count: 2}
Graph-->>User: ⚡ yield {"Writer": {draft: "二稿v2", revision_count: 2}}
Graph->>Editor: 4. Review again
Editor-->>Graph: Return update: {status: "APPROVED"}
Graph-->>User: ⚡ yield {"Editor": {status: "APPROVED"}}
Note over Graph, Editor: Conditional Edge: Approved, route to END
Graph-->>User: End StreamMake sense? Every ⚡ yield is a heartbeat emitted by LangGraph to the outside world. As long as the external business logic listens to these heartbeats, it can draw beautiful progress bars or real-time logs on the frontend: "Editor is reviewing...", "Editor rejected the article, Writer is making the 2nd revision...".
💻 Hands-On Code Practice
Enough talk, show me the code.
We will use Python to build this "Writer and Editor" loop workflow and capture it using stream_mode="updates".
To ensure everyone can run this directly, I've used mock functions for the LLM calls here, focusing primarily on the streaming architecture.
1. Define State and Node Logic
import time
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator
# ==========================================
# 1. Define the global State for our Agency
# ==========================================
class AgencyState(TypedDict):
topic: str
draft: str
feedback: str
revision_count: Annotated[int, operator.add] # Use reducer, accumulate on each update
status: str # "PENDING", "REJECTED", "APPROVED"
# ==========================================
# 2. Define Nodes
# ==========================================
def writer_node(state: AgencyState):
"""Writer node: Responsible for writing articles based on topic or feedback"""
topic = state.get("topic")
feedback = state.get("feedback", "")
current_count = state.get("revision_count", 0)
# Simulate the time taken for LLM thinking and writing
time.sleep(1.5)
if current_count == 0:
new_draft = f"【初稿】关于 {topic} 的文章。内容比较浅显。"
else:
new_draft = f"【第{current_count + 1}稿】关于 {topic} 的文章。已根据反馈『{feedback}』进行了深度优化。"
# Return incremental updates
return {
"draft": new_draft,
"revision_count": 1 # Because operator.add is used, returning 1 here will add 1 to the original base
}
def editor_node(state: AgencyState):
"""Editor node: Responsible for reviewing articles"""
draft = state.get("draft")
current_count = state.get("revision_count", 0)
# Simulate editor review time
time.sleep(1)
# Business logic: Force rejection for the first two times, approve on the third (simulating a strict editor)
if current_count < 3:
return {
"feedback": f"第 {current_count} 稿深度不够，给我回去重写！",
"status": "REJECTED"
}
else:
return {
"feedback": "这次写得不错，可以发布了。",
"status": "APPROVED"
}
# ==========================================
# 3. Define Conditional Edge
# ==========================================
def check_approval(state: AgencyState):
"""Determine if the editor approved"""
if state.get("status") == "APPROVED":
return "approved"
return "rejected"
2. Assemble the Graph and Enable Streaming Observation
Next is the moment to witness the magic. We will assemble the graph and use the stream method.
# ==========================================
# 4. Assemble the Graph
# ==========================================
workflow = StateGraph(AgencyState)
workflow.add_node("Writer", writer_node)
workflow.add_node("Editor", editor_node)
workflow.set_entry_point("Writer")
workflow.add_edge("Writer", "Editor")
# If rejected, go back to Writer; if approved, go to END
workflow.add_conditional_edges(
"Editor",
check_approval,
{
"rejected": "Writer",
"approved": END
}
)
app = workflow.compile()
# ==========================================
# 5. Witness the magic: Use stream_mode="updates"
# ==========================================
def run_agency_with_stream():
print("🚀 [Agency 系统启动] 接收到客户需求...")
initial_state = {
"topic": "LangGraph 流式输出原理解析",
"revision_count": 0
}
# The core is here! Call app.stream and specify the mode
# app.stream returns a Python Generator
stream_generator = app.stream(initial_state, stream_mode="updates")
step = 1
for event in stream_generator:
# The structure of event is a dictionary: { "Node Name": { State Increment } }
print(f"\n--- ⏳ 步骤 {step} ---")
# Iterate through event to get node name and output
for node_name, node_update in event.items():
print(f"👀 观测到节点执行完毕: 【{node_name}】")
# Based on different nodes, we can do different frontend UI rendering
if node_name == "Writer":
print(f"✍️ 写手提交了新稿件: {node_update.get('draft')}")
print(f"🔄 当前修改次数: {node_update.get('revision_count')} (这是增量累加后的结果)")
elif node_name == "Editor":
status = node_update.get('status')
if status == "REJECTED":
print(f"😡 主编大发雷霆，打回了稿件！反馈意见: {node_update.get('feedback')}")
else:
print(f"🎉 主编非常满意，审核通过！反馈意见: {node_update.get('feedback')}")
step += 1
time.sleep(0.5) # Pause slightly to make it easier to observe the output effect visually
print("\n✅ [Agency 系统完毕] 最终文章已交付客户！")
if __name__ == "__main__":
run_agency_with_stream()
3. Execution Output Demonstration (Terminal Output Mock)
When you run the code above, you no longer need to wait 5 seconds to see the final result. You will see a log pop up in the terminal every second or two, vividly displaying the internal "office politics" drama:
🚀 [Agency 系统启动] 接收到客户需求...
--- ⏳ 步骤 1 ---
👀 观测到节点执行完毕: 【Writer】
✍️ 写手提交了新稿件: 【初稿】关于 LangGraph 流式输出原理解析 的文章。内容比较浅显。
🔄 当前修改次数: 1 (这是增量累加后的结果)
--- ⏳ 步骤 2 ---
👀 观测到节点执行完毕: 【Editor】
😡 主编大发雷霆，打回了稿件！反馈意见: 第 1 稿深度不够，给我回去重写！
--- ⏳ 步骤 3 ---
👀 观测到节点执行完毕: 【Writer】
✍️ 写手提交了新稿件: 【第2稿】关于 LangGraph 流式输出原理解析 的文章。已根据反馈『第 1 稿深度不够，给我回去重写！』进行了深度优化。
🔄 当前修改次数: 2 (这是增量累加后的结果)
--- ⏳ 步骤 4 ---
👀 观测到节点执行完毕: 【Editor】
😡 主编大发雷霆，打回了稿件！反馈意见: 第 2 稿深度不够，给我回去重写！
--- ⏳ 步骤 5 ---
👀 观测到节点执行完毕: 【Writer】
✍️ 写手提交了新稿件: 【第3稿】关于 LangGraph 流式输出原理解析 的文章。已根据反馈『第 2 稿深度不够，给我回去重写！』进行了深度优化。
🔄 当前修改次数: 3 (这是增量累加后的结果)
--- ⏳ 步骤 6 ---
👀 观测到节点执行完毕: 【Editor】
🎉 主编非常满意，审核通过！反馈意见: 这次写得不错，可以发布了。
✅ [Agency 系统完毕] 最终文章已交付客户！
Imagine how great your user experience would be if you pushed this data to the frontend via WebSockets? Users can watch the AI team polish their work step-by-step, just like watching a progress bar. This sense of "participation" and "transparency" is something a monolithic LLM simply cannot provide.
Pitfalls & Best Practices
As a veteran with 10 years of architecture experience, I must give you some warnings here. While streaming output is great, there are quite a few pitfalls:
💣 Pitfall 1: Confusing updates and values leading to frontend data chaos
Symptom: The state received by the frontend sometimes lacks fields, and sometimes it's complete.
Solution: Remember, updates only returns the dictionary returned by the current node. If you only return {"draft": "xxx"} in writer_node, then in updates mode, the event you receive will not have topic and status. If you need to get the full state every time to render the page, please use stream_mode="values"; if you just want to trigger events (like popping up a Toast saying "Editor has reviewed"), updates is more lightweight.
💣 Pitfall 2: Confusing "Node-level streaming" with "Token-level streaming"
Symptom: The boss says they want the ChatGPT-style effect where words pop up one by one, but you only achieved an effect where nodes pop up one by one. Solution: These are two completely different dimensions!
- Node-level streaming (covered in this lesson): Focuses on which step the Agent workflow has reached.
- Token-level streaming: Focuses on the text generation process of a specific large model.
In advanced architectures, we usually combine the two. That is: use
stream_mode="updates"to report progress during graph transitions, while simultaneously enabling the LLM's asynchronous Token Streaming to push to the frontend when calling the LLM inside thewriter_node. We will dedicate a specific lesson to this in the upcoming "Advanced Communication" section.
💣 Pitfall 3: Double-triggering of Reducer accumulators
Symptom: revision_count inexplicably becomes 4, 8, 16.
Solution: When using Annotated[int, operator.add], you must be clear that every time a node returns a number, it will be added to the original base. If you accidentally also return {"revision_count": 1} in the editor_node, it will cause double accumulation. Strictly dividing the write permissions of each node for specific fields in the State is a core principle of multi-agent architecture design.
📝 Lesson Summary
Today, we integrated a "God's eye view" into our Agency project.
Through LangGraph's stream_mode="updates", we transformed a black-box system that was originally extremely time-consuming and likely to make people lose patience into a white-box system with clear steps and transparent states. This is not just an optimization at the code level, but a dimensional strike in product experience.
In actual commercial implementations, being able to expose internal dynamics like "Editor rejects and requests rewrite" to the user will not make the user feel the system is dumb; on the contrary, it will make the user feel your AI is "working hard" and "very professional". This is the best manifestation of technology feeding back into the product.
In the next lesson, we will dive into an even more exciting topic: Human-in-the-loop. If the Editor AI is unsure, how do we get a real human boss to intervene and approve? How does the Graph achieve "pause execution, wait for human input"?
Stay tuned for Part 11. Class dismissed! Remember to run today's code locally and experience the thrill of the generator spitting out data!