做多智能体项目时有一个很微妙的阶段单个 Agent 已经能跑通但一旦让它们协同工作各种问题就来了——上下文互相污染、任务反复推翻重做、对话绕圈子完全不像一个团队倒像几个新手在抢同一台电脑。这个名为 agency-agents 的项目本质上就是针对这类问题做的一套多智能体协作框架。它的核心思路不是去发明更聪明的 Agent而是把个体智能的问题转化为组织协调的问题像搭一个机构一样去搭 Agent 系统每个 Agent 有明确的岗位职责、汇报关系和协作流程靠结构和制度来弥补单个模型的不确定性。这篇文章会从设计思路拆解到代码实现再到调试过程中踩过的坑完整记录这套框架是怎么搭起来的以及它适合解决什么样的问题。如果你现在刚准备入手多智能体开发或者已经在做但总觉得协作效果不稳定这篇内容应该能帮你省掉不少弯路。1. 整体设计为什么要把 Agent 组织成机构而不是聚会1.1 单智能体到多智能体问题性质的变化单个 Agent 的本质是一个输入到输出的高智能映射你给它一个任务它借助大语言模型的推理能力拆解步骤、调用工具、产出结果。这种情况下质量瓶颈主要在模型能力和提示词设计上属于个体能力问题。但一旦有了多个 Agent情况就不同了。Agent 与 Agent 之间的交互会产生组合爆炸消息怎么传递、先后顺序怎么定、意见冲突时听谁的、失败时谁来兜底这些都不再是提示词能简单解决的。我见过不少项目把几个 Agent 丢在一起结果就是它们反复确认同一件事把 Token 白白烧掉最后给出的答案还不如单 Agent 直接跑一遍。这不是模型不够聪明而是组织方式出了问题。多智能体系统最大的风险不是单个 Agent 变笨了而是整个系统的混乱度呈指数上升。1.2 机构化思路用制度对抗不确定性agency-agents 这个名字英文里 agency 既有代理的意思也有机构的意思。这个项目刻意取了双关一方面它是一个代理Agent的集合另一方面它尝试把真实的组织管理经验移植到智能体系统里。真实机构是怎么保证一群能力参差不齐的员工产出稳定结果的靠流程、分工、协作规范。不是每个人都要全能但每个人都要清楚自己的职责边界、汇报对象和交付标准。这套逻辑放到 Agent 系统里是成立的而且很有效。具体来说这套框架的核心是三层结构顶层是调度层相当于项目经理负责任务接收、拆解和分发。中间是执行层由多个专职 Agent 组成每个 Agent 只负责一种类型的任务。底层是工具层封装了模型调用、数据读写、外部 API 等能力对上层透明。调度层不直接干活但掌控全局执行层不思考全局但深耕局部。这种分工方式有两个直接好处一是每个 Agent 的提示词可以做得非常聚焦不需要塞入一大堆上下文来维持全局观降低了提示词复杂度二是排查问题时定位非常快某个环节出错直接找对应的执行 Agent而不是在一条超长的提示词里层层推理。1.3 什么场景适合这种架构不是所有多 Agent 需求都要上这种框架。我自己的判断标准很简单如果任务链条短、步骤固定、依赖关系简单用传统的工作流引擎加单个模型调用就够了上多 Agent 反而徒增开销。agency-agents 适合的场景通常是这几个特征任务可以明显拆分为多个专业模块例如信息收集 - 分析 - 文案生成三段式。各模块之间需要传递中间结果而不是一次性给出完整需求。任务执行过程中可能出现不确定性需要反复修订例如生成的内容需要经过质量检查和返工。有并行处理需求多个独立模块可以同时推进。反过来如果任务是给出一段文本翻成英文这种单步骤任务硬拆成多 Agent 是完全没必要的纯属自找麻烦。2. 核心细节角色编排、消息协议与上下文隔离2.1 角色怎么划分才算合理角色划分是这套框架里最重要也最容易被忽视的一步。很多人喜欢把 Agent 划分得很细什么意图识别 Agent情感分析 Agent代码生成 Agent最后项目变得极其臃肿。我的经验是角色划分的粒度应该跟任务的专业化程度对齐而不是跟功能清单对齐。一个有用的标准如果一个子任务交给单个 Agent 直接处理时你仍然需要大量提示词来控制它的行为说明这个 Agent 的角色还太宽需要继续拆分如果一个子任务用一段简单的提示词就能稳定完成说明它已经足够聚焦再拆反而增加通信开销。以这个模拟项目里的实际案例来说一个内容生产流水线被拆成了四个角色资料收集者、内容策划者、撰稿人和质量审查人。资料收集者只输出事实信息列表不输出任何判断内容策划者只输出大纲和观点不写具体内容撰稿人只负责把大纲扩展成完整文稿质量审查人只输出问题和修改建议不直接改稿。每个角色的输出边界都被严格约束这样审查人的反馈才能准确落到撰稿人头上而不是一团混战。2.2 消息格式结构化比自然语言可靠多个 Agent 之间通信最自然的想法是让它们直接用自然语言对话但实际操作中发现并不可靠。自然语言本身有歧义模型生成的自然语言更甚。同样一句话不同模型对它的理解可能相差很远。而且自然语言消息很难做程序化校验一个 Agent 漏传某个字段另一个 Agent 也不会主动发现只会硬着头皮往下跑。因此在这个框架里Agent 之间传递的消息被强制定义成了结构化格式。每条消息包含三个固定字段发送者、接收者、任务类型以及一个 JSON 格式的数据载荷。任务类型决定了接收方应该用哪个处理函数去解析数据载荷则承载实际的业务数据。这样一来消息本身就带有了足够的元信息接收方不需要靠猜来理解对方意图而可以像解析 API 请求一样处理消息。结构化消息还有一个好处就是让日志变得非常可读。调试时打开消息流水一眼就能看出哪个 Agent 在什么时候给谁发了什么消息、数据长什么样。这在排查多智能体问题时能省下大量时间比盯着对话记录猜要直观得多。2.3 上下文隔离每个 Agent 只该看到自己需要的多智能体系统里最隐蔽的问题是上下文污染。当多个任务在同一个对话环境中流转时模型很容易把不同任务的信息混在一起造成输出逻辑混乱。用户明明只问了 A 项目的数据模型却把 B 项目的结论也带了进来这样的系统看起来懂了很多实际上非常危险因为输出结果不可控。agency-agents 的做法是物理隔离每个执行 Agent 维护自己独立的上下文存储调度层只传递必要的结构化数据绝不把整个历史对话倒给下一个 Agent。就像真实机构里财务部门不需要看到销售部门的全部通话记录只需要拿到对方签好的合同金额。这样既减少了 Token 浪费也避免了模型被无用信息干扰。实际操作中这个设计带来的效果非常明显。同样一个数据统计需求上下文隔离前模型时不时会把上一次任务的旧数据混进结果里隔离之后只要保证传入的数据载荷是干净的输出就基本不会再出现张冠李戴的情况。2.4 任务状态机让系统知道自己走到哪一步了多智能体系统很容易陷入一种走一步看一步的被动状态尤其是使用链式调用的时候。Agent A 返回结果Agent B 拿到后继续处理整个过程没有一个明确的进度概念。一旦中间环节失败或者结果异常整个链路就卡住了你也说不清卡在哪里。为了避免这种局面这套框架给每个任务引入了一个简单的状态机待处理、执行中、已完成、已失败、需修订。调度层维护一张任务表任务每流转一步状态就更新一次。这样做直接带来的好处是任何时刻你都能准确知道系统忙在哪、卡在哪而不是面对一条长长的日志大海捞针。状态机还让任务重试和结果修订变得规范化。某个 Agent 输出不符合要求时不需要把整个链路推翻重来只需要把对应任务的状态改成需修订重新放回调度队列让对应 Agent 针对修订意见再做一轮即可。这比整个流程从头跑一遍要高效得多尤其是在任务链较长的情况下。3. 实操过程从零搭一个最小可用的多 Agent 协作系统3.1 环境准备与框架选型这个项目的语言选型是 Python主要是生态成熟模型调用库多而且团队更熟悉。如果是从头搭建类似项目的话依赖很简单一个支持工具调用的模型 SDK加上 Python 内置的 asyncio 用来做并发调度就够了。项目初期不需要引入重量级框架避免被框架约束了思路等跑通后再考虑抽象成通用库。代码结构也刻意保持简单核心目录只有三个agents/放置各个执行 Agent 的实现每个 Agent 一个文件。core/消息协议、状态机、上下文存储这些基础设施。scheduler/调度器负责任务的接收与分发。下面是我在这个模拟项目中实际使用的调度器核心代码骨架去掉了业务细节保留了关键逻辑。import asyncio import json from typing import Dict, Any class TaskState: PENDING pending RUNNING running DONE done FAILED failed NEEDS_REVISION needs_revision class Message: def __init__(self, sender: str, receiver: str, task_type: str, payload: Dict[str, Any]): self.sender sender self.receiver receiver self.task_type task_type self.payload payload def to_dict(self) - Dict[str, Any]: return { sender: self.sender, receiver: self.receiver, task_type: self.task_type, payload: self.payload, } class Scheduler: def __init__(self): self.agents {} self.task_table {} def register_agent(self, name: str, agent): self.agents[name] agent async def dispatch(self, task_id: str, message: Message) - Any: self.task_table[task_id] TaskState.RUNNING agent self.agents.get(message.receiver) if not agent: self.task_table[task_id] TaskState.FAILED raise ValueError(fAgent {message.receiver} not found) try: result await agent.process(message) self.task_table[task_id] TaskState.DONE return result except Exception as e: self.task_table[task_id] TaskState.FAILED raise e调度器本身不执行业务逻辑它只做三件事找到消息接收者对应的 Agent、调用它、记录任务状态。这个设计保持了调度层的轻量也让新增 Agent 变得非常简单——只需要实现process方法并注册到调度器里。3.2 Agent 基类与执行器实现所有执行 Agent 都继承同一个基类。基类负责消息解析、模型调用、错误包装这些公共逻辑子类只需要实现核心的业务处理函数。class BaseAgent: def __init__(self, name: str, model_client): self.name name self.model_client model_client self.context_history [] async def process(self, message: Message) - Any: raise NotImplementedError async def call_model(self, system_prompt: str, user_content: str) - str: self.context_history.append({role: user, content: user_content}) response await self.model_client.chat( systemsystem_prompt, messagesself.context_history[-8:], # 控制上下文长度 ) self.context_history.append({role: assistant, content: response}) return response调用模型时只保留最近八条消息这是个非常粗暴但有效的上下文截断策略。它能防止上下文无限膨胀同时因为前面做了上下文物理隔离这个截断基本不影响结果质量。如果你想优化模型对长上下文的理解可以换成更精细的摘要机制但在这个项目规模下截断策略已经够用了。以资料收集 Agent 为例它的process方法是这样实现的class ResearchAgent(BaseAgent): async def process(self, message: Message) - Any: query message.payload.get(query) system_prompt ( 你是资料收集员。请围绕用户给出的问题检索并整理相关信息。 只输出事实性信息列表每条信息包含来源、时间、关键内容。 不要输出任何观点、结论和建议。 输出格式为JSON数组每个元素包含 source, time, summary 三个字段。 ) raw await self.call_model(system_prompt, query) return self._parse_json(raw)可以看到这个 Agent 的提示词非常聚焦约束了输出范围、输出格式还严格限定了不要输出观点。限制越清晰模型的输出越稳定后续的策划 Agent 拿到数据时也就越容易处理。3.3 多轮协作任务拆分与结果汇总示例接下来用一个具体的场景把各模块串起来。假设任务是撰写一篇关于本地社区健身需求的分析报告。调度层收到这个任务后会把它拆成三个子任务资料收集查找社区健身相关的公开数据、政策文件和案例。内容策划根据收集到的资料提炼核心观点形成报告大纲。撰稿执行按照大纲撰写完整报告。调度层拆完后依次向三个 Agent 派发消息。如果前一步结果不合格比如策划 Agent 发现收集到的数据不足以支撑观点它会给调度层返回一条需修订的消息附带缺什么信息的具体说明。调度层把对应任务状态更新为NEEDS_REVISION再把修订请求发回资料收集 Agent让它补查数据然后重新触发策划任务。下面这段代码模拟了调度器如何编排这两轮协作import asyncio async def run_pipeline(scheduler: Scheduler): # 任务一资料收集 research_msg Message( senderscheduler, receiverresearch_agent, task_typecollect, payload{query: 本地社区健身需求的相关信息, depth: medium}, ) research_result await scheduler.dispatch(task_collect, research_msg) # 任务二内容策划依赖任务一结果 plan_msg Message( senderscheduler, receiverplanning_agent, task_typemake_plan, payload{data: research_result, target: 社区健身需求分析报告}, ) plan_result await scheduler.dispatch(task_plan, plan_msg) # 如果策划结果要求补料则回到任务一 if plan_result.get(needs_more_data): supplement_msg Message( senderscheduler, receiverresearch_agent, task_typecollect, payload{query: plan_result[missing_query], depth: high}, ) extra_data await scheduler.dispatch(task_collect_extra, supplement_msg) plan_msg.payload[data].append(extra_data) plan_result await scheduler.dispatch(task_plan_revise, plan_msg) # 任务三撰稿 write_msg Message( senderscheduler, receiverwriter_agent, task_typewrite_report, payload{outline: plan_result, style: professional}, ) final_report await scheduler.dispatch(task_write, write_msg) return final_report这个流程最值得注意的地方是任务 ID 的写法。task_collect和task_collect_extra虽然都是让同一个 Agent 做资料收集但它们是两个独立任务状态互不影响。这样即使补料环节失败系统也能明确知道失败的是补料这一步而不是把整个流程标记为失败。3.4 质量审查环节把返工变成正规流程多智能体系统输出质量不稳定是常态如果只管生成不管审查最终结果往往达不到可用标准。因此这套框架里增加了一个独立的审查 Agent专门做质量把关不参与内容生成。审查 Agent 的输入是撰稿人产出的完整报告加上一份质量检查清单。它不会直接给出修改后的版本而是输出一个结构化的审查意见表逐条列出问题类型、所属章节、修改建议。供稿人拿到这张表之后只针对列出的问题做定向修改不重新展开全文。class ReviewAgent(BaseAgent): async def process(self, message: Message) - Any: report_content message.payload.get(report) checklist message.payload.get(checklist) system_prompt ( 你是质量审查员。检查报告是否符合以下要求完整性、数据一致性、逻辑连贯性、格式规范。 不要修改原文。输出JSON格式的问题列表 每个问题包含 issue_type, section, severity, suggestion 四个字段。 如果所有检查项均通过输出空列表。 ) raw await self.call_model(system_prompt, json.dumps({report: report_content, checklist: checklist})) return self._parse_json(raw)审查环节的价值不只是提高质量它还给整个流程引入了退出条件。没有审查时多 Agent 协作的产出是好是坏全凭感觉有了审查系统就对什么时候算做完有了明确的判断依据。审查通过就结束不通过就进入返工循环直到通过或达到最大修订次数上限。这里要提一个实操细节必须设置最大修订次数。如果不加限制系统在极端情况下会陷入无限循环审查员始终不满意撰稿人始终在改Token 持续燃烧。我在调试时设的是最多三轮修订超过就直接报质量未达标并把最后一次结果和审查意见一并返回给上层至少保证有产物出来还能说清楚为什么没达标。4. 常见问题与排查技巧实录4.1 上下文污染导致的串味问题最常遇到的现象报告里出现了跟本次任务完全无关的内容比如上次任务的旧数据、旧结论甚至旧措辞。问题根源是上下文的物理隔离没做好。排查时先检查调度层传参看是不是把历史消息或者旧的结果数据混在了 payload 里传给了 Agent再检查 Agent 内部的 context_history 是否被多个任务共用了。我在调试时踩过的一个坑是贪图省事让同一个 Agent 实例处理多个任务它的context_history一直在累积。虽然每次只取最近几条消息但模型看到了之前的任务痕迹输出就会穿越。解决办法很简单每次任务开始前重置context_history或者确保每个任务使用独立的 Agent 实例。我自己最终选择了后者因为多个任务共享实例虽然省点内存但心智负担太重随时要提防消息串场。4.2 循环调用与死锁系统为什么会绕圈子多智能体系统一旦角色之间可以互相反馈就很容易出现循环。A 让 B 改B 改完让 C 审C 审完不通过又回到 A然后 A 又发给 B如此循环。这种循环在任务逻辑不严格时特别容易出现。排查循环问题的第一步就是看消息流水。如果发现同一对收发方的消息反复出现且内容高度相似基本可以断定是循环了。根因通常是某个 Agent 的职责边界不清晰比如审查 Agent 既提出修改意见又要求补充内容而补充内容的动作会触发另一个 Agent 的连锁反应反馈路径就闭合了。解决方式是在调度层加上环路检测记录每条消息的哈希值如果同一任务类型在短时间内重复出现超过阈值直接终止并提示人工介入。这是兜底方案更根本的解法还是在角色设计时把反馈路径限定为单向——在项目里审查 Agent 的意见只会发给撰稿人不会发给调度层调度层只负责在超限时截断流程。4.3 Token 消耗失控成本怎么控制多智能体系统的 Token 消耗通常比想象中高得多主要原因不是单个调用贵而是调用次数多了。一个任务走完收集-策划-撰写-审查-修订五个环节可能产生几十次模型调用每次调用还带着越来越长的上下文。控制 Token 有几个有效手段。第一严格执行上下文截断不要把所有历史消息都传给模型。第二payload 里只塞必要数据不要图省事把中间结果全打包传下去。第三对简单子任务使用更小的模型比如格式转换、字段抽取这种任务完全不需要顶配模型来处理。这套框架在设计时就把模型名作为 Agent 的可配置参数这样不同 Agent 可以用不同规格的模型成本能省下来一大块。我实际观察过一个案例同一个任务不做任何优化时消耗约 12 万 Token做了上下文截断和精简 payload 后降到 4 万出头效果是立竿见影的。所以如果觉得成本高先别急着怀疑模型贵大概率是通信和上下文的浪费。4.4 调试利器消息流水与任务状态表多智能体系统最可怕的不是出错而是出错后不知道哪一步错了。传统单 Agent 系统只要看日志和输入输出就能定位多智能体系统则必须依赖持久化的消息记录。我在这个项目里加了一个简单的消息日志模块所有 Message 在发送和接收时都会追加到一张数据表里。字段包括时间戳、任务 ID、发送者、接收者、任务类型、消息摘要、状态。排查问题时只需要按任务 ID 过滤就能完整还原一个任务的执行时间线。这是整个项目里性价比最高的一项功能强烈建议任何做多 Agent 项目的人第一时间加进来。任务状态表同样重要。调度器维护的任务表会记录每个任务从PENDING到DONE或FAILED的流转过程。如果发现任务一直停在RUNNING基本就是 Agent 调用挂住了需要检查模型调用是否超时如果停在NEEDS_REVISION没后续动作则是调度器的反馈路径没走通需要检查修订任务是否被正确重新入队。来看一个实际的排查场景。某次运行中报告任务在策划节点反复失败。查看任务状态表发现task_plan一直是FAILED再看消息日志发现策划 Agent 收到了资料收集结果但 payload 里的数据格式不是 JSON 数组而是一个字符串。进一步追查发现资料收集 Agent 在模型输出非标准 JSON 时用了简单的字符串兜底导致下游解析失败。修复方式是在解析函数里增加容错逻辑遇到非标准格式时尝试提取代码块中的 JSON 片段。整个排查过程不到十分钟全靠消息日志把问题范围一步步缩窄。4.5 常见问题速查表现象可能原因排查方向解决参考Agent 输出包含旧任务内容上下文未隔离检查 context_history 是否跨任务复用任务级上下文隔离执行前重置任务链卡住不动模型调用超时或无限等待查看任务状态是否长期处于 RUNNING为模型调用设置超时时间系统来回执行相似任务角色边界不清晰反馈路径成闭环查看消息流水中的重复消息简化反馈路径加环路检测下游 Agent 解析失败上游输出格式不规范查看 payload 实际内容增加上游输出格式校验和容错解析Token 消耗异常偏高上下文无节流payload 冗余统计各环节调用量和 Token 数上下文截断精简 payload 字段返工循环到不了终点缺少修订次数上限查看任务修订次数设置最大修订次数超限终止流程这张表是我在实际调试中整理出来的高频问题。多智能体系统的常见故障其实就集中在几个点上跑通一次流程后把容易出错的环节提前用规范和兜底逻辑卡住系统稳定性会显著提升。5. 扩展思考这套框架还能往哪些方向走agency-agents 目前的实现还比较轻量但它留下的几个设计接口让我看到不少扩展空间。第一个方向是为 Agent 增加工具调用能力。当前框架中的 Agent 只能调用模型如果接入真实的数据查询、文件读写、API 请求等工具整个系统就能从分析讨论走向动手执行。实现上只需要在 BaseAgent 里增加一个工具注册表Agent 可以通过结构化消息申请调用工具然后由工具层执行并返回结果。第二个方向是引入人工介入的审批节点。有些场景下AI 系统产出的内容需要经过人工确认才能进入下一步比如发送对外邮件、发布正式文档。可以在调度器上加一个人工审核状态流经该节点的任务会先挂起等人工确认后再继续。这种模式特别适合对准确性要求高的业务场景。第三个方向是评估体系的建设。现在系统的做得好不好只靠审查 Agent 的主观判断这种判断本身也是模型生成的不保证稳定。长远来看应该沉淀一套带客观指标的效果评估体系例如报告内容的命中率、格式规范率、返工率等指标用统计数据分析系统质量而不是听模型自我感觉良好。第四个方向是复用和编排更复杂的组织形态。机构里不只有老板和员工还有部门、项目组、专家团对应的系统设计也可以往这个方向靠。比如同样是资料收集可以并行多个收集 Agent 再合并结果同样是审查可以让两个不同视角的审查 Agent 交叉验证。这些高级形态都从角色可替换、调度可编程这两个基础能力上生长出来所以基础框架的设计是否留足了余地非常关键。我在实际探索中发现系统复杂度上升到一定阶段后最重要的反而不再是模型能力而是整个系统的可预期性和可观测性。一个固定输出格式的小模型 Agent有时候比一个聪明但不可控的大模型 Agent 更让你省心。多智能体的本质不是堆砌模型而是用架构把能力组织成系统想清楚这一点后续的扩展都会顺畅很多。
