做“agency-agents”这种项目的时候,我第一反应不是某个具体仓库名,而是一类架构思路:与其把一个智能体喂成“什么都干、什么都记不住”的全能超人,不如把任务拆给一组各有专长的小智能体,再靠一个调度层把它们组织得像一家小机构那样协作。我第一次真正动手做这类系统,是在一个叫“模拟项目X”的多角色写作工具里。当时被长文本质量逼得没办法:单轮系统提示词写了三千多字,生成的报告还是顾头不顾尾,前面给的数据后面就忘了。后来把架构改成“情报员—规划师—写手—评审员”四个角色轮流协作,效果立刻不一样。
这篇文章不打算照着某个特定框架的安装文档说事,而是想把“机构式多智能体”从设计到落地过程中那些容易被忽略的东西捋一遍。内容包括角色怎么分、编排层怎么设计、工具和记忆怎么接入、实际跑起来会遇到哪些坑,最后给一个二十个文件左右的轻量实现轮廓,适合想自己动手复现的读者。
1. 为什么我会把单 Agent 换成“机构式”多 Agent 架构
1.1 单智能体的上下文天花板
单个智能体之所以在复杂任务里容易翻车,核心原因是它的上下文窗口再大也有限,而且单次生成过程本质上是一条长链。你要它同时完成“查资料、列大纲、写初稿、校对、润色”,它就必须把所有指令和资料都塞进同一段上下文里跟着走。前几步的信息随着对话加长会被慢慢稀释,尤其是涉及具体数字、专有名词、前后交叉引用的内容,越到后面越是模糊。
我在模拟项目X里做过一次很直观的对比:同样一份写作需求,单智能体模式下生成的第三章节和第一章数据对不上,日期差了三天,人名出现两种写法。而换成多角色架构后,负责资料整理的智能体把结构化信息单独落成一份中间文件,后面的写手只读这份文件,不再从冗长对话里“找记忆”,错误率明显下降。
这不是说单智能体不能用,而是说它更适合目标清晰、步骤少、上下文占用低的场景。一旦任务步骤超过四五个且互相依赖,单链式推理的脆弱性就会暴露。
1.2 可观测性带来的维护收益
机构式架构真正的好处,不只是“生成质量更高”,还有一个经常被低估的点:可观测性。单智能体模式里,用户只看到最后那一大段输出,中间每一步推理都揉在一个黑盒里。一旦结果不对,你很难判断是资料收集不全、规划跑偏,还是写作风格失控。
多智能体模式则完全不一样。每个角色的产出都是独立可查看的中间工件:情报员的检索摘要、规划师的任务拆解、写手的初稿、评审员的问题列表。调试的时候可以直接定位到具体节点,省掉大量“反复重新生成整个流程”的时间。
我自己的习惯是在每个角色执行完后写一条结构化日志,包含输入摘要、输出长度、耗时和 token 消耗。这套日志跑久了,甚至可以帮你反向优化角色提示词——哪个角色频繁返工,哪个角色输出格式不稳定,一眼就能看出来。
1.3 不是所有项目都该上多 Agent
先泼一盆冷水:多智能体架构有成本,也有复杂度。每个角色至少是一次独立的大模型调用,链路一长,延迟和费用都会成倍增长。如果你只是做一个简单的问答机器人,或者任务只有两个步骤,那多智能体纯粹是给自己找事。
我当时判断是否用多 Agent 的标准只有三条:任务是否超过四个独立阶段、阶段之间是否要求不同专业能力、每个阶段是否有可验收的中间产物。三条都满足才值得拆。否则宁可先用单智能体把流程跑通,也别为了架构而架构。
2. 先想明白分工:像公司部门一样拆角色
2.1 我常用的五种角色模板
不同业务场景的角色设计会不一样,但把大量项目经验抽象之后,我手头有一个相对通用的五种角色模板,基本可以覆盖大多数任务:
- 情报员:负责搜索、检索、抓取和整理原始信息,输出格式通常是“要点列表 + 来源结构化数据”。
- 规划师:把用户目标拆成可执行的子任务,并决定每个子任务应该交给哪个角色、按什么顺序执行。
- 执行者:真正干活的角色,根据规划师的指令完成写作、编码、分析或设计,输出满足验收标准的产物。
- 质检员:对上一环节的产物做批判性审查,指出事实错误、逻辑矛盾、格式偏差和安全风险。
- 汇总人:把多份中间结果整合成用户最终看到的统一输出,负责排版、摘要和一致性校验。
拿到一个新项目,我不会直接套五个角色,而是先画一张最简单流程图,看任务自然分成几步,再给每步配一个“必要角色”。比如一个编码助手,执行者和质检员是必须的,情报员可以退化成“读代码库索引的工具调用”,规划师也可以由执行者兼任。角色是动态配的,不是越多越好。
2.2 交互拓扑:链式、轮询还是层级
角色定下来之后,下一步是决定它们之间怎么通信。
链式结构最简单:A 做完给 B,B 做完给 C,线性推进。适合流程固定、顺序明确的任务,比如“查资料—写初稿—校对”。轮询结构是几个角色围绕一个问题来回讨论,像开评审会,适合需要多角度权衡的任务,比如方案选型、矛盾数据仲裁。但轮询容易陷入无限讨论,必须设置最大轮数和每轮输出长度上限。层级结构是规划师居中调度,把子任务派给多个执行者并行处理,最后汇总,适合“一个大目标 + 多个独立子任务”的场景。
从我自己的实践看,八成以上的项目用链式结构加少量回环就够了。我认识的一位开发者有个很精辟的观点:多智能体的交互是越高价值越复杂,但对大多数早期项目来说,链式是投入产出比最高的。
2.3 角色提示词不要写成长篇大论
角色提示词的写法直接决定系统稳不稳。我踩过最大一个坑是:某个角色提示词写了八百字,结果它对无关任务的“主动发挥”反而更多了。后来我统一改成四个段落:角色的目标、可使用的工具和权限、输入输出的格式要求、避免做什么。
“避免做什么”这一段尤其重要。比如质检员,如果不明确禁止,它会把所有无关紧要的风格问题都当缺陷抛出来,导致执行者返工到崩溃。再比如情报员,如果不限制输出格式,它会把检索结果写成散文,下游解析起来极其痛苦。角色边界说穿了就是“让你的智能体在看不见的地方也能收敛,而不是发散”。
3. 编排层:拆任务、路由消息、管上下文
3.1 任务拆解:把用户目标变成一张执行清单
编排层的第一件事,是把用户的一句话需求拆成一张可执行的任务清单。拆解的粒度非常关键。拆太粗,每个任务还是很大,角色依然会丢信息;拆太细,编排开销和 token 损耗又不划算。
我的经验是拆到“一个任务对应一次角色调用,且输出可以被某个结构化校验函数检查”的粒度。比如“写一份行业调研报告”,不拆成“调研行业”,而是拆成这种样子:收集最近三个月的市场数据、输出十条核心趋势要点、基于要点生成三级大纲、按大纲写第一章、写第二章、执行一致性检查。每个子任务都有明确的输入来源和输出型状,这样编排层才知道谁先谁后、谁依赖谁。
任务拆解这一步我是用大模型完成的,但不会让它直接决定最终执行顺序。大模型输出清单之后,我会用一个确定性图结构去约束顺序,避免规划师天马行空。
3.2 消息路由:不该广播的消息就不广播
多智能体系统里有个很常见的坏味道:每次任务开始,把所有上下文和所有角色输出都广播给所有参与者,等于把噪音无限放大。正确做法是建立显式消息路由规则,每条消息必须携带四个字段:发送者、接收者、任务编号、载荷数据。
路由规则可以做成一张静态表,也可以用条件判断。比如用户说“帮我读这个 PDF 并总结”,编排层就直连“文档解析工具→汇总人”,根本不需要启动情报员和规划师。该有多少步就放行多少步,多余的消息一律不发,这是节省 token 最立竿见影的手段。
3.3 上下文管理:让“记忆”转移到数据里
多智能体最复杂的部分其实是上下文。每个角色需要的信息范围不同,如果把全局上下文原封不动传给每个角色,不仅浪费,还会造成干扰。
我在项目里把上下文分成三层:全局上下文存用户目标、约束条件和全局指标;任务上下文存当前子任务所需的数据和前置输出;执行上下文是角色当前这一步的输入消息。每个角色默认只能看到任务上下文和执行上下文,全局上下文只在必要时以“摘要字段”方式注入。
这种设计的核心逻辑是:不要让模型靠“记住对话前面说了什么”来工作,而是让信息以结构化数据的形式存在消息载荷里。角色读取的是一份可检索、可校验的事实清单,而不是一段越来越模糊的聊天记录。这样即使某个环节失败了,从头重跑时也不需要回放全部历史。
4. 工具注册表与记忆系统:决定系统是“空聊”还是“能干”
4.1 工具注册表:有限暴露能力给角色
光让几个智能体互相发消息,产出的还是文字空转。要让系统真正“干活”,必须接工具。我的实现方式是一个工具注册表,每件工具用 JSON Schema 描述自己的名称、入参、出参和支持的操作,角色调用前只能看到注册表里的工具列表和说明,看不到其他角色内部逻辑。
工具注册表的粒度要设计得比函数更粗一点。比如“搜索资料”是一个工具,具体是调用搜索 API、读本地知识库还是查数据库,由工具内部决定。这样做的好处是角色不需要知道底层实现,编排层也能统一控制权限。对于安全性要求高的场景,工具白名单比让模型自由发挥更可靠,我宁可多写几个包装函数,也不让模型直接调用任意外部接口。
4.2 短期记忆与长期记忆分层
记忆系统我分成两层:短期记忆是当前任务链路里各角色产生的中间结果,存在内存对象里,链路结束就清理;长期记忆是过去项目里沉淀出来的知识,比如行业术语表、用户偏好、常见错误模式,存在轻量级向量库里。
长期记忆的写入不是自动的。每个角色执行完后,会有一个专门的“记忆提取”步骤,判断当前产物里是否有值得沉淀的内容。这个步骤我通常用一个小模型单独跑,不会让它影响主链路延迟。如果判断为“值得写入”,再去做向量化存储。
这里有个经验教训:长期记忆过载比没有记忆更可怕。我试过把所有中间产物全存进向量库,结果相似度检索时疯狂命中无关旧内容,输出全是噪音。后来只保存两类:高置信度事实和明确的错误修正记录,效果立刻正常。
4.3 工具失败后的容错策略
工具调用不可能每次成功。搜索超时、解析报错、数据库连接断开都是常事。多智能体系统里,工具失败不能直接让整个任务失败,应该做三层容错。
第一层是重试,对瞬时错误设置最长重试次数和指数退避间隔。第二层是降级,比如情报员搜索失败时,改用本地缓存数据;写手的渲染工具失败时,允许它退回纯文本输出。第三层是隔离,把失败的任务标记为“失败状态”,不让它有资格污染下游节点。
我实际遇到过一种更隐蔽的失败:工具明明调用成功,但模型编造了一个工具并不支持的参数。所以容错逻辑里我会强制校验模型发出的工具调用是否符合注册表里的 Schema,不符合直接拒绝并把校验错误回传给角色要求修正。这一步看起来简单,但能让整体可靠性上一个台阶。
5. 实际跑起来之后,我踩过的五个坑
5.1 Token 预算失控
多智能体系统的成本问题比很多人想象得严重。每个角色都需要接收一定的前置信息,角色越多,重复传递的固定开销越大。一个四角色的链路,如果每步都传输整份对话历史,token 消耗比单智能体高出一大截。
我后来在编排层强加了“每一步的输入裁剪”策略:传给角色的消息只保留当前任务所需字段,历史记录用摘要代替,而且摘要限制在两百字以内。实测下来,token 消耗直接减少了四成,而质量几乎没有变化。项目上线前我会先用几十条代表性请求跑一遍成本测试,给自己划一个单次任务 token 预算上限,超出就优化流程而不是硬扛。
5.2 信息逐层衰减
这是多智能体最典型的病:情报员找到十个关键点,汇总给规划师时只剩七个,再传给执行者时只剩四个,最后成品里就只剩两三个。信息衰减不是模型变笨了,而是每一层都在做“无损到有损”的压缩。
解决办法是改变通信方式。不是让上游角色用自然语言把信息“转述”下去,而是让上游角色输出结构化的数据对象,比如 JSON 数组、表格、明确命名字段,下游角色必须基于完整数据结构处理,而不是基于摘要性叙述。我在模拟项目X里把信息传递全部改成 JSON 对象后,跨章节数据不一致的问题基本绝迹。
5.3 循环调用和死锁
质检员存在的意义是找问题,但如果你不给它设定“满意阈值”,它就会一直找问题。我早期遇到过执行者写一版,质检员批评,执行者改一版,质检员又挑出新的风格问题,反复六轮还结束不了。
后来我在编排层加了两个限制:一是最大返工次数,默认三,超过就直接进入汇总阶段;二是“合理问题”过滤,质检员提交的问题必须标注严重级别,只有严重级别达到“事实错误”或“逻辑矛盾”才能触发返工,纯粹的风格偏好和措辞建议不构成返工理由。这样质检员依然严格,但不会再让整个系统原地打转。
5.4 评测困难:出了错不知道怪谁
单智能体出了问题,你盯着系统提示词调就行。多智能体出了问题,第一步要搞清楚是哪个角色错了,但这一步往往最难。我对付这个问题的办法很土但有效:给每一次任务生成一个追踪 ID,每个角色执行时都把输入摘要、完整输出、耗时、重试次数打上追踪 ID 写入本地日志。复现问题时,直接按追踪 ID 搜索日志,看哪一步的输出和预期相差最大。
这套日志体系跑了一段时间后,还能反哺优化方向。比如我发现某类任务里情报员的输出格式偶尔不合法,导致下游解析失败,就把格式校验从编排层提前到情报员内部,问题率降了七成。没有日志,这种优化无从谈起。
5.5 过度设计:一上来就画复杂的图
最后一个坑是心态上的。很多开发者包括我自己,容易一上来就设计一个带超多节点、条件分支、并行网关的多智能体编排图,结果写了两周还没跑通第一个端到端用例。多智能体本质上是分布式系统,复杂度是叠加出来的,不是设计出来的。
我现在的原则是:先做一条最粗的链式管道,三个角色两轮交互,把端到端跑通。跑通之后再逐步添加分支、记忆、工具和质检环节。每增加一个节点,都要用一批测试用例验证“这个节点真的带来了质量提升”。如果验证不出提升,就砍掉。这个原则帮我避免了很多无意义的架构成本。
6. 一个二十文件左右的轻量实现参考
6.1 工程结构规划
如果你也想自己复现一套最小可用的多智能体系统,我建议不要一上来就引入复杂框架,而是从一个“自己看得懂”的工程骨架开始。我参考下面是适合起步的目录结构,一共不到二十个文件:
agent-hub/ ├── config/ │ └── settings.yaml # 全局配置:模型名、超时时间、token 上限 ├── roles/ │ ├── base.py # 角色基类,统一输入输出 │ ├── researcher.py # 情报员 │ ├── planner.py # 规划师 │ ├── writer.py # 执行者 │ └── reviewer.py # 质检员 ├── core/ │ ├── router.py # 消息路由 │ ├── task_graph.py # 任务图定义 │ ├── context.py # 上下文分层管理 │ └── memory.py # 短期记忆和长期记忆 ├── tools/ │ ├── registry.py # 工具注册表 │ ├── search.py # 模拟搜索工具 │ └── calculator.py # 简单计算工具 ├── app/ │ └── main.py # 服务入口,接收用户请求 ├── tests/ │ ├── test_router.py # 路由单元测试 │ └── e2e_demo.py # 端到端演示脚本 └── README.md这个结构够小,每一个文件职责也都足够清楚。想扩展的时候,加角色就是加一个目录文件,加工具就是给注册表加一条声明。
6.2 核心编排代码示例
下面是我实际用过的编排核心简化版,删掉了大量细节,但保留了最关键的调度逻辑:任务拆解、角色执行、结果归集。你可以直接照着搭一个能跑的骨架。
# core/task_graph.py import json from dataclasses import dataclass, field from typing import Any, Callable @dataclass class Task: task_id: str role: str input_payload: dict[str, Any] dependencies: list[str] = field(default_factory=list) status: str = "pending" output: dict[str, Any] = field(default_factory=dict) class TaskGraph: def __init__(self): self.tasks: dict[str, Task] = {} def add_task(self, task: Task) -> None: self.tasks[task.task_id] = task def ready_tasks(self) -> list[Task]: ready = [] for task in self.tasks.values(): if task.status not in ("pending", "blocked"): continue deps_done = all( self.tasks[dep].status == "done" for dep in task.dependencies ) if deps_done: task.status = "ready" ready.append(task) return ready def complete_task(self, task_id: str, output: dict[str, Any]) -> None: task = self.tasks[task_id] task.output = output task.status = "done"# app/main.py from core.task_graph import TaskGraph, Task from roles import researcher, planner, writer, reviewer from core.memory import Memory class AgentRunner: def __init__(self, graph: TaskGraph): self.graph = graph self.results: dict[str, Any] = {} self.memory = Memory() def execute_task(self, task: Task) -> dict[str, Any]: if task.role == "researcher": return researcher.run(task.input_payload) if task.role == "planner": return planner.run(task.input_payload) if task.role == "writer": return writer.run({ "draft_context": task.input_payload, "memory": self.memory.pull_relevant(task.input_payload), }) if task.role == "reviewer": return reviewer.run(task.input_payload) raise ValueError(f"unknown role: {task.role}") def run(self, user_request: str) -> dict[str, Any]: planner_payload = {"request": user_request, "memory": self.memory.pull_relevant(user_request)} plan_result = planner.run(planner_payload) plan = plan_result["plan"] for item in plan: task = Task( task_id=item["id"], role=item["role"], input_payload=item["payload"], dependencies=item.get("depends_on", []), ) self.graph.add_task(task) for _ in range(10): ready = self.graph.ready_tasks() if not ready: break for task in ready: self.results[task.task_id] = self.execute_task(task) self.graph.complete_task(task.task_id, self.results[task.task_id]) return {"final_output": self.serialize_output()} def serialize_output(self) -> str: return json.dumps(self.results, ensure_ascii=False, indent=2)这段代码里有两个关键设计点你可能注意到了:一是全局用追踪 ID 和数据对象串联,不用自然语言长上下文传递;二是每个任务输入都从前置输出里取,不会要求角色“记住上面聊过什么”。这正是我在前面几节反复强调的核心原则。
6.3 从零跑通一个 Demo 的路线图
第一次跑通建议只用三个角色:情报员、写手、质检员,任务就选“写一篇某个城市介绍的小短文”。用最小化配置先打通消息流,再逐步把记忆、工具、并行执行加进来。
我自己在跑新框架时通常按四步走:第一步静态检查数据流,确认角色间消息格式一致;第二步用一个固定输入跑十遍,确认输出稳定;第三步加入工具调用,确认工具注册表里的 Schema 与实际执行对得上;第四步才开始做性能优化,比如并行执行子任务、压缩上下文摘要。
等这四步都跑通了,你自然会对角色分工、编排和上下文管理有直观认知。这时候再去看各种花哨的多智能体框架,会发现自己已经能读懂它们的设计动机了。
如果说这个项目让我印象最深的体会,那就是多智能体系统真正的难点永远不在“多个模型调用”本身,而在于如何设计一套让模型之间的协作可以被观察、被约束、被回收成本的数据流。先把这条主干立住,模型能力反而不是瓶颈。