第 5 章：设计可靠的 Agent Graph

本章解锁：你能设计一个包含结构化边协议、共享状态、失败恢复策略和人工确认节点的 Agent Graph 拓扑。
预计时间：25 分钟

---

起点与环境

前置章节：第 4 章（从 Loop 到 Graph 的跃迁）

你已经知道 Graph 的三要素——节点、边、共享状态——以及什么场景需要 Graph。现在我们进入 Graph 设计的具体工程问题：边上传什么？共享状态怎么结构化？节点失败了怎么办？人类在哪里介入？

这些问题没有标准答案，但有明确的设计原则。

---

为什么现在做

你可能画过一张漂亮的 Graph 拓扑图——几个方框、几条箭头、看起来逻辑清晰。但当你真正让这个 Graph 跑起来时，会发现纸上谈兵遗漏了关键细节：

• 架构评审 Agent 输出了一份 3000 字的评审报告，全部转发给安全评审 Agent。安全评审 Agent 的上下文窗口被占满，自己的工作反而做不了了——边上传递了太多无关信息。
• 三个评审 Agent 都往共享状态里写结论，有两个写的格式不一样，仲裁 Agent 解析不了——共享状态没有统一协议。
• 性能评审 Agent 在分析中崩溃了，整个 Graph 卡住——没有失败恢复机制。
• 仲裁 Agent 综合了三份评审直接做出"上线"决策，结果漏掉了安全评审中标记为"需要人工确认"的关键风险——缺少人工确认节点。

这些都是"看起来对、跑起来崩"的典型问题。接下来逐个解决。

---

动手实践

5.1 边上传递什么：结构化协议

这是 Graph 设计中最容易犯错的地方。

错误做法：把上游节点的全部对话历史转发给下游。

为什么错？因为一个 Agent 的内部对话中充满了试探、犹豫、自我纠正——这些对下游没有价值，反而会：

• 占用下游的上下文窗口
• 让下游 Agent 受到上游"思考过程"的暗示影响
• 使调试变得困难（你很难在几千字的对话历史中找到关键信息）

正确做法：设计结构化的边协议（Edge Protocol），只传递下游需要的信息。

{
  "edge_protocol": {
    "from": "architecture-reviewer",
    "to": "arbitrator",
    "payload": {
      "conclusion": "conditional_pass",
      "confidence": 0.85,
      "key_findings": [
        {
          "category": "module_design",
          "severity": "info",
          "finding": "服务拆分合理，接口定义清晰",
          "evidence": "参见方案文档 3.2 节的模块划分图"
        },
        {
          "category": "scalability",
          "severity": "warning",
          "finding": "数据库分片策略未明确，预计日均 1000 万请求下可能成为瓶颈",
          "evidence": "方案中未提及分片方案，当前设计假设单库承载"
        }
      ],
      "conditions_for_pass": [
        "补充数据库分片策略或提供单库压测数据"
      ],
      "metadata": {
        "review_duration_seconds": 45,
        "documents_analyzed": 3,
        "timestamp": "2026-07-20T10:30:00Z"
      }
    }
  }
}

协议设计的三个原则：

1. 结论先行：conclusion 字段在最前面。下游首先需要知道"结果是什么"，然后才决定是否需要看细节
2. 证据可追溯：每个 finding 都附带 evidence，指向原始信息源。这让下游和人类都能验证
3. 关注分离：只传递与下游角色相关的信息。安全评审的结果不需要包含架构细节，除非它与安全直接相关

你的实践任务：为下面这个场景设计一个边协议 JSON。

场景：安全评审 Agent 完成评审后，需要把结果传递给仲裁 Agent。安全评审发现了一个高危漏洞（SQL 注入风险）和两个低危问题。

上面展示了边协议的 JSON 结构，但在实际工程中，我们需要用代码来创建、序列化和校验这些协议。下面是一个基于 Python dataclass + Pydantic 的类型安全实现，确保边上的数据在传输前和接收后都经过校验。

