多智能体协作框架：从单Agent到机构化工作流的设计与实践

做多智能体这件事，我前后折腾过不少方案，从最朴素的“一个Agent干所有活”，到后来的“一群Agent互相乱喊”，最后才走到“给Agent们建一个组织”这条路。这个项目代号叫【agency-agents】，说白了就是一套把多个智能体编排成“虚拟机构”的轻量级框架。它能做的是：让不同角色的Agent像公司里不同部门一样，各管一摊、按流程协作、有上级分配任务也有同级互相转交，最后汇总成一个可控、可审查、可回滚的工作流。如果你已经被单Agent的上下文限制、工具切换混乱、任务一复杂就失忆这些问题折磨过，那这篇文章大概能帮你去掉不少弯路。

1. 为什么多智能体需要“机构化”：从单兵作战到组织协作

先聊一个反直觉的结论：把好几个Agent丢在一起，不等于它们就会协作。多数人第一次尝试多智能体，用的都是“链式调用”甚至“互相对话”这种粗放玩法，结果翻车率极高。我仔细复盘过自己的几次失败试验，发现问题的根源不在模型本身，而在于——我们给Agent安排的场景太像“一群自由职业者临时拼单”，而不是“一家公司里职责分明的部门”。

1.1 单Agent的边界：不是能力不够，是“一人多岗”会失控

单个Agent接入再多的工具，本质上也还是一个知识面有限的“通用雇员”。你用RAG塞给它三份文档，它能答得不错；但你要是让它同时做“市场调研 → 写分析报告 → 生成营销文案 → 质检校对”这么一条流水线，它大概率会在第三步就开始自由发挥，把调研数据编得像真的一样。原因很朴素：它在单轮上下文里既要记住原始数据，又要保持创作风格，还要自我纠错，Token预算和注意力分配早晚要透支。我在自己的测试里，单Agent执行超过四步的复合任务，信息失真率明显上升，与其说是模型退化了，不如说是我们让一个人干了一个团队的活。

1.2 从“多Agent混聊”到“Agency模式”的转变

我一开始也试过让几个Agent直接互相发消息，A说“我查完了，你写吧”，B说“好的，马上”，C再插一句“我觉得这里不对”。这种模式看起来自由，实际上灾难——没有谁对最终结果负责，也没有人决定“下一步该谁上场”。折腾几次之后，我意识到真正缺的是一套组织关系：谁是指挥者，谁是执行者，谁能调取谁的产出，任务失败之后找谁兜底。这就是agency-agents项目最初的设计动机：把Agent分组、定职、定流程，用一套“机构”的壳去约束它们的行为。

对比一下两种模式的差异，表格可能更直观：

这个表格不是凭空拍的，而是我跑了一堆测试后总结出的分界。后面你就会发现，只要把“谁负责什么”写清楚，系统行为的可预测性会立刻上一个大台阶。

1.3 什么项目适合用这套思路

不是所有场景都需要机构化。我的判断标准很简单：“这个任务能不能拆成专业分工？”如果答案是能，比如涉及数据采集、分析、生成、审校、发布这类多角色协作的，那就值得上Agency模式；如果只是“帮我写个周报”，一个Agent配一两个工具就够了。机构化是有成本的，每个Agent都要定义角色、配置工具、写提示词，还要处理它们之间的消息流转，所以别为简单任务上重型方案。

2. 核心架构拆解：Agency、Agent与消息总线的三角关系

整个框架的骨架，其实只有三个核心概念：Agency、Agent、MessageBus。很多复杂系统的名字听着玄乎，落地之后你会发现，把这三样理清楚，剩下的都是补充。

2.1 Agency：既是容器，也是规则制定者

Agency是这个虚拟机构的总称。它负责三件最关键的事：一是注册成员Agent，维护一份“人员花名册”；二是定义机构内的路由规则，比如“凡是带research前缀的任务，统一发给研究员Agent”；三是维护机构的共享状态，相当于公司内部的“公共数据库”，所有Agent都能读，但只有被授权的角色能写。

初始版本里我用一个简单的类来承载这个思路，核心结构大概是这样：

class Agency: def __init__(self, name: str): self.name = name self.agents = {} self.router = Router() self.shared_state = SharedState() def register(self, agent: Agent): self.agents[agent.role] = agent agent.attach_state(self.shared_state) def dispatch(self, task: Task): route = self.router.route(task, self.agents) return route.agent.handle(task)

