agency-agents是我近期在折腾的一个多智能体协作系统项目。简单来说就是不再追求一个超级Agent搞定一切而是把大模型拆成一群各司其职的小Agent再按一个数字机构的组织方式把它们组装起来有人拆解需求、有人干活、有人审质量、有人管知识库最后像流水线一样把项目交付出去。这篇文章就是把整个项目的设计思路、踩坑过程和落地经验完整记录下来适合正在做Agent应用、多智能体编排或者想用大模型搞自动化工作流的开发者参考。1. 项目背景为什么单个Agent不够用这个项目不是一开始就长这样的。最开始我就跟大多数人的直觉一样想做一个全栈 Agent——把行业分析、代码生成、数据爬取、报告写作全部塞进同一个系统提示词里让它一个人当一支队伍用。结果实测下来效果非常不稳定也让我彻底明白了为什么纯单Agent方案在真实业务流程里撑不住。1.1 单Agent的瓶颈在哪里单个Agent最直观的问题就是上下文窗口不够用。现在主流的大模型虽然上下文越做越长但实际用起来你会发现长了之后推理质量明显下降而且是哪里都不够精。我试过让一个Agent同时处理市场调研任务它既要理解行业背景、又要写Python代码抓竞品数据、还要把数据整理成分析文案这些步骤挤在同一段上下文里互相干扰特别严重。最后代码是对的但分析结论很肤浅数据引用的部分还出现了幻觉。第二个问题是缺少检查机制。一个Agent自己干活自己验收等于既当运动员又当裁判。它如果在一个逻辑错误上扎了根后面的所有输出都会顺着错误往下走而且它会非常自信地继续编下去没有任何环节能把它拦住。第三个问题是并发能力差。真实业务不可能一个个任务排队做。我最初让单Agent处理一个月的任务列表它做第三项的时候已经忘了第一项的交付标准。如果不做任务级别的隔离和上下文管理Agent越往后执行质量越不可控。1.2 机构化思路的由来我后来换了个思路既然真实公司不是让员工一个人包全流程而是有产品经理、工程师、测试、项目经理这些分工那我为什么不能让多个Agent也这么干活于是这个项目被改名为agency-agents核心思想就是把一个Agent团队当成一个虚拟机构来设计。项目经理Agent负责拆解任务执行Agent负责具体产出质检Agent负责验收知识库Agent负责提供背景资料。每个Agent只有单一职责不让它跨界。这套思路带来的直接好处有三个每个Agent的上下文变得很干净只用操心它这个环节的数据任何一个环节出问题都能独立重跑不用把整个流程打翻重来可以像管员工一样管Agent给它定KPI交付标准、做评审质检环节、做复盘任务日志。当然这不是什么原创新概念多智能体系统早就有但我从机构这个视角去重新规划角色确实比单纯堆Agent要更有条理得多。2. 整体架构设计与角色拆解2.1 核心角色定义在一个标准的agency-agents系统里我目前稳定在用的有四个角色。第一个是项目经理AgentPM Agent。它不负责具体产出而是接收用户需求把需求拆解成原子任务列表每个任务都带清楚的目标和验收标准然后把任务分派给对应的执行Agent。项目经理Agent的系统提示词里我会明确写几条规则不允许自己动手执行任何任务拆分任务时必须给出可验证的输出物如果需求信息不足必须先向用户提问不能瞎猜。第二个是执行Agent。这其实是后面可以水平扩展的一组Agent比如代码Agent、文案Agent、数据Agent。它们的定位是只接受已经被PM Agent拆好的任务不自己理解用户原始需求。每个执行Agent只暴露它能处理的函数比如生成代码文件产出数据分析表其他一概不管。第三个是质检AgentQA Agent。它专门负责挑毛病不被允许直接修改执行结果只负责打回、给修改意见。质检Agent我会给它一份红线清单比如代码不存在语法级错误报告中每个数字都有数据来源结论和论据必须一致。一旦命中红线就返工。第四个是知识库Agent。这个角色充当的是团队的老员工通过检索为PM和执行Agent提供背景资料降低幻觉风险。2.2 协作机制任务分解、消息路由、结果聚合角色确定之后协作机制就是关键。常见的有三种顺序流水线、星型调度和层级分解。顺序流水线适合任务链路固定、前后依赖很强的场景比如生成文案-配图-排版-发布。优点是简单缺点是只要一个环节卡住整条链都停摆。星型调度是一个中心Agent把所有子任务派给外围的独立Agent然后收集所有人的结果。这种方式并行度高但中心节点容易成为瓶颈而且这种模式更适合每个Agent独立做不同事项的场景不适合前后强依赖的流程。我最终选择的是层级分解也是现在项目里在跑的方案PM Agent把大任务拆成几层子任务每个子任务还可以继续下放给更细的执行Agent。这样做的好处是任务之间的依赖关系完全由PM Agent维护调度器Scheduler)只负责按依赖顺序执行不用理解业务本身。一旦某个子任务失败只需要把这个分支重新跑一遍不会影响其他分支。在任务分派这一步有一点特别重要任务的描述格式必须完全结构化。我用的字段是task_id、parent_task_id、agent_role、input、output_format、acceptance_criteria。如果没有明确的验收标准执行Agent就很容易给出看起来差不多但没法用的结果而质检Agent也会因为没有标准而盲目放行。2.3 为什么选择角色化而非自由对话业内不少多Agent演示看起来很酷几个Agent围在一起自由聊天互相争论最后得出一个结论。但放到实战环境里这种自由对话模式有致命的缺陷不可控、不可追踪、成本高。自由对话意味着你不知道哪一轮对话会产生决定性结论也不知道该在什么节点人为介入。每个Agent之间的交互都是隐式的出问题了很难定位是哪一段逻辑错了。而角色化设计相当于给每个Agent定义了边界和交付物项目经理Agent的交付物是任务拆解表执行Agent的交付物是具体文件或数据质检Agent的交付物是通过或打回意见。每个Agent的输出不再是一段聊天文本而是结构化的工件。说得直白点聊天是过程工件是结果。做业务流程自动化我们关心的是工件不关心过程。角色化强制所有Agent把结果落成链路里的一个节点这样整个系统从外部看就不再是一个会聊天的黑箱而是一条可以审计、可以局部重试的流水线。这是我认为这个项目最核心的取舍。3. 实操过程从零搭建一套agency-agents系统3.1 环境准备与框架选型在技术选型上我并没有纠结太久。当时能选的路有三条直接用某个大模型平台的Agent能力、用开源Agent编排框架、完全自研调度层。大模型平台的Agent能力上手最快但有两个问题一是角色和流程逻辑被平台锁死二是多Agent协作的粒度不够细。开源编排框架也试用过几款它们的生态确实丰富但也给了我一种过度封装的感觉想让Agent完全按照我的流程走往往得迁就框架的抽象。最后我选了折中方案底层大模型走统一的API网关中间层用当前项目里最顺手的Web框架搭一个调度控制台状态存储用关系数据库记录任务流转消息队列负责角色之间的异步通信。不是自夸这套组合让我跑业务逻辑的时候几乎不用在意框架限制每个环节都能看得见、改得动。环境准备方面需要准备的事项如下Python 3.10建议虚拟环境隔离项目依赖关系数据库用于存任务状态和执行记录消息队列用于角色间传递任务通知大模型API网关的统一Key方便切换不同模型预留向量库存历史经验非必须项目中期再加也来得及。3.2 核心代码实现先把最底层的Agent基类和任务数据结构定义清楚。我用一段简化版代码说明核心逻辑。from dataclasses import dataclass, field from enum import Enum from typing import Any, Optional import uuid class TaskStatus(Enum): PENDING pending RUNNING running DONE done FAILED failed BLOCKED blocked dataclass class Task: task_id: str field(default_factorylambda: uuid.uuid4().hex) parent_task_id: Optional[str] None role: str executor # 由哪个角色处理 input_data: dict field(default_factorydict) output_format: str text # text / code / json / file acceptance_criteria: list field(default_factorylist) status: TaskStatus TaskStatus.PENDING result: Optional[Any] None error_message: str depth: int 0 class BaseAgent: def __init__(self, name: str, role: str, model_cfg: dict): self.name name self.role role self.model_cfg model_cfg self.tools {} def register_tool(self, name: str, func: callable): 统一挂载工具的入口方便调度层做权限控制。 self.tools[name] func def run(self, task: Task) - Task: raise NotImplementedError def _call_llm(self, system_prompt: str, user_prompt: str) - str: 实际实现时对接大模型API网关。 pass然后实现一个简化版调度器。调度器的职责很简单从任务队列里取一个任务根据role字段找到对应的Agent调用Agent的run方法更新任务状态。class Scheduler: def __init__(self, agents: dict): self.agents agents self.task_queue [] def submit_task(self, task: Task): self.task_queue.append(task) def run_loop(self, max_steps: int 50): for _ in range(max_steps): if not self.task_queue: break task self.task_queue.pop(0) agent self.agents.get(task.role) if agent is None: task.status TaskStatus.FAILED task.error_message funknown role: {task.role} continue task.status TaskStatus.RUNNING try: task agent.run(task) except Exception as e: task.status TaskStatus.FAILED task.error_message str(e) # 如果任务产生子任务就回填队列 if task.result and isinstance(task.result, list): for st in task.result: self.task_queue.append(st)这个调度器写得非常基础但足够说明问题。真实项目里我会加上并发控制同一时刻最多只允许N个任务并行防止所有Agent同时打爆API配额。3.3 指令、工具与记忆配置多Agent系统里每个Agent的系统提示词要有明确的职责声明和输出约束。举个例子项目经理Agent的提示词开头我会写成你是数字机构中的项目经理。你负责接收用户目标输出结构化任务拆解表。你不负责执行具体任务。每个任务必须包含以下字段role、input_data、output_format、acceptance_criteria。如果你认为需求不明确请直接输出question不要自行编造任务。你的输出必须是一个JSON数组。质检Agent的提示词则是另一个极端你是质量检查员。你只能对输入的工作成果进行验证。你必须根据acceptance_criteria逐条判定给出pass或fail。如果有任意一条不满足输出fail并列出具体修改意见。你无权修改成果本身。这里实际情况是模型对指令的遵循能力有限所以关键信息我会尽量通过结构化字段传入而不是全部塞在自然语言里。比如验收标准字段就会作为一个独立参数传给质检Agent来自任务本身的acceptance_criteria属性。工具的记忆配置方面每个Agent都有短期和长期两层记忆。短期记忆就是在处理当前任务时只携带这个任务链条内传递下来的数据不允许把上一个任务的数据带进来否则就是上下文污染。长期记忆则是把每次任务的经验教训写入向量库下次遇到类似任务时PM Agent会先检索这些经验作为任务拆解的参考。工具注册我用的装饰器方式和Flask里的路由注册很类似def tool(func): 把普通函数注册为Agent可调用工具。 func._is_tool True return func tool def fetch_web_content(url: str) - str: # 爬取网页正文 return 在调用大模型时我会把工具声明转成JSON Schema格式并强制模型限制工具调用遵循这个模式。这一步直接决定了后面工具调用成功率不能偷懒。3.4 运行效果与注意事项我拿一个具体任务测试让这套数字机构生成一篇某消费品牌年度营销方案。完整的流转过程大致是这样的PM Agent先拆出子任务行业背景调研、竞品分析、用户画像、传播策略建议、预算分配竞品分析需要数据Agent去查公开资料传播策略建议需要文案Agent产出方向草案。每个子任务都被标记了依赖关系策略草案必须等到竞品数据产出后才能开始。实际跑一轮下来最大的感受是并行度确实上来了数据Agent在等爬虫结果的同时用户画像Agent已经开始跑调研问卷的结构设计然后质检Agent把所有产物统一过了一遍找到了竞品页里两个过时数据打回让数据Agent重新更新。整套流程没有人工干预但产出质量比我之前单Agent方案高出一截。跑通之后有几个注意事项非常关键不要一次性放开所有Agent的并发权限。我第一版所有角色都是并发执行的结果质检Agent在数据Agent还没跑完时就去验收白跑了一堆无效调用。后来改成下游任务依赖上游输出的规则才把误跑率降下来。任务深度必须限制。如果PM Agent拆得不够彻底就会出现子任务下面还有子任务层级无限加深。我给嵌套深度设了一个上限超过就直接结束并提示PM Agent把拆解粒度放大。所有Agent的调用日志必须完整记录。这个项目的调错能力完全建立在日志上没有日志等于没有排查入口。我后来给每个Agent都加了自动记录任务开始、结束、调用模型、工具调用参数的结构化日志。4. 常见问题与排查技巧实录多Agent系统跑起来之后真正让人头大的永远不是模型能力而是工程层面的稳定性问题。我在这个项目里踩过好几个坑挑典型的几个说。4.1 任务分配死循环现象是任务在Agent之间来回传递始终没有一个Agent产出最终结果。最典型的一次PM Agent把一个总结报告任务派给执行Agent执行Agent觉得需要更多输入数据于是把任务派回给PM Agent当作数据补充任务PM Agent又认为这属于执行工作再次派给执行Agent两个角色之间反复互踢直到触发步数上限。排查思路是先看任务链日志确认每个任务的parent_task_id是否指向自己或形成环形。解决手段有三个给调度器加任务深度上限到点就掐断给每个Agent明确不允许转派任务的规则只能标记为blocked并等待外部输入在路由层做白名单明确每个角色只能接收特定role前缀的任务跨角色转交必须经过调度器审核。4.2 上下文污染执行Agent在处理多个任务时把上一个任务的历史数据带到了新任务里。比如跑完A公司的竞品分析紧接着跑B公司时模型会把A公司的数据当成B公司的写进报告里而且毫无违和感。这个问题排查起来不容易因为模型不会主动告诉你它记串了。我是靠抽样检查生成的文档才发现的。根治措施是每次任务执行前重新创建一个独立的上下文会话把该任务相关的数据作为新的系统输入注入不同任务之间不共享任何会话历史。短期记忆必须按任务隔离不能按Agent隔离。4.3 工具调用失败与重试机制工具调用失败最常见的两种原因参数没填全以及填了不存在的参数名。模型在调用工具时偶尔会自作主张生成一个看似合理但实际不存在的参数或者把必填参数漏掉。第一次遇到时我以为是大模型的Bug后来发现是我工具Schema写得太粗糙。解决办法是把所有工具参数声明为JSON Schema严格模式必填字段全部列出并对每个参数补充说明让模型理解含义。重试机制上我给工具调用加了三档退避失败一次立刻重试、再失败等1秒重试、第三次失败后挂起任务并通知PM Agent调整执行方案。实际跑下来工具失败率从最初的25%降到了3%以内。4.4 成本和时延失控多Agent系统最大的隐性成本不是模型推理本身而是无效对话。一个任务反复被打回重试每个Agent又重新读一遍所有上下文Token消耗成倍上涨但产出并没有变多。我做过一次统计一次完整营销方案任务主流程模型调用次数是7次但因为质检打回和上下文传参方式不当实际调用次数到了18次。查出问题后发现有两个Agent每次对话都把整个项目背景重新塞进去明明只需要传增量数据。优化方案是给Agent对话增加会话轮次上限默认3轮超过就必须由调度器介入同时建立缓存如果同一个输入之前已经产出了结果直接复用跳过调用。5. 进阶如何扩展为真正可用的数字机构基础版本跑通后agency-agents还只能算是一个技术demo。要想真正承担生产环境的业务还需要补上三个能力人工审批、工作流状态机、可视化追踪。5.1 引入人工审批与工单机制纯自动化的Agent流程在执行高风险任务对外发布、涉及预算、引用未知数据时还是需要人的确认。我的做法是在任务数据结构里增加一个requires_approval字段。如果PM Agent拆任务时判断这个子任务需要人工确认调度器就会把任务状态置为等待审批而不是直接丢给执行Agent。审批流程我参考了工单系统的设计每个需要确认的任务就是一个工单人工在控制台里看到工单可以批准、打回或修改输入。这使得Agent系统不是黑箱自动运行而是人机协作的流程工具尤其适合在团队里推广。实际用下来人的压力也不大大多数常规任务一遍通过只有少数关键节点需要参与。5.2 引入工作流状态机把Agent变成流程里的执行节点一开始我就是用代码逻辑控制任务流转状态一多就开始乱。后来我引入了状态机模型把整个业务流程显式定义成一张图每个节点是Agent调用人工审批消息通知或逻辑判断节点之间的连线是转移条件。这套改造最大的收益是让业务流程和Agent解耦了。业务流程是稳定的Agent是可以替换的。以后要换一个大模型或者换一个新Agent只需要在配置中心里把节点对应的Agent映射改掉流程本身不用动。对于要长期维护的项目来说这个分层的价值远大于省掉一点配置功夫。5.3 结果可追踪与复盘多Agent系统出问题时最大的难点是定位是哪一步的哪一次调用导致最终产出出了问题。所以我后来把所有的任务流转事件、Agent调用记录、工具返回结果、质检判定结果都持久化存储并给每个最终交付物生成了一个完整的生产过程报告。复盘时可以直接看报告某个子任务被质检打回两次原因都是数据来源缺失数据Agent平均每次任务要调四次工具PM Agent的拆解结果中有一半任务被质检判定为验收标准不清晰。这些数据反过来又能指导调整提示词或流程参数形成闭环。这也是我做这个项目最深的体会Agent本身的能力不是瓶颈流程设计和工程管控才是。把Agent当成一个需要管理、需要验收、需要复盘的“临时员工”而不是一个神奇黑箱这套系统才能稳定地跑出业务价值。 从单Agent到多智能体协作：搭建可落地的数字机构系统
agency-agents是我近期在折腾的一个多智能体协作系统项目。简单来说就是不再追求一个超级Agent搞定一切而是把大模型拆成一群各司其职的小Agent再按一个数字机构的组织方式把它们组装起来有人拆解需求、有人干活、有人审质量、有人管知识库最后像流水线一样把项目交付出去。这篇文章就是把整个项目的设计思路、踩坑过程和落地经验完整记录下来适合正在做Agent应用、多智能体编排或者想用大模型搞自动化工作流的开发者参考。1. 项目背景为什么单个Agent不够用这个项目不是一开始就长这样的。最开始我就跟大多数人的直觉一样想做一个全栈 Agent——把行业分析、代码生成、数据爬取、报告写作全部塞进同一个系统提示词里让它一个人当一支队伍用。结果实测下来效果非常不稳定也让我彻底明白了为什么纯单Agent方案在真实业务流程里撑不住。1.1 单Agent的瓶颈在哪里单个Agent最直观的问题就是上下文窗口不够用。现在主流的大模型虽然上下文越做越长但实际用起来你会发现长了之后推理质量明显下降而且是哪里都不够精。我试过让一个Agent同时处理市场调研任务它既要理解行业背景、又要写Python代码抓竞品数据、还要把数据整理成分析文案这些步骤挤在同一段上下文里互相干扰特别严重。最后代码是对的但分析结论很肤浅数据引用的部分还出现了幻觉。第二个问题是缺少检查机制。一个Agent自己干活自己验收等于既当运动员又当裁判。它如果在一个逻辑错误上扎了根后面的所有输出都会顺着错误往下走而且它会非常自信地继续编下去没有任何环节能把它拦住。第三个问题是并发能力差。真实业务不可能一个个任务排队做。我最初让单Agent处理一个月的任务列表它做第三项的时候已经忘了第一项的交付标准。如果不做任务级别的隔离和上下文管理Agent越往后执行质量越不可控。1.2 机构化思路的由来我后来换了个思路既然真实公司不是让员工一个人包全流程而是有产品经理、工程师、测试、项目经理这些分工那我为什么不能让多个Agent也这么干活于是这个项目被改名为agency-agents核心思想就是把一个Agent团队当成一个虚拟机构来设计。项目经理Agent负责拆解任务执行Agent负责具体产出质检Agent负责验收知识库Agent负责提供背景资料。每个Agent只有单一职责不让它跨界。这套思路带来的直接好处有三个每个Agent的上下文变得很干净只用操心它这个环节的数据任何一个环节出问题都能独立重跑不用把整个流程打翻重来可以像管员工一样管Agent给它定KPI交付标准、做评审质检环节、做复盘任务日志。当然这不是什么原创新概念多智能体系统早就有但我从机构这个视角去重新规划角色确实比单纯堆Agent要更有条理得多。2. 整体架构设计与角色拆解2.1 核心角色定义在一个标准的agency-agents系统里我目前稳定在用的有四个角色。第一个是项目经理AgentPM Agent。它不负责具体产出而是接收用户需求把需求拆解成原子任务列表每个任务都带清楚的目标和验收标准然后把任务分派给对应的执行Agent。项目经理Agent的系统提示词里我会明确写几条规则不允许自己动手执行任何任务拆分任务时必须给出可验证的输出物如果需求信息不足必须先向用户提问不能瞎猜。第二个是执行Agent。这其实是后面可以水平扩展的一组Agent比如代码Agent、文案Agent、数据Agent。它们的定位是只接受已经被PM Agent拆好的任务不自己理解用户原始需求。每个执行Agent只暴露它能处理的函数比如生成代码文件产出数据分析表其他一概不管。第三个是质检AgentQA Agent。它专门负责挑毛病不被允许直接修改执行结果只负责打回、给修改意见。质检Agent我会给它一份红线清单比如代码不存在语法级错误报告中每个数字都有数据来源结论和论据必须一致。一旦命中红线就返工。第四个是知识库Agent。这个角色充当的是团队的老员工通过检索为PM和执行Agent提供背景资料降低幻觉风险。2.2 协作机制任务分解、消息路由、结果聚合角色确定之后协作机制就是关键。常见的有三种顺序流水线、星型调度和层级分解。顺序流水线适合任务链路固定、前后依赖很强的场景比如生成文案-配图-排版-发布。优点是简单缺点是只要一个环节卡住整条链都停摆。星型调度是一个中心Agent把所有子任务派给外围的独立Agent然后收集所有人的结果。这种方式并行度高但中心节点容易成为瓶颈而且这种模式更适合每个Agent独立做不同事项的场景不适合前后强依赖的流程。我最终选择的是层级分解也是现在项目里在跑的方案PM Agent把大任务拆成几层子任务每个子任务还可以继续下放给更细的执行Agent。这样做的好处是任务之间的依赖关系完全由PM Agent维护调度器Scheduler)只负责按依赖顺序执行不用理解业务本身。一旦某个子任务失败只需要把这个分支重新跑一遍不会影响其他分支。在任务分派这一步有一点特别重要任务的描述格式必须完全结构化。我用的字段是task_id、parent_task_id、agent_role、input、output_format、acceptance_criteria。如果没有明确的验收标准执行Agent就很容易给出看起来差不多但没法用的结果而质检Agent也会因为没有标准而盲目放行。2.3 为什么选择角色化而非自由对话业内不少多Agent演示看起来很酷几个Agent围在一起自由聊天互相争论最后得出一个结论。但放到实战环境里这种自由对话模式有致命的缺陷不可控、不可追踪、成本高。自由对话意味着你不知道哪一轮对话会产生决定性结论也不知道该在什么节点人为介入。每个Agent之间的交互都是隐式的出问题了很难定位是哪一段逻辑错了。而角色化设计相当于给每个Agent定义了边界和交付物项目经理Agent的交付物是任务拆解表执行Agent的交付物是具体文件或数据质检Agent的交付物是通过或打回意见。每个Agent的输出不再是一段聊天文本而是结构化的工件。说得直白点聊天是过程工件是结果。做业务流程自动化我们关心的是工件不关心过程。角色化强制所有Agent把结果落成链路里的一个节点这样整个系统从外部看就不再是一个会聊天的黑箱而是一条可以审计、可以局部重试的流水线。这是我认为这个项目最核心的取舍。3. 实操过程从零搭建一套agency-agents系统3.1 环境准备与框架选型在技术选型上我并没有纠结太久。当时能选的路有三条直接用某个大模型平台的Agent能力、用开源Agent编排框架、完全自研调度层。大模型平台的Agent能力上手最快但有两个问题一是角色和流程逻辑被平台锁死二是多Agent协作的粒度不够细。开源编排框架也试用过几款它们的生态确实丰富但也给了我一种过度封装的感觉想让Agent完全按照我的流程走往往得迁就框架的抽象。最后我选了折中方案底层大模型走统一的API网关中间层用当前项目里最顺手的Web框架搭一个调度控制台状态存储用关系数据库记录任务流转消息队列负责角色之间的异步通信。不是自夸这套组合让我跑业务逻辑的时候几乎不用在意框架限制每个环节都能看得见、改得动。环境准备方面需要准备的事项如下Python 3.10建议虚拟环境隔离项目依赖关系数据库用于存任务状态和执行记录消息队列用于角色间传递任务通知大模型API网关的统一Key方便切换不同模型预留向量库存历史经验非必须项目中期再加也来得及。3.2 核心代码实现先把最底层的Agent基类和任务数据结构定义清楚。我用一段简化版代码说明核心逻辑。from dataclasses import dataclass, field from enum import Enum from typing import Any, Optional import uuid class TaskStatus(Enum): PENDING pending RUNNING running DONE done FAILED failed BLOCKED blocked dataclass class Task: task_id: str field(default_factorylambda: uuid.uuid4().hex) parent_task_id: Optional[str] None role: str executor # 由哪个角色处理 input_data: dict field(default_factorydict) output_format: str text # text / code / json / file acceptance_criteria: list field(default_factorylist) status: TaskStatus TaskStatus.PENDING result: Optional[Any] None error_message: str depth: int 0 class BaseAgent: def __init__(self, name: str, role: str, model_cfg: dict): self.name name self.role role self.model_cfg model_cfg self.tools {} def register_tool(self, name: str, func: callable): 统一挂载工具的入口方便调度层做权限控制。 self.tools[name] func def run(self, task: Task) - Task: raise NotImplementedError def _call_llm(self, system_prompt: str, user_prompt: str) - str: 实际实现时对接大模型API网关。 pass然后实现一个简化版调度器。调度器的职责很简单从任务队列里取一个任务根据role字段找到对应的Agent调用Agent的run方法更新任务状态。class Scheduler: def __init__(self, agents: dict): self.agents agents self.task_queue [] def submit_task(self, task: Task): self.task_queue.append(task) def run_loop(self, max_steps: int 50): for _ in range(max_steps): if not self.task_queue: break task self.task_queue.pop(0) agent self.agents.get(task.role) if agent is None: task.status TaskStatus.FAILED task.error_message funknown role: {task.role} continue task.status TaskStatus.RUNNING try: task agent.run(task) except Exception as e: task.status TaskStatus.FAILED task.error_message str(e) # 如果任务产生子任务就回填队列 if task.result and isinstance(task.result, list): for st in task.result: self.task_queue.append(st)这个调度器写得非常基础但足够说明问题。真实项目里我会加上并发控制同一时刻最多只允许N个任务并行防止所有Agent同时打爆API配额。3.3 指令、工具与记忆配置多Agent系统里每个Agent的系统提示词要有明确的职责声明和输出约束。举个例子项目经理Agent的提示词开头我会写成你是数字机构中的项目经理。你负责接收用户目标输出结构化任务拆解表。你不负责执行具体任务。每个任务必须包含以下字段role、input_data、output_format、acceptance_criteria。如果你认为需求不明确请直接输出question不要自行编造任务。你的输出必须是一个JSON数组。质检Agent的提示词则是另一个极端你是质量检查员。你只能对输入的工作成果进行验证。你必须根据acceptance_criteria逐条判定给出pass或fail。如果有任意一条不满足输出fail并列出具体修改意见。你无权修改成果本身。这里实际情况是模型对指令的遵循能力有限所以关键信息我会尽量通过结构化字段传入而不是全部塞在自然语言里。比如验收标准字段就会作为一个独立参数传给质检Agent来自任务本身的acceptance_criteria属性。工具的记忆配置方面每个Agent都有短期和长期两层记忆。短期记忆就是在处理当前任务时只携带这个任务链条内传递下来的数据不允许把上一个任务的数据带进来否则就是上下文污染。长期记忆则是把每次任务的经验教训写入向量库下次遇到类似任务时PM Agent会先检索这些经验作为任务拆解的参考。工具注册我用的装饰器方式和Flask里的路由注册很类似def tool(func): 把普通函数注册为Agent可调用工具。 func._is_tool True return func tool def fetch_web_content(url: str) - str: # 爬取网页正文 return 在调用大模型时我会把工具声明转成JSON Schema格式并强制模型限制工具调用遵循这个模式。这一步直接决定了后面工具调用成功率不能偷懒。3.4 运行效果与注意事项我拿一个具体任务测试让这套数字机构生成一篇某消费品牌年度营销方案。完整的流转过程大致是这样的PM Agent先拆出子任务行业背景调研、竞品分析、用户画像、传播策略建议、预算分配竞品分析需要数据Agent去查公开资料传播策略建议需要文案Agent产出方向草案。每个子任务都被标记了依赖关系策略草案必须等到竞品数据产出后才能开始。实际跑一轮下来最大的感受是并行度确实上来了数据Agent在等爬虫结果的同时用户画像Agent已经开始跑调研问卷的结构设计然后质检Agent把所有产物统一过了一遍找到了竞品页里两个过时数据打回让数据Agent重新更新。整套流程没有人工干预但产出质量比我之前单Agent方案高出一截。跑通之后有几个注意事项非常关键不要一次性放开所有Agent的并发权限。我第一版所有角色都是并发执行的结果质检Agent在数据Agent还没跑完时就去验收白跑了一堆无效调用。后来改成下游任务依赖上游输出的规则才把误跑率降下来。任务深度必须限制。如果PM Agent拆得不够彻底就会出现子任务下面还有子任务层级无限加深。我给嵌套深度设了一个上限超过就直接结束并提示PM Agent把拆解粒度放大。所有Agent的调用日志必须完整记录。这个项目的调错能力完全建立在日志上没有日志等于没有排查入口。我后来给每个Agent都加了自动记录任务开始、结束、调用模型、工具调用参数的结构化日志。4. 常见问题与排查技巧实录多Agent系统跑起来之后真正让人头大的永远不是模型能力而是工程层面的稳定性问题。我在这个项目里踩过好几个坑挑典型的几个说。4.1 任务分配死循环现象是任务在Agent之间来回传递始终没有一个Agent产出最终结果。最典型的一次PM Agent把一个总结报告任务派给执行Agent执行Agent觉得需要更多输入数据于是把任务派回给PM Agent当作数据补充任务PM Agent又认为这属于执行工作再次派给执行Agent两个角色之间反复互踢直到触发步数上限。排查思路是先看任务链日志确认每个任务的parent_task_id是否指向自己或形成环形。解决手段有三个给调度器加任务深度上限到点就掐断给每个Agent明确不允许转派任务的规则只能标记为blocked并等待外部输入在路由层做白名单明确每个角色只能接收特定role前缀的任务跨角色转交必须经过调度器审核。4.2 上下文污染执行Agent在处理多个任务时把上一个任务的历史数据带到了新任务里。比如跑完A公司的竞品分析紧接着跑B公司时模型会把A公司的数据当成B公司的写进报告里而且毫无违和感。这个问题排查起来不容易因为模型不会主动告诉你它记串了。我是靠抽样检查生成的文档才发现的。根治措施是每次任务执行前重新创建一个独立的上下文会话把该任务相关的数据作为新的系统输入注入不同任务之间不共享任何会话历史。短期记忆必须按任务隔离不能按Agent隔离。4.3 工具调用失败与重试机制工具调用失败最常见的两种原因参数没填全以及填了不存在的参数名。模型在调用工具时偶尔会自作主张生成一个看似合理但实际不存在的参数或者把必填参数漏掉。第一次遇到时我以为是大模型的Bug后来发现是我工具Schema写得太粗糙。解决办法是把所有工具参数声明为JSON Schema严格模式必填字段全部列出并对每个参数补充说明让模型理解含义。重试机制上我给工具调用加了三档退避失败一次立刻重试、再失败等1秒重试、第三次失败后挂起任务并通知PM Agent调整执行方案。实际跑下来工具失败率从最初的25%降到了3%以内。4.4 成本和时延失控多Agent系统最大的隐性成本不是模型推理本身而是无效对话。一个任务反复被打回重试每个Agent又重新读一遍所有上下文Token消耗成倍上涨但产出并没有变多。我做过一次统计一次完整营销方案任务主流程模型调用次数是7次但因为质检打回和上下文传参方式不当实际调用次数到了18次。查出问题后发现有两个Agent每次对话都把整个项目背景重新塞进去明明只需要传增量数据。优化方案是给Agent对话增加会话轮次上限默认3轮超过就必须由调度器介入同时建立缓存如果同一个输入之前已经产出了结果直接复用跳过调用。5. 进阶如何扩展为真正可用的数字机构基础版本跑通后agency-agents还只能算是一个技术demo。要想真正承担生产环境的业务还需要补上三个能力人工审批、工作流状态机、可视化追踪。5.1 引入人工审批与工单机制纯自动化的Agent流程在执行高风险任务对外发布、涉及预算、引用未知数据时还是需要人的确认。我的做法是在任务数据结构里增加一个requires_approval字段。如果PM Agent拆任务时判断这个子任务需要人工确认调度器就会把任务状态置为等待审批而不是直接丢给执行Agent。审批流程我参考了工单系统的设计每个需要确认的任务就是一个工单人工在控制台里看到工单可以批准、打回或修改输入。这使得Agent系统不是黑箱自动运行而是人机协作的流程工具尤其适合在团队里推广。实际用下来人的压力也不大大多数常规任务一遍通过只有少数关键节点需要参与。5.2 引入工作流状态机把Agent变成流程里的执行节点一开始我就是用代码逻辑控制任务流转状态一多就开始乱。后来我引入了状态机模型把整个业务流程显式定义成一张图每个节点是Agent调用人工审批消息通知或逻辑判断节点之间的连线是转移条件。这套改造最大的收益是让业务流程和Agent解耦了。业务流程是稳定的Agent是可以替换的。以后要换一个大模型或者换一个新Agent只需要在配置中心里把节点对应的Agent映射改掉流程本身不用动。对于要长期维护的项目来说这个分层的价值远大于省掉一点配置功夫。5.3 结果可追踪与复盘多Agent系统出问题时最大的难点是定位是哪一步的哪一次调用导致最终产出出了问题。所以我后来把所有的任务流转事件、Agent调用记录、工具返回结果、质检判定结果都持久化存储并给每个最终交付物生成了一个完整的生产过程报告。复盘时可以直接看报告某个子任务被质检打回两次原因都是数据来源缺失数据Agent平均每次任务要调四次工具PM Agent的拆解结果中有一半任务被质检判定为验收标准不清晰。这些数据反过来又能指导调整提示词或流程参数形成闭环。这也是我做这个项目最深的体会Agent本身的能力不是瓶颈流程设计和工程管控才是。把Agent当成一个需要管理、需要验收、需要复盘的“临时员工”而不是一个神奇黑箱这套系统才能稳定地跑出业务价值。