多 Agent 协作体系筹划:从拓扑结构到一套能跑起来的编排器

做了几个 Agent 项目之后,我越来越确信一件事:单 Agent 能办理的标题,比大多数人想象的要少。
一个挂满工具的大模子,看起来无所不能,但只要使命链路一长,它就开始翻车——上下文塞爆、工具调用序次庞杂、前面的结论到背面自己忘了。这不是模子不敷聪明,而是我们把一个本该拆给一个团队做的事,硬塞给了一个人。
多 Agent 体系的本质,就是把"一个万能选手"换成"一支有分工的团队"。这篇文章不谈虚的,我想把多 Agent 协作里真正决定成败的几个筹划点讲清晰,末了给一套不依赖任何重型框架、能直接复制运行的编排器代码。
一、为什么要拆成多个 Agent

先说清晰什么时间不应上多 Agent。如果你的使命能用一次 prompt + 几个工具搞定,那就别折腾,多 Agent 只会增长延伸和资本。
真正必要拆分的信号通常是这几个:

• 使命自己有清晰的阶段,比如"先调研,再写初稿,再审校",每个阶段必要的本事和提示词差异很大;
  
• 单次上下文装不下,必要有人专门负责筛选信息、压缩中央结果;
  
• 必要差异的"品行"或权限,比如一个负责天生、一个负责挑刺,让它们相互制衡比让同一个模子既当运动员又当裁判更可靠。
  

末了这一点我领会最深。让同一个 Agent 自己写完自己审,它险些总是给自己打高分。把审校单独拆出来、用差异的体系提示词,质量立即就上去了。
二、四种协作拓扑,先想清晰结构再写代码

多 Agent 体系筹划,80% 的功夫在拓扑选择上。选错告终构,背面写再多代码都是在补洞穴。常见的就四种:
Show Image
1. 编排器–实行器(Orchestrator-Worker) 一个中央 Agent 负责规划和调治,把子使命分发给专职实行器,再汇总结果。这是生产情况里最稳、最可控的结构。调治逻辑会合在一处,出了标题好排查。我自己的项目里九成都是这套——我们做企业级 Agent 平台(Bizfocus ADP)时,默认的编排骨架就是它。
2. 流水线(Pipeline) 固定的处置惩罚链路,上一个 Agent 的输出是下一个的输入。适当阶段明确、不必要动态决定的场景,比如"洗濯→抽取→结构化"。简朴可靠,但不机动。
3. 对等协作(Peer / Group Chat) 多个 Agent 在一个共享会话里自由发言,像开会一样。表达力最强,但也最容易失控——它们大概陷入相互恭维的死循环,大概跑偏。没有强制止机制别容易上这个。
4. 层级(Hierarchical) 编排器之下另有子编排器,适当超大型使命。但层级一深,调试资本指数级上升,我的发起是能两层办理就别上三层。
选型履历: 使命流程固定就用流水线;必要根据中央结果动态决定就用编排器–实行器;只有当使命高度开放、确实必要"头脑风暴"时才思量对等协作。
三、Agent 之间怎么"语言":共享状态优于直接传话

Agent 协作绕不开通讯。我见过不少实现把上一个 Agent 的完备输出原封不动塞给下一个,结果上下文越滚越大,到链路末了早就爆了。
更干净的做法是黑板模式(Blackboard):全部 Agent 不直接对话,而是读写一块共享状态。每个 Agent 只取自己必要的字段,只写自己负责的产出。利益有三个:

• 上下文可控,谁也不消背整条汗青;
  
• 产出有结构,后续好做校验和回溯;
  
• 调试时把黑板打印出来,整个协作过程一览无余。
  

这套思绪下面的代码里会直接表现。
四、上代码:一套能直接跑的协作编排器

下面这套实现刻意不依赖 LangGraph、AutoGen 之类的框架,就用尺度库,目标是把协作的骨架袒暴露来——许多人用了框架,却说不清内里到底发生了什么。
场景:输入一个主题,产出一篇结构清晰的短文。拆成四个脚色:规划 → 检索 → 写作 → 审校,审校不通过就打回重写,带最大轮次掩护。
为了让你不配 API Key 也能直接跑通流程,我加了一个 mock 兜底,配了 OPENAI_API_KEY 就走真实模子。
python