你说它是微内核也好，是容器也罢，总之整个系统对外只暴露一个dispatch入口。这样上层业务不用关心内部有几个Agent、谁先谁后，只要把Task扔进来，等结果就行。这一层带来的实际好处是，后续想加新Agent、换模型供应商、调整流程顺序，都不会炸到调用方。

2.2 Agent：职责、工具与边界的一次封装

每个Agent本质上就是一个“专业雇员”的封装。在它的内部，除了常规的系统提示词，至少要写明三样东西：职责边界（只做什么，不做什么）、可用工具（只有职责内的工具）、产出格式（直接喂给下游Agent能解析的结构化结果）。

我习惯把Agent定义成一个数据类，方便统一配置：

@dataclass class Agent: role: str # "researcher" system_prompt: str # 描述职责边界与行事风格 tools: List[Tool] max_iterations: int # 内部可以循环多少次，防死循环 output_schema: dict # 下游Agent消费用的JSON结构

请注意output_schema这一项，很多初学者会忽略。多Agent协作最容易翻车的地方之一，就是A说“我把结果放在北区了”，B根本找不到北区在哪。强约束产出结构之后，下游Agent可以直接用JSON字段取值，而不是靠模型瞎猜。

2.3 MessageBus：让Agent之间“对上话但不吵架”

Agent之间不能直接互相调用对方的函数，只能通过MessageBus发消息。这么做不是为了增加一层延迟，而是让所有通信都可追踪、可过滤、可打断。每条消息带上task_id、sender、receiver、payload、status，这样调试的时候你能像看聊天记录一样复盘整个协作过程。

MessageBus最基本的行为就是队列投递和按需订阅，实现上并不复杂：

class MessageBus: def __init__(self): self._queue = deque() self._handlers = {} def publish(self, msg: Message): self._queue.append(msg) def subscribe(self, role: str, handler: Callable): self._handlers.setdefault(role, []).append(handler) def tick(self): while self._queue: msg = self._queue.popleft() for handler in self._handlers.get(msg.receiver, []): handler(msg)

如果你嫌自研太糙，直接用Redis Stream或者RabbitMQ也能顶替，但核心思路不变：消息要过中间层，谁消费、什么时候消费、消费失败怎么办，都要有规矩。

也许你会说，这不是把简单事情搞复杂吗？我当时的想法也一样，但真正跑起来才发现，没有中间层的多Agent系统，就像没有项目经理的项目组，每个人都不知道别人干到哪一步了，最后产出的东西大概率要返工。MessageBus的那一点点性能损耗，换来的是全链路可观测性，这笔账划算。

3. 落地实现：角色定义、任务编排与调度器的具体设计

架构讲清楚了，接下来就是动手环节。这一节我重点讲角色定义怎么写、任务编排有哪几种模式、调度器怎么选，以及每个环节里容易踩的暗坑。

3.1 角色定义不是写“人设”，是写“岗位说明书”

很多人接触Agent时，会把大量精力花在给Agent写性格。但在多智能体协作里，最重要的不是性格，而是边界。比如“研究员Agent”，你不能只写“你是一个研究助理”，要写清楚：可以调用哪些搜索工具、只输出包含source和summary的JSON、不允许生成营销建议，那是文案Agent的事。

岗位说明书的组成部分我总结为五个要素：

1. 职责声明：一句话说明这个岗位负责什么。
2. 禁止事项：超过边界就拒绝执行，别硬扛。
3. 工具白名单：只挂这个岗位需要的工具。
4. 协作对象：通常和谁交接、从谁那里收结果。
5. 输出格式：用JsonSchema或示例约束，不给模型自由发挥的空间。

我自己在写角色定义时总结出一个经验：与其写“你是资深的研究员，请严谨细致”，不如写“如果你找不到权威来源，请明确说明‘未找到’，不要编造”。把运行规则直接写进提示词，比抽象的性格描述有效得多。说到底，我们不是在捏一个角色玩，而是在定义一套可靠运行的岗位逻辑。

3.2 三种常见的任务编排模式：串行、并行、主管代理

实际项目里，任务编排基本跳不出三种模式，我建议一开始先背下来，后面再按需混用。

