Multi-Agent事务一致性：六大Saga补偿引擎实战拆解

from agent_hypervisor import SagaOrchestrator, SagaStep

saga = SagaOrchestrator(
    steps=[
        SagaStep(
            name="book_flight",
            agent=flight_agent,
            compensation="cancel_flight",  # 补偿合约
            timeout=30,
            retry_policy={"max_attempts": 3, "backoff": "exponential"}
        ),
        SagaStep(
            name="book_hotel",
            agent=hotel_agent,
            compensation="cancel_hotel",
            timeout=45,
            validation_fn=validate_booking  # 输出验证函数
        ),
        SagaStep(
            name="send_confirmation",
            agent=email_agent,
            compensation="send_cancellation_notice",
            timeout=20
        )
    ],
    on_compensation_failure="alert_human",  # 补偿失败的兜底策略
    max_compensation_cost_usd=0.50          # 补偿成本上限
)

result = await saga.execute(travel_request)

from sagallm import SagaWorkflow, AgentStep, Validator, Compensator

class BookingValidator(Validator):
    """确定性验证器——不依赖LLM"""
    def validate(self, output: dict) -> bool:
        return (
            "confirmation_code" in output
            and len(output["confirmation_code"]) == 6
            and output.get("status") == "confirmed"
        )

class BookingCompensator(Compensator):
    """补偿动作——可以是LLM Agent或确定性API"""
    async def compensate(self, step_output: dict) -> dict:
        return await cancel_booking_api(step_output["confirmation_code"])

workflow = SagaWorkflow([
    AgentStep(agent=flight_agent, validator=FlightValidator(),
              compensator=FlightCompensator()),
    AgentStep(agent=hotel_agent, validator=HotelValidator(),
              compensator=HotelCompensator()),
])

from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres import PostgresSaver

# 配置PostgreSQL持久化
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://localhost:5432/agent_saga"
)

graph = StateGraph(BookingState)
graph.add_node("search_flights", search_flights_node)
graph.add_node("book_flight", book_flight_node)
graph.add_node("book_hotel", book_hotel_node)
graph.add_node("compensate_flight", cancel_flight_node)
graph.add_node("compensate_hotel", cancel_hotel_node)

# 条件边：失败时跳转到补偿节点
graph.add_conditional_edges(
    "book_hotel",
    lambda state: "compensate_flight" if state["hotel_error"] else "confirm",
    {
        "confirm": "send_confirmation",
        "compensate_flight": "compensate_flight",
    }
)

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["book_flight"]  # 人工审批门控
)

# 执行时自动快照
config = {"configurable": {"thread_id": "booking-123"}}
result = await app.ainvoke(input_data, config)

# 失败后从检查点恢复
await app.ainvoke(None, config)  # 从上次中断处继续

from temporalio import workflow
from temporalio.workflow import saga

@workflow.defn
class TravelBookingWorkflow:
    @workflow.run
    async def run(self, request: TravelRequest) -> BookingResult:
        # Temporal原生Saga支持
        async with saga.Saga(
            compensation_timeout=timedelta(seconds=30),
            parallel_compensation=False  # 补偿按逆序串行执行
        ) as s:
            # 每个execute_with_compensation注册补偿动作
            flight_result = await s.execute_with_compensation(
                # 正向动作
                workflow.execute_child_workflow(
                    FlightBookingWorkflow.run, request.flight
                ),
                # 补偿动作
                workflow.execute_child_workflow(
                    FlightCancelWorkflow.run,
                    ArgRef(0)  # 引用正向动作的输出
                )
            )

            hotel_result = await s.execute_with_compensation(
                workflow.execute_child_workflow(
                    HotelBookingWorkflow.run, request.hotel
                ),
                workflow.execute_child_workflow(
                    HotelCancelWorkflow.run, ArgRef(0)
                )
            )

            return BookingResult(flight=flight_result, hotel=hotel_result)

import restate
from restate import VirtualObject, Workflow

booking_service = VirtualObject("BookingService")

