Skip to content

第 10 章 编排模式 ​

这一章解决控制权放在哪的问题:多智能体系统里每分派一次就是做一次选择,下一步交给谁、带什么上下文进它的窗口、干完怎么收,这些选择既能由工程师在代码里事先定死,也能留给模型在运行时当场决定,编排模式(orchestration,多个智能体分工协作的组织方式与控制结构)就是在这两端之间选一个位置。10.1 画这条线,一端显式控制,一端涌现控制,各种资料里五花八门的模式名都归到线上;10.2 与 10.3 沿线讲七种模式,工作流四式加自主三式,每一种的结构、适用场景和代价;10.4 讲终止与人类委派两个横切话题,它们不属于任何单独一种模式,却左右每一种模式能不能安全上线;10.5 不借框架,拿原生 SDK 手写监督者编排,完整跑通一条中文场景流水线。读完这一章,框架文档里的宣传词无论是路由、状态图、群聊还是交接,都能在这条线上找到位置,看清它把控制权放在哪一端、要你付什么代价。

10.1 显式控制与涌现控制 ​

编排模式的核心是一个决定,下一步由谁来定。把常见模式排到同一条线上:一头是工作流模式,执行路径跟控制流都是开发者事先定义的;另一头是自主模式,模型在运行时边推理边通信,现场决定下一步。拿两个做同一件事的团队看最清楚,同样是给社区评论区做自动审核,A 团队把流程写死在代码里,先查广告,再查辱骂,最后查低俗,顺序固定,任何一条评论进来都走同一条路;B 团队只写了一个循环,每步查什么让模型当场判断,广告嫌疑重的就先查广告,语义含糊的就多查一轮。两边用同样的模型、同样的三个审核智能体,两个系统跑出来的行为却大不相同,差别跟智能体数量、模型强弱都没关系,只在下一步由谁来定这一件事上。这两头分别叫显式控制(explicit control)与涌现控制(emergent control,emergent 指系统行为在运行时生成,不预先写定;第 9 章讲的涌现,在控制权这条线上就属于这一头)。另有资料把这两头称作 Agent 流程链与自主 Agent 协作,说的还是同一条线。

各家厂商给模式的叫法差别很大,可背后说的其实只有一件事,自主度高低。还要分清一层,这一章说的“模式”指的不是软件工程里的设计模式:ReAct、计划式(Plan-and-Execute)这些编排模式,形态上是提示策略与工作流的形状,跟 GoF 设计模式那套带规范参与者、失败模式定义的词汇是两回事,两套词分属不同层次,混着用会乱,第 19 章会把这层对照正面讲清楚。

两端都是极端的纯粹形态,生产系统大多在中间某处。分支条件写进代码的条件式工作流靠左;拓扑固定成星型、路由判断交给模型的监督者式居中;交接与对话驱动把话语权逐步让给模型,位置逐步偏右。第 9 章的升级阶梯回答的是要不要升级,这条线回答的则是升级之后控制权给多少。阶梯第三级的工作流整体贴着左端,第四级的自主多智能体贴着右端,生产系统最常用的位置反而是中间地带。

图 10-1 控制权分布,七种模式的位置与别名归位,左端显式控制,右端涌现控制

工作流模式贴着这条线的左端,描述它的惯用工具是图。节点代表一个计算单元,可以是一个函数、一个智能体,甚至可以是一整个智能体团队;边代表控制流,规定两个节点的先后,A 的输出变成 B 的输入。边上还能挂条件,条件满足了转移才发生,条件的依据范围很宽,上一步输出、共享状态、外部信号和人工输入,或者几者的组合都行。图建模的好处有三个。执行路径能在运行前验证,智能体之间的关系能画出来给人看,执行行为是确定的。

这套画图语言不是新东西,传统工作流引擎已经用了很多年。Airflow、Temporal、Prefect 这类引擎就靠有向无环图(DAG)管依赖与状态同步,到今天仍是实现工作流模式的可靠基础。

图 10-2 控制词汇族对照,图、有限状态机、行为树各画一个微例

图这门语言再往前翻两代,是两套更老的词汇,读框架文档时经常撞见。一套是有限状态机(Finite State Machine,FSM)。系统任一时刻处在有限个状态之一,事件触发转移,守卫在转移前做一道检查,决定这条边走不走得通,动作在转移中干活,终止条件负责收尾。编译器和网络协议用它管了半个世纪。状态一多就得分层,超状态把一组子状态连同公共规则打包,并行区域让几组子状态各自推进,历史节点记住离开时停在哪个子状态,回来从那里接着走。LangGraph 的子图与检查点,谱系上正对着超状态与历史节点。另一套是行为树(behavior tree),机器人与游戏行业用了几十年。选择器、序列、条件、动作、装饰器、并行六类节点搭成一棵树,每个节拍(tick)从根往下行一遍,节点各自回报成功、失败或运行中,选择器见一个成功就收,序列见一个失败就停。它的长处是模块可拼装、可复用,LLM 编排很少直接用它,但按节点回执成败、按树形组织流程的思路,在各家框架里都有沿用。

把智能体放进来,有四件事跟着变了。推理不再靠规则,改用自然语言;规划能依据中间结果动态调整;输出是概率性的,错误检测需要新思路;成本按 token 计,通信开销随对话长度往上涨。四条差异越往线的右端走越明显。

资料里常把多智能体架构归成五类,Router、Parallel、Sequential、Circular、Dynamic。Sequential 与 Parallel 是最典型的左端模式;Router 不过是监督者的别名,位置居中。Circular 把顺序式的末端接回起点,成了一条反馈环,这条环仍属开发者预先画好的边,位置靠左;本书不把它单列一式,而是归进条件式,10.2 讲条件式的例子时会碰到它。Dynamic 站在最右,所有成员全互连、自由协商。

还有一套说法把协作分作三种,三种也都能在这条线上找到位置。共享思考链指的是多个智能体读写同一份消息草稿;中介者说穿了就是星型拓扑下的监督者;分层团队则是监督者一层套一层的组合。

资料里常提起的还有黑板(blackboard)这套拓扑。知识源智能体彼此不直接对话,把带类型、带版本、带置信度与时间戳的事实和假设都贴到中央板子上;一个控制器管住发布、评估、整合三个阶段,直到解法收敛。医疗诊断就是它的典型场景。症状分析智能体把“发热且有皮疹”发上板,皮肤科智能体一见皮疹就接手,回一条标 0.8 置信度的假设;病毒科智能体读到假设,再发一条血检请求;控制器持续评估,直到最终诊断的置信度足够高才合成出来。它在这条线上的位置在中间,谁接手由板子上的状态决定,控制器只负责把好阶段与收敛这道关。

这块板子最实用的性质是只增不改。历史因此天然留着一条可审计的轨迹,谁在何时贡献了什么、置信度是多少,事后都查得清。代价同样集中在中央。写入与评估都要经过同一个点,比直接消息传递慢,板子一旦没有分片或索引,控制器就先一步变成吞吐瓶颈。陈旧与失效的事实还要靠一套清理机制处理,清理不真删,靠标记失效或归档实现,审计轨迹不被破坏;不清理的话,噪声越积越多,拖累的是读板的智能体。

板子这个思路还有个更去中心化的变体,蜂群架构的任务板(task board)。任务发布到共享板上,谁的手空了谁拉走一件,干完把状态更新掉,流程里的下一个专职智能体接着拉。黑板好歹留着一个管阶段的控制器,任务板连这个都不留,派活全靠任务状态的流转;第 11 章的合同网是同一类问题的另一条解法,靠投标竞标,不靠先到先得。

共享状态不一定上黑板,存进数据库也是一条路。共享表里存的是结构化事实与语义嵌入,外加智能体之间的承诺(commitment),这类承诺指锁定资源、记录任务进度之类的约定。更新靠数据库的原子性兜底,一步要么全成、要么全不动,多个智能体同时改同一份共享状态也产生不了竞态条件。

资料里还有一族事件驱动的说法也常一起撞见,靠外部信号触发下一步,位置散布在靠左到中间这一段。边上挂外部信号的条件式是它的最简形态,黑板与任务板算大一号的样本。谁接手仍由板上的状态决定,只是触发器换成了事件,自主度并不跟着涨。

拿到任何外部材料,第一步先找它在这条线上的哪个位置,术语上的差异就不会再来添乱。

位置模式下一步谁定
左端顺序式、并行式代码里的图定死,运行前可验证
靠左条件式、Circular图仍定死,边带条件或反馈环
中间监督者(Router、中介者同位)拓扑固定,路由判断交模型
中右计划式模型出计划,步骤按计划走
右端交接、对话驱动(共享思考链、Dynamic 更在外沿)模型运行时现场定

这条线上还藏了一条容易被漏掉的原则。节点看着是个整体,里面却能再装一整套多智能体系统。计划式里的编码智能体,内部可能走着检索、编码、测试三步顺序工作流;并行工作流里那个法务审核节点,内部说不定又是一组各读各的、并发跑的子智能体。对外,节点只露一个入口,内部怎么对话、怎么编排,一概不外泄。

分层组合的价值正在于此。每层都能挑自己合适的模式,顶层只看摘要,底层干具体活,整个系统不必绑死在同一种编排上。第 7 章的 Isolate 把子任务隔离进独立上下文、只交回摘要,8.4 的子智能体把这种隔离做成了可复用的构件,这两处都是这条原则的现成例子。

七种模式看上去纷繁,运行时干的事却只有一套,就是下面这个五步的编排器循环。

text
while 未满足终止条件:
    选出下一个智能体            # 各模式各自实现
    为它准备上下文              # 各模式各自实现
    执行该智能体
    用它的产出更新共享状态
    检查终止条件

这七种模式的差别,基本都在头两步的实现上。顺序式的“选下一个”不过让计数器加一;对话驱动的“选下一个”得做一次模型调用;监督者式的“准备上下文”,备的是一份结果摘要;共享思考链的“准备上下文”,备的是整段历史。10.5 要手写的监督者,写的也正是这个循环。

两端的失败形态刚好对称,一头一个典型样本。

左端代表性的失败是需求漂移。流程定死的那一刻,业务规则就只在代码里。评论区上线三个月,运营加了条新规,高风险账号要先查历史发言,可图上没有这条边。改一次图、发一版、回归一轮,业务变得快的时候,这套固定流程反而成了拖累。适用边界也是从这份代价里推出来的,工作流的可靠与可预测,是拿适应变化的灵活性换来的。