第一种：串行流水线。典型的做法如上文提到的调研→分析→文案→审校。前一站的输出直接作为后一站的输入，不需要调停者。串行的优点是好理解、好追踪；缺点是整条链的耗时等于所有环节之和，而且任何一个环节出错都会中断。适合固定流程、稳定性优先的场景。

第二种：资深模式。这是我自己常说的“主管+专员”结构。一个主管Agent负责拆解需求，然后把小任务分给多个专员Agent并行执行，最后把大家的产出汇总成一份答卷。这个模式特别适合“让我写10个选题”之类的任务——一个主管拆出10个需求，10个专员Agent各写一个，最后总经理汇总。并行带来的收益在任务数量大时非常明显，代价是你需要额外写一个合并逻辑，以及处理部分Agent失败后的重试策略。

第三种：层级模式。再复杂一点，让主管Agent自己也有子级。其实就是上面的模式递归一层。不是所有项目都需要，但有些多步骤、跨部门的任务，确实要靠两级调度才能既保持分工又控制跨度。我一般建议能力不够成熟时先不要上这种，除非你已经有一套清晰的降级和重试机制。

3.3 调度器：别把所有任务都“轮询”给所有Agent

调度器就是那个决定“任务该进哪个门”的模块。初期版本里，我直接写死了路由表：task_type等于“search”就发给research_agent，等于“write”就发给copywriter_agent。这个方式简单可靠，适合流程固定的场景。

后来任务变多变杂，我开始引入了一个简单的“技能匹配评分”机制：每个Agent在注册时声明自己的skills，调度器根据任务的标签和技能列表做一个匹配度打分，取分高的Agent投递。这套思路类似一个简化版的服务发现，逻辑不复杂但很实用。

我还是要提醒一下，调度器不要做得太“聪明”。我一度希望能用另一个Agent来当“智能调度员”，让它在每次任务进来时现场决定分给谁。结果是，那个智能调度员自己就开始纠结，不仅拖慢了整体速度，偶发还会把任务发给完全无关的Agent。除非你的任务集合极度动态，否则用一个确定性规则加一个简单评分，已经远超够用。

这个阶段最容易忽略的是“拒绝”机制。每个Agent应该有可能说出“这不是我的活”的能力。如果调度器把一类任务分发错了，Agent直接把消息退回，并附上一句“请发给xxx”，整个系统就具备一定的自愈能力，而不是硬着头皮执行下去。

4. 跑通一个真实案例：市场洞察与内容生成流水线

理论说了一堆，我拿一个跑通的实际案例来讲，你会更清楚这套东西怎么用。这个案例是一个内容营销场景：给一个假想的SaaS产品做一份“行业洞察+竞品分析+推广文案”的每周自动报告。整个过程完全由agency-agents框架驱动。

4.1 任务拆解与Agent配置

我把任务拆成了四个角色，对应一个“市场部小团队”：

每个角色的Prompt我都按照前面说的“岗位说明书”写法配置，尤其是禁止事项写得很严格。研究员不评价竞品，只罗列事实；文案师不做数据核验，只负责表达；审校员一旦发现引用的数据来源不明，直接标记“延迟发布”。

4.2 任务编排：先并行后串行的混合流程

这个任务的编排是一个混合式流程。研究员和竞品分析师这两个岗位互不依赖，所以调度器让它们并行开工，等到两份报告都归档到共享状态后，文案师才被触发，读取两份分析文件，开始写文案。写完的初稿继续走下一站——审校员。

伪代码大概长这样：

agency = Agency("marketing_department") research = Agent(role="insights_researcher", tools=[WebSearch()]) analyst = Agent(role="competitor_analyst", tools=[SpiderTool(), DBReader()]) writer = Agent(role="content_writer", tools=[TemplateLib()]) reviewer = Agent(role="content_reviewer", tools=[SourceCheck()]) for a in [research, analyst, writer, reviewer]: agency.register(a) task = Task(type="weekly_marketing_report", payload={...}) agency.dispatch(task)

这里有个细节，dispatch这个动作本身并不会阻塞，它把任务交给调度器后，调度器按编排规则把两个初始任务并行发出。框架内部通过回调或者异步事件感知两个任务都完成了，才向writer发出第三阶段的触发信号。第一次跑通这个流程时，我最大的感受就是：我根本不用去管Agent们内部的对话，只要盯住它们发布到共享状态里的最终产物，整个系统就像一个自动化流水线一样往前滚。

4.3 实测观察：最值钱的是“审校员”这道闸

