LangChain 入门 Memory 会话记忆

发布时间:2026/6/28 23:32:25
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( modelopenai:gpt-5.4, tools[get_user_info], checkpointerInMemorySaver(), ) # 创建一个线程会话根据 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: Whats 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_provideropenai, # 指定模型提供者阿里兼容 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( modelmodel, tools[get_user_info], checkpointerInMemorySaver(), ) # 创建一个线程会话根据 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 LangChainuv 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:postgreslocalhost:5432/postgres?sslmodedisable 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], checkpointercheckpointer, )以Sqlite为例导入依赖初始化 checkpointer自动建表创建 Agent,指定 checkpointeruv 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 [ langchain1.3.4, langchain-deepseek1.0.1, langchain-tavily0.2.18, langgraph-checkpoint-sqlite3.1.0, notebook7.5.6, python-dotenv1.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_threadFalse connection sqlite3.connect(resources/checkpoint.db, check_same_threadFalse) # 初始化 checkpointer checkpointer SqliteSaver(connection) # 自动建表 checkpointer.setup() # 加载环境变量 load_dotenv() # 获取环境变更值 base_url os.getenv(DASHSCOPE_BASE_URL) api_key os.getenv(DASHSCOPE_API_KEY) print(初始化模型)