自主Agent中断恢复与人机交接六大引擎实战拆解

关键数据点（2026年多项研究汇总）：
- 长程Agent任务平均步骤数：12-47步
- 无持久化方案的任务完成率：31-42%
- 有检查点方案的任务完成率：78-89%
- 传统全量快照的I/O浪费：>75%
- Cold Handoff后的人类平均上下文重建时间：8-15分钟
- Warm Handoff后的上下文重建时间：<30秒

flowchart TD
    A[用户提交任务] --> B[规划节点]
    B --> C{interrupt?}
    C -->|需要审批| D[暂停 - 状态写入Checkpointer]
    D --> E[人类审核]
    E -->|批准| F[resume - 状态恢复]
    F --> G[执行节点]
    C -->|无需审批| G
    G --> H{还有下一步?}
    H -->|是| B
    H -->|否| I[完成]

    style D fill:#f9f,stroke:#333
    style F fill:#9f9,stroke:#333

from langgraph.graph import StateGraph
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt

def plan_step(state):
    plan = llm.invoke(f"制定计划: {state['task']}")
    # 在执行前暂停，等待人类审批
    approved = interrupt({"plan": plan, "request": "请审批此计划"})
    if not approved:
        return {"status": "rejected"}
    return {"plan": plan, "status": "approved"}

def execute_step(state):
    result = tool_executor.run(state["plan"])
    return {"results": [result]}

graph = StateGraph(AgentState)
graph.add_node("plan", plan_step)
graph.add_node("execute", execute_step)
graph.add_edge("plan", "execute")

# 检查点持久化
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = graph.compile(checkpointer=checkpointer)

# 首次运行 - 执行到interrupt()暂停
config = {"configurable": {"thread_id": "task-001"}}
result = app.invoke({"task": "迁移数据库"}, config)

# 人类审批后恢复
app.invoke(None, config)  # 恢复执行

# Google ADK 暂停恢复模式
from google.adk import Agent, Runner
from google.adk.sessions import InMemorySessionService

agent = Agent(
    name="long_runner",
    model="gemini-2.5-pro",
    instruction="你是一个合规审查Agent"
)

runner = Runner(agent=agent, session_service=InMemorySessionService())

# 第一次运行 - 执行到需要审批的操作时自动暂停
session = await runner.run(
    user_id="user-001",
    session_id="compliance-001",
    message="审查这笔跨境交易"
)
# session.state == "suspended"

# ... 人类审批操作 ...

# 恢复执行 - 上下文自动重建
session = await runner.resume(
    session_id="compliance-001",
    event={"type": "approval", "approved": True}
)

flowchart LR
    subgraph "Agent层"
        A[Agent Logic]
    end

    subgraph "Temporal层"
        W[Workflow Orchestrator]
        E[Event History]
        Q[Task Queue]
    end

    subgraph "基础设施"
        W1[Worker 1]
        W2[Worker 2]
        W3[Worker N]
    end

    A -->|注册步骤| W
    W -->|追加事件| E
    W -->|分发任务| Q
    Q --> W1
    Q --> W2
    Q --> W3
    W1 -->|返回结果| W
    W2 -->|返回结果| W
    W3 -->|返回结果| W

# Temporal + Agent 的持久执行
from temporalio import workflow, activity

@activity.defn
async def call_llm(prompt: str) -> str:
    return await llm_client.invoke(prompt)

@activity.defn
async def call_tool(tool_name: str, params: dict) -> dict:
    return await tool_executor.run(tool_name, params)

@activity.defn
async def human_approval(decision: dict) -> bool:
    # Temporal支持等待外部信号（Signal）
    # 人类可以在几秒或几天后批准
    return await wait_for_approval(decision)

@workflow.defn
class AutonomousAgentWorkflow:
    @workflow.run
    async def run(self, task: str):
        plan = await workflow.execute_activity(
            call_llm, f"制定计划: {task}", start_to_close_timeout=timedelta(minutes=5)
        )

        for step in parse_plan(plan):
            if step.requires_approval:
                approved = await workflow.execute_activity(
                    human_approval, step, 
                    start_to_close_timeout=timedelta(days=7)  # 最多等7天
                )
                if not approved:
                    return {"status": "rejected", "step": step}

            result = await workflow.execute_activity(
                call_tool, step.tool, step.params,
                retry_policy=RetryPolicy(max_attempts=3)
            )

        return {"status": "completed"}

sequenceDiagram
    participant A as Agent
    participant B as 交接缓冲区
    participant H as 人类

    A->>B: 阶段1: 上下文封装
    Note over B: 已完成步骤摘要<br/>当前决策点<br/>候选方案及置信度<br/>推荐行动

    B->>H: 阶段2: 意图预览
    Note over H: 30秒内理解<br/>Agent的意图和理由

    H->>B: 阶段3: 接管指令
    Note over B: 批准/修改/拒绝<br/>附带指导信息

    B->>A: 阶段4: 控制权转移
    Note over A: 根据人类指令<br/>继续或回滚

