Hcc的Blog

嵌入式 · AI · 折腾不止

0%

LangChain 学习(5) LangChain 的对话记忆、跨会话存储、人工介入以及多 Agent 的实现

发表于 分类于 ，

LangChain 对话记忆 ———— Checkpointer

  在默认情况下，每次 agent.invoke() 都是独立的，Agent 不记得之前聊过什么。Checkpointer 让 Agent 能够记住对话历史，实现真正的多轮对话。
  在没有 Checkpinter 的时候，第二轮调用 agent 回答时，不会记住第一轮说过的话。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    system_prompt="你是菜鸟教程 RUNOOB 的助手。",
)

# 第一轮
result1 = agent.invoke({
    "messages": [HumanMessage(content="我叫小明")]
})
print(f"第一轮: {result1['messages'][-1].content}")

# 第二轮——Agent 不记得第一轮的内容！
result2 = agent.invoke({
    "messages": [HumanMessage(content="我叫什么名字？")]
})
print(f"第二轮: {result2['messages'][-1].content}")

  这时候我们可以使用 Checkpointer 来记住对话内容，示例如下：

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
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

# 创建一个内存 Checkpointer
checkpointer = InMemorySaver()

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    checkpointer=checkpointer,  # 传入 Checkpointer
    system_prompt="你是菜鸟教程 RUNOOB 的助手。",
)

# 使用 thread_id 来标识对话线程
config = {"configurable": {"thread_id": "user-001"}}

# 第一轮
result1 = agent.invoke(
    {"messages": [HumanMessage(content="我叫小明，我在学 Python")]},
    config=config,
)
print(f"第一轮: {result1['messages'][-1].content}")

# 第二轮——使用相同的 thread_id，Agent 记住了！
result2 = agent.invoke(
    {"messages": [HumanMessage(content="我叫什么名字？我在学什么？")]},
    config=config,
)
print(f"第二轮: {result2['messages'][-1].content}")

  我们先创建一个内存，在 Agent 创建的时候传入该内存，然后在 config 中使用 thread_id 来标记对话线程，这样 Agent 就会知道哪些话是同一对话里的，可以做到自动关联。

  Checkpointer 的工作原理是这样的，它在每次 Agent 执行后自动保存状态快照（checkpoint）。下一次使用相同 thread_id 调用时，自动从最近的 checkpoint 恢复状态。具体工作流程：调用 agent.invoke()，传入 config（含 thread_id），Agent 检查是否有该 thread_id 的 checkpoint，如果有，加载历史消息，追加新消息后继续，执行完成后，自动保存新的 checkpoint。

  Checkpointer 一般有三种类型如下：

类型 存储位置 持久化 适用场景
InMemorySaver 内存 否 开发调试、测试
SqliteSaver SQLite 文件 是 单机部署、小规模应用
PostgresSaver PostgreSQL 是 生产环境、多实例共享

LangChain 跨会话存储 ———— store

  Checkpointer 解决了”单个对话内记忆”的问题。但如果你需要在不同对话之间共享数据——比如用户偏好、学习进度——就需要用到 Store。

  Store 使用 命名空间 + 键 的层级结构来组织数据，其基本操作如下：

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
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# 写入数据：put(namespace, key, value)
# namespace 是元组，key 是字符串，value 是字典
store.put(
    ("users", "user_001"),           # 命名空间
    "preferences",                    # 键
    {                                 # 值
        "theme": "dark",
        "language": "zh-CN",
        "level": "入门",
    }
)

store.put(
    ("users", "user_001"),
    "progress",
    {
        "completed_courses": ["HTML 基础", "Python 基础"],
        "total_hours": 35,
    }
)

# 读取数据：get(namespace, key)
prefs = store.get(("users", "user_001"), "preferences")
print(f"偏好设置: {prefs.value}")

progress = store.get(("users", "user_001"), "progress")
print(f"学习进度: {progress.value}")

# 搜索数据：search(namespace)
all_user_data = store.search(("users", "user_001"))
print(f"\n用户的所有数据 ({len(all_user_data)} 项):")
for item in all_user_data:
    print(f"  {item.key}: {item.value}")

# 删除数据：delete(namespace, key)
store.delete(("users", "user_001"), "preferences")
deleted = store.get(("users", "user_001"), "preferences")
print(f"\n删除后: {deleted}")  # None

  将 Store 传给 create_agent()，Agent 中的所有工具都能通过 InjectedStore 访问它：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@tool
