解析 ‘Human-in-the-loop’：如何在 LangGraph 中设置检查点（Checkpoints）等待人工审批后再继续执行？

引言：AI时代的“人机协作”与LangGraph的核心价值

在人工智能日益渗透我们工作与生活的今天，大型语言模型（LLMs）以其强大的生成和理解能力，正在重塑诸多行业。然而，LLMs并非万能，它们可能产生幻觉、输出不准确信息、甚至生成带有偏见或不当内容。在许多关键业务场景，如金融审批、医疗诊断辅助、法律文书审查、内容发布审核等，完全自动化决策的风险是不可接受的。这时，“人机协作”（Human-in-the-loop, HITL）范式应运而生，它旨在将人类的判断力、常识和伦理洞察力引入AI工作流，形成一个智能与人工优势互补的闭环系统。

LangChain作为构建LLM应用的事实标准，提供了丰富的工具链。而LangGraph，作为LangChain生态系统中的一个强大扩展，专注于通过图结构来编排复杂、有状态的多代理（multi-agent）工作流。它的核心优势在于能够清晰地定义流程中的各个步骤（节点）、数据流向（边）以及状态的演变。更重要的是，LangGraph提供了精妙的“检查点”（Checkpoints）机制，这正是实现高度灵活、可中断、可恢复的人机协作工作流的关键。

本讲座将深入探讨如何在LangGraph中利用检查点机制，构建一个能够暂停执行、等待人工审批、并根据人工反馈继续执行的复杂系统。我们将从LangGraph的基础概念讲起，逐步深入到检查点的内部机制，并通过一个详尽的实战案例，演示如何从零开始搭建一个具备人工审批功能的内容生成与审核工作流。

LangGraph核心概念速览

在深入检查点之前，我们首先快速回顾LangGraph的核心概念，这些是理解后续内容的基础。

LangGraph将一个复杂的应用程序建模为一个有向图。

图结构与节点（Nodes）

• 节点（Nodes）：图中的基本处理单元。每个节点通常封装了一个特定的操作，例如调用一个LLM、执行一个工具、或执行一段自定义的Python代码。在LangGraph中，节点可以是简单的函数，也可以是更复杂的LangChain Runnable。

边与条件路由（Edges & Conditional Routing）

• 边（Edges）：连接节点，定义了数据流动的路径。
  • 普通边（Normal Edges）：一个节点执行完毕后，无条件地将控制权和状态传递给下一个指定节点。
  • 条件边（Conditional Edges）：在源节点执行完毕后，根据其输出内容或当前状态，动态地决定接下来应该跳转到哪个节点。这通过一个“选择器”函数实现，该函数接收当前状态作为输入，返回一个或多个目标节点名。这是实现复杂决策逻辑和分支流程的关键。

状态管理（State Management）

• 状态（State）：LangGraph是“有状态”的。整个图的执行过程围绕一个共享的“图状态”（Graph State）进行。每个节点都可以读取和更新这个状态。状态通常是一个Python字典或一个自定义的Pydantic模型。每次节点执行后，其对状态的修改会合并到全局状态中。
  • StateGraph: 这是定义图和状态的基类。你需要定义一个继承自TypedDict的类来作为你的图状态。

运行器（Runners）

• 图的运行器（Graph Runner）：一旦图被定义并编译，就可以通过调用其invoke或stream方法来执行。这些方法接收一个初始输入，并返回最终的状态或状态流。

这些概念共同构成了LangGraph强大而灵活的编排能力。

为什么需要检查点？人机交互的挑战

在传统的编程模型中，一个程序的执行是线性的、连续的。一旦启动，它会尽可能地运行直到完成或遇到错误。然而，人机协作工作流的本质恰恰是“非线性”和“中断式”的。

考虑以下场景：一个AI系统生成了一篇营销文案，需要人工审核。

1. AI系统完成文案生成。
2. 系统需要将文案呈现给人类审核员。
3. 系统必须“暂停”执行，等待审核员阅读、思考，并给出“批准”、“驳回”或“修改建议”。
4. 这个等待过程可能持续几秒、几分钟、几小时甚至几天。
5. 一旦审核员给出反馈，系统需要从上次暂停的地方“恢复”执行，并根据反馈采取下一步行动。

