“agency-agents”这个词我第一次真正重视起来是在一个跨部门协作的项目里。当时我们接到需求要搭建一套内容审校系统把一篇长文拆给不同角色去处理——有人查事实、有人挑逻辑、有人补案例、还有人做最终风格统一。一开始试的是“超级单体Agent”上下文几千行单次回答又长又乱效果非常不稳定。后来切换到“agency-agents”思路——不是让一个智能体做完所有事而是像搭建一个虚拟内容公司那样让多个角色化、职责互斥的Agent在同一套机制里有次序地协作。效果立刻不一样了。这篇文章就来拆解这类“智能体机构”的设计核心、工程实现和实战避坑适合正在做多Agent系统、或者准备从单Agent往复杂协作分布式架构迁移的开发者参考。1. agency-agents的核心设计思路拆解1.1 先搞清楚它解决的到底是什么问题很多人在刚接触“agency-agents”时第一反应是“多Agent嘛我跑几个Agent并行调用不就行了”。实际远不止于此。单Agent的能力边界在三个地方会被卡死上下文窗口不够用、单一思维链路容易漏事实、以及没有“校对/制衡”机制导致错误会一路滚下去。我见过最典型的失败案例是让一个Agent同时做资料检索、观点提炼、段落改写和数据校验。它既要理解业务背景又要输出严谨结论还得时刻记得约束条件。结果就是上下文越堆越长前面的关键指令到后面已经被稀释掉模型开始“自由发挥”。agency-agents的核心思路是把“全能者”切换成“专业团队”每个Agent只做一类事职责边界清晰任务通过结构化消息传递。这种设计解决的问题非常直接上下文隔离每个Agent只处理自己领域的输入输出不背负全流程信息错误抑制下游Agent可以检测上游输出问题形成制衡并行与顺序可编排可以按依赖关系安排任务哪些环节串行、哪些并行可以自由控制可追踪可回放每步都有独立结果定位出错环节很容易1.2 关键词拆解Agency不是“群聊”是“机构”英文里“agency”容易让人误解为“代理”的复数其实不对。它更接近“代理机构”。这意味着你设计的不是一堆Agent乱聊天而是它们遵守相同的协作协议、有明确汇报路径、有任务工单流转。我在设计系统时一直用一套“虚拟公司”类比老板AgentPlanner负责拆任务执行岗AgentWorker负责具体活儿审核岗AgentCritic负责挑错还有一个调度器Dispatcher维护一个任务队列决定下一个任务分给谁。这个类比在向业务方解释方案时特别管用因为他们不需要理解模型原理只需要理解“有人想方案、有人干活、有人质检”。关键的组成要素有四个要素作用类比角色定义明确每个Agent的职责边界和目标岗位JD任务工单结构化描述做什么、输入什么、期望输出什么派工单消息路由确定结果传给谁、什么时候传内部工作流反馈回路下游对上游提出修改建议审校意见1.3 为什么即使有了大模型仍需要这套结构直接给一个超强提示词行不行行但不可持续。我实测过一个场景把任务拆成四步交给同一个Agent去做前三步结果都正常到第四步需要引用前文数据时它开始胡编。后来我换成三个不同的Agent实例协同每个只做一步且都拿到上一步的结构化输出错误率降低了将近一半。原因在于模型的注意力机制决定了长上下文的利用率是递减的。而Agency这种结构本质上是在用系统设计弥补模型短板——不让模型“记太多”只让它在有限范围内做到专注和可靠。2. 角色职责划分与编排策略2.1 最小可行角色集怎么定刚开始搞agency-agents的人特别容易犯一个毛病角色设得过多甚至一个Agent专门负责“生成标题”另一个专门负责“生成副标题”。这会造成大量的信息传递损耗和接口开销。我的经验是最小可行系统先配四个角色Planner理解用户目标输出一份任务分解方案Executor按方案执行单一子任务输出结果片段Reviewer审核执行结果是否符合要求输出通过/修改建议Aggregator把所有通过的结果组合成最终产出这四个角色基本覆盖了“拆解—执行—验收—汇总”的完整闭环。如果任务领域比较特殊再按需增加辅助角色比如RAG场景里的Retriever、代码场景里的Runner。2.2 编排模式顺序、并行、还是协商编排方式决定了整个系统的吞吐量和稳定性。我常用的三种模式顺序管线Pipeline适合有强依赖的任务比如“先生成提纲再写正文再校勘”扇出汇聚Fan-out/Fan-in适合子任务相互独立的方式比如“多路并行搜集素材最后统一汇总”协商评审Critic loop适合对质量要求极高的场景比如“执行Agent产出评审Agent打回循环直到通过”我见过有人把协商评审搞成无上限循环结果系统在半夜跑了几百轮仍没有结果。解决办法是设置最大评审轮次比如三到五次超出后自动降级采用当前版本并打上“待人工复核”标记。2.3 如何选择合理的编排方案选择编排模式前先画出任务依赖图。凡是存在上下游数据依赖的位置必须用顺序没有依赖的全部设计成并行。这里有一个实操技巧先用文字写下“每步需要什么输入、产出什么输出”判断依赖关系时画一个简单的二维矩阵行是输入列是输出非对角线有值则说明有关联要串行排队。实际上后端实现时再使用dag库或图数据库来承担依赖表达。3. 工程实现一套可落地的agency-agents最小系统3.1 环境与基础组件准备这一步主要是选择语言和基础框架。我现在常用的技术栈是Python 异步事件循环 消息队列本地开发用内存队列即可生产可以替换成Redis Stream或RabbitMQ LangGraph或自研状态机。如果你是从零开始优先推荐自研一个轻量调度内核因为框架层抽象太高时查问题反而困难。需要的外部依赖就只有大模型API封装业务逻辑用Pydantic定义消息体。消息结构非常关键错误的大多数来源于消息字段定义不清导致下游Agent不知道从哪里拿数据、拿到的数据是什么类型。3.2 消息协议和角色基类设计我在这里分享一个核心代码骨架已经过简化但结构完整。首先定义消息类型from datetime import datetime from typing import Optional, Any, List from enum import Enum from pydantic import BaseModel, Field class TaskStatus(str, Enum): PENDING pending RUNNING running REVIEWING reviewing DONE done FAILED failed class TaskMessage(BaseModel): task_id: str Field(..., description全局唯一任务标识) task_type: str Field(..., description任务类型对应Agent角色职责) payload: dict Field(default_factorydict, description实际任务内容) meta: dict Field(default_factorydict, description元数据优先级、超时时间) status: TaskStatus TaskStatus.PENDING created_at: datetime Field(default_factorydatetime.now) updated_at: datetime Field(default_factorydatetime.now) class WorkResult(BaseModel): task_id: str status: TaskStatus output: Any None error: Optional[str] None review_comment: Optional[str] None然后定义一个Agent基类所有具体Agent只要继承并实现run方法即可from abc import ABC, abstractmethod class BaseAgent(ABC): name: str base def __init__(self, model_client, system_prompt: str): self.model_client model_client self.system_prompt system_prompt abstractmethod async def run(self, message: TaskMessage) - WorkResult: 处理传入的任务消息返回结构化结果 pass async def call_model(self, prompt: str, **kwargs) - str: 封装对大模型接口的调用支持超时与重试 response await self.model_client.completion( messages[ {role: system, content: self.system_prompt}, {role: user, content: prompt}, ], **kwargs, ) return response.choices[0].message.content这样设计的初衷很简单把Agent变成“输入输出都是结构化消息”的纯函数。任何一个Agent都可以在单独进程中测试不需要和其他Agent耦合推理。3.3 Planner、Executor、Reviewer、Aggregator 四个角色的实现示例Planner的system prompt要写得很具体要求它输出JSON格式的任务分解方案并且明确指定每个子任务的类型和数据依赖。PLANNER_PROMPT 你是一个项目拆解专家。请根据用户提供的目标生成一份可执行的子任务计划。 要求 1. 输出JSON数组数组每个元素包含task_type搜索/写作/审校/汇总、description、dependencies依赖哪些任务ID。 2. 每个子任务必须边界清晰不得重复描述同一件事。 3. 子任务数量控制在3-8个不要拆得过于碎片化。 4. 返回示例 [{task_id: t1, task_type: search, description: 查找2024年行业数据, dependencies: []}] 只返回JSON不返回其他文字。 Executor采用通用实现根据task_type内部分发到不同处理函数class ExecutorAgent(BaseAgent): name executor def __init__(self, model_client): system_prompt 你是具体任务的执行者。只根据输入完成任务不要跳出职责。按要求输出内容。 super().__init__(model_client, system_prompt) async def run(self, message: TaskMessage) - WorkResult: task_type message.payload.get(task_type) if task_type search: # 此处可调用内部检索服务 content f模拟检索结果{message.payload[description]} elif task_type writing: content await self.call_model(f请撰写一段内容。要求{message.payload[description]}) else: content await self.call_model(f完成任务{message.payload[description]}) return WorkResult(task_idmessage.task_id, statusTaskStatus.DONE, outputcontent)Reviewer的职责是发现问题并给出可修改建议。这里给Agent设置了明确的否决/通过机制在prompt中写明输出格式REVIEWER_PROMPT 你是质量审核员。请对执行结果进行审核。 如果通过回复{pass: true, comment: 通过}。 如果不通过回复{pass: false, comment: 具体修改建议要求可执行}。 审核维度事实准确性、逻辑一致性、格式合规性、与目标的相关性。 只返回JSON。 Aggregator的最后一步组装则简单一些把所有通过审校的子结果按既定模板拼接。3.4 调度内核和任务编排调度器是整个系统的核心我常将其设计成一个异步循环。伪代码如下import asyncio from collections import deque class AgencyScheduler: def __init__(self, agents: dict[str, BaseAgent]): self.queue deque() self.agents agents self.results {} async def submit(self, message: TaskMessage): self.queue.append(message) async def run_until_empty(self, max_review_rounds3): while self.queue: message self.queue.popleft() agent_name message.payload.get(assigned_agent) agent self.agents[agent_name] result await agent.run(message) self.results[message.task_id] result # 如果审校不通过并且没有超出最大轮次则重新投递到执行Agent if result.status TaskStatus.REVIEWING and result.review_comment: if result.meta.get(round, 0) max_review_rounds: new_task TaskMessage( task_idmessage.task_id, task_typemessage.task_type, payloadmessage.payload, meta{...}, ) self.queue.append(new_task) # 如果通过则投递到Aggregator # ...真实线上运行的时候我会额外加上任务超时控制、Agent异常重试、每个任务的生命周期日志。用状态机来防止消息重复进入同一位置。3.5 错误处理与回退策略系统跑起来后最怕的不是模型答错而是接口超时或返回格式解析失败。我的策略是所有非预期格式的输出统一定级为“可重试错误”最多重试两次两次后标记为FAILED并由聚合器统一收集到“待人工处理批次”。Reviewer否决后重新投递的任务在meta中带round计数字段防止无限循环超过最大轮次的直接用当前版本加“quality warning”字段而不是阻塞全流程。这里要特别注意在prompt中强调“如果结果已经比上一版好允许通过”避免系统为了改而改。4. 常见问题与排查技巧实录4.1 任务死循环最典型的“卡死”现场现象系统在某个子任务上反复执行队列数量不降日志显示同一task_id不停流转。排查方法先看Reviewer返回的comment。如果comment是空泛的“不够好”“需要提升”大概率是prompt中缺少“可执行修改意见”的硬性约束。我在实际调试时会把Reviewer的输出打印出来发现模型有时候根本不按要求输出JSON而是回到自然语言。这个问题的根子是模型“不知道该给什么”此时给一个修改模板例如“请补充XX方面的数据至少300字并在结尾总结”循环会明显收敛。另一个原因是某个Agent的system prompt过于宽松导致无论收到什么输入都输出同一类结果。这时需要检查各角色的prompt是否互相冲突。4.2 上下文溢出和信息丢失扇出汇聚模式下常见的坑运气好时任务细分到十几路并行但汇总时Aggregator要把十几段结果全部放进prompt一次请求就超长了。解决方案有两个方向截断策略设置每段输入的最大长度超出部分摘要化分段汇总聚合器分层先小聚合再大汇总避免单次调用吞太多内容实际操作里我更偏向“分段汇总”因为摘要化会损失细节。比如50个子结果先分成5组每组10个生成5个中间汇总再汇总这5个中间结果最终输出。这样每次调用上下文都在可控范围内。4.3 数据质量下降但不报错多Agent跑一段时间后质量会悄悄下降表现为输出变得平淡且模式化。这个问题的根源通常是自我反馈机制失效了——Reviewer开始习惯性放行。我的应对办法是“交叉审校”每隔几个任务轮换一次Reviewer的system prompt或者在系统中加入随机抽查角色随机对通过的任务重新质检。这个“抽查者”角色的prompt可以这样写“你现在不是审核员而是一个怀疑一切的第三方。寻找三位不同的独立的证据支持或反驳这段输出。” 这样既能打破审美疲劳又能有效拦截质量滑坡。4.4 排查清单速查表症状可能原因先看哪里任务卡住不推进依赖关系成环调度器日志里的task_id流转输出JSON解析失败prompt没约束好Agent调用模型前的原始返回审校一直不通过修改意见不可执行Reviewer的comment文本汇总结果丢失细节单次上下文过长Aggregator输入截断策略系统整体变慢串行依赖过多并行度统计画出依赖图质量逐渐下降反馈机制失效Reviewer通过率曲线我给自己的代码里加了一个简易的“通过率监控”模块每五十个任务统计一次Reviewer平均通过率。如果通过率长期高于90%说明审校已经变成走过场低于30%说明执行Agent和审校Agent出现系统性矛盾。这两个信号都需要人工介入。5. 评估、调优与成本控制的实战经验5.1 从哪些维度评估一套agency-agents系统不要只看“任务完成率”。我建议至少记录这四个维度成功率任务在最大轮次内完成并输出的比例返工率需要Reviewer打回修改的任务占比单任务交互轮数平均每个任务在Agent之间传递的次数上下文吞吐量每个任务消耗的token总量按角色拆分统计这四个维度结合起来才能反映系统是否健康。返工率不是越低越好适当返工会提升最终质量但轮数太多就说明拆解阶段做得不好。5.2 成本优化的典型方案token统计通常会发现一个惊人事实Reviewer角色消耗的token往往比执行者还多因为它每次都要读全文有时还要反复读。降低成本的办法按优先级排列任务粗粒度化减少不必要的子任务拆分结构化输出强制JSON输出避免模型啰嗦解释局部审校不让Reviewer看全文只传关键段落混合模型策略执行Agent用较强模型审校Agent可以用相对低成本模型最后这一点在实践中很管用。比如规划任务和内容生成使用高能力模型审校阶段可以先用中等能力模型做一轮粗筛只有标记了“通过”或“存疑”的结果再升级到高能力模型复评。实测这样的混合策略能在保持质量的同时减少约30%到40%的高规格调用量。5.3 一个实测中的调优记录在某个模拟项目X的内容生产系统中初始配置的通关体验很差一个任务平均要跑七轮才结束成本也超预算。定位后发现Planner把“内容生成”拆得太细连续拆出“标题生成”“导语生成”“正文生成”“结尾生成”四个子任务而每个都要调用一次模型成本高且冗余。后调整拆解策略把相关度高的步骤合并三段式结构开头—主体—结尾合并为一个任务内完成。子任务总数从7个降到4个平均轮数从7降到3成本下降约50%而人工评分显示最终质量没有下降。这个经验说明agency-agents系统的瓶颈往往不在模型而在任务建模。任务建模越干净系统越省力。5.4 可观测性把每一次Agent行为都记录下来多Agent系统调试最痛苦的是“不知道中间发生了什么”。强烈建议从第一天开始就记录结构化日志至少包含任务ID、Agent名称、输入摘要、输出摘要、耗时、token用量、审校结论。开发的时候可以把每次调用都存到本地JSON文件生产环境则直接放到日志系统里。日志里多一个“为什么”字段每个Agent在返回结果时必须额外提供一两条理由比如“因为用户要求包含数据所以我补充了图表”。这样排查时能快速判断是执行错误还是目标理解偏差。这个小习惯帮我省了大量排查时间。6. 从“能跑”到“好用”的最后一公里代码能跑通只算完成了40%。剩下的是工程化的活儿提示词版本管理、Agent配置中心、灰度发布。我现在会把所有角色的system prompt放在一个配置目录里用Git管理版本。每次改动Prompt必须顺手跑一遍回归集把前后结果对比防止“改好了一个角色搞垮了另一个”。在实际落地过程中我发现最值得投入的是评估数据集。准备二十到三十个典型任务固定好输入作为回归基线。任何Prompt修改、任何模型切换、任何编排策略调整先跑一遍基线看成功率、返工率和成本三个指标是否恶化。一个没有基线的多Agent系统改来改去纯靠感觉最终一定会在线上突然崩掉。关于“agency-agents”这个方向我自己踩过最大的坑是早期过于迷信模型的推理能力把所有决策全部交给模型结果发现模型的随机性会导致系统行为不可预期。后来逐渐把关键决策逻辑用代码固化下来比如调度、重试、超时、降级、轮次控制这些都不让模型判断只有内容生成和语义判断才交给模型。这套“代码控制流程、模型控制内容”的混合架构是我做了多个项目之后最认可的形态。以后如果要做复杂业务接入还有两件事值得投入一个是可插拔的角色注册中心让业务方能在界面上配置新Agent另一个是更细粒度的预算控制让每个任务在启动前就明确自己的token上限。这些细节会让系统从“实验室玩具”真正变成“生产工具”也希望你少走一些我当年走过的弯路。