去年年底我接手了一个内部自动化项目,起初只打算用一个大模型智能体把所有流程跑通。结果跑了不到两周就被各种问题缠住:上下文越聊越长导致输出质量波动、角色职责混在一起后提示词怎么调都别扭、某个分支任务出错时整条链路的定位成本高得离谱。后来我把方案彻底改成了多个各司其职的智能体协作模式,也就是这里要讲的 agency-agents 思路——不是把一个 agent 塞进所有场景,而是像经营一家公司那样,把任务拆给一组角色明确的智能体,再靠调度层把它们组织起来。这篇文章我会从我自己的踩坑和重构过程出发,聊聊多智能体协作的几种组织模式、一个手搓调度框架的核心逻辑、通信协议怎么设计才不容易失真,以及真正跑起来之后最容易翻车的那几个坑。
1. 为什么单Agent撑不住复杂任务:从上下文天花板说起
1.1 一个Agent包打天下的隐形代价
大多数人对“一个Agent处理一切”的第一印象是:只要把任务描述清楚,再给足工具调用权限,模型自己就能完成计划、执行、反思的循环。这个思路在小范围Demo里确实成立,比如“帮我查资料并写一份周报”这种单点任务。
但一旦任务变成“调研市场动态—产出选题—生成初稿—校对合规—输出发布文案”这样多环节的流程,单个Agent的问题就会暴露得很彻底。
最直接的瓶颈是上下文窗口。流程越长,中间产物越多,Agent的记忆空间就塞得越满。当关键信息挤在几十段历史里,模型开始“记不住”最开始的需求边界,输出的风格和准确性都会滑坡。我实测过一个内部任务:让Agent同时承担“撰写方案”和“检查错误”两个角色,结果它写出来的内容越来越像“自我表扬稿”——因为它在自我纠错的同时,也在尽力维护自己先前产出的合理性,这在认知上本身就是一种拉扯。
更深层的问题是角色混淆。提示词里同时塞了管理者角色、执行者角色、质检角色,模型虽然在文本层面“理解”了这些角色,但执行时很难在同一个上下文中灵活切换。你需要严格审核、它偏向宽松;你需要发散创意、它又过度收敛。这不是模型能力不足,而是职责边界缺失。
1.2 多Agent协作的工程本质:更像微服务,而不是聊天群
把多个Agent组合起来,本质上不是“开一个群聊让AI们互相对话”,而是把任务链路拆成若干个有清晰输入输出接口的节点,每个节点由一个具备特定角色、工具和记忆空间的Agent负责,再用一个调度核心把它们串起来。
你可以把它理解成微服务架构:每个服务(Agent)自治、有明确接口、可以独立扩展,服务之间通过消息传递协作。这样做的好处很明显——职责单一、故障隔离、可单独调优。
我用一个曾经的方案对比来说明:
- 单Agent方案:一次 Prompt 包含“数据分析+文案生成+格式整理”
- 多Agent方案:分析Agent输出结构化数据 → 写手Agent基于数据分析结果生成文案 → 编审Agent做事实与合规检查
差异在于:第二套方案里,每个Agent的Prompt可以保持极短、极专注,且每个Agent的输出都有独立的检测标准。哪一步出了问题,直接定位到对应Agent,重跑那一段即可,而不是把整条链子推翻。
1.3 “Agency”不是概念包装,它是生产组织方式
“agency-agents”这个词组最有价值的地方在于“agency”这个组织化隐喻。单个Agent更像自由职业者:接单、干活、交付。而Agency像一家代理公司:有总监拆解需求、有专员执行细分任务、有质检员把关、有项目经理盯着进度。
做工程落地的时候,真正要思考的问题不是“哪个Agent更聪明”,而是“这个Agency的组织架构怎么搭、任务怎么分配、消息怎么流转、失败怎么恢复”。
这三种协作模式是我在真实项目中反复对比后确定的:
主管-专家模式(Orchestrator-Worker)这是最稳、适用范围最广的模式。一个Planner Agent负责任务理解与拆解,把整体任务切分成子任务,分配给不同专家Agent执行,再由Reviewer Agent验收汇总。类似项目经理带专业团队。
我实际用它搭过一个自动调研系统:Planner把“竞品分析”拆成“价格数据收集”“功能对比分析”“用户评论情绪分析”三个子任务,分别交给三个专家Agent,最后合并。效果很稳定,因为每个专家只需要专注自己那块,不需要理解全局。
自由讨论模式(Debate/Collaborative)多个Agent围绕同一议题多轮发言、互相提问或辩论,适合创意发散、方案评审类任务。但如果没有主持人Agent和迭代轮次上限,它很容易变成“AI相声”——互相客气或者互相抬杠,最后产出一堆共识废话。
我试过让三个Agent自由讨论“如何提升某产品的留存率”,不加约束地聊了十轮,结果大部分内容在重复观点。加了主持人Agent和轮次限制之后,输出质量才明显提升。
流水线模式(Pipeline)按固定顺序,一个Agent的输出作为下一个Agent的输入,各节点只负责单一动作。适合流程相对固化、顺序敏感的任务,比如“数据清洗 → 特征提取 → 模型训练 → 报告生成”。
三种模式各有适用场景,实际项目中经常混用:主流程用主管-专家,其中某个创意环节临时切换成自由讨论,再回到流水线继续。调度器设计得好不好,就看能不能灵活支持这种混编。
2. 手搓迷你Agency框架:一个事件驱动的调度核心
2.1 先定场景:三个角色的“内容生产小作坊”
为了让调度逻辑可落地,我把内部一个真实场景简化成三个Agent的协作:
- 主编Agent(Planner):接收用户需求,拆解为选题方向、内容大纲、写作规范
- 写手Agent(Writer):根据大纲产出正文初稿
- 编审Agent(Reviewer):检查事实性、格式、风格,输出修订意见或直接返回通过
这个结构小巧但五脏俱全:有规划、有执行、有验收,覆盖了调度器需要关心的全部环节——任务下发、消息传递、状态管理、结果校验。
2.2 消息与状态的数据结构设计
先定义两条核心数据结构:消息(Message)和任务(Task)。
消息流转的载体,我设计成字典形式,核心字段包括:
{ "msg_id": "unique-id", "sender": "planner", "recipient": "writer", "msg_type": "task_assignment", "content": {...}, "context_ref": ["task-001"], "timestamp": 1700000000 }msg_type区分“任务下发”“结果回传”“请求补充信息”等类型,方便调度器做不同处理。content保持结构化而非纯文本,这是避免传话走样的关键。
任务状态池用字典保存每个任务的当前状态:
task_status = { "task-001": { "phase": "assigned", # pending / assigned / running / succeeded / failed "current_agent": "writer", "agent_output": None, "retry_count": 0, "created_at": ..., "finished_at": None } }2.3 事件循环:队列 + 路由表 + 回调
调度核心采用简单的“事件循环”,三个关键组件:
- 消息队列(Queue):存放待分发消息
- 路由表(Registry):映射每种
msg_type或每个Agent的能力标签,决定消息由哪个Agent处理 - 任务状态池(State Store):记录所有任务进度,支持失败重试与超时控制
核心调度循环的伪代码我保留在这里,逻辑非常直白:
import queue class AgencyDispatcher: def __init__(self, agents: dict): self.agents = agents # {"planner": planner_obj, "writer": writer_obj, "reviewer": reviewer_obj} self.message_queue = queue.Queue() self.task_status = {} self.routing_table = { "task_assignment:writer": "writer", "task_assignment:reviewer": "reviewer", "review_result": "planner", "draft_revision": "writer", } def submit_task(self, task_id: str, first_agent: str, input_payload: dict): self.task_status[task_id] = { "phase": "assigned", "current_agent": first_agent, "retry_count": 0, } self.message_queue.put({ "recipient": first_agent, "content": input_payload, "task_id": task_id, }) def run_loop(self, max_iterations=50): for _ in range(max_iterations): if self.message_queue.empty(): break msg = self.message_queue.get() agent = self.agents.get(msg["recipient"]) if not agent: continue result = agent.run(msg["content"], task_id=msg["task_id"]) next_msg = self._decide_next_step(msg, result) if next_msg: self.message_queue.put(next_msg) def _decide_next_step(self, msg, result): task_id = msg["task_id"] if result["status"] == "success": # 根据路由表决定下一跳 next_recipient = self.routing_table.get(msg["msg_type"] + ":done") # 实际逻辑比这个复杂,要看各Agent返回里的意图字段 elif result["status"] == "retry": # 重试计数与重新入队 ...真实实现里_decide_next_step会根据Agent返回的“意图字段”决定流向:写手说“请审核”,路由到编审;编审说“需要修改”,回到写手;编审说“通过”,回到主编去收尾。
2.4 观察层与关键节点回调
多Agent系统的排错难度是单Agent的几倍。只给每个Agent塞日志打印远远不够,我习惯在调度循环里做三件事:
- 每个Agent调用前后记录耗时、输入摘要、输出摘要
- 每个任务状态变更时触发回调,便于外部监控或告警
- 对每个输出做schema校验,防止脏数据进入下一环
观察层能做到“链路追踪”:一句话从主编下发开始,一直到编审验收,每一步的流转耗时、token消耗、内容指纹都留痕。出了问题,点开对应task_id的日志,立刻知道卡在哪个环节。
3. 通信协议是隐藏的工程重点
3.1 纯文本传消息,为什么越传越跑偏
我做第一版多Agent协作时,用的是最简单的方式:Agent A的最终答复文本直接塞给Agent B。结果出现一个非常典型的“传话游戏”效应——写手Agent产出一篇完整文章,编审Agent接到文章后没有原始需求,只能凭感觉审;它退回的修改意见又很笼统,写手Agent拿不到上下文,只能猜着改。
这种纯文本传递本质上有两个问题:
- 压缩失真:上游Agent并没有把“需求约束”完整体现在最终文本里
- 意图丢失:下游Agent无法判断上游这条消息是“最终结果”还是“中间结果”“需要反馈”
3.2 结构化消息协议:让“接口”约束行为
工程上的解法其实不新鲜——给Agent之间定义接口协议。我用一套极简的结构化消息格式解决了大部分问题:
{ "purpose": "inform/request/review/approve", "sender": "planner", "recipient": "writer", "task_id": "task-20250101-001", "payload": { "topic": "智能客服系统市场调研", "sections": ["背景", "竞品", "机会点"], "constraints": ["全文不超过3000字", "必须有数据引用"] }, "context_refs": ["research-data-v3.json"], "format": {"output": "markdown", "schema": "article_schema_v1"} }每个字段都对应一个工程意图:purpose让接收方瞬间知道这条消息只是“通知”还是“需要输出”;constraints显式携带被压缩掉的需求边界;format.schema约束输出结构,方便下游解析和校验。
3.3 摘要编码:让信息被压缩但不被扭曲
除了协议结构的约束,还需要处理上下文长度问题。多Agent协作里,最典型的糟糕做法是:Agent B把Agent A的全部对话历史原封不动塞给Agent C。这个做法的后果是token爆炸,而且历史里的噪声会被模型“一本正经地误解”。
我的做法是引入一个“上下文摘要器”:上游Agent产出结果之后,调度器把结果的关键信息压缩成结构化摘要,再作为下游Agent的输入。比如编审Agent审完初稿后,并不会把全文再发给主编Agent,而是发一条:
{ "purpose": "review_summary", "result": "approve", "score": 86, "key_issues": [ {"severity": "medium", "position": "section-2", "desc": "竞品数据未标注来源"}, {"severity": "low", "position": "section-4", "desc": "结论部分衔接生硬"} ] }主编Agent只需要看到结构化反馈就足够做决策,不需要重读全文。这样既压缩了上下文,也减少了信息失真。上下文感知在工程上的核心原则是:让Agent看到“对当前决策最必要的信息”,而不是“能看到的所有信息”。
3.4 用SCHEMA校验替代“请你检查一下”
写代码的人都知道,接口不校验,系统早晚长毛。多Agent系统同理。如果Agent返回的内容不符合约定结构,调度器应当拦截并触发重试,而不是抱着“也许下一个能理解”的侥幸传下去。
我常用 json-schema 做这个事,给每个Agent的输出定义一个明确schema:
writer_output_schema = { "type": "object", "properties": { "title": {"type": "string"}, "sections": {"type": "array", "items": {"type": "object"}}, "word_count": {"type": "integer"}, }, "required": ["title", "sections", "word_count"] }一旦校验失败,调度器可以选择重试一次、请求Agent修复,或者记录错误信息后告警。重点是:坏的输出不应该走到下一环,“垃圾进垃圾出”在Agent协作里会被放大数倍。
4. 真实跑起来之后,我踩过的五个“翻车点”
4.1 死循环与讨论失控:两个AI互相“纠正”了一夜
场景:自由讨论模式下,Agent A提出一个观点,Agent B表示“有道理,但补充一点”,Agent A又说“好的,我修正”,Agent B又说“很好的补充”……如果没有轮次上限,这个对话可以无限延续,token烧光也不会收敛。
根因:缺少终止条件。自由讨论模式必须有一个明确的外部“主持人”负责喊停。
解决:我在调度器中强制设置了max_discussion_rounds,比如5轮。轮次结束后,要求参与Agent各自输出一个“结论摘要”,而不是继续对话。主持人Agent会对这些摘要做合并和投票,产出最终结果。
4.2 上下文爆炸:一次协作跑掉了日常一天的成本
场景:四个Agent协作,每个Agent每轮都要把前一步的完整输出传下去,一篇文章的流程跑下来,token消耗高得离谱。
根因:消息传递时选择了“全文透传”,而不是“摘要传递”。
解决:全面切换成“结构化摘要传递”,并且在每个Agent的定义里明确“输入边界”——告诉Agent它只需要关注消息中的哪些字段,其他字段不要处理。这一条优化直接把单次任务的token成本压到原来三分之一左右。
4.3 任务漂移:为什么我的Agent写着写着跑偏了
场景:主管Agent最初的需求是“分析华东市场,输出决策建议”,但经过两轮迭代之后,写手Agent开始大篇幅描述全国市场背景,还加了历史沿革的内容。单看每一轮,输出都是合理的,但最终产物跟原始需求已经偏离。
根因:需求约束在长链路中逐渐衰减,每个Agent基于自己的上下文理解局部目标,忘记了全局目标。
解决:强制在每个Agent的输入里带上“原始需求摘要”字段,并且在流程的关键节点(比如从第2步进入第3步时)插入一个对齐检查Agent,专门对比“当前输出”和“原始需求”,输出偏差评分,评分超过阈值时回到Planner重新规划。
4.4 失败反馈黑洞:子Agent超时后整个链路卡死
场景:某个专家Agent调了一个外部工具,结果工具卡住,Agent迟迟没有返回。因为没有超时机制,消息队列里后续的所有任务全部排队等待,整条流程死锁。
根因:把Agent调用当成了“一定能返回”的同步调用,没有设置超时、降级和补偿。
解决:给循环里的每个Agent调用加上了超时控制。超过预期时间仍未返回时,调度器直接标记任务为失败,并走“降级路径”——比如跳过该子任务、使用默认值,或者通知Planner调整方案。保证整个系统不会因为单个节点失联而整体瘫痪。
4.5 Token预算炸掉:一个复杂的任务吃掉了本来够跑一天的预算
场景:一次长文档分析任务,多个Agent反复读取完整的原始文档,最终token消耗超出预期10倍以上。
根因:没有做“Token预算管理”和“输入裁剪”。
解决:我给每个任务在创建时分配一个token预算(根据任务优先级设定),调度器在实际调用Agent前,会按预算裁剪输入。文档先做分块和检索:能通过检索拿到相关片段,就不再全量塞给Agent。这招对长文档类任务非常有效。
5. 框架选型与生产落地建议
5.1 LangGraph、CrewAI、AutoGen 怎么选
手搓调度器能帮你彻底理解多Agent系统的运作机制,但生产环境里,我更建议在成熟框架上做二次开发,省去大量基础工程工作。我拿三个主流框架做个对照:
| 框架 | 编排模型 | 状态管理 | 易用性 | 最适合的场景 |
|---|---|---|---|---|
| LangGraph | 图结构,节点+边 | 内置状态池,适合复杂状态机 | 中上,概念多一点但有清晰文档 | 需要精细控制流转、有复杂分支判断的生产系统 |
| CrewAI | 角色+任务委派 | 偏轻量,靠Agent间消息流转 | 高,上手快 | 快速原型、角色分工清晰的团队协作类任务 |
| AutoGen | 对话驱动,多Agent可互相发消息 | 依赖对话轮次管理 | 高,适合交互式场景 | 需要Agent间多轮讨论、研究探索型任务 |
我的建议是:先想清楚你的流程是“图”还是“对话”。状态流转复杂、有分支判断,优先LangGraph;角色固定、任务链条清晰,CrewAI最快;需要多Agent自由讨论或人机协作,AutoGen更顺手。
5.2 成本与延迟控制的三板斧
即使框架选好了,上线之后成本和延迟仍然要专门治理。我常用的方法有三招:
- 级别路由:并不是所有请求都需要多Agent协作。先由一个轻量分类器判断任务复杂度,简单任务直接走单Agent快车道,只有复杂任务才进入Agency流程。这个入口判断能省下大量费用。
- 结果缓存:同类任务的产出结果(尤其调研、分析类)可以按语义相似度缓存复用。跑过一次“竞品功能分析”任务,下次遇到高度相似需求可以直接返回旧结果,不重新调用模型。
- 并行分支:主管Agent拆解出的多个子任务,只要相互没有依赖,调度器就应该并发执行而非串行。能并联的环节并联起来,延迟能直接少一半。
5.3 人和Agent的交接设计
最后想提醒一个容易忽略的工程点:多Agent系统不是全自动的“黑箱”,它和真实团队一样,需要“人和AI之间”的交接点。
我现在的方案里,每个任务流转经过几个关键阶段时,都会设置一个“人审”开关:比如对外输出、涉及关键数据、犯错代价高的操作,必须由人来确认“允许继续”。Agent负责提出建议和草拟内容,人负责最终拍板。这个设计既不对抗AI也不盲信AI,符合工程系统追求“可控制”的底线。
多Agent系统做起来不难,难的是把组织设计、通信协议、异常处理这些工程细节一块块拼起来。如果看完这篇文章你想动手试,我建议从最简单的三个Agent、一条固定流水线开始,先把链路跑通,再逐步加分支和并行。别一上来就做复杂的自动规划系统,先让架构在可控范围内运转起来,比一开始就追求“全智能”要靠谱得多。