这种“暂停-等待-恢复”的模式对传统无状态或简单状态管理的系统提出了巨大挑战：

• 状态持久化：当系统暂停时，当前的执行状态（例如，LLM生成的文案、历史对话、任务上下文等）必须被安全地保存起来，以免丢失。
• 异步性：人工审核是一个异步操作。系统不能阻塞等待，而应该能够处理其他任务，并在收到人工反馈时被异步唤醒。
• 可恢复性：系统必须能够精确地从上次保存的状态点恢复执行，而不是从头开始。
• 多线程/多任务：在生产环境中，可能同时有成百上千个这样的审批流程在进行。系统需要能够独立地管理和跟踪每一个流程的状态。

LangGraph的检查点机制正是为了解决这些挑战而设计的。它提供了一种优雅的方式来保存和恢复图的执行状态，从而完美支持人机协作中的中断与恢复模式。

LangGraph检查点机制深度解析

LangGraph的检查点（Checkpoints）是其状态管理的核心扩展，尤其适用于需要长时间运行、可中断或需要外部干预的图执行。

BaseCheckpointSaver接口

所有检查点保存器都必须实现BaseCheckpointSaver接口。这个接口定义了以下核心方法：

• get(self, config: RunnableConfig) -> Optional[Checkpoint]：根据传入的配置（通常包含thread_id），获取指定线程的最新检查点。
• put(self, config: RunnableConfig, checkpoint: Checkpoint) -> RunnableConfig：保存当前的检查点。它会返回一个更新后的配置，通常包含保存的thread_id和checkpoint_id。
• list(self, config: RunnableConfig) -> Sequence[CheckpointTuple]：列出给定线程的所有检查点（如果支持版本历史）。

Checkpoint对象本身包含了图的当前状态、配置、最近执行的步骤信息等。thread_id是检查点机制中的关键概念，它允许你为每个独立的图执行路径（例如，每个用户会话或每个审批请求）分配一个唯一的标识符，从而隔离和管理它们的状态。

内置实现：MemorySaver与SQLiteSaver

LangGraph提供了几个开箱即用的BaseCheckpointSaver实现：

1. MemorySaver：

   • 特点：将所有检查点存储在内存中。
   • 优点：速度快，简单易用，适合开发和测试环境。
   • 缺点：非持久化，一旦程序重启，所有检查点数据将丢失。不适合生产环境或需要长时间等待人工干预的场景。

2. SQLiteSaver：

   • 特点：将检查点存储在SQLite数据库文件中。
   • 优点：持久化，即使程序重启数据也不会丢失。适合本地开发、单机部署的生产环境，或作为更复杂分布式存储的过渡方案。
   • 缺点：不适用于高并发、分布式部署的场景，因为它基于本地文件系统，锁机制可能成为瓶颈。

在生产环境中，你可能需要实现自定义的BaseCheckpointSaver，将其与PostgreSQL、Redis、MongoDB等分布式数据库集成，以满足高可用、可伸缩和并发的需求。

配置与使用检查点

要在LangGraph中使用检查点，你需要：

1. 实例化一个BaseCheckpointSaver的子类（例如SQLiteSaver）。
2. 在编译图时，将这个saver实例传递给StateGraph.compile()方法。
3. 在运行图时，通过config参数传入一个包含{"configurable": {"thread_id": "your_unique_thread_id"}}的字典。thread_id是用于标识和恢复特定执行路径的关键。

当图在执行过程中遇到需要中断的地方（例如，等待人工输入），它会自动保存当前状态作为一个检查点。当你再次使用相同的thread_id运行图时，LangGraph会自动加载最新检查点，并从该点继续执行。

实战：基于LangGraph检查点构建人工审批工作流

我们将通过一个“内容创作与审核系统”的案例来演示LangGraph检查点的强大功能。

场景设定：内容创作与审核系统

假设我们正在构建一个内部内容管理系统，用于生成博客文章草稿。

1. 用户提供一个主题。
2. LLM生成一篇博客文章草稿。
3. 这篇草稿需要人工审核。审核员可以：
   • 批准（Approve）：文章可以直接发布。
   • 驳回（Reject）：文章不符合要求，需要彻底重写或放弃。
   • 修订（Revise）：文章基本可用，但需要LLM根据审核员的建议进行修改。
