LangGraph持久化：支持恢复运行的检查点与HITL机制

本指南旨在为“LangGraph持久化：让其他一切成为可能的功能”重建一条可操作的实现路径。重点关注契约、校验机制以及那些无需猜测意图即可直接放入代码库的代码。在修改代码之前，应先明确输入参数、该步骤的负责人以及终止标准。操作人员应当能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。相比庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应当能够指向单一的责任主体。

目前我们所了解内容的简要回顾

为快速回顾目前所了解的内容，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 在成本较高的步骤之后设置检查点，以避免重复调用同一模型而产生额外费用。

持久化的真正含义

要了解什么是真正的持久性，应在修改代码之前明确输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间和成本。在共享环境中，提前了解情况可避免出现意外的费用账单。在高成本步骤之后设置检查点，这样重试时就不会再次对同一模型调用收费。

实际运作方式：检查点

关于其实际运作方式：在修改代码之前，需明确检查点、输入参数、该步骤的负责人以及终止标准。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。配置信息应置于应用程序代码之外，环境文件、密钥存储以及功能标志都应存放于操作员可审计的统一位置。在成本较高的步骤之后设置检查点，以避免重复执行时再次调用同一模型。关于其实际运作方式：在修改代码之前，需明确检查点、输入参数、该步骤的负责人以及终止标准。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相较于庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向单一责任主体。

多线程：隔离多个执行任务

在多线程处理方面：为确保多次执行相互独立，应在修改代码之前明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功标准，杜绝无声的半完成状态。 持久性使得演示文稿能够转化为真正的系统：多线程、人工干预以及时间回溯调试都依赖于它。

代码（其实比你想象的要简单）

对于代码部分（其实比你想象的要简单），在修改代码之前先明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 在功能结果旁记录执行时间和成本。提前了解这些信息可以避免在共享环境中出现意外账单。 持久性使得演示文稿能够转化为真正的系统：线程、人工干预以及时间回溯调试都依赖于它。

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import InMemorySaver
from typing import TypedDict

llm = ChatOpenAI()

class JokeState(TypedDict):
    topic: str
    joke: str
    explanation: str

def generate_joke(state: JokeState):
    response = llm.invoke(f"Generate a joke on the topic: {state['topic']}")
    return {"joke": response.content}

def generate_explanation(state: JokeState):
    response = llm.invoke(f"Explain this joke: {state['joke']}")
    return {"explanation": response.content}

# Build graph
graph = StateGraph(JokeState)
graph.add_node("generate_joke", generate_joke)
graph.add_node("generate_explanation", generate_explanation)
graph.add_edge("__start__", "generate_joke")
graph.add_edge("generate_joke", "generate_explanation")
graph.add_edge("generate_explanation", "__end__")

# Add persistence - this is the key line
checkpointer = InMemorySaver()
workflow = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}

result = workflow.invoke(
    {"topic": "pizza"},
    config=config
)

# Get the final state
workflow.get_state(config)

# Get ALL intermediate states
workflow.get_state_history(config)

config_2 = {"configurable": {"thread_id": "2"}}
result_2 = workflow.invoke({"topic": "pasta"}, config=config_2)

利用此工具可以构建什么

在修改代码之前，需先明确使用该工具能够构建什么功能，包括输入参数、各步骤的负责人以及终止标准。操作人员应能从已知的检查点重新运行相应步骤，而无需猜测隐藏状态。配置信息应置于应用程序代码之外，环境文件、密钥存储及功能开关都应集中在操作人员可审计的统一位置。持久化机制能让演示版本转变为真正的系统——线程处理、人工干预以及时间回溯式调试都依赖于它。在修改代码之前，需先明确使用该工具能够构建什么功能，包括输入参数、各步骤的负责人以及终止标准。操作人员应能从已知的检查点重新运行相应步骤，而无需猜测隐藏状态。相比庞大的脚本，更应优先选择小型且易于测试的单元。当某个步骤失败时，故障原因应能明确指向单一责任主体。

1. 聊天机器人中的短期记忆

1. 对于聊天机器人的短期记忆功能，在修改代码之前需明确输入内容、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 保持图结构的状态简洁且具有类型约束。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在中断后导致无法继续执行。

2. 容错能力

2. 容错性方面，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间和成本。在共享环境中，提前了解情况可避免出现意外费用。要保持图结构的层次简单且类型明确，嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在中断后导致无法继续执行。

import time

def step_two(state):
    print("Step 2 running...")
    time.sleep(30)  # simulating a slow operation
    return {"step_two": "done"}

workflow.invoke(None, config=config)

3. 人工干预机制

对于3.人工干预模式，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能开关都应存放在操作人员可审计的统一位置。 保持图结构的状态简洁且具有类型约束。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会导致中断后无法继续执行。 对于3.人工干预模式，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏状态。 相较于庞大的脚本，应优先使用小型且易于测试的单元。当某个步骤失败时，故障原因应能明确指向单一的责任主体。

4.时间回溯

对于时间旅行功能，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将此阶段视为输入与验证后输出之间的契约。为相关成果命名，定义成功判定标准，杜绝默许的半完成状态。 在成本较高的步骤之后设置检查点，以避免重复执行时再次产生相同的模型调用费用。

# Go back to a specific checkpoint and change the topic
workflow.update_state(
    {"configurable": {"thread_id": "1", "checkpoint_id": "<checkpoint_id>"}},
    {"topic": "samosa"}
)

# Re-run from that point
workflow.invoke(None, config={"configurable": {"thread_id": "1", "checkpoint_id": "<new_checkpoint_id>"}})

整体视角

从整体视角来看，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需在功能结果旁记录执行时间和成本。提前透明化信息可避免在共享环境中出现意外收费。 在成本较高的步骤之后设置检查点，以避免重复执行时再次产生相同的模型调用费用。

超越博客的结构化学习

在实施“超越博客的结构化学习”时，应在修改代码之前明确输入数据、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能开关都应存放于操作人员可审计的统一位置。 在成本较高的步骤之后设置检查点，以避免重复执行时再次调用同一模型。

课程

对于课程类功能，修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。

运营检查清单

针对运营检查清单，修改代码之前同样需要明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。

需同时记录正常流程与异常恢复流程。重试机制及错误处理都是产品功能的一部分。

持久化功能可将演示版本转化为完整系统：线程管理、人工干预以及时间回溯调试等功能都依赖于它。

在预算允许的情况下，为关键路径添加带有固定装置的烟雾测试到持续集成流程中。

应优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向单一的责任模块。

持久化功能将演示版本转化为完整系统：线程、人工干预以及时间回溯调试都依赖于它。

在推广整套技术栈之前，需冻结版本、为关键路径记录标准操作流程，并确认回滚步骤。共享环境需要设置速率限制、进行租户检查，同时明确密钥轮换的负责人。