右端代表性的失败是成本按乘法走。串行的轮次把延迟一圈圈叠高,B 团队让模型现场判断的做法,每多判一轮,全量上下文就再收一次费,延迟与 token 消耗跟着轮数一起涨。

两种失败摆在一起,选位置选的就是接受哪一种:越靠右不代表越高级,要紧的是任务更可能栽在哪一种失败上。

越靠左越可预测,越靠右越灵活,选位置看的是任务特征,判断方法第 9 章已经给过。

10.2 工作流四式 ​

工作流这类模式工程师早就熟悉,数据管道、CI 流水线都是同类:节点换成大模型,核心概念基本没动。软件工程里现成的扩缩容、测试、重试经验,直接搬过来就能用,不用给智能体另起炉灶。

顺序式(Sequential)是四式里最省事的,A 干完把结果交给 B,B 干完再给 C,一路串到底。电商商品详情页的生成就是现成的例子。

text
选题 ──> 检索素材 ──> 撰写文案 ──> 生成配图 ──> 详情页发布

选题步骤产出的商品卖点列表,喂给检索步骤当查询输入。检索攒下的素材,成为文案撰写的依据。文案定了稿,配图才有内容可配。每步的产出都是下一步少不了的前置输入,这种天然的分段依赖,正好是顺序式的适用场景。

这一式写成代码最短,几个函数按固定次序串起来,前一步的返回值直接进下一步的参数。

python
"""示意代码:顺序式工作流的最小骨架,python3 可直接运行。"""

# 假数据:选品库里的候选商品(配图与发布两步照同样写法延长)
PRODUCTS = [
    {"title": "便携榨汁杯", "reviews": 3210},
    {"title": "保温焖烧罐", "reviews": 1580},
]

def pick_topic(products: list) -> list:
    """第一步选题:按评论量挑重点商品,产出卖点列表。"""
    top = max(products, key=lambda p: p["reviews"])
    return [f"{top['title']}热度高", "便携场景是核心卖点"]

def search_materials(selling_points: list) -> list:
    """第二步检索素材:拿卖点当查询词(桩),回一堆佐证。"""
    return [f"{point}的用户实测佐证" for point in selling_points]

def write_copy(materials: list) -> str:
    """第三步撰写文案(桩):素材在手,文案才有内容可写。"""
    return f"标题:通勤路上的新鲜果汁。正文引用素材 {len(materials)} 条。"

points = pick_topic(PRODUCTS)          # 外层把三步串成一条链,
materials = search_materials(points)   # 前一步的返回值
copy = write_copy(materials)           # 直接进下一步的参数
print(copy)

好处有两处。执行时序完全可预期,总延迟就是各步之和,很好算;错误隔离也清楚,哪一步挂了,去日志里一眼就能定位。

五步各花八秒,整条链就是四十秒。同样这五步要是互相独立、改成并发,总耗时只看最慢那一路,只有八秒。

代价也是结构性的,不能跳步,不能并行,任何一步失败整条停,最慢的一步拖住全链。什么时候不该用顺序式,判据也在这份代价里。步骤之间没有数据依赖还串着走,等来的全是白付的延迟,这是这一式里最常见的浪费。

条件式(Conditional)往边上加判断,按条件挑分支。内容审核的初筛,画出来就是这样。

text
              ┌── 风险分低 ──> 自动放行
评论 ──> 风险初筛 ─┤
              └── 风险分高 ──> 深度审核 ──┬── 通过 ──> 放行
                                        └── 不通过 ──> 转人工

代码开发是另一个例子。测试通过就进部署,测试失败就折回编码,流程折成了带反馈的环,正是 Circular 那一类;条件式也能带上这种环的形态。

条件式能用的前提是路径有限、可以枚举,写代码的时候就能把分支画完。

工程上有条纪律得守住,图归图、参数归参数,两层别混在一起。分支条件读的是上一步结构化判决里的字段,比如风险初筛回一份带风险分的 JSON,边上的条件只看分数字段,阈值放到配置里。这么一分,调参不用动图,改阈值也不用碰代码。

照这条纪律写出来,条件式的骨架长这样,边只读判决字段,阈值单独放在最上面。

python
"""示意代码:条件式工作流的最小骨架,python3 可直接运行。"""

THRESHOLD = 0.6   # 阈值放配置里,调参不用动图

def risk_screen(comment: str) -> dict:
    """风险初筛(桩):真实实现是一次模型调用,回带风险分的 JSON 判决。"""
    risky_words = ("加微信", "优惠券", "辱骂")
    score = sum(w in comment for w in risky_words) / len(risky_words)
    return {"risk_score": round(score, 2)}

def deep_review(comment: str) -> str:
    """深度审核(桩):换一份更细的提示再判一次。"""
    return "不通过" if "加微信" in comment else "通过"

def moderate(comment: str) -> str:
    """分支全在这一个函数里,路径有限、事先枚举得完。"""
    verdict = risk_screen(comment)
    if verdict["risk_score"] < THRESHOLD:      # 边一:低分自动放行
        return "自动放行"
    if deep_review(comment) == "通过":          # 边二:高分深审后放行
        return "放行"
    return "转人工"                             # 边三:深审不过

if __name__ == "__main__":
    for c in ("这鞋质量真好,推荐大家去店里试试。",
              "加微信 xxx123 领内部优惠券。"):
        print(moderate(c))

代价要等规模上来才现形。条件一多,分支组合数跟着爆炸,图就没人看得懂了;冒出新的情况,就得改代码、发新版。条件定得越死,系统越僵硬。好在条件本身不设限,上一节提过,它可以是上一步输出、共享状态、外部信号甚至人工输入的任意组合。

并行式(Parallel)管的是互相独立的子任务,图整体是有向无环图,先扇出后汇入,扇出(fan-out)把一份工作拆成多路同时下发,汇入(fan-in)再把各路结果收拢合并。

text
        ┌──> 品牌A 舆情统计 ──┐
简报任务 ─┼──> 品牌B 舆情统计 ──┼──> 汇总成稿 ──> 简报发布
        └──> 品牌C 舆情统计 ──┘

公关团队要一份三品牌舆情简报,三个品牌的统计彼此没有依赖,并发开三路去跑,总耗时看最慢的一路,而不是三路相加。

扇出与汇入写成代码,就是几路并发跑出去,汇入时带着缺口标注合回来。

python
"""示意代码:并行式的最小骨架,python3 可直接运行。"""

from concurrent.futures import ThreadPoolExecutor

# 假数据:三个品牌的舆情统计源,品牌B这一路取不回来
SOURCES = {"品牌A": ["好评", "差评", "好评"], "品牌B": None, "品牌C": ["好评"]}

def collect_brand(brand: str, posts: list | None) -> dict:
    """单路统计(桩):None 模拟这一路数据源超时。"""
    if posts is None:
        raise TimeoutError(f"{brand} 数据源超时")
    return {"brand": brand, "negative": posts.count("差评"),
            "total": len(posts)}

def build_brief() -> str:
    """扇出几路并发,汇入时带着缺口标注合并,简报照常出。"""
    with ThreadPoolExecutor(max_workers=3) as pool:            # 扇出
        futures = {b: pool.submit(collect_brand, b, posts)
                   for b, posts in SOURCES.items()}
        sections = []
        for brand, fut in futures.items():                     # 汇入
            try:
                r = fut.result(timeout=5)
                sections.append(f"{brand} 负面 {r['negative']}/{r['total']}")
            except TimeoutError:                               # 降级支路
                sections.append(f"{brand} 数据缺失,本期刊载从略")
    return "三品牌舆情简报:" + ";".join(sections)

if __name__ == "__main__":
    print(build_brief())

并行式什么时候划算,条件摆得很明白,子任务彼此独立,谁也不跟谁共享副作用。代价全堆在汇入这一头,一是所有分支收齐才能合并,部分失败的分支得预先备好补跑或降级;二是合并的 token 开销,三路各回五千 token,汇入这一轮输入就是一万五,第 7 章说的注意力预算在这里照样管用。

补跑与降级就是把两个既有机制拼起来。补跑靠的是步级重试,开源多智能体框架 PicoAgents 给每个步骤都配了重试上限,失败就在原地重试,超过上限才算真失败。本章后文提到的“参考实现”,说的都是它。

降级交给条件边去办。失败或质量不过线的分支,引到降级支路;品牌 B 的统计要是实在取不出来,汇入时就带着缺口标注一起合并,简报照常发,缺的那块注明数据缺失,好过整份简报一起出不来。

同一道题发给两个模型并行跑,还会冒出一类汇入独有的失败,两路都正常开工,一路按时回了,另一路超时,汇入方只等到一半。这不算哪一路的临时故障,是汇合这个环节自己的结构性失败,重试机制处理不了。得另想办法,给汇合设一个等待上限,到点凑不齐,就把已到的那路带上质量标记先发出去,好过什么都不返回;缺席的那路单独放进隔离区,留给运维排查和事后重放。两路结果怎么合,取优、拼成一体、还是拿一路兜底,做成运行时可换的策略,汇入的骨架一行不用改。

汇入还有一条投票的路可走。几路智能体对同一个问题各交一份答案,简单多数票、加权票,或者让一个裁判智能体挑最终结果,单个智能体的偶发错误与偏见压不过票数,诊断与评估这类任务常用它。失效模式也一目了然,多数派会自信地一起错。几路智能体共享同一种偏见、同读了一段错误材料时,票数越集中,错误越难被发现。解法是让参与者本身有多样性。换不同厂商的模型,训练数据、对齐与推理风格一并跟着换,多样性最大,成本也最高。退一档,用不同的提示立场,一个乐观、一个怀疑,便宜但效果弱一截。再有一条,是让各路去读不同的证据子集。

这种一起错,可靠性工程里有个现成的名字,叫共因失效(common cause failure)。冗余的本意是互相备份,一路坏了,剩下的照样能干活;可几路要是共享同一个失效原因,就会一起坏。投票恰恰容易栽在这里。同一个模型的几个实例,带着相似的上下文进来,做着相似的选择,意见自然往一处偏。几路审查看上去是几票,实际算一票。所以数票之前得先问一句独立不独立,这几份意见是不是真的各走各的路。对策是把几路真正拆开,前面那几条多样性做法,针对的正是这个。

投票还有个多轮化的扩展,叫共识迭代,各路把答案广播出去,读了别人的再调整自己那份,连着几轮的变动都在容差之内就停。上面一轮定胜负的多数票,算是它最简的形态。

