六大Multi-Agent状态持久化引擎：从崩溃失忆到确定性回放

State_v0 → [Agent_A] → State_v1 → [Agent_B] → State_v2 → [Agent_C] → State_v3

from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # Reducer: 追加而非覆盖
    current_agent: str
    shared_context: dict

# 构建图
graph = StateGraph(AgentState)
graph.add_node("researcher", researcher_agent)
graph.add_node("writer", writer_agent)
graph.add_node("reviewer", reviewer_agent)
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")

# 挂载PostgreSQL Checkpointer
async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
    await checkpointer.setup()
    compiled = graph.compile(checkpointer=checkpointer)

    # 执行——每个节点执行后自动保存Checkpoint
    config = {"configurable": {"thread_id": "user-123"}}
    result = await compiled.ainvoke(
        {"messages": [("user", "分析2026年AI Agent市场趋势")]},
        config=config
    )

    # 崩溃恢复——从最近Checkpoint继续
    # 重启后用相同thread_id调用即可
    resumed = await compiled.ainvoke(None, config=config)

graph TD
    A[Workflow启动] --> B[Event History追加]
    B --> C{执行Agent节点}
    C -->|需要LLM调用| D[发起Activity]
    D -->|崩溃!| E[Temporal重启Worker]
    E --> F[重放Event History]
    F --> G[到达崩溃点继续执行]
    C -->|Agent完成| H[状态转换事件]
    H --> B
    G --> C

from temporalio import workflow, activity
from datetime import timedelta

@activity.defn
async def call_research_agent(query: str) -> dict:
    """Activity: 可重试、可超时的LLM调用"""
    response = await llm_client.chat(query)
    return {"findings": response.content, "sources": response.citations}

@activity.defn
async def call_writer_agent(research: dict) -> str:
    """Activity: 写作Agent"""
    prompt = f"基于以下研究写一篇分析文章:\n{research['findings']}"
    response = await llm_client.chat(prompt)
    return response.content

@workflow.defn
class MultiAgentResearchWorkflow:
    """持久执行的Multi-Agent工作流"""

    @workflow.run
    async def run(self, query: str) -> str:
        # 1. 研究Agent（崩溃后自动重试，最多3次）
        research = await workflow.execute_activity(
            call_research_agent, query,
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=RetryPolicy(maximum_attempts=3)
        )

        # 2. 写作Agent
        article = await workflow.execute_activity(
            call_writer_agent, research,
            start_to_close_timeout=timedelta(minutes=10)
        )

        # 3. 审核——支持人工审批（可持续数天）
        approved = await workflow.execute_activity(
            human_review, article,
            start_to_close_timeout=timedelta(days=7)  # 等待人工!
        )

        return approved

# 任何时刻崩溃，重启后自动从Event History恢复
# 无需任何额外代码！

import * as restate from "@restatedev/restate-sdk";

const agentWorkflow = restate.workflow({
  name: "multiAgentResearch",
  handlers: {
    run: async (ctx: restate.WorkflowContext, query: string) => {
      // 持久化的LLM调用——崩溃后自动重放
      const research = await ctx.run(() => 
        callResearchAgent(query)
      );

      // 持久化状态——自动写入Journal
      ctx.set("research_result", research);

      // 持久化定时器——可设置小时/天级别等待
      const writerResult = await ctx.run(() =>
        callWriterAgent(research)
      );

      // 支持Signal——外部事件注入
      const humanApproval = await ctx.promise("approval");

      return writerResult;
    },

    // 外部可随时发送Signal
    approve: async (ctx: restate.WorkflowSharedContext) => {
      ctx.resolve("approval", true);
    }
  }
});

graph TB
    subgraph "CrewAI统一记忆架构"
        A[短期记忆<br/>当前对话上下文] --> B[长期记忆<br/>跨会话持久化]
        B --> C[实体记忆<br/>特定实体/用户画像]
    end
    subgraph "Flows状态层"
        D[Flow State<br/>工作流级别共享状态]
        D --> A
    end
    subgraph "存储后端"
        E[Mem0 / ChromaDB] --> B
        F[PostgreSQL] --> D
    end

from crewai import Agent, Crew, Task
from crewai.flow import Flow, listen, start, router

class ResearchFlow(Flow):
    """事件驱动的Multi-Agent研究工作流"""

    @start()
    def research(self):
        """触发研究Agent"""
        crew = Crew(
            agents=[research_agent],
            tasks=[research_task]
        )
        result = crew.kickoff()
        # 写入共享State
        self.state["research_findings"] = result
        return result

    @listen("research")
    def write_and_review(self, research_result):
        """监听研究完成，触发写作+审核"""
        # 共享State在多个Crew间传递
        context = self.state.get("research_findings", {})

        writing_crew = Crew(agents=[writer_agent], tasks=[write_task])
        article = writing_crew.kickoff(inputs={"context": context})

        review_crew = Crew(agents=[reviewer_agent], tasks=[review_task])
        final = review_crew.kickoff(inputs={"article": article})

        self.state["final_article"] = final
        return final

