以编程方式使用文档

生产环境中的软件需要变更。新的需求、错误修复和重构最终都会落到您的图代码中。因为 LangGraph 使用最新部署的图针对已 持久化的 对于现有线程的 state,您部署的每个变更实际上都是相对于现有检查点的向后兼容 API 变更。

与将运行固定到其启动时代码版本的工作流引擎不同,LangGraph 立即将最新图应用到 *每个* 线程,包括新线程和从检查点恢复的线程。这很方便:错误修复会传播到进行中的对话和代理,无需仪式性的操作。这也意味着您必须思考每个变更如何与在代码前一版本下启动的运行进行交互。

需要关注三类兼容性问题,大致按您遇到它们的顺序排列:

  1. 技术兼容性:最常见的情况;新代码必须仍能针对现有 State 进行加载和执行。
  2. 业务兼容性:不太常见;现有运行应该继续遵循旧的业务逻辑,即使代码已更改。
  3. Non-determinism:仅适用于 函数式 API.

技术兼容性

技术兼容性相当于微服务中的 API 破坏性变更。这里的 "API" 是指你的图代码与已由 检查点存储器 为现有线程保存的状态。当线程恢复时,LangGraph 反序列化保存的状态,按名称将其分派到节点,并期望节点返回符合状态模式的值。

常见的技术性破坏:

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

边拓扑本身是 *不* 保存在检查点中的。在仍存在的节点之间添加、删除或重新路由边对于正在运行的线程是安全的。根据 图迁移 总结,唯一可能破坏中断线程的拓扑变更是重命名或删除节点。

推荐模式

  • - 将新状态字段添加为 NotRequired (or Optional[...] = None),旧检查点仍可验证:
  from typing import NotRequired
  from typing_extensions import TypedDict

  class State(TypedDict):
      messages: list
      summary: NotRequired[str]  # [!code ++]
  
  • - 将删除视为弃用。即使没有节点读取它,也至少在状态上保留一个排空周期的字段定义,以便现有检查点继续加载。
  • - 通过以下方式重命名 *add-then-remove*。在新字段或节点旁添加旧字段或节点,在弃用窗口期间双向写入或路由到两者,然后确认没有运行中线程依赖后删除旧的。
  • - 保持节点函数对未知键的容错。 TypedDict 在运行时忽略额外的键,因此旧代码版本的剩余状态不会引发错误,除非节点显式读取缺失的键。
  • - 使用 时间穿越graph.get_state 在暂存部署中对新代码进行抽查现有线程后再推出。

检测运行中线程

在移除节点、重命名状态键或进行旧线程无法容忍的更改之前,您需要知道当前是否有线程停在即将删除的代码版本上。LangGraph 本身不维护线程状态的搜索索引,因此答案取决于您的图运行的位置。

**如果部署到 LangSmith.** 使用 Agent Server 的线程搜索按状态筛选。该 status 字段接受 idle, busy, interruptederror,因此您可以批量查询 interrupted or busy 线程,可选择使用元数据筛选器缩小范围。参见 按线程状态筛选列出线程.

任何 LangGraph 运行的地方。 使用 LangSmith 追踪 监控生产中正在进入和退出的节点。这是节点或状态字段在当前活动代码路径中不再可达的最可靠信号。

**当您已有 thread_id.** 直接检查该单个线程:

如有疑问,请保留已弃用的节点或字段,直到 Agent Server 线程列表和追踪均显示不再有活动。

业务兼容性

有时变更在技术上是有效的(每个现有检查点仍可加载,每个节点仍可解析),但 *含义* 新图与旧图的不同。新行为对新线程是正确的,而您不希望将其追溯应用到在旧逻辑下启动的线程。

例如,假设您的图运行 intake → triage → respond,而您决定在 policy_checktriage 之间插入一个新的 respond:

  • - 已经通过的线程 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()

恢复后的旧线程 triage 读取 flow_version 其保存的状态(或采用 v1 默认值)并跳过 policy_check。新线程从 intake开始,带有 flow_version=2标记,并运行新路径。一旦所有 v1 线程完成,您可以移除版本标志和条件边。

此模式仅在您在线程启动时设置版本时才有效 *在*时才能正常工作(在任何需要版本化的分支之前)。稍后设置意味着现有线程在需要时不会拥有它。

Non-determinism

此类别仅适用于 函数式 API 以及 **任务** or interrupt 调用在 图 API **节点**中。普通图 API **节点** 从节点函数的开头重新运行 ;设计副作用时要保持幂等性,但除非您在那个 **任务** or interrupt 使用 **节点**.

一个函数式 API **入口点** 编译为单个 **节点** ,当运行恢复时从头重放入口点主体,使用缓存的 @task 结果来跳过已完成的工作。有两类变更会破坏此模式:

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

更多带示例的详细说明,请参阅 确定性常见陷阱 (位于 Functional API 指南中)。

如果您需要对一个 进行重大代码更改 @entrypoint ,其中存在运行中的任务,最安全的选项是:

  • - 在部署更改之前,让运行中的任务完成。
  • - 将任何新逻辑包装在一个新的 @task 中,以便其结果被独立检查点保存。
  • - 在 langgraph.json 中以新的图名称注册一个新的入口函数用于新行为,并将新线程路由到该入口函数。