这个流水线跑了一周之后，真正的功臣是那个审校员Agent。统计下来，它拦截了三次严重的数据张冠李戴——研究员把A公司的融资金额记到B公司头上了，这种错误文案师根本看不出来，但审校员用它的SourceCheck工具一核对就发现了，直接返回“不通过，请核对数据源”。以前单Agent模式下，这种错几乎是必漏的。多Agent的每一步都可能引入噪声，所以“验收节点”不只是走流程，它真正兜住了系统的可信度下限。

如果你也想照这个模式搭类似流程，我建议不要一上来就搞5个Agent，先用3个跑通端到端，再逐步加角色。角色越少越容易定位是哪一步出了问题。

5. 多智能体系统最常见的五个翻车点及排查思路

这部分全是我自己踩过坑之后总结出来的，价值不低。多智能体系统看着好玩，真正跑起来，翻车率相当高。我把常见的五个问题列出来，连排查思路一起讲透。

5.1 上下文爆炸：所有Agent都在背同一本历史书

一开始我让所有Agent共享一份全局对话历史，结果每个Agent的上下文窗口都被成倍地浪费——研究员看文案师收到的东西，文案师看研究员的原始对话，Token消耗让我一度怀疑是不是被黑客盗号了。后来方案改成按需拉取：每个Agent只被授予“与自己岗位相关的那一段状态”，比如研究员只读原始数据和查询结果，文案师只读研究员最终产出的报告摘要，而不是整套聊天记录。

排查方法也简单：打开每次Agent调用的token用量日志，如果发现某Agent的上下文输入量一直在线性增长，那就说明它被塞了大量与其职责无关的信息。解决方案就是“最小状态原则”——下游Agent永远只消费上游Agent的结构化产物摘要，不消费原文上下文。

5.2 死循环与互相踢皮球：A请求B，B又请求A

最典型的现象是日志一直转，任务永远不会结束。我遇到过研究员Agent给文案师Agent发消息“你先把大纲给我”，文案师回“我得先有你的数据才能写大纲”，然后两边就一个劲地“礼貌对话”。问题根源在于没有“应答链条”的概念，也没有超时熔断。

我的解决方案有两层。第一层，给每条任务的流转设置了最大转手次数，比如8次，超过直接标记为“审核异常”，转入人工队列；第二层，在Agent的提示词里明确写出“收到任务后，能执行就立即执行，缺的信息向上游申请补充一次，如果上游无法在指定轮次内补全，就先基于已知信息产出并标注风险，严禁反复强制索取。”这一条写进Prompt之后，疑似死循环的比例降得很明显。

5.3 状态同步不及时：下游读到了“昨天的数据”

分布式系统里大家都吃过脏数据的亏，多Agent系统一样。我们的框架里状态是聚合同步的，但因为复制和更新的时机不一致，下游Agent可能在上游还没完成最终提交时就触发了读取。结果就是文案师拿到半截调研报告，写得还挺像样。

这个问题的排查方法就是在MessageBus上给每个任务加一个ready事件，只有上游产出发送completed信号后，调度器才放行下游触发。我后来把“发布-订阅”改成了“状态机驱动”，每个任务节点有pending、running、completed、rejected四个状态，下游订阅的是completed，不是“有新数据”这个宽泛的事件。状态机的好处是转移条件清晰，不会误触发。

5.4 失败重试的雪崩效应：一次超时，全员重跑

一开始我们给每个Agent调用模型设置了自动重试，Agent A失败了就重试，重试可能让A的下游B也重新跑了一遍，B一重跑，C也被迫重来。整条流水线在高峰期甚至可能反复从头开始，资源消耗直接翻好几倍。

改进的方案是“分级重试”：小任务小模型调用超时，只重试该步骤本身，最多三次；大流程级失败不自动重试，而是把异常状态发送到人工面板，由人来决定是重跑整个流水线还是修正某个环节。重试一定要有“幂等标识”，每个任务带唯一task_id，重试时相同task_id的结果直接复用，避免重复计算。

5.5 调试黑洞：Agent内部出了错，但外面看不出来

文本模型的一个特性是出错方式不固定，同样的输入可能产生完全不同的输出格式，导致下游解析失败。我曾因为一个字段从“source”变成“sources”，让下游工具整整两天没跑通，而系统日志里只有一句“JSON解析失败”。