4. 如果需要修订，LLM会根据审核员的建议生成新版本，并再次进入人工审核流程，直到被批准或驳回。

这个流程是一个典型的多轮人机协作场景，完美展示了检查点的应用。

步骤一：定义工作流状态

首先，我们定义一个TypedDict作为我们的图状态。它将存储整个工作流中的关键信息。

from typing import List, Literal, TypedDict, Optional
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableConfig
from langchain_community.chat_models import ChatOllama # 或者使用 OpenAI/Anthropic/Google 等
from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver, SQLiteSaver
import os

# 确保环境变量设置，例如 Ollama 的模型名称
# 如果使用 OpenAI，则需要设置 OPENAI_API_KEY
# os.environ["OPENAI_API_KEY"] = "your_openai_api_key"

# 定义图的状态
class ContentReviewState(TypedDict):
    """
    代表内容审核工作流的状态。
    """
    topic: str  # 用户输入的主题
    draft: str  # LLM生成的当前草稿内容
    review_status: Optional[Literal["pending", "approved", "rejected", "revising"]] # 审核状态
    reviewer_feedback: Optional[str] # 审核员的反馈或修改建议
    revisions_count: int # 修订次数
    messages: List[BaseMessage] # 存储对话历史，用于LLM的上下文

状态字段解释：

• topic: 用户最初输入的文章主题。
• draft: 当前LLM生成的文章草稿。
• review_status: 记录当前草稿的审核状态，这对于条件路由至关重要。
  • pending: 等待人工审核。
  • approved: 已批准，可以发布。
  • rejected: 已驳回，流程结束。
  • revising: 正在根据反馈进行修订。
• reviewer_feedback: 审核员提供的具体反馈或修订建议。
• revisions_count: 记录草稿被修订的次数，可用于防止无限循环修订。
• messages: 一个BaseMessage列表，用于存储与LLM的对话历史，确保LLM在修订时有上下文。

步骤二：构建核心节点

接下来，我们定义工作流中的各个节点。每个节点都是一个接受当前状态并返回一个更新状态的函数。

# 初始化LLM模型
# 这里使用 Ollama，你可以替换为 ChatOpenAI 或其他模型
# 请确保你的 Ollama 已经运行，并且拉取了相应的模型，例如 'llama3'
llm = ChatOllama(model="llama3", temperature=0.7) # 降低温度以获得更稳定的输出

# 1. 内容生成节点
def generate_content_node(state: ContentReviewState) -> ContentReviewState:
    """
    根据主题生成文章草稿。
    """
    print("---生成内容节点：开始生成草稿---")
    topic = state["topic"]
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个专业的博客文章撰写者。请根据用户提供的主题撰写一篇详细的博客文章草稿。"),
        ("human", f"主题：{topic}")
    ])
    chain = prompt | llm | StrOutputParser()

    # 第一次生成时，messages为空
    # 如果是修订，messages中会有之前的对话
    new_draft = chain.invoke({"topic": topic})

    # 更新状态
    return {
        "draft": new_draft,
        "review_status": "pending", # 生成后立即进入待审核状态
        "revisions_count": state.get("revisions_count", 0), # 初始化或保持
        "messages": [
            HumanMessage(content=f"主题：{topic}"),
            AIMessage(content=new_draft)
        ]
    }

# 2. 内容修订节点
def revise_content_node(state: ContentReviewState) -> ContentReviewState:
    """
    根据审核员的反馈修订文章草稿。
    """
    print("---修订内容节点：根据反馈修订草稿---")
    draft = state["draft"]
    feedback = state["reviewer_feedback"]
    messages = state["messages"]

    # 将最新的草稿和反馈添加到消息历史中
    messages.append(HumanMessage(content=f"请根据以下反馈修改文章草稿：n{feedback}nn当前草稿：n{draft}"))

    # 构建修订提示
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个专业的博客文章撰写者。请根据用户提供的反馈修订文章草稿。请直接输出修订后的文章内容，不要包含额外说明。"),
        *messages # 包含历史消息作为上下文
    ])

    chain = prompt | llm | StrOutputParser()
    revised_draft = chain.invoke({"feedback": feedback, "draft": draft}) # 实际通过messages传递上下文

    # 更新状态
    return {
        "draft": revised_draft,
        "review_status": "pending", # 修订后再次进入待审核状态
        "revisions_count": state.get("revisions_count", 0) + 1,
        "messages": messages + [AIMessage(content=revised_draft)] # 添加修订后的草稿到历史
    }

