架构定位

工作流与编排

把复杂任务拆成可恢复的步骤，并通过明确合同协调执行单元。

如何拆解复杂任务，并协调步骤或多个 Agent？

受这些层约束

概念校准

子 Agent

它是什么: 被委派明确目标、输入、权限和返回合同的受限执行单元。
它不是什么: 不等于简单并行；没有合同的角色拆分只会放大成本和冲突。

规划

它是什么: 为多步骤、高成本或需要恢复的任务建立可检查、可更新的执行结构。
它不是什么: 不是所有 Agent 的必备步骤，简单任务常常不需要独立规划器。

验证与评测

它是什么: 用测试、Schema、引用、状态和验收规则检查单次结果，并用数据集衡量系统表现。
它不是什么: 不等于让模型对自己的输出再给一次意见；验证必须落到外部可检查证据。

多 Agent 协作与任务编排

通过消息合同、共享状态、冲突处理和成本评测判断何时拆分角色。

本章只做一件事加入研究、实现和评审角色，并用对照实验说明是否值得。

开始前，Forge 已经具备已完成 Module 11 的“Planner/Executor 与 Agentic 工作流”，其通过验收的 solution 是本章起点。

完成后，Forge 将能够增加研究、实现和评审角色，并证明何时多 Agent 不值得使用。

卡住时的最小恢复点退回单 Agent 基线，逐个加入角色并记录新增成本。

下载本章实验仓库Python 3.12 · pytest · Pydantic · SQLite · deterministic Mock

阶段能力链

01用明确目标、输入、权限和输出 Schema 定义受限角色02使用可关联、可验证的消息封装 Agent 间请求和结果03通过乐观并发控制防止 Agent 静默覆盖彼此的状态更新04对互斥提案应用确定性安全优先规则，并保留决议理由05按能力、权限和当前负载选择最小权限 Agent，无法安全委派时明确拒绝06根据必需角色和最小成功数判断任务可继续、降级或失败07组合最小权限路由、消息关联、共享状态和部分失败策略，生成可审计协作结果

每一步只增加一种可验证能力；后一步建立在前一步已经通过的代码和测试上。

概念校准\n\n在开始操作代码前，先把本章涉及的概念放回正确的工程边界。\n\n
\n\n### 子 Agent sub-agent\n\n- 它是什么: 被委派明确目标、输入、权限、预算和返回合同的受限执行单元，完成后把可验证结果交还协调者。\n- 它不是什么: 它不等于简单并行，也不是为每个角色复制完整上下文；没有边界和返回合同的角色拆分会增加冲突与成本。\n- 与相邻概念的关系: Planning 决定可委派节点，Harness 限制权限与预算，版本化合同交换状态，Supervisor 处理冲突和部分失败。\n- 在 HeatStack Forge 中的位置: Forge 将研究、实现和评审角色限制在不同工具与目录中，每个角色返回结构化证据，协调者验证后才接受。\n- 典型误用与修正: 误用是同时启动多个角色并期待自然协作。修正方法是先定义任务独立性、消息合同与冲突策略，再比较单 Agent 基线。\n\n### 规划 planning\n\n- 它是什么: 为多步骤、高成本或需要恢复的任务建立可检查、可更新的执行结构，明确依赖、预算、状态和重新规划条件。\n- 它不是什么: 它不是所有 Agent 的必备阶段；简单低风险任务可能直接执行更快、更可靠，独立规划器必须证明价值。\n- 与相邻概念的关系: Agent Loop 可直接选择下一动作或执行计划节点；Verification 决定节点完成，State 与 Checkpoint 支持局部重试。\n- 在 HeatStack Forge 中的位置: Forge 将复杂请求转成带依赖、预算、验收和恢复点的任务图，执行器只领取可运行节点，失败后按证据局部重新规划。\n- 典型误用与修正: 误用是生成很长的静态计划并严格照做。修正方法是保持计划最小、状态可验证，并允许输入或工具变化时局部更新。\n\n### 验证与评测 evaluation\n\n- 它是什么: Verification 用测试、Schema、引用、状态或验收规则检查一次结果；Evaluation 用数据集和指标衡量多次运行表现。\n- 它不是什么: 它不是让模型对自己的回答再给一次意见；通过与否必须落到外部可检查证据。\n- 与相邻概念的关系: Agent Loop 依赖 Verification 决定停止，RAG 需要分层评测，Planning 需要节点验收，Observability 提供运行记录。\n- 在 HeatStack Forge 中的位置: Forge 为单次运行保存合同测试与 evidence，再用固定场景集比较质量、延迟、成本、权限和恢复能力。\n- 典型误用与修正: 误用是把一个综合分数当成全部结论。修正方法是定义分层指标、失败样本和版本基线，并结合产品风险评审。\n\n

STAGE 01

用明确目标、输入、权限和输出 Schema 定义受限角色

先看一个具体问题

委派时权限失控

在 Module 11 的 Planner/Executor 工作流中，执行器可以执行任意动作，没有按角色限制权限。

现在要引入多 Agent 角色，但直接委派时可能把超出角色能力的权限交给子 Agent，导致越权操作。

例如一个只读研究员角色被委派了写代码的权限，就会破坏系统安全边界。

本阶段需要定义 AgentRole 和 delegate 函数，确保委派只能缩小权限，不能扩大。

先做判断

在 delegate 函数中，如果调用者请求的 allowed_capabilities 包含角色没有的权限，应该发生什么？

直接创建委派，因为调用者知道自己在做什么
抛出 ValueError，因为委派不能扩大角色权限
自动忽略多余的权限，只保留角色已有的
记录警告但继续执行

判断依据: 正确选项是抛出 ValueError。委派必须受角色边界约束，任何越权请求都应立即失败，而不是静默忽略或信任调用者。

它是什么

角色合同

角色合同是一个不可变的数据结构，明确声明角色的能力集合和可产生的输出键集合。

委派是角色的一个受限实例，它从角色继承身份，但只能使用角色能力的一个子集，并且必须产出角色可产出的键。

delegate 函数是合同的执行点，它验证委派请求是否在角色边界内，通过后才创建委派对象。

它不是什么

不是随意授权

角色合同不是简单的权限列表，它必须同时约束能力和输出，防止角色被要求做它做不到的事。

委派不是复制角色，它不能拥有角色没有的权限，也不能承诺输出角色无法生成的键。

delegate 函数不是信任调用者的便捷构造器，而是强制边界的安全检查点。

它与相邻概念的关系

角色与委派的关系

AgentRole 定义角色的静态边界，Delegation 是运行时的一次具体任务分配，必须满足角色的边界。

delegate 函数连接两者：它接收角色和委派参数，检查 allowed_capabilities 是角色 capabilities 的子集，expected_output_keys 是角色 output_keys 的子集。

如果检查失败，函数抛出 ValueError，阻止不合法的委派产生，从而保证系统安全。

本阶段的边界决定: 当委派请求的权限或输出键超出角色边界时，必须抛出异常而不是静默调整，因为静默调整会掩盖调用者的错误并可能引入安全漏洞。

实验 ID: multi-agent-orchestration-01-role-contracts
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/multi-agent-orchestration/01-role-contracts/
验收: 待验收

教师示范

目标: 演示如何用 AgentRole 和 delegate 创建受限委派，并验证权限子集检查。

假设我们有一个研究员角色，能力是 read 和 search，输出键是 sources 和 summary。

现在要创建一个委派，只允许 read 能力，期望输出 sources。

我们调用 delegate 函数，传入角色和这些参数，函数会检查 allowed_capabilities 是角色能力的子集，expected_output_keys 是角色输出键的子集。

检查通过后，返回一个 Delegation 对象，其 allowed_capabilities 被冻结为 frozenset({‘read’})。

实施步骤

定义 AgentRole 实例：role = AgentRole(‘researcher’, frozenset({‘read’,‘search’}), frozenset({‘sources’,‘summary’}))
调用 delegate：item = delegate(role, ‘d-1’, ‘collect evidence’, {‘topic’:‘retries’}, {‘read’}, {‘sources’})
断言 item.allowed_capabilities == frozenset({‘read’})

真实 solution 代码: forge/multiagent/__init__.py

from .roles import AgentRole, Delegation

__all__ = ["AgentRole", "Delegation"]

真实 solution 代码: forge/multiagent/roles.py

from __future__ import annotations

from dataclasses import dataclass

@dataclass(frozen=True)
class AgentRole:
    role_id: str
    capabilities: frozenset[str]
    output_keys: frozenset[str]

@dataclass(frozen=True)
class Delegation:
    delegation_id: str
    role_id: str
    goal: str
    inputs: dict[str, object]
    allowed_capabilities: frozenset[str]
    expected_output_keys: frozenset[str]

