W7-Day2｜Supervisor 模式：让一个 Agent 负责调度，多个 Worker 负责执行

昨天主要是在画 Supervisor 模式的图，今天开始真正把它写成代码。

今天的 Demo 是一个 Supervisor + 3 个 Worker 的多 Agent 流程：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
                    ┌──────────────┐
                    │  supervisor  │
                    │  决定下一步   │
                    └──────┬───────┘
                           │
             ┌─────────────┼─────────────┐
             ▼             ▼             ▼
        researcher       writer       reviewer
             │             │             │
             └─────────────┴─────────────┘
                           │
                           ▼
                       supervisor
                           │
                    FINISH → END

三个 Worker 分工：

1
2
3
researcher → 查资料
writer     → 写文章
reviewer   → 审文章

最重要的不是“有三个 Agent”，而是：

Worker 负责执行，Supervisor 负责决定下一步。

所以实际运行不是一条固定流水线：

1
researcher → writer → reviewer

而是一个循环：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
supervisor
    ↓
researcher
    ↓
supervisor
    ↓
writer
    ↓
supervisor
    ↓
reviewer
    ↓
supervisor
    ↓
FINISH
    ↓
END

Worker 干完以后必须重新回到 Supervisor。

一、先看今天的 State

今天的 State：

1
2
3
class SuperState(MessagesState):
    next: str
    rounds: int

在 MessagesState 自带的 messages 之外，又增加两个字段：

1
2
3
messages → 工作内容和历史
next     → Supervisor 的下一步决策
rounds   → 已经循环多少次

这三个字段的职责并不一样。

messages：数据流

它保存整个任务的上下文。

例如：

1
2
3
4
5
6
7
用户：写一篇 LangGraph 介绍

researcher：
LangGraph 是……

writer：
LangGraph 是一个……

Worker 每次都是从这里读取前面的工作结果。

next：控制流

例如：

1
next = "writer"

表示：

下一步执行 writer。

所以 next 是给图的控制逻辑看的，而不是业务内容。

rounds：安全控制

Supervisor 是循环：

1
Supervisor → Worker → Supervisor → Worker → ...

所以必须记录轮数，防止模型一直循环。

二、为什么要有两个 LLM 变量

代码：

1
2
llm = make_llm()
router_llm = make_llm()

当前它们使用的是同一个模型，但职责不同：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
llm
 ↓
Worker
 ↓
真正干活


router_llm
 ↓
Supervisor
 ↓
决定下一步

Worker 的任务是：

1
查资料 / 写文章 / 审稿

Supervisor 的任务是：

1
现在应该找谁？

虽然都是 LLM，但从系统设计上已经是两个不同角色。

三、MEMBERS：为什么专门维护一个名单

1
2
3
4
5
MEMBERS = [
    "researcher",
    "writer",
    "reviewer"
]

这个列表相当于 Supervisor 的“花名册”。

后面很多代码都依赖它：

1
2
3
4
5
6
7
Supervisor 能选择谁
        ↓
条件边能连接到谁
        ↓
Worker 的名字是否合法
        ↓
Worker 回边连接给谁

所以它最好成为：

单一事实来源。

以后增加：

1
"fact_checker"

就应该优先改这里，而不是到程序各处手写名字。

四、Route：Supervisor 怎么把“思考结果”交给程序

接下来：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
class Route(BaseModel):
    next: Literal[
        "researcher",
        "writer",
        "reviewer",
        "FINISH"
    ]

    reason: str = Field(
        default="",
        description="一句话理由, 方便调试"
    )

这里解决的是一个核心问题：

模型说完“下一步该谁”以后，程序怎么可靠地拿到这个答案？

不能让模型随便返回一句：

1
我觉得现在应该让 writer 处理一下……

而是约束成：

1
2
3
4
{
    "next": "writer",
    "reason": "researcher 已经完成资料整理"
}

其中真正影响图运行的是：

1
decision.next

而 reason 主要用于调试。

因此这里形成了一条链：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
LLM
 ↓
结构化输出
 ↓
Route
 ↓
decision.next
 ↓
State["next"]
 ↓
