LangChain 入门 Memory 会话记忆
·
在 LangChain 短期记忆是通过 AgentState 实现的,而会话历史(也就是消息列表)是 AgentState 的一部分
LangChain 提供了 Checkpointer 对象来保存 AgentState,每一次用户与AI的交互都会生成一个快照,记录为一个 checkpoint。
每次请求时,会拿到最近一次快照,再加上当前的消息,拼接好,发给模型,可以基于历史消息进行回答了

时间穿梭
代码可以回滚到历史的某个位置
随着会话次数越来越多,历史记录也会越来越多,这时候会对同一个会话的多个 checkpoint 开成一个组,用同一个 thread_id 来标记。{"thread_id":"1"},这边的 thread_id 也可以理解成会话ID

短期记忆-基于内存
from langchain.agents import create_agent
# 导入 checkpoint 的依赖
from langgraph.checkpoint.memory import InMemorySaver
def get_user_info() -> str:
"""Look up information about the current user."""
return "No user profile on file."
agent = create_agent(
model="openai:gpt-5.4",
tools=[get_user_info],
checkpointer=InMemorySaver(),
)
# 创建一个线程(会话),根据 thread_id 查找历史记录,拼接后发给模型
thread_config = {"configurable": {"thread_id": "1"}}
response = agent.invoke(
{"messages": [{"role": "user", "content": "Hi! My name is Bob."}]},
thread_config,
)["messages"][-1].content
print(response) # "Hi Bob! Nice to see you here. How are you doing?"
response = agent.invoke(
{"messages": [{"role": "user", "content": "What's my name?"}]},
thread_config,
)["messages"][-1].content
print(response) # "You are Bob!"
完整代码
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
# 导入 checkpoint 的依赖
from langgraph.checkpoint.memory import InMemorySaver
from dotenv import load_dotenv
import os
# 加载环境变量
load_dotenv()
# 获取环境变更值
base_url = os.getenv("DASHSCOPE_BASE_URL")
api_key = os.getenv("DASHSCOPE_API_KEY")
print("初始化模型")
# 初始化模型
model = init_chat_model(
model = "qwen3.6-plus",
model_provider="openai", # 指定模型提供者(阿里兼容 openai)
base_url = base_url,
api_key = api_key,
temperature = 1.5,
top_p = 0.9,
)
def get_user_info() -> str:
"""Look up information about the current user."""
return "No user profile on file."
agent = create_agent(
model=model,
tools=[get_user_info],
checkpointer=InMemorySaver(),
)
# 创建一个线程(会话),根据 thread_id 查找历史记录,拼接后发给模型
thread_config = {"configurable": {"thread_id": "thread_1"}}
response = agent.invoke(
{"messages": [HumanMessage(content="你好,我是VipSoft,我喜欢太极!")]},
thread_config,
)["messages"][-1].content
print(response) # "Hi Bob! Nice to see you here. How are you doing?"
response = agent.invoke(
{"messages": [HumanMessage(content="我喜欢什么运动!")]},
thread_config,
)["messages"][-1].content
print(response)
D:\OpenSource\Python\VipLangChain\.venv\Scripts\python.exe D:\OpenSource\Python\VipLangChain\memory.py
初始化模型
你好 VipSoft!很高兴认识你。🥋 系统显示你是新访问的用户(No user profile on file),很高兴能在这个会话中为你提供服务!
太极拳是一项非常修身养性的运动,讲究刚柔并济、行云流水。你是练习了很久,还是刚刚开始接触这项传统武术呢?如果有关于太极招式、练习心得或者养生方面的话题,我也很乐意和你交流!
你刚才告诉我了呀!你喜欢 **太极** (Tai Chi)! 🥋
Process finished with exit code 0
长期记忆-基于数据库
在生产环境中,一般采用数据库的方式
Short-term memory - Docs by LangChain
uv add langgraph-checkpoint-postgres
from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver
def get_user_info() -> str:
"""Look up information about the current user."""
return "No user profile on file."
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup() # auto create tables in PostgreSQL
agent = create_agent(
"gpt-5.5",
tools=[get_user_info],
checkpointer=checkpointer,
)
以Sqlite为例,
- 导入依赖
- 初始化 checkpointer
- 自动建表
- 创建 Agent,指定 checkpointer
uv add langgraph-checkpoint-sqlite
pyproject.toml
[project]
name = "vip-langchain"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"langchain>=1.3.4",
"langchain-deepseek>=1.0.1",
"langchain-tavily>=0.2.18",
"langgraph-checkpoint-sqlite>=3.1.0",
"notebook>=7.5.6",
"python-dotenv>=1.2.2",
]
[[tool.uv.index]]
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
default = true
完整代码
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
# 导入 checkpoint 的依赖
from langgraph.checkpoint.memory import InMemorySaver
from dotenv import load_dotenv
import os
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
# 连接 sqlite
# sqlite 默认会检查创建连接时用的线程,和后面使用的线程是不是同一个,如果不是就报错。所以 check_same_thread=False
connection = sqlite3.connect("resources/checkpoint.db", check_same_thread=False)
# 初始化 checkpointer
checkpointer = SqliteSaver(connection)
# 自动建表
checkpointer.setup()
# 加载环境变量
load_dotenv()
# 获取环境变更值
base_url = os.getenv("DASHSCOPE_BASE_URL")
api_key = os.getenv("DASHSCOPE_API_KEY")
print("初始化模型")更多推荐


所有评论(0)