前阵子接了个自动化调研的活儿需求看起来很简单让AI定期帮我产出竞品动态周报。我一开始也是老思路单Agent加一段超长Prompt把搜索、分析、总结全塞进去。跑了两周就发现不对劲上下文越堆越长输出质量忽上忽下有时候连数据来源都开始编。后来我把这套东西彻底推倒重新按agency-agents的思路组织——说白了就是不再养一个全能选手而是养一支各司其职的AI小团队有做研究的、有做分析的、有做质检的再由一个调度角色统一分派任务。这篇文章就聊聊我完整搭建这套多智能体协作体系的过程从架构设计到核心代码再到实测数据和一个又一个的坑。适合那些已经跑通过单Agent、正打算往多Agent编排方向走的开发者也适合想搞清楚Agent协作到底怎么落地的技术决策者。1. 为什么我会自建agency-agents而不是直接堆Prompt先说结论单Agent不是不能用但凡是需要多步骤、多角色、可复现的任务硬塞给一个Agent就是在给自己埋雷。1.1 单Agent模式的三个致命瓶颈我那段时间天天跟一个超长Prompt较劲网上的、文档里的、别人分享的各种结构都试过。表面看是提示词工程问题实际是架构问题。具体来说有三个瓶颈特别明显第一是上下文膨胀。任务链条一长前面搜到的资料、中间的分析草稿、后面的总结措辞全堆在同一个上下文窗口里。Token占用指数级上涨不说模型越往后越容易忘记开头的要求经常出现前面说A方案、后面又按B方案总结的情况。第二是职责混乱。既让Agent做事实检索又让它做观点提炼还让它做格式排版——这等于让一个人同时当记者、编辑和美术设计。每切换一次角色模型就要做一次思维转换转换不彻底就会串味。我实测下来串味最典型的症状是分析部分里忽然冒出一段以下是搜索结果或者总结部分开始重复搜索关键词。第三是返工成本。单Agent模式下如果中间某一步质量不行你没法精准定位是哪一环出了问题只能整体重跑。而整体重跑又意味着从头搜资料、从头生成费时费力费钱最后还不一定比上次好。1.2 一个后厨团队的类比想明白这个事之后我换了个思路。你看一个餐厅后厨不可能让一个厨师既配菜又掌勺又洗碗——配菜师只负责切配炒锅师傅只负责调味洗碗工只负责清洁各干各的最后通过传菜口协作。放到AI这边一模一样检索Agent只负责搜集和整理原始素材分析Agent只负责在素材基础上提炼观点编审Agent只负责核查前后矛盾和事实错误。每个Agent的Prompt都短而专上下文干净角色不会串出了问题也能精准定位是哪一环。这就是agency-agents的核心思想把一群各有所长的Agent组织成一个代理机构有管理岗、有执行岗、有质检岗通过明确的通信协议协作。1.3 这种模式适合什么场景不是所有任务都必须上多Agent。我梳理了一下适合agency-agents的任务有几个特征子任务之间有明确的边界、不同子任务需要不同的技能或知识侧重、任务结果需要多次检查迭代。比如周期性的行业报告、数据处理流水线、需要调用多个外部工具的复杂业务流程。反之如果是单轮问答、简单的文本改写、一次性小任务直接单Agent反而更快更省。多Agent不是银弹它解决的是复杂任务的可控性代价是更高的系统复杂度和推理成本。我后面会详细算这笔账。2. 体系架构拆解角色注册、消息总线与任务编排是怎么串起来的想清楚了要组建AI团队下一步就是搭骨架。我把这套体系拆成了五个核心组件每个组件只干一件事彼此之间通过标准化的消息格式通信。组件职责对照表组件职责类比RoleRegistryAgent的注册、查找和健康检查公司花名册MessageBus任务消息的分发与回传公司内部邮箱系统TaskScheduler任务分解、依赖管理和调度执行项目经理SharedContext跨Agent共享中间产物与状态共享文档库MemoryStore长期记忆与历史结果沉淀档案室这五个组件的分工逻辑我展开说一下。RoleRegistry是基础所有Agent启动时都要在这里登记说明自己是谁、擅长什么、接收什么格式的消息。TaskScheduler是大脑接到用户请求后先做任务分解拆成若干子任务再根据依赖关系决定哪些能并行、哪些必须串行。MessageBus不承载业务逻辑只负责把消息从A传到B这样Agent之间不会互相直接调用降低耦合。SharedContext用来传递中间产物比如检索Agent找了30条资料分析Agent需要读这30条资料就通过SharedContext拿不用重新搜索一遍。MemoryStore则是超长期记忆比如上周的周报结论、上个月的错误教训下次任务可以直接引用。2.1 数据流长什么样我按一次完整的任务走一遍数据流这样更容易理解各个组件是怎么配合的用户请求先进入入口Agent。入口Agent不干活只做意图识别判断任务类型后交给TaskScheduler。TaskScheduler把任务拆成调研-分析-编审-成稿四个子任务按依赖关系排好顺序。调研类子任务交给专门负责检索的Agent该Agent从MessageBus拿到任务干完活把结果写进SharedContext再通过MessageBus回传一个完成信号。TaskScheduler收到信号后把分析子任务投递给分析Agent。依此类推最后一个Agent产出最终结果回传给用户。2.2 为什么消息不能直接点对点传我刚开始偷懒让TaskScheduler直接拿到每个Agent的对象引用然后一个个调用。跑通是能跑通但很快发现三个问题第一新增一个Agent要改调度器代码违背开闭原则第二Agent之间如果产生间接依赖代码里查都查不出来第三想加一个任务日志审计的功能发现无从下手因为调用散落在各处。改成MessageBus之后这些问题迎刃而解。所有通信走统一通道调度器只需要知道发什么消息给谁不需要知道谁在什么时候具体怎么处理。日志审计也简单在MessageBus挂一个全局监听器所有消息记录一目了然。这就是为什么我宁可多写几百行基础设施代码也不省这个事。3. 核心代码路径从Agent基类到Director调度器的落地实现这套体系我用Python实现核心依赖只有ASGI和Pydantic没上重量级框架——后面会解释为什么。先看最底层的BaseAgent。3.1 Agent基类设计from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Optional dataclass class AgentRequest: task_id: str role: str payload: dict context_id: Optional[str] None max_retries: int 2 dataclass class AgentResult: task_id: str role: str success: bool output: dict error: Optional[str] None iterations: int 1 cost: float 0.0 class BaseAgent(ABC): 所有Agent的基类声明角色、定义能力边界、实现统一入口 role: str general description: str def __init__(self, model_name: str qwen-plus, temperature: float 0.3): self.model_name model_name self.temperature temperature self.system_prompt self.build_system_prompt() abstractmethod def build_system_prompt(self) - str: 每个Agent必须定义自己的系统提示词这是角色边界的核心 pass abstractmethod def process(self, context: dict) - dict: 核心业务逻辑接收上下文产出结果 pass def run(self, request: AgentRequest, context_store) - AgentResult: 统一执行入口处理重试、超时、成本统计 for attempt in range(request.max_retries 1): try: context context_store.get(request.context_id) if request.context_id else {} context.update(request.payload) output self.process(context) result AgentResult( task_idrequest.task_id, roleself.role, successTrue, outputoutput, iterationsattempt 1 ) return result except Exception as e: if attempt request.max_retries: return AgentResult( task_idrequest.task_id, roleself.role, successFalse, errorstr(e), iterationsattempt 1 )这个基类有几个设计点值得说。role字段是Agent的身份标识调度器靠它路由任务build_system_prompt是强制要求每个Agent自己定义角色Prompt不允许在实例化的时候临时传一个这样能保证角色边界在设计期就被焊死run方法内部统一处理了重试和成本统计业务子类只需要实现process不用关心这些横切逻辑。3.2 Director调度器任务分解与路由Director是整个体系里最关键的组件它的职责是把用户请求翻译成子任务序列再按Agent能力路由出去。class DirectorAgent: def __init__(self, registry: dict[str, BaseAgent], model): self.registry registry self.model model def decompose(self, user_request: str) - list[dict]: 将用户请求拆解为子任务序列返回[{role: ..., prompt: ...}] # 这里走一次LLM调用让模型基于当前注册的Agent列表做任务规划 agent_roles , .join(self.registry.keys()) planning_prompt f 你是任务规划器。当前可用的Agent角色有{agent_roles}。 请将这个用户请求拆解为不超过5个子任务每个子任务指定一个最适合的角色。 用户请求{user_request} 只输出JSON数组格式 [{{role: researcher, prompt: 具体指令}}, ...] raw self.model.chat(planning_prompt) return self._parse_json_safe(raw) def dispatch(self, request: AgentRequest, subtasks: list[dict], context_store): 按依赖顺序执行子任务允许并行 # 简化版按顺序执行实际可以按依赖拓扑并行 for st in subtasks: agent self.registry[st[role]] sub_request AgentRequest( task_idf{request.task_id}:{st[role]}, rolest[role], payload{task: st[prompt], context_id: request.context_id}, context_idrequest.context_id ) result agent.run(sub_request, context_store) # 把每个子任务的输出写回共享上下文供后续任务读取 context_store.append(request.context_id, st[role], result) final_result context_store.get(request.context_id) return final_resultDirector的核心是一个规划-执行循环。每次拿到用户请求先让LLM基于当前注册了哪些角色来拆分任务这一步很关键——任务分解不是硬编码规则而是动态适配的。你新增了一个数据可视化Agent下次用户请求里涉及图表需求时规划器就会自动把生成图表拆成一个子任务无需改代码。3.3 注册机制与消息规范化Agent注册我用了一个简单的装饰器实现这样新增能力只需要写一个类加一行注册语句Director那边不用动AGENT_REGISTRY {} def register_agent(cls): AGENT_REGISTRY[cls.role] cls() return cls register_agent class ResearcherAgent(BaseAgent): role researcher def build_system_prompt(self): return 你是一名研究员负责检索并整理事实性资料只输出原文摘录和来源链接不做主观判断。 def process(self, context): # 调搜索API把结果去重、提取摘要 query context[task] results search_web(query, top_k10) return {materials: dedupe(results)}每个Agent通过register_agent把自己登记进AGENT_REGISTRYDirector启动时加载这个字典。这套做法看起来不起眼但它保证了两件事Agent列表是运行时可见的规划器能实时感知可用能力新增Agent不影响任何既有代码路径。消息格式我全部用Pydantic的dataclass约束字段不可随意增删。这么做看起来啰嗦实际好处是每个Agent收到的都是结构一致的任务包不会出现A传了b字段、B期待c字段的对接错位后续加消息追踪、审计、重放全都基于这套格式做不用返工。4. 一次真实任务的完整旅程从任务拆解到报告产出架构讲得再多不如看一次实际执行过程。我带了一个具体的业务场景走完整流程。4.1 场景设定竞品动态周报某团队需要每周产出一份竞品动态周报内容包括主要竞品的产品发布动态、价格或策略调整、行业趋势信号、以及对自家产品的潜在影响评估。之前这个活靠人工每人每周要花半天。我的目标是用agency-agents把这套流程跑起来。4.2 四个Agent的职责定义我注册了四个Agent角色分别是researcher研究员、analyst分析师、reviewer编审、writer撰稿人。researcher对接搜索API搜集一周内目标竞品的公开动态输出事实清单附带原始链接和时间戳。analyst接收研究员的事实清单结合产品背景提炼出哪些动态值得关注以及可能的影响输出分析要点。reviewer核查分析师输出的合理性检查是否有夸大、前后矛盾、事实缺乏支撑输出核查意见。writer把分析要点和核查意见整合成一篇结构完整的周报输出Markdown正文。这四个角色的顺序是严格依赖的researcher干完analyst才能干analyst干完reviewer和writer才有输入。我把前三个当作串行链writer放在最后。4.3 实际执行数据跑了三轮完整测试其中一轮是模拟突发竞品发布会当天的数据采集。执行情况如下表环节平均耗时平均Token消耗质量自评任务规划Director3.5s约800稳定researcher22s约6000事实清单较全偶有重复项analyst18s约4500洞察质量高偶尔出现跨领域脑补reviewer12s约3500能抓住大部分问题有少量误报writer15s约5000成稿稳定格式统一整个链路跑下来约70秒。对比人工方式需要半天效率提升接近两个数量级。Token消耗约2万左右按通用模型的成本折算单次生成成本在几毛钱级别。这个成本换一个团队半天的工时性价比相当可观。4.4 为什么坚持这个顺序而不是让一个Agent全干我在跑测试的过程中试过合并角色的偷懒做法让analyst同时兼任reviewer也就是自己分析、自己检查。结果发现一个问题分析Agent给自己挑错的时候挑错意愿明显下降很多明显的前后矛盾它自己看不出来。后来我专门查了一下这跟模型的自我纠错能力限制有关模型在自己生成的内容上做批判性审视效果远不如让它去审视另一个角色的输出。这也是agency-agents为什么强调角色分离的一个具体原因——不只是为了让Prompt短更是为了利用不同视角带来的制衡效应。reviewer Agent的Prompt里有一句关键设定你负责代表用户质疑分析师的观点不要轻易认同。这句话让它在大概率上保持挑剔姿态。4.5 最终产出效果成稿一篇周报大约800到1200字结构固定为概览、竞品动态要点、影响分析、下周关注。人工审阅后初稿可用率在八成左右剩余两成主要是有些动态抓取不全或者影响判断过于保守。相比纯单Agent生成这套体系的优势在于事实核查环节能把编造数据的概率压到很低因为reviewer会要求每个结论给出支撑来源成稿格式始终稳定writer不像单Agent那样隔几篇就突然改变排版风格。5. 跑通之后的排障实录上下文污染、任务死锁与格式翻车任何系统只要跑起来问题都会冒出来。我集中说三个最典型的附带完整排查链路。5.1 上下文污染Agent串味是怎么回事现象analyst的输出里出现researcher风格的句子比如直接输出以下是搜索到的原始URL列表但它根本没搜过网。排查过程我先看MessageBus日志确认消息路由没有问题analyst确实只收到了researcher的产出摘要。然后又检查了SharedContext发现我一开始设计的是所有子任务共享同一个上下文对象researcher把30条原始资料全塞了进去analyst拿到的上下文里既有整理好的事实清单也有未清洗的搜索摘要。我给analyst的Prompt说基于研究资料进行分析模型一看上下文里有一大堆URL原文就倾向于直接引用它们。根因找到了Sharing Context时没做内容隔离中间产物的形态不合格。修复每次子任务之间传递时不直接传原始上下文对象而是通过一个extract_for_role(context, role)函数该函数按照目标角色的需求把上下文裁剪成该角色需要的最小集合。比如传给analyst的只保留researcher整理后的fact_list丢弃所有原始搜索摘要和中间草稿。修完后串味现象彻底消失。这个教训后来让我养成了一个习惯SharedContext里必须显式声明谁写的、给谁读的并加一条隔离规则——默认情况下其他Agent只能读取指定字段看不到完整内容。5.2 任务死循环两个Agent互相甩锅现象有一轮测试迟迟不返回结果查日志发现Director一直在循环调度任务卡了快10分钟。排查MessageBus日志显示reviewer在核查时发现了analyst的一个疑点于是它把需要补充分析的信令写回了SharedContext。我那个版本的调度器逻辑是只要上下文里有未处理信令就继续投递于是analyst又收到补充分析的任务补完之后reviewer又发现新的疑点……两人来回互相派活形成死循环。修复我在Director里加了两道保险。第一道是每轮调度检查该子任务的最大执行次数上限超了就强行终止并标记需人工介入第二道是引入超时熔断机制整个任务的执行时间超过预设阈值我设的5分钟就自动终止避免无声无息地耗着。这个坑本质上是我在设计任务依赖时没有考虑回环的情况。任务有向无环图DAG不是口头说说要真正在做调度的时候校验依赖关系哪怕用最简单的方式——每轮执行前检查这个子任务是否已经被执行过N次——也能挡住绝大部分死循环。5.3 模型输出格式翻车让Agent输出JSON它非要加Markdown现象Director的_parse_json_safe函数频繁报错一看原始返回模型偶尔会在JSON外面加一段解释性文字比如以下是JSON格式的结果导致解析失败。排查不是Prompt问题。我试过在提示词里强调只输出JSON不要其他内容正确率从70%涨到85%但尾部仍有15%的概率翻车。再试了降低temperature从0.7降到0.3正确率继续提升到92%但仍然有漏网。这说明单靠提示词和参数控制解决不了这个随机性问题必须靠代码兜底。修复_parse_json_safe里做三层兜底。第一层尝试直接json.loads第二层用正则提取第一个[和最后一个]之间的内容再解析第三层如果还是失败就把这个子任务标记为格式异常重新提交给同一个Agent并附带上次你返回的内容格式不对请重新输出严格JSON的错误反馈。这三层下来正确率逼近100%。这条经验让我对Agent系统有个新认识不要把稳定性押注在模型的输出纪律上而是要让系统能容忍模型偶尔的调皮用解析兜底和自动重试去吸收这部分不确定性。6. 事后复盘这套体系到底值不值我用下来的三点建议跑通这套agency-agents我最大的感受不是AI真厉害而是工程化真重要。模型能力强只是一个前提真正让系统稳定可依赖的是角色边界、消息规范、重试机制和日志追踪这一整套工程手段。6.1 什么时候不要用多Agent如果只是单轮问答或者任务链条很短且不需要交叉验证直接单Agent就好省事省钱还快。多Agent的价值只有在任务复杂到单Agent无法可靠完成时才体现。你可以用这个标准判断如果任务链条超过三个步骤且中间有需要不同技能的环节且结果需要人工审核——这时候才值得上agency-agents。达不到这个复杂度所有的编排代码都是负担。6.2 三条实操建议第一先把流程写死再上调度器。我第一次做就想让Director完全动态规划任务结果每次执行路径都不一样排障全靠猜。后来老老实实先写好固定的执行链比如固定的搜索、分析、核查、成稿四步骤跑熟之后再让Director做局部动态调整。稳定的主干加上有限的灵活性远比全动态可靠。第二把每个Agent的能力边界钉死。这体现在Prompt设计上也体现在输入输出格式上。我给每个Agent都规定死了输入类型和输出字段researcher只能输出事实清单analyst只能输出分析要点谁也不能越界。边界一旦模糊系统的不确定性就会指数上升——因为模型会自动补全你没说明的部分而补全的方向你不可控。第三日志必须能看清谁在什么时候干了什么。我建议在MessageBus挂一个全局日志监听记录每一次消息的发起角色、接收角色、耗时、成本、成功失败状态。这套日志平时不显眼一旦要排查问题它就是救命的。我排上下文污染和死循环的两次经历全靠日志一步步反推没有日志的话这两个问题几乎没法定位。6.3 后续可以怎么玩这套体系跑顺之后我下一步打算做两件事一是把MemoryStore用起来让每周周报自动参考上个月的结论实现跨任务的连续性二是加上并行调度目前子任务还是顺序执行实际上researcher可以拆成多个并行搜索任务来压缩耗时。这个方向做下来可能会让整条链路从70秒压到40秒以内我到时候有新发现再回来分享。