def query_course_info(
    course_name: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """查询菜鸟教程 RUNOOB 中课程的详细信息。

    Args:
        course_name: 课程名称
    """
    item = store.get(("runoob", "courses"), "catalog")
    catalog = item.value if item else {}

    if course_name in catalog:
        info = catalog[course_name]
        return (
            f"《{course_name}》- 价格：{info['price']}，"
            f"时长：{info['hours']}小时，难度：{info['level']}"
        )
    return f"未找到课程《{course_name}》"

  InMemoryStore 的数据在程序重启后丢失。生产环境可以使用 PostgresStore 等持久化方案：

1
2
3
4
5
6
7
# 开发阶段
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()

# 生产环境（需要 PostgreSQL）
# from langgraph.store.postgres import PostgresStore
# store = PostgresStore.from_conn_string("postgresql://...")

人工介入

  在生产环境中，有些操作需要人工确认，例如发送邮件、执行删除、处理支付等等。人工介入（HITL，Human-in-the-Loop）让 Agent 在关键时刻暂停，等待人工审批后继续。

interrupt() —— 在工具中暂停执行

  interrupt() 函数可以让工具执行到一半时暂停，等待外部输入后再继续，示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langgraph.types import interrupt

# 在工具中使用 interrupt() 暂停
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件（需要人工审批）"""
    # 暂停执行，向外部发送审批请求
    approval = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "请确认是否发送此邮件？"
    })

    # 等待外部传入 approval 后继续
    if approval.get("approved"):
        return f"邮件已发送给 {to}"
    else:
        return f"邮件发送已被拒绝：{approval.get('reason', '用户取消')}"

  interrupt() 的工作流程如下：

  • 工具调用 interrupt()，Agent 暂停执行。
  • 外部系统获取中断信息，展示给用户
  • 用户做出决定后，通过 Command(resume=…) 恢复执行
  • interrupt() 返回用户传入的值，工具继续执行

  审批流程的完整示例如下：

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
from dotenv import load_dotenv
load_dotenv()

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage


@tool
def delete_course(course_name: str) -> str:
    """删除课程（需要审批）。

    Args:
        course_name: 要删除的课程名称
    """
    # 暂停并等待审批
    approval = interrupt({
        "action": "delete_course",
        "course": course_name,
        "message": f"确认删除课程《{course_name}》？此操作不可撤销。"
    })

    if approval.get("confirmed"):
        return f"课程《{course_name}》已删除"
    else:
        return f"删除操作已取消"


checkpointer = InMemorySaver()
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[delete_course],
    checkpointer=checkpointer,
    system_prompt="你是菜鸟教程 RUNOOB 的管理员助手。",
)

config = {"configurable": {"thread_id": "admin-001"}}

# 第一步：发起删除请求（会触发中断）
print("=== 开始执行 ===")
result = agent.invoke(
    {"messages": [HumanMessage(content="请删除课程《过时的 Java 教程》")]},
    config=config,
)

  除了在工具中使用 interrupt()，我们还可以在 create_agent() 中设置全局中断点，示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[some_tool],
    checkpointer=checkpointer,

    # 在工具节点之前暂停（每次调用工具前都需要审批）
    interrupt_before=["tools"],

    # 在模型节点之后暂停（每次模型回复后都可以检查）
    # interrupt_after=["model"],
)

  下面是一个实际 Web 应用中的人工干预实例：

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
# 后端：接收用户消息，处理到中断点，返回中断信息
def handle_user_message(thread_id: str, message: str):
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        {"messages": [HumanMessage(content=message)]},
        config=config,
    )
    state = agent.get_state(config)

    # 检查是否在等待审批
    if state.tasks and state.tasks[0].interrupts:
        return {
            "status": "pending_approval",
            "interrupt": state.tasks[0].interrupts[0].value,
            "thread_id": thread_id,
        }

    return {
        "status": "completed",
        "reply": result["messages"][-1].content,
    }

# 后端：处理用户审批
def handle_approval(thread_id: str, approved: bool, reason: str = ""):
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        Command(resume={"confirmed": approved, "reason": reason}),
        config=config,
    )
    return {"status": "completed", "reply": result["messages"][-1].content}

LangChain 多 Agent

  当一个任务太复杂，单个 Agent 难以胜任时，我们可以创建多个各司其职的 Agent，让它们像团队一样协作。

  单个 Agent 可能会存在以下问题：system_prompt 太长会导致模型注意力分散，工具太多会增加模型选择工具的出错概率，不同类型的任务需要不同的专业知识和行为风格。

方式一、子 Agent 作为工具

  将 Agent 编译成 CompiledStateGraph，然后作为一个工具注册给父 Agent：

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
70
71
72
73
74
75
76
77
78
79
80
81
from dotenv import load_dotenv
load_dotenv()

from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)

# 子 Agent 1：天气专家
@tool
def get_weather(city: str) -> str:
    """查询天气"""
    data = {"杭州": "晴，25°C", "北京": "多云，18°C"}
    return data.get(city, f"{city}: 数据暂缺")

weather_agent = create_agent(
    model=model,
    tools=[get_weather],
    name="weather_expert",  # 名字用于标识和日志
    system_prompt="你是天气专家，专门回答天气相关问题。回答要简洁。",
)

# 子 Agent 2：计算专家
@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    result = eval(expression, {"__builtins__": {}}, {})
    return f"{expression} = {result}"

math_agent = create_agent(
    model=model,
    tools=[calculate],
    name="math_expert",
    system_prompt="你是数学专家，专门进行数学计算。回答要简洁。",
)

# 父 Agent：协调者
# 将子 Agent 作为工具注册
@tool
def ask_weather_expert(question: str) -> str:
    """向天气专家咨询天气相关问题。

    Args:
        question: 关于天气的问题
    """
    result = weather_agent.invoke(
        {"messages": [HumanMessage(content=question)]}
    )
    return result["messages"][-1].content

@tool
def ask_math_expert(question: str) -> str:
    """向数学专家咨询数学计算问题。

    Args:
        question: 数学计算问题
    """
    result = math_agent.invoke(
        {"messages": [HumanMessage(content=question)]}
    )
    return result["messages"][-1].content

coordinator = create_agent(
    model=model,
    tools=[ask_weather_expert, ask_math_expert],
    system_prompt="""你是协调助手。根据用户问题选择合适的专家：
- 天气相关问题 → 使用 ask_weather_expert
- 数学计算问题 → 使用 ask_math_expert
- 如果同时涉及多个领域，依次咨询各个专家""",
)

# 测试复合问题
result = coordinator.invoke({
    "messages": [HumanMessage(
        content="杭州今天天气怎么样？如果温度是 25 度，换算成华氏度是多少？"
        "（公式：华氏度 = 摄氏度 × 9/5 + 32）"
    )]
})
print(result["messages"][-1].content)

方式二、用 name 参数区分 Agent

  当你将子 Agent 作为工具嵌入时，设置 name 参数有助于追踪消息来源：

1
2
3
4
5
6
7
8
9
10
# name 参数的作用：
# 1. 编译后的图中使用该名称
# 2. 作为子图节点嵌入父图时使用该名称
# 3. 所有 AI 消息被标记为该名称

agent = create_agent(
    model=model,
    tools=[...],
    name="customer_service",  # 给 Agent 命名
)

方式三、Middleware 实现 Agent 路由

  更加复杂的多 Agent 场景可以通过 Middleware 实现动态路由：

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
from langchain.agents.middleware import before_model


# 定义不同专家使用的工具集
general_tools = [tool_a, tool_b]
admin_tools = [tool_c, tool_d]

@before_model
def route_by_user_role(state, runtime):
    """根据用户角色动态切换可用工具"""
    context = runtime.context
    if context is None:
        return None

    user_role = context.get("user_role", "user")

    # 不同角色看到不同的工具
    if user_role == "admin":
        available_tools = general_tools + admin_tools
    else:
        available_tools = general_tools

    # 注意：before_model 不能直接修改 tools，
    # 需要配合 wrap_model_call 或 request.override 来实现
    return None

  多 Agent 有以下几种架构：

模式 结构 适用场景
协调者模式 一个父Agent —> 多个子 Agent 工具 任务类型明确可分类
接力模式 Agent A 的输出 -> Agent B 的输入 流水线式处理(生成 -> 审核 -> 润色)
辩论模式 多个 Agent 并行输出 -> 汇总决策 需要多角度分析问题