如果你最近在折腾大模型应用，应该对“多代理编排”这个词不陌生。我去年花了不少时间打磨一个内部工具，项目代号就叫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(), key=lambda x: x[1], reverse=True) 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_retries=2): 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/10 | 9/10 | 18.5 分钟 | 6.2 分钟 | 42k token | 11k token |
| 1000 行日志异常分类 | 6/10 | 10/10 | 25 分钟 | 7 分钟 | 38k token | 9k token |
| 12 页产品文档质检 | 5/10 | 8/10 | 32 分钟 | 14 分钟 | 51k token | 13k token |
成功率的提升主要来源于上下文隔离和短工具链。上下文从 40k 降到 10k 级别后，模型产生幻觉和角色冲突的概率明显下降。但第二个任务里出现过两次奇怪的结局，就是通过下面的排障链路解决掉的。
4.2 排查一：执行代理互相刷屏，终止信号迟迟不来
第一个诡异现象是：日志异常分类完成后，负责汇总的结果生成代理居然还在给数据清洗代理发“结果确认”消息，而且往返了十几轮。调度代理一直没叫停，整个流程陷入死循环。
排查链路是这样的：先在黑板里打印每个代理的消息记录，发现结果生成代理的每条消息都带着同一个request_id，这不是任务派发 ID，而是它自己在内部生成的一个确认码。再往深挖，是结果生成代理的系统提示词里写了“持续向数据清洗代理确认字段定义，直到对方确认无误为止”，模型把“确认”理解为“必须反复询问”，而不是“确认一次即可”。
修复方案是在该代理的提示词里增加明确的终止条件：“如果收到对方包含{"ack": true}的响应，立即停止发送新消息。”同时在整个消息循环外层加一个最大轮次max_rounds=5，超过即强制中断并发出告警。这类问题不在模型能力上，而在指令歧义上，模型只是严格遵循了你的字面要求。
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 根本没有成就感，真正的难题全在“跑通了之后”——任务漂移、上下文膨胀、格式抖动、代理间死循环，每一个都会在最想不到的时间点冒出来。我最后的建议是，先做一个小规模最小系统，把注册、分发、验收、超时这四件事用最朴素的数据结构跑通，再慢慢扩展角色和工具。代理数量控制在三个到五个之间，超过这个量级，调度层本身的复杂度和决策成本会反超收益。
如果这篇文章让一些读者少走两三个坑，或者让你下定决心“再改一次自己的框架”，我就觉得值了。