往质量那边再走一步,汇入这一头还有三个骨架,分别管竞争、辩论、评测。竞争式(competitive)最直白,同一道题原样发给几个智能体,各自独立作答,谁也不看谁,收齐后按事先写定的标准挑一份最好的。两个代码智能体各试一种优化思路,一个比较智能体评完选效率高的那种,就是这一式的典型场景。它跟第 3 章的自洽性是同一招的两种规模,自洽性在单智能体里多采样几次、按人头取答案,竞争式把每次采样升级成一次带工具的完整运行,判据也从“取多数”换成“按标准择优”。多数派错得整齐的时候,最好的答案可能恰恰在少数派手里,投票会漏掉它,竞争式还能把它挑出来。

竞争式的骨架更短,同一道题发几路,收齐后按事先写定的标准挑一份。

python
"""示意代码:竞争式的最小骨架,python3 可直接运行。"""

# 两个代码智能体各试一种优化思路,各自独立作答,谁也不看谁(桩)
SOLUTIONS = [
    {"author": "方案一", "approach": "缓存查询结果", "runtime_ms": 40},
    {"author": "方案二", "approach": "批量取数", "runtime_ms": 85},
]

def judge(solutions: list) -> dict:
    """比较智能体(桩):真实实现读代码跑基准,择优标准写死在代码里。"""
    return min(solutions, key=lambda s: s["runtime_ms"])   # 运行时间越短越好

best = judge(SOLUTIONS)
print(f"采信 {best['author']}({best['approach']}),运行 {best['runtime_ms']} 毫秒")

辩论式(debate)给结论会打架的场合用。正方反方各领一个立场,各自举证,还读得到对方上一轮说过什么,专门挑着反驳。辩满几轮,一个中立裁判读完整记录,出一份结构化裁决,写清哪方论据成立、信息缺在哪、最终采信什么。它跟投票分工明确,投票换置信度,辩论换结论质量。事实有争议、两边都有说得通的论据时,辩论比投票顶用,票数压不过论证,少数派手里的正确意见,在投票里拗不过票数,在辩论里还有机会把对方驳倒。裁判要读全量辩词,token 开销比投票贵;9.3 提过的“多智能体多轮辩论能改进推理与事实性”那条研究结论,说的正是这一式。

这一式写成代码分两段,正反方带着立场辩几轮,中立裁判最后出结构化裁决。

python
"""示意代码:辩论式的最小骨架,python3 可直接运行。"""

import json

# 场景:赔付纠纷的责任辩论,正反方各带立场(发言内容都是桩)
DEBATE_ROUNDS = [
    ("正方-商家", "承诺发货日是 12 日,揽收记录落在 14 日,货交晚了两天。"),
    ("反方-承运", "揽收后滞留三天,物流轨迹为证,慢在运输环节。"),
    ("正方-商家", "滞留发生在交货之后,用户等的天数大头在发货延迟上。"),
]

def judge(transcript: list) -> dict:
    """中立裁判(桩):读完整记录,出结构化裁决,写清采信什么、缺什么。"""
    return {"adopted": "正方-商家",
            "reason": "交货延迟在前,主因在商家",
            "missing": "承运方内部调度记录未取得"}

def run_debate(rounds: list) -> str:
    """辩满几轮收工,每一轮都读得到对方上一轮说过什么。"""
    transcript = []
    for side, point in rounds:
        transcript.append(f"[{side}] {point}")
    return json.dumps(judge(transcript), ensure_ascii=False)

if __name__ == "__main__":
    print(run_debate(DEBATE_ROUNDS))

评测者循环(evaluator loop)管产出要打磨的活,一个智能体创作,另一个拿结构化评分卡挑刺,不达标就把意见带回去重写,循环到达标为止,写作、代码、报告都适用。它跟第 3 章的反思不一样。反思是同一个智能体自查,自己挑自己的毛病常常下不去手;拆成两个角色,创作者和评审者各一份提示、各一份评分标准,评审只管按卡打分,反而挑得动。评审者自己怎么校准,12.3 讲裁判偏差的那套办法直接搬过来用。

评测者循环的骨架是一个创作一个评审,评分卡给结构化分,不达标带着意见重写。

python
"""示意代码:评测者循环的最小骨架,python3 可直接运行。"""

# 场景:店铺周报成稿,创作者与评审者各一份提示(两处模型调用都是桩)

def create_draft(feedback: str | None) -> str:
    """创作者(桩):首轮凭数据写稿,重写轮带着评审意见改。"""
    if feedback is None:
        return "本周营收 320 万,环比上升。"
    return "本周营收 320 万(环比 +8%,基期 296 万),同店口径。"

def review_draft(draft: str) -> dict:
    """评审者(桩):按评分卡逐项打分,先理由后分数。"""
    passed = ("环比" in draft and "基期" in draft and "口径" in draft)
    return {"score": 9 if passed else 6,
            "feedback": None if passed else "缺基期数字与统计口径,补齐再交。"}

def iterate(max_rounds: int = 3) -> str:
    """循环到达标为止,轮数上限兜底,防评审永不满足。"""
    draft = ""
    feedback = None
    for _ in range(max_rounds):
        draft = create_draft(feedback)
        verdict = review_draft(draft)
        if verdict["score"] >= 8:
            return draft
        feedback = verdict["feedback"]
    return draft                 # 上限耗尽,交出当前最好的一版

if __name__ == "__main__":
    print(iterate())

三个骨架没有一个是便宜的,同一份活跑几遍、辩几轮、改几稿,调用都按倍数翻,9.3 第一道闸那句“动手之前先把成本算清楚”,在这里照样成立。适用判据也一致,产出错了代价高、改对了收益大的地方才值得上,诊断结论、对外公告、要负责任的文书都在这一档;日常批量产出,一次调用配一个好提示,多半够了。

比汇入更早的关口在扇出之前。什么都不先约定好就扇出去,各路智能体对目标各有一套理解,各自做假设,用互不兼容的办法去解决重叠的问题,事后光解决冲突花掉的时间,就比并行省下的还多。解法叫方案先行。站在中央拆活派活的编排智能体,先花约 10,000 token 造一份方案,再放智能体出去并行。

方案里写四件事。任务边界,哪个智能体负责哪些文件;接口契约,哪些函数与 API 必须兼容;依赖顺序,什么必须在什么之前完成;验收标准,每个智能体怎么知道自己做完了。跑偏返工的团队要烧 500,000 token 甚至更多,相比之下方案先行的 10,000 token 花得很值。

给个真实例子。上线前给博客做 QA,领队把任务劈成五路,页面响应、博文渲染、链接完整性、SEO 元数据和可访问性各归一个智能体,全跑在更便宜的模型上,直接用 curl 抓页面、解析 HTML。五路都完工后,领队把发现汇总成一份按严重程度排序的报告。从提示词到最终报告大约三分钟,五个智能体测完 146 个以上 URL、83 篇博文;成本比单会话顺序干同一件事要高,墙钟时间却只剩后者的零头。

并行流是并行式的另一个叫法。它跟层级协调在图上长得很像,怎么分只看一条判据,扇出和汇入这对原语,能不能顶替掉编排智能体。拆的是同质任务、汇入只做简单聚合,原语就够用,这叫并行流。要给各路分别下指令、盯进度、调和互相冲突的结果,原语干不了,只能算层级协调,下一式的监督者就是它的典型形态。

监督者式(Supervisor,有的资料叫它中介者,本书统一称监督者)算条件式的一个变体,把分支判断收拢进一个中央控制节点。它对每个进来的请求做评估,路由给合适的专门智能体,结果回流后再定下一步。拓扑固定成星型,一个中心连多个分支。路由判断能写成规则,也能交给模型,一旦交给模型,它就挪到了中间位置。

电商客服就是这类形态的标准样本。退款、物流、售后咨询各有各的专门智能体,监督者按问题类型把人分过去。Router 模式是生产环境里出现最多的多智能体形态,预订、查价、退款这类需求花样多的聊天产品,大多长这个样。

它适合任务类型多、又能归成几类的场景。代价一样明摆着。中央节点既是吞吐瓶颈,也是单点;监督者路由的准确率就是整个系统的上限,它一旦判错,后面的专门智能体再强也白搭。

监督者路由还能把成本一起算进去,派活的时候连模型档位一起派,查物流、对规则这类轻活落到便宜的快模型,纠纷调解这种难啃的才上旗舰。路由这次调用本身也一样,输入短、输出就一个名字,用低档模型跑足够。成本进了路由的判断,就不用等费用单出来再补救。

这一族还挂着另一个名字,叫协调者-工人-委派者(CWD)模式。三个角色分得很清,协调者盯大局,委派者管拆活、派活和收拢成果,工人埋头执行。合起来看,协调者加委派者做的正是监督者那些事;任务在智能体间转手的那一下,就是 10.3 要讲的交接。

四式在控制流、自主度、开发者控制、实现复杂度上怎么对照,见下表。

模式控制流智能体自主度开发者控制实现复杂度
顺序式线性 A 到 B 到 C低高低
条件式条件分支路由低高低到中
并行式并发扇出与汇入低高中
监督者式中央节点路由中中中

四式怎么挑,任务特征直接给出对应关系。步骤之间分段依赖,前一步的产出是后一步的输入,用顺序式;路径有限、事先枚举得完,用条件式;子任务彼此独立、互不共享副作用,用并行式;任务类型多、又能归成有限的几类,用监督者式。还有一条分界,监督者是每个请求进来挑一次角色,挑完各干各的;计划式是每个任务先拆一份计划再照着走。请求要是连归类都归不进去,每个任务都得单独拆计划,工作流四式就到了边界,再往右是 10.3 的自主三式。

工作流模式靠三个前提撑着,缺一个都不稳。解法是已知的,任务拆成哪几步,事先全定下来;资源得够,有时间和专业人力把操作序列做出来;任务还得是静态的,运行中不需要临时调整。三条都成立,工作流才给得出最高的可预测性,每一步都能单独测试。只要有一条不成立,固定流水线给不出需要的灵活性,就该把目光往右端挪了。

10.3 自主三式 ​

路径事先列不全时,“下一步派谁”只能交给模型现场定,交出去有三种方式,自主程度一种比一种高。电商大促的客服分流就是典型:活动规则三天两头在变,客诉类型越来越多,退款纠纷、价保争议、物流破损和恶意下单,事先列不全分支,条件式连那张图都画不出。