from dataclasses import dataclass, field
from typing import Any
from pydantic import BaseModel, ValidationError

# ---- Pydantic 模型：用于运行时类型校验 ----
class Finding(BaseModel):
    category: str
    severity: str          # info | warning | high
    finding: str
    evidence: str

class EdgePayload(BaseModel):
    conclusion: str                  # pass | conditional_pass | fail
    confidence: float                # 0.0 ~ 1.0
    key_findings: list[Finding]
    conditions_for_pass: list[str] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)

# ---- Dataclass：轻量级边协议容器 ----
@dataclass
class EdgeProtocol:
    """节点间传递的结构化消息协议"""
    from_node: str
    to_node: str
    payload: EdgePayload

    def serialize_output(self) -> dict[str, Any]:
        """序列化为可传输的 JSON 字典"""
        return {
            "edge_protocol": {
                "from": self.from_node,
                "to": self.to_node,
                "payload": self.payload.model_dump(),
            }
        }

    @classmethod
    def parse_input(cls, raw: dict[str, Any]) -> "EdgeProtocol":
        """从 JSON 字典反序列化并做类型安全校验"""
        ep = raw.get("edge_protocol", raw)
        payload = EdgePayload.model_validate(ep["payload"])  # 校验失败抛 ValidationError
        return cls(from_node=ep["from"], to_node=ep["to"], payload=payload)

serialize_output() 在发送端生成 JSON，parse_input() 在接收端自动校验字段类型和必填项。如果上游传了格式错误的数据，Pydantic 会立即抛出 ValidationError，而不是让错误静默传播到下游节点。

5.2 共享状态设计

共享状态是 Graph 的"公共白板"，所有节点都能读写。设计不好的共享状态会变成"公共垃圾场"。

共享状态应该包含四类信息：

shared_state:
  # 1. 任务输入：所有节点都需要访问的原始资料
  task_input:
    proposal_document: "path/to/proposal.md"
    related_code: "path/to/repo"
    requirements: "path/to/requirements.md"

  # 2. 已确认决策：各节点达成的结论
  confirmed_decisions:
    - id: "decision-001"
      topic: "技术选型"
      decision: "使用 PostgreSQL 作为主数据库"
      decided_by: "architecture-reviewer"
      timestamp: "2026-07-20T10:30:00Z"
      rationale: "团队 PostgreSQL 经验丰富，且满足事务一致性需求"

  # 3. 未解决争议：需要仲裁的冲突
  open_disputes:
    - id: "dispute-001"
      topic: "缓存策略"
      positions:
        - agent: "architecture-reviewer"
          position: "使用 Redis 做分布式缓存"
          rationale: "减少数据库压力"
        - agent: "security-reviewer"
          position: "缓存层增加了攻击面，建议先不引入"
          rationale: "缓存中毒风险"
      status: "pending_arbitration"

  # 4. 证据链：关键判断的依据
  evidence_chain:
    - id: "evidence-001"
      claim: "当前 QPS 峰值约 500"
      source: "monitoring-dashboard"
      collected_by: "performance-reviewer"
      timestamp: "2026-07-20T10:45:00Z"

设计要点：

• 已确认决策和未解决争议要分开。如果混在一起，后续节点不知道哪些可以直接采纳、哪些还在争论中
• 证据链独立维护。当仲裁 Agent 需要判断谁的观点更有依据时，它需要看到原始证据，而不只是各方的"说法"
• 每条记录都有时间戳和来源。这对调试和审计至关重要

5.3 失败恢复策略

Graph 中的节点会失败。网络超时、模型输出格式错误、外部 API 不可用——你需要为每种失败准备策略。

三种失败恢复模式：

模式	含义	适用场景
重试（Retry）	同一个节点重新执行	临时性错误（网络超时、API 限流）
回退（Fallback）	使用降级方案代替	非关键节点可以跳过或简化
整图失败（Graph Fail）	整个 Graph 停止，通知人类	关键节点失败且无法恢复

