如果你最近在折腾大模型应用应该对“多代理编排”这个词不陌生。我去年花了不少时间打磨一个内部工具项目代号就叫agency-agents目的很简单把单一智能体干不好的复杂任务拆给一组各司其职的代理去协作完成。实测下来同样的任务从“勉强能用”变成了“稳定可交付”但也踩了一堆文档里根本不会写的坑。这篇就把我的设计思路、核心代码结构、实测对比和排障过程完整拆开讲适合正在做 LLM 应用、想落地多代理方案的开发者参考。先说结论多代理不是银弹但当你遇到“上下文塞爆、角色冲突、工具链不稳定”这三件事同时发生时它可能是性价比最高的解法。下面我从问题出发一步步讲清楚 agency-agents 到底是怎么工作的。1. 单代理模型的上限一次复杂任务把我逼到墙角1.1 上下文被“角色技能”撑爆回答开始自相矛盾我最开始做的是一个竞品分析工具输入一份产品说明让它输出市场洞察、竞品对比、数据库表结构建议三份结果。用单个智能体跑的时候我把“产品专家”“数据分析师”“后端架构师”三个角色全塞进一条系统提示词里再加上工具调用说明、输出格式约束、历史对话示例上下文瞬间堆到接近窗口上限。结果很尴尬让它先做市场分析它会在中间突然开始画 ER 图让它写数据库表结构它又回过头来补充竞品定价策略。原因不难理解——单个模型在一段上下文里同时扮演多个角色时不同角色的指令会互相干扰越到后面越容易“串味”。我试过把提示词拆成多个版本也试过用 temperature 调低减少随机性但本质问题在于所有角色的所有指令都写在同一个上下文里模型没有物理边界去隔离它们。后来我把这三个角色拆成三个独立代理每个代理只保留自己的角色说明、工具集和输出模板上下文从几千 token 骤降到几百 token。效果立竿见影但精力也从“写提示词”转移到了“写代理之间的通信协议”。1.2 工具调用链越长失败概率越是成指数上涨另一个痛点是工具调用的稳定性。单代理做数据分析时典型链路是读取文件 → 清洗字段 → 计算聚合值 → 生成图表 → 写结论。每一步都要让模型自己决定调用哪个工具、传什么参数只要中间有一次参数格式解析错误整条链就崩了而且经常是在最后一步才暴露问题。我统计过一份 50 次运行的日志单代理模式下平均 8 次工具调用才能完成一次任务其中至少 2 次会因为 JSON 参数转义错误或字段名拼错而重试。越长的链路重试成本越高——每次重试要把前面几步的结果重新塞回上下文上下文又变长模型更糊涂恶性循环。agency-agents 的思路是把长链路拆成短链路每个执行代理只负责一两个工具调用比如“数据清洗代理”只做格式转换“图表生成代理”只调绘图接口。单条链路短了失败概率自然降下来即使失败重试范围也限定在一个代理内部不会影响其他环节的已完成结果。1.3 串行执行的时间账一个任务能把一整天耗尽单代理还有一个隐藏成本几乎所有步骤都是串行的。读取文件、分析、写报告、校对每一步都要等模型完整输出后才能进入下一步。遇到一次“想太多”的输出光等待响应就能拖几分钟。我在实测里遇到过单代理跑一个 500 行日志的分析任务从开始到出结果花了 25 分钟其中大部分时间消耗在长上下文的反复处理和无关输出上。拆成多代理之后调研类和计算类任务可以并行执行整体耗时基本等于最慢的那个代理耗时。还是那个日志分析任务拆成“异常提取”“异常分类”“报告生成”三个代理前两个并行跑最后汇总总耗时从 25 分钟压到了 7 分钟。2. 设计一个轻量“代理机构”调度、执行、黑板三层分工2.1 调度代理的职责拆任务、派单、验收结果把单代理改成多代理本质上是在模仿一个真实团队有人拆活、有人干活、有人对结果负责。agency-agents 里对应着三类角色——调度代理、执行代理、共享黑板。调度代理相当于项目经理负责把用户请求拆解成可执行的子任务然后按照每个执行代理的能力声明去派单。这里有一个关键设计调度代理不直接执行任何业务逻辑它只做三件事。第一把原始需求拆成结构化的子任务列表每个子任务带明确的目标、输入参数和期望输出第二根据执行代理的能力注册信息选一个或者多个合适的代理去接单第三接收返回结果检查是否符合协议必要时触发重试。我踩过的坑是让调度代理“顺便”干点活比如让它也参与数据分析。一旦调度层开始执行具体任务它的上下文立刻被大量中间信息占满后续拆解任务时就会出现记忆偏差导致漏派单。所以我的原则是调度层必须保持上下文干净它只读任务清单和结果摘要不看具体执行过程。2.2 执行代理的能力注册表没有登记的能力不接单每个执行代理都有一个能力注册信息格式很简单就是一个包含name、description、input_schema、output_schema的字典。调度代理通过比较任务需求与description的语义相似度来选人而不是硬编码代理 ID。这样做的好处是新增代理只需要注册一下调度层不用改代码。比如我的项目里有这三个执行代理代理名能力描述输入要求输出要求数据清洗器处理 CSV/JSON 格式转换、字段重命名、缺失值填充原始文件路径 清洗规则清洗后的结构化数据异常分类器从日志文本中提取异常事件并按类型分组日志文本 分类维度JSON 数组每组含严重级别文档质检员检查文档是否存在前后矛盾、错别字、格式问题待检文档 检查清单问题列表 修改建议注册表的作用不只是“选人”它还承担了任务边界的功能。调度代理看到能力描述里写了“输入要求”会主动把任务参数规范化看到“输出要求”就知道该以什么格式验收结果。如果某个任务没有匹配到任何执行代理调度代理直接返回“能力不足”的提示而不是硬着头皮让某个代理乱做。2.3 共享黑板任务状态和结果摘要的同步机制多代理协作最头疼的问题是状态同步A 代理跑完了B 代理需要它的结果但没有一个统一的地方存放中间产物。agency-agents 里我引入了一个中心化的“黑板”blackboard组件本质上是一个支持 JSON 读写的内存存储。黑板里主要存三类内容任务状态、结果摘要、冲突标记。任务状态标记每个子任务是“待派单”“执行中”“已完成”“失败”结果摘要只保留每个代理返回的关键字段不保留原始长文本冲突标记用于标记两个代理对同一数据的结论不一致后续由调度层决定用谁的。为什么只放摘要不放全文因为共享区域一旦堆满原始信息模型读取时的注意力会被分散我实测下来黑板每条记录超过 500 token 后调度代理的决策质量明显下降。3. 核心代码落地注册中心、任务协议与结果回收3.1 能力注册与检索让“谁擅长干什么”变成可查询数据先放一段简化版的注册中心实现我用 Python 写核心就是注册、检索两步。注册时把代理实例和它的能力 schema 绑定检索时用向量相似度匹配任务描述和能力描述。from typing import Dict, Any, List import json import numpy as np from sentence_transformers import SentenceTransformer class AgentRegistry: def __init__(self): self._agents {} self._encoder SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) def register(self, agent_id: str, capability: Dict[str, Any]) - None: cap_text f{capability[name]} {capability[description]} self._agents[agent_id] { meta: capability, embedding: self._encoder.encode(cap_text), } def match(self, task_description: str, top_k: int 1) - List[str]: task_emb self._encoder.encode(task_description) scores { aid: np.dot(task_emb, meta[embedding]) for aid, meta in self._agents.items() } ranked sorted(scores.items(), keylambda x: x[1], reverseTrue) return [aid for aid, _ in ranked[:top_k]]你可能会问为什么不直接用关键词匹配。我试过但真实任务描述里很少包含“清洗”“分类”这种明确的动词更多是“帮我把这批数据整理一下”“看看日志里有什么异常”这类自然语言。向量匹配对这种模糊表达更友好而且注册时把name和description拼在一起编码相当于把简称和长描述都纳入检索范围。3.2 任务分发协议一份带目标与约束的临时合同调度代理拆解完任务后会生成一个 JSON 任务包格式大致如下。这个协议我迭代了三个版本最终确定用“目标 输入 约束 期望输出”四段式缺一不可。{ task_id: ab34f2, type: execute, target: 对 2024Q4 日志做异常分类, input: { log_path: /data/logs/2024q4.txt, group_by: error_type }, constraints: { timeout_seconds: 120, max_retries: 1 }, expected_output: { format: json, schema: { type: array, items: { error_type: string, count: int, samples: [string] } } } }这里有一个很容易忽略的点约束条件里的 timeout 和 max_retries 是必填的。如果哪天某个执行代理因为外部接口响应缓慢而卡住没有超时机制整个任务会无限等待。我在早期版本里就没这个东西一次线上事故直接让整个流水线挂了两小时。后来把所有代理任务都强制加上超时上限超时后调度代理自动标记失败并重派给其他可用代理。3.3 执行代理的消息循环从接单到交付的完整动作执行代理的内部实现是一个消息循环接收调度代理发来的任务包执行自己的工具函数最后返回标准化结果。核心代码逻辑大约长这样class ExecutorAgent: def __init__(self, agent_id: str, executor_fn, registry: AgentRegistry): self.agent_id agent_id self.executor_fn executor_fn registry.register(agent_id, executor_fn.capability) def handle(self, task: Dict[str, Any]) - Dict[str, Any]: try: # 1. 校验输入是否符合声明的 schema self._validate_input(task[input]) # 2. 调用真实的业务函数 result self.executor_fn(**task[input]) # 3. 返回标准化封装 return { task_id: task[task_id], agent_id: self.agent_id, status: success, data: result, summary: self._summarize(result), } except Exception as e: return { task_id: task[task_id], agent_id: self.agent_id, status: failed, error: str(e), }关键在于summary字段。我不会把完整结果直接返回给调度代理而是先在执行代理内部生成一段不超过 200 token 的摘要。调度代理做后续决策时只读摘要只有需要最终交付物时才去取完整数据。这个设计源于实测中的教训多个代理的完整结果堆积在调度上下文里会严重挤压模型注意力。上下文是最贵的资源能少放就少放。3.4 调度代理的验收逻辑校验结构、裁剪噪声、触发重试调度代理拿到执行代理的返回结果后不是直接扔给用户而是先做三层校验。第一层校验status不是success就进入重试逻辑第二层校验data是否符合expected_output.schema不符合则让同一代理重新生成一次第三层校验summary是否包含关键字段防止代理输出一堆空话。def validate_and_retry(task, result, registry, max_retries2): if result[status] ! success: return dispatch_with_fallback(task, registry) if not conforms_to_schema(result[data], task[expected_output][schema]): for attempt in range(max_retries): result re_dispatch(task) if conforms_to_schema(result[data], task[expected_output][schema]): break else: return {status: failed, reason: schema_validation_failed} return result为什么重试要让同一个代理重做而不是换个代理因为大多数结构错误是模型输出不稳定导致的同一代理的领域知识最匹配重试成功率很高。只有连续失败两次才换另一个候选代理避免因为某个代理对数据格式有系统性的认知偏差。4. 三组实测数据与四个排障链路跑通之后才是麻烦的开始4.1 实测场景竞品调研、日志异常分类、文档质检光说架构可能感受不到差异我列出三组实际跑过的任务分别对比单代理和 agency-agents 的表现。每组跑 10 次记录成功率、平均耗时和上下文消耗。任务单代理成功率agency-agents 成功率单代理耗时agency-agents 耗时单代理峰值上下文agency-agents 峰值上下文竞品调研 数据库表结构建议4/109/1018.5 分钟6.2 分钟42k token11k token1000 行日志异常分类6/1010/1025 分钟7 分钟38k token9k token12 页产品文档质检5/108/1032 分钟14 分钟51k token13k token成功率的提升主要来源于上下文隔离和短工具链。上下文从 40k 降到 10k 级别后模型产生幻觉和角色冲突的概率明显下降。但第二个任务里出现过两次奇怪的结局就是通过下面的排障链路解决掉的。4.2 排查一执行代理互相刷屏终止信号迟迟不来第一个诡异现象是日志异常分类完成后负责汇总的结果生成代理居然还在给数据清洗代理发“结果确认”消息而且往返了十几轮。调度代理一直没叫停整个流程陷入死循环。排查链路是这样的先在黑板里打印每个代理的消息记录发现结果生成代理的每条消息都带着同一个request_id这不是任务派发 ID而是它自己在内部生成的一个确认码。再往深挖是结果生成代理的系统提示词里写了“持续向数据清洗代理确认字段定义直到对方确认无误为止”模型把“确认”理解为“必须反复询问”而不是“确认一次即可”。修复方案是在该代理的提示词里增加明确的终止条件“如果收到对方包含{ack: true}的响应立即停止发送新消息。”同时在整个消息循环外层加一个最大轮次max_rounds5超过即强制中断并发出告警。这类问题不在模型能力上而在指令歧义上模型只是严格遵循了你的字面要求。4.3 排查二任务漂移执行代理私改目标第二个问题是竞品调研任务里负责数据采集的执行代理在跑完后文本摘要里出现了“建议增加用户访谈”这类不属于它职责的内容。调度代理看到摘要后竟然真的把“用户访谈”加进了后续任务清单导致整个报告方向跑偏。根因是数据采集代理的输出模板里有一个“补充建议”字段模型觉得有义务提出额外建议。而调度代理验收摘要时只看字段名不校验“建议”字段是否在预期 schema 里。修复分两步一是从输出模板中移除所有非 schema 内的自由文本字段二是调度验收逻辑增加“如果摘要字段不在 expected_output 中则丢弃该字段并记录警告”。这里我想提醒的是多代理系统里任何自由文本都是潜在的任务漂移源。一旦某个代理输出里有不受约束的开放式内容调度代理很容易被带偏。如果确实需要代理提建议应该在 schema 里单独定义suggestions字段并规定最多三条而不是让模型随意发挥。4.4 排查三上下文膨胀把底层模型直接拖垮第三个问题出现在文档质检任务中。调度代理需要汇总三个执行代理的检查结果每个结果完整版都有一万字符左右虽然黑板里存的是摘要但调度代理在“汇总”阶段被要求读取完整结果来交叉验证结果上下文峰值直接飙到接近模型上限生成速度肉眼可见地变慢。我看了一眼日志发现是汇总阶段的提示词里写了“请结合所有原始检查结果进行交叉验证”模型于是把完整字段全拉进来了。这不是黑板设计的问题而是提示词引导的问题。改成“请只使用黑板内 summary 字段进行交叉验证完整结果仅在确认为最终报告时拉取”之后上下文峰值从 61k 掉到 24k。多代理系统里不是所有步骤都需要完整数据能让我访问完整数据的通道不等于代理必须使用完整数据。4.5 排查四结果格式不匹配调度层签收失败最后一个问题比较隐蔽。某个执行代理返回的error_type字段有时候是NullPointerException有时候是{ type: NullPointerException }格式不一致。conforms_to_schema校验器要求字段类型为string遇到第二种情况就报 schema 错误触发重试。但重试后模型又可能返回格式一来回横跳。排查发现是执行代理内部用了两个不同的代码路径组装结果一条路径直接返回error_type字符串另一条路径先组装 dict 再序列化。修复办法是在执行代理的executor_fn里做了统一的数据规范化无论内部逻辑怎么走最终返回前都强制调用一次normalize_result()。这种问题不是模型输出的锅是工程实现里数据来源不统一。5. 自研还是上框架我按这四个标准做的选型5.1 自研的优势只保留最小协议阈值掌握在自己手里市面上的多代理框架很多各有各的抽象。我选择自研而不是直接套框架主要是四个原因。一是协议可控任务分发格式、回归校验、超时策略都是自己定义的出了问题知道去哪找。二是依赖极简生产环境只依赖一个向量模型做能力检索其余全部是 Python 标准库。三是性能可控框架层越薄单次调用的额外延迟越少我实测某些框架光代理间通信就要损耗 300 毫秒以上自研后压到 80 毫秒以内。四是调试点位清晰黑板上直接可见所有任务状态和摘要排查问题时没有黑盒层。5.2 现成框架的考察点别被流程列表骗了不是说框架不能用而是选型时要注意四个细节。第一框架是否允许自定义任务协议如果只能按它内置的任务流转方式走后面想加“重试给另一个代理”这种逻辑会非常痛苦。第二框架如何处理中间结果是集中存储还是完全分散如果框架没有中心黑板排查问题时大概率要靠肉眼翻日志。第三框架的失败重建机制是自动还是手动我看过不少框架只提供“失败后重跑整个流程”粒度太粗重跑几个代理比重跑全部更常见。第四框架的可观测性到什么程度有些框架的日志只记录“调用了哪个代理、耗时多久”完全不记录模型内部的指令摘要出了问题根本定位不了。5.3 什么时候根本不需要多代理多代理有价值但不少场景单代理加几段工具调用就够了。我给你一个简单的判断标准如果任务的所有步骤共享同一个上下文且步骤之间几乎不产生中间结果那就用单代理。比如“把一段 Markdown 转成 HTML”单代理完全可以胜任拆成多代理纯属增加复杂度。多代理真正适用的条件是步骤之间存在明显的角色隔离需求、工具调用链超过四次、或者某些步骤可以并行执行。如果三个条件一个都不满足我建议先做单代理不要提前造轮子。我犯过的错误是刚开始所有任务都想套多代理架构结果有些简单任务跑起来比单代理慢一倍还频繁触发不必要的重试。后来我定了一条规则先用单代理写一个最小可用版本只有当它因为上下文或角色冲突导致质量问题再拆成代理团队。这个规则帮我省掉了很多无意义的调试时间。6. 落地半年后我想退回重写的几个代码位置6.1 调度层的提示词不该越写越长要结构化早期我为了让调度代理拆解任务更准确不断往提示词里加规则比如“请确保任务互相独立”“请避免重叠”“请考虑数据依赖关系”结果提示词膨胀到 3000 多字模型反而开始过度思考经常生成八个子任务其中一半根本不需要。后来我把这些规则全部移出提示词放进任务分发的校验逻辑里用代码保证子任务之间不重叠而不是让模型“自觉”遵守。提示词只负责表达意图规则用代码实现这是我在半年的项目里体会最深的一点。模型擅长的是理解需求和生成内容不擅长严格遵守一堆否定性规则。把规则外置之后子任务数量从平均 8 个降到了 4 个质量反而高了。6.2 代理注册信息要能动态更新不能写死在常量里第一个版本我把三个执行代理的能力信息写死在 Python 文件里每次要调能力描述、改输入 schema都得改代码重新部署。后来新增第四个代理时发现没改代码根本没法用于是重构成从配置文件加载注册信息。再后来又发现配置更新滞后改成允许运行时通过管理接口动态注册和注销代理。这个能力在早期看起来像过度设计但当代理数量超过五个之后没有动态注册机制每一次新增能力都意味着一次上线流程非常拖节奏。6.3 可观测性建设把每个代理的“思考摘要”刷出来多代理系统最大的排查障碍是不透明调度代理做了什么决策执行代理为什么选择这个路径两者之间的消息到底流转了几轮我的解决方案是在黑板上增加了一个轻量的“审计日志”通道每次调度代理做决策时记录决策原因每次执行代理返回时记录摘要和原始结果的前几十个字符。排查问题时直接按 task_id 拉出整条消息链比翻底层模型日志高效得多。最后再分享一个实际体会多代理编排这类系统一开始跑通 demo 根本没有成就感真正的难题全在“跑通了之后”——任务漂移、上下文膨胀、格式抖动、代理间死循环每一个都会在最想不到的时间点冒出来。我最后的建议是先做一个小规模最小系统把注册、分发、验收、超时这四件事用最朴素的数据结构跑通再慢慢扩展角色和工具。代理数量控制在三个到五个之间超过这个量级调度层本身的复杂度和决策成本会反超收益。如果这篇文章让一些读者少走两三个坑或者让你下定决心“再改一次自己的框架”我就觉得值了。