★ FEATURED ARTICLE 去年年底我接手了一个内部自动化项目起初只打算用一个大模型智能体把所有流程跑通。结果跑了不到两周就被各种问题缠住上下文越聊越长导致输出质量波动、角色职责混在一起后提示词怎么调都别扭、某个分支任务出错时整条链路的定位成本高得离谱。后来我把方案彻底改成了多个各司其职的智能体协作模式也就是这里要讲的 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的协作主编AgentPlanner接收用户需求拆解为选题方向、内容大纲、写作规范写手AgentWriter根据大纲产出正文初稿编审AgentReviewer检查事实性、格式、风格输出修订意见或直接返回通过这个结构小巧但五脏俱全有规划、有执行、有验收覆盖了调度器需要关心的全部环节——任务下发、消息传递、状态管理、结果校验。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_iterations50): 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_idmsg[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的输出定义一个明确schemawriter_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、一条固定流水线开始先把链路跑通再逐步加分支和并行。别一上来就做复杂的自动规划系统先让架构在可控范围内运转起来比一开始就追求“全智能”要靠谱得多。
阅读完成 · 觉得有帮助？
