AI Agent持久化执行六大引擎架构对比

@activity.defn
async def call_llm(prompt: str) -> str:
    response = await openai.chat.completions.create(
        model="gpt-4", messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

@workflow.defn
class ResearchAgentWorkflow:
    @workflow.run
    async def run(self, topic: str) -> str:
        # Step 1: 搜索 - 自动持久化，崩溃后从此处恢复
        search_results = await workflow.execute_activity(
            call_llm, f"搜索关于{topic}的最新资料",
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(
                maximum_attempts=5,
                initial_interval=timedelta(seconds=1),
                backoff_coefficient=2.0,
            ),
        )
        # Step 2: 分析
        analysis = await workflow.execute_activity(
            call_llm, f"分析以下搜索结果：{search_results}",
            start_to_close_timeout=timedelta(minutes=2),
        )
        # Step 3: 人工审核 - 暂停等待外部Signal
        await workflow.wait_condition(lambda: self._approved is not None)
        if not self._approved:
            return "用户拒绝了报告"
        return analysis

// Restate: 将Agent定义为Virtual Object
const agentRouter = restate.router({
  // 每个researchSession有唯一的execution guarantee
  research: async (ctx: restate.Context, topic: string) => {
    // 调用LLM - 自动持久化，崩溃后自动重放
    const result = await ctx.run(() =>
      openai.chat.completions.create({
        model: "gpt-4",
        messages: [{ role: "user", content: `研究: ${topic}` }],
      })
    );

    // 持久化中间结果
    ctx.set("researchResult", result);

    // Awakeable: 暂停等待外部事件（如人工审核）
    const approval = ctx.awakeable<boolean>();
    // 发送审核请求给外部系统
    await ctx.run(() => sendForApproval(approval.id, result));
    const { value: approved } = await approval.promise;

    if (approved) {
      const report = await ctx.run(() => generateReport(result));
      return report;
    }
    return "rejected";
  },
});

from dbos import DBOS

@DBOS.workflow()
def research_agent_workflow(topic: str) -> str:
    # Step 1: 搜索 - 自动持久化到Postgres
    search_result = search_step(topic)
    # Step 2: 分析 - 如果崩溃，恢复时自动跳过已完成的步骤
    analysis = analysis_step(search_result)
    # Step 3: 生成报告
    report = report_step(analysis)
    return report

@DBOS.step(retries_allowed=True, max_attempts=3)
def search_step(topic: str) -> str:
    """每个step自动记录到Postgres，崩溃后精确恢复"""
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": f"搜索{topic}"}]
    )
    return response.choices[0].message.content

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_step: str
    research_data: str

def research_node(state: AgentState) -> dict:
    """研究节点 - 执行后自动checkpoint"""
    result = llm.invoke(f"研究: {state['messages'][-1]}")
    return {"research_data": result, "next_step": "analyze"}

def analyze_node(state: AgentState) -> dict:
    return {"messages": [llm.invoke(f"分析: {state['research_data']}")]}

# 构建状态图
graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("analyze", analyze_node)
graph.add_edge("research", "analyze")
graph.add_edge("analyze", END)
graph.set_entry_point("research")

# 挂载Postgres检查点 - 自动持久化
checkpointer = PostgresSaver.from_conn_string("postgresql://...")
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["analyze"],  # 人工介入点
)
# 执行 - thread_id实现会话隔离
result = app.invoke(
    {"messages": ["研究AI Agent发展趋势"]},
    config={"configurable": {"thread_id": "session-123"}},
)

from flink_agents.api import Agent, Event, Action

class RealTimeMonitorAgent(Agent):
    """实时监控Agent - 持续处理事件流"""
    def on_event(self, event: Event) -> list[Action]:
        # Flink保证exactly-once处理语义
        anomaly = self.llm_analyze(event.data)
        if anomaly.detected:
            return [Action(type="alert", payload=anomaly.report)]
        return []

    def llm_analyze(self, data: str):
        # LLM调用自动checkpoint
        return self.invoke_llm(f"分析异常: {data}")

<!-- BPMN定义Agent工作流 -->
<serviceTask id="analyze" name="AI分析Agent">
  <extensionElements>
    <zeebe:taskDefinition type="ai-agent" />
    <zeebe:taskHeaders>
      <zeebe:header key="agentType" value="analyzer" />
      <zeebe:header key="model" value="gpt-4" />
    </zeebe:taskHeaders>
  </extensionElements>
</serviceTask>

// Java工作者执行Agent任务
@ZeebeWorker(type = "ai-agent")
public void executeAgent(JobClient client, ActivatedJob job) {
    String prompt = job.getVariablesAsMap().get("input").toString();
    String result = llmClient.chat(prompt);
    client.newCompleteCommand(job.getKey())
          .variables(Map.of("result", result))
          .send().join();
}

[启动] → [搜索✓] → [分析✓] → 💥崩溃
                                    ↓ 恢复
[启动] → [搜索] → [分析] → [生成报告] → [完成]
         ↑重放跳过  ↑重放跳过   ↑从这里继续

Checkpoint 1          Checkpoint 2          Checkpoint 3
[State A] ──────────→ [State B] ──────────→ [State C]
                                              ↑ 崩溃恢复点