计划式(Plan-and-Execute)在自主三式里结构感最强。先由一个编排智能体把任务整体规划成一份执行计划,计划写明分派与依赖,然后照着计划逐步执行、逐步验收。列任务清单、派活、盯进度、验收,全让这一个角色包了。

依赖这两个字,要写成程序能读的格式,每个任务带上 blocks 与 blockedBy 字段,写清它挡着谁、被谁挡着,执行器拿到计划就不用整条排成串行。互不依赖的任务同时放行,有依赖的排到前序完成之后,整份计划一波一波往前推,一波收完再放下一波。计划写得再长,并行度不用人操心,从字段里读得出来。

blocks 与 blockedBy 写成代码,执行器从字段里读出波次,整份计划自己往前推。

python
"""示意代码:计划式的最小骨架,python3 可直接运行。"""

# 场景:店铺运营周报。计划由编排智能体产出(桩),依赖全写在字段里
PLAN = [
    {"id": "fetch_a", "task": "抓取平台A销售数据", "blocks": ["analyze"], "blockedBy": []},
    {"id": "fetch_b", "task": "抓取平台B销售数据", "blocks": ["analyze"], "blockedBy": []},
    {"id": "analyze", "task": "同环比分析", "blocks": ["draft"], "blockedBy": ["fetch_a", "fetch_b"]},
    {"id": "draft", "task": "套模板成稿", "blocks": [], "blockedBy": ["analyze"]},
]

def execute(task: str) -> str:
    """执行一件活(桩):真实实现里各节点挂各的模型调用或工具。"""
    return f"{task}完成"

def run_plan(plan: list) -> list:
    """波次执行:互不依赖的同时放行,有依赖的排到前序完成之后。"""
    done, log = set(), []
    while plan:
        wave = [s for s in plan
                if all(dep in done for dep in s["blockedBy"])]
        if not wave:             # 依赖成环,计划本身有毛病,停在那里
            raise ValueError("计划里有互相等待的步骤")
        label = " / ".join(s["task"] for s in wave)
        print(label + ("(同一波,可并行)" if len(wave) > 1 else ""))
        for step in wave:
            log.append(execute(step["task"]))
            done.add(step["id"])
        plan = [s for s in plan if s["id"] not in done]
    return log

if __name__ == "__main__":
    run_plan(PLAN)

这个形态的知名实现有微软的 Magentic-One。一个编排智能体带四个智能体干活,靠两张清单驱动,任务清单记要做什么,进度清单记做到哪了。它曾在 GAIA 榜单登顶(GAIA 是一个公开基准,测推理、工具使用与多步执行,12.4 展开)。同一形态的另一个样本是 Anthropic 的多智能体研究系统,主智能体按问题复杂度动态开出子智能体,最少一个,最多十多个。官方工程博客给过一组数字,主模型带子智能体协作的系统,在内部研究评测上比单智能体方案提升了 90.2%。

这 90.2% 只是单一内部评测上的对比,看个方向就好,别当成普适常数。

机制分三个环节。计划生成这一步,得靠结构化输出兜底,第 3 章的 JSON Schema 约束到这里就不是可选项了,自由文本写出来的计划根本进不了程序。步骤验收同样不能省,每做完一步就让模型对产出给一份结构化判决,写清楚这一步成没成、失败原因、置信度和改进建议,判决进得了程序,重试才知道该改哪。失败之后的重试,也不是把原样再跑一遍,而是把失败分析与改进建议塞进下一轮的上下文,带着反馈重跑,命中率比盲目重跑高出一截。

图 10-3 语义反向传播,文本反馈从失败点沿执行图往回流

带着反馈重跑,还有一个容易漏掉的细节,反馈该记在谁头上。多智能体系统里最磨人的事故不是哪一步失败,而是改好了 A、弄坏了 B,对一个节点明明正确的修复,顺着数据流波及下游,出事的位置离动手的位置隔好几个节点。研究界给这类问题起过一个借来的名字,语义反向传播(semantic backpropagation),把文本反馈当梯度用,让它从失败点沿执行图往回流,流到哪个节点,哪个节点照着改。名字借的是神经网络的反向传播,搬走的是“错误信号沿计算路径回传”这个思路,传的不是数字,是一段写清楚哪里错、该怎么改的文字。

光往回流还不够,归因得准,这一步叫邻居感知。给某个节点写反馈,除了它自己的输出和下游报上来的错误,还得把下游的其他输入一并放进反馈里,不然会张冠李戴。拿门店周报的流水线说,抓取节点从报表里抽出“本月营收 320 万”,换算节点把它折成同比数字,校验节点发现增长率对不上,报了错。报错只点了增长率,增长率的第一来源是抓取出的营收数,反馈就顺着这条线找到了抓取节点。给抓取节点的反馈要是只写“你抽的数导致增长率错了”,它多半会得出“下次抽个别的数”的结论,修错了地方;增长率错在换算节点拿错了基期,而校验节点之所以认得出基期不对,靠的是日期节点喂给它的另一路输入。把这几路邻居的输出一起写进反馈,责任才落到换算节点头上,抓取节点原样保留。这套做法有工程门槛,反馈要沿图回流,前提是执行图的依赖关系先显式记下来,节点的输入输出都留得下档,现成工具多数还停在研究原型,生产里通常得自己搭个最小版本。

这套验收加改计划的活,业界有现成的角色名。讲 Plan-and-Execute 的资料常把它拆成规划器、执行器、重规划器三件,重规划器(Replanner)收执行器的反馈,动态调整剩下的计划,前面讲的步骤验收、Magentic-One 两张清单的更新,干的正是这个角色的活。

计划式最完整的公开参照物,是各家厂商都在做的深度研究。6.3 从检索那头看过它的两级编排,这里换到循环这一头,把实现机制拆成四件事。

头一件,把计划与状态搬出智能体,住进循环外的代码里。研究目标、已得结论、查过的来源、待追问的问题和每个子课题的进度,全装进两份结构化对象,循环每转一轮读一次、写一次。窗口装不下的历史靠对象装,智能体下一轮会忘的细节靠对象记,窗口里只放当前要用的那份摘要,要用原文再取。

第二件,让每一轮的产出可解析。模型转完一轮,交一份结构化报告,本轮发现了什么、下一步想查什么、目标算不算达成,全写成固定字段。循环控制器读字段做决定,续跑、换题还是收尾。第 3 章的结构化输出到这里是刚需,自由文本的报告进不了程序,循环等于失控。

第三件,终止闸门分层摆。硬上限与预算是保底层,缺一不可;模型自评目标达成是理想出口,可自评有偏,要紧的场合再配一个独立判断的角色;顶上加一道停滞检测,连续几轮产出几乎不变就停,防的是原地打转。这几层正好对上 10.4 的两族终止条件,多出来的只有停滞这一层,它的阈值做法见 14.5。

第四件,收尾把探索与整理分开。循环攒下的是一堆快照式的发现,彼此可能重复、可能打架,收尾让一个专门的综合角色读全量状态,出一篇成形的报告。两件事别挤在一个智能体里,边探索边写报告的智能体,探索质量会往下掉;综合角色只跑一次,不在循环里,可以放心把全部发现都喂给它。

代价也明摆着:编排智能体既是单点,也是瓶颈,它推理的天花板就是整个系统的天花板。翻费用记录的话,每一步至少两次模型调用,规划、执行和验收各自计费,它的调用成本在七种模式里排得靠前。

计划式让模型临场出计划,经典人工智能还留了一条不劳模型出计划的路。经典规划算法里排得上号的名字有一长串。STRIPS 拿逻辑谓词给状态和动作建模,条件只有真和假两种取值,可自然语言里“部分理解了”“大致可以接受”这类状态写不成二元谓词,拿来管 LLM 就太僵;HTN 把任务层层分解成子任务再排执行,跟语言模型拆解任务的方式合拍,是老方法里最能用的一个。介于两者之间的还有一条 GOAP(Goal-Oriented Action Planning,目标导向行动规划),每个动作声明自己的前置条件与效果,规划器拿这些声明做搜索,自动求出一条从现状通到目标的执行序列,顺序既不用人手写,也不用模型现编。落到智能体框架里,前置与效果往往就是每个智能体声明的输入和输出,目标就是整个系统要交的产出。代价藏在搜索算法的天性里,它按最短路径求解,结构上不允许环,反思、迭代改进这类要绕回来的流程塞不进去,得在框架外另搭。

交接(Handoff)不设中央节点,智能体之间平级互派。每个智能体只认得一部分同伴,凭自己对任务的局部判断,把控制权和相关状态显式交给下一位。看客服场景最清楚。一线客服办不了的就转技术专员,专员搞不定再转高级工程师,工程师处理完还能交回一线收尾。

实现起来,这一式省事:智能体互为工具,交接不过是一次工具调用,第 4 章讲透的函数调用机制原样就能用。一个智能体眼里的同事,就是它工具清单里的一个函数。

交接时递什么过去,要先经过筛选。新智能体接手时默认能看到整段对话历史,历史里多半有它不认识的工具、用不上的中间结果,原样继承,这些内容会一路占着它的窗口。第 7 章的压实讲的是同一份上下文里的通则,跨智能体交接是它最该用上的场景。在交接处放一道输入过滤器,把前任的工具调用记录清掉,或者只保留最近几条消息,新智能体拿到的是手头的活,不是前任的全部过程记录。10.5 的监督者只读结论摘要、不读工人推理,用的是同一个道理。

交接写成代码,智能体互为工具,交接就是一次函数调用,入口处先过那道输入过滤器。

python
"""示意代码:交接式的最小骨架,python3 可直接运行。"""

MAX_HANDOFFS = 2   # 防循环:来回转手的次数上限

def specialist(clean: dict) -> dict:
    """技术专员(桩):拿到过滤后的上下文,回结论与处置。"""
    return {"issue": clean["issue"],
            "resolution": "接口限流,已提工单给平台侧,两小时内恢复。"}

def handoff(session: dict, receiver) -> dict:
    """交接本体:一次函数调用,入口先过滤前任的过程记录。"""
    session["handoffs"] += 1
    if session["handoffs"] > MAX_HANDOFFS:
        session["resolution"] = "转手次数超限,升级人工处理。"
        return session
    clean = {"issue": session["issue"],       # 只递手头的活
             "verified": session["verified"]}  # 与已核实的事实
    return receiver(clean)                    # 前任的过程记录不跟着走