def delegate(role: AgentRole, delegation_id: str, goal: str, inputs: dict[str, object], allowed_capabilities: set[str], expected_output_keys: set[str]) -> Delegation:
    if not delegation_id.strip() or not goal.strip():
        raise ValueError("delegation id and goal are required")
    if not allowed_capabilities <= set(role.capabilities):
        raise ValueError("delegation exceeds role capabilities")
    if not expected_output_keys <= set(role.output_keys):
        raise ValueError("role cannot satisfy output contract")
    return Delegation(delegation_id, role.role_id, goal, dict(inputs), frozenset(allowed_capabilities), frozenset(expected_output_keys))

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

delegate 函数首先检查 delegation_id 和 goal 非空，然后检查 allowed_capabilities 是角色 capabilities 的子集。

在这个例子中，{‘read’} 是 {‘read’,‘search’} 的子集，所以通过。

接着检查 expected_output_keys 是角色 output_keys 的子集，{‘sources’} 是 {‘sources’,‘summary’} 的子集，也通过。

最后创建 Delegation 对象，将 allowed_capabilities 和 expected_output_keys 转换为 frozenset 以保证不可变性。

带提示练习

补全 delegate 函数中缺失的输出键检查，使测试通过。

打开 forge/multiagent/roles.py，找到 delegate 函数。
在现有权限子集检查之后，添加输出键子集检查：if not expected_output_keys <= set(role.output_keys): raise ValueError(‘role cannot satisfy output contract’)
运行测试 python -m pytest -q tests/test_stage.py，确认三个测试全部通过。

逐级提示

提示1：使用集合的 <= 运算符检查子集关系。
提示2：错误消息应包含 ‘output’ 以匹配测试中的 match 参数。
提示3：检查应放在创建 Delegation 对象之前。

完成标准

test_output_contract_must_be_satisfiable 通过
test_delegation_cannot_escalate_permissions 通过
test_delegation_is_bounded_by_role_capabilities 通过

故障注入与排查

在故障实验室中，我们提供了一个有缺陷的 forge/multiagent/roles.py 文件，其中 delegate 函数缺少权限子集检查。运行测试 test_delegation_cannot_escalate_permissions 会失败，因为一个只读研究员角色被允许获得 write 权限。你需要诊断并修复这个缺陷。

本阶段已经学习了角色合同和委派边界，知道 delegate 必须检查 allowed_capabilities 是角色 capabilities 的子集。故障实验室中的缺陷正是缺少这个检查，因此你可以应用刚学到的知识来定位并修复它。

已验证的故障合同

目标测试: test_delegation_cannot_escalate_permissions
故障文件: forge/multiagent/roles.py

诊断顺序

运行故障测试命令：python -m pytest -q ../tests/test_stage.py::test_delegation_cannot_escalate_permissions
观察输出：测试失败，提示 ‘Failed: DID NOT RAISE ValueError’，说明 delegate 没有抛出异常。
检查 forge/multiagent/roles.py 中的 delegate 函数，发现缺少检查 allowed_capabilities 是否为角色 capabilities 子集的代码。
确认根因：函数直接信任传入的 allowed_capabilities，没有与角色边界比较。

修复原则: 在 delegate 函数中添加权限子集检查：if not allowed_capabilities <= set(role.capabilities): raise ValueError(‘delegation exceeds role capabilities’)。这样当请求的权限超出角色能力时，函数会抛出 ValueError，测试将通过。

python -m pytest -q ../tests/test_stage.py::test_delegation_cannot_escalate_permissions

迁移挑战

在办公研究场景中，你有一个 researcher 角色，能力为 read，输出键为 summary。现在需要创建一个委派，要求读取文档并输出 summary，但错误地请求了 write 权限。

改变的条件: 将委派请求的 allowed_capabilities 从 {‘read’} 改为 {‘write’}，而角色能力仍为 {‘read’}。

在办公研究场景中，创建 researcher 角色（能力为 read，输出键为 summary），然后调用 delegate 函数，传入 allowed_capabilities={‘write’} 和 expected_output_keys={‘summary’}。观察并确认 delegate 抛出 ValueError，且错误消息包含 ‘exceeds’。

完成标准

必须调用 delegate，且 allowed_capabilities 包含 ‘write’，而角色能力只有 ‘read’。
必须确认 delegate 抛出 ValueError，且错误消息包含 ‘exceeds’。
必须使用 researcher 角色和指定的参数，不得修改角色能力或输出键。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么委派不能扩大角色权限？请从安全边界和系统可预测性的角度解释。

一个有力回答应包含

委派扩大权限会破坏角色的最小权限原则，导致越权操作风险。
如果委派可以任意增加权限，系统的安全边界将无法静态验证，增加攻击面。
限制委派权限为角色能力的子集，可以保证任何委派的行为都在角色定义的安全范围内。

掌握证据

所有三个测试通过：test_delegation_is_bounded_by_role_capabilities、test_delegation_cannot_escalate_permissions、test_output_contract_must_be_satisfiable。
delegate 函数在 allowed_capabilities 不是角色 capabilities 子集时抛出 ValueError，错误消息包含 ‘exceeds’。
delegate 函数在 expected_output_keys 不是角色 output_keys 子集时抛出 ValueError，错误消息包含 ‘output’。

STAGE 02

使用可关联、可验证的消息封装 Agent 间请求和结果

先看一个具体问题

响应必须来自被委派的 Agent

在 Stage 01 中，我们定义了角色契约，但 Agent 之间交换的请求和结果还没有统一的消息结构。

如果任何 Agent 都能回复一个关联 ID 匹配的消息，那么 reviewer 可以冒充 researcher 提交结果，导致任务被错误地标记为完成。

本阶段需要实现 create_message 和 correlate 函数，确保响应消息的发送者与原始请求的接收者一致，并且消息类型受限。

先做判断

在实现 correlate 函数时，以下哪个条件对于防止响应伪造是必要的？

响应消息的 correlation_id 必须与请求相同
响应消息的 sender 必须等于请求的 recipient
响应消息的 recipient 必须等于请求的 sender
响应消息的 kind 必须是 result 或 error

判断依据: 正确答案是第二个选项。虽然所有条件都是有效关联的一部分，但防止伪造的关键是验证响应发送者身份：响应必须来自请求的接收者。如果只检查关联 ID 和接收者，攻击者可以伪造发送者。

它是什么

消息契约是什么

消息契约是一组规则，确保 Agent 之间的通信可关联、可验证。

每个消息包含 message_id、correlation_id、sender、recipient、kind 和 payload，其中 correlation_id 用于关联请求和响应。

correlate 函数验证响应是否合法：关联 ID 匹配、发送者和接收者身份反转、消息类型为 result 或 error。

它不是什么

消息契约不是什么

消息契约不是简单的数据容器，它强制实施身份和类型约束。

它不是任何 Agent 都可以随意回复的机制；响应必须来自被委派的 Agent。

它不保证消息内容正确，只保证消息来源和类型符合协议。

它与相邻概念的关系

与其他概念的关系

消息契约建立在 Stage 01 的角色契约之上，角色定义了谁可以做什么，消息契约定义了如何交换信息。

create_message 在创建时验证消息的合法性，correlate 在关联时验证响应身份，两者共同维护通信完整性。

消息类型限制（request、result、error、cancel）防止协议混乱，确保只有预期类型的消息被处理。

本阶段的边界决定: 在 correlate 中必须检查 response.sender == request.recipient，否则任何 Agent 都可以伪造响应；同时，create_message 必须拒绝未知类型和自发送消息。

实验 ID: multi-agent-orchestration-02-message-contracts
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/multi-agent-orchestration/02-message-contracts/
验收: 待验收

教师示范

目标: 演示 create_message 和 correlate 如何验证响应发送者身份。

假设 supervisor 向 researcher 发送一个请求，然后 researcher 返回结果。

我们使用 create_message 创建请求和响应，然后调用 correlate 验证它们是否匹配。

如果 reviewer 试图冒充 researcher 发送响应，correlate 应该返回 False。

实施步骤

创建请求消息：request = create_message("m1", "job-1", "supervisor", "researcher", "request", {})
创建合法响应：valid = create_message("m2", "job-1", "researcher", "supervisor", "result", {})
创建伪造响应：spoofed = create_message("m3", "job-1", "reviewer", "supervisor", "result", {})
验证：correlate(request, valid) 返回 True，correlate(request, spoofed) 返回 False。

真实 solution 代码: forge/multiagent/messages.py

from __future__ import annotations

from dataclasses import dataclass