# 3. 人工审核等待节点 (这是一个“虚拟”节点，实际等待发生在外部)
def human_review_node(state: ContentReviewState) -> ContentReviewState:
    """
    此节点主要用于展示当前草稿和等待人工输入。
    它不修改状态，只是提供一个暂停点。
    """
    print("---人工审核节点：等待人工输入---")
    print(f"n---当前草稿 ({state['revisions_count']}次修订) ---n{state['draft']}n---等待审核员反馈---n")
    # 在实际系统中，这里不会有 input() 调用，而是将草稿发送到UI或API，等待异步回调
    # 为了演示，我们在此打印并模拟暂停
    return state # 不修改状态，等待外部输入来驱动下一个决策

关于human_review_node的说明：

在LangGraph中，当图执行到达一个需要人工干预的节点时，我们实际上不会在节点函数内部调用input()来等待。这样做会阻塞图的执行线程，这不是一个健壮的异步人机协作模式。

真正的模式是：

1. 图运行到human_review_node。
2. LangGraph保存检查点并暂停。
3. 外部系统（例如一个Web服务）通过thread_id获取到当前检查点，提取出draft内容。
4. Web服务将draft展示给用户界面。
5. 用户在UI中输入反馈和决策（批准/驳回/修订）。
6. Web服务收到用户输入后，再次调用LangGraph的invoke方法，传入thread_id和用户输入作为下一个节点（决策节点）的额外参数。

为了本讲座的演示目的，human_review_node会打印出当前草稿，并模拟一个暂停点。实际的用户输入将通过invoke方法的input参数传入，并在后续的决策节点中处理。

4. 决策路由节点 (这是一个条件路由函数，而不是一个实际节点)

这个函数将根据人工审核结果（由外部传入）决定下一个节点。

def route_review_decision(state: ContentReviewState) -> str:
    """
    根据审核状态决定下一步的路由。
    """
    status = state["review_status"]
    if status == "approved":
        print("---决策路由：草稿已批准---")
        return "approved_node"
    elif status == "rejected":
        print("---决策路由：草稿已驳回---")
        return "rejected_node"
    elif status == "revising":
        print("---决策路由：草稿需要修订---")
        # 检查修订次数，防止无限循环
        if state["revisions_count"] >= 3: # 例如，最多允许3次修订
            print("---修订次数过多，流程结束，请人工介入---")
            return "rejected_node" # 超过修订次数上限，视为驳回
        return "revise_content"
    else:
        # pending状态意味着仍在等待人工输入，这个路由不应该被触发
        # 如果触发，说明逻辑有问题，或者需要继续等待
        print("---决策路由：状态异常或仍在等待人工输入---")
        # 实际情况中，pending状态下，图会暂停，不会走到这里
        # 这里为了演示，可以作为错误处理或者回到等待状态
        return "human_review" # 重新回到人工审核等待

5. 最终输出节点

def approved_node(state: ContentReviewState) -> ContentReviewState:
    """
    处理已批准的文章。
    """
    print("---文章已最终批准，可以发布！---")
    print(f"最终文章：n{state['draft']}")
    return state

def rejected_node(state: ContentReviewState) -> ContentReviewState:
    """
    处理已驳回的文章。
    """
    print("---文章已被驳回，流程结束。---")
    return state

步骤三：组装图与定义边

现在我们使用StateGraph将所有节点和路由连接起来。

# 构建图
workflow = StateGraph(ContentReviewState)

# 添加节点
workflow.add_node("generate_content", generate_content_node)
workflow.add_node("human_review", human_review_node)
workflow.add_node("revise_content", revise_content_node)
workflow.add_node("approved_node", approved_node)
workflow.add_node("rejected_node", rejected_node)