if __name__ == "__main__":
    session = {"issue": "用户反馈下单接口报错",
               "verified": ["订单 8899001 状态正常"],
               "trace": "先查订单,又试下单接口(一线的过程记录)",
               "handoffs": 0}
    result = handoff(session, specialist)
    print(f"结论:{result['resolution']}")
    print(f"专员上下文里的字段:{sorted(result)}")   # trace 不在其中

它适用的场景,是领域边界清楚,什么情况该转给谁,能说得明白。

风险出在没人看全局。两个智能体可能互相推,也可能把任务来回转包,谁都不愿意接。防循环的计数与上限,这一式照样不能省。

对话驱动(Conversation-Driven,也叫群聊模式)把自主度再抬高一级。全体智能体共用同一份对话,编排就是在轮流发言的过程里自然形成的,没有预定路径,也没有结构化交接。

它为什么管用?模型本来就按对话轮训练,读历史、接下一句是它的本行。编排机制说穿了,就是大家都往同一个消息列表里追加内容。

最简的形态里,一位智能体发言一次就对应一次 LLM 调用。它把共享对话史和自己的系统提示读一遍,产出一句发言,编排一轮正好等于一次调用。要是带工具的智能体,发言内部还套着自己的工具循环。

当参与者只剩一位、手上又握着工具时,这个模式就退回第 3 章的 ReAct 循环,共享对话史成了那条推理与行动交替的消息流;第 3 章讲过的 Reflexion 走的也是这条路,反思也只是往这份历史里再追加一条文字。这条线在这里闭合,单智能体循环是群聊的最小特例,群聊是它的多人展开。

轮次还有一组现成定法。固定轮询照名单转圈,实现只要一个下标;AI 选人则是让一个模型读当前对话,挑下一位最该发言的。AutoGen 把做法收得很全,群聊发言者选择内置 auto、manual、random、round_robin 四种选项,还留了自定义函数的接口。有个客户洞察项目就靠这个,保证研究员与每位客户交替发言,每人次数固定。

轮询这一路写成代码,一份共享消息列表加一个选人函数就是全部,AI 选人只是把选人函数换成一次模型调用。

python
"""示意代码:对话驱动的最小骨架,python3 可直接运行。"""

# 场景:售后工单三方会诊,全员共用同一份消息列表(发言内容都是桩)
SPEAKERS = {"客服": "物流轨迹显示 14 日已揽收,晚于承诺发货日。",
            "物流专员": "揽收后 48 小时无更新,卡在承运环节。",
            "售后主管": "TERMINATE"}

def next_speaker(order: list, last: str | None) -> str:
    """固定轮询照名单转圈,实现只要一个下标。"""
    if last is None:
        return order[0]
    return order[(order.index(last) + 1) % len(order)]

def run_group_chat(order: list, max_turns: int = 6) -> list:
    """一位发言一次对应一次调用,共享历史人人读得到。"""
    history, last = [], None
    for _ in range(max_turns):                 # 预算型兜底照旧
        name = next_speaker(order, last)
        history.append({"speaker": name, "text": SPEAKERS[name]})
        if "TERMINATE" in SPEAKERS[name]:      # 语义型终止,文本信号
            break
        last = name
    return history

if __name__ == "__main__":
    for m in run_group_chat(["客服", "物流专员", "售后主管"]):
        print(f"[{m['speaker']}] {m['text']}")

AutoGen 后来跟 Semantic Kernel 并进了 Microsoft Agent Framework,2026 年 4 月正式发布。发言者选择这套思路,在新框架里依旧成立。

透明是对话驱动的一个长处。模型容易顺着对话往下走,全员同一份上下文,谁说了什么都能看见;模型升级换代,协作质量跟着涨,代码一行不用改。短处也出在同一处。token 随对话长度线性往上走,上下文迟早溢出,后进窗口的发言会把早期信息顶出去。治理的办法还是第 7 章那套压实,搬进共享历史里多出一个新问题,由谁来压。各位参与者读的是同一份历史,压实就得统一做,不然各个智能体压各个智能体的,下一轮大家对不上号。常见的做法是编排层挂一道滚动摘要,历史一超阈值,就把最早的几轮缩成一段摘要,拿摘要换掉原文,新发言照常往后追加。写摘要这活交给单独一次便宜模型的调用,比让当前发言者顺带写更稳,摘要的口径就不会跟着发言者的立场走。还有一条更隐蔽,研究里提醒过,对话里共享的只有各个智能体说出口的发言,每个智能体内部推理过程没有跟着进去,所以哪怕共享同一份对话,智能体还是可能做出相互矛盾的决定。上下文共享了,产出照样可能各说各话、拼不到一起,碎片化问题不会自己消失。

两种拓扑的通信形状完全不同,画成下面这张图。

图 10-4 两种拓扑的通信开销,全互连一轮 n(n-1) 次阅读,星型一轮 2(n-1) 次

这份开销除了随对话长度涨,还随与会人数涨,对话驱动里全员共享同一份历史,一位发言,其余每个人都得读一遍,把谁读谁的关系画出来就是全互连。全员各说一句算一轮,这一轮累计 n(n-1) 次阅读;人数翻一倍,这份开销涨到原来的四倍上下,增长阶数写作 O(n²)。

监督者则把通信收拢成星型,工人只和中央节点说话,每人向中央一进一出算一轮,是 2(n-1) 次,随人数线性涨,阶数 O(n)。大 O 这个框架拿来比方案,不能算延迟常数,也算不出 token 单价,但选拓扑的时候管用。人一多,哪边的上下文先装不下,增长阶数先给出判断。

反过来看,星型省下通信,代价是全部消息都收进中央节点,监督者自己既是瓶颈又是单点。

三式样子差得远,拆回 10.1 那个编排器循环里,把各模式都绕不开的更新状态与检查终止放一边,核心只剩三步。挑下一位发言的智能体,给它备好上下文,让它执行。

轮流、AI 选人、计划式三种做法的差别,全在头两步,一步是挑人,一步是备上下文。轮流按名单取下一位的下标,AI 选人读当前对话挑最该说的那位,计划式照着任务清单派活。

拿这个循环回头看本节三式,一样对得上。对话驱动的轮次选择在第一步;交接把挑下一位的权力交给当前发言者,每次交接都是一次工具调用。

循环一旦统一,换编排策略就不用动基础设施。流式输出、取消和状态管理这类公共能力全收在编排器基类,每种模式只写自己的挑选逻辑。所有模式也因此共享一个观察视角。任取一式,看编排都从循环记录看起,选中了谁、带了什么上下文、说了什么,三栏一对齐,该调哪里立刻清楚。

失控的风险随自主度一起加码,得靠三件安全设施收住。编排器循环把最大轮数写死,给迭代上限兜底,参考实现(PicoAgents)的默认值是 50,终止条件万一失灵,这就是系统最后一道保险。终止条件本身也要组合着用,语义判断管该不该停,预算管最多能跑多久,细节放到 10.4 去讲。自主模式每步交出来的产出,也得进得了程序判断、写成结构化判决,这样验收、重试、升级交给人才有依据;自由文本的产出在这套流程里走不通。

这个基础之上还能再套一层。分层团队模式里的监督者,手下每个节点依旧是个监督者,比如一个写文章的团队,内部又挂检索、筛选、构思、撰写和检查五个角色。第 8 章子智能体讲的那套上下文隔离思想,在这里直接照搬进了编排。第 7 章说 Isolate 策略时就提过,它省下的 token 在各策略里排第一。

自主度上去以后,还有一类事故开始冒头,两个智能体的计划正面相撞。物流智能体安排货车进窄街,仓储智能体在同一时段给同一条街排了卸货;吞吐智能体想把批次直接推进审批,公平智能体把同一批扣下来要做细致核查。冲突放着不管,轻则互相等出死循环,重则各带着失效的计划跑完全程。工程做法分两步。先检测,智能体的计划在执行前先登记,两份计划动同一份资源、改同一条状态,就判互斥。再裁决,四条路按控制权从紧到松排。层级式最直接,监督者拍板,结果可预期,合规要求高的场合默认走这条。基于策略把裁决写成预定义规则,比如安全攸关的智能体永远优先于效率优化的智能体,规则放在系统外面,看得见、改得动,事后有据可查。协商让冲突双方自己谈,报价让步找出折中,适合有双赢空间的场合;交接式里两个智能体互相推活,算这类冲突的温和形态,靠计数与上限兜住。博弈论式最贵,把冲突建成一局形式化的博弈,给候选行动配收益、找稳定解,更适合在设计期算一次,不是每次冲突都算。无论走哪条,裁决都得留一条审计轨迹,把为什么这么裁写进日志,事后排查和合规审阅都指着这份记录。第 11 章讲博弈论与合同网的那一节,会把后面两条路的底细正面展开。

自主三式换来灵活,也把什么时候停这个原本该在设计期解决的问题拖进了运行时。

10.4 终止与人类委派 ​

什么时候停,是多智能体系统最要紧的环节,这一节给两族终止条件与人类委派的两条路。跑飞的场景很具体:两个智能体彼此客套,一个说信息不全请补充,另一个补完再请对方确认,来回二十几轮,每一次调用都在计费;单智能体聊进死循环已经很贵,多智能体把这份贵翻成乘法,终止条件不可靠,对话就会一直跑下去。

两族终止条件,头一族是预算型终止(budget-based termination),时间、成本和迭代次数三项资源各设上限,超了就停。参考实现把三类上限的现成终止条件都写好了,消息数、token 用量、超时,指个名就能用。

预算型的特点在于行为确定。超了上限就停,记录可审计,要向合规与成本部门交代时,这份记录就是现成的解释。它也有个先天盲区,不知道任务到底做完没有。预算烧完那一刻,工作恰好完成与刚做一半两种情形,在它眼里没有区别。预算型只管止损,不管判断。

条件之间还能互相组合,参考实现把组合重载成了运算符。竖线代表任一满足即停,与号代表全部满足才停。三个条件,20 条消息、有人说 TERMINATE、超过 5 分钟,用竖线串起来就是先到先停;两个条件,跑满 10 条消息、有人明确说出 COMPLETE,用与号串起来就是缺一不可。

研究任务里另有一组常见配置,10 轮迭代或模型判定质量过线,先到先停。组合的好处在于每个条件自己仍然简单可测,复杂的只是搭配关系,而这层关系明摆在面上,测试与审计都够得着。

实现上还有个省 token 的门道。终止检查只喂新增的消息,不反复扫全量历史,条件盯的是最近动向,检查本身也不给上下文添负担。

