项目标题: 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_idagent_id, rolerole, 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: funknown agent: {agent_id}} msg { message_id: fmsg_{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这个项目未来我还会继续扩展的方向包括接入更细粒度的技能市场、增加更复杂的回退策略、把记忆压缩做成完全自动化的后台服务。如果你也在做类似的多智能体协作项目欢迎一起聊聊你们在角色编排和上下文管理上是怎么处理的。