flowchart TD
    subgraph "传统方案（全量快照）"
        T1[Step 1: 思考] --> S1[完整快照 - 2.3GB]
        T2[Step 2: 思考] --> S2[完整快照 - 2.3GB]
        T3[Step 3: 执行命令] --> S3[完整快照 - 2.3GB]
        T4[Step 4: 思考] --> S4[完整快照 - 2.3GB]
    end

    subgraph "Crab（语义感知）"
        C1[Step 1: 思考] --> N1[跳过 - 无恢复状态]
        C2[Step 2: 思考] --> N2[跳过 - 无恢复状态]
        C3[Step 3: 执行命令] --> D3[增量快照 - 47MB]
        C4[Step 4: 思考] --> N4[跳过 - 无恢复状态]
    end

    style S1 fill:#f99
    style S2 fill:#f99
    style S3 fill:#f99
    style S4 fill:#f99
    style N1 fill:#9f9
    style N2 fill:#9f9
    style D3 fill:#ff9
    style N4 fill:#9f9

Agent执行了一个5步任务：
Step 1: 创建数据库记录 ✓
Step 2: 发送邮件通知 ✓  
Step 3: 调用支付API ✓
Step 4: 更新库存系统 ✗ (失败)
Step 5: 生成报告 (未执行)

传统回滚：撤销Step 1-3 → 但邮件已发出，支付已完成
正确做法：执行补偿操作 → "已发货退款"、"撤回通知邮件"

# RAC 补偿日志模式
class CompensationLog:
    def __init__(self):
        self.log = []

    def record(self, action, compensation):
        self.log.append({
            "action": action,        # 正向操作
            "compensation": compensation,  # 补偿操作
            "executed": False
        })

    def compensate(self, failed_step):
        # 从失败步骤反向执行补偿
        for entry in reversed(self.log[:failed_step]):
            if not entry["executed"]:
                execute(entry["compensation"])
                entry["executed"] = True

# Agent执行时自动记录补偿
log = CompensationLog()
log.record(
    action="INSERT INTO orders VALUES (...)",
    compensation="DELETE FROM orders WHERE id=..."
)
log.record(
    action="send_email(to='customer@...', body='订单确认')",
    compensation="send_email(to='customer@...', body='订单已取消')"
)
log.record(
    action="charge_credit_card(amount=999)",
    compensation="refund_credit_card(amount=999, reason='order_cancelled')"
)

flowchart TD
    START[你的Agent任务] --> Q1{任务平均步数?}
    Q1 -->|≤5步| FAST[简单重试即可<br/>无需复杂持久化]
    Q1 -->|5-20步| Q2{需要人类审批?}
    Q1 -->|>20步| Q3{基础设施预算?}

    Q2 -->|是| LG[LangGraph中断恢复<br/>最简单的审批模式]
    Q2 -->|否| Q4{失败代价?}

    Q4 -->|低| FAST
    Q4 -->|高| RAC[RAC补偿模式<br/>每步预定义回滚]

    Q3 -->|充足| TEMP[Temporal持久执行<br/>工业级容错]
    Q3 -->|有限| ADK[Google ADK<br/>事件溯源方案]

    LG --> Q5{运行在沙箱?}
    Q5 -->|是| CRAB[叠加Crab<br/>语义检查点]
    Q5 -->|否| FINAL[方案确定]

    TEMP --> FINAL
    ADK --> FINAL
    RAC --> FINAL
    CRAB --> FINAL
    FAST --> FINAL

    style FINAL fill:#4CAF50,color:white
    style TEMP fill:#2196F3,color:white
    style LG fill:#FF9800,color:white
    style RAC fill:#9C27B0,color:white
    style CRAB fill:#607D8B,color:white

推荐组合：
┌──────────────────────────────────────────┐
│          Temporal (持久执行底层)            │
│  ┌──────────┐  ┌──────────┐  ┌────────┐ │
│  │ LangGraph │  │ Warm     │  │  RAC   │ │
│  │ 中断恢复  │  │ Handoff  │  │ 补偿   │ │
│  └──────────┘  └──────────┘  └────────┘ │
│       ↕              ↕              ↕     │
│  ┌─────────────────────────────────────┐ │
│  │      Agent Business Logic           │ │
│  └─────────────────────────────────────┘ │
└──────────────────────────────────────────┘

层级关系：
- Temporal负责基础设施级容错（进程崩溃、网络中断）
- LangGraph中断负责业务级暂停（人类审批）
- Warm Handoff负责人机交接的上下文保全
- RAC负责失败后的确定性回滚