另一族是语义型终止(semantic termination),判定权交给模型,任务到底做完没有,由它自己说。最省事的形态是文本信号,发言里出现约定标签就停。参考实现把这做成了文本提及终止条件,只扫新增消息里的触发词。

AutoGen 体系惯用的标签是 TERMINATE。实践里也有人把用户说“谢谢”当作触发条件,智能体照这个条件生成 TERMINATE,把会话结束掉。

这些条件落到代码,每个都是一个函数,收完整历史,返回真或假。参考实现靠运算符重载组合它们,这里的代码换走一层布尔包装,语义相同。

python
import time

def max_messages(n: int):
    """消息数上限。"""
    return lambda history: len(history) >= n

def text_mention(trigger: str):
    """文本触发词,只扫上一次检查之后的新增消息。"""
    seen = 0
    def check(history: list) -> bool:
        nonlocal seen
        fresh = history[seen:]
        seen = len(history)
        return any(trigger in m["content"] for m in fresh)
    return check

def timeout(seconds: float):
    """超时上限。"""
    deadline = time.monotonic() + seconds
    return lambda history: time.monotonic() > deadline

def any_of(*conditions):
    """任一满足就停;要全部满足才停,把 any 换成 all。"""
    return lambda history: any(c(history) for c in conditions)

# 20 条消息、有人说 TERMINATE、超过 5 分钟,三个条件先到先停
should_stop = any_of(max_messages(20), text_mention("TERMINATE"), timeout(300))

text_mention 里就有省 token 的门道。seen 记着上次扫到哪,每次只过新增的几条,历史再长,检查本身的开销也不涨。

再往上一步是判决式。照上面的格式再加一个条件函数,函数内用一次模型调用读任务描述与当前产出,判断质量达不达标,返回真或假。判定结果写进结构化字段,方便跟程序里其他条件拼接。

语义型在两个方向上有风险。模型可能永不满足,越改越花,预算一路烧到顶;也可能提前宣布胜利,活没干完就收工。所以语义型必须跟预算型并用。语义管该不该停,预算管最多跑多久,单独用任何一族都缺一层保护。

停下来的第三条通道在系统外,走的是外部信号。人按下停止,或运维接口触发取消,都算这一类。参考实现另有一种交接终止,智能体主动声明移交完毕,会话就地收尾。

会话一旦跑起来,常常几分钟都停不下来。可中途偏偏有两种情况需要打断,用户改了主意,或智能体进了死循环。参考实现把优雅取消列为编排系统必备能力,编排器循环每转一圈,都要给取消检查留出位置。

取消怎么收场?信号以异常的形式进来,参考实现就地把它接住再往上抛,于是和正常终止走上同一条收场通道。收场的动作固定四件,已产生的全部消息原样打包;从消息里倒着找最后一条模型回复,当作最终结果;把各智能体的耗时与用量逐项加总,汇成一份汇总;再附一条停止消息,写明退出原因。正常终止、预算到顶还是外部取消,记录上都分得清。

用户中途叫停,拿到手的也不是残缺的半成品,而是一份带着完整轨迹与费用单的阶段性交付。接着排查,或换个提示词重跑,都有据可依。

终止解决的是什么时候停,另有一类决策要往人那里交。系统运行中遇到拿不准的事,升级给人拍板,这就是人类委派。做法有两条。

规则式委派的升级条件直接写在代码里,比如金融操作超过一万美元、请求涉及医疗或法律领域、同一任务连续失败三次。一旦触发就升级,行为确定、记录可审计,监管与合规部门认的正是这类。

模型式委派把这套评估交给智能体自己,风险、复杂度、置信度都自己掂量。常见信号像客诉里提到可能的法律行动、可能影响品牌声誉。这类信号规则列不全,模型却判断得出来。

两条路不用二选一。工程上稳妥的组合,是规则打底、模型补充。高风险动作一律让规则拦下,送人工处理;模糊地带让模型给升级建议,拍板留给人。

任务委派出去之后,事情只成了一半。人审完了,结论要怎么回到系统里继续跑,这一环才是工程上最容易卡住的地方。

这条路得分两层走。机制那层靠状态持久化。LangGraph 每一步后自动存一份检查点,会话就停在断点上,人能查看状态,甚至动手改状态,再从断点接着跑。持久化加人为介入,是它写进文档的内置能力。

语义那层靠共享状态更新。LangGraph 的设计原则里有一条叫可中断,它的行程规划案例把整个闭环走了一遍。用户中途打断,报出真实档期,新信息写进共享上下文,那个正等消息的费用智能体读到修正后的状态,自动接着跑。

检查点攒下的历史,除了兜故障还有别的用处。从任意一个历史检查点把线程分叉出去,改一个输入或换一个决定,再跑一遍,就得到一条对照路径,“如果我当初选 B 会怎么样”这类问题,靠分叉不用重跑整段历史。验收新提示词、对比两版方案的时候,分叉比另起一个会话省得多,任务历史越长,省得越多。

真到实现,有两处细节得照着做。头一处,审批结论不能只回准或不准,得连理由一起写回共享状态,后续智能体读到才是个能解释的决定,不是光秃秃的判定。

另一处,断点搁在哪儿,编排设计期就得定下来。高风险动作执行之前,是天然的断点位置。真到运行时才想插断点,已经晚了,执行不会停下来等人,高风险动作已经放出去,想回填只能从上一个检查点重放。人机协同的四种形态和渐进自主的做法,第 15 章细讲。

除了委派,还有一类出错是不响的,发生在指令往下传的那一步。高层智能体把子任务交给下级,下级只顾优化自己的局部目标,可能误读、可能漏掉更高层的约束,子任务看着干成了,整体结果却跟业务意图对不上。对策是在交接处安排一个审计智能体。它不干活,只把工人的输出和原始指令逐条比对,确认所有约束都满足,才放行定案。

电商里就有现成例子。原始指令是,仅当用户的订单是首单,才给 15% 折扣。折扣智能体盯着主要动作,折扣照给,首单检查却跳过了,回报成功。审计智能体拿原始指令对照输出,再去查客户数据库,发现该用户已经有过好几笔订单,于是判失败、撤销折扣,把一次违反政策的操作拦在定案前。审计不用处处设卡,只盯高风险的交接就行,真要每步都审,token 与延迟的开销先吃不消。

上线前拿三问自检。最多跑多少步、停下来谁说了算、停不住时找谁,逐一答得上来,编排才算有一套用得上的安全设计。

编排模式怎么算成功,不同模式评的不是同一个东西,因为各模式的失败形态本来就不一样。监督者式路由一错全错,评的是派得对不对;并行式怕汇入缺了口、几路一起错;计划式怕步骤蒙混过关;辩论式与评测者循环,成败全看最后采纳的那份裁决。给每种模式配上对症的核心指标,模式本身的毛病才在评测里现形,各模式的指标与算法列在下表。

模式核心指标怎么算
监督者式路由准确率抽一批请求,统计监督者派给的智能体与应有归属一致的比例
并行式汇入完整率、共因失效检出完整率看汇入时各路结果齐不齐;共因失效靠埋一个共同错误源,看几路同错有没有被认出来
计划式步骤验收通过率逐步按验收标准判成没成,统计通过的步骤占全计划的比例
辩论式、评测者循环裁决质量对最终采纳的结论人工抽检,统计抽检合格率
顺序式、条件式整体通过率不单独设指标,按任务本身的结果判,跟单智能体任务一个评法

这些指标怎么落进自建任务集、怎么随坏案例滚动,12.5 展开。

10.5 手写一个监督者编排 ​

这一节手写一个评论区自动审核的监督者编排;这类审核是中文社区每天都要面对的真实需求。每条评论进来,系统要回答三件事,违不违规、违反哪一类、怎么处置;违规分三类,广告引流、辱骂攻击和低俗内容,彼此判断交叉不多,正好一个工人智能体管一类;再配一个不碰内容、只管调度的监督者,人工复审通道留一条兜底;三名审核员都给出结论后,最后单独跑一次汇总调用,合成处置建议。这正是 10.2 的监督者式,站在中间位置,监督者的分派次数压进预算里。

开工之前,架构上有三件事要拍板。工人智能体的上下文彼此隔离是头一件,系统提示只写自己那一类,输出固定成 JSON 判决,第 7 章 Isolate 策略和第 8 章子智能体思想在这里做了个最小实现。三个工人互相看不到对方推理,监督者只读结果摘要,谁也不用背别人的上下文。

第二件是监督者怎么分派,答案是用一个虚拟路由函数。说它虚拟,是因为代码里根本没有这个函数,只有一份工具定义摆在模型面前。参数 next 的取值是个枚举,三名工人、human_review,再加 FINISH。模型调用它,代码从调用参数里取出 next,这次分派就算完了。有个智能报告系统的实战案例正是这么做的,用 LangGraph 把 route 函数绑给监督者并强制调用,再用解析器抽出 next 字段。工具定义里列的可选值是一个 OPTIONS 枚举,成员名单之外再加 FINISH。

这里不用框架。原生 SDK 的 tool_choice 直接指名 route,效果一样,代码还更短。DeepSeek 的 Chat API 支持四种 tool_choice 取值,none、auto、required 与指名函数都在其中。

第三件,终止与委派照着 10.4 的三问落实。分派次数封顶属于预算型,FINISH 属于语义型,置信度低于阈值转人工属于规则式委派;预算耗尽还没等来 FINISH,同样落到人工,流程不会悬在半空。

代码完整放在下面。装好 openai 库,配好 DEEPSEEK_API_KEY 环境变量就能跑起来,模型名以官方文档为准。

python
import json
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="<DeepSeek 兼容端点,见官方文档>",
)
MODEL = "deepseek-v4-flash"  # 模型名以官方文档为准

WORKERS = {
    "ad_detector": (
        "社区评论的广告与引流审核员,只判断评论是否在发广告、拉客、导流,"
        "看推销话术、联系方式、外站链接、优惠诱导,不评价其他方面。"
        '只输出 JSON,形如 {"violation": false, "confidence": 0.93, "reason": "一句话依据"}。'
    ),
    "abuse_detector": (
        "社区评论的辱骂攻击审核员,只判断评论是否含辱骂、人身攻击、恶意挂人,"
        "语气差但没有攻击对象不算违规。"
        '只输出 JSON,形如 {"violation": false, "confidence": 0.93, "reason": "一句话依据"}。'
    ),
    "obscenity_detector": (
        "社区评论的低俗内容审核员,只判断评论是否含低俗、色情暗示内容,"
        "正常讨论健康话题不算违规。"
        '只输出 JSON,形如 {"violation": false, "confidence": 0.93, "reason": "一句话依据"}。'
    ),
}