failure_policies:
  architecture-reviewer:
    on_failure:
      - strategy: retry
        max_attempts: 3
        backoff: exponential  # 1s, 2s, 4s
      - strategy: fallback
        action: "使用上一次架构评审的结论（如果存在）"
      - strategy: graph_fail  # 重试和回退都失败后
        action: "通知人类，标记为需要人工架构评审"
  
  security-reviewer:
    on_failure:
      # 安全评审不能跳过——安全是底线
      - strategy: retry
        max_attempts: 3
      - strategy: graph_fail
        action: "安全评审失败，整个评审暂停，不允许继续"
  
  performance-reviewer:
    on_failure:
      - strategy: retry
        max_attempts: 2
      - strategy: fallback
        action: "跳过性能评审，在最终报告中标注'性能未评估'"
        # 性能评审不是上线的硬性阻断条件

设计要点：

• 根据节点的关键程度选择策略。安全评审失败应该阻断整个流程，性能评审失败可以降级处理
• 回退策略要明确标注。如果跳过了某个节点，最终结果中必须说明"这部分未评估"
• 重试要有限度。无限重试等于没有失败恢复

上面用 YAML 描述了失败策略，但策略需要代码来执行。下面是一个带指数退避重试和 fallback 降级的节点执行器，它将 YAML 策略转化为可运行的控制流。

import asyncio
import functools
import logging
from typing import Any, Callable, Awaitable

logger = logging.getLogger("graph.executor")

class GraphFailError(Exception):
    """整图失败异常：重试和降级都耗尽后抛出"""
    pass

def with_retry(
    max_attempts: int = 3,
    base_delay: float = 1.0,
    exceptions: tuple = (Exception,),
):
    """指数退避重试装饰器：1s → 2s → 4s …"""
    def decorator(fn: Callable[..., Awaitable[Any]]) -> Callable[..., Awaitable[Any]]:
        @functools.wraps(fn)
        async def wrapper(*args, **kwargs) -> Any:
            last_exc = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return await fn(*args, **kwargs)
                except exceptions as exc:
                    last_exc = exc
                    if attempt < max_attempts:
                        delay = base_delay * (2 ** (attempt - 1))
                        logger.warning(
                            "节点第 %d 次执行失败，%ds 后重试: %s",
                            attempt, delay, exc,
                        )
                        await asyncio.sleep(delay)
            raise last_exc  # 重试耗尽，抛出最后一个异常
        return wrapper
    return decorator

async def execute_with_policy(
    node_fn: Callable[..., Awaitable[Any]],
    fallback_fn: Callable[..., Awaitable[Any]] | None = None,
    max_retries: int = 3,
    node_name: str = "unknown",
) -> Any:
    """按失败策略执行节点：先重试，再降级，最后整图失败"""
    retried = with_retry(max_attempts=max_retries)(node_fn)
    try:
        return await retried()
    except Exception as exc:
        logger.error("节点 [%s] 重试 %d 次后仍失败: %s", node_name, max_retries, exc)
        if fallback_fn is not None:
            logger.info("节点 [%s] 启用 fallback 降级方案", node_name)
            return await fallback_fn()
        # 没有降级方案 → 整图失败
        raise GraphFailError(
            f"节点 [{node_name}] 不可恢复地失败，Graph 终止: {exc}"
        ) from exc

with_retry 负责指数退避重试，execute_with_policy 编排完整的失败恢复流程。安全评审等关键节点不传 fallback_fn，重试耗尽后直接抛 GraphFailError 终止整图；性能评审等非关键节点传入降级函数，失败后返回带标注的默认结果。

5.4 人工确认节点（Human-in-the-loop）

不是所有决策都应该由 Agent 自动做出。某些决策的风险太高、后果不可逆，或者涉及业务判断而非技术判断——这些需要人类确认。

在 Graph 中嵌入人工确认节点：

人工确认节点的设计：

