LangGraph 高级实战（五）：记忆持久化与人机协作 (Human-in-the-Loop)

1. 现实世界的需求

在实验室里跑 Demo 和在生产环境运行 App 是两码事。在生产环境中，我们面临两个巨大的挑战：

  1. 长时记忆（Session Management）：用户今天问了“我的名字叫张三”，明天问“我叫什么？”，Agent 必须能回答。这需要我们将 State 保存到硬盘或数据库中。
  2. 安全围栏（Safety Guardrails）：如果用户让 Agent “给老板发一封骂人的邮件”，或者“转账 100 万”，我们不能让 Agent 直接执行。我们需要一个机制，让程序在执行危险工具前暂停，等待人类管理员点击“批准”或“拒绝”。

LangGraph 原生支持这两个功能，且实现起来令人惊讶地简单。

2. 环境准备

我们需要 langgraph-checkpoint 库来管理状态保存。

1
2
3
4
5
# 安装 checkpoint 库
pip install langgraph-checkpoint

# 确保核心库
pip install langgraph langchain langchain-openai

3. 实战一：给 Agent 装上“海马体” (Persistence)

我们将使用 LangGraph 自带的 MemorySaver。它是一个内存数据库（类似于 Python 的字典），用于演示。在实际生产中，你可以轻松替换为 PostgresSaver 或 SqliteSaver。

我们将复用上一篇的代码结构，但稍作修改以支持持久化。

代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver # <--- 新引入

# 1. 定义工具和模型
@tool
def search_weather(city: str):
    """查询天气"""
    return f"{city} 的天气是晴朗，气温 25 度。"

tools = [search_weather]
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# 2. 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 3. 定义节点逻辑
def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

tool_node = ToolNode(tools)

def tools_condition(state: State):
    if state["messages"][-1].tool_calls:
        return "tools"
    return END

# 4. 构建图
graph = StateGraph(State)
graph.add_node("agent", chatbot)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", tools_condition, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

# --- 关键改变：引入 Checkpointer ---
# 初始化一个内存保存器
memory = MemorySaver()

# 在编译图时，传入 checkpointer
app = graph.compile(checkpointer=memory)

# 5. 运行测试
# 为了区分不同的对话，我们需要提供一个 thread_id (线程 ID)
# 这就像你在 ChatGPT 左侧栏看到的每一个独立的 Chat History
config = {"configurable": {"thread_id": "user_123_session_1"}}

print("--- 第一轮对话 ---")
input_1 = {"messages": [HumanMessage(content="你好，我叫小明，住在上海。")]}
# 注意：我们必须传入 config
app.invoke(input_1, config=config)
print("Agent 回复完毕。")

print("\n--- 模拟程序重启或第二天 ---")
# 我们不需要传递上下文，只问新问题
# 只要 thread_id 一样，Agent 就能读取之前的状态
input_2 = {"messages": [HumanMessage(content="我住在哪里？那里天气怎么样？")]}

for event in app.stream(input_2, config=config):
    for key, value in event.items():
        if "messages" in value:
            print(f"[{key}] 输出: {value['messages'][-1].content}")

深度解析

当你运行这段代码，你会发现第二轮对话中，Agent 准确知道了用户的地点是“上海”，并调用了天气工具。 这是因为 MemorySaver 在每一步运行后，都会把当前的 State 快照保存下来。下次运行时，LangGraph 会先根据 thread_id 从内存加载 State，然后再把新消息 append 进去。

4. 实战二：人机协作 (Human-in-the-Loop)

现在，我们给 Agent 一个危险的工具：transfer_money（转账）。我们要求：Agent 在决定调用这个工具后，必须暂停，等待我们在控制台输入“yes”才能继续，否则拒绝执行。

修改代码：添加中断机制

我们只需要修改 compile 阶段的参数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
# ... (前面的代码保持不变，除了新增一个工具)

@tool
def transfer_money(amount: int, to_user: str):
    """转账给指定用户。这是一个敏感操作！"""
    return f"成功转账 {amount} 元给 {to_user}。"

# 更新工具列表
tools = [search_weather, transfer_money]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)

# 重新构建图 (逻辑不变)
graph = StateGraph(State)
graph.add_node("agent", chatbot)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", tools_condition, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