ROUTE_TOOL = {
    "type": "function",
    "function": {
        "name": "route",
        "description": "选择下一步动作,返回下一个审核员的名称、human_review 或 FINISH。",
        "parameters": {
            "type": "object",
            "properties": {
                "next": {
                    "type": "string",
                    "enum": list(WORKERS) + ["human_review", "FINISH"],
                    "description": "下一个动作",
                }
            },
            "required": ["next"],
        },
    },
}

SUPERVISOR_PROMPT = (
    "社区评论审核的调度员,管理三名审核员:ad_detector 查广告引流,"
    "abuse_detector 查辱骂攻击,obscenity_detector 查低俗内容。调度规则:"
    "三位审核员各看一次通常就够,已出结论的不必重复分派;"
    "结论之间证据矛盾,或拿不准是否还遗漏违规类型时,选 human_review;"
    "需要的审核都完成且证据一致时,选 FINISH。只做调度,不自己判断评论内容。"
)

SUMMARY_PROMPT = (
    "社区评论处置决策员。根据三位审核员的结论给出最终处置建议,"
    "四选一:放行、删除、折叠展示、转人工复审,并用一句话说明理由。"
)

def parse_json(text: str) -> dict:
    """剥掉可能存在的 Markdown 围栏后解析 JSON,失败时抛 ValueError。"""
    if text.startswith("```"):
        text = text.split("```")[1].removeprefix("json").strip()
    return json.loads(text)

def supervisor_next(comment: str, results: list) -> str:
    """监督者只调用虚拟路由函数,从调用参数里取出下一步动作。"""
    done = (
        "\n".join(
            f"{r['worker']}:{'违规' if r['violation'] else '正常'},"
            f"置信度 {r['confidence']:.2f},{r['reason']}"
            for r in results
        )
        or "(尚无审核结果)"
    )
    messages = [
        {"role": "system", "content": SUPERVISOR_PROMPT},
        {
            "role": "user",
            "content": f"待审评论:{comment}\n\n已完成的审核:\n{done}\n\n请选择下一个动作。",
        },
    ]
    resp = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=[ROUTE_TOOL],
        tool_choice={"type": "function", "function": {"name": "route"}},
        temperature=0,
    )
    return parse_json(resp.choices[0].message.tool_calls[0].function.arguments)["next"]

def run_worker(name: str, comment: str) -> dict:
    """工人智能体在独立上下文里干活,只回结构化判决。"""
    messages = [
        {"role": "system", "content": WORKERS[name]},
        {"role": "user", "content": f"请审核以下评论:\n{comment}"},
    ]
    resp = client.chat.completions.create(model=MODEL, messages=messages, temperature=0)
    verdict = parse_json(resp.choices[0].message.content.strip())
    return {
        "worker": name,
        "violation": bool(verdict["violation"]),
        "confidence": float(verdict["confidence"]),
        "reason": verdict["reason"],
    }

def summarize(comment: str, results: list) -> str:
    lines = "\n".join(
        f"{r['worker']}:{'违规' if r['violation'] else '正常'},置信度 {r['confidence']:.2f}"
        for r in results
    )
    messages = [
        {"role": "system", "content": SUMMARY_PROMPT},
        {"role": "user", "content": f"待审评论:{comment}\n\n审核结论:\n{lines}\n\n请给出处置建议。"},
    ]
    resp = client.chat.completions.create(model=MODEL, messages=messages, temperature=0)
    return resp.choices[0].message.content.strip()

MAX_ROUTES = 4  # 预算型终止:分派次数封顶
CONFIDENCE_FLOOR = 0.7  # 规则式委派:置信度低于该值一律转人工,不靠模型自觉

def moderate(comment: str) -> dict:
    results = []
    for _ in range(MAX_ROUTES):
        next_action = supervisor_next(comment, results)
        if next_action == "FINISH":
            break
        if next_action == "human_review":
            return {"decision": "human_review", "results": results}
        results.append(run_worker(next_action, comment))
    else:
        # 预算耗尽仍未 FINISH,同样收敛到人工,流程不悬空
        return {"decision": "human_review", "results": results}
    if any(r["confidence"] < CONFIDENCE_FLOOR for r in results):
        return {"decision": "human_review", "results": results}
    return {"decision": summarize(comment, results), "results": results}

if __name__ == "__main__":
    for comment in [
        "这鞋质量真好,穿了一个月没开胶,推荐大家去店里试试。",
        "加微信 xxx123 领内部优惠券,都是傻子才原价买。",
        "楼主说得对,这电影的节奏确实慢,但摄影是真好看。",
    ]:
        print(json.dumps(moderate(comment), ensure_ascii=False, indent=2))

代码按四块来读。监督者的全部逻辑收在 supervisor_next 里。它把 tool_choice 强制成指名 route,模型没有别的输出形态可选,返回的值直接就是下一个动作。10.3 那句“智能体互为工具”,在这里就是只有五个取值的枚举。

工人的一切都装在 run_worker。独立上下文、领域提示、JSON 判决全在这一处,判决里的 confidence 字段就是规则式委派要看的依据。

第三块,10.1 的编排器循环在 moderate 里落实成了代码。选下一个、准备上下文、执行、更新状态、检查终止,五步一步不少,for 循环的 else 分支还兜住预算耗尽的情况。三问的答案全写在常量区。最多派四步,FINISH 和预算哪个先到按哪个停,停不住就进人工队列。

第四块是收官的 summarize,它拿着三位审核员的结论跑一次汇总提示,给出四选一的最终处置。

parse_json 会先剥掉 Markdown 围栏再解析,原因是模型输出不保证是纯 JSON。第 2 章就提过 arguments 字段可能不合法,普通文本输出也有同样的问题,所以工程代码给解析永远包一层壳。监督者的上下文只放待审评论原文和各工人的结论摘要,不放工人的推理过程,这是有意做的取舍。监督者干的是调度判断,把全文喂进去既烧 token 又分心,第 7 章的注意力预算原则在编排层一样算数。

三处模型调用都把 temperature 钉在 0,依据是第 1 章那条分档,输出给程序消费的用低温。队伍一大,这个参数就从全局变成了按角色配。裁判、路由和守门这类岗位,输出直接改判分派和结论,波动就是事故,温度锁死在低档;写文案、起标题这类要多样性的岗位,可以放宽到高档。第 1 章按“给程序还是给人”分档,落到编排里就细化成按角色分档,一个角色温度设多少,看它的判决责任有多重,而不是整队共用一个数。

这套结构还能再往上加一种机制,名字叫信任评分。假如哪一类违规的量变大,同一组工人要一起干同一种活,监督者就给每个工人记分。干成且质量高加分,失败、超时和输出差减分,分数随时间衰减回基线,近期表现权重大。派活时优先挑分高的,可靠的越拿越多,退化的慢慢靠边。要当心的失效模式是智能体饿死(agent starvation)。一个智能体失败过几次,分数跌穿阈值后就再也分不到任务,哪怕问题早修好了、能力还在。调整的办法有两条,衰减别设太狠,让旧失败随时间淡出;或是留个复活口,定期分一小段流量给低分智能体。

给谁派活这个题目,多智能体研究里有个正经名字,叫资源分配。集中分配器统一派、拍卖价高者得、公平分割保人人有份,还得防着有成员永远轮不上,留着可抢占的口子,前面那个智能体饿死,就是这条没防住的失效形态。落到软件场景,能分的资源主要就是配额与预算,模型调用额度、token 上限都算;上面这套信任评分,其实就是自建的一个集中分配器。

这段代码一个框架也没借。LangGraph 1.0 于 2025 年 10 月发布,官方保证 2.0 前不破坏兼容,拿它搭同样的结构并不费劲。那为什么还手写一遍?为的是看清框架在背后替人定下的三处决定,监督者读多少历史、工人的输出怎么约束、FINISH 之后归谁收尾。参考实现把所有对话历史拼成一个字符串喂给下一位,简单直接,但对话一长未必最优。这些选择一旦塞进框架就看不见,自己写过一遍,再翻框架源码才知道该盯着哪里。

本章小结 ​

  • 10.1 显式控制与涌现控制:所有编排模式排成一条线,一端开发者定死路径,一端模型运行时现场定。Router 是监督者的别名,Circular 归进条件式,黑板居中,Dynamic 在最外沿;判断任何模式先找它在线上的位置。节点内部可以再装一整套多智能体系统,分层组合允许每层各选各的模式。
  • 10.2 工作流四式:顺序式管步骤分段依赖,总延迟是各步之和;条件式管路径有限可枚举,纪律是图归图、参数归参数;并行式管子任务彼此独立,代价集中在汇入(部分失败要备补跑与降级,合并 token 按路数相加,扇出前先花约 10,000 token 造方案);监督者式管任务多而可归类,中央节点既是吞吐瓶颈也是单点,路由准确率就是系统上限,派活可连模型档位一起派。三个前提:解法已知、资源够、任务静态。
  • 汇入的质量骨架:竞争式按事先标准择优,辩论式换结论质量(token 比投票贵),评测者循环管打磨,调用都按倍数翻;共因失效是投票的死穴,数票之前先问意见独立不独立,多样性要主动做进参与者。
  • 10.3 自主三式:计划式用 blocks/blockedBy 声明依赖,步骤验收靠结构化判决,失败反馈沿执行图回流(语义反向传播加邻居感知);交接不设中央节点,智能体互为工具,交接处放输入过滤器;对话驱动全员共享历史,通信量 O(n²),监督者星型 O(n)。经典规划算法里 HTN 最能用,GOAP 自动求执行序列但结构上不允许环。
  • 10.4 终止与人类委派:预算型管止损(消息数、token 和超时,运算符组合先到先停),语义型管收工(文本信号、判决式),两族必须并用;委派规则打底、模型补充,审批结论连理由写回共享状态,断点要在设计期定好;指令保真靠交接处的审计智能体,只盯高风险交接。上线前三问:最多跑多少步、停下来谁说了算、停不住时找谁。
  • 模式怎么算成功:失败形态不同,指标各配各的,监督者式看路由准确率,并行式看汇入完整率加共因失效检出,计划式看步骤验收通过率,辩论式与评测者循环看裁决抽检合格率,顺序式与条件式不单独设指标、跟任务整体通过率走;指标进任务集的做法 12.5 展开。
  • 10.5 手写监督者:虚拟路由函数加 tool_choice 指名强制调用,工人独立上下文、回 JSON 判决,分派次数封顶与置信度阈值转人工写成常量;温度按角色分档,判决责任重的岗位锁低温。