条件边
 ↓
真正执行哪个节点

五、为什么最后用 `json_mode`

最开始写：

1
router_llm.with_structured_output(Route)

但实际跑 DeepSeek 时，默认路径触发了 json_schema，返回：

1
This response_format type is unavailable now

后来又试了：

1
method="function_calling"

结果变成：

1
Thinking mode does not support this tool_choice

最后通过探针验证，当前环境真正可用的是：

1
2
3
4
router = router_llm.with_structured_output(
    Route,
    method="json_mode"
)

所以今天这里真正学到的并不是“DeepSeek 不支持结构化输出”。

而是：

同一个结构化输出接口，底层可能走完全不同的协议路径。

可以画成：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
with_structured_output
        │
        ├── json_schema
        │      ↓
        │    当前环境不支持
        │
        ├── function_calling
        │      ↓
        │    thinking mode 拒绝 tool_choice
        │
        └── json_mode
               ↓
             可用

所以以后遇到这类兼容问题，先写一个最小 Probe，确认模型实际支持什么，不要凭印象猜。

六、Worker 为什么用工厂函数创建

这部分源码初看有一点绕：

1
2
3
4
5
6
7
8
9
def make_worker(name, tools, prompt):
    agent = create_react_agent(
        llm,
        tools,
        prompt=prompt
    )

    def node(state):
        ...

它其实是在做一件事：

把“创建 Worker”这件事封装成一个模板。

然后：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
worker_nodes = {
    "researcher": make_worker(
        "researcher",
        [web_search],
        WORKER_PROMPTS["researcher"]
    ),

    "writer": make_worker(
        "writer",
        [],
        WORKER_PROMPTS["writer"]
    ),

    "reviewer": make_worker(
        "reviewer",
        [],
        WORKER_PROMPTS["reviewer"]
    ),
}

最终得到三个不同的 Worker：

1
2
3
researcher
writer
reviewer

它们本质上结构类似，只是：

1
2
3
名字不同
Prompt 不同
工具不同

所以这里的工厂函数主要是为了：

避免重复写三遍几乎一样的 Worker 创建代码。

七、三个 Worker 为什么职责不同

Prompt：

1
2
3
4
5
WORKER_PROMPTS = {
    "researcher": "你是研究员……",
    "writer": "你是写手……",
    "reviewer": "你是审稿人……",
}

工具：

1
2
3
researcher → web_search
writer     → 无工具
reviewer   → 无工具

于是职责就很清楚：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
researcher
    ↓
产生资料

writer
    ↓
消费资料，产生文章

reviewer
    ↓
消费文章，产生审稿结果

这里实际上已经有一个很清楚的数据加工链：

1
2
3
4
5
6
7
原始任务
   ↓
资料
   ↓
文章
   ↓
评审结果

Supervisor 只是负责控制这个加工链什么时候进入下一阶段。

八、真正执行 Worker 时，State 是怎么进去的

Worker 的核心代码：

1
2
3
result = agent.invoke({
    "messages": state["messages"]
})

这里特别容易误解。

它不是：

1
把整个 LangGraph State 原封不动塞给 Agent

而是：

1
2
3
4
5
LangGraph State
      │
      └── state["messages"]
                 ↓
            Worker Agent

所以 Worker 真正拿到的核心数据是：

当前 State 中的消息历史。

如果此时 State 是：

1
2
3
messages:
    用户要求
    researcher 结果

那么 writer 就可以根据这些内容继续工作。

九、Worker 干完以后为什么还要重新包装一次消息

Worker 内部 Agent 最后得到：

1
2
last = result["messages"][-1]
text = str(last.content)

接着：

1
2
3
4
5
6
7
8
return {
    "messages": [
        HumanMessage(
            content=f"[{name} 交活]\n{text}",
            name=name
        )
    ]
}

这一步非常重要。

因为 Worker 内部的最终结果不能只停留在它自己的 Agent 里。

它必须进入 LangGraph 的 State。

于是：

1
2
3
4
5
6
7
Worker 内部
    ↓
生成结果
    ↓
包装为 HumanMessage
    ↓
