外观
一文搞懂多智能体主从编排：任务分发与结果汇总实战
问题背景：单体 Agent 为什么不够用
在真实的 AI 编程助手或研发自动化场景里，一个需求往往不是“让大模型写一段代码”这么简单。例如“为订单服务补一个退款接口，并同步更新测试和文档”，至少包含需求澄清、接口设计、代码实现、单测编写、文档更新、结果审查六类工作。如果全部塞进一个 Agent，提示词会迅速膨胀，上下文互相污染，模型容易遗漏约束，也难以并行和重试。
multi-agent（多智能体）方案把不同职责拆给专职 Worker，再由一个 Supervisor 统一调度。supervisor-worker 主从编排模式的核心价值不是“让更多 Agent 看起来很酷”，而是把复杂任务变成可控的工程流程：可拆解、可并行、可观测、可重试、可汇总。
方案对比：主从编排、自由协商与固定工作流
常见多智能体协作模式主要有三类。
第一种是自由协商模式，多个 Agent 互相发言、投票、辩论。它适合方案评审、创意发散，但缺点也明显：轮次不可控、成本高、容易跑题，不适合确定性的开发流水线。
第二种是固定工作流模式，由代码硬编码节点顺序，比如“分析→编码→测试”。它稳定、便宜、易测试，但适应性差；一旦任务只需其中两个节点，仍可能走完整链路。
第三种就是本文重点的主从编排模式。Supervisor 负责理解目标、生成任务 DAG、分发任务、处理失败并汇总结果；Worker 只关注单一能力，例如代码 Worker、测试 Worker、文档 Worker。它在灵活性和可控性之间取得平衡，特别适合多智能体协作开发。
设计关键：Supervisor 只做编排，不替 Worker 干活
主从模式最常见的失败原因，是 Supervisor 既当调度器又当执行者。它一边拆任务，一边顺手写代码、改测试、润色文档，最终又退化成单体 Agent。清晰边界应当是：Supervisor 负责决策与协议，Worker 负责专业执行。
建议 Supervisor 至少维护五件事：
- 原始目标：用户要解决什么问题，验收标准是什么。
- 任务列表：每个任务有唯一 ID、负责人、依赖、输入、状态。
- 共享上下文：只传递必要信息，避免把全部历史无差别塞给 Worker。
- 失败策略：超时、重试、降级、人工确认要有明确规则。
- 汇总契约：所有 Worker 输出统一结构，方便机器合并和审查。
Worker 的设计则应遵循“小而专”。一个 Worker 最好只拥有一个明确能力，输入输出通过 Pydantic 模型或 JSON Schema 约束。这样既能并发执行，也能在失败时单独重放，而不是重跑整个多智能体流程。
可运行示例：用 Python 实现任务分发与结果汇总
下面示例不依赖特定大模型厂商，先用本地可运行的模拟 Worker 演示编排骨架。接入真实 LLM 时，只需把 worker.run 内部替换为模型调用即可。
python
import asyncio
import json
from dataclasses import dataclass, field, asdict
from enum import Enum
from typing import Any, Dict, List, Optional
class TaskStatus(str, Enum):
PENDING = "pending"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
@dataclass
class Task:
id: str
name: str
worker: str
payload: Dict[str, Any]
dependencies: List[str] = field(default_factory=list)
status: TaskStatus = TaskStatus.PENDING
retries: int = 0
result: Optional[Dict[str, Any]] = None
error: Optional[str] = None
class BaseWorker:
"""所有 Worker 必须返回结构化字典，避免自由文本难以汇总。"""
name = "base"
async def run(self, payload: Dict[str, Any]) -> Dict[str, Any]:
raise NotImplementedError
class CodeWorker(BaseWorker):
name = "code"
async def run(self, payload: Dict[str, Any]) -> Dict[str, Any]:
endpoint = payload["endpoint"]
return {
"artifact": "code",
"summary": f"实现 {endpoint} 接口",
"files": [f"api/{endpoint}.py"],
"risks": ["需要补充参数校验"],
}
class TestWorker(BaseWorker):
name = "test"
async def run(self, payload: Dict[str, Any]) -> Dict[str, Any]:
return {
"artifact": "test",
"summary": "补充成功与失败路径单元测试",
"files": ["tests/test_refund.py"],
"risks": [],
}
class DocWorker(BaseWorker):
name = "doc"
async def run(self, payload: Dict[str, Any]) -> Dict[str, Any]:
return {
"artifact": "doc",
"summary": "更新接口文档和错误码说明",
"files": ["docs/refund.md"],
"risks": [],
}
class Supervisor:
def __init__(self, workers: Dict[str, BaseWorker], max_retries: int = 1):
self.workers = workers
self.max_retries = max_retries
self.tasks: Dict[str, Task] = {}
def add_task(self, task: Task) -> None:
self.tasks[task.id] = task
def _ready_tasks(self) -> List[Task]:
ready = []
for task in self.tasks.values():
if task.status != TaskStatus.PENDING:
continue
deps = [self.tasks[d] for d in task.dependencies]
if all(d.status == TaskStatus.SUCCESS for d in deps):
ready.append(task)
return ready
async def _run_one(self, task: Task) -> None:
worker = self.workers.get(task.worker)
if worker is None:
task.status = TaskStatus.FAILED
task.error = f"未知 Worker: {task.worker}"
return
task.status = TaskStatus.RUNNING
try:
# 真实场景可在此注入超时：asyncio.wait_for(worker.run(...), timeout=30)
task.result = await worker.run(task.payload)
task.status = TaskStatus.SUCCESS
except Exception as exc:
task.error = str(exc)
if task.retries < self.max_retries:
task.retries += 1
task.status = TaskStatus.PENDING
else:
task.status = TaskStatus.FAILED
async def execute(self) -> Dict[str, Any]:
while True:
ready = self._ready_tasks()
if not ready:
break
await asyncio.gather(*(self._run_one(task) for task in ready))
return self.summarize()
def summarize(self) -> Dict[str, Any]:
files: List[str] = []
risks: List[str] = []
summaries: List[str] = []
failed: List[Dict[str, str]] = []
for task in self.tasks.values():
if task.status == TaskStatus.FAILED:
failed.append({"id": task.id, "error": task.error or "unknown"})
if task.result:
files.extend(task.result.get("files", []))
risks.extend(task.result.get("risks", []))
summaries.append(task.result.get("summary", ""))
return {
"success": not failed,
"changed_files": sorted(set(files)),
"risks": risks,
"summary": "；".join(summaries),
"failed_tasks": failed,
"trace": [
{"id": t.id, "worker": t.worker, "status": t.status.value, "retries": t.retries}
for t in self.tasks.values()
],
}
async def main() -> None:
supervisor = Supervisor({
"code": CodeWorker(),
"test": TestWorker(),
"doc": DocWorker(),
})
# 测试和文档依赖代码产物，因此先实现，再并行补充。
supervisor.add_task(Task(
id="t1", name="实现退款接口", worker="code",
payload={"endpoint": "refund"}
))
supervisor.add_task(Task(
id="t2", name="补充退款测试", worker="test",
payload={"endpoint": "refund"}, dependencies=["t1"]
))
supervisor.add_task(Task(
id="t3", name="更新退款文档", worker="doc",
payload={"endpoint": "refund"}, dependencies=["t1"]
))
result = await supervisor.execute()
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
asyncio.run(main())运行后会得到统一的汇总结果，包括变更文件、风险、失败任务和执行轨迹。这个 trace 很重要：它能把多智能体从“黑盒聊天”变成可观测的任务系统，后续可接入日志、评测和成本统计。
接入 LLM：让 Supervisor 动态生成任务计划
真实项目中，任务列表不一定由代码写死，可以让 Supervisor 根据用户需求生成结构化计划。为了降低解析失败率，应强制模型输出 JSON，而不是自然语言。
可复用提示词模板如下：
text
你是多智能体系统的 Supervisor，只负责任务编排，不直接编写最终产物。
用户目标：{user_goal}
可用 Worker：
- code：编写或修改代码
- test：编写单元测试和集成测试
- doc：更新文档
请输出 JSON，不要输出 Markdown，也不要解释：
{
"tasks": [
{
"id": "t1",
"name": "任务名称",
"worker": "code|test|doc",
"payload": {"key": "value"},
"dependencies": ["依赖任务 id"],
"reason": "为什么需要该任务"
}
],
"acceptance_criteria": ["可验证的验收标准"]
}
约束：
1. 任务粒度应能在一个 Worker 中独立完成。
2. 没有依赖关系的任务才允许并行。
3. 如果信息不足，创建 worker 为 human 的确认任务。
4. 不允许创建不存在的 Worker。接模型时建议使用 function calling、structured output 或 Pydantic 校验。即使模型偶发输出多余文本，也可以通过 JSON 提取和 schema 验证兜底。校验失败时不要盲目继续执行，应带着错误信息要求模型重新规划。
汇总策略：不要简单拼接 Worker 回复
很多初版 multi-agent 系统会把各 Worker 的文本用换行拼起来当作最终答案，这会造成重复、冲突和格式混乱。更稳妥的方式是“结构化产物 + 最终裁决”。
每个 Worker 返回产物类型、文件列表、摘要、风险、依赖信息和置信度。Supervisor 先做机器合并，例如去重文件、聚合风险、检查依赖缺口；再交给 Reviewer 或 Supervisor 自身做最终审查。审查关注点包括：代码与测试是否一致、文档是否反映真实接口、是否存在未解决风险、失败任务是否影响交付。
避坑总结：主从编排落地的 5 个注意事项
1. 任务依赖写错导致 Worker 读取不到产物
常见现象是测试任务先于代码任务执行，最终返回“找不到接口”或生成一套与实现不匹配的测试。修正方法是在调度器中显式维护 dependencies，只有依赖任务全部成功才置为可运行。不要依赖模型“记得先后顺序”，顺序必须由代码状态机保证。
2. Worker 输出自由文本，汇总阶段频繁解析失败
典型报错是汇总器拿不到文件列表，或从长文本中误提取出普通说明文字。修正方法是定义统一输出 schema，例如 artifact、files、summary、risks、confidence。LLM Worker 也要通过 structured output 约束，本地 Worker 则直接返回字典或 Pydantic 模型。
3. Supervisor 把全部上下文广播给所有 Worker
这会带来 token 成本飙升、上下文污染和提示注入扩散。比如文档 Worker 不需要知道测试失败堆栈，代码 Worker 也不应读取无关密钥配置。修正方法是按任务构造最小 payload，只传依赖任务的必要产物，并对敏感字段做脱敏。
4. 只做重试，不区分错误类型
网络超时适合重试，但 schema 错误、权限不足、需求歧义不会因为重复调用而消失，反而增加成本。修正方法是对异常分类：瞬时错误自动重试，参数错误要求 Supervisor 重新规划，信息不足进入人工确认，安全相关操作默认拒绝并升级处理。
5. 没有执行轨迹，出问题只能看聊天记录
多智能体一旦并行，仅靠最终回复很难判断哪个 Worker 失败、重试了几次、成本花在哪里。修正方法是从第一版就记录任务 trace，包括任务 ID、Worker、状态、重试次数、输入摘要、输出摘要、耗时和模型版本。这样任务编排才能进入可测试、可回放、可评测的工程状态。
结语
supervisor-worker 并不是把一个大模型拆成多个聊天窗口，而是一种以状态、契约和可观测性为核心的任务编排架构。Supervisor 负责规划、调度和裁决，Worker 负责专业执行，结构化结果负责连接二者。只要控制好任务粒度、依赖关系、最小上下文和失败策略，multi-agent 就能从演示型 Demo 走向稳定的 AI 编程工作流。
免责声明
本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。