项目标题: agency-agents
做智能体协作类项目有一段时间了，踩过不少坑，也重构过好几轮。今天把“agency-agents”这个项目从设计初衷到落地实现完整拆开聊一聊。这个项目的核心不是再做一个单一对话机器人，而是把多个具备不同专长的AI代理组合成一个能自主分工、互相串流结果的工作系统。简单说，就是让会规划的去规划、会写代码的去写代码、会挑毛病的去验收，最后再由一个调度核心汇总产出。这篇文章既适合想搞懂多智能体架构的人，也适合正在做类似项目、需要找一个可落地的工程参考的开发者。里面大部分方案都是我们在实际搭建过程中反复调过的，可以直接拿去做蓝图。
1. 整体思路：为什么要把“一个会干活的AI”拆成“一群各司其职的AI”
1.1 单智能体的天花板，是多智能体的起点
最早做这个项目之前，我们的原型其实就是一个“超级助手”，把工具调用、内网搜索、代码生成、文档改写全部塞进同一个系统提示词里。刚开始效果还行，但越跑越不对劲。任务一复杂，角色之间就开始互相打架：明明是在写代码，模型突然开始解释需求文档；明明只需要给一段SQL，它偏要顺带做安全审计。这不是某个模型能力不行，而是单一上下文把所有职责挤在一起，注意力根本分配不过来。
后来尝试过把这一个智能体拆成若干个“功能组件”，每次按流程串行调用。但很快问题又来了——组件无状态，每个步骤都重头开始，丢失了前半程的决策信息。比如设计评审环节明明否决了某个数据库方案，到了生成环节它又自己把原来的方案写回来了。
这正是我决定做“agency-agents”的直接原因：把“能力”和“权利”分开，让不同的智能体各自持有自己的上下文、记忆和输出约束，用一层轻量级的调度机制完成信息交接。有人把这叫“多智能体框架”，有人叫“AI编排”，本质上都是一回事——把一个庞大的任务空间切分成若干子空间，每个子空间由一个专业性更强的代理负责。
1.2 把系统拆成“机构”而不是“工具库”
“agency-agents”这个命名的核心隐喻，是把系统当成一家公司而不是一堆工具函数。工具函数是调用即返回，没有记忆、没有取舍、没有自我检查。而机构里的每个成员都有三个共同点：有明确的岗位职责、有上下级的汇报关系、有自己的工作记录。
这个设计映射到代码层面，就是每个agent都包含四条核心约束：
- 角色定义：我是谁，我擅长什么，我不做什么。
- 输入契约：我需要收到什么类型的信息，格式是什么。
- 输出契约：我产出什么结构的结果，如何被下游消费。
- 决策边界：什么情况我自己可以直接决断，什么情况必须上报调度器。
这套模型带来的最大好处是“局部改动的成本极低”。比如之前我们需要把代码生成环节从“给一段建议”改成“直接改文件并跑测试”，只需要替换执行者这一个agent的定义，其他角色的接口完全不用动。
1.3 解决一个最容易被忽略的问题：信息失真
多智能体系统里最常见的问题是信息在传递过程中衰减。A说了十个要点，B只收到了八个，其中两个还理解偏了。我们在第一版就吃了这个亏。一开始以为只要把A的输出文本拼到B的上下文里就行，结果发现B的决策质量大幅下降，经常答非所问。
后来我们换了思路：定义一套结构化的“任务描述语言”，每个智能体的输出先经过一个序列化层，转成带字段的结构（目标、约束、产物、风险、待确认项），再交给下游。这以后信息失真问题基本消失。所以如果你的多智能体系统正在出现“越传越歪”的现象，大概率不是模型的问题，而是你们的传输协议太随意了。
2. 角色模型与协作流程的设计细节
2.1 四个核心角色：规划者、执行者、评审者、守护者
我们的系统里最常用的是四类角色，互相之间保持明确的边界。
**规划者（Planner）**负责把用户的目标拆解成可执行的任务列表。它的输出不是一段建议，而是一份结构化任务书，包含任务编号、依赖关系、验收标准。规划者的退化模式是“什么事都大包大揽”，所以我们把它的工具权限控制得很窄，不允许它直接去操作文件或者发起外部请求。
**执行者（Executor）**是干活的人。它接收规划者的任务书，调用实际工具完成编码、写文档、生成SQL、抓取网页等具体动作。执行者是一个可以有很多个实例的角色，不同实例绑定不同的工具集。比如代码执行者绑定了Python运行时和Git接口，内容执行者绑定了文档模板和排版工具。
**评审者（Reviewer）**负责找茬。它的系统提示词里明确写了“你的工作是发现错误，不是修复错误，发现之后写清楚问题即可”。这是个很关键的设计决策——评审者一旦动起手来改东西，它的标准和产出就会混在一起，最后没人说得清问题到底是谁解决的。
**守护者（Guardian）**是全局的安全网，相当于“最终验收+熔断器”。它检查评审结果是否全绿、执行产物是否满足验收标准、是否存在合规风险。如果连续两次评审都不过，守护者会主动中止流程，把问题升级回给用户，而不是让系统死循环地“改-评-改”。
四个角色用一句话串联起来就是：规划者定方案，执行者出活，评审者挑刺，守护者把关。
2.2 任务编排状态机
整个系统的运行过程是一台状态机，而不是简单的消息串接。我们定义了五个核心状态：
| 状态 | 含义 | 进入条件 | 退出条件 |
|---|---|---|---|
| IDLE | 空闲等待 | 系统启动 | 收到新任务 |
| PLANNING | 规划中 | 任务队列非空 | 规划者产出任务书 |
| EXECUTING | 执行中 | 任务书通过校验 | 执行者提交产物 |
| REVIEWING | 评审中 | 产物进入评审队列 | 评审者给出结论 |
| DONE / FAILED | 结束或失败 | 评审通过或熔断 | 清理资源，回到IDLE |
实际运行中这个状态机有两个关键细节。第一，每个状态都有超时控制，比如EXECUTING默认超时是120秒，超时后自动进入评审且打上“疑似超时”的标记。第二，评审不通过不会从头再来，而是生成一条“回退指令”，只针对对应子任务重新进入执行状态，这样不会把整个流程推倒重来。
这个设计的价值在于可观测性很强。每次状态流转都会打点记录，我们后来做的可视化面板，基本就是把状态机的流转日志重新画了一遍图。
2.3 消息协议：智能体之间如何“说话”
智能体之间传递信息我们设计了一套轻量级协议，字段不多，但覆盖了协作需要的全部语义。
{ "message_id": "msg_001", "from": "planner", "to": "executor.code", "task_ref": "task_002", "type": "assignment", "payload": { "goal": "实现用户登录接口", "acceptance_criteria": ["返回200", "密码使用bcrypt加密", "错误提示不泄露用户是否存在"], "context": "前端已提交auth/schema.json" }, "requires_ack": true }每个消息都有明确的来源、目的地、所属任务和承载数据。最核心的字段是requires_ack，它保证了消息送达后，接收方必须回执确认。如果超过重试阈值还没收到回执，调度器会认为接收方异常，触发隔离和重启。
有人说这种设计太重了，有点“把简单的事情搞复杂”。但我的体会是，多智能体系统的复杂度是必然存在的，问题只在于你选择把复杂度放在“消息结构”里，还是放在“排查Bug时的猜测”里。我选择前者。
3. 核心代码与实操落地过程
3.1 智能体基类：所有的角色都从这里继承
写代码之前先定义抽象基类。这个基类我们不追求大而全，它只做三件事：接收事件、维护记忆、产出结果。其他全部交给子类自由发挥。
from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Any, Optional @dataclass class AgentContext: agent_id: str role: str memory: dict current_task: Optional[dict] = None class BaseAgent(ABC): """所有智能体的基类""" def __init__(self, agent_id: str, role: str): self.agent_id = agent_id self.role = role self.context = AgentContext(agent_id=agent_id, role=role, memory={}) def remember(self, key: str, value: Any) -> None: """把关键信息写入自己的记忆区""" self.context.memory[key] = value def recall(self, key: str) -> Any: """从自己的记忆区取出信息""" return self.context.memory.get(key) @abstractmethod def handle(self, message: dict) -> dict: """处理一条消息，返回结果消息""" raise NotImplementedError我特意把“记忆”设计成最朴素的字典，是因为绝大多数场景不需要搞一个重型向量数据库放在智能体内部。你需要记忆的不过是任务的来龙去脉和几个关键决策点，用字典就够了。真想上向量检索，那也是外面的记忆服务去做的事，不应该塞进agent内部。
3.2 编排器调度器：整个系统的“经理”
编排器是系统里唯一一个不是智能体的组件，它是调度中枢。它维护着任务队列、各智能体健康状态、消息重试机制和状态机转换。
import asyncio from collections import deque class Orchestrator: def __init__(self): self.agents = {} self.task_queue = deque() self.state = "IDLE" self.running = False def register_agent(self, agent: BaseAgent) -> None: """注册智能体到调度中心""" self.agents[agent.agent_id] = agent agent.context.memory["home"] = "orchestrator" async def dispatch(self, task: dict) -> dict: """入口：提交一个任务并等待它最终结束""" self.state = "PLANNING" self.task_queue.append(task) while self.task_queue and self.state != "FAILED": current = self.task_queue.popleft() agent_id = current.get("to", "planner") if agent_id not in self.agents: return {"status": "error", "reason": f"unknown agent: {agent_id}"} msg = { "message_id": f"msg_{asyncio.get_event_loop().time()}", "from": "orchestrator", "to": agent_id, "payload": current, "requires_ack": True, } result = await self.agents[agent_id].handle(msg) # 根据智能体返回类型决定下一步状态 if result.get("status") == "blocked": self.state = "BLOCKED" return result if result.get("status") == "failed": self.state = "FAILED" return result self.state = "DONE" return {"status": "done"}这里有一个比较重要的设计：dispatch方法返回的是“所有任务都完成”之后的最终结果，但这个结果不一定是用户看到的样子。实际项目中，我们会把每一条子任务的中间产物都写入一个共享的任务仓库，前端或下游系统按需拉取。这样避免了“结果太大塞不进队列”的问题。
3.3 提示词模板管理：角色的“人设”该如何存放
很多同类的项目会把系统提示词直接写在代码里，我个人非常不建议这么干。提示词是变化最频繁的配置，你今天改一个角色的语气，明天调一段约束，然后就得重新走一次构建流程，不划算。我们的做法是把提示词模板独立成.md文件，按角色存放在prompts/目录下。
# 角色：代码执行者 你是专注的代码实现工程师。你的唯一目标是完成规划者分派给你的编码任务。 你不负责设计方案，不负责验收测试，更不负责优化全局架构。 ## 工作原则 1. 只改动与任务相关的文件，禁止顺手优化无关代码。 2. 必须处理异常路径，不能只写happy path。 3. 提交代码时附带一段简短的自测结果说明。 ## 不允许做的事 - 不要修改其他模块的代码 - 不要把简单功能拆成一堆抽象类 - 不要忽略验收标准中的任何一条这些模板文件通过一个加载器读入，再与动态参数拼接。改模板不需要改动任何代码逻辑，这在实际维护中省了非常多的事。尤其是当你要给同一个角色调“性能模式”和“稳定模式”的时候，只需要在模板里加一个可替换的变量控制“发挥程度”，不用在程序上做任何分支。
3.4 配置驱动的应用场景：YAML里定义整个协作流
我们支持用户通过配置文件来描述自己的智能体协作流，这是项目从“demo”走向“可产品化”的重要一步。配置结构长这样：
project: 模拟项目X workflow: - step: 1 name: 拆分需求 agent: planner next: [exec_code, exec_doc] - step: 2 name: 并行执行 agents: - exec_code - exec_doc next: reviewer - step: 3 name: 合并评审 agent: reviewer on_fail: goto 2 next: guardian - step: 4 name: 终审 agent: guardian on_fail: fail next: done agents: planner: llm: model_x temperature: 0.2 template: prompts/planner.md exec_code: llm: model_y tools: [python3, git] allow_parallel: true这个配置文件干了两件事：一是定义流程拓扑，二是定义角色参数。也就是说，你想加一个“安全审计”的角色，只需要在agents里加一段配置，再在workflow里插入一步。整个系统的扩展从改代码变成了改配置。我们后来做的一些客户定制项目，基本就是靠这批配置文件快速搭出不同行业的多智能体流程。
4. 上下文管理、记忆与并发调优
4.1 三段式上下文：全局、任务、局部
多智能体系统里最容易失控的就是上下文。一开始我们让每个智能体都“拥有”完整任务上下文，结果模型直接被冗余信息淹没。后来我们把上下文明确切成三层。
全局上下文是项目基本信息，比如项目代号、技术栈、部署环境，所有智能体共享一份，但只读。任务上下文是当前任务的任务书、验收标准、相关文件路径，和执行该任务的智能体绑定。局部上下文则是智能体在单次执行过程中的临时变量，比如刚刚读取的文件内容、临时生成的代码片段。
这三层各有各的存储策略。全局上下文放Redis，任务上下文放内存对象，局部上下文随智能体实例消亡而释放。我的体会是：99%的“智能体闲聊说错话”问题，都是因为把不该放全局的信息扔进了全局。
4.2 超长任务的记忆压缩方案
模型都有上下文窗口上限，不管窗口多大，做一些长流程任务时都会触顶。我们的解法是“记忆压缩”。当任务记录超过窗口阈值的70%时，触发一次压缩：把已经完成的关键动作、已确认的决策、不再改变的事实，浓缩成几条摘要放进一个summary字段里，替换掉原始长文本。这个压缩在前端是不可见的，但token消耗几乎减半。
压缩策略听上去简单，坑在于“什么时候该压缩”。我们最开始每到一个固定阈值就压缩，结果发现有些细节在后续阶段还会反复用到，被压掉之后只能回来翻日志。后来改成“到达阈值 + 当前状态处于完成事件较多的时间点”双条件才触发压缩。而且压缩摘要由另一个智能体来生成，而不是简单切句子，效果好很多。
4.3 并发执行与资源隔离
并行执行是提升系统吞吐的关键，也是引入Bug的温床。我们系统里支持多个执行者同时处理不同子任务，但并发要解决两个问题：共享资源写冲突、下游合并顺序。
共享资源冲突的解决思路是给每个执行者分配独立的临时目录和独立的数据库schema，互不干扰。执行完成后，把产物以“工件”的形式登记到统一仓库，再由下游智能体拉取。合并顺序问题则通过给每个工件打上依赖标记，只要有依赖关系的工件必须等它依赖的工件先登记完，才能进入评审环节。
还有一个看起来很小但实际很影响体验的点：并发执行时，模型的调用超时要独立设置。否则一个智能体卡住了，整个调度进程都会卡在那等。我们把模型的超时从“全局5秒”改成了“每角色独立可配置”，代码执行型角色的超时放宽到60秒，规划型角色保持5秒。
4.4 模型选型与温度参数的取舍
不同角色的参数配置直接决定了协作效果。我们总结了一套经验值，列出来供参考：
| 角色 | 建议模型档位 | 温度 | 备注 |
|---|---|---|---|
| 规划者 | 逻辑推理强的模型 | 0.2 | 温度低了不容易发散，拆解更稳定 |
| 执行者 | 综合能力均衡的模型 | 0.3~0.4 | 需要一点创造性但不要跑偏 |
| 评审者 | 细节觉察能力强的模型 | 0.1 | 严格模式，低温度增强找茬能力 |
| 守护者 | 安全与合规能力强的模型 | 0.0 | 基本让它按规则逻辑判断，不放飞 |
| 摘要压缩 | 速度快token省的模型 | 0.2 | 压缩摘要不需要太强的创造力 |
注意这只是基线，换成不同业务场景时参数还得微调。比如做创意文案类的执行者，温度可以拉到0.8以上；但做财务数据处理时，执行者的温度压到0.2都不为过。核心是理解每个参数在协作链路上的意义，而不是照抄一个固定值。
5. 常见问题与排查技巧实录
5.1 问题速查表
整理了这段时间实际踩过的高频问题，按“现象-原因-解法”列出来，可以存下来当参考。
| 现象 | 根本原因 | 解决思路 |
|---|---|---|
| 规划者输出的任务书格式不稳定，偶尔缺字段 | 输出契约只靠提示词约束，没有硬校验 | 增加JSON Schema校验，不通过的强制重生成一轮 |
| 执行者改了无关文件 | 全局上下文里有太多无关文件路径 | 收紧上下文注入，只给任务相关的文件清单 |
| 评审者永远打回，流程死循环 | 评审标准过于主观，且没有全局兜底 | 配置最大重试次数，到了阈值交给守护者熔断 |
| 多个执行者写同个文件互相覆盖 | 依赖关系定义缺失 | 在任务书里增加depends_on字段，严格依赖检查 |
| 任务结果太大，消息队列塞不下 | 把完整产物塞进了消息体 | 产物写工件仓库，消息里只传递工件ID |
| 智能体突然丢失历史决策 | 局部上下文被意外清理 | 增加快照机制，定期把关键决策写入任务记忆 |
5.2 排查技巧：从“玄学”到“可定位”
多智能体系统里出了问题，最怕的是“不可复现”。你这次跑挂了，下次再跑可能又好了。为了把玄学变成可定位问题，我们做了两个强制要求。
第一个要求：每次消息传递都带trace_id。这个ID从用户请求产生开始生成，贯穿所有智能体和消息队列，就像物流单号。只要接口出问题，靠trace_id就能把整条链路拉出来看。
第二个要求：每次智能体输出都先落盘，再往下一步走。所以我们有一个存储目录专门保存原始输出，按trace_id分目录存放。这样即使模型输出是什么妖魔鬼怪，我们也能在事后翻到当时的原文，不用靠猜。
这套机制建立之前，排查问题平均要花一整天，建立之后基本一小时内就能定位到具体环节。强烈建议每个人都做这两件事，成本极低但收益极大。
5.3 模拟场景：一次完整的多智能体协作过程
用实际例子看一下这个系统到底是怎么运作的。假设用户提交了一个任务：“为内部文档系统开发一个自动标签推荐功能，要求支持中文长文本的标签提取”。
第一步，规划者接到任务后生成如下结构：
{ "tasks": [ {"id": "t1", "desc": "调研现有文档结构，确认扩展点", "depends_on": []}, {"id": "t2", "desc": "开发标签提取基础服务", "depends_on": ["t1"]}, {"id": "t3", "desc": "开发接口层并接入前端调用", "depends_on": ["t2"]}, {"id": "t4", "desc": "编写单元测试与集成测试", "depends_on": ["t2", "t3"]} ], "risks": ["中文分词依赖较重，建议先做技术验证", "标签数量上限需要产品侧确认"] }第二步，执行者并行处理t1、t2。每个执行者都有独立的临时目录，产出的代码和文档分别登记为工件。第三步，评审者检查代码质量，发现标签提取服务对长文本的截断处理有隐患，打回并写了明确的问题描述。第四步，执行者针对评审意见修改，提交再次评审。第五步，守护者合并所有评审结论，确认全绿后返回用户。
这套流程如果串行跑大概需要三到五分钟，并行跑可以压到三分钟以内。核心的时间消耗不在模型生成上，而在多次评审和校验上。但是换来的是最终产物的可靠性大幅提升。
6. 一些个人心得：什么样的项目适合用多智能体
先说结论：不是所有任务都需要多智能体。单轮问答、简单文本改写、单个工具调用，老老实实一个智能体做完才是最优解。上多智能体协作架构，只有在任务必须分段完成、每段需要不同专业能力、且每段的产出会作为下一段输入的时候，收益才真正明显。
具体来说，有三类项目适合这个架构：一是跨领域的复杂任务，比如“分析数据→生成报告→绘制图表”一条龙；二是需要质量把关的生成任务，比如代码生成后必须有人评审；三是需要并行处理大量子任务的场景，比如一个季度几十个文档的批量处理。
如果项目符合这个特征，多智能体协作带来的收益是非常直接的：每个环节可以独立优化、独立扩展、独立测试，整个系统的可维护性比单体智能体好得多。而且随着角色越来越多，系统整体的任务承载力不是线性增长，而是接近乘法增长。
最后说一个我在实际项目里比较深的体会：多智能体系统最值钱的部分不是“让AI自己干活”，而是“让AI知道什么时候该把问题抛回给人类”。规划者拿不准需求的时候，评审者连续打回两次的时候，守护者检测到风险异常的时候，都应该主动生成一个人工介入请求，而不是硬着头皮自己决策。这种“留一个退出通道”的思维，才是多智能体协作真正成熟的表现。
agency-agents这个项目未来我还会继续扩展的方向包括：接入更细粒度的技能市场、增加更复杂的回退策略、把记忆压缩做成完全自动化的后台服务。如果你也在做类似的多智能体协作项目，欢迎一起聊聊你们在角色编排和上下文管理上是怎么处理的。