# 设置入口点 (Entry Point)
workflow.set_entry_point("generate_content")

# 添加边和条件路由
# 1. 从生成内容到人工审核
workflow.add_edge("generate_content", "human_review")

# 2. 从修订内容到人工审核
workflow.add_edge("revise_content", "human_review")

# 3. 从人工审核到决策路由 (这里的决策是基于外部输入更新状态后触发的)
# 实际上，从 human_review 节点出来后，我们期望外部系统会更新 review_status
# 然后再次调用 run_graph，这时候会通过 route_review_decision 进行路由
# 因此，human_review 节点后，我们直接连接到条件路由
workflow.add_conditional_edges(
    "human_review",           # 源节点
    route_review_decision,    # 条件路由函数
    {
        "approved_node": "approved_node",
        "rejected_node": "rejected_node",
        "revise_content": "revise_content",
        "human_review": "human_review" # 如果状态异常，回到等待状态
    }
)

# 4. 定义结束点 (END)
workflow.add_edge("approved_node", END)
workflow.add_edge("rejected_node", END)

# 编译图
# 使用 SQLiteSaver 实现持久化检查点
# checkpoint_saver = MemorySaver() # 内存保存器，非持久化
checkpoint_saver = SQLiteSaver.from_file("langgraph_checkpoints.sqlite") # 持久化到SQLite文件

app = workflow.compile(checkpointer=checkpoint_saver)

步骤四：配置检查点与运行图

现在我们来演示如何运行这个工作流，并在人工审核点暂停和恢复。

我们将模拟三次运行：

1. 第一次运行：从头开始，生成草稿，到达human_review节点并暂停。
2. 第二次运行：模拟人工“修订”操作，更新状态，并恢复执行，进入revise_content节点，再次到达human_review节点。
3. 第三次运行：模拟人工“批准”操作，更新状态，并恢复执行，流程结束。

重要的 config 参数：

app.invoke() 或 app.stream() 方法接受一个 config 参数。对于检查点，config 字典必须包含 {"configurable": {"thread_id": "your_unique_id"}}。这个 thread_id 是 LangGraph 用来查找和保存特定执行路径状态的关键。每次调用 invoke 或 stream 时，如果提供了 thread_id，LangGraph 会尝试加载该 thread_id 的最新检查点；如果没有找到，则会从头开始执行。

第一次运行：生成内容并等待审核

# 1. 第一次运行：生成内容并等待审核
print("n--- 第一次运行：生成草稿并等待人工审核 ---n")
initial_input = {"topic": "如何有效学习一门新编程语言"}
thread_id_1 = "content_review_123" # 为这个审批流程定义一个唯一的线程ID

# 运行图，它会在 human_review 节点暂停
# 注意：这里我们不传入 review_status 和 reviewer_feedback，因为是第一次运行
# 状态会在 generate_content 节点中被更新为 "pending"
final_state_1 = None
try:
    for s in app.stream(initial_input, {"configurable": {"thread_id": thread_id_1}}):
        if "__end__" not in s:
            print(s)
            final_state_1 = s # 记录每次迭代的最新状态
except Exception as e:
    print(f"第一次运行发生错误: {e}")

if final_state_1:
    print("n--- 第一次运行结束，图已暂停，等待人工审核 ---")
    print(f"当前草稿内容：n{final_state_1['human_review']['draft']}")
    print(f"当前审核状态：{final_state_1['human_review']['review_status']}")
    print(f"当前线程ID：{thread_id_1}")

# 此时，LangGraph 已经将状态保存到 langgraph_checkpoints.sqlite 文件中
# 我们模拟审核员看到这个草稿并决定需要修订

输出示例（部分，LLM生成内容会不同）：

---生成内容节点：开始生成草稿---
{'generate_content': {'draft': '学习一门新编程语言的有效策略...n[此处为LLM生成的文章内容]n', 'review_status': 'pending', 'revisions_count': 0, 'messages': [HumanMessage(content='主题：如何有效学习一门新编程语言'), AIMessage(content='学习一门新编程语言的有效策略...n[此处为LLM生成的文章内容]n')]}}
---人工审核节点：等待人工输入---

---当前草稿 (0次修订) ---
学习一门新编程语言的有效策略...
[此处为LLM生成的文章内容]
---等待审核员反馈---

