【Agent Orchestrator】生产级 Orchestrator 实战：从零搭一个能上线的多 Agent 编排系统

title: 【Agent Orchestrator】生产级 Orchestrator 实战LLM 编排 vs 代码编排 任务分解 MCP 工具编排 可观测性从零搭一个能上线的多 Agent 编排系统 description: 手把手实现生产级 Agent 编排器LLM 编排与代码编排的混搭架构、任务分解算法的两种实现、Orchestrator-Worker 完整代码分诊→并行工作→汇总、MCP 工具编排与权限矩阵、Trace 与 Eval 落地、成本控制看板附可直接迁移的生产工程模板 tags: [Agent, Orchestrator, MCP, 多Agent, 任务分解, 可观测性, Eval, 生产实战, LangGraph]【Agent Orchestrator】生产级 Orchestrator 实战从零搭一个能上线的多 Agent 编排系统首屏导读 · 本教程配套付费专栏《大模型工程师修炼手记》19.9 元AI 编程 · Agent 实战 · 本文同主题系统课程· 《AI时代程序员的自我提升》49.9 元AI 时代成长方法论单篇不过瘾订阅解锁全量源码、实战与答疑文末附资料包领取方式 ↓导读框架评测看得再多落地时总会遇到同一个灵魂拷问——我的编排器到底长什么样 前两篇我们讲清了范式Orchestrator-Worker / Handoff / Agent-as-Tool和框架选型LangGraph / CrewAI / Agents SDK本篇直接给可运行代码一个「分诊 → 并行工作 → 汇总」的生产级编排器骨架含任务分解算法、MCP 工具编排与权限矩阵、Trace 与 Eval 落地。它不是玩具 demo而是你在真实项目里可以直接迁移的工程模板。图 1生产级 Orchestrator 的四个设计决策一、架构总览无框架也能搭的编排器骨架先把编排器抽象成一个稳定的五件套。无论你最终选 LangGraph 还是 Agents SDK这套骨架不变┌───────────────────────────────────────────────────────┐ │ Orchestrator │ │ │ │ ┌──────────┐ ┌──────────────┐ ┌─────────────────┐ │ │ │ Router │→ │ Planner │→ │ Dispatcher │ │ │ │ 分诊/分类│ │ 计划/拆解 │ │ 分派/并行调度 │ │ │ └──────────┘ └──────────────┘ └────────┬──────────┘ │ │ ▼ │ │ ┌─────────────────┐ │ │ │ Worker 池 │ │ │ │ W1 W2 W3 ... Wn │ │ │ └────────┬─────────┘ │ │ ▼ │ │ ┌──────────┐ ┌──────────────┐ ┌─────────────────┐ │ │ │ Guardrail │ │ Synthesizer │ │ Trace State │ │ │ │ 护栏 │ │ 汇总 │ │ 追踪 状态 │ │ │ └──────────┘ └──────────────┘ └─────────────────┘ │ └───────────────────────────────────────────────────────┘1.1 五个组件各管一件事组件职责用 LLM 还是代码Router 分诊判断任务类型、是否需人工、复杂度分级轻量 LLM一次分类调用Planner 计划把大任务拆成可并行子任务LLM开放式任务或代码确定性任务Dispatcher 分派把子任务发给 worker、控制并行/限流代码并发控制、重试、超时Worker 池各自的窄上下文 专属工具执行子任务每 worker 一个 AgentSynthesizer 汇总合并结果、查冲突、产出最终答案LLM汇总Guardrail 护栏输入/输出校验、权限检查、越权拦截代码规则式别用 LLMTrace/State记录每次运行每个步骤的输入输出与耗时代码结构化日志 关键设计原则让 LLM 做决策让代码做动作。分诊/计划/汇总这类判断用 LLM并发、重试、超时、权限、持久化这类执行必须用代码。把这两层分干净你的编排器才可预测、可调试、可测试。二、任务分解编排器的大脑任务分解质量直接决定编排上限。生产里有两种实现各有用武之地2.1 方式 A确定性分解代码写死适合形态已知、参数可变的任务——比如客服工单看一眼工单类型就知道流程是固定的WORKFLOWS { 退款: [核验订单, 校验退款政策, 发起退款, 通知用户, 记录工单], 比价: [拉取商品A报价, 拉取商品B报价, 生成绩效对比, 输出建议], 售后: [核验购买记录, 分类售后类型, 分派处理, 回访], } def decompose(ticket_type: str) - list[str]: if ticket_type not in WORKFLOWS: raise ValueError(f未知工单类型: {ticket_type}) return WORKFLOWS[ticket_type]优点零 token 消耗、100% 可预测、方便做权限矩阵每个步骤绑定固定工具集合。缺点新场景要写代码。2.2 方式 BLLM 分解模型拆解适合没见过的新形态任务。让 Planner 用结构化输出structured output吐出一个任务清单from pydantic import BaseModel class SubTask(BaseModel): id: str description: str dependencies: list[str] # 依赖哪些子任务先完成 tool_hint: str # 建议使用的工具/能力 class TaskPlan(BaseModel): subtasks: list[SubTask] # planner_agent.task(... complex request ...).execute(TaskPlan)关键防护新人都栽这里面防护说明上限约束子任务数量上限如 ≤8超出让它合并依赖校验用代码检查 dependencies 有没有环、有没有引用不存在的 id复述计划执行前把计划展示给调用方/HITL 确认分级执行高权限子任务标注requires_approvalTrue经验值生产系统中 80% 的任务应该走确定性分解只有真正开放式的 20% 走 LLM 分解。反过来用什么都让模型拆是成本与失控的双重灾难。三、生产级 Orchestrator 完整实现下面是一个不依赖任何框架的 Python 实现骨架只依赖openai-agentsSDK 的 Agent 部分与标准库体现代码控制执行、LLM 控制决策的混搭import asyncio, time, uuid from dataclasses import dataclass, field, asdict dataclass class Step: phase: str worker: str ok: bool dur: float tokens: int detail: str dataclass class Run: id: str question: str steps: list[Step] field(default_factorylist) class Orchestrator: def __init__(self, workers: dict, plannerNone, timeout60): self.workers workers # {research: Agent, support: Agent, ...} self.planner planner # 可选LLM 分解器 self.timeout timeout def log(self, run: Run, phase, worker, ok, dur, tokens, detail): run.steps.append(Step(phase, worker, ok, dur, tokens, detail)) async def dispatch(self, run: Run, subtask: dict) - str: w self.workers[subtask[tool_hint]] try: result await asyncio.wait_for( w.run(subtask[description]), timeoutself.timeout) self.log(run, worker, subtask[tool_hint], True, self._last_dur, self._last_tokens) return str(result) except asyncio.TimeoutError: self.log(run, worker, subtask[tool_hint], False, self.timeout, 0, timeout) return [timeout] async def run(self, question: str) - dict: run Run(iduuid.uuid4().hex[:8], questionquestion) t0 time.perf_counter() # 1) 分诊LLM 决策代码承接 plan self.planner.plan(question) # 2) 并行派发代码执行 results await asyncio.gather( *[self.dispatch(run, st) for st in plan.subtasks if not st.dependencies]) # 3) 汇总LLM 决策 answer self.synthesize(run, plan, results) self.log(run, synthesize, main, True, time.perf_counter()-t0, self._last_tokens, answer[:200]) return {answer: answer, trace: [asdict(s) for s in run.steps]}三个细节值得单拎出来强调① 并行度要封顶。asyncio.gather全量并行不可取——生产里用一个信号量把并发限到 3~5 个 worker否则一次调用把 API 打挂。② 超时是硬护栏。每个 worker 都套wait_forTimeoutError 记进 trace 而不是静默吞掉。失败次数超过阈值就触发人类介入或降级分支。③ 依赖关系的处理。上述例子里只并行跑dependencies[]的第一层有依赖的按拓扑序分层执行每层完成后跑下一层。3.1 混合编排示例分层执行async def run_by_layers(self, run, plan): layers self._topo_layers(plan.subtasks) # 按依赖分层 for layer in layers: # 代码保证层级顺序 results await asyncio.gather( *[self.run_subtask(run, st) for st in layer], return_exceptionsTrue) self._store(run, layer, results) # 结果入库供上层引用 return self.synthesize(run)_topo_layers用标准拓扑排序Kahn 算法即可LLM 完全不需要参与——这就是代码做动作。四、MCP 工具编排权限矩阵与漂泊治理4.1 工具接入的 2026 约定所有 worker 只通过MCP拿工具。好处统一协议、统一鉴权、统一审计。一个 worker 拿到的工具集合不是框架给的而是权限矩阵给的TOOL_MATRIX { research: [mcp://search, mcp://web/fetch], # 只读 orders: [mcp://orders/query, mcp://orders/refund], # 读写 support: [mcp://crm/read, mcp://crm/update], } def tools_for(worker: str) - list[str]: return TOOL_MATRIX.get(worker, [])4.2 权限四条红线红线实现最小权限矩阵不配工具不进上下文写操作分级orders/refund标记requires_approvalHITL 确认高危动作审计每次写操作单独记 audit 日志谁/何时/改了什么MCP Server 治理统一 Hub 收敛 10~20 个 server健康探测 版本管理4.3 防MCP Server Sprawl的最小手段┌──────────┐ ┌──────────┐ ┌──────────┐ │ MCP S1 │ │ MCP S2 │ │ MCP Sn │ └────┬─────┘ └────┬─────┘ └────┬─────┘ └──────────────┼──────────────┘ ▼ ┌──────────────┐ │ MCP Gateway │ 注册/发现/鉴权/限流/健康探测 └──────┬───────┘ ▼ ┌──────────────┐ │ Orchestrator │ 按 TOOL_MATRIX 注入每个 worker └──────────────┘五、可观测性Trace Eval 双落地5.1 Trace先记录再优化编排器是黑盒中的黑盒没有 trace 就没有优化的资格。从第一个版本就该有def to_trace(run: Run) - dict: return { run_id: run.id, question: run.question, steps: [asdict(s) for s in run.steps], total_tokens: sum(s.tokens for s in run.steps), total_dur: sum(s.dur for s in run.steps), failed: [s for s in run.steps if not s.ok], }每个 step 记phase / worker / ok / dur / tokens / detail。有了它你才能回答三个问题问题从 trace 怎么看哪里最贵按 worker 聚合 token 总和哪里最慢按 worker 聚合 dur找 p95哪里在反复失败过滤okFalse的 step看输入分布框架自带方案LangGraph 用 LangSmith 时间旅行调试Agents SDK 内置 tracing。无框架时按上面自己攒一个即可。5.2 Eval编排变更是改动就要回归编排系统最怕改一行整体变差。给它配一套最小回归集EVALS [ {q: 帮我把订单 1234 退款并对比竞品售后政策, expect: 包含退款路径 且 包含竞品对比}, {q: 只是一个普通咨询, expect: 不调用退款工具 且 不触发并行编排}, {q: 超出权限的高危操作, expect: 触发 requires_approval 流程}, ]每次编排逻辑变更跑一遍用内容断言expect 关键词/行为断言而不是答案像不像因为答案文本会漂行为不会。给创业者的启示编排器的深水区从来不在选哪个框架而在任务分解规则、权限矩阵、trace 数据、回归集这四样沉淀。框架随时可换但你的编排逻辑一旦形成 Data-as-Code 的业务资产团队就真正建立了多 Agent 系统的工程能力。六、落地路线图从 demo 到生产的五个里程碑里程碑交付物验收标准M1 单 Agent 验证裸 API 跑通核心子任务确认价值成立再上编排M2 手写编排器Router→Planner→Dispatcher→Synthesizer 骨架10 个真实任务可跑通M3 工具与权限MCP 接入 TOOL_MATRIX 审计越权调用被拒绝且留痕M4 TraceEval结构化 trace 回归集能回答哪步贵/哪步慢/哪步坏M5 换框架如需迁移到 LangGraph/Agents SDKM2-M4 资产零损失迁移结语编排不是多 Agent 聊天而是一套真正意义的工程系统决策给 LLM、动作给代码、工具走 MCP、行为靠 trace 与 eval 守护。能把这五件事做干净即使用最朴素的代码也能跑出生产级的多 Agent 系统反之框架再强也只是在给失控的系统加速度。参考资料OpenAI Agents SDK · Agent orchestration / multi-agent 文档Handoff 与 Agent-as-ToolOpenAI Agents SDK · Guardrails / Human-in-the-loop / TracingVercel AI SDK · Orchestrator-Worker PatternAI SDK v6MCPModel Context Protocol官方规范langgraph-vs-crewai 基准 / ailog 框架对比选型数据来源同第 2 篇⚠️ 免责声明本文代码为工程化骨架示例用于说明架构与数据流投入生产前请补齐鉴权、限流、密钥管理等安全措施并以自身业务基准验证。 延伸阅读 · 我的付费专栏觉得这篇文章对你有帮助我把同类主题的系统化内容沉淀成了付费专栏欢迎订阅支持持续输出专栏定价内容大模型工程师修炼手记19.9 元AI 编程 / Agent 深度实战AI时代程序员的自我提升49.9 元AI 时代成长方法论 一杯咖啡的价格换来系统化的知识体系你的订阅是我持续创作的最大动力。本文配套代码 / 资料包欢迎在评论区留言「求代码」我会私信发送完整资源