Agent持久执行六大运行时实测：从秒级调用到天级任务的全链拆解

# Google AX 持久执行示例
from google.ax import AgentExecutor, step

executor = AgentExecutor("order-agent")

@step(checkpoint=True)
async def check_inventory(item_id: str):
    result = await db.query("SELECT stock FROM items WHERE id = ?", item_id)
    return result

@step(checkpoint=True, compensate="restore_inventory")
async def reserve_inventory(item_id: str, qty: int):
    await db.execute("UPDATE items SET stock = stock - ? WHERE id = ?", qty, item_id)

@step(checkpoint=True)
async def process_payment(order_id: str):
    payment = await payment_api.charge(order_id)
    if payment.failed:
        raise PaymentError(payment.reason)  # 触发Saga补偿
    return payment

async def restore_inventory(item_id: str, qty: int):
    await db.execute("UPDATE items SET stock = stock + ? WHERE id = ?", qty, item_id)

# 持久执行：崩溃后自动恢复，跳过已完成步骤
@executor.workflow
async def order_workflow(order: Order):
    await check_inventory(order.item_id)
    await reserve_inventory(order.item_id, order.qty)
    approval = await executor.wait_for_approval(
        "confirm_order", timeout="7d"
    )
    if approval.approved:
        await process_payment(order.id)

# Temporal 持久Agent工作流
from temporalio import workflow, activity
from datetime import timedelta

@activity.defn
async def analyze_document(doc_id: str) -> AnalysisResult:
    """活动：可被持久化、重试、超时控制"""
    doc = await doc_store.get(doc_id)
    return await llm.analyze(doc)

@activity.defn
async def human_review(analysis: AnalysisResult) -> ReviewDecision:
    """活动：等待人工审核，可持续数天"""
    review = await review_queue.submit(analysis)
    return await review.wait(timeout=timedelta(days=3))

@workflow.defn
class DocumentReviewWorkflow:
    """工作流：自动检查点，崩溃恢复，确定性重放"""

    @workflow.run
    async def run(self, doc_id: str) -> FinalResult:
        # 步骤1：自动分析（失败自动重试）
        analysis = await workflow.execute_activity(
            analyze_document, doc_id,
            start_to_close_timeout=timedelta(minutes=30),
            retry_policy=RetryPolicy(maximum_attempts=3)
        )

        # 步骤2：人工审批（可持续数天）
        review = await workflow.execute_activity(
            human_review, analysis,
            start_to_close_timeout=timedelta(days=3)
        )

        if review.approved:
            return await workflow.execute_activity(
                finalize_document, analysis,
                start_to_close_timeout=timedelta(minutes=10)
            )

# LangGraph 持久执行
from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres import PostgresSaver

# 配置持久化后端
checkpointer = PostgresSaver.from_conn_string("postgresql://...")

builder = StateGraph(AgentState)
builder.add_node("research", research_node)
builder.add_node("analyze", analyze_node) 
builder.add_node("report", report_node)
builder.add_node("human_review", human_review_node)

builder.add_edge("research", "analyze")
builder.add_edge("analyze", "human_review")
builder.add_edge("human_review", "report")

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review"]  # 执行到此节点前暂停
)

# 首次执行：到human_review前暂停
result = await graph.ainvoke(
    {"topic": "AI Agent trends"},
    config={"configurable": {"thread_id": "thread-123"}}
)

# 人工审核后恢复执行
result = await graph.ainvoke(
    None,  # 无需重新传入输入
    config={"configurable": {"thread_id": "thread-123"}}
)

// Inngest Agent Kit 持久执行
import { agent, step } from "inngest/agent-kit";

export const researchAgent = agent({
  name: "Research Agent",
  model: "claude-sonnet-4-6",

  steps: [
    step("search", async ({ query }) => {
      const results = await searchEngine.search(query);
      return results;
    }),
    step("analyze", async ({ searchResults }) => {
      const analysis = await llm.analyze(searchResults);
      return analysis;
    }),
    step("wait-for-approval", async ({ analysis }) => {
      // 可持久等待数天
      const approval = await waitForHumanInput(analysis, {
        timeout: "7d",
        channel: "slack"
      });
      return approval;
    }),
    step("generate-report", async ({ analysis, approval }) => {
      return await generateReport(analysis, approval.comments);
    })
  ],

  // 自动重试、检查点、并发控制
  config: {
    retries: 3,
    checkpoint: true,
    concurrency: { limit: 5, key: "user.id" }
  }
});