--- 第一次运行结束，图已暂停，等待人工审核 ---
当前草稿内容：
学习一门新编程语言的有效策略...
[此处为LLM生成的文章内容]
当前审核状态：pending
当前线程ID：content_review_123

第二次运行：模拟人工“修订”操作，恢复执行

现在，我们模拟人工审核员给出了反馈，并要求修订。我们需要将这些反馈作为输入传递给图，并更新状态以触发修订流程。

# 2. 第二次运行：模拟人工“修订”操作
print("n--- 第二次运行：模拟人工修订，恢复执行 ---n")
reviewer_feedback_2 = "文章内容不错，但缺少实际代码示例，请补充一些Python或JavaScript的代码片段来演示概念。"

# 准备更新状态，以模拟人工输入
# LangGraph 的 invoke/stream 方法允许你传入一个字典作为 input，
# 这个 input 会在下一个节点开始前，与当前状态合并。
# 这里我们直接更新了 review_status 和 reviewer_feedback
# 这样，当图从 human_review 节点恢复时，route_review_decision 就能看到这些更新
human_input_for_revision = {
    "review_status": "revising",
    "reviewer_feedback": reviewer_feedback_2
}

final_state_2 = None
try:
    for s in app.stream(human_input_for_revision, {"configurable": {"thread_id": thread_id_1}}):
        if "__end__" not in s:
            print(s)
            final_state_2 = s
except Exception as e:
    print(f"第二次运行发生错误: {e}")

if final_state_2:
    print("n--- 第二次运行结束，图已暂停，等待人工二次审核 ---")
    print(f"修订后的草稿内容：n{final_state_2['human_review']['draft']}")
    print(f"当前审核状态：{final_state_2['human_review']['review_status']}")
    print(f"当前修订次数：{final_state_2['human_review']['revisions_count']}")
    print(f"当前线程ID：{thread_id_1}")

输出示例（部分）：

---决策路由：草稿需要修订---
{'human_review': {'draft': '学习一门新编程语言的有效策略...n[此处为LLM生成的文章内容]n', 'review_status': 'revising', 'reviewer_feedback': '文章内容不错，但缺少实际代码示例，请补充一些Python或JavaScript的代码片段来演示概念。', 'revisions_count': 0, 'messages': [...]}}
---修订内容节点：根据反馈修订草稿---
{'revise_content': {'draft': '学习一门新编程语言的有效策略...n[此处为LLM根据反馈修订后的文章内容，包含代码示例]n', 'review_status': 'pending', 'reviewer_feedback': '文章内容不错，但缺少实际代码示例，请补充一些Python或JavaScript的代码片段来演示概念。', 'revisions_count': 1, 'messages': [...]}}
---人工审核节点：等待人工输入---

---当前草稿 (1次修订) ---
学习一门新编程语言的有效策略...
[此处为LLM根据反馈修订后的文章内容，包含代码示例]
---等待审核员反馈---

--- 第二次运行结束，图已暂停，等待人工二次审核 ---
修订后的草稿内容：
学习一门新编程语言的有效策略...
[此处为LLM根据反馈修订后的文章内容，包含代码示例]
当前审核状态：pending
当前修订次数：1
当前线程ID：content_review_123

第三次运行：模拟人工“批准”操作，流程结束

现在，我们模拟人工审核员对修订后的草稿表示满意，并批准发布。

# 3. 第三次运行：模拟人工“批准”操作
print("n--- 第三次运行：模拟人工批准，流程结束 ---n")
human_input_for_approval = {
    "review_status": "approved",
    "reviewer_feedback": "修订得很好，可以发布！" # 批准时反馈通常不重要，但也可记录
}

final_state_3 = None
try:
    for s in app.stream(human_input_for_approval, {"configurable": {"thread_id": thread_id_1}}):
        if "__end__" not in s:
            print(s)
            final_state_3 = s
except Exception as e:
    print(f"第三次运行发生错误: {e}")

if final_state_3:
    print("n--- 第三次运行结束，图已完成 ---")
    print(f"最终文章状态：{final_state_3[END]['approved_node']['review_status']}")
    print(f"最终文章内容：n{final_state_3[END]['approved_node']['draft']}")
    print(f"最终修订次数：{final_state_3[END]['approved_node']['revisions_count']}")
    print(f"当前线程ID：{thread_id_1}")