写回 State.messages

同时增加：

1
name=name

于是 Supervisor 下一轮就知道：

1
这是 researcher 交来的

而不是只能看到一条匿名消息。

十、这里可以第一次把“逻辑流”和“数据流”放在一起看

现在完整看一次 researcher：

逻辑流

1
2
3
4
5
6
7
supervisor
    ↓
decision.next = researcher
    ↓
条件边
    ↓
researcher

这是：

决定谁执行。

数据流

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
State.messages
    ↓
researcher Agent
    ↓
web_search
    ↓
研究结果
    ↓
HumanMessage(name="researcher")
    ↓
写回 State.messages

这是：

决定执行时拿什么，执行后留下什么。

两条流最后重新汇合：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
                     State
                ┌───────────────┐
                │ messages      │
                │ next          │
                │ rounds        │
                └───────┬───────┘
                        │
            ┌───────────▼───────────┐
            │      supervisor       │
            │       决定 next       │
            └───────────┬───────────┘
                        │
                     控制流
                        │
                        ▼
                      Worker
                        │
                     数据流
                        │
                        ▼
                  写回 messages
                        │
                        └────→ supervisor

这是今天我认为理解 Supervisor 最重要的一张图。

十一、Supervisor 节点到底在做什么

源码核心：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
def supervisor_node(state):
    rounds = state.get("rounds", 0) + 1

    if rounds > MAX_ROUNDS:
        return {
            "next": "FINISH",
            "rounds": rounds,
            ...
        }

    msgs = [
        SystemMessage(
            content=SUPERVISOR_PROMPT
        )
    ] + state["messages"]

    decision = router.invoke(msgs)

    return {
        "next": decision.next,
        "rounds": rounds,
    }

可以拆成四步。

第一步：计算轮数

1
rounds += 1

第二步：检查是否超限

如果已经超限：

1
直接 FINISH

连 LLM 都不再调用。

这是为了避免已经知道要结束了，还白花一轮模型费用。

第三步：把“Supervisor 的规则 + 当前消息历史”交给 LLM

1
2
3
msgs = [
    SystemMessage(content=SUPERVISOR_PROMPT)
] + state["messages"]

模型看到：

1
2
3
自己的职责
+
前面的工作历史

第四步：把模型决定写回 State

1
2
3
4
return {
    "next": decision.next,
    "rounds": rounds,
}

所以 Supervisor 这一轮的作用其实非常简单：

读取 State → 决定下一步 → 修改 State。

十二、条件边：模型只负责“说去哪”，图负责“真的去”

路由函数：

1
2
3
4
5
6
7
8
def route_from_supervisor(state):
    nxt = state.get("next") or "FINISH"

    return (
        nxt
        if nxt in MEMBERS or nxt == "FINISH"
        else "FINISH"
    )

然后：

1
2
3
4
5
builder.add_conditional_edges(
    "supervisor",
    route_from_supervisor,
    path_map
)

完整过程其实是：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
Supervisor LLM
       ↓
"writer"
       ↓
state["next"]
       ↓
route_from_supervisor()
       ↓
"writer"
       ↓
path_map
       ↓
writer 节点

所以一个非常值得记住的边界是：

LLM 做决策，Graph 执行决策。

模型不是直接调用：

1
writer()

它只是在 State 中留下：

1
next = "writer"

然后 LangGraph 根据图结构真正执行 writer。

十三、为什么 Worker 必须回 Supervisor

这里：

1
2
3
4
5
for name in MEMBERS:
    builder.add_edge(
        name,
        "supervisor"
    )

实际上是整个模式的循环核心。

形成：

1
2
3
researcher ─┐
writer     ─┼──→ supervisor
reviewer   ─┘

如果没有这组回边：

1
2
3
4
5
supervisor
    ↓
researcher
    ↓
结束

那么 Worker 做完以后，就没有人继续判断下一步。

所以 Supervisor 模式真正的图形是：

1
2
3
Supervisor → Worker
      ↑        │
      └────────┘

直到：

1
Supervisor → FINISH → END

十四、把一次完整运行串起来