human_checkpoint:
  name: "上线决策确认"
  trigger_condition: "仲裁 Agent 完成综合评审后"
  
  # 展示给人类的信息
  presentation:
    summary: "仲裁 Agent 的综合评审结论（一段话）"
    details:
      - "各评审 Agent 的结论和关键发现"
      - "未解决的争议及仲裁结果"
      - "风险等级评估"
    evidence: "证据链中的关键项"
  
  # 人类可以做的操作
  actions:
    - id: "approve"
      label: "确认上线"
      effect: "继续执行 Graph，进入部署节点"
    - id: "reject"
      label: "打回修改"
      effect: "将反馈写入共享状态，通知方案作者"
      requires_comment: true  # 打回必须说明原因
    - id: "request_info"
      label: "需要更多信息"
      effect: "指定哪个评审 Agent 补充分析，Graph 回到该节点"
  
  # 超时处理
  timeout:
    duration: "24h"
    action: "提醒人类，如无回应则自动打回"

何时需要人工确认：

需要人工确认	不需要人工确认
影响生产环境的决策	纯信息收集和分析
涉及业务判断的决策	技术性的自动验证（测试是否通过）
安全相关的重大风险	低风险的自动化操作
多个 Agent 产生不可调和的冲突	Agent 之间的冲突可以用规则自动仲裁

5.5 何时不该用 Graph

Graph 的价值必须通过结果证明，不是通过节点数量证明。

Josh Simmons 在文章中特别提到了几种不该用 Graph 的情况：

1. 任务路径高度开放：如果你不知道 Agent 在过程中会遇到什么、需要做什么决策，那你也画不出 Graph。Graph 适合路径相对可预测的任务
2. 角色边界不清晰：如果你无法明确定义每个节点的职责和输入输出，强行画 Graph 只会制造混乱
3. 协调成本超过收益：三个 Agent 之间的协调（边协议、共享状态、失败恢复）本身有成本。如果一个 Agent 花 5 分钟能做完的事，拆成三个 Agent 需要 15 分钟加上调试协调的时间，那就不该拆

一个有用的检验标准：如果你画完 Graph 拓扑后，发现大部分边都是"简单的数据转发"，没有涉及格式转换、冲突解决或条件判断——你大概不需要 Graph。

5.6 综合实践：设计方案评审 Graph

现在把这一章的所有内容综合起来，设计一个完整的多 Agent 方案评审系统。

场景：一份技术方案需要架构、安全、性能三方面的评审，最终由仲裁 Agent 综合意见，交给人类做上线决策。

你的实践任务：先尝试自己设计，然后对比参考方案。你需要定义：

1. 节点清单和每个节点的职责
2. 边协议（至少一条边的完整 JSON）
3. 共享状态的数据结构
4. 失败恢复策略
5. 人工确认节点的位置

参考方案：

节点定义：

nodes:
  initializer:
    type: function  # 非 LLM 节点
    task: "解析方案文档，提取关键章节，填充共享状态的 task_input"
    
  architecture_reviewer:
    type: agent
    model: "claude-sonnet"
    skill: "architecture-review-skill.md"
    memory: "arch-review-memory.md"
    loop_config:
      # 这个节点内部是一个小型 Loop
      steps: ["读取方案", "分析模块划分", "评估接口设计", "检查扩展性", "生成报告"]
      verification: "报告是否覆盖了所有评审维度"
    output_protocol: "edge-protocol-review.json"
    
  security_reviewer:
    type: agent
    model: "claude-sonnet"
    skill: "security-review-skill.md"
    memory: "sec-review-memory.md"
    loop_config:
      steps: ["识别数据流", "检查认证授权", "评估攻击面", "检查合规性", "生成报告"]
      verification: "是否覆盖 OWASP Top 10"
    output_protocol: "edge-protocol-review.json"
    critical: true  # 标记为关键节点
    
  performance_reviewer:
    type: agent
    model: "claude-sonnet"
    skill: "performance-review-skill.md"
    memory: "perf-review-memory.md"
    loop_config:
      steps: ["识别热路径", "评估资源需求", "检查瓶颈点", "生成报告"]
      verification: "是否给出了量化的性能预估"
    output_protocol: "edge-protocol-review.json"
    critical: false  # 非关键节点，可以降级

  arbitrator:
    type: agent
    model: "claude-opus"  # 仲裁用更强的模型
    task: |
      综合三份评审报告：
      1. 识别各报告中的一致意见 → 直接确认
      2. 识别矛盾意见 → 分析双方证据，给出仲裁
      3. 生成综合评审结论和风险等级
    reads_from: "shared_state.confirmed_decisions + shared_state.open_disputes"

  human_checkpoint:
    type: human
    config: "参见 5.4 节的设计"