输出示例（部分）：

---决策路由：草稿已批准---
{'human_review': {'draft': '学习一门新编程语言的有效策略...n[此处为LLM根据反馈修订后的文章内容，包含代码示例]n', 'review_status': 'approved', 'reviewer_feedback': '修订得很好，可以发布！', 'revisions_count': 1, 'messages': [...]}}
---文章已最终批准，可以发布！---
{'approved_node': {'draft': '学习一门新编程语言的有效策略...n[此处为LLM根据反馈修订后的文章内容，包含代码示例]n', 'review_status': 'approved', 'reviewer_feedback': '修订得很好，可以发布！', 'revisions_count': 1, 'messages': [...]}}
--- 第三次运行结束，图已完成 ---
最终文章状态：approved
最终文章内容：
学习一门新编程语言的有效策略...
[此处为LLM根据反馈修订后的文章内容，包含代码示例]
最终修订次数：1
当前线程ID：content_review_123

处理驳回（Rejected）情况

如果审核员在任何时候决定“驳回”文章，流程会直接结束。这通过route_review_decision函数中的elif status == "rejected": return "rejected_node"实现。

# 模拟驳回情况 (假设在第一次审核时就驳回了)
print("n--- 模拟驳回情况：新流程 ---")
thread_id_2 = "content_review_456"
initial_input_reject = {"topic": "一篇关于宇宙大爆炸理论的科普文章"}

# 第一次运行，生成草稿
for s in app.stream(initial_input_reject, {"configurable": {"thread_id": thread_id_2}}):
    pass # 忽略打印，直接运行到 human_review 节点

# 模拟人工驳回
human_input_for_reject = {
    "review_status": "rejected",
    "reviewer_feedback": "主题不适合当前受众，请重新选择主题。"
}

final_state_reject = None
try:
    for s in app.stream(human_input_for_reject, {"configurable": {"thread_id": thread_id_2}}):
        if "__end__" not in s:
            print(s)
            final_state_reject = s
except Exception as e:
    print(f"驳回运行发生错误: {e}")

if final_state_reject:
    print("n--- 驳回流程结束 ---")
    print(f"最终文章状态：{final_state_reject[END]['rejected_node']['review_status']}")
    print(f"当前线程ID：{thread_id_2}")

输出示例（部分）：

---决策路由：草稿已驳回---
{'human_review': {'draft': '宇宙大爆炸理论科普...n', 'review_status': 'rejected', 'reviewer_feedback': '主题不适合当前受众，请重新选择主题。', 'revisions_count': 0, 'messages': [...]}}
---文章已被驳回，流程结束。---
{'rejected_node': {'draft': '宇宙大爆炸理论科普...n', 'review_status': 'rejected', 'reviewer_feedback': '主题不适合当前受众，请重新选择主题。', 'revisions_count': 0, 'messages': [...]}}
--- 驳回流程结束 ---
最终文章状态：rejected
当前线程ID：content_review_456

通过这个实战案例，我们可以清晰地看到LangGraph的检查点机制如何与thread_id、状态更新和条件路由协同工作，实现了复杂的人工审批流程。SQLiteSaver确保了即使程序在等待人工审批期间被关闭，其状态也能被完整地保存和恢复。

高级考量与最佳实践

在生产环境中部署基于LangGraph检查点的人机协作系统时，还需要考虑更多高级因素。

持久化策略选择

CheckpointSaver	优点	缺点	适用场景
MemorySaver	简单、快速，无需额外配置	非持久化，进程重启数据丢失	开发、测试、概念验证
SQLiteSaver	持久化，易于设置，文件存储	不支持高并发，不适合分布式系统，性能有限	本地开发、单机部署的低并发应用
自定义Saver	可与任何数据库（PostgreSQL, Redis, MongoDB等）集成	需要额外开发工作	生产环境、高并发、分布式系统、需要特定数据存储和查询功能