@dataclass(frozen=True)
class AgentMessage:
    message_id: str
    correlation_id: str
    sender: str
    recipient: str
    kind: str
    payload: dict[str, object]

def create_message(message_id: str, correlation_id: str, sender: str, recipient: str, kind: str, payload: dict[str, object]) -> AgentMessage:
    values = [message_id, correlation_id, sender, recipient]
    if any(not value.strip() for value in values):
        raise ValueError("message identity fields are required")
    if sender == recipient:
        raise ValueError("sender and recipient must differ")
    if kind not in {"request", "result", "error", "cancel"}:
        raise ValueError("unsupported message kind")
    return AgentMessage(message_id, correlation_id, sender, recipient, kind, dict(payload))

def correlate(request: AgentMessage, response: AgentMessage) -> bool:
    return response.correlation_id == request.correlation_id and response.sender == request.recipient and response.recipient == request.sender and response.kind in {"result", "error"}

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

create_message 在创建时检查消息字段非空、发送者与接收者不同、消息类型合法。

correlate 检查关联 ID 匹配、发送者等于请求接收者、接收者等于请求发送者、消息类型为 result 或 error。

伪造响应中发送者是 reviewer，而请求接收者是 researcher，因此 correlate 返回 False。

带提示练习

补全 correlate 函数中缺失的发送者检查，使测试通过。

打开 forge/multiagent/messages.py 文件。
找到 correlate 函数，当前实现缺少发送者身份检查。
添加条件 response.sender == request.recipient。
运行测试验证。

逐级提示

检查响应发送者是否等于请求接收者。
确保所有条件用 and 连接。
不要改变其他条件。

完成标准

correlate(request, valid) 返回 True
correlate(request, spoofed) 返回 False
所有三个测试通过

故障注入与排查

在故障夹具 fixtures/failure.json 中，correlate 函数忽略了响应发送者身份，导致 test_response_must_reverse_sender_and_recipient 测试失败。具体地，当 reviewer 冒充 researcher 发送响应时，correlate 错误地返回 True，因为只检查了关联 ID 和接收者，而没有检查发送者。可执行故障夹具修改 forge/multiagent/messages.py，并由 test_response_must_reverse_sender_and_recipient 稳定复现。已验证机制标识为响应发送者和请求接收者。

本阶段已经学习了消息契约和身份验证机制，因此能够诊断出缺失的发送者检查。通过查看 correlate 函数的实现，可以识别出缺少响应发送者等于请求接收者这一条件，并修复它。

已验证的故障合同

目标测试: test_response_must_reverse_sender_and_recipient
故障文件: forge/multiagent/messages.py

诊断顺序

运行故障命令 python -m pytest -q ../tests/test_stage.py::test_response_must_reverse_sender_and_recipient，观察测试失败。
查看测试代码，发现 spoofed 消息的发送者是 reviewer，而请求接收者是 researcher。
检查 correlate 函数实现，发现缺少 response.sender == request.recipient 条件。
确认根因：关联只检查了关联 ID 和接收者，没有验证发送者身份。

修复原则: 在 correlate 函数中添加条件 response.sender == request.recipient，确保响应来自请求的接收者。修复后的函数应返回 response.correlation_id == request.correlation_id and response.sender == request.recipient and response.recipient == request.sender and response.kind in {"result", "error"}。

python -m pytest -q ../tests/test_stage.py::test_response_must_reverse_sender_and_recipient

迁移挑战

在音乐创作场景中，supervisor 向 researcher 发送一个请求，要求整理灵感。然后尝试用 reviewer 作为发送者创建响应，验证 correlate 返回 False。

改变的条件: 将场景从软件交付改为音乐创作，但保持消息结构和验证逻辑不变。

创建请求消息，其中 sender 为 supervisor，recipient 为 researcher，kind 为 request，correlation_id 为 music-1。然后创建响应消息，其中 sender 为 reviewer，recipient 为 supervisor，kind 为 result，correlation_id 为 music-1。调用 correlate 验证返回 False。

完成标准

请求消息的 sender 是 supervisor，recipient 是 researcher，correlation_id 是 music-1。
响应消息的 sender 是 reviewer，recipient 是 supervisor，correlation_id 是 music-1。
correlate(request, response) 返回 False。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么响应必须反转发送者和接收者？如果只检查关联 ID 和接收者，会有什么安全风险？

一个有力回答应包含

响应必须反转发送者和接收者，以确保响应来自请求的接收者，防止其他 Agent 伪造响应。
如果只检查关联 ID 和接收者，攻击者可以伪造发送者，导致任务被错误地标记为完成。
反转检查是消息契约的核心，它建立了请求-响应的身份绑定。

掌握证据

所有三个测试通过：test_response_must_reverse_sender_and_recipient、test_unknown_message_kind_is_rejected、test_self_addressed_message_is_rejected。
correlate 拒绝伪造发送者：当响应发送者不是请求接收者时返回 False。
create_message 拒绝未知类型和自发送消息：未知类型抛出 ValueError，发送者等于接收者时抛出 ValueError。

STAGE 03

通过乐观并发控制防止 Agent 静默覆盖彼此的状态更新

先看一个具体问题

两个 Agent 基于同一旧版本更新导致丢失更新

在办公研究场景中，研究员 Agent A 和 Agent B 同时读取共享状态，都看到版本号为 0。

Agent A 先写入 research 键并成功更新版本到 1，但 Agent B 仍基于版本 0 写入 review 键，覆盖了 Agent A 的更新。

最终状态中 research 键丢失，系统没有检测到任何冲突，导致数据不一致且难以追踪。

你需要实现一个 SharedState 类，要求更新时必须提供读取时的版本号，否则抛出冲突异常，从而防止静默覆盖。

先做判断

如果两个 Agent 都基于版本 0 读取状态，然后依次更新不同的键，会发生什么？

两个更新都会成功，最终状态包含两个键
第二个更新会抛出冲突异常，因为版本已变化
第二个更新会静默覆盖第一个更新，导致数据丢失
系统会自动合并两个更新，无需版本检查

判断依据: 正确答案是第二个更新会抛出冲突异常。乐观并发控制要求每次更新必须携带读取时的版本号，如果版本不匹配则拒绝更新，从而避免丢失更新。

它是什么

乐观并发控制

乐观并发控制假设冲突很少发生，因此不提前加锁，而是在更新时检查版本号是否与读取时一致。

如果版本一致，则应用更改并将版本号加一；如果不一致，则抛出冲突异常，由调用方决定重试或放弃。

这种机制允许多个 Agent 并发读取，只在写入时进行冲突检测，提高了并发性能。

它不是什么

不是悲观锁或直接更新

它不是在读取时锁定资源，也不是在更新时无条件覆盖现有值。

它不保证所有更新都会成功，而是要求调用方处理冲突异常。

它不能解决所有并发问题，例如需要原子性复合操作时可能需要更复杂的机制。

它与相邻概念的关系

与快照和版本号的关系

SharedState 维护内部版本号和值字典，read 方法返回一个不可变的 StateSnapshot，包含当前版本号和值的深拷贝。

update 方法接收期望版本号和更改字典，首先比较期望版本号与当前版本号，如果不匹配则抛出 RuntimeError。

如果匹配，则应用更改、递增版本号，并返回新的快照。快照的深拷贝确保外部修改不会影响内部状态。

本阶段的边界决定: 更新必须提供读取时的版本号，否则抛出冲突异常；快照必须隔离可变状态，防止外部修改影响内部数据。

实验 ID: multi-agent-orchestration-03-versioned-shared-state
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/multi-agent-orchestration/03-versioned-shared-state/
验收: 待验收

教师示范

目标: 演示 SharedState 的乐观并发控制：读取版本，更新，再次使用旧版本更新触发冲突。

我们创建一个 SharedState 实例，初始版本为 0。

首先调用 read 获取快照，其版本为 0。

然后调用 update(0, {“research”: “done”})，因为期望版本 0 与当前版本 0 匹配，更新成功，版本变为 1。

再次调用 update(0, {“review”: “done”})，期望版本 0 与当前版本 1 不匹配，抛出 RuntimeError(“shared state version conflict”)。

实施步骤

创建 SharedState 实例。
调用 read 获取初始快照，记录版本号。
使用该版本号调用 update 更新一个键，观察版本号递增。
再次使用旧版本号调用 update，观察抛出冲突异常。

真实 solution 代码: forge/multiagent/shared_state.py

from __future__ import annotations

from dataclasses import dataclass, field

@dataclass(frozen=True)
class StateSnapshot:
    version: int
    values: dict[str, object]

