分布式事务 Saga 模式在多 Agent 协作写操作中的落地-平芜编程栈

分布式事务 Saga 模式在多 Agent 协作写操作中的落地

在多智能体系统（Multi-Agent System）从只读查询分析迈向**执行真实业务写操作（如电商下单、资金划转、机票酒店预订、发票开具）的过程中，分布式事务的“最终一致性与故障回滚”**成为了最严峻的架构挑战。

在大模型概率性决策与多微服务跨库调用的复杂场景下，传统的两阶段提交（2PC / XA）由于需要长时间持有数据库行级排他锁，在耗时数秒的大模型链路中会导致严重的死锁与性能瘫痪。

而在缺乏事务保障的系统中，经常发生致命的**“半吊子不一致故障（Inconsistent Partial Failures）”**：

Order_Agent成功在订单库创建了订单；
Payment_Agent成功扣减了用户的微信余额 500 元；
紧接着Hotel_Agent在调用第三方酒店 API 锁定房源时突然报错（库存不足）；
此时如果没有自动回滚补偿机制，用户的 500 元已经被扣除，而酒店房间却没有订上，引发严重的资损客诉！

借鉴微服务分布式事务经典理论中的Saga 模式（Saga Pattern），构建一套**“正向动作执行（Forward Actions） + 逆向补偿动作（Compensating Actions） + 状态机持久化协调器（Saga Orchestrator）”**的多智能体事务中枢，是保障多 Agent 跨系统写操作实现 100% 最终一致性的终极标准。

一、多 Agent 分布式事务 Saga 协调器全景拓扑模型

[ 用户下达指令: "帮我预订明天上海希尔顿酒店并完成支付" ] │ ▼ ┌────────────────────────────────────────────────────────────────────────┐ │ 多智能体 Saga 事务协调器 (Saga Orchestration Engine) │ ├────────────────────────────────────────────────────────────────────────┤ │ 步骤 1: 正向执行 [T1: 锁定酒店库存] ──► 成功 ✅ (登记补偿动作: C1) │ │ 步骤 2: 正向执行 [T2: 扣减资金账户] ──► 成功 ✅ (登记补偿动作: C2) │ │ 步骤 3: 正向执行 [T3: 触发短信通知] ──► ❌ 失败崩溃! │ └─────────────────────────┬──────────────────────────────────────────────┘ │ (触发 Saga 逆向补偿回滚流水线!) ▼ ┌────────────────────────────────────────────────────────────────────────┐ │ 逆向补偿阶段 (Backward Compensation Pipeline): │ │ ├── 1. 执行 [C2: 逆向原路退款 500 元] ──► 资金 100% 恢复原状 ✅ │ │ └── 2. 执行 [C1: 逆向释放酒店锁定库存] ──► 房源 100% 释放 ✅ │ └─────────────────────────┬──────────────────────────────────────────────┘ │ ▼ [ 系统安全归纳为终态: CANCELLED (0 资损、0 脏数据残留!) ]

二、生产级 Python 多 Agent Saga 事务协调器实现实操

import time from typing import List, Dict, Any, Callable from pydantic import BaseModel class SagaStep: def __init__( self, name: str, action_fn: Callable[[dict], dict], # 正向执行动作 compensate_fn: Callable[[dict], None] # 逆向补偿动作 (必须具备幂等性!) ): self.name = name self.action = action_fn self.compensate = compensate_fn class SagaOrchestrator: def __init__(self, saga_id: str): self.saga_id = saga_id self.steps: List[SagaStep] = [] self.executed_steps: List[SagaStep] = [] self.context: Dict[str, Any] = {} def add_step(self, step: SagaStep): self.steps.append(step) return self def execute_transaction(self, initial_payload: dict) -> bool: self.context = initial_payload.copy() print(f"🎬 【Saga 事务启动 💳】SagaID: [{self.saga_id}] | 总步骤数: {len(self.steps)}") # 1. 顺序执行正向动作 (Forward Execution) for step in self.steps: try: print(f" ▶ 正在执行正向动作: [{step.name}]...") output = step.action(self.context) self.context.update(output) self.executed_steps.append(step) except Exception as e: print(f"🚨 【步骤 [{step.name}] 发生故障 🛑】: {str(e)} ──► 立即启动逆向补偿回滚！") self._rollback_compensations() return False print(f"🎉 【Saga 事务圆满提交 ✅】所有步骤执行成功，最终一致性达成。") return True def _rollback_compensations(self): """逆向依序执行已完成步骤的补偿逻辑 (LIFO 栈式回滚)""" print(f"🔄 【启动逆向补偿回滚】共需补偿 {len(self.executed_steps)} 个已完成动作...") for step in reversed(self.executed_steps): try: print(f" ↩️ 正在执行逆向补偿: [{step.name}]...") step.compensate(self.context) print(f" ✅ 补偿成功: [{step.name}] 数据已恢复。") except Exception as ce: # 生产环境若补偿失败，必须记录至死信日志并触发人工紧急介入！ print(f"💥 【致命补偿异常】[{step.name}] 补偿失败: {ce}！")

三、真实业务场景下的 Saga 步骤组装实战

# 1. 定义酒店服务正向与补偿 def book_hotel(ctx: dict) -> dict: print(" [Hotel Agent] 成功锁定上海希尔顿大床房 1 晚") return {"hotel_booking_id": "HT_2026_9981"} def cancel_hotel_booking(ctx: dict): booking_id = ctx.get("hotel_booking_id") print(f" [Hotel Agent] 已释放房源锁定: {booking_id}") # 2. 定义支付服务正向与补偿 def deduct_payment(ctx: dict) -> dict: print(f" [Payment Agent] 成功从用户余额扣减 {ctx['amount']} 元") return {"payment_tx_id": "TX_500_OK"} def refund_payment(ctx: dict): print(f" [Payment Agent] 已向用户原路退款 {ctx['amount']} 元！") # 3. 组装并运行 Saga 事务 saga = SagaOrchestrator(saga_id="saga_order_20260912_01") saga.add_step(SagaStep("BookHotel", book_hotel, cancel_hotel_booking))\ .add_step(SagaStep("DeductPayment", deduct_payment, refund_payment)) # 启动事务 success = saga.execute_transaction({"user_id": "u1001", "amount": 500.0})