对于生产环境，强烈推荐实现一个自定义的BaseCheckpointSaver，将其数据存储在具备高可用性、可扩展性和事务支持的数据库中。这通常涉及将LangGraph的Checkpoint对象序列化（例如，为JSON或Pickle）并存储到数据库中。

并发与多线程管理：thread_id的妙用

thread_id是管理并发人机协作流的关键。每个独立的审批请求或用户会话都应该有一个唯一的thread_id。

• 当一个新请求开始时，生成一个新的thread_id。
• 当用户界面或外部系统需要获取特定审批任务的状态时，它会使用该thread_id去查询检查点。
• 当人工反馈到来时，它会携带相应的thread_id，系统使用此thread_id恢复并继续执行图。

这种机制使得你可以同时运行成千上万个独立的审批流程，而它们的状态互不干扰。

与外部系统集成：API接口设计思路

在实际应用中，LangGraph通常作为后端服务的一部分运行。为了与前端UI或外部系统交互，你需要设计相应的API接口。

示例API接口设想：

API Endpoint	HTTP Method	功能	请求体示例	响应体示例
/start_review	POST	启动一个新的内容审核流程	{"topic": "..."}	{"thread_id": "...", "initial_draft": "..."}
/get_pending_reviews	GET	获取所有等待人工审核的任务列表	(无)	[{"thread_id": "...", "current_draft": "...", "revisions_count": ...}, ...]
/submit_review_feedback	POST	提交人工审核反馈并继续流程	{"thread_id": "...", "decision": "approved"|"rejected"|"revising", "feedback": "..."}	{"thread_id": "...", "new_status": "...", "final_output": "..." (如果流程结束)}
/get_review_status/{thread_id}	GET	获取特定thread_id的最新状态	(无)	{"thread_id": "...", "status": "...", "draft": "...", "history": [...]}

后端服务会：

1. 接收API请求。
2. 根据请求调用LangGraph的invoke或stream方法，并传入相应的thread_id和input。
3. 处理LangGraph的输出（新的状态或最终结果）。
4. 将结果格式化后通过API响应返回。

错误处理与超时机制

• 错误处理：在节点函数中应该包含健壮的错误处理逻辑。如果LLM调用失败或外部API响应异常，应该捕获并记录错误，并更新状态以指示错误发生，而不是让整个图崩溃。LangGraph的图结构可以设计专门的错误处理路径。
• 超时机制：人工审批可能需要很长时间。你需要考虑：
  • 人工响应超时：如果人工审核员在预设时间内（例如24小时）没有给出反馈，系统应该如何处理？是自动驳回、发送提醒、还是将任务标记为“过期”？这可以通过在外部调度器中监控检查点的时间戳来实现。
  • LLM调用超时：确保LLM客户端配置了合理的超时时间，避免LLM响应过慢导致节点长时间阻塞。

审计与追踪

对于所有人工审批流程，记录完整的审计日志至关重要：

• 谁在何时启动了流程？
• LLM生成了什么内容？
• 哪个审核员在何时给出了什么反馈？
• 流程的每个阶段状态是什么？
• 最终决策是什么？

LangGraph的检查点本身就提供了部分审计功能（通过list方法可以查看历史检查点，如果你的Saver支持版本历史）。在此基础上，结合外部日志系统和数据库记录，可以构建一个全面的审计追踪系统。

安全与授权

在多用户环境中，确保只有授权用户才能：

• 启动新的审批流程。
• 查看待审批的任务。
• 提交审批反馈。
• 访问敏感内容（如草稿或反馈）。

这需要在API层和LangGraph的业务逻辑层都实现适当的认证（Authentication）和授权（Authorization）机制。例如，在提交审批反馈时，后端应验证提交者是否有权限审批该thread_id对应的任务。

LangGraph赋能人机协作的强大工具

LangGraph的检查点机制为构建复杂、有状态、可中断的人机协作AI工作流提供了坚实的基础。通过其灵活的图结构、强大的状态管理以及对持久化检查点的支持，开发者能够设计出既能充分发挥LLM智能，又能有效引入人类判断和干预的智能应用。这不仅提高了AI系统的可靠性和安全性，也极大地扩展了AI在关键业务场景中的应用边界。掌握LangGraph的检查点，无疑是现代AI工程师在构建企业级LLM应用时的必备技能。