这几年做 AI Agent 项目,我踩过最大的坑不是模型选型,不是提示词调优,而是单 Agent 的天花板。单个 Agent 看起来能聊天、能调用工具、能写代码,但一旦任务复杂起来——要同时处理调研、方案、执行、质检——它就手忙脚乱,经常把上下文搞成一锅粥。后来我转向了多 Agent 协作的方向,做了一个内部代号叫agency-agents的项目,核心思路不复杂:把一堆 Agent 当成一家公司来组织。有人当主管,有人当执行者,有人当质检员,互相之间通过消息协作,而不是一个巨型 Agent 包揽所有事。
这篇文章就是我对这个项目的完整复盘。我会从设计思路、架构拆解、核心实现、真实踩坑这几个维度展开,项目里的代码结构我会用简化后的伪代码和配置片段来说明。如果你也在做多 Agent 系统,或者准备从单 Agent 往多 Agent 迁移,这篇文章应该能帮你少走不少弯路。
1. 项目背景与设计思路:为什么要把 Agent 组织成“机构”
1.1 从单 Agent 到多 Agent:到底解决了什么问题
先说一个真实的例子。我之前做过一个智能客服机器人,单 Agent 接所有问题:查订单、退换货、投诉、咨询售后政策。看起来功能齐全,实际上问题很多。第一个问题是上下文污染,用户聊到一半说“那上次那个呢”,单 Agent 根本不知道“上次”指的是订单还是投诉;第二个问题是职责不分,一个模型既要懂业务逻辑又要懂话术,提示词写得越来越长,模型行为越来越不稳定;第三个问题是难以扩展,想加一个“价格保护”功能,得改动整套提示词,风险极高。
agency-agents 想解决的,就是这一类“职责杂糅”和“上下文混乱”的问题。我把它设计成一个小型的组织架构,每个 Agent 只负责一件事,Agent 之间通过任务消息传递协作,而不是把所有逻辑塞进一个系统提示词里。这样做的好处很直接:每个 Agent 的提示词短小专注,职责边界清晰,出了问题可以单独替换某个 Agent,不用动全局。
做这个项目之前,我调研了市面上几种主流的多 Agent 框架,各有特点,但总有些地方不适合我的场景。有的框架太重,光调度层就要配一套消息队列;有的框架只支持对话流,没法做任务队列和异步回调;有的框架把 Agent 定义得太死板,我想让一个 Agent 既当执行者又当复核者,结果它不支持。所以最后我决定自己写一个轻量级的编排层,核心就一句话:Agent 是函数,消息是数据,编排是循环。
1.2 组织架构类比:主管、专员、质检员如何映射到系统
我一直觉得,拿公司组织架构来类比 Agent 协作是最容易理解的。一家公司要完成一个大项目,不会让一个人从头干到尾,而是分给不同部门,每个部门里又分角色。agency-agents 就把这个逻辑搬到了代码里。
整个系统里有三类角色:
- 主管 Agent(Manager):接收任务,拆解任务,决定派给哪个执行 Agent,汇总结果,决定任务是否算完成。它相当于项目负责人,但不亲自下场干具体活。
- 执行 Agent(Worker):负责具体某个领域的工作,比如写代码、写文案、查资料、算数据。它只接受主管派发的任务,做完之后把结果交回去。
- 质检 Agent(Reviewer):负责验收执行 Agent 的产出。它如果觉得不合格,会打回让 Worker 重做,并且附上修改意见。
这三类角色之间不直接乱聊天,所有沟通都走任务消息。我一开始也尝试过让 Worker 之间直接互相讨论,结果发现两个问题:一是消息量爆炸,一个 100 行的任务能产生几百条对话记录;二是责任不清,出了错不知道是哪个 Agent 的问题。改成主管统一调度之后,任务流转路径变得非常清晰:主管接收 → 拆解 → 派发 → 工人执行 → 质检验收 → 打回或提交。这种方式很像公司里的工作流,管理和追责都容易。
实际跑起来之后,我发现这个类比还有个额外的好处:新 Agent 的接入成本极低。你想加一个“翻译 Agent”,只需要注册一个 Worker,告诉主管它能处理什么类型的任务,主管在拆解任务时就会自动把它考虑进去。整个系统像搭积木一样,每个 Agent 就是一块积木,主管负责把积木拼起来。
2. 架构设计与核心机制:系统是怎么运转起来的
2.1 Agent 注册表:如何描述一个 Agent 能干什么
agency-agents 的架构核心是一个Agent 注册表(Registry)。每个 Agent 在启动时要注册自己的信息,包括名字、角色类型、擅长领域、输入输出格式、以及异步回调地址。注册表的工作类似公司的通讯录加职位说明书,主管接到任务之后,先去注册表里查一圈,看哪些 Agent 能接这个活。
注册信息我用的是 JSON Schema 描述,这样后续可以自动校验消息格式。举个例子,一个“文案撰写 Worker”的注册信息大概是这个样子:
{ "agent_id": "copywriter_worker_001", "name": "文案专员", "role": "worker", "skills": ["文案创作", "广告语撰写", "内容改写"], "input_schema": { "type": "object", "properties": { "topic": { "type": "string" }, "tone": { "type": "string" }, "max_length": { "type": "integer" } }, "required": ["topic"] }, "output_schema": { "type": "object", "properties": { "draft": { "type": "string" }, "summary": { "type": "string" } }, "required": ["draft"] }, "callback_url": "http://internal.agent-network/copywriter/result" }这里需要注意几个点。skills字段很关键,它是主管 Agent 做任务分派时的匹配依据。我一开始想用自然语言描述技能,比如“擅长写有创意的文案”,但后来发现用短标签匹配最靠谱,因为大模型做标签匹配比做语义匹配稳定得多。input_schema和output_schema是给主管用的,它拆解任务之后必须生成符合input_schema的字段,否则 Worker 那边解析会出错。
注册表本身我用了内存版加 Redis 持久化两层。内存版保证速度快,Redis 保证多实例部署时各个 Agent 能互相找到对方。实际部署时,每个 Agent 都是一个独立服务,启动后向 Redis 里写自己的注册信息,同时开一个 HTTP 服务等回调。这种分布式结构看起来复杂,实际上每个 Agent 就是一个很简单的 Web 服务,关键逻辑都在主管那里。
2.2 任务消息设计:状态机驱动的流转过程
任务消息是整个系统的血液。我把它定义成一个带状态的对象,流转过程就是一个状态机。任务对象大致长这样:
{ "task_id": "task_8f3a2b", "parent_task_id": null, "manager_id": "manager_agent_001", "assignee_id": "copywriter_worker_001", "reviewer_id": "reviewer_agent_001", "status": "pending_review", "input": { "topic": "夏季促销文案", "tone": "活泼", "max_length": 200 }, "output": { "draft": "夏日炎炎,折扣不停..." }, "review_comment": "", "attempts": 2, "created_at": "2024-05-20T10:00:00Z", "updated_at": "2024-05-20T10:05:30Z" }状态机上定义了这些状态:pending(等待分派)、assigned(已派发)、executing(执行中)、pending_review(等待质检)、approved(已通过)、rejected(被打回)、done(完成)、failed(失败)。每次状态变更都会触发一个事件,主管 Agent 监听事件并做对应处理。
这里我想强调一个设计细节:任务重试次数attempts一定要有上限。我最初没有设上限,结果一个质量不行的文案被质检 Agent 打回,Worker 改了一遍还是不行,再改还是不行,形成了一个死循环。后来我加了max_attempts,默认 3 次,超过次数就直接转人工处理,或者由主管降级接受。
任务消息的传递我用的是 HTTP 回调加消息队列两种方式。内部 Agent 之间通信比较多,我就用轻量级的消息队列,保证消息不丢。如果 Agent 是外部服务,就用 HTTP 回调。设计原则很简单:内部通信要快,外部通信要稳。
2.3 主管 Agent 的任务拆解逻辑:从“一句话需求”到“可执行动作”
主管 Agent 是整个系统里最核心的部分,它的任务拆解质量直接决定了整个系统的效果。我一开始想直接让大模型做拆解,给他一句话“帮我把夏季促销方案做了”,它输出一个长长的计划,看起来很专业,但无法直接落地。后来我改成两步走:第一步,用模型生成一个高层次计划;第二步,用一个确定性代码模块把计划转换成具体的任务对象。
举个例子,“做夏季促销方案”这个需求,模型可能生成这样的计划:
- 调研竞品促销策略
- 设计促销主题与文案
- 制定折扣方案
- 生成推广渠道建议
然后确定性模块会把每一条转成任务对象,分配给对应技能的 Worker。调研任务分配给“调研 Worker”,文案任务分配给“文案 Worker”,折扣方案分配给“策略 Worker”,推广建议分配给“渠道 Worker”。
第二步的转换过程为什么用确定性代码而不是模型?因为我发现模型在这步容易“自由发挥”,生成一些字段名对不上的任务对象。确定性代码虽然死板,但可靠。我写了一个简单的规则引擎,基于skills标签去注册表里匹配 Worker,匹配不到就把任务标记为unassigned,发回给主管重新拆解。
主管 Agent 还需要处理一个特殊情况:任务之间的依赖关系。促销主题设计完了,才能写具体文案,所以文案任务要等主题设计任务完成之后才能执行。我在任务对象里加了一个depends_on字段,主管派发任务时必须检查依赖是否满足,不满足就先挂起。这个依赖检查逻辑我也用确定性代码实现,不靠模型判断,因为模型的判断不稳定,容易把依赖搞反。
3. 核心实现细节:代码层面的关键设计与踩坑记录
3.1 提示词工程的分解:每个 Agent 的提示词该怎么写
多 Agent 系统的提示词设计和单 Agent 完全不同。单 Agent 追求把场景、规则、例子、知识一次性塞进去;多 Agent 追求每个 Agent 的提示词小而精,并且彼此之间要衔接。
我给每个 Worker 写的提示词模板一般分四块:
- 角色定义:一句话说清楚你是谁,负责什么。
- 任务输入说明:说明你会收到什么字段,每个字段是什么意思。
- 输出要求:明确输出的结构、长度、风格。
- 边界声明:明确告诉你遇到什么情况该拒绝执行或请求主管介入。
这里最容易被忽视的是第四块。我之前做的一个“代码生成 Worker”,用户让它生成一段 Python 脚本,它生成完之后还自作主张加了一堆单元测试,结果多出来的部分既不符合格式要求,又拖慢了响应时间。后来我在边界声明里加了这么一句:“你只负责生成脚本本身,不要生成测试代码、不要解释代码、不要建议额外功能,除非任务输入中明确要求。”加了这句话之后,输出稳定了非常多。
主管 Agent 的提示词又不一样。它的核心能力是“拆解”和“决策”,不是“生成”。它的提示词里除了角色定义,还需要一个分配策略说明,比如:
你是任务主管。你收到一个整体目标后,需要拆解为多个独立子任务。 每个子任务必须指定一个 assignee。 分配原则: - 优先选择 skills 匹配度最高的 Worker; - 如果多个 Worker 匹配,选择当前负载最低的; - 如果没有任何 Worker 匹配,将子任务标记为 unassigned,并在注释中说明缺失技能。这段提示词看起来简单,但它能管住主管不出乱子。没有这个约束的时候,主管经常把任务派给不相关的人,比如让文案 Worker 去算折扣数据,让数据 Worker 去写文案。加了分配原则之后,这类问题大幅减少。
质检 Agent 的提示词核心是“标准”。我把质检标准直接写进提示词里,比如:
你是质检专员。你只负责检查任务输出是否满足要求,不负责修改内容。 检查维度包括: 1. 是否严格符合 output_schema 的字段要求; 2. 内容是否完整覆盖任务输入中的所有要求; 3. 语言风格是否和输入中的 tone 描述一致; 4. 是否存在明显的逻辑错误或事实错误。 你的输出必须是 JSON 格式,包含 approved(布尔值)和 comment(修改建议)。质检 Agent 的提示词是关键中的关键。它如果太宽泛,什么都能打回,那系统就没法用了;它如果太严格,输出永远过不了。我的经验是,质检标准要从任务输入里动态生成,而不是写死一套通用标准。比如输入里写了“语气活泼”,质检标准的第一条就检查文案里有没有感叹号、网络新词等“活泼”信号;输入里写了“字数不超过 200”,质检标准就明确检查字数上限。
3.2 工具调用规范:Agent 能碰什么工具,碰之前先过一道闸
多 Agent 系统里,工具调用是一个容易失控的地方。Worker 能访问数据库、能调第三方 API、能执行代码,如果权限控制不严,一次错误的调用就能造成大问题。
我在 agency-agents 里实现了工具白名单机制。每个 Agent 注册时声明自己允许调用的工具列表,主管在派发任务时会把工具权限也一起传下去。Worker 在实际调用工具前,还要经过一层参数校验,防止注入和越权访问。
举个例子,我系统里有一个“数据查询 Worker”,它允许调用数据库查询工具,但不允许调用删除和更新工具。它的工具权限配置大概是:
{ "agent_id": "data_query_worker", "allowed_tools": [ { "name": "database.query", "args_schema": { "sql": "string" } }, { "name": "http.get", "args_schema": { "url": "string", "headers": "object" } } ], "denied_tools": ["database.execute", "file.write", "http.post"] }这个白名单在派发任务时绑定到任务上下文里,Worker 的代码里有一个统一入口,每次调工具前都检查一下当前任务上下文是否允许调用这个工具。不允许就直接返回一个工具错误,并触发主管介入。
还有一点很容易被忽略:外部 API 调用必须有超时和重试机制。我踩过一次坑,一个 HTTP 调用的 Worker 去请求一个第三方接口,对方响应很慢,Worker 就一直等,导致整个任务链卡住。后来我给所有工具调用加了超时控制,默认 30 秒,超时直接返回错误。重试逻辑放在任务状态机里,而不是放在工具层,因为重试次数和退避策略应该由主管统一控制,而不是每个 Worker 各自为战。
3.3 状态管理与上下文传递:如何避免多 Agent 之间的信息丢失
多 Agent 系统最常见的运行问题之一就是上下文丢失。A Agent 生成了某个结论,B Agent 不知道这个结论是怎么来的,只能看到结论本身,导致后续任务质量下降。
我在设计状态管理时,给每个任务对象加了一个context字段,专门存放“背景信息”。主管在拆解任务时,会把相关的背景信息复制到子任务的 context 里。Worker 执行完成后,它要返回两样东西:一是output(最终产出),二是execution_summary(执行摘要),执行摘要里要写清楚它做了哪些步骤、使用了哪些数据、得出了什么中间结论。
具体的状态保存我用了 Redis 加 JSON 存储。每个任务的完整历史都保存下来,这样主管做汇总时能回溯到每个子任务的执行过程。不这样做的话,主管生成最终总结时就只能看到结果,看不到过程,一旦结果有问题,很难定位是哪一步出的问题。
这里分享一个实用技巧:给每条上下文加一个来源标签。比如context["source"] = "task_8f3a2b",这样后续任何 Agent 使用这段上下文,都知道它来自哪个任务。排查问题时,顺着来源标签就能理清上下文链条。
状态管理还有一个坑:并发更新导致的状态覆盖。两个 Worker 同时完成各自的任务,各自回调更新同一个父任务对象,如果没有锁机制,后更新的会把先更新的覆盖掉。我给任务状态更新加了版本号机制,每次更新前检查版本号是否匹配,不匹配就重新读取再更新。这个机制在单机部署时可能用不到,但一旦上多实例,就非常必要。
4. 实操过程:从零搭起 agency-agents 的最小可用版本
4.1 环境准备与模块拆分:先搭骨架再填血肉
agency-agents 这个项目我用 Python 写的,因为生态里现成的 Agent 工具和模型接口最全。核心依赖就三样:一个 Web 服务框架(我用的是 FastAPI),一个 Redis 客户端,一个 LLM 调用封装。数据库用来存任务对象和执行历史,因为任务对象是 JSON 结构,所以我直接用 Redis 加一个本地 JSON 文件存档。如果想上生产,建议加一个正式的文档型数据库,比如 MongoDB,但最小可用版本阶段没必要。
项目结构我按角色和服务来组织,大概长这样:
agency-agents/ ├── core/ │ ├── registry.py # Agent 注册表管理 │ ├── task_manager.py # 任务状态机与流转逻辑 │ ├── dispatcher.py # 主管的任务拆解与分配 │ └── message_bus.py # 内部消息队列封装 ├── agents/ │ ├── manager.py # 主管 Agent 实现 │ ├── reviewer.py # 质检 Agent 实现 │ ├── worker_research.py # 调研 Worker │ ├── worker_copy.py # 文案 Worker │ └── worker_data.py # 数据 Worker ├── schemas/ │ └── task_schema.json # 任务对象的 JSON Schema └── main.py # 服务入口这个结构看起来很规整,但我要说,它是我重构了两次之后才得到的。第一次我图省事,把所有 Agent 写在一个文件里,结果代码膨胀得非常快,改一处崩一片。第二次我按“角色”分文件,但把公共逻辑和各个 Agent 的实现混在一起,后来单独抽了core目录才清爽。我的建议是:先把公共逻辑抽干净,再写具体 Agent 的实现。
环境上我用的是 Docker 加 Docker Compose,一个容器跑 Redis,一个容器跑所有 Agent 服务。最开始我把每个 Agent 都单独打包一个容器,后来发现调试太麻烦,日志分散在十来个容器里,找一条报错得翻很久。最小可用版本阶段,单个容器跑所有 Agent 服务足够了,上生产再拆。拆容器有一个折中方案:同一个容器里不同 Agent 用不同的端口,日志统一收集到一个目录,排查问题时一次能看完所有日志。
4.2 第一版代码实现:主管怎么拆任务、Worker 怎么接活
我先说最核心的代码流程,主管拆解并派发任务的逻辑。
主管 Agent 的核心是一个循环:接收一个整体目标,调用 LLM 生成计划,用规则引擎把计划转成子任务,然后逐个派发。这里的代码我用简化后的伪代码展示:
async def manager_loop(goal: str, registry, task_manager): # 步骤1:生成高层次计划 plan = await llm_generate(manager_prompt, goal) plan_items = parse_plan(plan) # 步骤2:将计划转成子任务 subtasks = [] for item in plan_items: task = build_task_from_plan_item(item, registry) subtasks.append(task) # 步骤3:检查依赖,逐个派发 for task in subtasks: if all(dep.status == "done" for dep in task.depends_on): await task_manager.dispatch(task) else: task_manager.hold(task)build_task_from_plan_item这个函数就是我用确定性代码实现的规则引擎部分。它做的事情是:从计划条目里提取关键词,匹配 Worker 的 skills 标签,生成符合 input_schema 的任务对象,找不到匹配就标记为 unassigned。
Worker 这边,核心逻辑非常简单,就是一个“收任务 → 干活 → 返回结果”的循环。以“文案 Worker”为例,它的处理函数大概是:
async def copywriter_worker(task: Task): # 校验输入是否符合约定 validate_input(task.input, input_schema) # 生成文案 draft = await llm_generate(copywriter_prompt, task.input) # 返回结果 task.status = "done" task.output = {"draft": draft, "summary": summarize(draft)} task.execution_summary = "使用了输入中的主题和语气要求,生成了一版文案草稿。" await task_manager.report(task)这套流程看起来简单,但我实际写的时候遇到了一个很纠结的问题:Worker 判断自己有没有完成,应该看什么?我一开始让 Worker 自己判断,结果它经常“觉得”自己完成了,但产出的内容跟要求差十万八千里。后来我把完成判断交给了质检 Agent,Worker 只负责生成内容和生成执行摘要,是否合格完全由质检 Agent 说了算。这个改动让系统的整体质量上了一个台阶。
质检 Agent 的代码也简单,关键是它要能访问到任务输入、Worker 产出和质检标准:
async def reviewer_agent(task: Task, review_criteria): review_result = await llm_generate( reviewer_prompt, { "input": task.input, "output": task.output, "criteria": review_criteria } ) if review_result["approved"]: task.status = "approved" await task_manager.finish(task) else: task.status = "rejected" task.review_comment = review_result["comment"] await task_manager.reassign(task)这里我特别要说一下质检标准的来源。最初质检标准是硬编码在提示词里的,后来我发现不同任务的需求差异很大,便改成了“主管生成任务时附带质检标准”。主管在拆解任务时,会同时生成一段任务专属的质检标准,比如“字数不超过 200”“必须包含 3 个营销关键词”“语气必须活泼”。质检 Agent 收到任务后,直接用这段标准做检查。这样质检更有针对性,也不会过度泛化。
4.3 运行效果与调优:一次真实的压测记录
我把三类的 Agent 部署起来之后,跑了一个模拟场景:让系统生成一份“夏季校园推广方案”。任务目标是“针对高校市场,设计一个夏季促销活动方案,包含主题、折扣和推广渠道”。我故意用了比较模糊的描述,想看看系统的拆解能力。
第一轮跑下来,主管生成了一份五步计划:调研市场、确定主题、设计折扣、选择推广渠道、生成总结。前四步都匹配到了对应的 Worker,第五步“生成总结”没有匹配到任何 Worker。主管标注了 unassigned,并且生成了一条注释:“该步骤过于笼统,请重新指定具体内容。”
这个结果我非常满意,因为主管正确地识别出了“生成总结”不是一个具体任务,而且没有强行硬塞给某个 Worker。我修改了系统提示词,让主管在拆解计划时直接跳过“总结”“汇报”这类抽象动作,把重点放在具体的可执行动作上。改完之后,第二轮跑得干净多了。
再往后就是压测。我模拟了并发 50 个任务同时进入系统,场景是“每个人提交一篇产品文案需求,系统自动分配给文案 Worker,写完再由质检审核”。最初跑的时候,系统频繁出现两个问题:一是 Redis 连接数被打满,二是任务状态出现脏写。
Redis 连接数的问题好解决,加连接池就行。脏写的问题花了点功夫,现象是两个 Worker 同时完成修改同一个父任务的状态,结果 A 的状态覆盖了 B 的状态。这就是我之前提到的版本号机制发挥作用的地方。我重新设计了任务更新函数,要求每次更新带一个期望版本号,更新时先检查,不匹配就重试。修完之后,并发场景下的状态一致性就稳定了。
调优过程中还有一个比较隐蔽的问题:主管 Agent 的响应延迟。由于所有任务都要先经过主管拆解,主管成了整个系统的瓶颈。我做了两步优化:一是把主管的 LLM 调用从同步改成异步,拆解任务时不用等全部完成再返回;二是给主管加了一个简单的任务池,任务进来先入池,主管从池里批量取任务拆解,减少模型调用次数。这两步优化把系统吞吐量提升了不少。
5. 常见问题与排查技巧实录
5.1 Agent 之间的死循环:如何识别并彻底消除
多 Agent 系统最折磨人的问题就是死循环,两个 Agent 互相打回、互相重试,日志刷了几百条还没结束。我遇到最典型的一次是:质检 Agent 认为文案语气不够活泼,打回让文案 Worker 改;文案 Worker 改完之后质检 Agent 还是认为不够活泼,再次打回;Worker 一脸懵,因为它的提示词里根本没有“活泼”的量化标准,只能反复改措辞碰运气。
这个问题的根源在于质检标准不可量化。后来我把质检标准从笼统的“语气活泼”改成了具体规则:“文案中是否包含至少 3 个感叹号”“是否包含至少 2 个网络流行语”“是否有至少 1 个反问句”。质检 Agent 按规则逐条检查,不合格就给出具体哪一条不满足。文案 Worker 收到具体反馈之后,修改就非常有针对性,不再盲目兜圈子。
死循环还有一个常见原因:重试机制没有上限。我给任务加上了max_attempts字段之后,循环最多转三次,第四次直接进入人工处理队列。这个人工处理队列在项目早期就是一封邮件加一个待办,后来接了一个简单的审批界面。我觉得做多 Agent 系统必须保留一个人工兜底的出口,因为模型行为不可能 100% 可控,总有模型无论如何都绕不过去的坑。
另外我总结了一个死循环识别的经验:监控任务的attempts字段,如果某个任务在短时间内连续增加 3 次以上,基本可以判定进入了异常循环,立刻触发告警。在告警里把任务的完整状态和历史执行摘要打出来,排查起来会很快。
5.2 上下文丢失与“失忆”问题:Agent 老是忘了前面说过什么
多 Agent 系统中,“失忆”现象非常普遍。现象是:主管要求 Worker 写文案时必须引用前面某个 Agent 的调研数据,但 Worker 最终产出的内容跟那个数据完全对不上,因为它根本没接收到那段数据。
这个问题的根因在我的设计里很明确:主管在派发任务时,没有把相关的上下文一并传给 Worker。我最初只在任务对象里放了input,当 Worker 需要背景资料时,它只能自己猜。后来我强制在任务派发时复制一段context给子任务,并且把“是否携带必要上下文”作为质检标准之一。如果 Worker 的产出与背景信息有明显矛盾,质检直接打回。
这里有一个取舍问题:context 太多会让模型的注意力分散,太少又会让 Worker 缺乏关键信息。我的经验是,context 只放与当前子任务直接相关的信息,不要一股脑把整个项目的所有历史都塞进去。比如“写折扣方案”这个子任务,它需要的 context 是调研结果里的竞品折扣区间和用户价格敏感度,其他不相干的调研细节可以省略。
我还做了一个“上下文摘要”机制。父任务完成之后,如果它要作为后续任务的 context,主管会先让汇总 Agent 把它压缩成一个简要摘要,而不是直接传整个原始输出。这样既保留了关键信息,又不会让 context 无限膨胀。总结一下就是:上下文传递要“按需供给”,不是“全部转储”。
5.3 工具误调用与权限越界:一次“危险操作”的排查实录
这个坑我印象太深了。我早期做的一个系统里,数据分析 Worker 被授权能执行代码,有一次它为了算一个平均值,直接写了一段 Python 脚本,里面调用了一个系统接口,结果误删了一个临时表。虽然只是临时表,但那次事故让我下定决心做工具权限白名单。
白名单机制上线之后,又发现了一个新问题:Worker 可能会“绕道”调用工具。比如数据 Worker 不能直接执行删除操作,但它可以先调用一个“查询接口”拿数据,再通过“写文件接口”把数据写到本地脚本里,间接实现删除。这其实是 Agent 能力组合造成的越权,单看每次调用都合理,组合起来就危险了。
我的对策是两层:第一层是工具层的白名单,拦掉大多数明显的违规调用;第二层是日志监控,记录每次工具调用的参数和目的,并用一个规则引擎识别可疑操作组合。比如“查询接口 + 写文件接口 + 环境变量读取接口”这个组合,如果出现在同一个 Worker 的同一次任务里,就会触发告警。
还有一个容易被忽略的坑:工具参数里的注入攻击。Worker 通过 LLM 生成 SQL 查询时,可能把用户输入直接拼进 SQL 里,造成 SQL 注入风险。我给所有工具的参数校验层加了“参数类型检查和关键词过滤”,凡是 SQL 字符串里出现DROP、DELETE、;都会被拦下并告警。这套机制虽然不能完全防住所有攻击,但能挡住绝大多数常见试探。
6. 经验总结与后续扩展方向
6.1 我重新认识到的三件事:架构、提示词、监控缺一不可
agency-agents 从想法到跑通,到压测稳定,用了大概一个多月时间。这个过程让我对多 Agent 系统有了几个很深的体会。
第一,架构设计决定了系统的上限。单 Agent 的问题靠堆提示词解决不了,必须从架构上拆分职责。agent-agents 的“主管-工人-质检”架构看起来朴实无华,但它确实让每个 Agent 的职责清晰、上下文可控、任务可追踪。如果你现在也深陷于一个巨型 Agent 的泥潭,我建议你先把它的职责拆开,哪怕拆出来的每个 Agent 都很简单,也比一个什么都干的大 Agent 强。
第二,提示词工程在多 Agent 系统里不是终点,而是起点。单 Agent 的提示词可以穷举规则,多 Agent 的提示词必须靠任务结构和质检标准来兜底。agent-agents 的实践证明,给每个 Agent 写一个“小而专注”的提示词,比把所有需求塞进一个提示词里要可靠得多。
第三,运维和监控必须从第一天就做起来。多 Agent 系统的运行状态并发度高、流转路径长,没有好的日志和监控,出了问题根本无从查起。我后来给系统加了一个任务追踪面板,能看到每个任务当前停在哪个状态、经过哪些 Agent、每次操作耗时多久。这个面板成了排查问题的头号工具。
6.2 后续可以扩展的方向:自适应分工与记忆沉淀
做完这个最小可用版本之后,我脑子里还有几个明确想扩展的方向。第一个是自适应分工:现在主管拆解任务依赖硬编码的规则引擎,技能的标签匹配很死板。我想让主管能根据任务动态调整子任务粒度,比如“这个任务拆成三步就够了”“那个任务需要再拆细一点”。这就需要一个反馈回路,主管根据历史任务的质检结果来调整拆解策略。
第二个是记忆沉淀机制:现在每次任务完成的执行摘要只存在任务记录里,没有被充分利用。我想把这些执行摘要汇总成某种“组织记忆”,当类似的新任务进来时,主管可以先检索组织记忆,把过去的经验带进新任务的上下文里。这个机制如果做出来,系统的学习能力会上一个台阶。
第三个是质量评估闭环:当前质检 Agent 只有“过/不过”两档判断,我想改成多维评分,并且让评分结果反过来影响任务派发策略。如果一个 Worker 的文案质量分长期偏低,主管可以自动降低它的任务优先级,或者给它分配更多的时间去修改。这些扩展方向要想全部落地,还需要不少时间,但核心架构已经给这些扩展留好了接口。
最后分享一个实操上的小建议:做多 Agent 系统不要一上来就追求“全自动、零人工”。先保留一个人工干预的口子,让系统在关键时刻可以暂停、可以求助。我见过很多项目因为过度追求自动化,最后系统在无人值守的状态下越跑越偏,收拾残局比手工处理还累。好的多 Agent 系统应该像是带着一个靠谱的团队干活,而不是丢给一台永不停止的机器。
项目推进到现在,我觉得最有成就感的不是系统跑得多快、任务完成率多高,而是我终于理解了“组织”这个词在 Agent 系统里的深刻含义。把 Agent 组织成机构,不是简单的角色划分,而是让每个 Agent 在清晰的边界和明确的协作关系里,发挥出自己最大的价值。这一点,和带真实团队工作的道理很像。