在 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-postgresfrom 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-sqlitepyproject.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("初始化模型")