@dataclass
class SharedState:
    _version: int = 0
    _values: dict[str, object] = field(default_factory=dict)

    def read(self) -> StateSnapshot:
        return StateSnapshot(self._version, dict(self._values))

    def update(self, expected_version: int, changes: dict[str, object]) -> StateSnapshot:
        if expected_version != self._version:
            raise RuntimeError("shared state version conflict")
        self._values.update(changes)
        self._version += 1
        return self.read()

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

update 方法首先检查 expected_version 是否等于当前 _version，如果不相等则抛出 RuntimeError。

如果相等，则应用更改并将 _version 加一，然后返回新的快照。

这样，任何基于旧版本的更新都会被拒绝，从而防止丢失更新。

带提示练习

补全 update 方法中缺失的版本比较，使测试通过。

打开 forge/multiagent/shared_state.py 文件。
在 update 方法中，添加 if expected_version != self._version: 检查。
如果版本不匹配，抛出 RuntimeError(“shared state version conflict”)。
运行测试验证。

逐级提示

提示1：在应用更改之前检查版本。
提示2：使用 != 比较 expected_version 和 self._version。
提示3：抛出异常时消息中包含 ‘conflict’ 以匹配测试。

完成标准

test_update_requires_the_version_that_was_read 通过
test_successful_updates_increment_version 通过
test_snapshots_do_not_alias_mutable_state 通过

故障注入与排查

在故障实验室中，我们提供了一个有缺陷的 SharedState 实现，它忽略了 expected_version 参数，直接更新状态。运行测试 test_update_requires_the_version_that_was_read 会失败，因为第二次更新没有抛出冲突异常。你需要诊断并修复这个缺陷。可执行故障夹具修改 forge/multiagent/shared_state.py，并由 test_update_requires_the_version_that_was_read 稳定复现。已验证机制标识为 expected_version。

本阶段已经学习了乐观并发控制的基本原理，并实现了版本检查。故障实验室中的缺陷正是缺少版本比较，因此你可以通过添加 if expected_version != self._version: 检查来修复。

已验证的故障合同

目标测试: test_update_requires_the_version_that_was_read
故障文件: forge/multiagent/shared_state.py

诊断顺序

运行测试 test_update_requires_the_version_that_was_read，观察失败信息：DID NOT RAISE RuntimeError。
检查 update 方法，发现没有比较 expected_version 和 self._version。
确认缺陷：更新时忽略了版本检查，导致旧版本更新被接受。
修复：在 update 方法开头添加版本比较，如果不匹配则抛出 RuntimeError。

修复原则: 在 update 方法中添加版本检查：if expected_version != self._version: raise RuntimeError(“shared state version conflict”)。这样，任何基于旧版本的更新都会被拒绝，测试将通过。

python -m pytest -q ../tests/test_stage.py::test_update_requires_the_version_that_was_read

迁移挑战

在办公研究场景中，两个研究员同时读取状态版本 0，第一个更新 research 键成功，第二个尝试用版本 0 更新 review 键，预期抛出 RuntimeError。

改变的条件: 将场景从通用键值更新改为办公研究中的具体键（research 和 review），并模拟两个研究员并发操作。

编写一个测试或脚本，模拟两个研究员：研究员 A 读取状态版本 0，更新 research 键；研究员 B 也读取状态版本 0，然后尝试更新 review 键。验证研究员 B 的更新抛出 RuntimeError，且最终状态中 research 键存在，review 键不存在。

完成标准

研究员 B 的更新抛出 RuntimeError，消息包含 ‘conflict’。
最终状态中 research 键的值为 ‘done’，review 键不存在。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么需要版本号来防止丢失更新？请解释版本号在乐观并发控制中的作用，以及如果没有版本检查会发生什么。

一个有力回答应包含

版本号记录了状态的每次修改，更新时比较期望版本和当前版本可以检测到并发修改。
如果没有版本检查，后提交的更新会覆盖先提交的更新，导致数据丢失且无法察觉。
乐观并发控制通过版本号实现轻量级的冲突检测，避免使用锁带来的性能开销。

掌握证据

所有三个测试通过：test_update_requires_the_version_that_was_read、test_successful_updates_increment_version、test_snapshots_do_not_alias_mutable_state。
update 方法在版本不匹配时抛出 RuntimeError，防止丢失更新。
快照的深拷贝确保外部修改不会影响内部状态。

STAGE 04

对互斥提案应用确定性安全优先规则，并保留决议理由

先看一个具体问题

多个 Agent 提出互斥方案时，如何安全地选择？

在上一阶段，多个 Agent 通过共享状态交换了提案，但还没有决定采用哪一个。

现在有两个 Agent 提出了互斥方案：一个要删除数据（不可逆），另一个要归档数据（可逆）。

如果直接比较证据数量，删除方案可能因为证据更多而胜出，但这会带来不可逆的风险。

我们需要一个确定性的规则：不可逆提案必须显式批准才能参与，否则按风险等级和证据数量排序。

先做判断

在冲突解决中，如果有一个不可逆提案和一个可逆提案，且不可逆提案未获批准，你认为应该如何处理？

直接选择证据数量更多的提案
排除未批准的不可逆提案，然后选择风险最低的提案
随机选择一个提案
让 Agent 再次投票

判断依据: 正确答案是排除未批准的不可逆提案，然后选择风险最低的提案。因为不可逆操作一旦执行就无法撤销，必须显式批准才能参与；否则按风险等级和证据数量确定性排序。

它是什么

确定性安全优先规则

冲突解决是一个纯函数，输入提案列表和是否批准不可逆提案的标志，输出一个决议对象。

它首先过滤掉未批准的不可逆提案，然后在剩余提案中按风险等级升序、证据数量降序、Agent ID 升序排序，选择第一个作为胜者。

如果过滤后没有提案，则返回 reason 为 “approval_required” 的决议，表示需要批准不可逆提案才能继续。

它不是什么

不是简单的证据数量比较

它不是仅仅比较证据数量，因为风险等级更高的提案即使证据更多也不能自动胜出。

它也不是随机选择或让模型再次投票，而是基于明确规则的确定性排序。

它不会自动批准不可逆提案，除非调用者显式传入 approval_for_irreversible=True。

它与相邻概念的关系

与共享状态和消息的关系

冲突解决函数接收来自共享状态的提案列表，这些提案由不同 Agent 通过消息传递产生。

决议结果可以写回共享状态，供后续阶段使用，例如执行胜出的提案。

风险等级映射 RISK_ORDER 定义了 read_only、reversible、irreversible 的排序，确保低风险优先。

本阶段的边界决定: 当提案列表为空时返回 no_proposals；当过滤后为空时返回 approval_required；否则返回 lowest_risk_then_evidence 的胜者。

实验 ID: multi-agent-orchestration-04-conflict-resolution
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/multi-agent-orchestration/04-conflict-resolution/
验收: 待验收

教师示范

目标: 演示 resolve 如何排除未批准的不可逆提案并选择最低风险方案。

假设有两个提案：builder 提出删除数据（不可逆，证据 9），reviewer 提出归档数据（可逆，证据 2）。

调用 resolve 时未批准不可逆提案，因此删除提案被过滤掉，只剩归档提案。

归档提案风险为 reversible，证据 2，成为唯一候选，因此胜者是 reviewer。

实施步骤

创建 Proposal 列表，包含 builder 的不可逆删除提案和 reviewer 的可逆归档提案。
调用 resolve(proposals, approval_for_irreversible=False)。
函数内部过滤掉不可逆提案，因为 approval_for_irreversible 为 False。
剩余提案按风险排序，reversible 风险低于 irreversible，因此 reviewer 胜出。
返回 Resolution(winner=reviewer_proposal, reason=“lowest_risk_then_evidence”)。

真实 solution 代码: forge/multiagent/conflicts.py

from __future__ import annotations

from dataclasses import dataclass

@dataclass(frozen=True)
class Proposal:
    agent_id: str
    value: object
    risk: str
    evidence_count: int

@dataclass(frozen=True)
class Resolution:
    winner: Proposal | None
    reason: str

RISK_ORDER = {"read_only": 0, "reversible": 1, "irreversible": 2}

def resolve(proposals: list[Proposal], approval_for_irreversible: bool = False) -> Resolution:
    if not proposals:
        return Resolution(None, "no_proposals")
    eligible = [item for item in proposals if item.risk != "irreversible" or approval_for_irreversible]
    if not eligible:
        return Resolution(None, "approval_required")
    winner = sorted(eligible, key=lambda item: (RISK_ORDER[item.risk], -item.evidence_count, item.agent_id))[0]
    return Resolution(winner, "lowest_risk_then_evidence")

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

