1. 单Agent吃不下所有活儿这套框架的出发点我最初做LLM应用时和大多数人的路径一样把一个超长系统提示词塞给同一个模型让它扮演产品经理、数据分析师、文案写手、代码审查员……结果大家都懂——提示词越长模型越容易在某个环节上犯迷糊回复风格漂移、格式不稳定、上下文一长就忘事。后来我把任务拆成多个独立脚本用硬编码流程串起来又发现流程写死了之后换个业务场景就要改一大坨if-else维护成本高得离谱。真正让我下定决心去写agency-agents这套框架的契机是我在做一个信息收集类的需求时遇到了一个两难单一Agent要同时做查资料、整理摘要、生成结构化报告三件事结果摘要部分总是丢字段而手写流水线又很难应对资料在某个站点拿不到时自动换来源这种动态决策。于是我换了个思路能不能模拟一个真实的中介事务所前台接单根据业务类型转派给不同部门部门之间可以互相协作干不了的活儿按规定上报最后由一个负责人汇总交付。这套模式本质上就是多Agent编排只是把公司组织架构翻译成了路由表转派协议任务上下文隔离。agency-agents就是这么一套轻量级的编排框架原型核心解决三个问题让每个Agent只干一件专业的事、让转派有明确的协议可循、让跨Agent协作不会把上下文烧穿。这篇文章我会把框架的骨架设计、核心代码、踩坑链路和实测数据都摊开来讲。适合已经在跑单Agent应用、被上下文和流程维护折磨过的开发者参考也适合刚接触多Agent编排、想理解这类框架内部到底在做什么的同学。下面所有实现都是我自己的实验代码不带任何商业包装。2. 先定骨架转派协议、Agent注册表与上下文隔离动手写代码之前我花了一周时间只做设计。因为多Agent框架最怕的不是写不出来而是写出来之后转派像没头苍蝇Agent之间互相踢皮球。所以我把整个骨架拆成三个核心概念协议、注册表、隔离区。2.1 任务信封一切协作都走统一协议跨Agent协作的第一步是定义对话时说什么。我借鉴了消息队列的做法设计了一个叫TaskEnvelope的通用信封结构每次转派都是一个完整的信封而不是一段自然语言。信封字段包括task_id全局唯一任务编号用于链路追踪from和to来源Agent与目标Agentintent转派意图例如research、summarize、reviewpayload携带的数据只允许JSON序列化对象provenance转派路径记录例如[orchestrator, research_agent, writer_agent]deadline任务截止时间为什么不用自然语言对话直接转派我实测过自然语言转派在两次以内还好一旦链路超过三个Agent模型会把任务背景请求内容历史结论混在一起目标Agent根本分不清哪些是给它干活的指令、哪些只是背景资料。信封结构强制每个Agent接收到的都是意图数据的干净组合相当于给每个Agent配了一个标准工单极大降低了歧义。2.2 注册表Agent的能力声明与路由依据第二个关键设计是注册表。每个Agent在启动时向注册表登记三样东西名字、能力标签、转派偏好。能力标签我建议用动词开头的短标签比如fetch_webpage、extract_entities、draft_email而不是多功能助手这种大而全的描述。原因很简单编排器做路由时要用标签匹配标签越精确匹配越稳定。# registry.py AGENT_REGISTRY { research_agent: { capabilities: [fetch_webpage, search_sources, extract_facts], model: gpt-4o-mini, max_payload_size: 8000, }, writer_agent: { capabilities: [draft_article, rewrite_text, summarize], model: gpt-4o-mini, max_payload_size: 20000, }, review_agent: { capabilities: [check_facts, lint_format, detect_bias], model: gpt-4o, max_payload_size: 20000, }, }这里有个我当时没想明白、后来踩了坑才懂的点路由匹配不能只做包含判断。比如research_agent声明了search_sources但writer_agent也可能需要查资料来补充案例如果编排器只按标签包含来选人就会出现两个Agent抢活儿。所以我在注册表里额外加了一个priority字段同一标签下的匹配按优先级排序高优先级的Agent优先低优先级的作为后备。2.3 上下文隔离每个Agent只带任务行李出发这是agency-agents和很多多人对话式框架最大的区别。很多方案让所有Agent共享同一个完整对话历史Agent A说了什么后面所有Agent都能看到。听起来很酷但实际跑起来你会发现上下文窗口烧得飞快而且Agent B会被A的风格带偏。我采用的方案是隔离总结式交接。每个Agent只接收信封里的payload和一份由上一个Agent生成的handoff_summary不超过800字的交接摘要不接收完整历史。Agent干完活之后自己整理输出结果并写一份交接总结交给下一个Agent。这样每个Agent的上下文窗口里都只有我这次需要的数据别人给我的简报我的系统提示词干净利落。打个比方这就像公司里跨部门协作财务部把报表交给市场部时不会把整个财务系统日志都拷过去只会给一份关键数字注意事项的交接文档。上下文隔离让每个Agent都能把注意力集中在自己的专业领域上。3. 核心代码实现编排器、Worker与握手逻辑设计定了之后实现反而比较快。我整个框架只写了三个核心文件agent.pyAgent基类、orchestrator.py转派编排器、protocol.py信封与校验。单机运行后续想扩展成分布式也很容易因为通信协议本身就是JSON。3.1 Agent基类跑任务的标准化接口所有Worker都继承同一个基类基类里封装了接收信封→拼提示词→调用模型→校验输出→生成交接总结五个步骤。这样每个业务Agent只需要实现execute()这一个方法不用重复处理模型调用和错误重试。# agent.py class Agent: def __init__(self, name: str, capabilities: list[str], system_prompt: str): self.name name self.capabilities capabilities self.system_prompt system_prompt async def handle(self, envelope: TaskEnvelope) - AgentResult: # 1. 从信封中取出本Agent需要的输入 payload envelope.payload # 2. 拼装本次调用的提示词payload序列化后截断到max_payload_size user_prompt self._build_prompt(payload, envelope.handoff_summary) # 3. 调用模型带上重试机制最多重试2次 raw await self._call_model(self.system_prompt, user_prompt) # 4. 解析为结构化输出解析失败则抛出AgentOutputError output self._parse_output(raw) # 5. 生成交接总结记录关键结论和为下游准备的数据 handoff_summary self._summarize(output) return AgentResult(agent_nameself.name, outputoutput, handoff_summaryhandoff_summary)设计上的关键点在于_build_prompt()。我要求每个Agent在接收payload之前先做一轮裁切只保留与自己能力相关的字段。比如writer_agent写文章时需要research_agent给的事实清单和来源URL但完全不需要原始网页的HTML实体。裁切规则我直接写在每个Agent的类属性INPUT_FIELDS里哪个Agent要哪些字段一目了然。3.2 编排器转派的闭环逻辑编排器是整条链路的大脑。它的职责是接收初始任务通过意图解析找到能力匹配的Agent按照高优先级→后备的顺序执行处理异常并决定是否需要上升转派。# orchestrator.py async def orchestrate(initial_task: dict, max_depth: int 4): agent route_by_intent(initial_task[intent]) envelope TaskEnvelope( task_iduuid4().hex, fromorchestrator, toagent.name, intentinitial_task[intent], payloadinitial_task[payload], provenance[orchestrator], ) depth 0 while depth max_depth: result await execute_with_retry(envelope) if result.status ok: next_intent decide_next_step(result, initial_task[goal]) if next_intent is None: return compile_final_output(result) next_agent route_by_intent(next_intent) envelope build_next_envelope(result, next_agent, next_intent) else: # 失败上升把错误摘要发回编排器重新路由或结束 envelope build_fallback_envelope(result, envelope) depth 1 raise OrchestrationTimeoutError(ftask {envelope.task_id} exceeded max_depth)这个循环的核心是decide_next_step()它让当前Agent不只产出一个最终结果还要产出一个下一步建议通常是一个意图。比如research_agent查完资料后建议draft_articlewriter_agent写完初稿后建议review_agent做检查。这本质上是让模型自己决定流程怎么走但由编排器做最终裁决避免了越权乱跳。3.3 握手校验转派前的最后一道闸门我在跑通第一版时发现Agent偶尔会生成一个意图但对应的目标Agent不存在或者能力不匹配。比如writer_agent建议转给email_sender但注册表里根本没有这个Agent。所以我在build_next_envelope()里加了一层握手校验转派前先查注册表目标不存在时把当前结果标记为endpoint_unresolved由编排器直接走兜底完成逻辑——用当前Agent的结果作为最终交付不再尝试转派。def build_next_envelope(result, next_agent, next_intent): if next_agent is None: return None # 触发兜底完成 if next_agent.name not in AGENT_REGISTRY: raise HandoffError(funknown agent: {next_agent.name}) return TaskEnvelope( task_idresult.task_id, fromresult.agent_name, tonext_agent.name, intentnext_intent, payloadresult.output, handoff_summaryresult.handoff_summary, provenanceresult.provenance [next_agent.name], )这套校验在正式跑任务之前先帮我拦下了至少二十次无效转派模型随机性导致的幻觉Agent名问题基本被根治了。我的经验是多Agent框架里对模型输出做结构化校验的优先级比调提示词更高。提示词再严谨也拦不住模型的随机性但一个if name not in registry就能让整个流程稳定一个数量级。4. 跑通之后踩到的三个大坑完整排查链路与修复框架写出来、Demo跑通只是万里长征第一步。真正让它能扛住真实任务的是后面那一周填坑的过程。这里三个坑是我印象最深的我把排查链路完整写出来因为只看结论不看过程的话下次换个场景你还是会踩。4.1 上下文膨胀交接摘要越传越大token翻了三倍症状第50个任务之后我统计token消耗发现每个任务平均消耗比预期高出约3倍。仔细看日志writer_agent收到的手工摘要越来越长有的甚至把原始网页正文复制了过来。排查链路第一步我给每次交接加了一个summary_length日志字段把每个任务的交接摘要长度打点。数据出来之后发现摘要长度呈阶梯式上涨第一个Agent写了300字第二个Agent基于第一个的摘要再加自己的输出写成800字第三个Agent再堆到2000字。第二步我看具体内容发现疫情是Agent把交接摘要理解成了汇总报告每经过一个节点就把上一轮的所有信息重新复述一遍。第三步定位根因_summarize()方法的提示词没有明确只写本节点新增结论不重复历史模型当然按自己的惯性来。修复我在_summarize()里加了一个硬约束输出必须严格符合新增结论 下游需要注意的风险 建议下一步三段式并且用代码强制截断到800字超了就按关键段截断。同时把压缩历史的职责从Worker上移到编排器每次转派前编排器对provenance路径上的旧摘要做一次合并只保留最新一份。改完之后摘要长度稳定在400到800字token消耗恢复到了预期水平。这个坑给我的教训是多Agent框架里对话历史的膨胀不是靠增大上下文窗口解决的要靠协议约束。4.2 循环转派A找B、B又找A任务永远结束不了症状某个整理行业报告的任务跑了一个多小时还没结束token消耗持续增长。我看日志发现research_agent转给writer_agentwriter_agent又转回research_agent两个Agent来回踢了十几轮。排查链路我先给每个信封加了provenance路径打印日志里清晰地看到orchestrator → research → writer → research → writer → ...的交替循环。第二步我把两个Agent的decide_next_step()原始输出拉出来看它们到底为什么互相推荐发现research_agent的提示词里写了资料不够充分时请交给其他人补充writer_agent的提示词里写了遇到不确定的事实请返回给研究人员核实。这俩规则单独看都没问题合在一起就成了死循环。修复我做了三重保险。第一编排器中硬性规定max_depth 4超过直接结束、返回当前最优结果并附带cycle_warning。第二在provenance里检查循环检测如果某个Agent在路径中出现两次接下来禁止再次转派给它只能转给新Agent。第三也是最根本的我修改了两个Agent的提示词把请交给其他人改成了请列出具体缺失信息并缓存为待办禁止在本次流程中再次请求同一对象。事实上第三点才是解决循环的关键前两重只是兜底止血。这个坑让我意识到Agent之间的协作规则本质上是一套协议协议里必须有禁止回路的显式条款。4.3 工具调用失败模型幻觉出根本不存在的工具名症状research_agent尝试抓取网页时连续三次返回同一个错误错误信息是调用工具fetch_page_from_url失败原因找不到该工具。但我注册表里明明定义的是fetch_webpage。排查链路第一反应是改提示词把工具名清单再强调一遍结果失败率只降了一点点。第二步我打断点看模型原始输出发现模型返回的内容里工具名是对的但参数结构里url字段被写成了Url——系统提示词里用的是小写开头模型却按Python类名的习惯大写。工具解析器拿到Url之后找不到对应参数就报了个找不到工具。真正的问题不在工具名而在参数Schema的容错性。修复我写了一个容错解析层在把模型输出传给工具执行器之前做一轮字段名归一化把首字母大写的字段转成小写驼峰遇到多余字段直接忽略而不是报错必填字段缺失时才重试。改完之后工具调用成功率达到97%以上剩下的3%靠一次重试基本能消化。这个坑给我的经验是工具调用的失败根因大多不在模型能力而在你的解析器太脆弱。做好Schema校验和字段容错比反复调提示词高效一百倍。4.4 三个坑之外的隐形坑并发与超时同一条链路里如果多个任务并发跑共享注册表倒是没问题但共享同一个openai客户端的连接池就会出问题并发一高偶尔出现连接重置。我把客户端的max_connections调大并给每个任务加了独立的超时控制默认60秒超时触发重试。这个改动看起来不起眼却是上线之后最救命的一个配置。编码时注意一点编排器的循环里每轮转派都要单独算超时而不是整个任务算一个总超时否则前一个Agent卡死会占用后面所有环节的时间。5. 一百轮实测后的调优清单数据、取舍与最终效果修完三个大坑之后我拿agency-agents跑了一百个不同类型的任务做压测包括资讯整理、邮件草拟、数据提取、文案润色。我把任务分成两组A组用单Agent加很长系统提示词B组用这套多Agent框架。结果有几个数据让我印象很深。5.1 核心指标的对比指标单Agent长提示词agency-agents多Agent任务成功率按交付可用度判定71%93%平均单任务耗时18秒32秒平均token消耗42005800最大上下文长度接近窗口上限稳定在11000以内格式异常率字段缺失/JSON解析失败14%3%成功率提升22个百分点代价是耗时和token消耗各增加约40%。这个取舍我认为完全值得因为多Agent的价值不在省钱而在把复杂任务跑通。单Agent做不了的事多Agent能做单Agent能做的事多Agent做得更稳。如果你的任务场景非常简单一次调用就能完成完全没必要上多Agent框架纯属浪费token。5.2 几个让我意外的调优结论第一给每个Agent减少系统提示词效果反而更好。research_agent最初的系统提示词有1200字我把所有背景解释删掉只留职责工具表输出格式缩到400字它的指令遵循率从89%升到96%。模型需要的不是长篇大论的解释而是清晰的边界。第二交接摘要的质量直接影响下游效果。我发现writer_agent写出的事实性错误八成不是因为模型本身而是因为research_agent的交接摘要里混淆了事实和推测。后来我要求研究Agent在摘要里用[FACT]和[INFER]显式标注信息类型下游写作时的错误率立刻下降了一半。这个技巧成本为零收益却相当可观。第三失败时要让Agent看到真实错误信息。最初我在重试时只告诉模型上次执行失败了请重试结果它反复给出同样的错误做法。改成把异常堆栈和工具返回的错误原文喂回去之后模型自己就会调整参数。AI Agent的容错能力很大程度上取决于你给它的反馈信息质量。5.3 框架可以继续扩展的方向agency-agents目前还只是个原型我自己列了几个后续想做的方向把注册表改成动态注册支持在业务运行时添加新Agent给编排器加一个成本预算参数token花超就自动降级到轻量模型把转派协议扩展成支持异步回调这样上游Agent可以先交付一个暂存结果等下游完成后异步更新最终输出。这些方向里我最推荐优先做动态注册因为多Agent系统的维护成本主要来自新增一个Agent要改编排器而动态注册能把这种耦合彻底解开。最后分享一个我在填坑过程中形成的小习惯给每个Agent跑日志时加一个provenance字段的断言每轮转派前断言路径无重复、长度未超限。这套断言在上线初期帮我抓出了至少五个隐藏的循环转派场景。多Agent框架的复杂度很大一部分在链路治理上能用代码约束的东西就别依赖模型自觉这是我在这个项目里最深刻的体会。
