去年我在做内部自动化工具的时候第一次认真思考agency-agents该怎么落地。当时的需求并不复杂给出一份杂乱的需求文档让大模型把它拆成可执行的开发任务再针对每个任务写出实现说明最后整体自检一遍。听起来很常规可单Agent模式跑了不到两周问题就集中爆发了。后来我干脆把项目名改成agency-agents——一个像虚拟团队一样工作的多Agent协作层。这篇文章就把它的设计思路、核心代码和实测踩坑一次性讲清楚。这个项目适合三类人正在做AI应用开发、想尝试多Agent协作但不知道怎么组织的朋友手里已经接了多个模型服务、想让不同角色各司其职的开发者以及单Agent方案反复撞墙、想换一种协作方式的人。就算你暂时不写代码光看架构思路也会对多Agent到底怎么协作有一个比很多教程都更清楚的理解。1. 为什么我会把多个Agent组织成一家虚拟公司1.1 单Agent模式在复杂任务面前的四个信号第一个信号是角色混乱。让模型既当产品经理、又当架构师、还当测试它会在同一轮回答里一会儿改需求、一会儿写代码、一会儿给自己挑毛病最后产出一锅乱炖。第二个信号是上下文稀释任务越往后最开始定义的目标和约束条件越容易被遗忘我曾在第三轮之后反复看到它把数据权限按团队隔离这种硬需求漏掉。第三个信号是错误难定位。一段错误的输出到底是在哪个环节被引入的完全说不清只能把整段历史翻出来人工核对。第四个信号是token浪费——每轮都要把全部历史对话重传一遍成本随着轮次线性上涨效果却逐步下降。这四个信号背后其实是同一个原因单个模型实例在同一个上下文里连续扮演多个角色超过了它有效注意力能够覆盖的边界。打个比方让一个人同时当项目经理、文案、财务和客服忙起来必然顾此失彼但如果你让四个人分工每个人只需要记住自己那一亩三分地整体反而更稳。我就是在这个对比里决定把架构改成多角色协作的。1.2 agency模式借鉴的是现实组织的分工逻辑agency这个词的灵感来自现实中的广告公司和咨询公司。它们很少让一个人从头干到尾客户经理负责对接需求策略团队出方案创意团队做执行最后还有一层质检。员工各管一摊、交接标准明确、有人对总结果负责。这套逻辑搬到Agent上就是三件事角色边界清楚、交接物标准化、调度与质检集中。所以我把这个项目定位成一层编排层不是一个大模型应用。它不关心你接的是哪个商用模型接口也不替Agent写业务知识它只负责几件小事定义角色、派发任务、传递上下文、回收结果。项目因此很轻核心代码没有动辄几千行却把多个Agent怎么协作这个最常见的问题管住了。在动手前我也想了很多市面上已经有那么多Agent框架自己再造一遍轮子有意义吗答案是看你想解决的问题是什么。如果你只想快速搭一个能对话的Agent现有框架够用但如果你的痛点是多个角色如何稳定交接、任务状态如何可审计、模型服务如何按角色分配那你就需要这样一层轻量编排逻辑。我的判断是永远别让框架替你决定协作模式你自己的业务才该决定协作模式。2. agency-agents的核心架构注册中心、任务调度与协作上下文开始写代码之前我先把整个系统的运作模式拆开——不是画漂亮的流程图而是明确谁手里有什么数据、谁对什么负责。最终沉淀下来四个模块下面这张表是我在项目文档里反复改了四五版才定下来的。模块职责关键产出注册中心 Registry登记所有Agent的身份、能力、输入输出声明能力索引调度器 Dispatcher把任务匹配到合适的Agent驱动整条协作链任务分派记录上下文存储 Context Store保存任务中间结果、角色之间的交接物结构化上下文Agent Worker真正调用大模型接口的执行单元结构化结果Registry解决有哪些角色、各自能干什么Dispatcher解决某件事该找谁Context Store解决上个角色说了什么、下个角色该知道什么Worker解决最终调用模型干活。四者缺一不可。最容易被忽略的是Context Store——很多人以为多Agent就是多个模型互相传消息忽略了上下文本身需要统一管理结果协作起来到处丢信息。2.1 Agent的能力声明让调度器看懂每个角色为了让Dispatcher不写死逻辑每个Agent在上线前都要提交一份能力声明。我用的格式是YAML既好读又方便直接做成配置文件。name: requirements_analyst role: 需求分析员 description: 将杂乱需求文档整理成结构化任务清单 capabilities: - requirement_analysis - task_decomposition accepts: type: raw_document max_size: 20000 emits: type: task_plan schema: task_plan_v1 model: provider: generic_commercial_llm temperature: 0.2 max_retries: 2这份声明里几个字段很关键。capabilities是给调度器做匹配用的accepts和emits定义了交接物的类型和格式这是标准化交接的基础model允许不同角色用不同模型服务比如需求分析用便宜快速的模型就够质量检查用能力更强的模型效果更好max_retries是执行单元的重试上限。新增一个Agent不需要改调度器代码只需要在注册中心登记一份这样的声明这是整个项目可扩展性的来源。为什么用YAML而不是在代码里写死因为能力声明本质上是一份接口文档。把接口从代码里抽出来意味着你可以用配置驱动的方式动态加入新角色。我在后来接入一个数据清洗Agent时就是靠加一份声明文件完成的没有动一行调度代码。这种配置优先的思路会省掉后期大量的改造成本。2.2 Job对象与状态机让任务干净地交到下一个人手里每次任务进来Dispatcher会生成一个Job对象。Job不只是消息它携带完整状态。{ job_id: job_7f3a, trace_id: trace_8e21, type: task_plan, state: pending, payload: {}, result: null, attempts: 0, assigned_to: null }state字段是我反复强调绝对不要省略的部分。它让每个任务在任意时刻都是可审计的pending表示等待派发running表示有Agent在处理done表示完成failed表示执行失败needs_review表示结果需要人工介入。显式状态机带来的好处是你可以随时回答这个任务现在到底怎么样了排错时一眼就能看清停在哪一步。很多多Agent项目做得乱就是因为任务在Agent之间传来传去却没有任何状态记录出了问题只能靠猜。我甚至要求所有环节都只能通过Job状态来判断下一步动作禁止任何Agent看到对方返回了就顺手改别人状态这种旁路操作。状态只属于Job本身这算是这个项目里最有用的一条纪律。2.3 任务路由与协作上下文Dispatcher怎么决定下一个环节Dispatcher在收到Agent完成事件后会读取Context Store里该Job的最新产物类型再根据一张简单的路由表决定下一步。ROUTE_TABLE { (requirements_analyst, done): solution_designer, (solution_designer, done): quality_checker, (quality_checker, done): final, (quality_checker, needs_review): human_gate, }路由表的key是上一个Agent加它的结束状态value是下一个处理者。这种写法直观且容易改你想在中间加一道人工确认就加一行路由你想让质量不合格的方案退回设计师而不是走向终稿就改一个映射。注意这里没有写任何业务逻辑只有谁完成后交给谁。业务判断发生在Agent内部流程控制发生在路由表里各管一段改起来互不干扰。3. 关键模块的代码实现与设计取舍架构想清楚之后写代码反而很快。核心类就两个Agent基类和Dispatcher调度器外加一个上下文存储模块。我把关键实现和取舍逻辑放在这一节。3.1 Agent基类封装调用、重试与错误隔离先看基类代码。我用Python实现核心思想是把调用大模型的通用流程沉淀在基类里子类只负责写提示词和结果解析。import json from abc import ABC, abstractmethod class BaseAgent(ABC): def __init__(self, config, context_store, llm_client): self.name config[name] self.config config self.context_store context_store def handle(self, job): state running attempts job.get(attempts, 0) try: payload self.context_store.read(job[job_id], in) prompt self.build_prompt(payload, job) result_raw self.llm_client.chat( messagesprompt, temperatureself.config[model][temperature], max_tokens1024, ) result self.parse_result(result_raw) self.context_store.write(job[job_id], out, result) state done except Exception as e: attempts 1 if attempts self.config.get(max_retries, 2): state failed self.context_store.write(job[job_id], error, str(e)) else: state pending finally: job[state] state job[attempts] attempts return job abstractmethod def build_prompt(self, payload, job): ... abstractmethod def parse_result(self, result_raw): ...基类里做了三件事读上下文、调模型、回写结果。比较重要的设计决策是把错误处理和状态流转都放在基类里。这样每个Agent子类只需要专注写自己的提示词和解析函数重试、异常、状态更新这类横切逻辑不会散落在各个Agent中。错误隔离也体现在这里一个Agent抛异常不会连累别的任务任务只会变成failed或者回到pending等待下一次重试。这里有一个容易踩的细节重试逻辑放进基类之后要小心无限重试问题。我在早期版本没有设置上限结果网络故障时一个Job能被重试几十次下游收到一堆重复结果。所以max_retries字段从一开始就写进能力声明把重试次数当成一种资源按需配置。3.2 调度器拉模型与推模型之间的折中调度器我试过两种思路都有明显缺点。推模型是Dispatcher直接调用Agent的handle方法代码简单直接但问题也很明显一个Agent处理慢会把整个调度线程卡住Agent内部的重试会让调度器被迫空等测试时想单独模拟某个Agent的行为必须把整条链都跑起来。拉模型是Agent处理完就主动向Dispatcher申请下一个任务好处是天然有背压——Agent忙就不会积压新任务系统不容易被打爆坏处是调度逻辑被拆散到各个Agent里协作链路不好观察出了问题更难定位。我最终采用的是折中的队列模型Dispatcher负责匹配任务并把Job投递到对应Agent的队列中Worker空闲时从自己的队列里拉取。调度策略集中在Dispatcher里执行节奏又掌握在每个Worker手里两头兼顾。投递动作是原子的同一个Job只会进入一条队列不会出现两个Worker同时抢任务的情况。import queue class Dispatcher: def __init__(self, agents, route_table): self.agents agents self.route_table route_table self.queues {name: queue.Queue() for name in agents} def submit(self, job): target self.agents.get(job.get(type)) if target: self.queues[target].put(job) def on_agent_done(self, job): key (job.get(assigned_to), job.get(state)) next_agent self.route_table.get(key) if next_agent: job[assigned_to] next_agent self.queues[next_agent].put(job) else: job[state] job.get(state, done)实际应用时我在on_agent_done里还加了路由日志每个Job每经过一次路由就追加一行从X到Y状态Z的记录。调试的时候把这一串日志拉出来基本就是整个协作链路的骨架执行过程一目了然。3.3 上下文存储的三个设计原则多Agent系统里最容易被低估的是上下文管理。踩了几次数据错乱的坑之后我定下三个原则。第一个原则是命名空间隔离。每个Job有自己的独立命名空间不存在全局共享变量Agent只能读写自己这个Job的数据碰不到别人的。第二个原则是结构化交接。Agent之间传递的是结构化字段比如{task_plan: {...}}不是一大段散文式对话历史散文没法被下一环节稳定解析字段可以。第三个原则是交接前压缩。Agent在回写结果时会生成一份摘要下个Agent优先读摘要而不是原文。class ContextStore: def __init__(self): self._data {} def write(self, job_id, key, value): self._data.setdefault(job_id, {})[key] value def read(self, job_id, key): return self._data.get(job_id, {}).get(key) def snapshot(self, job_id): return dict(self._data.get(job_id, {}))实现本身不复杂难的是坚持这几个原则。我见过很多多Agent实现上下文就是一个全局变量所有Agent往里塞对话历史最后模型被一坨互相矛盾的信息搞晕质量崩盘。宁可少传、传精华也不要多传、传垃圾——这句话我写在项目文档第一行。4. 跑通第一个多Agent协作Demo的完整过程讲完设计理念用一个实际Demo说明整个链路。目标让三个Agent协作完成把一段需求描述变成一份产品方案并完成自查。4.1 最小配置三个角色加一份任务清单我先定义了三个角色requirements_analyst负责读需求原文、输出结构化任务清单solution_designer根据任务清单输出产品方案quality_checker负责检查方案输出问题清单和修订建议。三个角色用YAML声明注册再启动Dispatcher。整个初始化流程就是读取配置、注册Agent、开启队列、等待任务提交四步。这一步我建议不要贪多三个Agent已经足够跑通机制角色越多越难排查。这次Demo提交的需求是一段真实场景描述做一个面向团队成员的周报汇总工具支持自动收集每个人提交的周报并按项目维度和时间维度生成汇总视图。要求操作简单数据权限按团队隔离。4.2 从提交任务到拿到成品的完整链路任务提交之后Dispatcher根据type字段把它路由给requirements_analyst。整个过程一共五步requirements_analyst收到raw_document输出结构化任务清单。Dispatcher看到结果是task_plan将其作为输入路由给solution_designer。solution_designer生成产品方案包含信息架构、核心页面、数据权限说明。Dispatcher把方案交给quality_checker做自检。quality_checker输出问题清单若问题严重则退回设计师修订一次否则输出终稿。我实测跑完这条链路产物质量比单Agent直接生成高出不少。尤其数据权限按团队隔离这条需求三个角色接力之后被完整保留而单Agent方案在后期经常把这类约束忘掉。这其实不是巧合每个角色只负责一个环节约束作为结构化字段被显式携带自然更不容易丢。4.3 调试多Agent系统的三板斧多Agent系统调试起来和单Agent完全不同。我摸索出三个办法。第一个是给每个任务分配trace_id全链路日志都带上它排查时一条命令就能拉出完整执行轨迹。第二个是上下文快照每完成一个环节把Context Store里该Job的命名空间整体导出人工检查中间产物是否合理比如看看任务清单有没有漏掉关键需求。第三个是重放把某一份快照喂给单个Agent单独验证它的处理逻辑避免每次都要跑完整链路才能复现问题。这三个办法配合Job状态机基本解决多Agent系统里九成的问题任务走到哪了这个Agent拿到了什么输入它输出了什么哪个环节出的错。建议你在自己的项目里尽早把这些观测能力做进去别等到出问题再补那是真的来不及。5. 实测中踩过的坑与优化经验这个项目跑了几周之后我从能跑进入跑得稳的阶段。这一阶段踩的坑最有价值下面三个问题值得展开讲。5.1 链路加深导致的质量滑坡与控制手段最开始我把协作链设计得很长需求分析、代码设计、测试用例、文档生成四五个环节接力。结果发现环节越多最终结果反而越差。原因不复杂每一跳的模型都在转述上一环节的内容转述过程中信息必然损耗环节越多损耗越大最后只剩下一些表面内容真正关键的细节反而丢了。我用的控制手段是交接物压缩和关键约束显式携带。每个Agent回写结果时除了结构化产物还必须把上游明确的关键约束写进constraints字段原样传给下一环节。这样一来中间步骤的信息损耗再大最核心的约束也不会丢。同时我把链路从五跳压成三跳能合并的环节直接合并牺牲一点角色的单一职责换回整体质量的大幅提升。5.2 上下文膨胀与遗忘之间的平衡第二个坑正好和第一个相反。有时候为了让信息不丢我让Context Store把上游所有内容都传给下一个Agent结果上下文飞速膨胀。一个Job处理到第三环节时传给模型的token量已经是初始的几倍响应变慢模型对关键信息的注意力反而被稀释。信息不丢和信息可用完全是两回事。我的解决方案是分级记忆Agent的输入分成必读字段和参考字段。必读字段是结构化关键信息直接进Prompt参考字段可能很大只给摘要Agent需要时再按需读取。这也逼着我在设计交接物时更克制——真正需要传给下一环的通常就那么几个字段其余都是噪音。配置好必读和参考分类之后上下文体积基本稳定在初始的一倍多一点质量也没再因为信息太多而下滑。5.3 死循环、重复执行与并发风暴多Agent系统跑起来之后还会出现三类失控问题我每个都遇到过至少一次。死循环某Agent发现结果不合格把Job退回上一环节上一环节重新处理后结果还是不合格又退回来循环往复。我加了最大回流次数限制默认1次超过就把Job标记成needs_review转人工处理。退回机制一定要画上限否则模型偶尔的坚持己见就会让整条链卡住。重复执行同一个Job因为网络超时被重试结果下游收到两份几乎一样的产物。解决方式是给Job加幂等键下游在写Context Store之前先检查同样的job_id加产物类型是否已经存在。宁可丢弃一个可疑请求也不要让重复产物污染下游。并发风暴多个Worker同时向同一个模型服务发起高并发请求触发限流反而拖慢所有任务。我在每个Worker外加了信号量限制同时进行的任务数后来又加了一个简单的熔断器连续失败超过阈值就暂停新任务。别看这两个机制简单它们能保证系统在模型服务不稳定时不会自己把自己拖死。6. 项目后续还能怎么扩展把基础链路做稳之后我在几个方向上做了扩展效果都不错列出来供你参考。6.1 人工介入节点让系统在该停的时候停下性价比最高的扩展是加人工介入节点。我在Dispatcher里加了一类gate节点任务流到某个环节时先挂起等人工确认再继续。比如quality_checker给出needs_review时系统会带着问题清单请求人工裁决而不是闷头自己改。多Agent系统最怕的不是出错而是信心满满地错。人工闸门就是在关键分歧点上叫停加了这个之后长链路任务的最终质量明显提升。6.2 成本追踪与角色-模型解耦第二个值得做的扩展是成本追踪。每个Agent调用模型时记录token数和耗时汇总到每个Job的成本账单里。有了账单你能直观发现哪个环节最贵、哪次跑链路是无谓浪费。再往后我给不同角色配置了不同的模型服务需求分析用便宜快速的模型就够质量检查用能力更强的模型效果更好。这其实是能力声明里model字段带来的自由度——角色和模型解耦按需分配性价比。6.3 给新手的最后几条建议如果让我给打算动手的人几句实在话别一上来铺开十个Agent先用两三个角色把一条最关键的链路跑通把状态机、上下文交接、质量检查这三件事做扎实再谈扩展。我在实际使用中发现多数场景下三到五个角色已经能覆盖大部分需求多Agent的价值从来不是数量而是清晰的协作结构和可控的交接质量。对于一个Agent就能完成的任务不要硬拆只有当单一角色确实顾不过来时分工才有意义。
去年我在做内部自动化工具的时候第一次认真思考agency-agents该怎么落地。当时的需求并不复杂给出一份杂乱的需求文档让大模型把它拆成可执行的开发任务再针对每个任务写出实现说明最后整体自检一遍。听起来很常规可单Agent模式跑了不到两周问题就集中爆发了。后来我干脆把项目名改成agency-agents——一个像虚拟团队一样工作的多Agent协作层。这篇文章就把它的设计思路、核心代码和实测踩坑一次性讲清楚。这个项目适合三类人正在做AI应用开发、想尝试多Agent协作但不知道怎么组织的朋友手里已经接了多个模型服务、想让不同角色各司其职的开发者以及单Agent方案反复撞墙、想换一种协作方式的人。就算你暂时不写代码光看架构思路也会对多Agent到底怎么协作有一个比很多教程都更清楚的理解。1. 为什么我会把多个Agent组织成一家虚拟公司1.1 单Agent模式在复杂任务面前的四个信号第一个信号是角色混乱。让模型既当产品经理、又当架构师、还当测试它会在同一轮回答里一会儿改需求、一会儿写代码、一会儿给自己挑毛病最后产出一锅乱炖。第二个信号是上下文稀释任务越往后最开始定义的目标和约束条件越容易被遗忘我曾在第三轮之后反复看到它把数据权限按团队隔离这种硬需求漏掉。第三个信号是错误难定位。一段错误的输出到底是在哪个环节被引入的完全说不清只能把整段历史翻出来人工核对。第四个信号是token浪费——每轮都要把全部历史对话重传一遍成本随着轮次线性上涨效果却逐步下降。这四个信号背后其实是同一个原因单个模型实例在同一个上下文里连续扮演多个角色超过了它有效注意力能够覆盖的边界。打个比方让一个人同时当项目经理、文案、财务和客服忙起来必然顾此失彼但如果你让四个人分工每个人只需要记住自己那一亩三分地整体反而更稳。我就是在这个对比里决定把架构改成多角色协作的。1.2 agency模式借鉴的是现实组织的分工逻辑agency这个词的灵感来自现实中的广告公司和咨询公司。它们很少让一个人从头干到尾客户经理负责对接需求策略团队出方案创意团队做执行最后还有一层质检。员工各管一摊、交接标准明确、有人对总结果负责。这套逻辑搬到Agent上就是三件事角色边界清楚、交接物标准化、调度与质检集中。所以我把这个项目定位成一层编排层不是一个大模型应用。它不关心你接的是哪个商用模型接口也不替Agent写业务知识它只负责几件小事定义角色、派发任务、传递上下文、回收结果。项目因此很轻核心代码没有动辄几千行却把多个Agent怎么协作这个最常见的问题管住了。在动手前我也想了很多市面上已经有那么多Agent框架自己再造一遍轮子有意义吗答案是看你想解决的问题是什么。如果你只想快速搭一个能对话的Agent现有框架够用但如果你的痛点是多个角色如何稳定交接、任务状态如何可审计、模型服务如何按角色分配那你就需要这样一层轻量编排逻辑。我的判断是永远别让框架替你决定协作模式你自己的业务才该决定协作模式。2. agency-agents的核心架构注册中心、任务调度与协作上下文开始写代码之前我先把整个系统的运作模式拆开——不是画漂亮的流程图而是明确谁手里有什么数据、谁对什么负责。最终沉淀下来四个模块下面这张表是我在项目文档里反复改了四五版才定下来的。模块职责关键产出注册中心 Registry登记所有Agent的身份、能力、输入输出声明能力索引调度器 Dispatcher把任务匹配到合适的Agent驱动整条协作链任务分派记录上下文存储 Context Store保存任务中间结果、角色之间的交接物结构化上下文Agent Worker真正调用大模型接口的执行单元结构化结果Registry解决有哪些角色、各自能干什么Dispatcher解决某件事该找谁Context Store解决上个角色说了什么、下个角色该知道什么Worker解决最终调用模型干活。四者缺一不可。最容易被忽略的是Context Store——很多人以为多Agent就是多个模型互相传消息忽略了上下文本身需要统一管理结果协作起来到处丢信息。2.1 Agent的能力声明让调度器看懂每个角色为了让Dispatcher不写死逻辑每个Agent在上线前都要提交一份能力声明。我用的格式是YAML既好读又方便直接做成配置文件。name: requirements_analyst role: 需求分析员 description: 将杂乱需求文档整理成结构化任务清单 capabilities: - requirement_analysis - task_decomposition accepts: type: raw_document max_size: 20000 emits: type: task_plan schema: task_plan_v1 model: provider: generic_commercial_llm temperature: 0.2 max_retries: 2这份声明里几个字段很关键。capabilities是给调度器做匹配用的accepts和emits定义了交接物的类型和格式这是标准化交接的基础model允许不同角色用不同模型服务比如需求分析用便宜快速的模型就够质量检查用能力更强的模型效果更好max_retries是执行单元的重试上限。新增一个Agent不需要改调度器代码只需要在注册中心登记一份这样的声明这是整个项目可扩展性的来源。为什么用YAML而不是在代码里写死因为能力声明本质上是一份接口文档。把接口从代码里抽出来意味着你可以用配置驱动的方式动态加入新角色。我在后来接入一个数据清洗Agent时就是靠加一份声明文件完成的没有动一行调度代码。这种配置优先的思路会省掉后期大量的改造成本。2.2 Job对象与状态机让任务干净地交到下一个人手里每次任务进来Dispatcher会生成一个Job对象。Job不只是消息它携带完整状态。{ job_id: job_7f3a, trace_id: trace_8e21, type: task_plan, state: pending, payload: {}, result: null, attempts: 0, assigned_to: null }state字段是我反复强调绝对不要省略的部分。它让每个任务在任意时刻都是可审计的pending表示等待派发running表示有Agent在处理done表示完成failed表示执行失败needs_review表示结果需要人工介入。显式状态机带来的好处是你可以随时回答这个任务现在到底怎么样了排错时一眼就能看清停在哪一步。很多多Agent项目做得乱就是因为任务在Agent之间传来传去却没有任何状态记录出了问题只能靠猜。我甚至要求所有环节都只能通过Job状态来判断下一步动作禁止任何Agent看到对方返回了就顺手改别人状态这种旁路操作。状态只属于Job本身这算是这个项目里最有用的一条纪律。2.3 任务路由与协作上下文Dispatcher怎么决定下一个环节Dispatcher在收到Agent完成事件后会读取Context Store里该Job的最新产物类型再根据一张简单的路由表决定下一步。ROUTE_TABLE { (requirements_analyst, done): solution_designer, (solution_designer, done): quality_checker, (quality_checker, done): final, (quality_checker, needs_review): human_gate, }路由表的key是上一个Agent加它的结束状态value是下一个处理者。这种写法直观且容易改你想在中间加一道人工确认就加一行路由你想让质量不合格的方案退回设计师而不是走向终稿就改一个映射。注意这里没有写任何业务逻辑只有谁完成后交给谁。业务判断发生在Agent内部流程控制发生在路由表里各管一段改起来互不干扰。3. 关键模块的代码实现与设计取舍架构想清楚之后写代码反而很快。核心类就两个Agent基类和Dispatcher调度器外加一个上下文存储模块。我把关键实现和取舍逻辑放在这一节。3.1 Agent基类封装调用、重试与错误隔离先看基类代码。我用Python实现核心思想是把调用大模型的通用流程沉淀在基类里子类只负责写提示词和结果解析。import json from abc import ABC, abstractmethod class BaseAgent(ABC): def __init__(self, config, context_store, llm_client): self.name config[name] self.config config self.context_store context_store def handle(self, job): state running attempts job.get(attempts, 0) try: payload self.context_store.read(job[job_id], in) prompt self.build_prompt(payload, job) result_raw self.llm_client.chat( messagesprompt, temperatureself.config[model][temperature], max_tokens1024, ) result self.parse_result(result_raw) self.context_store.write(job[job_id], out, result) state done except Exception as e: attempts 1 if attempts self.config.get(max_retries, 2): state failed self.context_store.write(job[job_id], error, str(e)) else: state pending finally: job[state] state job[attempts] attempts return job abstractmethod def build_prompt(self, payload, job): ... abstractmethod def parse_result(self, result_raw): ...基类里做了三件事读上下文、调模型、回写结果。比较重要的设计决策是把错误处理和状态流转都放在基类里。这样每个Agent子类只需要专注写自己的提示词和解析函数重试、异常、状态更新这类横切逻辑不会散落在各个Agent中。错误隔离也体现在这里一个Agent抛异常不会连累别的任务任务只会变成failed或者回到pending等待下一次重试。这里有一个容易踩的细节重试逻辑放进基类之后要小心无限重试问题。我在早期版本没有设置上限结果网络故障时一个Job能被重试几十次下游收到一堆重复结果。所以max_retries字段从一开始就写进能力声明把重试次数当成一种资源按需配置。3.2 调度器拉模型与推模型之间的折中调度器我试过两种思路都有明显缺点。推模型是Dispatcher直接调用Agent的handle方法代码简单直接但问题也很明显一个Agent处理慢会把整个调度线程卡住Agent内部的重试会让调度器被迫空等测试时想单独模拟某个Agent的行为必须把整条链都跑起来。拉模型是Agent处理完就主动向Dispatcher申请下一个任务好处是天然有背压——Agent忙就不会积压新任务系统不容易被打爆坏处是调度逻辑被拆散到各个Agent里协作链路不好观察出了问题更难定位。我最终采用的是折中的队列模型Dispatcher负责匹配任务并把Job投递到对应Agent的队列中Worker空闲时从自己的队列里拉取。调度策略集中在Dispatcher里执行节奏又掌握在每个Worker手里两头兼顾。投递动作是原子的同一个Job只会进入一条队列不会出现两个Worker同时抢任务的情况。import queue class Dispatcher: def __init__(self, agents, route_table): self.agents agents self.route_table route_table self.queues {name: queue.Queue() for name in agents} def submit(self, job): target self.agents.get(job.get(type)) if target: self.queues[target].put(job) def on_agent_done(self, job): key (job.get(assigned_to), job.get(state)) next_agent self.route_table.get(key) if next_agent: job[assigned_to] next_agent self.queues[next_agent].put(job) else: job[state] job.get(state, done)实际应用时我在on_agent_done里还加了路由日志每个Job每经过一次路由就追加一行从X到Y状态Z的记录。调试的时候把这一串日志拉出来基本就是整个协作链路的骨架执行过程一目了然。3.3 上下文存储的三个设计原则多Agent系统里最容易被低估的是上下文管理。踩了几次数据错乱的坑之后我定下三个原则。第一个原则是命名空间隔离。每个Job有自己的独立命名空间不存在全局共享变量Agent只能读写自己这个Job的数据碰不到别人的。第二个原则是结构化交接。Agent之间传递的是结构化字段比如{task_plan: {...}}不是一大段散文式对话历史散文没法被下一环节稳定解析字段可以。第三个原则是交接前压缩。Agent在回写结果时会生成一份摘要下个Agent优先读摘要而不是原文。class ContextStore: def __init__(self): self._data {} def write(self, job_id, key, value): self._data.setdefault(job_id, {})[key] value def read(self, job_id, key): return self._data.get(job_id, {}).get(key) def snapshot(self, job_id): return dict(self._data.get(job_id, {}))实现本身不复杂难的是坚持这几个原则。我见过很多多Agent实现上下文就是一个全局变量所有Agent往里塞对话历史最后模型被一坨互相矛盾的信息搞晕质量崩盘。宁可少传、传精华也不要多传、传垃圾——这句话我写在项目文档第一行。4. 跑通第一个多Agent协作Demo的完整过程讲完设计理念用一个实际Demo说明整个链路。目标让三个Agent协作完成把一段需求描述变成一份产品方案并完成自查。4.1 最小配置三个角色加一份任务清单我先定义了三个角色requirements_analyst负责读需求原文、输出结构化任务清单solution_designer根据任务清单输出产品方案quality_checker负责检查方案输出问题清单和修订建议。三个角色用YAML声明注册再启动Dispatcher。整个初始化流程就是读取配置、注册Agent、开启队列、等待任务提交四步。这一步我建议不要贪多三个Agent已经足够跑通机制角色越多越难排查。这次Demo提交的需求是一段真实场景描述做一个面向团队成员的周报汇总工具支持自动收集每个人提交的周报并按项目维度和时间维度生成汇总视图。要求操作简单数据权限按团队隔离。4.2 从提交任务到拿到成品的完整链路任务提交之后Dispatcher根据type字段把它路由给requirements_analyst。整个过程一共五步requirements_analyst收到raw_document输出结构化任务清单。Dispatcher看到结果是task_plan将其作为输入路由给solution_designer。solution_designer生成产品方案包含信息架构、核心页面、数据权限说明。Dispatcher把方案交给quality_checker做自检。quality_checker输出问题清单若问题严重则退回设计师修订一次否则输出终稿。我实测跑完这条链路产物质量比单Agent直接生成高出不少。尤其数据权限按团队隔离这条需求三个角色接力之后被完整保留而单Agent方案在后期经常把这类约束忘掉。这其实不是巧合每个角色只负责一个环节约束作为结构化字段被显式携带自然更不容易丢。4.3 调试多Agent系统的三板斧多Agent系统调试起来和单Agent完全不同。我摸索出三个办法。第一个是给每个任务分配trace_id全链路日志都带上它排查时一条命令就能拉出完整执行轨迹。第二个是上下文快照每完成一个环节把Context Store里该Job的命名空间整体导出人工检查中间产物是否合理比如看看任务清单有没有漏掉关键需求。第三个是重放把某一份快照喂给单个Agent单独验证它的处理逻辑避免每次都要跑完整链路才能复现问题。这三个办法配合Job状态机基本解决多Agent系统里九成的问题任务走到哪了这个Agent拿到了什么输入它输出了什么哪个环节出的错。建议你在自己的项目里尽早把这些观测能力做进去别等到出问题再补那是真的来不及。5. 实测中踩过的坑与优化经验这个项目跑了几周之后我从能跑进入跑得稳的阶段。这一阶段踩的坑最有价值下面三个问题值得展开讲。5.1 链路加深导致的质量滑坡与控制手段最开始我把协作链设计得很长需求分析、代码设计、测试用例、文档生成四五个环节接力。结果发现环节越多最终结果反而越差。原因不复杂每一跳的模型都在转述上一环节的内容转述过程中信息必然损耗环节越多损耗越大最后只剩下一些表面内容真正关键的细节反而丢了。我用的控制手段是交接物压缩和关键约束显式携带。每个Agent回写结果时除了结构化产物还必须把上游明确的关键约束写进constraints字段原样传给下一环节。这样一来中间步骤的信息损耗再大最核心的约束也不会丢。同时我把链路从五跳压成三跳能合并的环节直接合并牺牲一点角色的单一职责换回整体质量的大幅提升。5.2 上下文膨胀与遗忘之间的平衡第二个坑正好和第一个相反。有时候为了让信息不丢我让Context Store把上游所有内容都传给下一个Agent结果上下文飞速膨胀。一个Job处理到第三环节时传给模型的token量已经是初始的几倍响应变慢模型对关键信息的注意力反而被稀释。信息不丢和信息可用完全是两回事。我的解决方案是分级记忆Agent的输入分成必读字段和参考字段。必读字段是结构化关键信息直接进Prompt参考字段可能很大只给摘要Agent需要时再按需读取。这也逼着我在设计交接物时更克制——真正需要传给下一环的通常就那么几个字段其余都是噪音。配置好必读和参考分类之后上下文体积基本稳定在初始的一倍多一点质量也没再因为信息太多而下滑。5.3 死循环、重复执行与并发风暴多Agent系统跑起来之后还会出现三类失控问题我每个都遇到过至少一次。死循环某Agent发现结果不合格把Job退回上一环节上一环节重新处理后结果还是不合格又退回来循环往复。我加了最大回流次数限制默认1次超过就把Job标记成needs_review转人工处理。退回机制一定要画上限否则模型偶尔的坚持己见就会让整条链卡住。重复执行同一个Job因为网络超时被重试结果下游收到两份几乎一样的产物。解决方式是给Job加幂等键下游在写Context Store之前先检查同样的job_id加产物类型是否已经存在。宁可丢弃一个可疑请求也不要让重复产物污染下游。并发风暴多个Worker同时向同一个模型服务发起高并发请求触发限流反而拖慢所有任务。我在每个Worker外加了信号量限制同时进行的任务数后来又加了一个简单的熔断器连续失败超过阈值就暂停新任务。别看这两个机制简单它们能保证系统在模型服务不稳定时不会自己把自己拖死。6. 项目后续还能怎么扩展把基础链路做稳之后我在几个方向上做了扩展效果都不错列出来供你参考。6.1 人工介入节点让系统在该停的时候停下性价比最高的扩展是加人工介入节点。我在Dispatcher里加了一类gate节点任务流到某个环节时先挂起等人工确认再继续。比如quality_checker给出needs_review时系统会带着问题清单请求人工裁决而不是闷头自己改。多Agent系统最怕的不是出错而是信心满满地错。人工闸门就是在关键分歧点上叫停加了这个之后长链路任务的最终质量明显提升。6.2 成本追踪与角色-模型解耦第二个值得做的扩展是成本追踪。每个Agent调用模型时记录token数和耗时汇总到每个Job的成本账单里。有了账单你能直观发现哪个环节最贵、哪次跑链路是无谓浪费。再往后我给不同角色配置了不同的模型服务需求分析用便宜快速的模型就够质量检查用能力更强的模型效果更好。这其实是能力声明里model字段带来的自由度——角色和模型解耦按需分配性价比。6.3 给新手的最后几条建议如果让我给打算动手的人几句实在话别一上来铺开十个Agent先用两三个角色把一条最关键的链路跑通把状态机、上下文交接、质量检查这三件事做扎实再谈扩展。我在实际使用中发现多数场景下三到五个角色已经能覆盖大部分需求多Agent的价值从来不是数量而是清晰的协作结构和可控的交接质量。对于一个Agent就能完成的任务不要硬拆只有当单一角色确实顾不过来时分工才有意义。
