编排模式：中央协调 + 子任务分配

编排模式：中央协调 + 子任务分配

概念速查

编排模式（Orchestration Pattern）由一个中央协调 Agent 充当控制中枢，负责接收用户请求、将复杂任务分解为可执行的子任务、按能力分发给专业化的子 Agent，并聚合各子 Agent 的返回结果。区别于层级模式中 Manager 可以递归下放管理权，编排模式只有一级协调层——中央协调 Agent 直管所有子 Agent，不允许多层嵌套。通信拓扑恒为星形，所有信息流必经协调节点，子 Agent 之间不建立旁路通信。

底层原理

编排模式的核心运转机制是分解-分配-执行-聚合四段式循环。中央协调 Agent 维护一张任务依赖图（DAG），完整工作流如下：

1. 分解：协调 Agent 将输入任务拆解为原子级子任务，分析子任务间的数据依赖，确定并行边界。分解策略可以是基于预定义模板的静态拆分，也可以是 LLM 驱动的动态拆分，后者更灵活但对协调 Agent 的推理能力要求更高。

2. 分配：根据子 Agent 注册的能力标签（如code_writer、data_analyzer、reviewer）将子任务匹配到最合适的执行者。分配阶段同时确定执行顺序——无依赖的任务全量扇出，有依赖链的任务按拓扑排序分批派发（staged dispatch）。

3. 执行：子 Agent 异步执行任务。执行结果携带状态标识（成功、失败、部分完成）和结构化输出，写回协调 Agent 维护的共享上下文。子 Agent 在隔离的上下文窗口中运行，无法感知其他子 Agent 的存在。

4. 聚合：协调 Agent 持续轮询各子 Agent 的执行状态，待所有依赖满足后执行结果合并。聚合阶段必须处理三类异常——超时、格式错误、部分失败。典型应对策略包括：超时降级（放弃阻塞等待，基于已有结果做便聚合）、重试（对可重入子任务重新派发）、默认值填充（对非关键路径失败任务使用预设兜底值）。

编排模式的核心瓶颈在协调 Agent 的上下文窗口大小。协调 Agent 需要同时追踪所有子任务的状态、中间结果和依赖关系，上下文窗口决定了它能管理的最大子 Agent 数量。当子 Agent 数量超过窗口容量时，必须引入分页式上下文管理或状态压缩机制。

架构设计原则

原则一：协调 Agent 不做执行。协调 Agent 只做分解与聚合，不参与具体子任务。若协调 Agent 同时承担执行工作，会导致"裁判兼球员"的角色冲突，在拆任务和做任务之间频繁切换上下文，降低决策质量。协调 Agent 应保持轻量，核心能力是状态机维护和调度逻辑，而非领域技能。

原则二：子 Agent 全隔离。子 Agent 之间不直接通信，所有信息交换经过协调 Agent 中转。这个约束保证了通信拓扑的简洁性——N 个 Agent 的通信复杂度恒为 O(N)，而非全连接下的 O(N²)。协调 Agent 对子 Agent 的输入/输出做格式标准化，避免各 Agent 使用不同数据方言导致的兼容性问题。

原则三：聚合必须健壮。编排模式的可靠性取决于聚合阶段的容错能力。若某个子 Agent 返回异常但协调 Agent 的设计是彻底失败，整体可用性会急剧下降。推荐策略是"尽力而为"聚合——即便部分子任务失败，协调 Agent 也应基于已完成的结果返回有意义的输出，并在响应中附带失败说明和重试建议。

原则四：分解粒度适中。粒度过粗导致子 Agent 上下文溢出；粒度过细则增加协调 Agent 的调度开销。经验法则是每个子任务的处理时间应大致相等，且不超过协调 Agent 轮询间隔的 80%，避免调度空转。通常一个子任务的复杂度目标为"一个普通工程师能在 15 分钟内完成"的工作量。

以下是一个简化编排引擎的 Python 实现（Python 3.11+）：

# Python 3.11+fromdataclassesimportdataclass,fieldfromtypingimportAny,Callableimportasyncio@dataclassclassSubAgent:name:strcapability:strfn:Callable[[dict],dict]@dataclassclassTask:input:dictassigned_to:strstatus:str="pending"result:dict|None=NoneclassOrchestrator:def__init__(self,agents:list[SubAgent]):self.agents={a.capability:aforainagents}asyncdefrun(self,request:dict)->dict:subtasks=self._decompose(request)pending=[]fortaskinsubtasks:agent=self.agents.get(task.assigned_to)ifnotagent:continuepending.append(self._execute(agent,task))results=awaitasyncio.gather(*pending,return_exceptions=True)returnself._aggregate(subtasks,results)def_decompose(self,request:dict)->list[Task]:return[Task(input=v,assigned_to=k)fork,vinrequest.items()]asyncdef_execute(self,agent:SubAgent,task:Task)->dict:task.status="running"task.result=awaitasyncio.to_thread(agent.fn,task.input)task.status="done"returntask.resultdef_aggregate(self,tasks:list[Task],results:list)->dict:return{t.assigned_to:rfort,rinzip(tasks,results)ifnotisinstance(r,Exception)}

该实现对应编排模式的四个阶段：_decompose分解任务到能力标签的映射；_execute通过asyncio.to_thread避免阻塞事件循环；asyncio.gather实现并行扇出；_aggregate过滤异常结果实现容错聚合。生产级实现还应补充超时控制（asyncio.wait_for）、重试策略和任务依赖图的拓扑排序。