这里我建议大家一定要尽早做两件事：一是所有Agent的产出在进入MessageBus之前做schema校验，这一步能把大多数格式错误挡在最外层；二是为每次任务生成一份trace记录，包含每个Agent的输入摘要、输出摘要、耗时、token数和重试次数。排查问题时，我基本都靠这份trace定位到是哪一步、哪个字段出了问题，而不是漫无目的地翻Prompt。

翻车问题讲完了，最后说一些更进阶的东西。你可能已经发现，多Agent系统的难点不在于“造出几个有技能的Agent”，而在于如何让它们在同一个上下文里高效协作。等这一套跑顺之后，可以考虑接入工具调用、记忆增强和人工介入这些更“职业化”的能力。

6. 进阶玩法：工具调用、记忆增强与人工介入机制

基础框架稳定之后，我开始做加法。这个阶段你要睁大眼睛挑选，哪些功能值得做，哪些是锦上添花但多数时候用不上。

6.1 工具注册表：把“会做什么”变成可枚举的能力

工具是Agent的“手”，但手多了也会打架。我在框架里设计了一个工具注册表，每个Agent申明自己需要的工具，进而由一个统一的ToolExecutor来执行。工具执行结果会自动带上一个“调用时间戳”和“原始返回结果”，这样不管哪个Agent用到这些数据，都知道它的新鲜度，不会拿三个月前的搜索当昨天的答案。

常见的工具有搜索、数据库查询、API调用、日历/任务管理等。工具调用的鉴权和隔离一定要做扎实，因为在多Agent里，工具权限失控等同于内部员工越权，不得不防。

6.2 记忆增强：短期任务记忆与长期组织记忆分离

Agent是否需要长期记忆？我的答案是分情况。短期记忆靠MessageBus的任务上下文就够；长期记忆建议用向量库做组织级记忆，比如“本周已经生成过哪20个选题”“竞品分析表的最近更新时间”。这种组织级记忆多半是任务间的关联知识，不应塞进每一个Agent的上下文，而应该在需要时让特定角色查询。

实现上，我让Agent自己决定“要不要去查记忆”，如果它发现自己遇到上一个任务遗留的引用，就调用记忆检索工具。这算是一种懒加载策略，比把所有历史一股脑塞给每个Agent安全得多。这个思路很像是人类员工的“项目档案+同事打听”结合体。

6.3 人工介入：不是打断，而是设置“审批卡点”

全自动流程听上去美好，但涉及到对外发布、付款、发送邮件这类动作，还是需要人拍板。我在框架里实现了“人工审核节点”，也就是流程走到某一步时，任务状态变为awaiting_human，等待人工在后台面板确认。这个节点并非阻断整个流水线，而是只阻断需要审批的那个分支，其他不相关的分支继续跑。

初期做的时候，我发现审批面板写得顺手很重要，否则一个节点的等待就可能让整个系统被吐槽“还不如我自己做”。所以要让待办卡片上能直接看到Agent的原始产出、引用来源和风险提示，让人不用回到日志里翻上下文。

6.4 效果评估与日志审计：让系统越跑越稳

最后一件值得做的事，是给框架加一个“评估环”。每一轮任务结束后，我会把最终产出和用户的验收结果回传，并记录到评估表中。常见的指标包括：任务完成率、平均耗时、失败环节Top榜、重试率。用这些数据去反推Prompt或路由规则哪里需要调整，而不是靠感觉“哪个Agent看起来聪明”。

尤其是失败环节的分布，能非常直接地告诉你系统瓶颈在哪。比如如果数据总是研究员Agent漏来源，那就去查它的工具权限或搜源范围，而不是去改文案师的措辞。数据驱动的优化逻辑，虽然听着像废话，但大多数多Agent项目其实都没有正经跑起来过。这个评估表，我建议从第一天开始就埋好。

写到这里，主体内容差不多讲完了。最后我想说句实在的——做多智能体项目，最有价值的事不是我写了多少行代码，而是学会了“用管理团队的思路设计AI”。这套agency-agents框架即使被换掉，背后那些关于角色边界、任务编排、状态机、失败重试、人工卡点的经验，依然能迁移到任何一套Agent系统上。你在自己项目中踩的每一个坑，都会成为下一版设计里最宝贵的设计约束。我现在回头看，最正确的是没有一开始就沉迷“智能涌现”，而是老老实实把流程和责任先建好。