4.5 幂等性与状态恢复

4.5 幂等性与状态恢复

重跑会不会翻倍

本节在全册位置：持久化带来「重放」，重放必须幂等，否则状态错乱。幂等指同一输入多次执行结果一致、副作用不重复。LangGraph 靠检查点使重跑从断点继续，而非从头，天然降低重复副作用。我们强调：检查点降低重复，但不等于自动幂等。

机制：从断点而非从头

无检查点时，重试要从 START 重跑，已调用外部 API 的节点会再调一次（重复发信）。有检查点时，失败后从最近检查点 resume，已完成的节点不重跑。对不可幂等副作用（如支付），应在节点内用「去重键」或放 interrupt 人工确认。类比金融：幂等像交易流水号，重复提交不会被记两次。去重键要落在状态里，才能跨重试可见。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class S(TypedDict):
    sent: list

def sendmail(s: S):
    # 用已发集合去重，保证幂等
    if "u1" in s["sent"]:
        return {}
    return {"sent": ["u1"]}

b = StateGraph(S)
b.add_node("sendmail", sendmail)
b.add_edge(START, "sendmail")
b.add_edge("sendmail", END)
g = b.compile(checkpointer=MemorySaver())
cfg = {"configurable": {"thread_id": "p1"}}
g.invoke({"sent": []}, cfg)
#    4.5 幂等性与状态恢复
​

# 去重键 + 中断 是处理写操作的两道保险
​

案例：重复发信事故

背景：节点因超时被判失败，重试又发了一封相同的信，用户投诉。

操作：节点内用已发集合去重，并检查点保证不重跑已完成步。

结果：用户只收到一封，事故归零。

解读：幂等是持久化的配套要求，不能只靠检查点，业务层去重键必须落地。

变式：把去重键提升为全局「动作日志」，可审计每次外部调用，合规与排障都受益。

工程清单

• 无检查点时重试从 START 重跑，已调用外部 API 会再调一次造成重复副作用。

• 有检查点时从最近检查点 resume，已完成节点不重跑，天然降低重复副作用。

• 不可幂等副作用（支付）用去重键或 interrupt 兜底，检查点降低重复但不等于自动幂等。

常见误区

假设检查点自动保证幂等，结果支付类节点被重放两次；幂等要业务层去重键配合，把去重键提升为全局动作日志还能顺带满足审计。

去重键与动作日志

幂等的核心是「同一动作不重复产生副作用」，最实用的实现是状态里的去重键集合，必要时升级为动作日志。

# 去重键集合：已发就不再发
class S(TypedDict):
    sent: list

def notify(s):
    if "u1" in s["sent"]:
        return {}            # 幂等：已处理，空更新
    return {"sent": s["sent"] + ["u1"]}

# 升级为动作日志：可审计每次外部调用
class S2(TypedDict):
    log: list
def call_api(s, name):
    if any(e["name"] == name for e in s["log"]):
        return {}
    return {"log": s["log"] + [{"name": name, "ok": True}]}
# 重试时日志已含该动作，节点直接跳过
​

⚠️ 常见坑：以为检查点 resume 就自动幂等，结果支付节点被重放两次；resume 只保证不重跑完成的节点，不保证节点内部不发第二次。

💡 关键直觉：幂等是业务层责任，去重键必须落在状态里，才能跨重试可见、跨进程一致。

业务幂等键怎么选

去重键不是随便取个字段就完事，选错键会让去重失去意义。幂等键要满足「同一业务动作始终生成同一键」：

判断标准一句话：同一笔业务只应产生一个键，重试多少次都是同一个键。时间戳、随机数、自增计数都不能当幂等键。把幂等键的生成逻辑放在节点入口，先查后写，天然满足「重复提交不重复执行」。

重试窗口与去重集合的边界

去重集合放状态里是必须的，但要理解它的生命周期，才知道边界在哪：

1. 去重集合随检查点持久化，进程重启不丢，跨重试可见。
2. 去重集合是 thread 维度的，不同 thread 的去重集合相互独立。
3. 同一 thread 内永远有效，除非手动清空或 thread 被删除。

# 跨 thread 也要去重时，把去重集合放进全局 store
# 而非状态，状态是 per-thread 的
​

理解第 2 条很关键：如果同一个业务可能在多个 thread 里重复触发，只靠状态里的集合拦不住跨 thread 的重复。此时需要把去重集合升级到共享存储（全局 store 或业务库），以业务幂等键为唯一约束。判断用哪种：先问「重复发生在同一 thread 内还是跨 thread」，前者状态集合够用，后者必须走共享存储。

动作日志：从去重到审计

把去重集合升级成动作日志后，它同时回答「这个动作做了没」和「这个动作发生过什么」。日志条目要包含四要素：动作名、幂等键、时间戳、结果状态。

class S(TypedDict):
    action_log: Annotated[list, operator.add]

def call_api(s, name, payload):
    # 用幂等键查日志
    if any(e["name"] == name and e["key"] == payload["order_id"]
           for e in s["action_log"]):
        return {}                       # 已做过，直接跳过
    result = _do_call(payload)          # 真正执行
    return {"action_log": [{
        "name": name, "key": payload["order_id"],
        "ts": time.time(), "ok": result.ok}]}
​

动作日志的收益是双份的：重试时跳过去重，排查时翻出每次外部调用的时间与结果。日志只增不删，配合检查点天然可审计。唯一要控制的是日志字段大小——只记必要元数据，别把请求体全文塞进去，否则又回到状态体积失控的老问题。