过滤条件 item.risk != "irreversible" or approval_for_irreversible 确保未批准的不可逆提案被排除。

排序键 (RISK_ORDER[item.risk], -item.evidence_count, item.agent_id) 实现风险优先、证据次之、Agent ID 作为最终平局裁决。

带提示练习

补全 resolve 函数中缺失的不可逆提案过滤，使测试通过。

打开 forge/multiagent/conflicts.py 文件。
找到 resolve 函数中的 eligible 赋值行。
将 eligible = list(proposals) 改为 eligible = [item for item in proposals if item.risk != "irreversible" or approval_for_irreversible]。
运行测试 python -m pytest -q tests/test_stage.py 验证。

逐级提示

注意过滤条件需要同时考虑风险类型和 approval_for_irreversible 标志。
如果过滤后列表为空，应返回 approval_required。
排序逻辑已经存在，不需要修改。

完成标准

test_unapproved_irreversible_proposal_cannot_win 通过
test_all_irreversible_proposals_require_approval 通过
test_equal_risk_prefers_more_evidence 通过

故障注入与排查

在故障实验室中，我们提供了一个有缺陷的 conflicts.py 文件，其中 resolve 函数没有过滤未批准的不可逆提案。运行测试 test_all_irreversible_proposals_require_approval 会失败，因为所有不可逆提案都被包含，导致返回了一个胜者而不是一个表示需要批准的决议。你需要诊断并修复这个缺陷。可执行故障夹具修改 forge/multiagent/conflicts.py，并由 test_all_irreversible_proposals_require_approval 稳定复现。已验证机制标识为 approval_required、approval_for_irreversible。

本阶段学习的核心机制就是不可逆提案的过滤规则。通过理解 resolve 函数的预期行为，你可以识别出缺陷在于缺少过滤步骤，并应用正确的修复。

已验证的故障合同

目标测试: test_all_irreversible_proposals_require_approval
故障文件: forge/multiagent/conflicts.py

诊断顺序

运行故障测试命令，观察 AssertionError：result.winner 不是 None。
检查 resolve 函数实现，发现 eligible = list(proposals) 没有过滤不可逆提案。
确认当所有提案都是不可逆且未批准时，应该返回 approval_required。
修复：添加过滤条件，排除未批准的不可逆提案。

修复原则: 将 eligible = list(proposals) 替换为 eligible = [item for item in proposals if item.risk != "irreversible" or approval_for_irreversible]，这样未批准的不可逆提案会被排除，如果过滤后为空则返回 approval_required。

python -m pytest -q ../tests/test_stage.py::test_all_irreversible_proposals_require_approval

迁移挑战

在办公场景中，两个 Agent 提出不同的文档保留策略：合规官提出永久保留（不可逆，证据 7），档案员提出保留 7 年后归档（可逆，证据 3）。未批准不可逆提案时，resolve 应选择档案员的提案。

改变的条件: 场景领域从软件交付变为办公文档保留，但风险等级（irreversible 与 reversible）和证据数量保持不变，且冲突解决逻辑无需修改即可复用。

编写一个测试函数，创建两个 Proposal：合规官的永久保留（risk=“irreversible”, evidence_count=7）和档案员的保留 7 年后归档（risk=“reversible”, evidence_count=3）。调用 resolve 且 approval_for_irreversible=False，断言胜者的 agent_id 为 “archivist”，且 reason 为 “lowest_risk_then_evidence”。

完成标准

测试中合规官的提案风险为 irreversible 且未获批准，因此被过滤。
档案员的提案风险为 reversible，成为唯一候选，胜者 agent_id 为 “archivist”。
测试断言 reason 为 “lowest_risk_then_evidence”，并且测试在现有 resolve 实现下通过。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么不可逆提案需要显式批准？请解释风险等级和证据数量如何共同决定胜者，并说明这种设计的边界。

一个有力回答应包含

不可逆操作一旦执行就无法撤销，因此必须由有权力的实体显式批准，防止自动决策造成不可挽回的损失。
风险等级是首要排序键，低风险优先；证据数量是次要排序键，在风险相同时选择证据更多的提案。
边界：如果所有提案都是不可逆且未批准，则返回 approval_required，强制人工介入。

掌握证据

所有三个测试通过：test_unapproved_irreversible_proposal_cannot_win、test_all_irreversible_proposals_require_approval、test_equal_risk_prefers_more_evidence。
resolve 在未批准不可逆提案时返回 approval_required，且胜者为 None。
resolve 在风险相同时选择证据更多的提案，例如 test_equal_risk_prefers_more_evidence 中 agent b 胜出。

STAGE 05

按能力、权限和当前负载选择最小权限 Agent，无法安全委派时明确拒绝

先看一个具体问题

如何选择既能完成任务又拥有最少权限的 Agent？

当前系统中，多个 Agent 角色拥有不同权限集合，例如 reader 只有 read 权限，admin 拥有 read、write、process 权限。

当收到一个只读任务时，如果仅按当前负载选择，空闲的 admin 会被选中，导致不必要的权限暴露。

我们需要实现一个路由函数 choose_worker，在满足能力要求的前提下，优先选择权限最少的 Agent，以降低安全风险。

先做判断

在多个 Agent 都能完成任务时，你认为应该优先选择哪个 Agent？

负载最低的 Agent
权限最少的 Agent
角色 ID 最小的 Agent
随机选择一个 Agent

判断依据: 正确答案是“权限最少的 Agent”。因为最小权限原则要求只授予完成任务所必需的权限，减少潜在的安全风险。负载和角色 ID 仅用于打破平局。

它是什么

最小权限路由是什么

最小权限路由是一种选择策略：在所有能够满足任务所需能力的 Agent 中，优先选择权限集合最小的那个。

它通过比较每个 Agent 的能力集合大小（即权限广度）来排序，确保选中的 Agent 不会拥有超出任务需要的额外权限。

它不是什么

最小权限路由不是什么

它不是简单的负载均衡：负载最低的 Agent 可能拥有过多权限，即使它当前空闲，也不应被选中执行低权限任务。

它也不是随机选择或按角色 ID 排序：这些方法可能忽略权限差异，导致不必要的权限暴露。

它与相邻概念的关系

与其他概念的关系

最小权限路由建立在角色契约（Stage 01）之上：每个 Agent 角色明确定义了其能力集合，路由函数依赖这些能力集合进行筛选和排序。

它与冲突解决（Stage 04）互补：冲突解决处理多个 Agent 之间的意见分歧，而路由负责在任务开始前选择合适的执行者。

本阶段的边界决定: 选择必须首先最小化权限广度，然后才考虑负载和角色 ID 打破平局。

实验 ID: multi-agent-orchestration-05-supervisor-routing
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/multi-agent-orchestration/05-supervisor-routing/
验收: 待验收

教师示范

目标: 演示 choose_worker 如何在 reader 和 admin 之间选择 reader 执行只读任务。

假设我们有两个 WorkerStatus 对象：reader 角色拥有 read 权限，负载为 2；admin 角色拥有 read、write、process 权限，负载为 0。

任务需要 read 能力。两个 Agent 都满足能力要求，但 reader 的权限集合更小（只有 read），因此应该选择 reader。

在实现中，我们首先筛选出所有满足能力的候选者，然后使用 min 函数，以权限数量、负载、角色 ID 作为排序键，从而选出 reader。

实施步骤

定义 WorkerStatus 数据类，包含 role 和 active_tasks 字段。
在 choose_worker 中，筛选出 required_capabilities 是 worker.role.capabilities 子集的候选者。
如果没有候选者，抛出 ValueError。
使用 min 函数，key 为 (len(worker.role.capabilities), worker.active_tasks, worker.role.role_id)，返回选中的 role。

真实 solution 代码: forge/multiagent/supervisor.py

from __future__ import annotations

from dataclasses import dataclass
from .roles import AgentRole

@dataclass(frozen=True)
class WorkerStatus:
    role: AgentRole
    active_tasks: int

def choose_worker(required_capabilities: set[str], workers: list[WorkerStatus]) -> AgentRole:
    candidates = [worker for worker in workers if required_capabilities <= set(worker.role.capabilities)]
    if not candidates:
        raise ValueError("no worker can satisfy the delegation")
    chosen = min(candidates, key=lambda worker: (len(worker.role.capabilities), worker.active_tasks, worker.role.role_id))
    return chosen.role

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

筛选步骤确保只有具备所需能力的 Agent 才会被考虑，避免选择无法完成任务的 Agent。

排序键中，权限数量（len(worker.role.capabilities)）是第一优先级，因此权限最少的 Agent 会被选中。

负载和角色 ID 作为后续键，用于在权限数量相同时打破平局，保证选择的确定性。

