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

image

时间穿梭
代码可以回滚到历史的某个位置

随着会话次数越来越多,历史记录也会越来越多,这时候会对同一个会话的多个 checkpoint 开成一个组,用同一个 thread_id 来标记。{"thread_id":"1"},这边的 thread_id 也可以理解成会话ID

image

短期记忆-基于内存

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("初始化模型")
Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