跳到内容

持久化执行

持久化执行 是一种技术,其中进程或工作流在其关键点保存进度,使其能够暂停并在稍后从中断处精确恢复。这在需要 人机环路 的场景中尤其有用,用户可以在继续之前检查、验证或修改流程,以及在可能遇到中断或错误(例如,调用 LLM 超时)的长时间运行的任务中。通过保留已完成的工作,持久化执行使流程能够恢复,而无需重新处理之前的步骤——即使在显著延迟之后(例如,一周后)。

LangGraph 内置的 持久性 层为工作流提供持久化执行,确保每个执行步骤的状态都保存到持久化存储中。此功能保证,如果工作流被中断——无论是由于系统故障还是为了 人机环路 交互——它都可以从其最后记录的状态恢复。

提示

如果您正在使用带有检查点程序的 LangGraph,则您已经启用了持久化执行。您可以随时暂停和恢复工作流,即使在中断或故障之后也是如此。为了充分利用持久化执行,请确保您的工作流被设计为 确定性的幂等的,并将任何副作用或非确定性操作包装在 任务 中。您可以从 StateGraph(Graph API)函数式 API 中使用 任务

要求

要在 LangGraph 中利用持久化执行,您需要

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

确定性和一致性重放

当您恢复工作流运行时,代码不会从执行停止的同一行代码恢复;相反,它将识别一个合适的 起始点,从中继续执行。这意味着工作流将从 起始点 重放所有步骤,直到达到停止的点。

因此,当您编写用于持久化执行的工作流时,您必须将任何非确定性操作(例如,随机数生成)和任何具有副作用的操作(例如,文件写入、API 调用)包装在 任务节点 中。

为了确保您的工作流是确定性的并且可以一致地重放,请遵循以下准则

  • 避免重复工作:如果一个 节点 包含多个具有副作用的操作(例如,日志记录、文件写入或网络调用),请将每个操作包装在一个单独的 任务 中。这确保了当工作流恢复时,操作不会重复执行,并且其结果将从持久性层检索。
  • 封装非确定性操作: 将任何可能产生非确定性结果的代码(例如,随机数生成)包装在 任务节点 中。这确保了在恢复时,工作流遵循完全记录的步骤序列,并具有相同的结果。
  • 使用幂等操作:尽可能确保副作用(例如,API 调用、文件写入)是幂等的。这意味着如果在工作流中发生故障后重试操作,它将具有与第一次执行时相同的效果。这对于导致数据写入的操作尤其重要。如果 任务 开始但未能成功完成,则工作流的恢复将重新运行 任务,依靠记录的结果来保持一致性。使用幂等键或验证现有结果以避免意外重复,从而确保平稳且可预测的工作流执行。

有关要避免的一些陷阱示例,请参阅函数式 API 中的 常见陷阱 部分,其中展示了如何使用 任务 组织代码以避免这些问题。相同的原则适用于 StateGraph(Graph API)

在节点中使用任务

如果一个 节点 包含多个操作,您可能会发现将每个操作转换为 任务 比将操作重构为单独的节点更容易。

from typing import NotRequired
from typing_extensions import TypedDict
import uuid

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

# Define a TypedDict to represent the state
class State(TypedDict):
    url: str
    result: NotRequired[str]

def call_api(state: State):
    """Example node that makes an API request."""
    result = requests.get(state['url']).text[:100]  # Side-effect
    return {
        "result": result
    }

# Create a StateGraph builder and add a node for the call_api function
builder = StateGraph(State)
builder.add_node("call_api", call_api)

# Connect the start and end nodes to the call_api node
builder.add_edge(START, "call_api")
builder.add_edge("call_api", END)

# Specify a checkpointer
checkpointer = MemorySaver()

# Compile the graph with the checkpointer
graph = builder.compile(checkpointer=checkpointer)

# Define a config with a thread ID.
thread_id = uuid.uuid4()
config = {"configurable": {"thread_id": thread_id}}

# Invoke the graph
graph.invoke({"url": "https://www.example.com"}, config)
from typing import NotRequired
from typing_extensions import TypedDict
import uuid

from langgraph.checkpoint.memory import MemorySaver
from langgraph.func import task
from langgraph.graph import StateGraph, START, END
import requests

# Define a TypedDict to represent the state
class State(TypedDict):
    urls: list[str]
    result: NotRequired[list[str]]


@task
def _make_request(url: str):
    """Make a request."""
    return requests.get(url).text[:100]

def call_api(state: State):
    """Example node that makes an API request."""
    requests = [_make_request(url) for url in state['urls']]
    results = [request.result() for request in requests]
    return {
        "results": results
    }

# Create a StateGraph builder and add a node for the call_api function
builder = StateGraph(State)
builder.add_node("call_api", call_api)

# Connect the start and end nodes to the call_api node
builder.add_edge(START, "call_api")
builder.add_edge("call_api", END)

# Specify a checkpointer
checkpointer = MemorySaver()

# Compile the graph with the checkpointer
graph = builder.compile(checkpointer=checkpointer)

# Define a config with a thread ID.
thread_id = uuid.uuid4()
config = {"configurable": {"thread_id": thread_id}}

# Invoke the graph
graph.invoke({"urls": ["https://www.example.com"]}, config)

恢复工作流

一旦您在工作流中启用了持久化执行,您可以在以下场景中恢复执行

  • 暂停和恢复工作流: 使用 interrupt 函数在特定点暂停工作流,并使用 Command 原语使用更新后的状态恢复它。有关更多详细信息,请参阅 人机环路
  • 从故障中恢复: 在异常之后(例如,LLM 提供商中断),从上次成功的检查点自动恢复工作流。这涉及使用相同的线程标识符执行工作流,方法是为其提供 None 作为输入值(请参阅函数式 API 的此 示例)。

恢复工作流的起始点

  • 如果您正在使用 StateGraph(Graph API),则起始点是执行停止的 节点 的开头。
  • 如果您在节点内进行子图调用,则起始点将是调用被暂停的子图的节点。在子图内部,起始点将是执行停止的特定 节点
  • 如果您正在使用函数式 API,则起始点是执行停止的 入口点 的开头。

评论