外观
软件在生产环境中也需要不断变更。新需求、bug 修复和重构最终都会落到你的图代码中。由于 LangGraph 会对现有线程已持久化的状态运行最新部署的图,因此你发布的每一项更改,对于现有检查点而言实际上都是一种向后兼容的 API 变更。
与将运行锁定在其启动时代码版本的工作流引擎不同,LangGraph 会立即将最新图应用于每一个线程,包括新线程和从检查点恢复的线程。这很方便:bug 修复可以毫无障碍地传播到正在进行的对话和智能体。这也意味着你必须仔细考虑每项更改如何与在旧版本代码下启动的运行交互。
需要注意三类兼容性问题,按你大致会遇到它们的顺序排列:
TIP
有关运行时默认支持的图拓扑和状态更改的简要总结,请参阅图的迁移。本页其余部分介绍了当更改超出该受支持集合时你可以应用的模式。
技术兼容性
技术兼容性相当于微服务中的 API 破坏性变更。这里的"API"是你的图代码与检查点持久化器为现有线程已持久化的数据之间的契约。当线程恢复时,LangGraph 会反序列化已保存的状态,按名称将其分派到某个节点,并期望该节点返回符合状态模式的值。
常见的破坏性技术问题:
- 重命名或删除节点:当线程在某个节点暂停或即将进入该节点时,例如在
interrupt处,或通过一条仍路由到旧名称的、已检查点化的条件边。恢复时,LangGraph 无法按保存的名称找到该节点,运行将失败。恢复运行的起点是执行停止处的节点开头,因此缺失的节点无处可恢复。 - 重命名或删除状态(State)键:旧检查点仍包含的、或下游节点仍会读取的状态键。
- 收紧状态(State)字段:例如将
Optional字段设为必填、收窄类型,或添加没有默认值的新必填字段。现有检查点将无法满足新模式。
边拓扑本身不持久化在检查点中。在仍然存在的节点之间添加、删除或重新路由边,对于正在进行的线程是安全的。根据图的迁移总结,唯一可能破坏被中断线程的拓扑更改是重命名或删除节点。
推荐模式
- 将新状态字段添加为
NotRequired(或Optional[...] = None),以便旧检查点仍能通过校验:
python
from typing import NotRequired
from typing_extensions import TypedDict
class State(TypedDict):
messages: list
summary: NotRequired[str] 将删除视为弃用。即使没有节点读取该字段,也至少在状态上保留该字段定义一个排空周期(drain cycle),以便现有检查点继续加载。
通过先添加后删除的方式进行重命名。在旧字段或旧节点旁边添加新字段或新节点,在弃用窗口期内进行双写或同时路由到两者,待确认没有任何正在进行的线程依赖旧项后,再将其删除。
让节点函数对未知键保持宽容。
TypedDict在运行时忽略多余的键,因此旧代码版本残留的状态不会引发错误,除非节点显式读取缺失的键。在正式上线前,先在预发布(staging)部署中使用时间旅行和
graph.get_state对照新代码抽查现有线程。将新状态字段标记为可选(
z.string().optional()或.nullish()),以便旧检查点仍能通过校验。将删除视为弃用:在模式上保留该字段至少一个排空周期(drain cycle),以便现有检查点继续加载。
通过先添加后删除的方式进行重命名:在旧字段或旧节点旁边添加新字段或新节点,在弃用窗口期内进行双写或同时路由到两者,待没有正在进行的线程依赖旧项后,再将其删除。
在正式上线前,先在预发布(staging)部署中使用时间旅行和
graph.getState对照新代码抽查现有线程。
检测正在进行的线程
在删除节点、重命名状态(State)键,或以其他方式做出旧线程无法容忍的更改之前,你需要知道是否有任何线程当前停留在即将删除的代码版本上。LangGraph 本身不维护线程状态的搜索索引,因此答案取决于你的图在哪里运行。
如果你部署到 LangSmith。 使用 Agent Server 的线程搜索按状态进行过滤。status 字段接受 idle、busy、interrupted 和 error,因此你可以批量查询 interrupted 或 busy 线程,并可选地使用元数据过滤器缩小范围。请参阅按线程状态过滤和列出线程。
只要 LangGraph 能运行的地方。 使用 LangSmith 追踪监控生产环境中哪些节点被进入和退出。这是判断某个节点或状态字段在任何活动代码路径中是否不再可达的最可靠信号。
当你已经有一个 thread_id 时。 直接检查该单个线程:
graph.get_state(config)返回最新的检查点,包括线程暂停在哪个节点以及任何待处理的中断。graph.get_state_history(config)返回该线程完整的按时间顺序排列的检查点列表。graph.getState(config)返回最新的检查点,包括线程暂停在哪个节点以及任何待处理的中断。graph.getStateHistory(config)返回该线程完整的按时间顺序排列的检查点列表。
如有疑问,请保留已弃用的节点或字段,直到 Agent Server 线程列表和追踪都显示其上不再有任何活动。
业务兼容性
有时某项更改在技术上是有效的(每个现有检查点仍能加载,每个节点仍能解析),但新图的含义与旧图不同。新行为对于新线程是正确的,而你不想将其追溯应用于在旧逻辑下启动的线程。
例如,假设你的图执行 intake → triage → respond,你决定在 triage 和 respond 之间插入一个新的 policy_check 步骤:
- 已经通过
triage的线程应直接继续到respond(旧流程)。 - 新线程应运行完整的新流程。
推荐的模式是在线程启动时在状态上记录相关的行为版本,然后使用条件边根据它进行分支:
python
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:
# 为新线程标记当前的流程版本。
# 越过 `intake` 恢复的现有线程会保留已保存的任何值。
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 线程都已完成,你就可以移除版本标志和条件边。
这种模式只有在线程启动时、在需要版本化的任何分支之前设置版本才有效。之后再设置意味着现有线程在需要时不会拥有该版本。
非确定性
此类问题仅适用于功能 API以及 Graph API 节点中的任务或 interrupt 调用。普通的 Graph API 节点在恢复时会从节点函数开头重新运行;请将副作用设计为幂等的,但除非你在该节点中使用任务或 interrupt,否则无需保持任务调用顺序。
功能 API entrypoint 会编译为单个节点,当运行恢复时,它会从开头重放 entrypoint 主体,并使用缓存的 @task 结果跳过已完成的工作。有两类更改会破坏此模型:
- 添加、删除或重新排序位于恢复点之前的
@task调用或interrupt调用。LangGraph 会根据调用在重放中的位置将缓存结果和恢复值与调用匹配,因此移动该位置可能导致错误的缓存值被重放到不同的调用上。 - 在
@task之外引入非确定性操作,例如time.time()、random.random(),或内联在 entrypoint 主体中的网络调用。重放时这些操作会产生与首次运行时不同的值,这可能会改变控制流。
如需更深入的讲解和示例,请参阅功能 API 指南中的确定性和常见陷阱。
如果你需要对具有正在进行的运行的 @entrypoint 进行重大代码更改,最安全的选项是:
- 在部署更改之前,让正在进行的运行全部完成排空。
- 将任何新逻辑包装在新的
@task中,以便其结果被独立检查点化。 - 在
langgraph.json中以新的图名称注册新的 entrypoint 以实现新行为,并将新线程路由到它。