// Restate 持久Agent
import { endpoint } from "@restatedev/restate-sdk";

const agentService = endpoint({
  name: "OrderAgent",
  handler: async (ctx: restate.Context, order: Order) => {
    // 每个调用自动持久化到Journal
    const inventory = await ctx.run("check-inventory", () =>
      checkInventory(order.itemId)
    );

    if (!inventory.available) {
      return { status: "out_of_stock" };
    }

    // 持久等待人工确认（可等待数天）
    const approval = await ctx.run("request-approval", () =>
      requestHumanApproval(order, { timeout: "3d" })
    );

    // Awakeable：外部事件唤醒
    const paymentResult = await ctx.awakeable<PaymentResult>();
    sendPaymentRequest(order, paymentResult.id);

    return { status: "completed" };
  }
});

// Microsoft Agent Framework Durable Agent
[Agent(AgentName = "DocumentProcessor")]
public class DocumentAgent : DurableAgent
{
    [AgentStep]
    public async Task<AnalysisResult> Analyze(
        [DurableActivity] IDurableActivityContext context)
    {
        var doc = context.GetInput<Document>();
        return await _llm.AnalyzeAsync(doc);
    }

    [AgentStep]
    public async Task<ReviewDecision> WaitForReview(
        [DurableTimer] IDurableOrchestrationContext context,
        AnalysisResult analysis)
    {
        // 持久定时器：等待外部事件或超时
        var timeout = context.CurrentUtcDateTime.AddDays(3);
        var reviewEvent = context.WaitForExternalEvent<ReviewDecision>("ReviewSubmitted");
        var timeoutEvent = context.CreateTimer(timeout, CancellationToken.None);

        var winner = await Task.WhenAny(reviewEvent, timeoutEvent);
        return winner == reviewEvent 
            ? reviewEvent.Result 
            : ReviewDecision.Timeout;
    }
}

graph TB
    subgraph "Agent应用层"
        A[Agent Loop] --> B[Tool Calls]
        B --> C[LLM Inference]
    end

    subgraph "持久执行层（六大引擎）"
        D[检查点管理器<br/>Checkpoint Manager]
        E[事件日志<br/>Event Journal]
        F[调度器<br/>Scheduler]
        G[补偿管理器<br/>Compensation/Saga]
    end

    subgraph "等待语义层"
        H[Human-in-the-Loop<br/>人工审批]
        I[Timer<br/>定时器]
        J[External Event<br/>外部事件]
        K[Signal<br/>信号通知]
    end

    subgraph "状态存储层"
        L[PostgreSQL]
        M[Cassandra]
        N[Redis]
        O[Cloud Spanner]
        P[Azure Storage]
    end

    C --> D
    D --> E
    E --> F
    F --> G

    G --> H
    G --> I
    G --> J
    G --> K

    H --> L
    H --> M
    H --> N
    H --> O
    H --> P

    style D fill:#4A90D9,color:#fff
    style E fill:#7B68EE,color:#fff
    style F fill:#20B2AA,color:#fff
    style G fill:#FF6B6B,color:#fff

graph LR
    subgraph "场景判断"
        S1{需要持久执行?}
        S2{运行时长?}
        S3{技术栈?}
        S4{基础设施?}
        S5{团队规模?}
    end

    S1 -->|秒级完成| X[无需持久执行]
    S1 -->|分钟到天级| S2
    S2 -->|< 30分钟| S3
    S2 -->|> 30分钟| S4
    S3 -->|Python| LangGraph
    S3 -->|TypeScript| Inngest
    S4 -->|已有GCP| Google AX
    S4 -->|已有Azure| Azure Durable Task
    S4 -->|自建| S5
    S5 -->|> 5人DevOps| Temporal
    S5 -->|< 5人| Restate