资讯详情

资讯详情

多Agent协作系统实战:从单Agent到组织化架构的完整复盘

这几年做 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 就把这个逻辑搬到了代码里。整个系统里有三类角色主管 AgentManager接收任务拆解任务决定派给哪个执行 Agent汇总结果决定任务是否算完成。它相当于项目负责人但不亲自下场干具体活。执行 AgentWorker负责具体某个领域的工作比如写代码、写文案、查资料、算数据。它只接受主管派发的任务做完之后把结果交回去。质检 AgentReviewer负责验收执行 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 自己判断结果它经常“觉得”自己完成了但产出的内容跟要求差十万八千里。后来我把完成判断交给了质检 AgentWorker 只负责生成内容和生成执行摘要是否合格完全由质检 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 在清晰的边界和明确的协作关系里发挥出自己最大的价值。这一点和带真实团队工作的道理很像。
觉得有用,分享给同行:

为您的企业打造数字门面

稳重轻奢商务风格,端正雅致视觉,长效耐看不易过时。

立即咨询 →