Skip to main content
持久化执行是一种技术，其中进程或工作流在关键点保存其进度，允许其暂停并在稍后从暂停处精确恢复。这在需要人机交互的场景中特别有用，用户可以在继续之前检查、验证或修改进程，以及在可能遇到中断或错误（例如，调用LLM超时）的长时间运行任务中。通过保存已完成的工作，持久化执行使进程能够在不重新处理先前步骤的情况下恢复——即使在显著延迟（例如一周后）之后也是如此。 LangGraph内置的持久化层为工作流提供持久化执行，确保每个执行步骤的状态都保存到持久化存储中。此功能保证，如果工作流被中断——无论是由于系统故障还是人机交互交互——都可以从其最后记录的状态恢复。
如果您正在使用带有检查点器的LangGraph，则已启用持久化执行。您可以随时暂停和恢复工作流，即使在中断或故障之后。 为了充分利用持久化执行，请确保您的工作流设计为确定性和幂等性，并将任何副作用或非确定性操作包装在任务中。您可以从StateGraph (Graph API)和Functional API中使用任务。

​
要求

要在LangGraph中利用持久化执行，您需要：
  1. 通过指定一个检查点器来启用工作流中的持久化，该检查点器将保存工作流进度。
  2. 在执行工作流时指定一个线程标识符。这将跟踪特定工作流实例的执行历史。
  3. 将任何非确定性操作（例如，随机数生成）或具有副作用的操作（例如，文件写入、API调用）包装在任务中，以确保当工作流恢复时，这些操作不会针对特定运行重复执行，而是从持久化层检索其结果。有关更多信息，请参阅确定性和一致重放。

​
确定性和一致重放

当您恢复工作流运行时，代码不会从执行停止的同一行代码恢复；相反，它将识别一个适当的起点以从该点继续执行。这意味着工作流将从起点重放所有步骤，直到达到停止点。 因此，在为持久化执行编写工作流时，您必须将任何非确定性操作（例如，随机数生成）和任何具有副作用的操作（例如，文件写入、API调用）包装在任务或节点中。 为确保您的工作流是确定性的并且可以一致重放，请遵循以下准则：
  • 避免重复工作：如果一个节点包含多个具有副作用的操作（例如，日志记录、文件写入或网络调用），请将每个操作包装在一个单独的任务中。这确保当工作流恢复时，操作不会重复，并且其结果从持久化层检索。
  • 封装非确定性操作：将任何可能产生非确定性结果的代码（例如，随机数生成）包装在任务或节点中。这确保在恢复时，工作流遵循精确记录的步骤序列并具有相同的结果。
  • 使用幂等操作：尽可能确保副作用（例如，API调用、文件写入）是幂等的。这意味着如果操作在工作流失败后重试，其效果将与第一次执行时相同。这对于导致数据写入的操作尤为重要。如果任务开始但未能成功完成，工作流的恢复将重新运行任务，依赖记录的结果来保持一致性。使用幂等键或验证现有结果以避免意外重复，确保工作流执行顺畅且可预测。
有关要避免的陷阱示例，请参阅功能API中的常见陷阱部分，该部分展示了如何使用任务构建代码以避免这些问题。相同的原则适用于StateGraph (Graph API)。

​
持久性模式

LangGraph支持三种持久性模式，允许您根据应用程序的要求平衡性能和数据一致性。更高的持久性模式会为工作流执行增加更多开销。您可以在调用任何图执行方法时指定持久性模式： 
graph.stream(
    {"input": "test"},
    durability="sync"
)
持久性模式从最不持久到最持久如下：
  • "exit"：LangGraph仅在图执行成功退出、出错或由于人机交互中断时持久化更改。这为长时间运行的图提供了最佳性能，但意味着中间状态未保存，因此您无法从执行过程中发生的系统故障（如进程崩溃）中恢复。
  • "async"：LangGraph在下一步执行时异步持久化更改。这提供了良好的性能和持久性，但存在一个小风险，即如果进程在执行期间崩溃，LangGraph可能不会写入检查点。
  • "sync"：LangGraph在下一步开始之前同步持久化更改。这确保LangGraph在继续执行之前写入每个检查点，以高持久性为代价提供一些性能开销。

​
在节点中使用任务

如果一个节点包含多个操作，您可能会发现将每个操作转换为任务比重构操作为单独的节点更容易。 
from typing import NotRequired
from typing_extensions import TypedDict
from langchain_core.utils.uuid import uuid7

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
import requests

# 定义一个 TypedDict 来表示状态
class State(TypedDict):
    url: str
    result: NotRequired[str]

def call_api(state: State):
    """示例节点，发出 API 请求。"""
    result = requests.get(state['url']).text[:100]  # 副作用  #
    return {
        "result": result
    }

# 创建 StateGraph 构建器并为 call_api 函数添加一个节点
builder = StateGraph(State)
builder.add_node("call_api", call_api)

# 将开始和结束节点连接到 call_api 节点
builder.add_edge(START, "call_api")
builder.add_edge("call_api", END)

# 指定一个检查点器
checkpointer = InMemorySaver()

# 使用检查点器编译图
graph = builder.compile(checkpointer=checkpointer)

# 定义一个带有线程 ID 的配置。
thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}

# 调用图
graph.invoke({"url": "https://www.example.com"}, config)

​
恢复工作流

一旦在工作流中启用了持久化执行，您就可以在以下场景中恢复执行：
  • 暂停和恢复工作流：使用interrupt函数在特定点暂停工作流，并使用Command原语以更新状态恢复它。有关更多详细信息，请参阅中断。
  • 从故障中恢复：在异常（例如，LLM提供商中断）后，自动从最后一个成功检查点恢复工作流。这涉及使用相同的线程标识符执行工作流，并为其提供None作为输入值（请参阅此示例使用功能API）。

​
恢复工作流的起点

  • 如果您正在使用StateGraph (Graph API)，起点是执行停止的节点的开始。
  • 如果您在节点内进行子图调用，起点将是调用已停止子图的父节点。 在子图内部，起点将是执行停止的特定节点。
  • 如果您正在使用功能API，起点是执行停止的入口点的开始。
在 GitHub 上编辑此页面 或 提交问题。
通过 MCP 将这些文档 连接到 Claude、VSCode 等，以获取实时答案。