## 动态拆解与多 Agent 协作 固定 `fork` 适合作者预先知道的分支。 当工作数量、负责人和依赖关系只能在运行时确定时，Planner Agent 可以通过 callback 返回受约束的 WorkItem 集合，再由 `fan_out` 建立动态但可恢复的执行作用域。 runtime 只接受闭合数据契约，不接受 Agent 临时发明图结构。 ### 本章目标 学完本章，你将能够： - 让 Planner callback 产生编译器拥有的 `work_items` 值。 - 使用 `fan_out`、item 边和配对 `join` 执行动态工作。 - 解释依赖调度、并发上限、失败收集与结果顺序。 - 区分 item 直接引用 `item.*`、依赖引用 `item.dependencies` 和 join 汇总引用 `items` 的作用域。 ### 适用边界 - **覆盖**：一批 WorkItem、成功依赖、有界并行、`failureMode: collect` 与一次配对 join。 - **不覆盖**：fan-out body 内嵌套 `fan_out`、失败后条件依赖、按 Agent 设置独立并发和 Agent 编写子图；当前 v1 不支持这些能力。配对 join 之后可以由静态图进入另一个 Planner callback 和第二个 `fan_out`。 - **假设**：Planner 与执行候选都来自调用输入的受信 `agent` 值，工作范围来自受权 `scope`。 ### WorkItem 是封闭契约 Plan 作者不能在 `input`、`state` 或普通 `callbackOutput` 中声明 `type: work_items`。 只有声明了 `sendMessage.workItems.allowedResponders` 的 callback 能产生编译器拥有的 `output.workItems`，并且该值必须直接进入一个 `fan_out`。 这样，Planner 只能选择允许列表内的响应者，不能绕过 Plan 的人员边界。 调用者必须提供至少一个允许的响应 Agent；空列表不能形成可执行的 WorkItem callback 契约。 一次 callback 最多返回 256 项，即公开常量 `MaxWorkItems=256`；超过上限会在打开 fan-out 前被拒绝。 每项 `selectOne` 候选池最多 64 个 Agent；runtime 按首次出现顺序去重并冻结有序池。分发时选择当下有容量的最高优先级候选；winner 一旦随消息 receipt 提交便被冻结，重试、恢复或重启都不能重新选人。 每个 WorkItem 具有稳定字段： - `key`：本批唯一键。 - `dependsOn`：直接依赖的 WorkItem 键。 - `workScope`：执行 Agent 的具体工作范围。 - `relatedScopes`：按需读取的相关范围。 - `responder`：允许列表内的固定 Agent，或 `{selectOne: [候选 Agent…]}`。 - `instruction`：本项工作的明确说明。 `fan_out` 的 item 边还提供 `item.dependencies`。 它按直接依赖键保存终态结果 envelope；其中 `value` 是依赖项到 closing join 的业务值，不是原动作的裸 `output` 命名空间。 深层依赖不会递归复制整棵结果图。 ### 有界并行覆盖整条 item lineage `maxConcurrency` 省略时，编译器冻结默认值 4；显式值必须在 1 到 64 之间。 这个上限覆盖一项 WorkItem 的完整 lineage，包括消息、等待、callback、retry 和静态子节点，不只限制第一条消息。 scheduler 按 callback 返回的源顺序访问 WorkItem。 只有所有直接依赖成功且有并发槽位时，pending 项才会启动。 如果依赖失败或被阻断，后继变成 `blocked`；`blockedBy` 保存按源顺序传播的失败根键。 没有依赖关系的其他项继续运行。 同一 Workspace 内由同一个响应 Agent 承担的 callback 会稳定按 FIFO 串行；不同 Agent 仍可并行。 因此 `maxConcurrency` 是全局 item lineage 上限，不承诺同一个 Agent 同时处理多个 WorkItem。 `failureMode` 表示失败模式，当前只能取值 `collect`。 item 失败不会立即让整个 Plan 失败；runtime 等所有项进入终态，再让配对 join 就绪。 正常 `collect` 完成时，join 结果只包含 `succeeded` 成功项、`failed` 失败项和 `blocked` 阻断项，后续 Reviewer 或控制逻辑必须检查这些 envelope 的状态。 若 Run 或承载 fan-out 的外层 scope 被取消，runtime 会把尚未终态的 item 标为 `canceled`，同时取消 closing join；这条取消路径不会生成 completed join 汇总。 > **❌ 反例：** 把 `collect` 写成 fail-fast，或把 join 完成解释为“所有业务项成功”。 > **✅ 正例：** 把正常 join 完成解释为“所有 WorkItem 已成为 `succeeded`、`failed` 或 `blocked`”，再由后续节点评估结果数组并决定成功、补救或人工兜底。 ### Planner 提示词 Planner 必须返回可执行、可审计且受允许响应者约束的 WorkItem。 ```text 把需求拆成最小可独立验收的 WorkItem。 每项必须给出唯一 key、直接 dependsOn、workScope、relatedScopes、responder 和 instruction。 responder 只能是一个允许的固定 Agent，或 {selectOne: [按优先级排列的允许 Agent]}；候选池非空、最多 64 项，重复项会按首次出现去重。 依赖必须引用同一批已存在的 key，且不能形成环。 一次最多返回 256 项。 不要返回第二层子图、脚本或新的 Agent 名单。 ``` runtime 校验 callback 的结构、键、依赖、响应者和资源上限。 提示词帮助 Planner 形成高质量拆分，但不能替代这些机器约束。 ### 从 Planner 到收集结果 动态协作流程分为五段： 1. `plan_work` 向 Planner Agent 发送需求，并用 `allowedResponders` 冻结可选执行者。 2. callback 返回 `output.workItems`，直接绑定到 `fan_out_work` 的必填输入。 3. `fan_out_work` 按依赖和 `maxConcurrency` 激活 `implement`。 4. item 边从 `item.*` 构造执行 Agent 的局部输入；callback 结果沿 item 端口到达配对 join。 5. 正常 `collect` 路径中，`join_work` 在全部 item 成为 `succeeded`、`failed` 或 `blocked` 后，以源顺序生成 `items` 数组并交付 Plan 输出；取消路径不执行这一步。 ### 可编译配置 ```yaml apiVersion: loop.plan/v1 kind: PlanVersion input: planner: {type: agent, required: true} developers: type: array required: true items: {type: agent, required: true} parent_task: {type: scope, required: true} requirement_scopes: type: array required: true items: {type: scope, required: true} output: results: type: array required: true items: {type: object, required: true} graph: entry: nodeId: plan_work port: in input: planner: {ref: plan.input.planner} developers: {ref: plan.input.developers} parent_task: {ref: plan.input.parent_task} requirement_scopes: {ref: plan.input.requirement_scopes} nodes: - id: plan_work type: send_message input: planner: {type: agent, required: true} developers: type: array required: true items: {type: agent, required: true} parent_task: {type: scope, required: true} requirement_scopes: type: array required: true items: {type: scope, required: true} sendMessage: scope: {ref: input.parent_task} responder: {ref: input.planner} text: "拆解需求并返回 WorkItem" completion: callback workScope: {ref: input.parent_task} relatedScopes: {ref: input.requirement_scopes} workItems: allowedResponders: {ref: input.developers} callbackOutput: summary: {type: string, required: true} - id: fan_out_work type: fan_out input: items: {type: work_items, required: true} fanOut: items: {ref: input.items} maxConcurrency: 2 failureMode: collect - id: implement type: send_message input: key: {type: string, required: true} dependsOn: type: array required: true items: {type: string, required: true} workScope: {type: scope, required: true} relatedScopes: type: array required: true items: {type: scope, required: true} responder: {type: responder, required: true} instruction: {type: string, required: true} dependencies: {type: object, required: true} sendMessage: scope: {ref: input.workScope} responder: {ref: input.responder} text: "{{ input.instruction }}" completion: callback workScope: {ref: input.workScope} relatedScopes: {ref: input.relatedScopes} callbackOutput: result: {type: object, required: true} - id: join_work type: join input: result: {type: object, required: true} join: fanOutId: fan_out_work mode: all result: results: type: array required: true items: {type: object, required: true} merge: results: {ref: items} - id: done type: end end: {outcome: succeeded} edges: - id: plan_to_fan_out source: {nodeId: plan_work, port: completed} target: {nodeId: fan_out_work, port: in} bindings: items: {ref: output.workItems} - id: fan_out_to_implement source: {nodeId: fan_out_work, port: item} target: {nodeId: implement, port: in} bindings: key: {ref: item.key} dependsOn: {ref: item.dependsOn} workScope: {ref: item.workScope} relatedScopes: {ref: item.relatedScopes} responder: {ref: item.responder} instruction: {ref: item.instruction} dependencies: {ref: item.dependencies} - id: implement_to_join source: {nodeId: implement, port: completed} target: {nodeId: join_work, port: item} bindings: result: {ref: output.result} - id: join_to_done source: {nodeId: join_work, port: completed} target: {nodeId: done, port: in} bindings: results: {ref: output.results} ``` 这个示例与 [`internal/workflow/plan/fan_out_v4_test.go: TestCompileWorkItemFanOutV4`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/plan/fan_out_v4_test.go)、[`internal/workflow/run/fanout_select_one_plantype_test.go: TestFanOutSelectOneItemDispatchesToCandidates`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/run/fanout_select_one_plantype_test.go) 和 [`internal/workflow/run/fanout_test.go: TestFanOutConcurrencyCollectsFailedItemAndPreservesSourceOrder`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/run/fanout_test.go) 的真实 fixture 对齐。 编译器由 [`internal/workflow/plan/fan_out_scope.go: validator.validateFanOutScopes`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/plan/fan_out_scope.go) 证明静态 body 边界和配对 join，由 [`internal/workflow/plan/work_items_provenance.go: validator.validateFanOutWorkItemsSources`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/plan/work_items_provenance.go) 证明 WorkItem 来源；[`internal/workflow/plan/budget.go: MaxWorkItems`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/plan/budget.go) 固定 256 项上限。 runtime 的调度与单一依赖读取见 [`internal/workflow/run/fanout.go: Kernel.scheduleFanOut / fanOutDependencyView`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/run/fanout.go)，终态 envelope 与 closing 物化见 [`internal/workflow/run/fanout_resolution.go: fanOutTerminalEnvelope / fanOutResultEnvelopes / Kernel.validateFanOutItemCandidate`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/run/fanout_resolution.go)。 取消 fence 与依赖值单次读取分别由 [`internal/workflow/run/fanout_task_watch_recovery_test.go: TestFanOutRecoveryCancelAndLateCompletionAreFenced`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/run/fanout_task_watch_recovery_test.go) 和 [`internal/workflow/run/fanout_test.go: TestFanOutMaterializesDependencyReferencesOnceAndFreezesChildInput`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/run/fanout_test.go) 覆盖。 ### 配对 join 的结果不是 callback 到达顺序 `join.fanOutId` 与 `join.forkId` 互斥，fan-out join 只能使用 `mode: all`。 `merge` 中的 `items` 是按 Planner callback 源顺序排列的终态 envelope 数组。 即使后面的 WorkItem 先完成，输出顺序也不改变。 零 WorkItem 会立即得到空数组，并只激活一次后续节点。 终态 WorkItem 是依赖读取的单一来源，DAG 不会沿每条依赖边复制业务值；形成 join output、下游 Activation input 和最终 Run output 时仍会分别物化所需值。 runtime 在恢复后重新校验冻结列表、依赖图、状态和 join 激活，再继续调度；重放不能新增、删除、重排或改派 WorkItem。 ### 常见异常与处置 | 现象 | 原因 | 处置 | | --- | --- | --- | | `fan_out.items_source` | `work_items` 不是来自允许的 Planner callback | 直接绑定 `output.workItems` 到唯一 fan-out | | callback 被拒绝 | responder 不在允许列表，键重复、依赖无效或超预算 | 修正同一 callback 的 WorkItem 集合 | | item 一直 pending | 直接依赖尚未成功，或并发槽位已满 | 查看依赖终态与 `maxConcurrency` | | item 变成 blocked | 某个依赖失败或被阻断 | 沿 `blockedBy` 找到失败根并决定补救 | | join 已完成但有失败 | 正常 `collect` 只保证全部项成功、失败或阻断后汇总 | 检查每个 envelope 状态，再做业务结论 | | 取消后没有结果数组 | Run 或外层 scope 取消了非终态 item 和 closing join | 把取消视为独立终态，不读取 completed join 汇总 | | 状态竞争诊断 | item body 写入共享 Run 状态 | 让每项返回局部值，在 join 后统一处理 | ### 审批 不要在每个 item 中默认创建无界人工审批。 若确有必要，应把审批人、响应时限和积压处置作为容量设计，并记住 `request_approval` 本身没有创作期超时。 常见做法是在 Planner 前审批拆分目标，或在 join 后由一个 Reviewer 评估完整终态数组；具体取舍是业务治理策略，不是 fan-out 自动语义。 ### 外部系统 `workScope` 与 `relatedScopes` 是 runtime 校验 callback scope 合同并向 Agent 暴露协作上下文的边界，不是 App 或 Extension 的全局权限沙箱。 具体 App/Extension action 是否允许执行，仍由该 action 自身的身份与授权检查决定。 WorkItem 的 `key` 只标识本批项目；外部业务键必须由可信调用输入或平台签发的“业务对象 + 来源批准版本或业务事件 + 期望变化 + 目标”构成，跨消息、Invocation 与 Run 保持不变。Run、Activation、Effect 和 WorkItem 的 `key` 可以关联审计，但不能靠加上这些技术 ID 来制造跨 Run 唯一性。 某项外部写入成功后 callback 丢失时，Agent 应先按完整业务键查询已有结果再补交，不能重复创建资源。 ### 验收 - [ ] `work_items` 只来自声明了允许响应者的 Planner callback。 - [ ] 能说明 item 启动条件、完整 lineage 并发上限和 `collect` 语义。 - [ ] 能区分 `item.*`、`item.dependencies..value` 和 join `items`。 - [ ] join 后检查终态 envelope，不把完成等同于全部成功。 - [ ] 能说明正常 collect 汇总与 Run/外层 scope 取消路径的区别。 - [ ] WorkItem responder body 使用 `type: responder`，并能说明固定 Agent、`selectOne` 的 64 项上限、去重、最高优先级可用者与 winner 冻结。 - [ ] 外部业务键来源可信且跨 Run 稳定，Run / Activation / Effect / WorkItem ID 仅作审计关联。 - [ ] 本章完整 Plan 原样编译后 `diagnostics=[]`。 ### 媒体槽位 建议图示包含 4 个带依赖的 WorkItem、2 个并发槽位、一个失败根、一个 blocked 后继和最终源顺序结果数组。 ### 小结 - [ ] Planner 返回的是封闭 WorkItem 数据，不是动态 Plan 子图。 - [ ] fan-out 按依赖和有界并发执行，失败模式只有 `collect`。 - [ ] 正常 `collect` 路径在全部项成为 `succeeded`、`failed` 或 `blocked` 后，由配对 join 按源顺序收集结果；取消路径不汇总。 下一章把幂等、恢复、最小权限、补偿和人工兜底组合成生产级治理方法。