【LangChain 记忆系统】记忆篇
·
LangChain 记忆系统
目录
-
-
问题引入：无记忆的对话
-
基础方案：InMemoryChatMessageHistory
-
进阶方案：RunnableWithMessageHistory
-
持久化方案：RedisChatMessageHistory
-
1. 记忆系统（Memory）
1.1 问题引入：无记忆的对话
核心问题
大语言模型本身是无状态的。每次对话请求都是独立的，模型不会自动记住之前的对话内容。
代码演示
from langchain.chat_models import init_chat_model
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
import os
# 初始化模型
llm = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 构建简单链
prompt = PromptTemplate.from_template("请回答我的问题：{question}")
parser = StrOutputParser()
chain = prompt | llm | parser
# 第一轮对话
print(chain.invoke({"question": "我叫张三，你叫什么?"}))
# 第二轮对话
print(chain.invoke({"question": "你知道我是谁吗?"}))
输出结果
# 第一轮：模型知道"我叫张三" 你好，张三！我叫通义千问（Qwen），是阿里云研发的超大规模语言模型。 # 第二轮：模型"遗忘"了 不知道你是谁。我是一个AI助手，没有能力识别或获取用户的身份信息。
问题分析
┌─────────────────────────────────────────────────────────┐
│ 独立调用（无记忆） │
├─────────────────────────────────────────────────────────┤
│ │
│ invoke({"question": "我叫张三..."}) ──► LLM ──► 输出 │
│ │ │
│ ▼ │
│ invoke({"question": "你知道..."}) ──► LLM ──► 输出 │
│ │
│ ⚠️ 两次调用完全独立，模型无法跨请求记住上下文 │
└─────────────────────────────────────────────────────────┘
1.2 基础方案：InMemoryChatMessageHistory
核心思想
手动将对话历史存储在内存中，每次调用时将历史消息一起发送给模型。
原理图
┌─────────────────────────────────────────────────────────┐
│ 手动管理历史消息（InMemory） │
├─────────────────────────────────────────────────────────┤
│ │
│ history.add_user_message("我叫张三...") │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ InMemoryChatMessageHistory │ │
│ │ ├── HumanMessage: 我叫张三... │ │
│ │ └── AIMessage: 我是通义千问... │ │
│ └─────────────────────────────────┘ │
│ │ │
│ ▼ │
│ llm.invoke(history.messages) ──► LLM │
│ │
└─────────────────────────────────────────────────────────┘
完整代码
from langchain.chat_models import init_chat_model
from langchain_core.chat_history import InMemoryChatMessageHistory
from loguru import logger
import os
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
# 初始化模型
llm = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 创建内存历史记录
history = InMemoryChatMessageHistory()
# ===== 第一轮对话 =====
# 添加用户消息
history.add_user_message("我叫张三，我的爱好是学习")
# 调用模型
ai_message = llm.invoke(history.messages)
logger.info(f"第一次回答\n{ai_message.content}")
# 将AI回复添加到历史
history.add_message(ai_message)
# ===== 第二轮对话 =====
# 添加新用户消息
history.add_user_message("我叫什么？我的爱好是什么？")
# 再次调用模型（携带完整历史）
ai_message2 = llm.invoke(history.messages)
logger.info(f"第二次回答\n{ai_message2.content}")
history.add_message(ai_message2)
# 查看完整历史
for message in history.messages:
logger.info(message.content)
输出示例
第一次回答 你好，张三！很高兴认识你！学习是一个非常棒的习惯... 第二次回答 你叫张三，你的爱好是学习。
优点与局限
| 优点 | 局限 |
|---|---|
| 简单直接 | 程序重启后历史丢失 |
| 适合单次会话 | 无法跨进程共享 |
| 调试方便 | 不适合生产环境 |
1.3 进阶方案：RunnableWithMessageHistory
文件：
-
07_memory/Memory_RunnableWithMessageHistory.py（全局 history 对象） -
07_memory/Memory_RunnableWithMessageHistoryV2.py（store 字典管理多会话）
核心思想
使用 RunnableWithMessageHistory 装饰器自动管理历史消息，无需手动调用 add_user_message() 和 add_message()。
方案一：全局 history 对象
文件：Memory_RunnableWithMessageHistory.py
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory, RunnableConfig
from langchain.chat_models import init_chat_model
from langchain_core.chat_history import InMemoryChatMessageHistory
from loguru import logger
import os
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
# 初始化模型
llm = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 定义 Prompt（使用 MessagesPlaceholder 注入历史）
prompt = ChatPromptTemplate.from_messages([
MessagesPlaceholder(variable_name="history"), # 历史消息注入点
("human", "{input}") # 用户当前输入
])
# 构建链
parser = StrOutputParser()
chain = prompt | llm | parser
# 创建内存历史
history = InMemoryChatMessageHistory()
# 包装为带历史的 Runnable
runnable = RunnableWithMessageHistory(
chain,
get_session_history=lambda session_id: history, # 获取历史的函数
input_messages_key="input", # 用户输入的 key
history_messages_key="history" # 历史消息的 key
)
# 清空历史（可选）
history.clear()
# 配置会话
config = RunnableConfig(configurable={"session_id": "user-001"})
# 对话
logger.info(runnable.invoke({"input": "我叫张三，我爱好学习。"}, config))
logger.info(runnable.invoke({"input": "我叫什么？爱好是什么？"}, config))
方案二：使用 store 管理多会话
文件：Memory_RunnableWithMessageHistoryV2.py
from langchain.chat_models import init_chat_model
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
import os
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
# 初始化模型
llm = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 全局会话存储（支持多用户/多会话）
store = {}
def get_session_history(session_id: str):
"""根据 session_id 获取或创建历史记录"""
if session_id not in store:
store[session_id] = InMemoryChatMessageHistory()
return store[session_id]
# 定义 Prompt
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个友好的中文助理，会根据上下文回答问题。"),
MessagesPlaceholder("history"), # 注入历史消息
("human", "{question}") # 当前用户输入
])
# 构建链
memory_chain = prompt | llm | StrOutputParser()
# 包装为带历史的 Runnable
with_history = RunnableWithMessageHistory(
memory_chain,
get_session_history, # 函数：获取/创建历史
input_messages_key="question", # prompt 中用户输入的变量名
history_messages_key="history" # prompt 中历史消息的变量名
)
# 使用 session_id 区分不同用户
cfg = {"configurable": {"session_id": "user-001"}}
# 第一次对话
print("用户：我叫张三。")
print("AI：", with_history.invoke({"question": "我叫张三。"}, cfg))
# 第二次对话（模型记得之前的内容）
print("\n用户：我叫什么？")
print("AI：", with_history.invoke({"question": "我叫什么？"}, cfg))
原理图
┌─────────────────────────────────────────────────────────┐
│ RunnableWithMessageHistory 工作流程 │
├─────────────────────────────────────────────────────────┤
│ │
│ 用户输入: "我叫什么？" │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ MessagesPlaceholder("history") │ ◄── 历史消息自动注入 │
│ └─────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ ChatPromptTemplate │ │
│ │ ├── system: 你是一个友好助理 │ │
│ │ ├── history: [张三说过的话...] │ │
│ │ └── human: 我叫什么？ │ │
│ └─────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ LLM (qwen-plus) │ │
│ └─────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ 输出: "你叫张三" │ │
│ └─────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ 自动保存到历史记录 │ │
│ └─────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
核心组件说明
| 组件 | 作用 |
|---|---|
MessagesPlaceholder("history") |
模板中的占位符，历史消息会自动填充到这里 |
input_messages_key |
指定用户当前输入在 prompt 中的变量名 |
history_messages_key |
指定历史消息在 prompt 中的变量名（与占位符名称一致） |
get_session_history |
函数，根据 session_id 返回对应的历史记录对象 |
1.4 持久化方案：RedisChatMessageHistory
核心思想
使用 Redis 作为持久化存储，实现跨进程、跨会话的记忆保存。
原理图
┌─────────────────────────────────────────────────────────┐ │ Redis 持久化记忆系统 │ ├─────────────────────────────────────────────────────────┤ │ │ │ ┌──────────┐ ┌─────────────┐ │ │ │ Client │ ──────► │ Redis │ │ │ └──────────┘ │ (持久化) │ │ │ └─────────────┘ │ │ ▲ │ │ │ │ │ ┌──────────────────────────────────┐ │ │ │ RedisChatMessageHistory │ │ │ │ • session_id: "user-001" │ │ │ │ • 自动从 Redis 读写消息 │ │ │ └──────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────┘
完整代码
from langchain.chat_models import init_chat_model
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableConfig
import os
import redis
from loguru import logger
from dotenv import load_dotenv
load_dotenv(encoding='utf-8')
# Redis 连接配置
REDIS_URL = "redis://localhost:26379"
redis_client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
# 初始化模型
llm = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 定义 Prompt
prompt = ChatPromptTemplate.from_messages([
MessagesPlaceholder("history"),
("human", "{question}")
])
def get_session_history(session_id: str) -> RedisChatMessageHistory:
"""获取或创建 Redis 中的会话历史"""
return RedisChatMessageHistory(
session_id=session_id,
url=REDIS_URL,
# ttl=3600 # 可选：设置过期时间（秒）
)
# 创建带历史的链
chain = RunnableWithMessageHistory(
prompt | llm,
get_session_history,
input_messages_key="question",
history_messages_key="history"
)
# 配置会话
config = RunnableConfig(configurable={"session_id": "user-001"})
# 对话循环
print("开始对话（输入 'quit' 退出）")
while True:
question = input("\n输入问题：")
if question.lower() in ['quit', 'exit', 'q']:
break
response = chain.invoke({"question": question}, config)
logger.info(f"AI回答: {response.content}")
redis_client.save() # 强制持久化到 dump.rdb
Redis 环境检查
文件：07_memory/RedisEnvCheck.py
import redis
try:
print(f"redis 版本: {redis.__version__}")
print("✅ redis 包导入成功！")
except ModuleNotFoundError:
print("❌ 未找到 redis 包，请先安装: pip install redis==5.3.1")
记忆系统对比总结
| 方案 | 存储 | 持久化 | 适用场景 |
|---|---|---|---|
| InMemoryChatMessageHistory | 内存 | ❌ | 开发调试、一次性会话 |
| RunnableWithMessageHistory + InMemory | 内存 | ❌ | 单进程多会话 |
| RedisChatMessageHistory | Redis | ✅ | 生产环境、分布式部署 |
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
19
0- 0
已为社区贡献24条内容
所有评论(0)