自主Agent持久执行六栈：从崩溃到秒级续跑的全线实测

@workflow.defn
class ResearchAgentWorkflow:
    @workflow.run
    async def run(self, research_topic: str) -> str:
        # Phase 1: 自主研究
        findings = await workflow.execute_activity(
            conduct_research, research_topic,
            start_to_close_timeout=timedelta(hours=2)
        )

        # Phase 2: 人工审批门控——暂停等待人类确认
        approval = await workflow.execute_activity(
            request_human_approval, findings,
            start_to_close_timeout=timedelta(days=7)  # 最长等7天
        )

        if approval.approved:
            # Phase 3: 执行获批操作
            return await workflow.execute_activity(
                publish_report, findings,
                start_to_close_timeout=timedelta(hours=1)
            )
        else:
            return "Research rejected"

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

# 三个存储后端：MemorySaver（开发）、SqliteSaver（本地）、
# AsyncPostgresSaver（生产）
checkpointer = AsyncPostgresSaver.from_conn_string(DB_URI)

graph = builder.compile(checkpointer=checkpointer)

# 带 interrupt 的人机审批
@node
def review_node(state):
    if state["risk_score"] > 0.8:
        return interrupt({"question": "高风险操作，是否继续？", 
                          "data": state["proposed_action"]})
    return {"approved": True}

[Function("ResearchAgentOrchestrator")]
public static async Task<string> RunOrchestrator(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    // Step 1: 自主分析
    var analysis = await context.CallActivityAsync<string>(
        "AnalyzeRequirements", context.GetInput<string>());

    // Step 2: 等待人工审批（可跨天）
    var approval = await context.WaitForExternalEvent<string>("Approval");

    // Step 3: 条件执行
    if (approval == "approved")
    {
        return await context.CallActivityAsync<string>(
            "ExecuteAction", analysis);
    }
    return "Cancelled";
}

from pydantic_ai import Agent
from pydantic import BaseModel

class ResearchResult(BaseModel):
    topic: str
    findings: list[str]
    confidence: float

research_agent = Agent(
    'openai:gpt-4o',
    result_type=ResearchResult,
)

# Kitaru持久化集成
from kitaru import durable

@durable(checkpoint_every=5)  # 每5轮自动checkpoint
async def run_research(topic: str) -> ResearchResult:
    result = await research_agent.run(topic)
    return result.data

import { Agent, Workflow } from '@mastra/core';
import { z } from 'zod';

const researcher = new Agent({
  name: 'researcher',
  instructions: 'You are a research assistant...',
  model: { provider: 'openai', name: 'gpt-4o' },
});

const workflow = new Workflow({
  name: 'research-pipeline',
  triggerSchema: z.object({ topic: z.string() }),
});

// Step持久化自动完成
workflow
  .step(researchStep, { agent: researcher })
  .then(analysisStep)
  .then(humanApprovalStep)  // 内置人工等待
  .then(publishStep);

// 崩溃后通过 workflow.run({ resumeFrom: 'analysisStep' }) 恢复

graph TB
    subgraph "Crab架构：语义感知检查点"
        A[Agent LLM] -->|发出操作| B[沙箱容器]
        B -->|系统调用| C[Host侧拦截层]
        C -->|语义过滤| D{状态是否恢复相关?}
        D -->|是| E[增量检查点存储]
        D -->|否| F[丢弃]
        E -->|崩溃恢复| G[精确还原沙箱]
        G -->|继续执行| B
    end

    style A fill:#4A90D9,color:#fff
    style C fill:#E67E22,color:#fff
    style E fill:#27AE60,color:#fff
    style G fill:#8E44AD,color:#fff

stateDiagram-v2
    [*] --> Pending: 提交任务
    Pending --> Running: 调度执行
    Running --> Checkpointing: 自动/手动保存点
    Checkpointing --> Running: 保存完成
    Running --> Suspended: 等待人工/外部事件
    Suspended --> Running: 事件到达
    Running --> Failed: 异常/超时
    Failed --> Recovering: 从Checkpoint恢复
    Recovering --> Running: 恢复成功
    Failed --> Dead: 重试耗尽
    Running --> Completed: 任务完成
    Completed --> [*]
    Dead --> [*]

    note right of Checkpointing
        状态快照
        - LangGraph: Node级
        - Temporal: Activity级
        - Mastra: Step级
    end note

    note right of Suspended
        人机门控
        - Temporal: Signal/Query
        - LangGraph: interrupt()
        - MAF: ExternalEvent
    end note

graph TD
    START[需要Agent持久执行] --> Q1{任务时长?}
    Q1 -->|>1天| Q2{是否Azure生态?}
    Q1 -->|<1天| Q3{团队主语言?}

    Q2 -->|是| AZ[Microsoft Durable Task<br/>+ Azure Functions]
    Q2 -->|否| Q4{需要跨语言?}

    Q4 -->|是| TEMP[Temporal<br/>工业级可靠]
    Q4 -->|否| LG[LangGraph Checkpoint<br/>Agent原生]

    Q3 -->|Python| Q5{类型安全优先?}
    Q3 -->|TypeScript| Q6{已有Web项目?}

    Q5 -->|是| PA[Pydantic AI + Kitaru<br/>类型安全]
    Q5 -->|否| LG2[LangGraph Checkpoint<br/>生态最全]

    Q6 -->|是| MA[Mastra Workflow<br/>全栈统一]
    Q6 -->|否| Q7{需要企业级SLA?}

    Q7 -->|是| TEMP2[Temporal Cloud<br/>免运维]
    Q7 -->|否| MA2[Mastra 自部署]

    style AZ fill:#0078D4,color:#fff
    style TEMP fill:#000,color:#fff
    style LG fill:#1A8B3F,color:#fff
    style PA fill:#E83E8C,color:#fff
    style MA fill:#8B5CF6,color:#fff