关键设计决策解析：

1. 初始化节点是非 LLM 节点：解析文档和填充共享状态不需要 LLM，用确定性代码更可靠
2. 三个评审节点并行执行：它们读取相同的输入但各自独立，不需要串行
3. 仲裁 Agent 用更强的模型：仲裁需要理解三个领域的结论并做出综合判断，认知复杂度更高
4. 每个评审节点内部是一个 Loop：节点不是执行一步就完了，它内部有自己的观察-规划-执行-验证循环
5. 安全评审标记为 critical：安全评审失败导致整图失败，性能评审失败可以降级
6. 人工确认在仲裁之后：让 Agent 先做完所有分析，人类只需要做最终决策，而不是参与中间过程

上面的设计都是声明式配置，要让 Graph 真正跑起来，还需要一个执行引擎。下面是一个简化的 Graph 执行器，包含拓扑排序、并行调度和共享状态管理，将本章的所有概念串联成可运行的代码。

import asyncio
import logging
from dataclasses import dataclass, field
from typing import Any, Callable, Awaitable

logger = logging.getLogger("graph.engine")

# ============ 共享状态 ============
@dataclass
class SharedState:
    """Graph 的公共白板：所有节点共享读写"""
    task_input: dict[str, Any] = field(default_factory=dict)          # 原始资料
    confirmed_decisions: list[dict] = field(default_factory=list)     # 已确认决策
    open_disputes: list[dict] = field(default_factory=list)           # 未解决争议
    evidence_chain: list[dict] = field(default_factory=list)          # 证据链
    node_outputs: dict[str, Any] = field(default_factory=dict)        # 各节点输出缓存

    def write_output(self, node_name: str, output: Any) -> None:
        self.node_outputs[node_name] = output

    def read_inputs(self, node_name: str, dependencies: list[str]) -> list[Any]:
        """读取上游节点的输出"""
        return [self.node_outputs.get(dep) for dep in dependencies]

# ============ 节点定义 ============
@dataclass
class GraphNode:
    """一个可执行的 Graph 节点"""
    name: str
    handler: Callable[[SharedState, list[Any]], Awaitable[Any]]
    depends_on: list[str] = field(default_factory=list)

@dataclass
class Graph:
    """有向无环图：节点 + 边（依赖关系）"""
    nodes: dict[str, GraphNode] = field(default_factory=dict)

    def add_node(self, node: GraphNode) -> None:
        self.nodes[node.name] = node

    def topological_order(self) -> list[list[str]]:
        """拓扑排序 → 返回可并行执行的层级列表
        例如 [['init'], ['arch', 'sec', 'perf'], ['arbitrator']]
        """
        resolved: set[str] = set()
        layers: list[list[str]] = []
        remaining = dict(self.nodes)
        while remaining:
            # 找出所有依赖已满足的节点 → 同层可并行
            ready = [
                name for name, node in remaining.items()
                if all(dep in resolved for dep in node.depends_on)
            ]
            if not ready:
                raise ValueError(f"检测到循环依赖: {list(remaining.keys())}")
            layers.append(ready)
            resolved.update(ready)
            for name in ready:
                del remaining[name]
        return layers