现在假设模型做出了一个比较理想的路径：

1
2
3
4
5
6
7
8
supervisor
→ researcher
→ supervisor
→ writer
→ supervisor
→ reviewer
→ supervisor
→ FINISH

状态变化可以粗略理解为：

初始

1
2
3
4
5
6
7
8
messages:
    用户要求

next:
    ""

rounds:
    0

Supervisor 第一次

1
2
next = researcher
rounds = 1

researcher 执行

1
2
3
messages:
    用户要求
    researcher 交活

Supervisor 第二次

读取：

1
2
用户要求
researcher 结果

决定：

1
2
next = writer
rounds = 2

writer 执行

1
2
3
4
messages:
    用户要求
    researcher 结果
    writer 文章

Supervisor 第三次

决定：

1
2
next = reviewer
rounds = 3

reviewer 执行

1
2
3
4
5
messages:
    用户要求
    researcher 结果
    writer 文章
    reviewer 意见

Supervisor 最后决定

1
next = FINISH

于是：

1
2
3
FINISH
  ↓
END

整个过程中：

1
messages

越来越丰富。

而：

1
next

负责控制当前应该去哪。

这就是今天最核心的：

数据流在 State 中积累，控制流围绕 State 动态前进。

十五、为什么要加 MAX_ROUNDS

Supervisor 本质上是动态循环。

例如 reviewer 可能说：

1
文章需要修改

于是：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
reviewer
 ↓
writer
 ↓
reviewer
 ↓
writer
 ↓
reviewer
……

如果没有限制，模型可以一直绕下去。

所以：

1
MAX_ROUNDS = 8

相当于：

给模型的自主决策设置一道硬上限。

而且今天特意做了 fallback 模式：

1
MAX_ROUNDS = 1

让流程进入：

1
2
3
4
supervisor
→ researcher
→ supervisor
→ 强制 FINISH

这样才能真正证明兜底逻辑生效，而不是只看代码说“应该能生效”。

十六、今天的一个重要调试经验：两次运行不能拿来互相证明

之前的思路是：

1
2
3
4
5
6
7
stream()
    ↓
拿流转链

invoke()
    ↓
拿最终 State

然后：

1
2
第一次：4 轮
第二次：6 轮

再拿两者比较。

问题是：

它们根本不是同一次运行。

LLM 是动态的，所以即使：

1
temperature = 0

也不能把两个独立运行当成同一个实验。

最后改成：

1
2
3
4
for mode, chunk in graph.stream(
    INITIAL,
    stream_mode=["updates", "values"]
):

于是一次运行同时得到：

1
2
updates → 流程链
values  → State 快照

这样才能比较：

1
这一次运行到底发生了什么。

这个思路其实比 Supervisor 本身还通用：

验证动态系统时，尽量让观察数据来自同一次实验。

十七、今天还有一个特别隐蔽的测试坑：空集合

比如：

1
2
for m in delivered:
    assert m.name in MEMBERS

如果：

1
delivered = []

那么：

1
2
3
4
5
循环执行 0 次
↓
assert 执行 0 次
↓
程序不报错

于是你会产生一种错觉：

“检查通过了。”

实际上：

根本没有检查。

所以正确做法是：

1
2
3
4
assert delivered

for m in delivered:
    assert m.name in MEMBERS

先证明：

1
确实存在需要验证的数据

再检查数据是否合法。

这就是今天说的“真空真”。

十八、验证器也可能看错地方

今天还有一个反例。

原本检测工具是否真的被调用，写成：

1
2
3
4
called = any(
    "真身被调用" in str(m.content)
    for m in out["messages"]
)

结果：

1
❌ 工具没有调用

但实际上工具真的调用了。

因为：

1
print(...)

走的是：

1
stdout

而：

1
return ...

才会进入：

1
ToolMessage

也就是：

1
2
3
4
5
6
7
8
9
print
 ↓
终端

return
 ↓
ToolMessage
 ↓
State.messages

这是两条完全不同的数据通道。

所以后面才改成：

1
2
3
CALLS
+
ToolMessage

两个证据一起看。

这个例子让我今天对测试又多了一层认识：