设计题。给自己手头的场景挑一个位置,动手实现,流程写死还是模型现场派单,边界要交代清楚。度量办法,拿固定的二十条输入,统计分流正确率与调用成本,再人为断掉一个节点,验证终止条件。预期产出是一份编排代码和两项数字。要是终止那一问答不上来,这个设计就不算完。

参考资料 ​

  • Anthropic 工程博客 How we built our multi-agent research system(orchestrator-worker 形态、子智能体数量随复杂度变化、内部评测 90.2% 对比)https://www.anthropic.com/engineering/multi-agent-research-system 核实于 2026-08
  • DeepSeek API 参考 Chat Completion(tool_choice 取值 none、auto、required、指名函数对象)https://api-docs.deepseek.com/api/create-chat-completion 核实于 2026-08
  • DeepSeek API 文档首页(模型名 deepseek-v4-flash、deepseek-v4-pro,OpenAI 兼容端点)https://api-docs.deepseek.com/ 核实于 2026-08
  • LangChain 官方博客 LangChain LangGraph 1.0(2025-10-22 发布、2.0 前不破坏兼容)https://www.langchain.com/blog/langchain-langgraph-1dot0 核实于 2026-08
  • Microsoft Learn 文档 Microsoft Agent Framework Overview(AutoGen 与 Semantic Kernel 的后继框架、工作流与显式执行路径)https://learn.microsoft.com/en-us/agent-framework/overview/ 核实于 2026-08
  • 《Designing Multi-Agent Systems》第 2 章 Multi-Agent Patterns(显式到涌现的模式谱系、工作流与自主模式、任务管理与终止及人类委派分类、Magentic-One(引 Fourney 等 2024 年论文,曾登顶 GAIA 榜单)与 Anthropic 研究系统引述),见 txt-ref-zh/designing-multi-agent-systems/
  • 《Designing Multi-Agent Systems》第 7 章 Building Autonomous Multi-Agent Orchestration(编排器循环五步、三模式同循环与基类可替换、终止条件组合与 delta 检查、计划式的步骤验收与重试、max_iterations 默认 50),见 txt-ref-zh/designing-multi-agent-systems/
  • 《Designing Multi-Agent Systems》第 3 章(可中断列成设计原则、共享状态更新与行程规划案例的中断续跑闭环),见 txt-ref-zh/designing-multi-agent-systems/
  • 《Designing Multi-Agent Systems》第 6 章(步骤级重试上限:失败就地重试、超限才算真失败),见 txt-ref-zh/designing-multi-agent-systems/
  • PicoAgents:开源多智能体框架,该书配套可运行库,正文与示例代码所称“参考实现”即它;GitHub https://github.com/victordibia/designing-multiagent-systems 与 PyPI https://pypi.org/project/picoagents/ 核实于 2026-09-01
  • 《AI Agent 开发实战》第 7 章 多 Agent 系统(Agent 流程链与自主协作两模式、共享思考链、中介者、分层团队三种协作模式、AutoGen 群聊与发言者选择策略、TERMINATE 标签实践),见 txt-ref/ai-agent开发实战/
  • 《AI Agent 开发实战》第 8 章 多 Agent 系统的实战案例(中介者架构、虚拟路由函数与 OPTIONS 枚举、FINISH 路由、强制调用与结果解析),见 txt-ref/ai-agent开发实战/
  • 《构建 Agentic AI 系统》第 6 章(协调者-工人-委派者 CWD 模式,协调者做战略监督、委派者负责任务分解分发与汇总、工人只管执行),见 txt-ref/构建AgenticAI系统/
  • Julian Soh《Building and Distributing Agentic AI》(Apress,2026)第 4 章(大 O 复杂度视角:对话驱动全互连的消息量 O(n²) 对监督者星型 O(n)),见 txt-ref-zh/building-distributing-agentic-ai/
  • 《Multimodal, Real-Time AI Agent Systems》第 1 章(Router、Parallel、Sequential、Circular、Dynamic 五种多智能体架构、Router 在生产环境最常见、智能体互为工具调用),见 txt-ref-zh/multimodal-realtime-agents/
  • 《Agentic Architectural Patterns for Building Multi-Agent Systems》(Dr. Ali Arsanjani、Juan Pablo Bustos,2026,Packt)第 5 章(黑板知识枢纽:带类型、版本、置信度、时间戳的事实发布,控制器三阶段收敛,只增不改可审计,清理机制)、第 6 章(指令保真审计与电商首单折扣示例)、第 7 章(信任衰减与评分、智能体饿死失效模式),见 txt-ref-zh/agentic-architectural-patterns/。
  • 《Agentic AI Data Architectures》(Blaize Stewart、Ed Huang,2026,O'Reilly)第 3 章(多智能体共享记忆:共享表存事实、嵌入与承诺,原子更新保证无竞态),见 txt-ref-zh/agentic-ai-data-architectures/。
  • 《AI Agents in Action》(Second Edition)(Micheal Lanham,2026,Manning)第 4 章(并行流与层级协调的编排智能体可替换性判据、投票/最佳 N 选一的共享偏见失效与多样性工程化)、第 9 章(深度研究循环的实现机制:研究状态与计划外置成结构化对象、强类型迭代输出驱动循环控制、分层终止闸门、综合智能体与探索职责分离),见 txt-ref-zh/ai-agents-in-action-2e/。
  • 《Claude Code: The Definitive Guide to Agentic Development》(2026,独立出版)第 4 章(方案先行的并行化:方案四要素、约 10,000 token 对跑偏的 500,000 token;QA 蜂群五智能体约三分钟测超 146 个 URL、83 篇博文),见 txt-ref-zh/claude-code-definitive/。
  • Tim O'Brien《Building Complex Multi-Agent Systems Using Pattern Prompting》(Packt 2026)第 8 章(编排模式是提示策略与工作流形状、不是 GoF 设计模式的方法论主张,工作流形状应另映射到消息拓扑与 GoF/EIP 词汇),见 txt/pattern-prompting-mas.txt(英文原文,第 8 章,行 4733-4749、4840-4868、4896-4907、5048-5194)。
  • 《AI Agents: The Definitive Guide》第 1 章(有限状态机与分层状态机的控制词汇:状态、事件、守卫、动作、终止,超状态、子状态、历史、并行区域,作为读智能体框架图模型的通用语言),见 txt-ref-zh/ai-agents-definitive-guide/。
  • 《AI Agents in Action》(第 1 版)(Micheal Lanham,Manning 2024 中译)第 6 章(行为树六类节点:选择器、序列、条件、动作、装饰器、并行,源自机器人与游戏领域的控制工具),见 txt-ref-zh/ai-agents-in-action/。
  • 《构建 Agentic AI 系统》第 5 章(规划算法谱系:STRIPS 以逻辑谓词建模状态与动作、二元真假条件难以刻画自然语言状态故不适合 LLM 智能体;HTN 分解任务与语言模型天然契合),见 txt/构建AgenticAI系统.txt。
  • 《AI Agents with Java》(Alex Soto Bueno、Markus Eisele、Mario Fusco,O'Reilly Early Release)第 1 章 Multi-Agent Coordination Patterns(GOAP:动作声明前置条件与效果、规划器自动求执行序列,最短路搜索结构上不允许环),见 txt/ai-agents-with-java.txt(英文原文)。
  • 《Agentic AI for Engineers: Architecting Goal-Driven Systems》(Dhivya Nagasubramanian)第 12 章(辩论式架构:正反方各持论点、可引用并反驳对方历史论据,中立裁判读完整记录出结构化裁决),见 txt/agentic-ai-for-engineers.txt(英文原文)。
  • 《The Bible of Agentic AI Systems》(Stefanos Drakos,2026 自出版)第 8 章(交接输入过滤器:新智能体继承全量历史的问题与清除工具调用记录的做法)、第 9 章(评测者循环:一个创作、一个按结构化评分挑刺、不达标带反馈重写),见 txt-ref-zh/agentic-ai-bible/。
  • 《Building Agent-Powered Applications》(Vasyl Zvarydchuk,Packt 2026)第 8 章(竞争式智能体模型:多个智能体独立解同一题、按既定标准择优,两个代码智能体各试一种方案加比较智能体的示例),见 txt/building-agent-powered-apps.txt(英文原文)。
  • 《Agentic GraphRAG》第 7 章(语义反向传播与邻居感知:文本反馈当梯度沿执行图回传、反馈须带相邻节点输出防错误归因,配套工具多为研究原型),见 txt-ref-zh/agentic-graphrag/。
  • 《Agentic Architectural Patterns for Building Multi-Agent Systems》第 5 章另四节(蜂群任务板:任务上板、空手者自取、状态流转驱动交接;共识迭代协议:多轮广播-调整-容差收敛;资源分配:集中分配器、拍卖、公平分割与防饥饿;冲突解决:检测互斥计划后按层级、策略、协商、博弈论四途径裁决、裁决留审计轨迹),见 txt-ref-zh/agentic-architectural-patterns/。
  • 《AI Agent 开发实战》(郑天民,2025)第 3 章(Plan-and-Execute 的规划器、执行器、重规划器三组件,重规划器接收执行反馈动态调整计划),见 txt-ref/ai-agent开发实战/。
  • 《AI Agents and Applications with LangChain, LangGraph and MCP》(Roberto Infante,Manning 2026 中译)第 5 章与第 14 章(从检查点重放工作流、从任意检查点分叉对话线程支撑 what-if 对照场景),见 txt-ref-zh/agents-langchain-langgraph-mcp/。
  • Tim O'Brien《Building Complex Multi-Agent Systems Using Pattern Prompting》(Packt 2026)第 3 章(按组件角色定温度:裁判、路由、执行者、守门、评审宜低温,文案编辑可开高)、第 8 章(双模型并行的部分结果失败类:等待窗口、部分结果隔离区、带质量标记发布、合并策略运行时可换),见 txt/pattern-prompting-mas.txt(英文原文)。
  • 《Claude Code: The Definitive Guide to Agentic Development》(2026,独立出版)第 4 章与第 9 章(任务系统以 blocks、blockedBy 做显式依赖声明:无依赖的任务并行、有依赖的按顺序,波次执行),见 txt-ref-zh/claude-code-definitive/。