# ============ 执行引擎 ============
class GraphEngine:
    """异步并行调度引擎"""

    def __init__(self, graph: Graph, state: SharedState):
        self.graph = graph
        self.state = state

    async def run(self) -> SharedState:
        layers = self.graph.topological_order()
        for i, layer in enumerate(layers):
            logger.info("执行第 %d 层（%d 个节点并行）: %s", i + 1, len(layer), layer)
            # 同层节点无依赖关系 → 用 asyncio.gather 并行执行
            tasks = [self._run_node(name) for name in layer]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            # 检查是否有节点失败
            for name, result in zip(layer, results):
                if isinstance(result, Exception):
                    logger.error("节点 [%s] 执行失败: %s", name, result)
                    raise result
        return self.state

    async def _run_node(self, name: str) -> Any:
        """执行单个节点：读取上游输出 → 调用 handler → 写回共享状态"""
        node = self.graph.nodes[name]
        upstream_outputs = self.state.read_inputs(name, node.depends_on)
        logger.debug("节点 [%s] 接收到 %d 条上游输入", name, len(upstream_outputs))
        output = await node.handler(self.state, upstream_outputs)
        self.state.write_output(name, output)
        return output

# ============ 示例：方案评审 Graph ============
async def demo():
    """演示：架构评审与安全评审并行 → 仲裁汇总"""
    async def init_handler(state: SharedState, _: list) -> dict:
        state.task_input = {"proposal": "技术方案 v1.md"}
        return {"status": "initialized"}

    async def arch_review_handler(state: SharedState, inputs: list) -> dict:
        # 模拟架构评审，实际中调用 LLM Agent
        await asyncio.sleep(0.1)
        return {"reviewer": "arch", "conclusion": "conditional_pass", "confidence": 0.85}

    async def sec_review_handler(state: SharedState, inputs: list) -> dict:
        await asyncio.sleep(0.15)  # 安全评审稍慢
        return {"reviewer": "sec", "conclusion": "fail", "confidence": 0.95}

    async def arbitrate_handler(state: SharedState, inputs: list) -> dict:
        arch_result, sec_result = inputs[0], inputs[1]
        # 仲裁 Agent 综合两份评审
        final = "reject" if sec_result["conclusion"] == "fail" else "approve"
        state.confirmed_decisions.append({
            "topic": "上线决策", "decision": final,
            "rationale": f"架构={arch_result['conclusion']}, 安全={sec_result['conclusion']}",
        })
        return {"final_decision": final, "details": [arch_result, sec_result]}

    # 构建 Graph
    graph = Graph()
    graph.add_node(GraphNode("init", init_handler))
    graph.add_node(GraphNode("arch_review", arch_review_handler, depends_on=["init"]))
    graph.add_node(GraphNode("sec_review", sec_review_handler, depends_on=["init"]))
    graph.add_node(GraphNode("arbitrator", arbitrate_handler,
                             depends_on=["arch_review", "sec_review"]))

    # 执行
    state = SharedState()
    engine = GraphEngine(graph, state)
    result = await engine.run()
    print(f"最终决策: {result.confirmed_decisions}")

# asyncio.run(demo())  # 取消注释即可运行

这段代码把本章的核心概念串联了起来：SharedState 管理四类信息，EdgeProtocol（上一段代码定义的）可嵌入 node_outputs 实现节点间类型安全的数据传递，topological_order() 自动算出可并行层级，asyncio.gather 同层并行调度。demo() 函数展示了架构评审和安全评审并行执行、仲裁节点汇总两份评审的完整流程。

---

理解检查

1. 设计题：为安全评审 Agent → 仲裁 Agent 的边写一个完整的 JSON 协议。假设安全评审发现了一个高危 SQL 注入风险和一个中危的不安全 HTTP 头配置。

2. 判断题：一个 Graph 中的所有节点都标记为 critical: true。这个设计有什么问题？

3. 辨析题：共享状态中的"已确认决策"和仲裁 Agent 做出的"仲裁结论"有什么区别？

参考答案：

1. 参考协议：