带提示练习

补全 choose_worker 中缺失的权限广度排序，使测试通过。

打开 forge/multiagent/supervisor.py 文件。
找到 choose_worker 函数中的 min 调用。
修改 key 参数，使其首先按权限数量排序：key=lambda worker: (len(worker.role.capabilities), worker.active_tasks, worker.role.role_id)。
运行测试 python -m pytest -q tests/test_stage.py 验证。

逐级提示

提示 1：查看当前 key 参数，它只考虑了 active_tasks 和 role_id。
提示 2：权限数量可以通过 len(worker.role.capabilities) 获得。
提示 3：确保权限数量是排序键的第一个元素。

完成标准

test_supervisor_prefers_least_privileged_capable_worker 通过。
test_load_breaks_tie_between_equally_privileged_workers 通过。
test_missing_capability_is_refused 通过。

故障注入与排查

在故障夹具中，choose_worker 的实现只按负载排序，忽略了权限广度。当任务需要 read 能力时，空闲的 admin（负载 0）被选中，而不是权限更少的 reader（负载 2）。这违反了最小权限原则，导致不必要的权限暴露。

本阶段的核心就是实现最小权限路由。通过分析故障夹具中的代码，我们可以发现排序键缺少权限数量，从而定位缺陷并修复。修复后，测试 test_supervisor_prefers_least_privileged_capable_worker 将通过。

已验证的故障合同

目标测试: test_supervisor_prefers_least_privileged_capable_worker
故障文件: forge/multiagent/supervisor.py

诊断顺序

运行故障命令：python -m pytest -q ../tests/test_stage.py::test_supervisor_prefers_least_privileged_capable_worker，观察断言失败。
查看 forge/multiagent/supervisor.py 中的 choose_worker 函数，发现 min 的 key 参数为 (worker.active_tasks, worker.role.role_id)。
注意到缺少权限数量排序，导致负载低的 admin 被选中。
修改 key 参数，加入 len(worker.role.capabilities) 作为第一排序键。

修复原则: 将 min 的 key 参数改为 lambda worker: (len(worker.role.capabilities), worker.active_tasks, worker.role.role_id)，确保权限最少的 Agent 优先被选中。

python -m pytest -q ../tests/test_stage.py::test_supervisor_prefers_least_privileged_capable_worker

迁移挑战

在音乐创作场景中，有三个 Agent：reader（权限 read，负载 2）、editor（权限 read/write，负载 1）和 admin（权限 read/write/process，负载 0）。任务需要 read 能力。

改变的条件: 场景从软件交付变为音乐创作，并引入第三个 Agent：editor（权限 read/write，负载 1）。任务需要 read 能力，此时 reader 和 editor 都满足能力，但 reader 权限更少，应被选中。

使用 choose_worker 函数，传入 required_capabilities={‘read’} 和包含 reader、editor、admin 的 WorkerStatus 列表，验证返回的角色 ID 是 ‘reader’。

完成标准

调用 choose_worker({‘read’}, [admin, editor, reader]) 返回的角色 ID 为 ‘reader’。
调用 choose_worker({‘write’}, [admin, editor, reader]) 返回的角色 ID 为 ‘editor’，因为 editor 权限少于 admin。
调用 choose_worker({‘process’}, [admin, editor, reader]) 返回的角色 ID 为 ‘admin’，因为只有 admin 满足能力。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么最小权限优先于负载均衡？请解释这种设计在安全性和资源利用之间的权衡。

一个有力回答应包含

最小权限原则减少攻击面：即使 Agent 被攻破，攻击者获得的权限也仅限于任务所需。
负载均衡是性能优化，但不应以牺牲安全为代价；在权限相同的情况下才考虑负载。

掌握证据

所有三个测试通过：test_supervisor_prefers_least_privileged_capable_worker、test_load_breaks_tie_between_equally_privileged_workers、test_missing_capability_is_refused。
choose_worker 在多个候选者中选择权限最少的 Agent，例如在 reader 和 admin 之间选择 reader。
当没有 Agent 满足能力要求时，choose_worker 抛出 ValueError，拒绝不安全委派。

STAGE 06

根据必需角色和最小成功数判断任务可继续、降级或失败

先看一个具体问题

当部分 Agent 失败时，如何决定协作是成功、降级还是失败？

在 Stage 05 中，Supervisor 已经能够将任务路由给不同的 Agent，但还没有定义当部分 Agent 失败时如何评估整体结果。

现在需要实现一个评估函数 assess_outcomes，它接收每个 Agent 的结果、必需角色集合和最小成功数，然后返回一个决策对象，状态为 complete、degraded 或 failed。

一个常见的错误是只检查成功数量是否达到阈值，而忽略了某些角色是必需的，即使数量足够，缺少必需角色也必须判定为失败。

本阶段将实现这个策略，并通过测试验证：必需角色失败导致整体失败，可选角色失败导致降级，全部成功才是完整。

先做判断

如果 required_roles 包含 builder，但 builder 失败了，而其他两个角色成功，且 minimum_successes=2，assess_outcomes 应该返回什么状态？

complete，因为成功数量达到了阈值
degraded，因为有一个角色失败
failed，因为必需角色 builder 失败
failed，因为成功数量不足

判断依据: 正确答案是 failed，因为必需角色失败必须导致整体失败，即使成功数量达到了阈值。这体现了角色语义优先于数量阈值的策略。

它是什么

部分失败评估策略

assess_outcomes 是一个纯函数，它根据三个输入（结果列表、必需角色集合、最小成功数）计算协作状态。

它首先收集所有成功角色的集合，然后计算缺失的必需角色（必需角色减去成功角色）。

如果缺失的必需角色非空，或者成功数量小于最小成功数，则返回 failed 状态，并附带缺失的必需角色元组。

否则，如果存在任何失败的角色（即结果状态不是 succeeded），则返回 degraded 状态。

如果所有角色都成功，则返回 complete 状态。

它不是什么

不是简单的数量阈值

这个策略不是仅仅比较成功数量和 minimum_successes；它必须优先检查必需角色是否全部成功。

它也不是对每个角色一视同仁；必需角色和可选角色在失败时的影响不同。

它不会因为成功数量足够而忽略必需角色的缺失；必需角色的失败是致命的。

它不会在存在失败角色时返回 complete；只有全部成功才是完整。

它与相邻概念的关系

与 Supervisor 和消息的关系

assess_outcomes 被 Supervisor 在收集所有 Agent 结果后调用，以决定下一步行动。

它依赖于 AgentOutcome 数据类，该数据类包含 role_id 和 status 字段。

它返回 CollaborationDecision 数据类，包含 status 和 missing_required 字段。

这个策略是 Stage 05 路由逻辑的补充，为 Supervisor 提供了失败处理能力。

本阶段的边界决定: 当必需角色失败时，无论成功数量多少，都必须返回 failed；当只有可选角色失败且成功数量达到阈值时，返回 degraded；当所有角色成功时，返回 complete。

实验 ID: multi-agent-orchestration-06-partial-failure-policy
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/multi-agent-orchestration/06-partial-failure-policy/
验收: 待验收

教师示范

目标: 演示 assess_outcomes 如何因必需角色失败而返回 failed。

假设有三个 Agent 结果：researcher 成功、builder 失败、reviewer 成功。

必需角色集合是 {"builder"}，最小成功数是 2。

虽然成功数量为 2，达到了阈值，但必需角色 builder 失败了，所以整体状态必须是 failed。

这个例子展示了必需角色检查优先于数量阈值。

实施步骤

创建结果列表：[AgentOutcome("researcher", "succeeded"), AgentOutcome("builder", "failed"), AgentOutcome("reviewer", "succeeded")]。
调用 assess_outcomes(outcomes, {"builder"}, minimum_successes=2)。
函数内部计算成功角色集合为 {"researcher", "reviewer"}。
计算缺失的必需角色：{"builder"} - {"researcher", "reviewer"} = {"builder"}，非空。
因此返回 CollaborationDecision("failed", ("builder",))。

真实 solution 代码: forge/multiagent/resilience.py

from __future__ import annotations

from dataclasses import dataclass

@dataclass(frozen=True)
class AgentOutcome:
    role_id: str
    status: str

@dataclass(frozen=True)
class CollaborationDecision:
    status: str
    missing_required: tuple[str, ...]

def assess_outcomes(outcomes: list[AgentOutcome], required_roles: set[str], minimum_successes: int) -> CollaborationDecision:
    successful = {item.role_id for item in outcomes if item.status == "succeeded"}
    missing = tuple(sorted(required_roles - successful))
    if missing or len(successful) < minimum_successes:
        return CollaborationDecision("failed", missing)
    if any(item.status != "succeeded" for item in outcomes):
        return CollaborationDecision("degraded", ())
    return CollaborationDecision("complete", ())

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

