Agent容错执行：六大引擎横评

# Temporal + OpenAI Agents SDK 集成示例
from temporalio import workflow
from datetime import timedelta

@workflow.defn
class DurableResearchAgent:
    @workflow.run
    async def run(self, query: str) -> str:
        # Step 1: 搜索（已完成则跳过）
        results = await workflow.execute_activity(
            search_web, query,
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=RetryPolicy(max_attempts=3)
        )
        # Step 2: 分析（崩溃恢复后从此处继续）
        analysis = await workflow.execute_activity(
            analyze_results, results,
            start_to_close_timeout=timedelta(minutes=10)
        )
        # Step 3: 等待人工审批（可等待数天）
        approval = await workflow.execute_activity(
            human_approval, analysis,
            start_to_close_timeout=timedelta(days=7)
        )
        return analysis if approval else "rejected"

// Restate: 持久化Agent审批流程
import { ObjectContext } from "@restatedev/restate-sdk";

export class ApprovalAgent {
  async run(ctx: ObjectContext, request: RefundRequest) {
    // 幂等执行：重复调用只执行一次
    const eligibility = ctx.run(() => checkEligibility(request));

    // 持久化等待：可等待数天，不消耗计算资源
    const approval = ctx.awakeable<ApprovalResult>();
    await notifyManager(request, approval.id);
    const result = await approval.promise;

    if (result.approved) {
      ctx.run(() => processRefund(request));
    }
  }
}

// Inngest: Step-by-step持久化Agent
import { inngest } from "./client";

export const researchAgent = inngest.createFunction(
  { id: "research-agent", retries: 3 },
  { event: "agent/research" },
  async ({ event, step }) => {
    // 每个step自动持久化，崩溃后从最后一个完成的step恢复
    const searchResults = await step.run("search", async () => {
      return await webSearch(event.data.query);
    });

    const analysis = await step.run("analyze", async () => {
      return await llmAnalyze(searchResults);
    });

    // 等待外部事件（如人工审批）
    const approval = await step.waitForEvent("wait-approval", {
      event: "agent/approved",
      timeout: "7d"
    });

    return approval ? analysis : "rejected";
  }
);

// Cloudflare: Durable Agent with Fibers
import { Agent } from "agents";

class ResearchAgent extends Agent {
  async runTask(query: string) {
    await this.runFiber("research", async (ctx) => {
      // 搜索阶段
      const results = await searchWeb(query);
      // 检查点：保存中间状态
      ctx.stash({ phase: "searched", results });

      // 分析阶段（崩溃后从此处恢复）
      const analysis = await analyzeResults(results);
      ctx.stash({ phase: "analyzed", analysis });

      // 生成报告
      await generateReport(analysis);
    });
  }

  // 崩溃恢复回调
  async onFiberRecovered(ctx) {
    const state = ctx.stash();
    console.log(`Recovered at phase: ${state.phase}`);
  }
}

# DBOS: Postgres原生持久化Agent
from dbos import DBOS

@DBOS.workflow()
def research_pipeline(query: str):
    # 每个事务自动持久化到Postgres
    results = search_step(query)
    analysis = analyze_step(results)

    if needs_approval(analysis):
        # 持久化等待人工审批
        approval = wait_for_approval(analysis)
    else:
        approval = True

    if approval:
        return generate_report(analysis)
    return "rejected"

@DBOS.step()  # 幂等步骤：重复调用返回缓存结果
def search_step(query: str):
    return web_search(query)

// Azure Durable Task: Agent循环
[DurableTask("research-agent")]
public class ResearchAgent : TaskOrchestrator<string, string>
{
    public override async Task<string> RunAsync(
        TaskOrchestrationContext ctx, string query)
    {
        var results = await ctx.CallActivityAsync<string[]>(
            "SearchWeb", query);

        // Agent循环：LLM决定是否继续
        bool complete = false;
        while (!complete)
        {
            var decision = await ctx.CallActivityAsync<AgentDecision>(
                "LLMDecide", new { results, query });

            if (decision.NeedsMore)
            {
                results = await ctx.CallActivityAsync<string[]>(
                    "DeepSearch", decision.NextQuery);
            }
            complete = decision.IsComplete;
        }

        return await ctx.CallActivityAsync<string>(
            "GenerateReport", results);
    }
}

graph TB
    subgraph "传统方案"
        A1[每个Turn全量检查点] --> A2[87%的检查点浪费]
        A2 --> A3[高I/O开销]
    end

    subgraph "Crab方案"
        B1[eBPF监控OS事件] --> B2{Turn产生恢复相关状态?}
        B2 -->|是| B3[精准检查点]
        B2 -->|否 75%Turn| B4[跳过检查点]
        B3 --> B5[与LLM等待时间重叠]
        B4 --> B5
        B5 --> B6[87%流量削减<br/>1.9%时间开销]
    end

    style A2 fill:#ff6b6b
    style B6 fill:#51cf66

graph LR
    subgraph "Agent层"
        A1[ReAct循环]
        A2[工具调用]
        A3[LLM推理]
    end

    subgraph "容错执行层"
        B1[事件历史日志]
        B2[Step幂等注册表]
        B3[检查点存储]
        B4[重放引擎]
    end

    subgraph "基础设施层"
        C1[Temporal Cloud]
        C2[Restate单二进制]
        C3[Cloudflare DO]
        C4[Postgres WAL]
        C5[Azure Task Hub]
    end

    A1 --> B1
    A2 --> B2
    A3 --> B4
    B1 --> C1
    B1 --> C2
    B2 --> C3
    B3 --> C4
    B4 --> C5

    subgraph "恢复流程"
        D1[崩溃检测] --> D2[事件重放]
        D2 --> D3[跳过已完成Step]
        D3 --> D4[恢复到崩溃点]
        D4 --> D5[继续执行]
    end

    B4 -.-> D1

graph TD
    Start{选择容错引擎} --> Q1{是否需要多语言支持?}
    Q1 -->|是| Q2{团队规模?}
    Q1 -->|仅TS| Q3{部署环境?}
    Q2 -->|>10人平台团队| Temporal
    Q2 -->|<10人| Q4{能否接受运行数据库集群?}
    Q4 -->|否| Restate
    Q4 -->|是| DBOS
    Q3 -->|Cloudflare| Cloudflare
    Q3 -->|Vercel/AWS| Q5{偏好运维模式?}
    Q5 -->|零运维Serverless| Inngest
    Q5 -->|Azure托管| Azure Durable Task

    style Temporal fill:#4dabf7
    style Restate fill:#51cf66
    style Cloudflare fill:#fcc419
    style DBOS fill:#ff922b
    style Inngest fill:#cc5de8
    style Azure fill:#20c997