# --- 关键改变：设置中断点 ---
memory = MemorySaver()

# interrupt_before=["tools"]: 意思是，在进入 "tools" 节点之前，强制暂停。
app = graph.compile(
    checkpointer=memory, 
    interrupt_before=["tools"] 
)

# ----------------- 运行流程 -----------------

thread_config = {"configurable": {"thread_id": "secure_session_01"}}

print("--- 步骤 1: 用户发起转账请求 ---")
app.invoke(
    {"messages": [HumanMessage(content="转账 500 块给张三")]}, 
    config=thread_config
)

# 此时，程序已经运行完 "agent" 节点。
# Agent 已经生成了 ToolCall (决定转账)，下一步本该去 "tools" 节点。
# 但因为 interrupt_before，程序在这里暂停了。

# 我们可以查看当前的 State
current_state = app.get_state(thread_config)
print("\n--- 程序已暂停。当前待执行的动作: ---")
last_message = current_state.values["messages"][-1]
print(f"工具调用: {last_message.tool_calls}")

# --- 步骤 2: 人类介入审核 ---
approval = input("\n管理员：是否批准这笔转账？(输入 yes/no): ")

if approval.lower() == "yes":
    print("--- 管理员批准。继续执行... ---")
    # 传入 None 表示不修改状态，只是恢复运行
    # Stream 会从暂停的地方（进入 tools 节点前）继续跑
    for event in app.stream(None, config=thread_config):
         for key, value in event.items():
            if "messages" in value:
                last_msg = value["messages"][-1]
                # 打印工具的执行结果
                if hasattr(last_msg, 'tool_calls'): 
                    pass # 这是 agent 的消息
                else:
                    print(f"[{key}] 结果: {last_msg.content}") # 这是 tool 的结果
else:
    print("--- 管理员拒绝。操作终止。 ---")
    # 在实际应用中，你可能会注入一条"用户拒绝操作"的消息给 Agent
    # 这里我们简单地什么都不做

深度解析：HITL 的工作原理

  1. 暂停 (Suspend)：当图运行到 tools 节点门口时，发现配置了 interrupt_before，于是它把当前的 State 保存到 Checkpointer 中，然后停止运行。
  2. 检查 (Inspect)：你可以使用 app.get_state(config) 来查看“快照”。在这个快照里，你可以清楚地看到 Agent 想要调用的工具和参数（比如 amount: 500）。
  3. 恢复 (Resume)：当你再次调用 app.stream(None, config) 时，LangGraph 发现这个 thread_id 处于暂停状态，它会读取快照，然后尝试跨过那个暂停点，继续执行 tools 节点。

5. 还能更高级吗？修改 Agent 的记忆

Human-in-the-Loop 不仅仅是“批准/拒绝”。你甚至可以篡改 Agent 的记忆。

比如，Agent 想转账 500。你可以修改 State，把 500 改成 50，然后让它继续运行。

1
2
3
4
5
6
7
8
9
10
11
12
# (伪代码示例)
if approval == "change":
    # 获取当前状态
    state = app.get_state(thread_config)
    # 修改状态里的 ToolCall 参数
    last_msg = state.values["messages"][-1]
    last_msg.tool_calls[0]['args']['amount'] = 50 
    
    # 更新状态数据库
    app.update_state(thread_config, {"messages": last_msg})
    # 继续运行
    app.invoke(None, thread_config)

这赋予了开发者上帝视角，可以在运行时动态纠正 AI 的错误。

6. 系列总结

恭喜你！你已经完成了 LangChain 和 LangGraph 的从入门到精通之旅。

让我们回顾一下我们的旅程：

  1. 第一篇：我们用 LCEL 替代了原生 API，实现了组件的模块化。
  2. 第二篇：我们引入 RAG，解决了知识截止和幻觉问题。
  3. 第三篇：我们初识 LangGraph，打破了线性链条，实现了循环工作流。
  4. 第四篇：我们构建了 ReAct Agent，让 AI 学会了自主使用工具。
  5. 第五篇：我们加上了 Persistence 和 HITL，让 Agent 具备了长期记忆和安全边界。

这五篇文章构成了现代 LLM 应用开发的完整骨架。接下来的路，就需要你在这个骨架上填充具体的业务逻辑了。祝你在 AI 开发的道路上好运！