{
  "edge_protocol": {
    "from": "security-reviewer",
    "to": "arbitrator",
    "payload": {
      "conclusion": "fail",
      "confidence": 0.95,
      "risk_level": "high",
      "key_findings": [
        {
          "category": "injection",
          "severity": "high",
          "finding": "用户输入直接拼接 SQL 查询，存在 SQL 注入风险",
          "evidence": "方案文档 4.3 节的查询实现示例使用字符串拼接",
          "recommendation": "使用参数化查询或 ORM",
          "blocking": true
        },
        {
          "category": "headers",
          "severity": "medium",
          "finding": "API 响应未配置 Content-Security-Policy 和 X-Frame-Options",
          "evidence": "方案中未提及 HTTP 安全头配置",
          "recommendation": "在网关层统一配置安全响应头",
          "blocking": false
        }
      ],
      "conditions_for_pass": [
        "修复 SQL 注入风险（blocking）",
        "补充 HTTP 安全头配置（non-blocking，可作为 follow-up）"
      ],
      "metadata": {
        "review_duration_seconds": 62,
        "owasp_categories_checked": 10,
        "timestamp": "2026-07-20T10:35:00Z"
      }
    }
  }
}

2. 所有节点都 critical 意味着任何一个节点失败，整个 Graph 都失败。这抹杀了 Graph 的弹性——连非关键节点的失败也会导致整个流程停止。应该只有真正不可或缺的节点才标记为 critical，其他节点应该有降级方案。

3. "已确认决策"是各节点各自做出的、不存在争议的结论（如"使用 PostgreSQL"）。"仲裁结论"是仲裁 Agent 针对各节点之间的冲突做出的判断（如"架构师建议用 Redis 缓存 vs 安全专家认为缓存增加攻击面"→ 仲裁结论："引入 Redis 但限制缓存内容不含敏感数据"）。前者是共识，后者是裁决。

---

发生了什么

这一章你深入到了 Graph 设计的工程细节。

核心概念回顾：

• 边协议：结构化传递结论、证据和置信度，不转发对话历史
• 共享状态四类信息：任务输入、已确认决策、未解决争议、证据链
• 三种失败恢复：重试、回退、整图失败——根据节点关键程度选择
• 人工确认节点：在高风险决策点嵌入，展示摘要而非全部细节
• 何时不该用 Graph：路径开放、边界不清、协调成本超过收益
• Graph 节点内部可能是 Loop：两层嵌套是正常的

---

常见误区

误区	纠正
"边上传越多信息越好"	信息过多会淹没下游节点。只传递结论、证据和元数据，不传思考过程。
"共享状态就是一个大 JSON，随便写"	共享状态需要严格的 schema。没有 schema 的共享状态会变成"各写各的"，仲裁 Agent 无法解析。
"所有节点都应该是 LLM Agent"	初始化、格式转换、条件路由这些确定性任务用普通代码更可靠。LLM 用在需要判断力的地方。
"人工确认越多越安全"	人工确认太多等于没有自动化。只在高风险、不可逆的决策点加人工确认。
"Graph 画得越复杂说明系统越强大"	Josh Simmons 指出：Graph 的价值通过结果证明，不通过节点数量证明。追求的是最小可行 Graph。

---

本章检查点

完成以下清单后，你可以进入下一章：

• [ ] 我能设计一个包含结论、证据和元数据的边协议 JSON
• [ ] 我能设计包含四类信息的共享状态结构
• [ ] 我能根据节点关键程度选择失败恢复策略
• [ ] 我知道人工确认节点应该放在哪里
• [ ] 我能识别"不该用 Graph"的场景
• [ ] 我已经完成（或仔细阅读）了方案评审 Graph 的设计

下一章预告：你已经掌握了 Loop 和 Graph 的设计方法。最后一章我们退后一步，用统一视角审视五层工程——它们如何叠加、何时选哪层、以及一个不容忽视的风险：Agentic Design Debt。

---

← 上一章：当一个 Loop 不够用 | 下一章：五层工程的统一视角与实践选择 →