函数首先检查缺失的必需角色，如果非空则立即返回 failed，不会继续检查数量阈值。

这确保了必需角色的失败不会被其他成功所掩盖。

返回的 missing_required 元组包含了所有缺失的必需角色，便于 Supervisor 了解哪些关键角色未完成。

带提示练习

补全 assess_outcomes 中缺失的必需角色检查，使测试通过。

打开 forge/multiagent/resilience.py 文件。
找到 assess_outcomes 函数，当前实现可能只检查了成功数量。
在函数开头添加对缺失必需角色的检查：计算 missing = tuple(sorted(required_roles - successful))，如果 missing 非空，则返回 CollaborationDecision("failed", missing)。
确保在检查数量阈值之前进行此检查。
运行测试 python -m pytest -q tests/test_stage.py 验证所有测试通过。

逐级提示

提示 1：使用集合差集 required_roles - successful 来找出缺失的必需角色。
提示 2：如果 missing 非空，直接返回 failed，不要继续执行后续逻辑。
提示 3：注意 missing 应该是一个排序后的元组，以保持确定性。

完成标准

test_missing_required_role_fails_even_when_quorum_is_met 通过。
test_optional_failure_produces_degraded_success 通过。
test_all_success_is_complete 通过。

故障注入与排查

在 fixtures/failure.json 中定义了一个故障场景：assess_outcomes 只检查成功数量是否达到阈值，而忽略了必需角色的失败。运行测试 test_missing_required_role_fails_even_when_quorum_is_met 会失败，因为当 builder 失败但成功数量达到 2 时，函数错误地返回了 degraded 而不是 failed。可执行故障夹具修改 forge/multiagent/resilience.py，并由 test_missing_required_role_fails_even_when_quorum_is_met 稳定复现。

本阶段已经学习了必需角色和最小成功数的概念，并且知道必需角色失败必须导致整体失败。通过检查 assess_outcomes 的实现，可以定位到缺少必需角色检查的问题，并修复它。

已验证的故障合同

目标测试: test_missing_required_role_fails_even_when_quorum_is_met
故障文件: forge/multiagent/resilience.py

诊断顺序

运行测试 python -m pytest -q ../tests/test_stage.py::test_missing_required_role_fails_even_when_quorum_is_met，观察断言失败：期望 failed 但得到 degraded。
检查 assess_outcomes 函数，发现它只检查了 len(successful) < minimum_successes，没有检查必需角色是否缺失。
确认当 builder 失败时，successful 集合不包含 builder，但成功数量为 2，满足阈值，因此函数返回了 degraded。
修复方法是在函数开头添加对缺失必需角色的检查，如果缺失则返回 failed。

修复原则: 在 assess_outcomes 函数中，计算 missing = tuple(sorted(required_roles - successful))，并在检查数量阈值之前添加 if missing: return CollaborationDecision("failed", missing)。这样，当任何必需角色缺失时，函数会立即返回 failed，而不会错误地返回 degraded。

python -m pytest -q ../tests/test_stage.py::test_missing_required_role_fails_even_when_quorum_is_met

迁移挑战

在软件交付场景中，有三个 Agent：researcher、builder 和 reviewer。builder 是必需角色，minimum_successes=2。现在 researcher 和 reviewer 成功，但 builder 失败。

改变的条件: 将场景从通用角色改为软件交付角色，并指定 builder 为必需角色，同时保持成功数量为 2。

使用 assess_outcomes 评估这个场景，并验证返回的状态是 failed，且 missing_required 包含 builder。

完成标准

调用 assess_outcomes 并传入结果列表 [AgentOutcome("researcher", "succeeded"), AgentOutcome("builder", "failed"), AgentOutcome("reviewer", "succeeded")]，必需角色集合 {"builder"}，最小成功数 2。
断言返回的决策状态为 failed，并且 missing_required 等于 ("builder",)。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么必需角色失败不能通过其他成功来弥补？请解释这种设计在协作系统中的因果权衡。

一个有力回答应包含

必需角色通常承担不可替代的关键功能，例如构建或核心实现，缺少它会导致最终产品无法交付。
如果允许其他成功弥补必需角色的失败，系统可能会错误地认为任务完成，但实际上关键部分缺失，导致后续阶段失败或交付不完整。
这种设计牺牲了灵活性，但保证了关键路径的可靠性，避免了在缺少核心组件时错误地宣布成功。

掌握证据

所有三个测试通过：test_missing_required_role_fails_even_when_quorum_is_met、test_optional_failure_produces_degraded_success、test_all_success_is_complete。
assess_outcomes 在必需角色缺失时返回 failed，即使成功数量达到阈值。
assess_outcomes 在可选角色失败且成功数量达到阈值时返回 degraded。

STAGE 07

组合最小权限路由、消息关联、共享状态和部分失败策略，生成可审计协作结果

先看一个具体问题

为什么顺序执行任务还不够？

在 Stage 06 中，你已经实现了部分失败策略，但还没有一个统一的协调器来串联所有机制。

当前代码中不存在 forge/multiagent/orchestrator.py，因此测试在导入 WorkItem 和 coordinate 时直接失败。

你需要创建一个协调器，它必须为每个工作项选择最小权限代理、记录必需角色、更新共享状态，并根据结果评估整体状态。

如果只是按顺序调用处理函数而不跟踪必需角色，那么当必需代理失败时，运行会被错误地降级而不是失败。

先做判断

在实现 coordinate 之前，请预测：如果某个必需工作项的处理函数抛出异常，但协调器没有记录该角色为必需角色，最终运行状态会是什么？

运行状态为 failed，因为异常被捕获并记录为失败。
运行状态为 degraded，因为失败的角色没有被标记为必需，评估逻辑将其视为可选失败。
运行状态为 complete，因为异常被忽略，其他工作项成功。
运行会直接崩溃，因为异常未被捕获。

判断依据: 正确答案是 degraded。因为 assess_outcomes 根据 required_roles 集合判断是否必须失败；如果集合为空，任何失败都只会导致降级。

它是什么

协调器是一个有状态的工作流引擎

协调器接收工作项列表、可用代理列表和处理函数映射，然后按顺序处理每个工作项。

对于每个工作项，它使用 choose_worker 根据所需能力选择最小权限代理，并记录分配结果。

如果工作项被标记为必需，协调器必须将所选代理的角色 ID 加入 required_roles 集合。

处理成功后，协调器将输出写入共享状态，并记录一个成功结果；处理失败则记录一个失败结果。

最后，协调器调用 assess_outcomes 评估整体状态，并返回包含状态、分配和状态版本的 CollaborationRun。

它不是什么

协调器不是简单的顺序执行器

协调器不能忽略工作项的必需性；必需角色的失败必须导致整个运行失败，而不是降级。

协调器不能跳过共享状态的更新；每个成功的工作项输出都必须提交到状态中，以便后续审计。

协调器不能自行决定代理选择逻辑；它必须依赖 choose_worker 来保证最小权限路由。

协调器不能在没有可用代理时静默分配；如果找不到满足能力的代理，它必须抛出 ValueError。

它与相邻概念的关系

协调器如何组合已有组件

choose_worker 来自 supervisor.py，它根据能力集合选择最小权限代理，如果找不到则抛出 ValueError。

SharedState 来自 shared_state.py，它维护版本化的键值存储，每次更新都会递增版本号。

assess_outcomes 来自 resilience.py，它根据结果列表和必需角色集合决定最终状态。

AgentOutcome 是结果的数据结构，包含角色 ID 和状态（succeeded 或 failed）。

协调器将这些组件串联起来，形成一个可审计的端到端运行。

本阶段的边界决定: 协调器必须记录每个必需工作项所选代理的角色 ID，并在最终评估时传递该集合；否则，必需代理的失败会被错误地视为可选失败。

实验 ID: multi-agent-orchestration-07-coordinated-run
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/multi-agent-orchestration/07-coordinated-run/
验收: 待验收

教师示范

目标: 演示 coordinate 如何路由工作项、提交结果并返回完整状态。

假设我们有两个工作项：evidence 需要 read 能力且为必需，patch 需要 write 能力且为必需。

可用代理有 researcher（具备 read 和 search）和 builder（具备 read、write 和 execute）。

处理函数映射为：researcher 返回 "sources"，builder 返回 "diff"。

协调器首先处理 evidence，choose_worker 会选择 researcher，因为它是满足 read 能力的最小权限代理。

