向后兼容性

1. 技术兼容性：最常见；新代码必须能够加载并针对现有 State 执行。
2. 业务兼容性：较少见；即使代码已更改，现有运行应继续遵循旧的业务逻辑。
3. 非确定性：仅适用于函数式 API。

​

技术兼容性

• 在线程暂停或即将进入某个节点时（例如在 interrupt 处或通过一个仍然路由到旧名称的检查点条件边），重命名或删除该节点。恢复时，LangGraph 无法通过其保存的名称找到该节点，运行将失败。恢复运行的起点是执行停止时所在节点的开头，因此缺少的节点将无法从任何地方恢复。
• 重命名或删除 State 键，而旧的检查点仍然包含该键，或者下游节点仍然读取该键。
• 收紧 State 字段，例如将 Optional 字段设为必填、收窄类型，或添加没有默认值的新必填字段。现有检查点将无法满足新模式。

​

推荐模式

• 将新 state 字段添加为 NotRequired（或 Optional[...] = None），以便旧检查点仍然有效：
  from typing import NotRequired
  from typing_extensions import TypedDict
  
  class State(TypedDict):
      messages: list
      summary: NotRequired[str]
  
• 将删除操作视为弃用。将字段保留在 state 定义中至少一个 drain 周期，即使没有节点读取它，以确保现有检查点继续加载。
• 通过先添加后删除进行重命名。将新字段或节点与旧的并列添加，在一个弃用窗口期内双写或同时路由到两者，一旦确认没有进行中的线程依赖旧字段或节点，再将其删除。
• 使节点函数能够容忍未知键。TypedDict 在运行时会忽略额外的键，因此旧代码版本的残留状态不会引发异常，除非节点显式读取了缺失的键。
• 使用时间旅行和 graph.get_state 在发布前于暂存环境中针对新代码抽查现有线程。

​

检测进行中的线程

• graph.get_state(config) 返回最新的检查点，包括线程暂停在哪个节点以及任何待处理的中断。
• graph.get_state_history(config) 返回该线程完整的按时间顺序排列的检查点列表。

​

业务兼容性

• 已经通过 triage 的线程应直接继续到 respond（旧流程）。
• 新线程应运行完整的新流程。

from typing import NotRequired
from typing_extensions import TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    request: str
    flow_version: NotRequired[int]
    response: NotRequired[str]


def intake(state: State) -> dict:
    # Stamp new threads with the current flow version. Existing threads
    # that resume past `intake` keep whatever value was already saved.
    return {"flow_version": state.get("flow_version", 2)}


def triage(state: State) -> dict: ...
def policy_check(state: State) -> dict: ...
def respond(state: State) -> dict: ...


def after_triage(state: State) -> str:
    if state.get("flow_version", 1) >= 2:
        return "policy_check"
    return "respond"


builder = StateGraph(State)
builder.add_node("intake", intake)
builder.add_node("triage", triage)
builder.add_node("policy_check", policy_check)
builder.add_node("respond", respond)
builder.add_edge(START, "intake")
builder.add_edge("intake", "triage")
builder.add_conditional_edges("triage", after_triage, ["policy_check", "respond"])
builder.add_edge("policy_check", "respond")
builder.add_edge("respond", END)

graph = builder.compile()

​

非确定性

• 在恢复点之前添加、删除或重新排序 @task 调用或 interrupt 调用。LangGraph 根据重放中的位置将缓存结果和恢复值与调用进行匹配，因此移动该位置可能会导致错误的缓存值被重放到另一个调用上。
• 在 @task 之外引入非确定性操作，例如 time.time()、random.random() 或在 entrypoint 体中内联的网络调用。在重放时，这些操作会产生与首次运行时不同的值，这可能会改变控制流。

• 在部署更改前让进行中的运行 drain 完毕。
• 将任何新逻辑包装在一个新的 @task 中，以便其结果被独立检查点。
• 在 langgraph.json 中为新行为注册一个具有新图名称的 entrypoint，并将新线程路由到它。