验证器本身也必须建立在正确的数据通道上。

十九、今天最终形成的理解

以前看到这种代码，容易先被很多 API 吓到：

1
2
3
4
5
6
7
8
StateGraph
add_node
add_edge
add_conditional_edges
with_structured_output
create_react_agent
HumanMessage
MessagesState

但今天把它还原成系统以后，其实就是：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
                 State
        ┌──────────┼──────────┐
        │          │          │
    messages      next      rounds
        │          │          │
        │          │          │
        ▼          ▼          ▼
     数据流     控制流      安全控制
        │          │
        └────┬─────┘
             ▼
        Supervisor
             │
          决定 next
             │
             ▼
          条件边
             │
             ▼
           Worker
             │
          干自己的活
             │
             ▼
      结果重新写入 State
             │
             └────→ Supervisor

所以今天我真正理解的，不只是：

“Supervisor 可以调度多个 Agent。”

而是：

Supervisor 是一个不断读取 State、修改 State，并依据 State 决定下一跳的动态控制器。

而整个系统可以分成两条主线：

1
2
3
4
5
控制流：
Supervisor → next → 条件边 → Worker → Supervisor

数据流：
messages → Worker → 工作结果 → messages

这两条流最后都通过 State 汇合。

以后再看类似的多 Agent 代码，我觉得可以先问自己两个问题：

第一：谁决定下一步？

第二：上一阶段产生的数据，是通过什么字段传给下一阶段的？

先找到这两个答案，源码就不会再是一堆孤立的 API。

完整 Demo

下面保留今天最终版本的核心代码。

为了方便之后复习，完整 Demo 放在文章最后；省略 import、包安装和断言代码，正文重点放在运行机制和代码结构。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
MODEL_NAME = "deepseek-chat"
BASE_URL = "https://api.deepseek.com"
API_KEY = os.getenv("DEEPSEEK_API_KEY")

MODE = sys.argv[1] if len(sys.argv) > 1 else "normal"
MAX_ROUNDS = 1 if MODE == "fallback" else 8
FAULT = os.getenv("FAULT") or None

MEMBERS = [
    "researcher",
    "writer",
    "reviewer"
]


def make_llm(temperature: float = 0):
    return ChatOpenAI(
        model=MODEL_NAME,
        base_url=BASE_URL,
        api_key=API_KEY,
        temperature=temperature,
    )


llm = make_llm()
router_llm = make_llm()


class SuperState(MessagesState):
    next: str
    rounds: int


class Route(BaseModel):
    next: Literal[
        "researcher",
        "writer",
        "reviewer",
        "FINISH"
    ]
    reason: str = Field(
        default="",
        description="一句话理由，方便调试"
    )


router = router_llm.with_structured_output(
    Route,
    method="json_mode"
)


@tool
def web_search(query: str) -> str:
    print(f">>> [web_search] query={query!r}")

    return (
        f"[模拟检索结果] 关于「{query}」："
        "LangGraph 是用于构建有状态多步 Agent 的框架；"
        "核心包括 StateGraph、条件边和持久化能力。"
    )


WORKER_PROMPTS = {
    "researcher": (
        "你是研究员。用 web_search 查资料，"
        "把关键要点整理成 2-3 条，不要写完整文章。"
    ),

    "writer": (
        "你是写手。把对话历史里 researcher "
        "交的资料整合成一篇 300 字左右的中文文章。"
        "只写文章，不要写评论或修改意见。"
    ),

    "reviewer": (
        "你是审稿人。检查上一位 writer 写的文章，"
        "指出事实性、结构、表达上的具体问题。"
        "如果文章已经可以定稿，明确说「可以定稿」。"
    ),
}


def make_worker(name: str, tools: list, prompt: str):
    agent = create_react_agent(
        llm,
        tools,
        prompt=prompt
    )

    def node(state: SuperState) -> dict:
        result = agent.invoke({
            "messages": state["messages"]
        })

        last = result["messages"][-1]
        text = str(last.content)

        if FAULT == "no_deliver":
            return {
                "messages": []
            }

        msg_name = (
            None
            if FAULT == "no_name"
            else name
        )

        return {
            "messages": [
                HumanMessage(
                    content=f"[{name} 交活]\n{text}",
                    name=msg_name
                )
            ]
        }

    node.__name__ = name
    return node