然后处理 patch，choose_worker 会选择 builder，因为只有它具备 write 能力。

两个工作项都成功，状态更新两次，最终状态版本为 2，运行状态为 complete。

实施步骤

创建 WorkItem 列表，包含 evidence 和 patch。
创建 workers 列表，包含 researcher 和 builder 的 WorkerStatus。
定义 handlers 字典，将角色 ID 映射到处理函数。
调用 coordinate(items, workers, handlers)。
断言返回的 CollaborationRun 状态为 complete，分配为 (('evidence', 'researcher'), ('patch', 'builder'))，状态版本为 2。

真实 solution 代码: forge/multiagent/orchestrator.py

from __future__ import annotations

from dataclasses import dataclass
from typing import Callable
from .resilience import AgentOutcome, assess_outcomes
from .roles import AgentRole
from .shared_state import SharedState
from .supervisor import WorkerStatus, choose_worker

@dataclass(frozen=True)
class WorkItem:
    item_id: str
    required_capabilities: frozenset[str]
    required: bool

@dataclass(frozen=True)
class CollaborationRun:
    status: str
    assignments: tuple[tuple[str, str], ...]
    state_version: int

def coordinate(items: list[WorkItem], workers: list[WorkerStatus], handlers: dict[str, Callable[[WorkItem], object]]) -> CollaborationRun:
    state = SharedState()
    outcomes: list[AgentOutcome] = []
    required_roles: set[str] = set()
    assignments: list[tuple[str, str]] = []
    for item in items:
        role = choose_worker(set(item.required_capabilities), workers)
        assignments.append((item.item_id, role.role_id))
        if item.required:
            required_roles.add(role.role_id)
        try:
            output = handlers[role.role_id](item)
            snapshot = state.read()
            state.update(snapshot.version, {item.item_id: output})
            outcomes.append(AgentOutcome(role.role_id, "succeeded"))
        except Exception:
            outcomes.append(AgentOutcome(role.role_id, "failed"))
    decision = assess_outcomes(outcomes, required_roles, minimum_successes=len(required_roles))
    return CollaborationRun(decision.status, tuple(assignments), state.read().version)

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

choose_worker 根据能力集合选择最小权限代理：对于 read，researcher 比 builder 权限更小，因此被选中。

每次成功处理后，协调器调用 state.update 提交输出，状态版本递增。

最终 assess_outcomes 收到两个成功结果和包含两个角色的必需集合，因此返回 complete。

带提示练习

补全 coordinate 中缺失的必需角色记录，使测试通过。

在 forge/multiagent/orchestrator.py 中创建 coordinate 函数的基本框架。
在循环内，使用 choose_worker 选择代理并记录分配。
检查当前工作项的 required 字段，如果为真，将所选代理的角色 ID 加入 required_roles 集合。
调用处理函数，捕获异常并记录结果。
最后调用 assess_outcomes 并返回 CollaborationRun。

逐级提示

提示 1：required_roles 是一个集合，使用 add 方法添加角色 ID。
提示 2：在调用处理函数之前添加角色记录，这样即使处理失败，角色也已被标记为必需。
提示 3：确保 assess_outcomes 的 minimum_successes 参数等于 len(required_roles)。

完成标准

test_required_agent_failure_fails_the_run 通过，运行状态为 failed。
test_orchestrator_routes_by_least_privilege_and_commits_results 通过，状态版本为 2。
test_missing_capability_is_not_silently_assigned 通过，抛出 ValueError。

故障注入与排查

在故障夹具 fixtures/failure.json 中，forge/multiagent/orchestrator.py 的 coordinate 函数缺少记录必需角色的逻辑。当必需代理 builder 在处理 patch 工作项时抛出异常，协调器没有将 builder 加入 required_roles，导致 assess_outcomes 认为没有必需角色失败，最终返回 degraded 而不是 failed。

本阶段要求组合最小权限路由、消息关联、共享状态和部分失败策略。故障夹具中的缺陷正是缺少必需角色记录，这是协调器实现的关键部分。通过本阶段的学习，你能够识别并修复这个缺陷，确保必需代理的失败导致整个运行失败。

已验证的故障合同

目标测试: test_required_agent_failure_fails_the_run
故障文件: forge/multiagent/orchestrator.py

诊断顺序

运行故障命令 python -m pytest -q ../tests/test_stage.py::test_required_agent_failure_fails_the_run，观察断言失败：期望 failed 但得到 degraded。
检查 coordinate 函数，发现循环内没有根据 item.required 将角色加入 required_roles。
查看 assess_outcomes 的实现，确认它依赖 required_roles 集合来决定是否必须失败。
修复：在调用处理函数之前，如果 item.required 为真，则执行 required_roles.add(role.role_id)。

修复原则: 在 coordinate 函数的循环内，添加条件判断：如果 item.required 为真，则将所选代理的角色 ID 加入 required_roles 集合。这样，当必需代理失败时，assess_outcomes 会返回 failed。

python -m pytest -q ../tests/test_stage.py::test_required_agent_failure_fails_the_run

迁移挑战

办公研究：把证据收集和简报生成作为两个必需工作项交给协调器完成。

改变的条件: 场景和工作项数据从软件交付改为办公研究；必需工作项策略与 coordinate API 保持不变。

使用现有的 WorkItem、WorkerStatus、AgentRole 和 coordinate API，创建包含两个必需工作项的运行：collect_evidence（需要 read）与 generate_brief（需要 write）。两个处理器都成功时，协调器必须返回 complete，且 state_version 为 2。

完成标准

分配结果必须是 (collect_evidence, researcher) 和 (generate_brief, builder)。
结果 status 必须为 complete，state_version 必须为 2。
使用现有 tests/test_stage.py 和验证命令；不得新增 test_transfer.py 或新的 API。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么必须记录必需角色？如果省略这个步骤，在什么情况下会导致错误的运行状态？

一个有力回答应包含

记录必需角色是为了让 assess_outcomes 知道哪些角色的失败必须导致整个运行失败。
如果省略，当必需代理失败时，评估逻辑会将其视为可选失败，返回 degraded 而不是 failed，从而掩盖关键错误。
这违反了部分失败策略的语义：必需工作项的失败必须阻止运行成功。

掌握证据

所有三个测试通过：test_orchestrator_routes_by_least_privilege_and_commits_results、test_required_agent_failure_fails_the_run、test_missing_capability_is_not_silently_assigned。
coordinate 在成功场景下返回 complete，分配正确，状态版本为 2。
coordinate 在必需代理失败时返回 failed，状态版本为 1。
coordinate 在找不到满足能力的代理时抛出 ValueError，错误信息包含 no worker。

完成本章

增加研究、实现和评审角色，并证明何时多 Agent 不值得使用。

FORGE / LOCAL CHECK

本地实验自检

未开始
2阅读中
3实验已下载
4测试结果已读取
5本地自检通过

verification.json 只在当前浏览器中解析，不会上传。这里验证的是实验合同，不是服务器认证或第三方背书。

选择 verification.json

主要层

依赖能力

为后续提供

受这些层约束

概念校准

子 Agent

规划

验证与评测

多 Agent 协作与任务编排

先看一个具体问题

委派时权限失控

先做判断

它是什么

角色合同

它不是什么

不是随意授权

它与相邻概念的关系

角色与委派的关系

先看一个具体问题

响应必须来自被委派的 Agent

先做判断

它是什么

消息契约是什么

它不是什么

消息契约不是什么

它与相邻概念的关系

与其他概念的关系

先看一个具体问题

两个 Agent 基于同一旧版本更新导致丢失更新

先做判断

它是什么

乐观并发控制

它不是什么

不是悲观锁或直接更新

它与相邻概念的关系

与快照和版本号的关系

先看一个具体问题

多个 Agent 提出互斥方案时，如何安全地选择？

先做判断

它是什么

确定性安全优先规则

它不是什么

不是简单的证据数量比较

它与相邻概念的关系

与共享状态和消息的关系

先看一个具体问题

如何选择既能完成任务又拥有最少权限的 Agent？

先做判断

它是什么

最小权限路由是什么

它不是什么

最小权限路由不是什么

它与相邻概念的关系

与其他概念的关系

先看一个具体问题

当部分 Agent 失败时，如何决定协作是成功、降级还是失败？

先做判断

它是什么

部分失败评估策略

它不是什么

不是简单的数量阈值

它与相邻概念的关系

与 Supervisor 和消息的关系

先看一个具体问题

为什么顺序执行任务还不够？

先做判断

它是什么

协调器是一个有状态的工作流引擎

它不是什么

协调器不是简单的顺序执行器

它与相邻概念的关系

协调器如何组合已有组件