flow = ResearchFlow()
result = flow.kickoff()

Agent A: increment("task_count", 1)  ──→  CRDT State
Agent B: increment("task_count", 2)  ──↗     ↓
                                          最终: task_count = 3
                                          （无论执行顺序）

from pydantic import BaseModel
from crdt_lib import LWWRegister, GCounter, ORMap

class MultiAgentSharedState(BaseModel):
    """基于CRDT的Multi-Agent共享状态"""

    # Last-Writer-Wins Register: 最后写入胜出
    current_phase: LWWRegister[str] = LWWRegister("idle")

    # Grow-only Counter: 单调递增计数器
    tasks_completed: GCounter = GCounter()

    # Observed-Remove Map: 可增删的字典
    agent_results: ORMap = ORMap()

    def apply_agent_update(self, agent_id: str, update: dict):
        """Agent并行更新，自动冲突消解"""
        if "phase" in update:
            self.current_phase.set(update["phase"], timestamp=update["ts"])

        if "task_done" in update:
            self.tasks_completed.increment(agent_id, 1)

        if "result" in update:
            self.agent_results.set(agent_id, update["result"])

        # 无需任何锁或协调！
        # 不同Agent的更新可以以任何顺序合并
        # 最终状态数学保证一致

graph LR
    subgraph "Agent团队"
        A1[研究Agent]
        A2[分析Agent]
        A3[写作Agent]
        A4[审核Agent]
    end

    subgraph "黑板 Blackboard"
        B1[问题定义区]
        B2[事实收集区]
        B3[分析结论区]
        B4[草稿区]
        B5[最终输出区]
    end

    A1 -->|写入事实| B2
    B2 -->|触发| A2
    A2 -->|写入结论| B3
    B3 -->|触发| A3
    A3 -->|写入草稿| B4
    B4 -->|触发| A4
    A4 -->|修改草稿| B4
    A4 -->|最终输出| B5

# 使用 blackboard-core 库
from blackboard_core import Blackboard, Agent, Controller

class ResearchBlackboard(Blackboard):
    """Multi-Agent共享工作空间"""
    levels = {
        "problem": {},    # 问题定义
        "facts": [],      # 收集的事实
        "analysis": {},   # 分析结论
        "draft": "",      # 文章草稿
        "final": ""       # 最终输出
    }

class FactCollector(Agent):
    """监控problem区，写入facts区"""
    watch = ["problem"]
    writes_to = ["facts"]

    def execute(self, blackboard):
        problem = blackboard.read("problem")
        facts = call_llm(f"收集关于以下问题的事实: {problem}")
        blackboard.write("facts", facts, author=self.name)

class Analyst(Agent):
    """监控facts区，写入analysis区"""
    watch = ["facts"]
    writes_to = ["analysis"]

    def execute(self, blackboard):
        facts = blackboard.read("facts")
        analysis = call_llm(f"分析以下事实: {facts}")
        blackboard.write("analysis", analysis, author=self.name)

# 控制器自动调度——哪个Agent的watch区域被更新了，就触发那个Agent
controller = Controller(
    blackboard=ResearchBlackboard(),
    agents=[FactCollector(), Analyst(), Writer(), Reviewer()]
)

# 写入初始问题，触发整个流水线
controller.start(problem="分析2026年AI Agent市场趋势")

graph TD
    A[Multi-Agent状态管理需求] --> B{需要确定性回放?}
    B -->|是| C{团队语言栈?}
    B -->|否| D{Agent数量?}

    C -->|Python| E[LangGraph Checkpointer<br/>+ PostgreSQL]
    C -->|TypeScript/Java| F{运维资源?}

    F -->|充足| G[Temporal<br/>企业级持久执行]
    F -->|有限| H[Restate<br/>轻量持久执行]

    D -->|<5个Agent| I{需要快速上线?}
    D -->|5-50个Agent| J{高并发写入?}
    D -->|50+Agent| K[Temporal + Redis<br/>分片Event History]

    I -->|是| L[CrewAI Flows<br/>事件驱动+统一记忆]
    I -->|否| E

    J -->|是| M[CRDT Shared State<br/>无冲突复制]
    J -->|否| N[Blackboard<br/>共享工作空间]

    style E fill:#4CAF50
    style G fill:#2196F3
    style H fill:#FF9800
    style L fill:#9C27B0
    style M fill:#F44336
    style N fill:#607D8B