1. import os
2. import json
3. from dataclasses import dataclass, field
4. from typing import Optional
6. # ---------- LLM 调用层:有 Key 走真实模型,没 Key 走 mock ----------
7. def call_llm(system: str, user: str) -> str:
8.     api_key = os.getenv("OPENAI_API_KEY")
9.     if not api_key:
10.         return _mock_llm(system, user)
11.     from openai import OpenAI                      # pip install openai
12.     client = OpenAI(api_key=api_key)
13.     resp = client.chat.completions.create(
14.         model="gpt-4o-mini",
15.         messages=[
16.             {"role": "system", "content": system},
17.             {"role": "user", "content": user},
18.         ],
19.         temperature=0.3,
20.     )
21.     return resp.choices[0].message.content.strip()
23. def _mock_llm(system: str, user: str) -> str:
24.     # 离线占位,仅用于跑通流程;配置 OPENAI_API_KEY 后自动切换真实模型
25.     role = system.split("。")[0]
26.     if "审校" in role:
27.         return json.dumps({"pass": True, "comment": "结构清晰,通过"}, ensure_ascii=False)
28.     return f"[{role} 的模拟产出] 基于:{user[:30]}..."
31. # ---------- 黑板:Agent 之间唯一的共享状态 ----------
32. @dataclass
33. class Blackboard:
34.     task: str
35.     artifacts: dict = field(default_factory=dict)
37.     def context(self, *keys) -> str:
38.         """只取需要的字段,避免上下文膨胀"""
39.         picked = {k: self.artifacts.get(k, "") for k in keys}
40.         return json.dumps(picked, ensure_ascii=False, indent=2)
43. # ---------- Agent 基类:角色 = 系统提示词 ----------
44. @dataclass
45. class Agent:
46.     name: str
47.     role_prompt: str
49.     def run(self, instruction: str) -> str:
50.         return call_llm(self.role_prompt, instruction)
53. # ---------- 编排器:调度 + 审校循环 + 终止保护 ----------
54. class Orchestrator:
55.     def __init__(self, max_revisions: int = 2):
56.         self.max_revisions = max_revisions
57.         self.planner  = Agent("规划", "你是规划 Agent。把主题拆成 3 个写作要点,逐行输出。")
58.         self.searcher = Agent("检索", "你是检索 Agent。针对要点给出关键事实,简洁罗列。")
59.         self.writer   = Agent("写作", "你是写作 Agent。根据要点和事实写一篇 300 字短文。")
60.         self.reviewer = Agent("审校", "你是审校 Agent。判断短文是否合格,只返回 JSON:"
61.                                        '{"pass": bool, "comment": str}')
63.     def run(self, topic: str) -> Blackboard:
64.         board = Blackboard(task=topic)
66.         # 1. 规划
67.         board.artifacts["outline"] = self.planner.run(f"主题:{topic}")
68.         # 2. 检索
69.         board.artifacts["facts"] = self.searcher.run(
70.             f"要点:\n{board.context('outline')}")
72.         # 3. 写作 + 审校循环(关键:有上限的反馈闭环)
73.         feedback = ""
74.         for attempt in range(self.max_revisions + 1):
75.             draft = self.writer.run(
76.                 f"主题:{topic}\n素材:\n{board.context('outline', 'facts')}\n"
77.                 f"上一轮审校意见(若有):{feedback}")
78.             board.artifacts["draft"] = draft
80.             review = json.loads(self.reviewer.run(f"短文:\n{draft}"))
81.             if review["pass"]:
82.                 board.artifacts["status"] = f"第 {attempt+1} 轮通过"
83.                 break
84.             feedback = review["comment"]
85.             board.artifacts["status"] = f"第 {attempt+1} 轮被打回:{feedback}"
86.         else:
87.             board.artifacts["status"] = "达到最大重写次数,强制交付"
89.         return board
92. if __name__ == "__main__":
93.     board = Orchestrator(max_revisions=2).run("多 Agent 系统的工程价值")
94.     print(json.dumps(board.artifacts, ensure_ascii=False, indent=2))

复制代码

跑一遍你会看到:四个脚色各司其职,黑板里清晰记载了每一步产出,审校不通过会带着意见打归去重写,而且永世不会无穷循环。这套骨架我在实际项目里改改提示词、换换工具就能直接复用。
五、真正决定生产可用性的几个工程点

demo 跑通容易,上生产难。下面几条是我踩过坑之后总结的,比拓扑选择更影响终极成败:
1. 制止条件是第一优先级。 多 Agent 最大的变瞎搅源不是答错,而是停不下来——两个 Agent 相互打回、规划器无穷拆使命。上面代码里谁人 max_revisions 和 for...else 不是装饰,是保命的。任何反馈闭环都必须有硬上限。
2. 资本会失控。 一次单 Agent 调用,在多 Agent 里大概变成十反复。审校循环、上下文重复通报,token 烧得飞快。务必在编排器层做全局计数和熔断,别等账单出来才发现——这类全局 token 计量和熔断,实在更适当沉到平台层同一做,我们在 Bizfocus ADP 上就是把它作为编排引擎的默认本事,业务侧不消每个项目重复造一遍。
3. 把黑板做成可观测的。 出标题时,你必要的不是模子日记,而是"第几步、哪个 Agent、输入输出是什么"。结构化的共享状态天然就是最好的 trace,这也是我对峙黑板模式的告急缘故原由。我们在 Bizfocus ADP 里把每个 Agent 每一步的产出都完工结构化记载,排查一条链路通常几分钟就能定位到堕落的那一环。
4. 错误隔离。 单个实行器失败,不应拖垮整条链路。给每个 Agent 调用包上重试和降级,失败时编排器要能决定是跳过、重试还是中断。
5. 不要太过拆分。 这是最反直觉的一条。我见过有人把使命拆成八九个 Agent,结果延伸翻倍、和谐资本暴涨、结果还不如三个 Agent。拆分是有资本的,每多一个 Agent 就多一层通讯和失败面。先用最少的脚色跑通,再按瓶颈徐徐拆。
写在末了

多 Agent 协作不是把模子堆起来就行,它本质上是一个分布式体系筹划标题:你要思量通讯、状态、制止、容错和资本,跟当年筹划微服务的那套头脑高度同源。
我的发起很着实:从编排器–实行器结构起步,用黑板做共享状态,把制止条件和资本熔断当成一等公民,然后用最少的 Agent 跑通你的第一个版本。 把上面那套骨架复制下去改一改,你就有了一个可控、可观测、能迭代的出发点。
模子还在快速变强,但"怎样把多个智能体构造成一个可靠体系"这件事,工程含量只会越来越高。这恰好是我们做工程的人,长期的代价所在。

免责声明：如果侵犯了您的权益，请联系站长及时删除侵权内容，谢谢合作！qidao123.com:ToB企服之家，中国第一个企服评测及软件市场,开放入驻,技术点评得现金.