我是叶秋。如果你在过去的几个月里一直在关注多智能体框架的演进，你会注意到一个明显的断层：市面上的教程还在教你怎么让两个Agent互道早安，而生产环境里，我们已经在用Agent集群处理跨越四个微服务、涉及三个外部SaaS系统的合同审批流了。这中间的鸿沟，不是靠增加几个API调用就能填平的。
我花了整整160个小时，在同一套业务逻辑上跑了三个框架——LangGraph 0.2.0、CrewAI latest、Google ADK 0.2.3——不是为了写测评，而是为了回答一个让我连续失眠的问题：当审批流里的第4个Agent因为下游发票系统超时而宕掉，而前3个Agent已经分别扣减了预算、锁定了库存、更新了供应商状态之后，谁来收拾这个烂摊子？微服务架构里我们用分布式事务协调器（TCC）或者Saga模式来解决的问题，在多智能体编排里，对应的解是什么？
答案比我想象的更丑陋，也更有趣。这篇文章就是一次彻底的技术复盘。
30秒速览
- - LangGraph的检查点机制RTO在秒级，适用于人机协同长运行流，不适合高吞吐自动审批
- - CrewAI共享记忆有300-800ms的写入延迟，高并发下背压机制会丢弃写入，导致补偿链静默失败
- - ADK的事件溯源+显式补偿链架构正确，但跨区域Firestore延迟可能达400ms，需要事件粒度优化
- - 压测数据显示ADK的失败请求恢复成功率达82%（RTO<500ms），远高于LangGraph的23%和CrewAI的47%
- - 最终落地需要显式Saga协调器，代价是代码量增加3倍和所有外部调用的幂等性改造
一次审批链挂了三个节点，我才意识到多Agent系统的一致性需求根本不是ACID
先看一个真实的失败场景。这是一条企业差旅审批链，在LangGraph上构建：
- Agent A：校验差旅政策（同步，调用内部规则引擎）
- Agent B：冻结部门预算（半异步，调用财务系统gRPC）
- Agent C：预订酒店（异步长事务，调用外部SaaS接口，平均耗时4.7秒）
- Agent D：生成行程单并发送邮件（同步，调用通知服务）
事故发生在某个周二的下午14:23。Agent B成功冻结了8500元预算，但Agent C在调用酒店SaaS接口时，对方返回了HTTP 502——不是彻底的连接失败，而是经过30秒超时后的网关错误。此时状态已经分裂：预算中心的冻结记录存在，但酒店系统里没有订单。更致命的是，LangGraph的默认图执行器把这视为“节点失败”，直接停止了图遍历——没有回滚，没有补偿，那8500元的冻结额度在接下来的72小时里成为一笔“幽灵资金”，直到财务月底对账才发现。（延伸阅读：我把代码重构的AI赌注押在JetBrains AI Assistant上：一个后端架构师的三个月实战复盘）
这就是多智能体审批的典型失败模式，我总结了五类：
| 失败类型 | 触发条件 | 状态后果 | 发生频率（按我的压测数据） |
|---|---|---|---|
| 中间节点超时 | 第三方API响应>阈值 | 前置操作已提交，后续操作中断 | 23%的失败率 |
| 部分失败 | 并行分支中某个Agent失败 | 已完成分支无法回滚 | 17% |
| 重复执行 | 重试机制缺乏幂等性保证 | 重复扣款/多次发邮件 | 11% |
| 悬挂事务 | Agent崩溃或无响应 | 锁定的资源永不释放 | 8% |
| 状态分歧 | 检查点与系统状态不一致 | 恢复后的决策基于过时数据 | 15% |
看清楚了吗？ACID在这里是行不通的。你没法要求一个长达30秒的酒店预订操作遵循原子性——对方系统根本不认识你的事务边界。多智能体系统的一致性需求，更接近微服务架构中的最终一致性，需要的不是锁，而是Saga模式里的补偿交易。
棋局解读：为什么Saga模式正在成为Agent事务的暗线标准
这盘棋走到现在，三方都在暗中布局：①LangChain在LangGraph 0.2中悄悄加入了checkpoint的pending_writes队列，这本质上是Saga协调器里的“操作日志”；②CrewAI用共享记忆（Shared Memory）做背压控制，这是用事件溯源（Event Sourcing）的思路解决状态同步；③Google ADK最激进，直接把事件触发器设计为状态机，每一步状态变更都记录在Event Store里。它们选了不同的切入点，但底层的共识已经形成——在多Agent长事务中，必须有一个明确的补偿路径。我判断接下来的三个月，这三家都会推出显式的Saga编排API，不再是现在这种“用底层机制间接实现”的方式。如果我错了，那说明LangChain、CrewAI和Google同时决定牺牲审批场景这块蛋糕——这在企业级市场争夺战中几乎不可能。
LangGraph的检查点不是银弹，我差点在生产日志里淹死
LangGraph对状态管理的思路很清晰：每个super-step执行后，状态自动保存到检查点（Checkpoint）。你可以配置SqliteSaver或者Postgres后端。官方文档把这描述为“容错机制”，但我踩过的坑告诉我，这只解决了状态保存，没有解决状态修复。
条件边里的重试逻辑，我写了三版才绕过悬挂问题
第一版重试策略是直接在节点内部用Python重试装饰器：
# 版本1：天真版节点内重试
import tenacity
@tenacity.retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def agent_c_hotel_booking(state):
result = hotel_saas_api.book(state['hotel_id'], state['room_type'])
if result.status_code != 200:
raise BookingException("Failed to book hotel")
state['booking_ref'] = result.json()['ref_id']
return state
问题在哪？如果hotel_saas_api.book已经成功创建了订单，但在返回HTTP响应时网络抖动导致超时，这个重试会创建重复订单。这是典型的没有幂等性保证的重试。更隐蔽的问题是，这个装饰器在LangGraph的节点执行上下文中是黑盒——主线程看不到重试次数，也无法在三次失败后触发补偿逻辑。
第二版我用了条件边（Conditional Edges）实现图级别的重试路由：
# 版本2：条件边控制重试
workflow = StateGraph(ApprovalState)
workflow.add_node("agent_b", agent_b_freeze_budget)
workflow.add_node("agent_c", agent_c_book_hotel)
workflow.add_node("compensate_b", compensate_b_unfreeze_budget)
workflow.add_node("terminal_failed", terminal_node)
workflow.add_conditional_edges(
"agent_b",
check_agent_b_status,
{
"success": "agent_c",
"retry": "agent_b", # 重试
"fatal": "compensate_b" # 触发补偿
}
)
workflow.add_conditional_edges(
"agent_c",
check_agent_c_status_with_counter,
{
"success": "agent_d",
"retry": "agent_c", # 重试，但有计数限制
"fatal": "compensate_b" # 回滚到B的补偿节点
}
)
这段代码看起来合理，但运行一个月后我发现一个致命缺陷：当图恢复时（比如服务重启），条件边里的重试计数器并没有持久化。如果Agent C刚刚执行完第三次重试，节点返回了“fatal”信号，但在条件边函数check_agent_c_status_with_counter执行之前，进程崩溃了，那么下次图从检查点恢复时，会重新执行Agent C——因为检查点保存的是执行Agent C之前的状态，而重试计数器是运行时变量。这导致一条订单一共被执行了5次，客户收到了3封行程确认邮件。
第三版我放弃了对图内置重试的依赖，转而把重试状态和补偿逻辑都显式地放在State对象中：
# 版本3：将重试与补偿状态显式化
from typing import TypedDict, Annotated
from langgraph.checkpoint import BaseCheckpointSaver
class ApprovalState(TypedDict):
budget_frozen: bool
hotel_booked: bool
booking_ref: str
retry_counts: dict[str, int] # 节点->重试次数
compensation_stack: list[callable] # 补偿函数栈
def agent_c_book_hotel(state: ApprovalState) -> ApprovalState:
max_retries = 3
current_retries = state['retry_counts'].get('agent_c', 0)
if current_retries >= max_retries:
state['hotel_booked'] = False
state['status'] = 'requires_compensation'
return state
try:
state['retry_counts']['agent_c'] = current_retries + 1
result = hotel_saas_api.book(
state['hotel_id'],
state['room_type'],
idempotency_key=state['idempotency_key'] # 幂等性保证
)
state['hotel_booked'] = True
state['booking_ref'] = result['ref_id']
state['compensation_stack'].append(
lambda: hotel_saas_api.cancel(state['booking_ref'])
)
except TimeoutError:
state['status'] = 'retry_or_compensate'
return state
# 使用SqliteSaver确保重试计数器也持久化
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = workflow.compile(checkpointer=checkpointer)
这个版本的关键改进在于：补偿栈作为状态的一部分持久化到Sqlite里。当Agent C超时后服务重启，检查点恢复的状态会包含之前已经入栈的所有补偿函数引用。我在实际调试中发现，函数引用的序列化是个大坑——Lambda无法pickle——所以最终换成了字符串标记+执行时的路由表。
检查点恢复的实测数据：RTO远不如预想
我用Locust压测了基于LangGraph Checkpoint的审批链，模拟10%的Agent C超时场景（每次超时后等待5秒然后从检查点恢复）：
- 检查点保存开销：平均23ms（Postgres后端，State对象2.4KB）
- 恢复时间目标（RTO）：从检查点恢复到继续执行，平均1.7秒——远高于微服务里典型的事务回滚（通常<100ms）
- 幽灵分支：在1000次压测中，出现了31次“从检查点恢复后状态与外部系统不一致”的情况，原因是检查点保存时外部酒店SaaS已经返回了成功，但在
return state之前崩溃，恢复后重试导致重复预订
数据很诚实：LangGraph的检查点机制，设计初衷是支持人机协同（human-in-the-loop）的长运行工作流，不是高强度事务场景。它的RTO在秒级别，而我的审批链要求RTO<500ms。（延伸阅读：Google ADK这把轻量级快刀，正在切开LangGraph没啃下的审批流骨头）
CrewAI用共享记忆做背压，但当六个Agent同时写状态时，竞态条件比你想的更脏
CrewAI采用了完全不同的架构路径。它没有LangGraph那种显式的图定义，而是通过Task、Agent间的委托（Delegation）和共享记忆实现协作。我原以为这会简化状态管理——毕竟所有Agent写同一块内存——实际测试的结果是，共享记忆在多写者场景下的竞态条件复杂度，是我见过的Python并发问题里最棘手的之一。
委托链里的状态传播延迟，让一个审批流变成了薛定谔的猫
这是我搭建的审批Crew配置：
from crewai import Agent, Task, Crew, Process
from crewai.memory import ShortTermMemory, LongTermMemory
# 构建Agent
approval_agent = Agent(
role='Approval Coordinator',
goal='Orchestrate entire approval chain',
backstory='Tracks overall process state',
allow_delegation=True,
verbose=True
)
budget_agent = Agent(
role='Budget Controller',
goal='Freeze department budget and record it in shared memory',
backstory='Must update shared memory immediately after freezing',
allow_delegation=False
)
hotel_agent = Agent(
role='Hotel Booker',
goal='Reserve hotel via external API',
backstory='Needs to check shared memory before and after booking',
allow_delegation=False
)
# 任务定义——依赖关系通过上下文传递
task_budget = Task(
description="Freeze budget of 8500 CNY and write the freeze_id to shared memory",
agent=budget_agent,
expected_output="Budget frozen with freeze_id"
)
task_hotel = Task(
description="Read freeze_id from shared memory. Book hotel. If booking fails, trigger unfreeze of that freeze_id.",
agent=hotel_agent,
context=[task_budget], # 声明上下文依赖
expected_output="Hotel booked with ref_id"
)
crew = Crew(
agents=[approval_agent, budget_agent, hotel_agent],
tasks=[task_budget, task_hotel],
process=Process.sequential,
memory=True,
short_term_memory=ShortTermMemory(),
long_term_memory=LongTermMemory()
)
result = crew.kickoff()
这段配置运行的前20次一切正常。第21次，酒店预订失败，hotel_agent读取共享记忆尝试获取freeze_id来解冻预算——但读到的值为None。排查日志后发现，budget_agent把冻结ID写进了自己的输出（result.output），但在共享记忆（ShortTermMemory）的写入有一个不可见的异步延迟，大约300-800ms。而hotel_agent在启动时构建上下文（context=[task_budget]）读取的并不是共享记忆，而是budget_agent的直接输出。当hotel_agent在booking失败后的补偿逻辑里再次查询共享记忆时，依赖的是一个尚未传播的状态。
这不是bug，是CrewAI的共享记忆模型在设计上没有保证写入的即时可见性。在LangGraph的State里，状态变更在同一个step内是同步的、强一致的。但在CrewAI里，共享记忆更接近一个最终一致性的数据存储——它适用于知识积累（比如让三个Agent都记下“客户偏好靠窗座位”），但不适用于事务补偿链（比如“我现在就要知道刚才冻结的ID是多少”）。
背压控制的副作用：当写请求积压，整个Crew开始降级
我在压测中发现，当并发审批请求达到8个Crew同时运行时，共享记忆的写入开始出现排队。CrewAI内部有一个背压机制——当写入队列长度超过阈值，新的写入会被丢弃，记录一条警告日志而不是抛出异常。这意味着什么？意味着高负载下，补偿操作可能因为无法写入日志而“静默失败”。我在监控里看到这样的循环：
- Budget Agent成功冻结，写入共享记忆——被背压丢弃
- Hotel Agent调用失败，读取共享记忆找freeze_id——读不到
- Hotel Agent以为自己不需要补偿，流程结束
- 30分钟后，财务系统里多了一笔冻结的预算，无人认领
这和LangGraph的问题形成鲜明对比：LangGraph的问题是RTO太长，CrewAI的问题是状态传播不可靠。一个慢，一个不准确。在分布式系统里，我们通常选慢但准确，但在多Agent场景里，用户等不了。（延伸阅读：我往 Gemini 1.5 Pro 里塞了 5 万行代码，它给我画了张循环依赖图，还顺手把重构 diff 写好了——但我差点被账单送走）
ADK的状态机设计，第一次让我看到Agent事务可以不靠“补丁”
Google ADK（Agent Development Kit）是最晚进入这场竞争的产品，但它的设计思路明显借鉴了前人的坑。ADK没有图的概念，也没有Crew的层级委托，它把每个Agent都设计为一个事件驱动的状态机。状态变更通过Event触发器传播，所有事件都被持久化到外部的Event Store。
事件溯源 + 补偿链的完整实验
我按照ADK的文档重构了审批链——这其实是我的第四次重构，前三版都在另外两个框架上打转。ADK的方式完全不一样：
# ADK状态定义——每个阶段都是确定性的状态
from google.adk import Agent, State, Event, Trigger, CompensationChain
class BudgetFrozen(State):
freeze_id: str
amount: float
class HotelReserved(State):
booking_ref: str
hotel_name: str
class BookingFailed(State):
error_code: str
requires_compensation: bool = True
# 定义补偿链——这是我在其他两个框架里找了四个星期都没找到的东西
compensation = CompensationChain()
compensation.register(
trigger_state=BookingFailed,
compensate_action=lambda ctx: unfreeze_budget(ctx['freeze_id']),
on_success=State("BudgetUnfrozen"),
on_failure=State("CompensationFailed_EscalateToHuman")
)
# 定义Agent，每个Agent都是纯状态转换函数
budget_agent = Agent(
name="budget_freezer",
initial_state="Idle",
states=[
State("Idle"),
State("Processing"),
BudgetFrozen,
State("Error")
],
transitions=[
Trigger("start_freeze", from_state="Idle", to_state="Processing"),
Trigger("freeze_success", from_state="Processing", to_state=BudgetFrozen),
Trigger("freeze_failure", from_state="Processing", to_state="Error")
],
# 关键：事件持久化到外部Store
event_store=FirestoreEventStore(collection="agent_events")
)
hotel_agent = Agent(
name="hotel_booker",
initial_state="Idle",
states=[
State("Idle"),
HotelReserved,
BookingFailed
],
transitions=[
Trigger("start_booking", from_state="Idle", to_state="Processing"),
Trigger("booking_success", to_state=HotelReserved),
Trigger("booking_timeout", to_state=BookingFailed)
],
event_store=FirestoreEventStore(collection="agent_events"),
compensation_chain=compensation # 补偿逻辑绑定在Agent上
)
这段代码跑了1000次，0次状态分歧。为什么？因为ADK的状态转换是原子操作——从BudgetFrozen到BookingFailed的转换，要么全部记录在Event Store，要么完全不可见。没有中间状态。当补偿链被触发时，unfreeze_budget函数从Event Store中读取freeze_id——不是从运行时内存，是从已经持久化的事件日志。这意味着即使hotel_agent在执行补偿前崩溃，重启后Event Store里已经有了一条BudgetFrozen事件和一条BookingFailed事件，补偿逻辑可以从中断点继续执行。
这是我在整个实验里唯一一次感到“架构对了”的时刻。
触发器延迟的实验数据，暴露了新问题
但ADK也不是没有代价。我把Event Store放在Firestore上（Google推荐的后端），在跨区域部署时，触发器延迟的数据如下：
- 同Region：事件触发到Agent响应，中位数31ms，P99 87ms
- 跨Region（us-central1到asia-east1）：中位数173ms，P99 402ms
- 补偿链执行：从触发到补偿完成，中位数2.1秒——因为补偿操作需要从Firestore读取完整事件历史
跨区域的延迟比同区域高了5.5倍，这在全球部署的场景下是个硬伤。我咨询了Google Cloud的解决方案架构师（在一次线下meetup上，不是官方渠道），得到的信息是“Firestore的多区域同步目前是最终一致性模型，ADK团队正在评估Cloud Spanner作为替代方案”。如果我需要在北京时间下午2点处理一个审批，而Event Store在新加坡，这个延迟可能意味着用户要盯着loading动画等2秒——这在审批体验上是不可接受的。（延伸阅读：我让Cursor写了一套KEDA规则和Spot切换器，推理成本从8万暴跌到1.7万——但挂了两次生产）
三框架横向压测：当我把50条审批链同时砸上去，数据说了我没想到的话
我搭建了一个统一的压测环境。三条相同的审批链分别运行在LangGraph 0.2.0（Postgres检查点后端）、CrewAI latest（Redis共享记忆后端）、ADK 0.2.3（Firestore事件存储）。压测工具是Locust，模拟50个并发用户持续提交审批请求，每个审批链包含4个Agent节点，其中Agent C有15%的概率超时（模拟外部SaaS不稳定）。运行时长：30分钟。
关键指标对比
| 指标 | LangGraph | CrewAI | ADK |
|---|---|---|---|
| 请求总完成数 | 2873 | 3412 | 2561 |
| 平均响应时间（P50） | 4.2s | 2.8s | 5.1s |
| P99响应时间 | 14.7s | 11.3s | 8.9s |
| 失败请求占比 | 12.4% | 8.1% | 3.7% |
| 其中：系统错误（5xx） | 5.8% | 4.2% | 1.1% |
| 其中：状态分歧（重复/丢失） | 6.6% | 3.9% | 2.6% |
| 恢复成功（RTO<500ms内补偿完成） | 23% | 47% | 82% |
| 最大吞吐量（请求/秒） | 14.2 | 18.7 | 10.3 |
| 资源消耗（CPU/内存均值） | 58% / 3.2GB | 72% / 4.7GB | 41% / 2.1GB |
这三列数据放在一起，有一个我一开始没注意到的模式：CrewAI胜在吞吐量和平均响应时间，因为它没有显式的检查点开销，Agent之间通过共享内存通信极快。但ADK胜在恢复成功率——82%的失败请求在500ms内完成了补偿，而LangGraph只有23%。CrewAI的恢复成功率是47%，看起来不错，但那是因为我手动在共享记忆上加了Redis持久化和TTL——如果用默认的内存后端，这个数字会掉到20%以下。
这意味着什么？如果你的审批链对延迟敏感、对偶发失败容忍度较高（比如内部工单审批），CrewAI是三个里面最快的。如果你做金融级审批，一次失败就可能造成资金损失，ADK的事件溯源模型是唯一合理的选择。LangGraph夹在中间——它的检查点机制让它在复杂图遍历（比如带有人工审批节点）时有优势，但在纯自动化的高可靠性场景里，性能和安全两头不靠岸。
一次调试中发现的隐藏瓶颈
在跑ADK压测的前十分钟，我发现一个奇怪的现象：P99延迟在稳步上升，从6秒涨到了22秒。排查发现不是Agent执行变慢，而是Firestore的写入操作在积累——ADK为每个状态转换写入一条事件记录，30分钟产生了约12万条文档。Firestore的写入吞吐有配额限制（默认1000次写入/秒），当并发状态转换超过这个阈值，写操作开始排队，导致整个Agent链的推进被阻塞。我后来调整了事件粒度，把多个内部状态转换合并成一个聚合事件，QPS从900降到了200，问题消失。（延伸阅读：为什么我把公司知识库的RAG Pipeline从LangChain迁到了裸Gemini API：一场关于长上下文与分块策略的架构决策复盘）
这个踩坑让我意识到一件事：事件溯源模型的理论很优雅，但如果你不对事件粒度做设计，存储层的成本会吃掉所有收益。
我把Saga模式完整地搬到Agent层，这是可用的代码和不可回避的代价
经过160小时的反复试验，我得出的结论是：无论底层框架选哪个，在多Agent长事务中，你最终都会需要一个Saga协调器。以下是直接可用的实现，基于LangGraph的检查点机制，但设计思路可以移植到ADK或CrewAI。
Saga模式的轻量级落地方案
from typing import TypedDict, List, Callable
from langgraph.graph import StateGraph, END
from langgraph.checkpoint import PostgresSaver
import asyncio
class SagaState(TypedDict):
saga_id: str
executed_steps: List[str]
compensation_stack: List[dict] # {"step": "agent_b", "compensate_fn": "unfreeze_budget", "params": {...}}
final_status: str # "completed", "compensating", "compensated", "failed"
def build_saga_coordinator(tasks_with_compensation: list) -> StateGraph:
"""
tasks_with_compensation: [
{"name": "freeze_budget", "execute": fn, "compensate": fn, "params": {...}},
{"name": "book_hotel", "execute": fn, "compensate": fn, "params": {...}},
]
"""
workflow = StateGraph(SagaState)
# 为每个任务创建执行节点和补偿节点
for idx, task in enumerate(tasks_with_compensation):
exec_node_name = f"exec_{task['name']}"
comp_node_name = f"comp_{task['name']}"
def make_exec_node(task=task):
def exec_node(state: SagaState) -> SagaState:
try:
result = task['execute'](**task['params'], saga_id=state['saga_id'])
state['executed_steps'].append(task['name'])
state['compensation_stack'].append({
"step": task['name'],
"compensate_fn": task['compensate'].__name__,
"params": result.get('compensation_params', {})
})
state['final_status'] = f"{task['name']}_completed"
except Exception as e:
state['final_status'] = "requires_compensation"
state['last_error'] = str(e)
return state
return exec_node
def make_comp_node(task=task):
def comp_node(state: SagaState) -> SagaState:
# 从补偿栈中取出对应的补偿函数并执行
comp_entry = next(
(c for c in state['compensation_stack'] if c['step'] == task['name']),
None
)
if comp_entry:
task['compensate'](**comp_entry['params'], saga_id=state['saga_id'])
state['final_status'] = f"{task['name']}_compensated"
return state
return comp_node
workflow.add_node(exec_node_name, make_exec_node())
workflow.add_node(comp_node_name, make_comp_node())
# 连接：执行成功->下一个执行；执行失败->补偿链
for idx, task in enumerate(tasks_with_compensation):
current_exec = f"exec_{task['name']}"
current_comp = f"comp_{task['name']}"
if idx 0 else None
# 条件边：成功则继续，失败则触发补偿
workflow.add_conditional_edges(
current_exec,
lambda state, task_name=task['name']: (
"continue" if state['final_status'] == f"{task_name}_completed"
else "compensate"
),
{
"continue": next_exec,
"compensate": current_comp
}
)
# 补偿完成后，继续补偿前一个步骤
if prev_comp:
workflow.add_edge(current_comp, prev_comp)
else:
# 最后一个任务
workflow.add_edge(current_exec, END)
workflow.add_edge(current_comp, END)
return workflow
# 使用示例
async def main():
checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@localhost/saga_db")
workflow = build_saga_coordinator([
{
"name": "freeze_budget",
"execute": freeze_budget_fn,
"compensate": unfreeze_budget_fn,
"params": {"amount": 8500, "department_id": "dept_42"}
},
{
"name": "book_hotel",
"execute": book_hotel_fn,
"compensate": cancel_booking_fn,
"params": {"hotel_id": "hilton_sf", "idempotency_key": "saga_abc123"}
}
])
app = workflow.compile(checkpointer=checkpointer)
initial_state = SagaState(
saga_id="saga_abc123",
executed_steps=[],
compensation_stack=[],
final_status="started"
)
result = await app.ainvoke(initial_state)
return result
这段代码的生产化需要解决三个问题：补偿函数的幂等性（如果补偿被重复调用必须无害）、SagaID的全局唯一性（我用的是Snowflake算法）、补偿栈的序列化（PostgresSaver默认用Pickle，需要替换为JSON以保证跨版本兼容）。但核心思路是明确的——把微服务Saga模式的“正向操作-补偿操作”对搬到了Agent执行层。
不可回避的代价
用Saga模式管理Agent事务，你付出了什么？
- 代码量增加3倍：每个Agent不仅要写正向逻辑，还要写补偿逻辑，还要测试补偿是否正常工作。
- 幂等性要求渗透所有外部调用：如果酒店API不支持幂等健（idempotency key），你就得自己实现去重——这通常意味着一个额外的Redis存储。
- 补偿失败的人工兜底：Saga模式在微服务里允许补偿失败后进入“待人工处理”状态，在Agent系统里你同样需要这个——这意味着你的Agent链不是全自动的，末端必须有一个human-in-the-loop节点。
我在部署到预发布环境的第一周，就遭遇了一次补偿失败：unfreeze_budget函数因为财务系统的token过期而失败，整个Saga进入了CompensationFailed状态。如果没有提前设计人工兜底节点，这条记录就会永远卡在补偿栈里。这让我学到了一课：Saga解决的是自动化失败问题，但Saga自身的失败需要人介入——这不是缺点，是分布式系统的物理定律。
我的判断 + 可能被打脸的风险
基于这160个小时的实验数据，我做出如下判断：
未来6个月内，多Agent框架的事务处理会分裂成两条路线：一条是轻量级路线（CrewAI的共享记忆+背压），服务于内部工具和低风险审批，追求吞吐量和开发效率；另一条是重量级路线（ADK的事件溯源+显式Saga API），服务于金融、合规、供应链等强一致性场景。LangGraph如果不解决检查点恢复的RTO问题，会在这两条路线之间被挤压。LangChain显然意识到了这一点——他们在最新的0.3.0 alpha版本的changelog里提到了“checkpoint compression”和“async saga coordinator”，但我还没拿到内测资格。
可能被打脸的风险：如果OpenAI的GPT-5.5（或他们下一个版本叫什么）内置了足够可靠的工具调用状态管理，让开发者不再需要显式的多Agent框架，这三家的市场份额都会被吃掉一块。另一个风险是：企业客户可能发现，与其在Agent框架上折腾分布式事务，不如退回到把审批逻辑写成一个单体服务+传统工作流引擎——这意味着我上面所有的分析都只适用于一个注定小众的市场。如果传统BPMN引擎（Camunda、Temporal）在未来三个月推出成熟的Agent插件，我预测的两条路线分裂就会被第三条路线截胡。
但即便如此，我依然认为这160个小时没有白费。因为多Agent系统迟早会长大到需要分布式事务——就像2015年没有人觉得微服务需要Saga，到了2018年Saga已经是微服务面试的必考题。Agent基础设施的成熟，永远比应用场景的爆发要慢半拍。你现在折腾的每一个补偿链，都是三个月后别人踩坑时的救命稻草。
