HoRain云--LangChain 对话记忆

📅 2026/7/14 0:37:55
HoRain云--LangChain 对话记忆
默认情况下每次 agent.invoke() 都是独立的——Agent 不记得之前聊过什么。Checkpointer检查点保存器让 Agent 能够记住对话历史实现真正的多轮对话。没有 Checkpointer 的问题先看看没有 Checkpointer 时的情况实例from dotenv import load_dotenvload_dotenv()from langchain.agents import create_agentfrom langchain.chat_models import init_chat_modelfrom langchain.messages import HumanMessagemodel init_chat_model(deepseek:deepseek-v4-flash, temperature0)agent create_agent(modelmodel,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 记住对话添加 Checkpointer 后同一 thread_id 下的对话会自动关联实例from langgraph.checkpoint.memory import InMemorySaverfrom langchain.agents import create_agentfrom langchain.chat_models import init_chat_modelfrom langchain.messages import HumanMessage# 创建一个内存 Checkpointercheckpointer InMemorySaver()model init_chat_model(deepseek:deepseek-v4-flash, temperature0)agent create_agent(modelmodel,checkpointercheckpointer, # 传入 Checkpointersystem_prompt你是菜鸟教程 RUNOOB 的助手。,)# 使用 thread_id 来标识对话线程config {configurable: {thread_id: user-001}}# 第一轮result1 agent.invoke({messages: [HumanMessage(content我叫小明我在学 Python)]},configconfig,)print(f第一轮: {result1[messages][-1].content})# 第二轮——使用相同的 thread_idAgent 记住了result2 agent.invoke({messages: [HumanMessage(content我叫什么名字我在学什么)]},configconfig,)print(f第二轮: {result2[messages][-1].content})运行结果第一轮: 你好小明Python 是一门很好的入门语言有什么需要帮助的吗 第二轮: 你叫小明你正在学习 Python。有什么具体问题我可以帮你吗thread_id 是关键。同一个 thread_id 下的对话是连续的不同 thread_id 之间的对话完全隔离。这让你可以用一个 Agent 实例同时服务多个用户。Checkpointer 的工作原理Checkpointer 在每次 Agent 执行后自动保存状态快照checkpoint。下一次使用相同 thread_id 调用时自动从最近的 checkpoint 恢复状态。具体工作流程调用 agent.invoke()传入 config含 thread_idAgent 检查是否有该 thread_id 的 checkpoint如果有加载历史消息追加新消息后继续执行完成后自动保存新的 checkpoint实例from langgraph.checkpoint.memory import InMemorySaverfrom langchain.agents import create_agentfrom langchain.chat_models import init_chat_modelfrom langchain.messages import HumanMessagecheckpointer InMemorySaver()agent create_agent(modelinit_chat_model(deepseek:deepseek-v4-flash, temperature0),checkpointercheckpointer,system_prompt你是菜鸟教程 RUNOOB 的助手。,)config {configurable: {thread_id: demo-001}}# 模拟多轮对话questions [我叫小明,我在学 Python,帮我总结一下关于我的信息,]for i, q in enumerate(questions, 1):result agent.invoke({messages: [HumanMessage(contentq)]},configconfig,)# 查看 checkpoint 状态state agent.get_state(config)print(f\n第 {i} 轮后:)print(f 消息数: {len(state.values.get(messages, []))})print(f 下一步: {state.next})print(f 回复: {result[messages][-1].content[:80]}...)运行结果第 1 轮后: 消息数: 2 下一步: () 回复: 你好小明很高兴认识你。 第 2 轮后: 消息数: 4 下一步: () 回复: Python 是一门非常流行的编程语言简单易学。 第 3 轮后: 消息数: 6 下一步: () 回复: 根据我们的对话你的信息如下你叫小明正在学习 Python。Checkpointer 类型类型存储位置持久化适用场景InMemorySaver内存否重启丢失开发调试、测试SqliteSaverSQLite 文件是单机部署、小规模应用PostgresSaverPostgreSQL是生产环境、多实例共享SqliteSaver 示例实例from langgraph.checkpoint.sqlite import SqliteSaver# SQLite 持久化——重启后数据不丢失checkpointer SqliteSaver.from_conn_string(conversations.db)agent create_agent(modeldeepseek:deepseek-v4-flash,checkpointercheckpointer,)# 用法和 InMemorySaver 完全相同config {configurable: {thread_id: user-001}}result agent.invoke({messages: [{role: user, content: 你好}]},configconfig,)# 重启程序后相同 thread_id 的对话依然存在管理对话线程查看对话状态实例# 查看对话状态state agent.get_state(config)print(f下一步: {state.next}) # () 表示空闲print(f消息数: {len(state.values.get(messages, []))})# 查看对话历史for msg in state.values.get(messages, []):print(f [{msg.type}] {str(msg.content)[:60]})创建新线程实例# 不同的 thread_id 不同的独立对话config_alice {configurable: {thread_id: alice}}config_bob {configurable: {thread_id: bob}}# Alice 的对话agent.invoke({messages: [HumanMessage(content我是 Alice)]},configconfig_alice,)# Bob 的对话——完全独立不知道 Alice 说了什么agent.invoke({messages: [HumanMessage(content我是 Bob)]},configconfig_bob,)# 验证隔离性alice_state agent.get_state(config_alice)bob_state agent.get_state(config_bob)print(fAlice 对话消息数: {len(alice_state.values[messages])})print(fBob 对话消息数: {len(bob_state.values[messages])})更新状态——手动修改对话有时你需要手动修改对话状态比如清空对话、插入系统消息等实例from langchain.messages import SystemMessage# 更新状态——插入一条系统消息agent.update_state(config,{messages: [SystemMessage(content用户升级到了 VIP 会员)]})# 之后的对话会包含这条插入的消息update_state() 的参数会通过 add_messages reducer 处理对 messages 字段而言所以新消息会追加而不是覆盖。如果需要覆盖整个状态需要使用不同的方法。