12.2 分支、合并与循环
12.1 那条 分类 → 检索 → 综合 是一条直线：每个节点只有一个后继，图从入口一路走到出口。真实流程远没有这么听话——「分类结果不同，后面查的表也不同」（分支）、「两个独立的检查同时做，最后合起来看」（合并）、「写出来不合格就打回去重写，最多试三次」（循环）。这一节把直线织成网。
本节把 minagent 的图引擎从「单后继」升级为「波次执行」：一批节点在同一波里运行，各自触发下游后再去重汇合。在此之上讲清三种结构——条件边、fan-out/fan-in、带退出条件的循环边——并用假模型真跑一个「生成 → 校验 → 不合格回炉」的循环，打印每轮迭代直到收敛（或撞上次数上限）。
12.2.1 从单后继到一波多节点
12.1 的 _next 只返回第一条满足条件的边，所以任何时刻只有一个当前节点。要支持「一个节点触发多个下游」，得把「找一条边」换成「找一批后继」，并把执行模型从「单节点循环」改成「波次（wave）」：
# minagent/workflow/graph.py —— 替换 12.1 的 _next 与 run
def _successors(self, name: str, state: State) -> list[str]:
out = []
for e in self.edges:
if e.src == name and (e.cond is None or e.cond(state)):
out.append(e.dst)
return out
def run(self, state: State, max_waves: int = 25, on_node=None) -> State:
wave = [self.entry]
w = 0
while wave:
w += 1
if w > max_waves:
raise RuntimeError(f"超过 max_waves={max_waves}，图未收敛：{wave}")
nxt: list[str] = []
for name in wave:
before = dict(state)
update = self.nodes[name].fn(state) or {}
state.update(update)
if on_node is not None:
on_node(w, name, before, dict(state))
for s in self._successors(name, state):
if s not in nxt: # 波内去重 = fan-in 的合并点
nxt.append(s)
wave = nxt
return state
Node / Edge / Graph 三个 dataclass 一字未改，改的只是 run 的调度方式。核心是 wave 这个列表：它装着「当前这一波要跑的节点」。
- 一波可以有一个或多个节点：入口波只有
[entry]，但某个节点一旦有多个后继，下一波就会变长。 - 波内去重是「合并」的实现：如果
research和outline都指向merge，那么处理完这一波后，nxt里merge只会出现一次——两个上游都跑完了，merge才在下一波跑一次。这就是 fan-in。 - 同波节点的执行顺序 = 列表顺序：本节的最小引擎是串行跑同波节点的（
for name in wave）。它们在语义上「互不依赖」，所以顺序不影响正确性；真并行是卷二/卷三的事，这里只要保证「都跑完才进入下一波」。 max_waves取代max_steps：计数单位从「节点」变成「波」，但作用一样——给带环的图兜底。
一句话：单后继是「状态机」，一波多节点是「数据流」。前者任何时刻只在一个状态，后者允许一批节点同时活跃。
12.2.2 条件边：按 state 字段走不同分支
条件边在 12.1 就留好了接口——Edge.cond 是一个 Callable[[State], bool]。升级到波次模型后，条件边的行为更值得说清：_successors 会返回所有条件为真的边，所以一个节点可以同时触发多个分支。
最常用的形态是「互斥分支」：根据 state 里某个字段，走且只走一条路。比如按风险等级路由：
def route(state):
level = state["risk"]
return {"route": "manual" if level == "high" else "auto"}
g.add_edge("assess", "manual_review", cond=lambda s: s["route"] == "manual")
g.add_edge("assess", "auto_approve", cond=lambda s: s["route"] == "auto")
两个条件恰好互补（route 非 manual 即 auto），所以任何时刻只有一条边成立。写互斥条件时一定要确认它们覆盖全部情况、且两两不相交——漏了一种情况，图就走到死胡同（_successors 返回空，直接结束）；重叠了，两条分支会同时触发，状态被写乱。
这正是 12.1 埋下的伏笔：classify 节点在真实模型下可能吐出意料之外的类别，此时不能靠「命中就查表、没命中就报错」，而应该加一条条件边——「识别为已知类别 → 走 retrieve；否则 → 走 fallback」。条件边把「模型的意外输出」变成了图里一个显式的分支，而不是散落在节点内部的 if。
12.2.3 fan-out / fan-in：并行触发再合并
fan-out 是一个节点触发多个下游，fan-in 是多个上游汇入同一个节点。它们合起来就是「分头干活、最后汇总」的经典模式：
def plan(state):
return {"topic": state["raw"]}
def research(state):
return {"research": f"{state['topic']} 的 3 个事实"}
def outline(state):
return {"outline": f"{state['topic']} 的 4 段结构"}
def merge(state):
parts = [state.get("research", ""), state.get("outline", "")]
return {"draft": " / ".join(p for p in parts if p)}
g = (
Graph()
.add_node("plan", plan)
.add_node("research", research)
.add_node("outline", outline)
.add_node("merge", merge)
.add_edge("plan", "research") # fan-out：plan 同时触发两个下游
.add_edge("plan", "outline")
.add_edge("research", "merge") # fan-in：两个下游都汇入 merge
.add_edge("outline", "merge")
.set_entry("plan")
)
真跑一遍，观察波次编号：
波 1 | plan | 新增 ['topic']
波 2 | research | 新增 ['research']
波 2 | outline | 新增 ['outline']
波 3 | merge | 新增 ['draft']
draft: GIL 的 3 个事实 / GIL 的 4 段结构
关键在第 2 波：research 和 outline 同属一波，它们都读 topic、各写各的字段，互不干扰。到了第 3 波 merge 才执行，此时它读到的 state 里两个上游的产物都在——这就是 fan-in 的语义保证：合并点只在上游全部完成后运行。如果引擎不按波执行、而是每碰到一条边就立刻跳过去，merge 可能在 outline 还没跑时就执行，读到半成品。
fan-out 有一个必须守的纪律：同波节点之间不能互相依赖。research 不该去读 outline 写的字段——它们在同一波，谁先跑不确定。凡是「B 依赖 A 的产物」的，就必须是 A → B 的边，把 B 推到下一波。这条纪律是波次模型正确性的前提。
顺带一提，fan-out/fan-in 也是「并行」的自然落点：既然同波节点互不依赖，它们理论上可以真的并发执行。本节为了确定性先串行跑，把「并发」留到卷二——一旦引入并发，state 的合并就要处理写冲突，那是另一个层次的问题。
12.2.4 带退出条件的循环边
循环边就是一条指向「已经走过的节点」的边。它让图可以回炉重做——这正是循环式 Agent 擅长的，如今在图里也能表达。但循环边天然带来死循环的风险，所以必须配一个退出条件。
经典形态是「重试直到合格，或达到次数上限」。用两条互补的条件边表达：
MAX_ATTEMPTS = 2
g.add_edge("validate", "generate",
cond=lambda s: not s["ok"] and s["attempts"] < MAX_ATTEMPTS) # 回炉边
g.add_edge("validate", "finish",
cond=lambda s: s["ok"] or s["attempts"] >= MAX_ATTEMPTS) # 出口边
两个条件再次互补：要么回炉（还不合格且没到上限），要么出口（合格了或到上限了）。注意出口条件写的是 s["ok"] or s["attempts"] >= MAX_ATTEMPTS——即便始终不合格，次数一到也强制走 finish，绝不无限回炉。这是循环边最重要的一条纪律：回炉条件里必须带上限判断，否则一个永远不合格的校验会让图转个不停。
12.2.5 真跑：生成 → 校验 → 不合格回炉
把三个节点接成一个带环的图：generate 调模型产出草稿，validate 用纯代码校验（长度是否超限），finish 收尾。回炉上限设为 2 次。
from minagent.fake import FakeModel
from minagent.types import Message
from minagent.workflow.graph import Graph
MAX_ATTEMPTS = 2
LIMIT = 40
D_LONG = "这是一段相当冗长的第一版草稿，里面塞了很多并不必要的修饰和解释，需要再精简一些才能通过长度校验。" # 48 字
D_OK = "精简后的第二版草稿，长度已达标。" # 16 字
MODEL = None # 每个场景重建，避免脚本串味
def generate(state):
resp = MODEL.complete([Message(role="user", content=state["task"])], [])
return {"draft": resp.message.content.strip(),
"attempts": state.get("attempts", 0) + 1}
def validate(state):
ok = len(state["draft"]) <= LIMIT
return {"ok": ok, "reason": f"长度 {len(state['draft'])} {'<=' if ok else '>'} {LIMIT}"}
def finish(state):
return {"final": state["draft"], "passed": state["ok"]}
g = (
Graph()
.add_node("generate", generate)
.add_node("validate", validate)
.add_node("finish", finish)
.add_edge("generate", "validate")
.add_edge("validate", "generate",
cond=lambda s: not s["ok"] and s["attempts"] < MAX_ATTEMPTS) # 回炉边
.add_edge("validate", "finish",
cond=lambda s: s["ok"] or s["attempts"] >= MAX_ATTEMPTS) # 出口边
.set_entry("generate")
)
跑两个场景：A 里模型第二次给出合格草稿，B 里两次都不合格。真跑结果（Python 3.14.6）：
--- 场景 A：第二次回炉后合格 ---
波 1 | generate | draft=48 字 (第 1 次)
波 2 | validate | ok=False (长度 48 > 40)
波 3 | generate | draft=16 字 (第 2 次)
波 4 | validate | ok=True (长度 16 <= 40)
波 5 | finish |
passed=True attempts=2 final='精简后的第二版草稿，长度已达标。'
--- 场景 B：两次都不合格，撞上限 ---
波 1 | generate | draft=48 字 (第 1 次)
波 2 | validate | ok=False (长度 48 > 40)
波 3 | generate | draft=48 字 (第 2 次)
波 4 | validate | ok=False (长度 48 > 40)
波 5 | finish |
passed=False attempts=2 final='这是一段相当冗长的第一版草稿，里面塞了很多并不必要的修饰和解释，需要再精简一些才能通过长度校验。'
场景 A 走了一个漂亮的环：第 1 波生成 48 字草稿，第 2 波校验不过，回炉边生效 → 第 3 波重新生成 16 字，第 4 波校验通过，出口边生效 → 第 5 波收尾。attempts 字段是循环的计数器，它由 generate 每轮自增，回炉条件靠它判断是否到顶。
场景 B 则是撞上限的结局：两次都 48 字，第 2 次校验后 attempts 已达 MAX_ATTEMPTS，回炉条件 attempts < 2 为假，出口条件 attempts >= 2 为真，于是不再回炉，直接 finish。注意 passed=False——图正常收敛了，只是结果是「不合格」。「收敛」不等于「成功」：图保证的是「一定会停下来并给出一个明确结局」，而不是「结局一定好」。把「失败」也建模成一个正常出口，比让它卡死或抛异常健康得多。
12.2.6 防死循环的三道闸
循环边最危险，多花点篇幅说清怎么防它跑飞。三道闸，从内到外：
| 闸 | 位置 | 作用 |
|---|---|---|
| 回炉条件里的上限 | 边本身 | 次数到了就不回炉，转出口 |
max_waves | 引擎 run | 波数到顶直接 raise，最后的保险丝 |
| 图结构审查 | 建图时 | 每条环都必须有一条到出口的路径 |
第一道是业务层的上限：attempts < MAX_ATTEMPTS。它最贴合语义，是主力。第二道是引擎层的兜底：max_waves 到了就抛错，专治「条件写漏、环没有出口」这类编码事故——它不该在正常流程里触发，一旦触发就是 bug 信号。第三道是设计层的纪律：建图时逐条环检查「它有没有出口」，这靠人和 code review，引擎帮不上。
还要警惕一种隐蔽的死循环：环里状态不变。如果 generate 每次产出一样的内容，validate 每次结论一样，那么即便 attempts 在涨，图也只是空转够次数才退出——次数上限能止损，但已经浪费了 N 次模型调用。判断环是否「有效」的标准不是「它会不会停」，而是「每转一圈状态有没有向出口靠近」。如果某条环的每一轮都不改变任何影响出口的字段，那它就不是重试，是空转。
12.2.7 真实模型下会怎样
本节两个 demo 都是假模型，generate 的两次输出写死在脚本里。真实 Claude 下，generate 每次产出都不同，于是环的圈数不再可预测：可能第一次就合格（0 次回炉），也可能要试满上限。这带来一个真实工程里的取舍——上限设多少。设太小，合格的草稿还没生成就放弃了；设太大，一次失败的请求要烧掉 N 次模型调用。经验做法是让上限随任务难度浮动（简单格式化任务 1～2 次，开放式写作 3～5 次），并给每次回炉带上失败原因（把 validate 的 reason 塞回下一轮的提示词），让模型知道「上一版哪里不合格」。
还有一点：本节 validate 是纯代码的确定性校验（数长度），真实系统里校验往往也要模型参与（判断是否引用了不存在的来源、是否夹带了敏感信息）。这时 validate 节点自己就成了一个不确定的组件，它的「合格/不合格」判断可能出错——所以对高风险的校验，最稳的出口不是自动 finish，而是转人工。这正是下一节 12.3 人类在环
的主题。
阅读导航：上一节：12.1 状态机式编排 · 下一节：12.3 人类在环 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。