worker_nodes = {
    "researcher": make_worker(
        "researcher",
        [web_search],
        WORKER_PROMPTS["researcher"]
    ),

    "writer": make_worker(
        "writer",
        [],
        WORKER_PROMPTS["writer"]
    ),

    "reviewer": make_worker(
        "reviewer",
        [],
        WORKER_PROMPTS["reviewer"]
    ),
}


SUPERVISOR_PROMPT = """
你是写作工作室的主编(supervisor)，手下三位成员：

- researcher：负责查资料，只交资料要点，不写正文
- writer：负责把资料整合成一篇结构清晰的中文文章
- reviewer：负责审稿，指出文章的问题

你的职责：每一步只决定“下一步交给谁”。
不要替任何成员干活，不要自己写文章。

标准流程：
researcher 拿资料
→ writer 成稿
→ reviewer 审稿
→ 若有重大问题，可以回 writer 修改
→ 都满意后输出 FINISH

判断依据：
看历史里带名的交活消息。

已经交过活的不必重复派。

只输出一个 JSON 对象，例如：

{
    "next": "researcher",
    "reason": "还没有资料"
}

next 只能是：
researcher / writer / reviewer / FINISH
"""


def supervisor_node(state: SuperState) -> dict:
    rounds = state.get("rounds", 0) + 1

    if rounds > MAX_ROUNDS:
        return {
            "next": "FINISH",
            "rounds": rounds,
            "messages": [
                (
                    "assistant",
                    f"【已达最大轮数 {MAX_ROUNDS}，强制结束】"
                )
            ],
        }

    msgs = [
        SystemMessage(
            content=SUPERVISOR_PROMPT
        )
    ] + state["messages"]

    decision = router.invoke(msgs)

    print(
        f"[supervisor] round={rounds} "
        f"next={decision.next} "
        f"({decision.reason})"
    )

    return {
        "next": decision.next,
        "rounds": rounds,
    }


def route_from_supervisor(state: SuperState):
    nxt = state.get("next") or "FINISH"

    return (
        nxt
        if nxt in MEMBERS or nxt == "FINISH"
        else "FINISH"
    )


def build_graph():
    builder = StateGraph(SuperState)

    builder.add_node(
        "supervisor",
        supervisor_node
    )

    for name, node in worker_nodes.items():
        builder.add_node(
            name,
            node
        )

    builder.add_edge(
        START,
        "supervisor"
    )

    path_map = {
        member: member
        for member in MEMBERS
    }

    path_map["FINISH"] = END

    builder.add_conditional_edges(
        "supervisor",
        route_from_supervisor,
        path_map
    )

    for name in MEMBERS:
        builder.add_edge(
            name,
            "supervisor"
        )

    return builder.compile()


INITIAL = {
    "messages": [
        (
            "user",
            "写一篇 300 字介绍 LangGraph 的文章，"
            "要求有事实依据"
        )
    ],
    "next": "",
    "rounds": 0,
}


def main():
    graph = build_graph()

    chain = []
    final_state = None

    for mode, chunk in graph.stream(
        INITIAL,
        stream_mode=["updates", "values"]
    ):
        if mode == "updates":
            for node_name in chunk:
                chain.append(node_name)

        elif mode == "values":
            final_state = chunk

    out = final_state

    print(
        "\n[流转链]",
        " → ".join(chain)
    )

    print("\n=== 成品 ===")
    print(
        str(
            out["messages"][-1].content
        )[:400]
    )

    print(
        f"[结果] "
        f"next={out['next']} "
        f"rounds={out['rounds']}"
    )

    mermaid = graph.get_graph().draw_mermaid()

    with open(
        "w7_supervisor_topology.md",
        "w",
        encoding="utf-8"
    ) as f:
        f.write(
            "```mermaid\n"
            + mermaid
            + "\n```\n"
        )


if __name__ == "__main__":
    main()