@booking_service.handler()
async def book_trip(ctx: restate.Context, request: TravelRequest):
    # 幂等LLM调用——崩溃重启不会重复执行
    flight_plan = await ctx.run(
        "plan-flight",
        lambda: call_llm(f"Plan flight: {request.destination}")
    )

    # 幂等API调用——带自动重试
    booking = await ctx.run(
        "book-flight",
        lambda: book_flight_api(flight_plan),
        retry_policy=restate.RetryPolicy(max_attempts=3)
    )

    # 调用另一个Agent（也是Restate服务）
    hotel_result = await ctx.service_call(
        HotelAgentService.book,
        HotelRequest(destination=request.destination, dates=request.dates)
    )

    return BookingResult(flight=booking, hotel=hotel_result)

from veritas_stream import EventStore, AgentEvent

store = EventStore(backend="redis")

# 记录Agent决策事件
await store.append(AgentEvent(
    agent_id="flight-agent-001",
    event_type="tool_call",
    tool_name="book_flight",
    input={"route": "SFO->NRT", "date": "2026-06-15"},
    output={"confirmation": "UA-88372", "cost": "$1,240"},
    reasoning="User requested non-stop flight, UA-837 is the only option",
    timestamp=datetime.utcnow(),
    parent_event_id="task-001-decompose"  # 因果链追踪
))

# 回放某个Agent的所有决策
events = await store.replay(agent_id="flight-agent-001")
for event in events:
    print(f"{event.timestamp}: {event.event_type} -> {event.tool_name}")
    print(f"  Reasoning: {event.reasoning}")
    print(f"  Cost: ${event.output.get('cost', 'N/A')}")

graph TB
    subgraph "第一类：运行时治理层"
        MS[Microsoft Agent Governance Toolkit<br/>Hypervisor级资源隔离]
        MS -->|Saga Orchestrator| MS_SAGA[编排式Saga<br/>补偿合约声明式]
        MS -->|Policy Engine| MS_POLICY[合规策略<br/>审批门控]
    end

    subgraph "第二类：持久执行层"
        TMP[Temporal<br/>通用持久执行]
        TMP -->|Durable Execution| TMP_DE[确定性重放<br/>崩溃恢复]
        TMP -->|Native Saga| TMP_SAGA[原生Saga编排<br/>逆序补偿]

        RST[Restate<br/>轻量持久执行]
        RST -->|Idempotent Run| RST_IDE[幂等执行<br/>无需独立集群]
    end

    subgraph "第三类：框架内建"
        LG[LangGraph Checkpoint<br/>状态图快照]
        LG -->|Subgraph| LG_SUB[事务边界隔离<br/>条件边补偿]
        LG -->|Checkpoint| LG_CP[PostgreSQL持久化<br/>断点恢复]
    end

    subgraph "第四类：审计溯源层"
        VS[Veritas-Stream<br/>不可变事件存储]
        VS -->|Event Sourcing| VS_ES[因果链追踪<br/>不可变审计]
    end

    subgraph "第五类：学术理论"
        SL[SagaLLM<br/>事务语义形式化]
        SL -->|Validator| SL_V[确定性验证<br/>剥离LLM不确定性]
        SL -->|Compensator| SL_C[补偿完整保证<br/>递归验证]
    end

    style MS fill:#4CAF50,color:#fff
    style TMP fill:#FF9800,color:#fff
    style RST fill:#FF5722,color:#fff
    style LG fill:#2196F3,color:#fff
    style VS fill:#9C27B0,color:#fff
    style SL fill:#607D8B,color:#fff

flowchart TD
    START[你的Multi-Agent系统<br/>需要事务一致性] --> Q1{Agent数量?}

    Q1 -->|≤5个Agent| Q2{失败代价?}
    Q1 -->|>5个Agent| Q3{有合规要求?}

    Q2 -->|低成本,可容忍丢失| A1[LangGraph Checkpoint<br/>最快上手,内置方案]
    Q2 -->|高成本,不可丢失| A2[Restate<br/>轻量持久化,30分钟集成]

    Q3 -->|是 金融/医疗/政务| Q4{预算?}
    Q3 -->|否| A3[Temporal + LangGraph<br/>两层架构,生产级]

    Q4 -->|充足| A4[Temporal + Agent Governance Toolkit<br/>完整治理栈]
    Q4 -->|有限| A5[LangGraph + Veritas-Stream<br/>最小化合规成本]

    style A1 fill:#2196F3,color:#fff
    style A2 fill:#FF5722,color:#fff
    style A3 fill:#FF9800,color:#fff
    style A4 fill:#4CAF50,color:#fff
    style A5 fill:#9C27B0,color:#fff