LangChain 消息管理与聊天历史存储
·
LangChain 系列文章目录
第一章 LangChain 简介和核心包；
第二章 LangChain 提示词工程；
第三章 LangChain 工作流；
第四章 LangChain服务部署与链路监控；
第五章 LangChain 消息管理与聊天历史存储
前言
本文主要整理 LangChain 消息管理与聊天历史存储相关内容，包括内存消息存储、会话唯一键配置、Redis 持久化、多种消息存储方式、消息裁剪以及历史对话总结记忆等。
LangChain消息管理与聊天历史存储
一、消息存储在内存
1.作用：将该历史对话通过session_id保存到内存中。
2.构建消息存储方法一
- 自建一个get_session_history()方法，用于将对话记录保存到自定义的store[session_id]中
- 输入参数为session_id，用于保存到指定会话中。
- 输出为抽象基类BaseChatMessageHistory的子类ChatMessageHistory实例，LangChain提供的保存对象。
- 创建带会话历史记录的Runnable（链条）
- 使用RunnableWithMessageHistory()方法创建一个可以输入历史记录的链条（Runnable）对象。
- 参数：
- runnable：(必需) 要包装的可运行对象，可以是Runnable接口的LangChain组件，也可以是链。
- get_session_history：(必需) 工厂函数，接收session_id并返回对应的聊天消息历史存储对象的函数名称
- 工厂函数：不直接实例化类，而是使用一个函数来生产对象
- input_messages_key： (可选) 指定用户消息在输入字典中的键名
- history_messages_key：(可选) 指定历史消息在输入字典中的键名
- 执行带会话历史记录的Runnable（链条）
- 使用invoke()执行链条。
- 参数
- input为输入的内容字典，字典的键为对应的提示词模板的模板变量
- config为配置项字典。
- 外层结构：包含一个configurable键
- 内层结构：session_id对话的唯一标识符。不同的session_id代表不同的会话，历史也会不一样。
代码
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai.chat_models import ChatOpenAI
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
# 可以尝试打印prompt，不过会内容会比较乱。非必要。
from langchain_core.callbacks import StdOutCallbackHandler
from langchain_core.messages import BaseMessage
# 1. 创建一个打印prompt的callback
class PrintPromptCallback(StdOutCallbackHandler):
def on_chat_model_start(self, serialized, messages, **kwargs):
# messages就是最终送进模型的完整prompt（list[BaseMessage]）
print("=== 第二次真正送进模型的 prompt ===")
for m in messages[0]: # messages是List[List[BaseMessage]]
print(f"{m.type}: {m.content}")
# 创建一个聊天提示词模版
prompt = ChatPromptTemplate.from_messages([
("system", "You're an assistant who's good at {ability}."),
MessagesPlaceholder(variable_name="history"),
("human", "{inputx}"),
])
model = ChatOpenAI(
model="Pro/deepseek-ai/DeepSeek-V3",
openai_api_key="************",
openai_api_base="https://api.siliconflow.cn/v1",
temperature=0.7,
)
runnable = prompt | model
# 用来存储会话历史记录
store = {}
# 定义一个获取会话历史的函数，入参是session_id，返回一个会话历史记录
def get_session_history(session_id: str) -> BaseChatMessageHistory:
if session_id not in store:
store[session_id] = ChatMessageHistory()
return store[session_id]
# 创建一个带会话历史记录的Runnable
with_message_history = RunnableWithMessageHistory(
runnable=runnable,
get_session_history=get_session_history,
history_messages_key="history",
)
# 调用带会话历史记录的Runnable
response = with_message_history.invoke(
input={"ability": "math", "inputx": "余弦是什么意思？"},
config={"configurable": {"session_id": "abc123"}},
)
print(response)
print("=======================")
# 相同session_id --> 记住了
response = with_message_history.invoke(
input={"ability": "math", "inputx": "什么?"},
config={
"configurable": {"session_id": "abc123"},
"callbacks": [PrintPromptCallback()] # 增加一个callback函数打印prompt
},
)
print(response)
print("=======================")
# 新session_id --> 不记得了
response = with_message_history.invoke(
{"ability": "math", "inputx": "什么?"},
config={"configurable": {"session_id": "def234"}},
)
print(response)
结果
content = "余弦是三角函数之一，表示邻边与斜边的比值。"
additional_kwargs = {"refusal": None}
response_metadata = {
"token_usage": {
"completion_tokens": 13,
"prompt_tokens": 23,
"total_tokens": 36,
"completion_tokens_details": None,
"prompt_tokens_details": None,
},
"model_name": "Pro/deepseek-ai/DeepSeek-V3",
"system_fingerprint": "",
"id": "0197065649d0cd9f34ea935878f474de",
"service_tier": None,
"finish_reason": "stop",
"logprobs": None,
}
id = "run--e4e6ee75-4d75-41c2-a095-c123a2a93b66-0"
usage_metadata = {
"input_tokens": 23,
"output_tokens": 13,
"total_tokens": 36,
"input_token_details": {},
"output_token_details": {},
}
=======================
...
=== 第二次真正送进模型的 prompt ===
system: You're an assistant who's good at math.
human: 余弦是什么意思？
ai: 余弦是三角函数之一，表示邻边与斜边的比值。
human: 什么?
...
content = "邻边比斜边，记作cos。"
additional_kwargs = {"refusal": None}
response_metadata = {
...
}
id = "run--4aee0646-9a23-4906-9971-8df9f8849d74-0"
usage_metadata = {
...
}
=======================
content = "有什么可以帮您？"
additional_kwargs = {"refusal": None}
response_metadata = {
"token_usage": {
"completion_tokens": 5,
"prompt_tokens": 22,
"total_tokens": 27,
"completion_tokens_details": None,
"prompt_tokens_details": None,
},
"model_name": "Pro/deepseek-ai/DeepSeek-V3",
"system_fingerprint": "",
"id": "0197065657872ab0cd92d3e280c480ec",
"service_tier": None,
"finish_reason": "stop",
"logprobs": None,
}
id = "run--b7965209-2a85-4b84-a890-0ca93618253d-0"
usage_metadata = {
"input_tokens": 22,
"output_tokens": 5,
"total_tokens": 27,
"input_token_details": {},
"output_token_details": {},
}
3.构建消息存储方法二
- 自建一个get_session_history()方法，用于将对话记录保存到自定义的store[session_id]中
- 输入参数为session_id，用于保存到指定会话中。
- 输出为InMemoryChatMessageHistory实例，LangChain提供的保存对象。
- InMemoryChatMessageHistory本质是ChatMessageHistory的旧写法
- 创建带会话历史记录的Runnable（链条）同上
- 因为get_session_history()方法返回的是InMemoryChatMessageHistory实例，所以无需指定history_messages_key历史消息在输入字典中的键名。所以MessagesPlaceholder的variable_name参数定义什么都可以。
- 执行带会话历史记录的Runnable（链条）同上
- 对于invoke()方法的input输入为一个HumanMessage对象时，input_key或history_key和MessagesPlaceholder的variable_name可以随便定义
代码
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import (
BaseChatMessageHistory,
InMemoryChatMessageHistory,
)
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
model="Pro/deepseek-ai/DeepSeek-V3",
# model="Qwen/Qwen2.5-VL-32B-Instruct",
openai_api_key="************",
openai_api_base="https://api.siliconflow.cn/v1",
temperature=0.7,
)
prompt = ChatPromptTemplate.from_messages([
("system", "你是个智能AI助手"),
MessagesPlaceholder(variable_name="messages"),
])
chain = prompt | model
# 存储会话
store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
if session_id not in store:
store[session_id] = InMemoryChatMessageHistory()
return store[session_id]
with_message_history = RunnableWithMessageHistory(
runnable=chain,
get_session_history=get_session_history,
)
# 用固定配置模拟同一会话
config = {"configurable": {"session_id": "abc2"}}
response = with_message_history.invoke(
input=HumanMessage(content="你好，我是张三。"),
config=config,
)
print(response)
print("=======================")
response2 = with_message_history.invoke(
input=HumanMessage(content="我叫什么名字？"),
config=config,
)
print(response2)
print("=======================")
print(response2.content)
结果
content = "你好，张三！很高兴认识你。我是你的智能助手，有什么可以帮你的吗？😊"
additional_kwargs = {"refusal": None}
response_metadata = {
"token_usage": {
"completion_tokens": 21,
"prompt_tokens": 13,
"total_tokens": 34,
"completion_tokens_details": None,
"prompt_tokens_details": None,
},
"model_name": "Pro/deepseek-ai/DeepSeek-V3",
"system_fingerprint": "",
"id": "01981289e155932d3167941a516d9fe3",
"service_tier": None,
"finish_reason": "stop",
"logprobs": None,
}
id = "run--7860cfb6-c4ca-4b54-9bca-72bb71c9e56b-0"
usage_metadata = {
"input_tokens": 13,
"output_tokens": 21,
"total_tokens": 34,
"input_token_details": {},
"output_token_details": {},
}
=======================
content = "你刚刚告诉我你的名字是**张三**哦！需要我帮你记住这个名字吗？或者有其他问题想聊？ 😄"
additional_kwargs = {"refusal": None}
response_metadata = {
"token_usage": {
"completion_tokens": 26,
"prompt_tokens": 41,
"total_tokens": 67,
"completion_tokens_details": None,
"prompt_tokens_details": None,
},
"model_name": "Pro/deepseek-ai/DeepSeek-V3",
"system_fingerprint": "",
"id": "01981289ea33521239e7037c4626328d",
"service_tier": None,
"finish_reason": "stop",
"logprobs": None,
}
id = "run--fa7b0be9-f01e-4021-805b-f8c72055a682-0"
usage_metadata = {
"input_tokens": 41,
"output_tokens": 26,
"total_tokens": 67,
"input_token_details": {},
"output_token_details": {},
}
=======================
你刚刚告诉我你的名字是**张三**哦！需要我帮你记住这个名字吗？或者有其他问题想聊？ 😄
二、配置会话唯一键
1.作用：将该历史对话通过user_id和conversation_id保存到内存中。
2.构建消息存储方法
- 构建一个get_session_history()方法，用于将对话记录保存到自定义的store[(user_id, conversation_id)]中
- 参数
- user_id，用于将记录保存到指定的用户id中。
- conversation_id：用于保存用户下的每次对话的id
- 返回为BaseChatMessageHistory实例，LangChain提供的基础保存对象。
3.创建一个带会话历史记录的Runnable（链条）
- 使用RunnableWithMessageHistory()方法，创建一个可以输入历史记录的链条（Runnable）对象。
- 参数：
- runnable：(必需) 要包装的基础可运行对象，可以是任何实现了 Runnable 接口的 LangChain 组件
- get_session_history：(必需) 工厂函数，接收 session_id 并返回对应的聊天消息历史存储对象的函数名称
- input_messages_key：(可选) 指定输入字典中用户消息的键名
- history_messages_key：(可选) 指定历史消息在输入字典中的键名
- history_factory_config：配置传递给聊天历史工厂（chat history factory）的字段参数。
- history_factory_config是一个ConfigurableFieldSpec对象的列表，用于定义需要传递给get_session_history函数的配置字段。
- 如果不指定history_factory_config，系统会使用默认的配置，即只包含一个session_id字段
- ConfigurableFieldSpec对象包含内容
- id：字段的唯一标识符，用于在配置中引用该字段。如：“user_id”、“conversation_id”
- annotation：字段的类型注解，用于指定该字段接受的数据类型。
- name：字段的显示名称，提供更友好的字段名称显示。如：“User ID”、“Conversation ID”
- description：字段的描述信息，用于解释该字段的用途。如：“用户的唯一标识符。”
- default：字段的默认值，当没有提供值时使用的默认值。如：“” 空字符串
- is_shared：字段是否共享，默认False
4.执行Runnable（链条）
- 使用invoke()执行链条。
- 参数
- input为输入的内容字典，字典的键为对应的提示词模板的模板变量
- config为配置项字典。
- 外层结构：包含一个configurable键
- 内层结构：user_id用户的唯一标识符，conversation_id对话的唯一标识符。键值对与之前定义history_factory_config中的ConfigurableFieldSpec配置相对应
- config作用
- 实际应用：
- 这些参数会被传递给get_session_history 函数
- 用于获取或创建特定用户和对话的聊天历史
- 在示例代码中，当user_id和conversation_id相同时（如"123"和"1"），会使用相同的聊天历史
- 当 conversation_id改变时（如改为 “2”），会创建新的聊天历史
- 作用效果：
- 从代码运行结果可以看到，当使用相同的user_id和conversation_id时，AI能够记住之前的对话内容
- 当改变conversation_id时，AI 就无法获取之前的对话内容，表现为不记得之前的对话
- 实际应用：
代码
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai.chat_models import ChatOpenAI
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables import ConfigurableFieldSpec
prompt = ChatPromptTemplate.from_messages([
("system", "You're an assistant who's good at {ability}."),
MessagesPlaceholder(variable_name="history"), # 历史消息占位符
("human", "{input}"),
])
model = ChatOpenAI(
model="Pro/deepseek-ai/DeepSeek-V3",
openai_api_key="************",
openai_api_base="https://api.siliconflow.cn/v1",
temperature=0.7,
)
runnable = prompt | model
store = {}
def get_session_history(user_id: str, conversation_id: str) -> BaseChatMessageHistory:
if (user_id, conversation_id) not in store:
store[(user_id, conversation_id)] = ChatMessageHistory()
return store[(user_id, conversation_id)]
with_message_history = RunnableWithMessageHistory(
runnable=runnable,
get_session_history=get_session_history,
input_messages_key="input",
history_messages_key="history",
history_factory_config=[
ConfigurableFieldSpec(
id="user_id",
annotation=str,
name="User ID",
description="用户的唯一标识符。",
default="",
is_shared=True,
),
ConfigurableFieldSpec(
id="conversation_id",
annotation=str,
name="Conversation ID",
description="对话的唯一标识符。",
default="",
is_shared=True,
),
],
)
response = with_message_history.invoke(
input={"ability": "math", "input": "余弦是什么意思？"},
config={"configurable": {"user_id": "123", "conversation_id": "1"}},
)
print(response)
# 记住
response = with_message_history.invoke(
input={"ability": "math", "input": "什么?"},
config={"configurable": {"user_id": "123", "conversation_id": "1"}},
)
print(response)
# 同一用户，不同对话，记不住
response = with_message_history.invoke(
input={"ability": "math", "input": "什么?"},
config={"configurable": {"user_id": "123", "conversation_id": "2"}},
)
print(response)
结果
content = "余弦是三角函数之一，表示邻边与斜边的比值。"
additional_kwargs = {"refusal": None}
response_metadata = {
...
}
id = "run--ff7be465-ba6b-4186-954d-6467862ef582-0"
usage_metadata = {
...
}
content = "余弦是直角三角形中邻边比斜边的比值。"
additional_kwargs = {"refusal": None}
response_metadata = {
...
}
id = "run--088ecf83-3e2d-4b5b-8cc9-203a069787cb-0"
usage_metadata = {
...
}
content = "啥?"
additional_kwargs = {"refusal": None}
response_metadata = {
...
}
id = "run--1565a8fd-0585-4024-a593-78a210f33207-0"
usage_metadata = {
...
}
三、消息持久化
1.作用：将该历史对话通过user_id保存到Redis中。
2.配置Redis环境
- 外部需要有redis服务。
- 内部需要安装redis的python操作库：pip install redis
3.构建消息存储方法
- 构建一个get_session_history()方法，用于将对话记录保存到自定义的store[session_id]中
- 输入参数session_id，用于指定需要保存的session_id。
- 返回为RedisChatMessageHistory(session_id, url=REDIS_URL)实例，LangChain提供的保存对象。
- url参数为Redis的地址字符串
4.创建一个带会话历史记录的Runnable（链条）
- 使用RunnableWithMessageHistory()方法创建一个可以输入历史记录的链条（Runnable）对象。
- 参数：
- runnable：(必需) 要包装的基础可运行对象，可以是任何实现了Runnable接口的LangChain组件
- get_session_history：(必需) 工厂函数，接收 session_id 并返回对应的聊天消息历史存储对象的函数名称
- input_messages_key：(可选) 指定输入字典中用户消息的键名
- history_messages_key：(可选) 指定历史消息在输入字典中的键名
5.执行Runnable（链条）
- 使用invoke()执行链条。
- 参数
- input为输入的内容字典，字典的键为对应的提示词模板的模板变量
- config为配置项字典。
- 外层结构：包含一个configurable键
- 内层结构：session_id对话的唯一标识符。不同的session_id代表不同的会话，历史也会不一样。
代码
# 引入redis聊天消息存储类
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai.chat_models import ChatOpenAI
prompt = ChatPromptTemplate.from_messages([
("system", "You're an assistant who's good at {ability}."),
MessagesPlaceholder(variable_name="history"), # 历史消息占位符
("human", "{input}"),
])
model = ChatOpenAI(
model="Pro/deepseek-ai/DeepSeek-V3",
openai_api_key="************",
openai_api_base="https://api.siliconflow.cn/v1",
temperature=0.7,
max_tokens=8000,
)
runnable = prompt | model
REDIS_URL = "redis://localhost:6379/0"
def get_message_history(session_id: str) -> RedisChatMessageHistory:
res = RedisChatMessageHistory(session_id, url=REDIS_URL)
return res
with_message_history = RunnableWithMessageHistory(
runnable=runnable,
get_session_history=get_message_history,
input_messages_key="input",
history_messages_key="history",
)
response = with_message_history.invoke(
input={"ability": "math", "input": "余弦是什么意思？"},
config={"configurable": {"session_id": "abc123"}},
)
print(response)
# 记住
response = with_message_history.invoke(
input={"ability": "math", "input": "什么?"},
config={"configurable": {"session_id": "abc123"}},
)
print(response)
# 新session_id --> 不记得了。
response = with_message_history.invoke(
input={"ability": "math", "input": "什么?"},
config={"configurable": {"session_id": "def234"}},
)
print(response)
结果
content = "余弦是三角函数之一，表示邻边与斜边的比值，记作cos。"
additional_kwargs = {"refusal": None}
response_metadata = {
...
}
id = "run--cf611be7-b56f-4451-890b-df2ef48a49cb-0"
usage_metadata = {
...
}
content = "余弦(cos)是三角函数，指直角三角形中邻边与斜边的比值。"
additional_kwargs = {"refusal": None}
response_metadata = {
...
}
id = "run--3fcc61b8-7ede-4923-8aa1-5b929222281a-0"
usage_metadata = {
...
}
content = "不清楚您的问题，请再详细说明。"
additional_kwargs = {"refusal": None}
response_metadata = {
...
}
id = "run--0ee42bbe-d6a1-4d53-8c4e-d3bfd8b70da6-0"
usage_metadata = {
...
}
6.Redis内有数据
7.简易使用user_id和conversation_id定义对话
- 其实就是使用user_id和conversation_id拼接成一个字符串，类似session_id输入到redis
- 在get_message_history()方法中增加user_id和conversation_id，并且将其拼接成一个字符串。
- 设置RunnableWithMessageHistory()方法的参数history_factory_config定义两个id：user_id和conversation_id
代码
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai.chat_models import ChatOpenAI
from langchain_core.runnables import ConfigurableFieldSpec
prompt = ChatPromptTemplate.from_messages([
("system", "You're an assistant who's good at {ability}."),
MessagesPlaceholder(variable_name="history"), # 历史消息占位符
("human", "{input}"),
])
model = ChatOpenAI(
model="doubao-seed-1-6-flash-250615",
openai_api_key="************",
openai_api_base="https://ark.cn-beijing.volces.com/api/v3",
)
runnable = prompt | model
REDIS_URL = "redis://localhost:6379/0"
def get_message_history(user_id: str, conversation_id: str) -> RedisChatMessageHistory:
all_id = user_id+"_"+conversation_id # 拼接字符串
res = RedisChatMessageHistory(all_id, url=REDIS_URL)
return res
with_message_history = RunnableWithMessageHistory(
runnable=runnable,
get_session_history=get_message_history,
input_messages_key="input",
history_messages_key="history",
history_factory_config=[
ConfigurableFieldSpec(
id="user_id",
annotation=str,
name="User ID",
description="用户的唯一标识符。",
default="",
is_shared=True,
),
ConfigurableFieldSpec(
id="conversation_id",
annotation=str,
name="Conversation ID",
description="对话的唯一标识符。",
default="",
is_shared=True,
),
],
)
response = with_message_history.invoke(
input={"ability": "math", "input": "那么正弦呢"},
config={"configurable": {"user_id": "abc123","conversation_id": "xxx123"}},
)
print(response)
# 记住
response = with_message_history.invoke(
input={"ability": "math", "input": "第二个问了什么问题?"},
config={"configurable": {"user_id": "abc123","conversation_id": "xxx123"}},
)
print(response)
8.使用user_id和conversation_id定义对话
- 其实就是使用user_id和conversation_id拼接成一个字符串，类似session_id输入到redis
- user_id和conversation_id使用:拼接时，redis会自动生成一个user_id文件夹来保存数据，用于区分不同用户。
- 最终redis的key为user_id:conversation_id
- 继承RedisChatMessageHistory类并修改：
- 修改session_id为user_id:conversation_id
- 修改add_message()用于设置最大保存对话记录数量
- 在get_message_history()方法中增加user_id和conversation_id参数
- 设置RunnableWithMessageHistory()方法的参数history_factory_config定义两个id：user_id和conversation_id
代码
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables import ConfigurableFieldSpec
from langchain_openai.chat_models import ChatOpenAI
# ----------------- 基础配置 -----------------
prompt = ChatPromptTemplate.from_messages([
("system", "You're an assistant who's good at {ability}."),
MessagesPlaceholder(variable_name="history"), # 历史消息占位符
("human", "{input}"),
])
model = ChatOpenAI(
model="Pro/deepseek-ai/DeepSeek-V3",
openai_api_key="************",
openai_api_base="https://api.siliconflow.cn/v1",
temperature=0.7,
max_tokens=8000,
)
# ---------- 1. 自定义 Redis 历史存储，带「最多 8 轮」限制 -------------
REDIS_URL = "redis://localhost:6379/0"
class RedisChatMessageHistoryWithKeys(RedisChatMessageHistory):
MAX_ROUNDS = 8
MAX_MSG = MAX_ROUNDS * 2 # 8 轮 = 16 条消息
def __init__(self, user_id: str, conversation_id: str, url: str = REDIS_URL):
super().__init__(session_id=f"{user_id}:{conversation_id}", url=url)
# 重写 add_message，在每次添加后做滑动窗口
def add_message(self, message):
super().add_message(message) # 先正常写入
while len(self.messages) > self.MAX_MSG: # 如果超过上限，从左侧弹出旧消息
self.messages.pop(0) # 移除最早的一条
self.redis_client.lpop(self.key) # 同步删除 Redis 里最左元素
# ----------------- 2. 获取会话历史 -----------------
def get_session_history(user_id: str, conversation_id: str):
return RedisChatMessageHistoryWithKeys(user_id, conversation_id, url=REDIS_URL)
# ----------------- 3. Runnable 包装 -----------------
with_message_history = RunnableWithMessageHistory(
runnable=prompt | model,
get_session_history=get_session_history,
input_messages_key="input",
history_messages_key="history",
history_factory_config=[
ConfigurableFieldSpec(
id="user_id", annotation=str, name="User ID", is_shared=True
),
ConfigurableFieldSpec(
id="conversation_id", annotation=str, name="Conversation ID", is_shared=True
),
],
)
# ----------------- 4. 简单测试：连续对话 10 轮 -----------------
for i in range(10):
resp = with_message_history.invoke(
{"ability": "math", "input": f"第 {i+1} 轮"},
config={"configurable": {"user_id": "u1", "conversation_id": "c1"}},
)
print(f"第 {i+1} 轮回复：{resp.content}")
redis结果，只保存有8轮对话
9.多种存储方式（举例说明，不详细说明）
- 文件存储：FileChatMessageHistory
- SQL存储：SQLChatMessageHistory
- MongoDB存储：MongoDBChatMessageHistory
四、裁剪消息
1.作用：LLM和聊天模型的上下文token有限，需要截取有效的长度。
2.没有截取历史消息的聊天模型链
- 使用ChatMessageHistory()的add_user_message()和add_ai_message()来增加用户和ai历史对话记录。记录会按循序增加下去。
- 使用ChatMessageHistory()的messages属性获取所有的记录
- 使用匿名方程lambda session_id: temp_chat_history来输入执行链的get_session_history属性
- 与自建一个get_session_history()方法的功能一样，用于将对话记录保存到自定义的store[session_id]中
代码
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
# 手动增加对话记录
temp_chat_history = ChatMessageHistory()
temp_chat_history.add_user_message("我叫Jack，你好")
temp_chat_history.add_ai_message("你好")
temp_chat_history.add_user_message("我今天心情挺开心")
temp_chat_history.add_ai_message("很好啊！")
temp_chat_history.add_user_message("我下午在打篮球")
temp_chat_history.add_ai_message("运动是一个不错选择")
print(temp_chat_history.messages)
prompt = ChatPromptTemplate.from_messages([
("system","你是一个乐于助人的助手。尽力回答所有问题。提供的聊天历史包括与您交谈的用户的事实。",),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
])
model = ChatOpenAI(
model="Pro/deepseek-ai/DeepSeek-V3",
openai_api_key="************",
openai_api_base="https://api.siliconflow.cn/v1",
)
chain = prompt | model
chain_with_message_history = RunnableWithMessageHistory(
runnable=chain,
get_session_history=lambda session_id: temp_chat_history,
input_messages_key="input",
history_messages_key="chat_history",
)
response = chain_with_message_history.invoke(
input={"input": "我今天心情如何?"},
config={"configurable": {"session_id": "unused"}},
)
print(response)
print(temp_chat_history.messages)
结果
[
HumanMessage(
content="我叫Jack，你好", additional_kwargs={}, response_metadata={}),
AIMessage(
content="你好", additional_kwargs={}, response_metadata={}),
HumanMessage(
content="我今天心情挺开心", additional_kwargs={}, response_metadata={}),
AIMessage(
content="很好啊！", additional_kwargs={}, response_metadata={}),
HumanMessage(
content="我下午在打篮球", additional_kwargs={}, response_metadata={}),
AIMessage(
content="运动是一个不错选择", additional_kwargs={}, response_metadata={}),
]
content = "根据之前的对话记录，你提到过今天心情挺开心，下午还去打了篮球呢！看来是个活力满满的日子~ 现在你的心情有变化吗？还是依然保 持好状态呢？ 🌞"
additional_kwargs = {"refusal": None}
response_metadata = {
...
}
id = "run--adef5c58-0be0-4df4-a556-5a0514f91bf2-0"
usage_metadata = {
...
}
[
HumanMessage(
content="我叫Jack，你好", additional_kwargs={}, response_metadata={}),
AIMessage(
content="你好", additional_kwargs={}, response_metadata={}),
HumanMessage(
content="我今天心情挺开心", additional_kwargs={}, response_metadata={}),
AIMessage(
content="很好啊！", additional_kwargs={}, response_metadata={}),
HumanMessage(
content="我下午在打篮球", additional_kwargs={}, response_metadata={}),
AIMessage(
content="运动是一个不错选择", additional_kwargs={}, response_metadata={}),
HumanMessage(
content="我今天心情如何?", additional_kwargs={}, response_metadata={}),
AIMessage(
content="根据之前的对话记录，你提到过今天心情挺开心，下午还去打了篮球呢！看来是个活力满满的日子~ 现在你的心情有变化吗？还是依然保持好状态呢？ 🌞",
additional_kwargs={"refusal": None},
response_metadata={
"token_usage": {
"completion_tokens": 43,
"prompt_tokens": 59,
"total_tokens": 102,
"completion_tokens_details": None,
"prompt_tokens_details": None,
},
"model_name": "Pro/deepseek-ai/DeepSeek-V3",
"system_fingerprint": "",
"id": "019707e82c23aa55d485540c671107cd",
"service_tier": None,
"finish_reason": "stop",
"logprobs": None,
},
id="run--adef5c58-0be0-4df4-a556-5a0514f91bf2-0",
usage_metadata={
"input_tokens": 59,
"output_tokens": 43,
"total_tokens": 102,
"input_token_details": {},
"output_token_details": {},
},
),
]
3.截取历史消息后的聊天模型链
- 创建一个trim_messages()方法，裁剪temp_chat_history中的消息。
- 方法必须有chain_input参数
- 使用temp_chat_history.clear()方法，清除所有的历史对话记录
- 使用temp_chat_history.add_message()方法，增加历史对话记录
- 返回True为有截取，返回False为无截取。
- 裁剪过程
- 先使用RunnablePassthrough.assign(messages_trimmed=trim_messages)将裁剪方法导入。
- 参数messages_trimmed为裁剪方法名称。
- 其实就是增加一个messages_trimmed的变量到链中，值为裁剪方法名称
- 再使用链，链接使用所有的历史对话记录
- 先使用RunnablePassthrough.assign(messages_trimmed=trim_messages)将裁剪方法导入。
- 问最近提及的问题，是能够回答
代码
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough
temp_chat_history = ChatMessageHistory()
temp_chat_history.add_user_message("我叫Jack，你好")
temp_chat_history.add_ai_message("你好")
temp_chat_history.add_user_message("我今天心情挺开心")
temp_chat_history.add_ai_message("很好啊！")
temp_chat_history.add_user_message("我下午在打篮球")
temp_chat_history.add_ai_message("运动是一个不错选择")
prompt = ChatPromptTemplate.from_messages([
("system","你是一个乐于助人的助手。尽力回答所有问题。提供的聊天历史包括与您交谈的用户的事实。",),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
])
model = ChatOpenAI(
model="Pro/deepseek-ai/DeepSeek-V3",
openai_api_key="************",
openai_api_base="https://api.siliconflow.cn/v1",
)
chain = prompt | model
def trim_messages(chain_input):
stored_messages = temp_chat_history.messages
if len(stored_messages) <= 2:
return False
temp_chat_history.clear()
for message in stored_messages[-2:]:
temp_chat_history.add_message(message)
return True
chain_with_message_history = RunnableWithMessageHistory(
runnable=chain,
get_session_history=lambda session_id: temp_chat_history,
input_messages_key="input",
history_messages_key="chat_history",
)
chain_with_trimming = (
RunnablePassthrough.assign(messages_trimmed=trim_messages)
| chain_with_message_history
)
response = chain_with_trimming.invoke(
{"input": "我今天下午在干嘛?"},
{"configurable": {"session_id": "unused"}},
)
print(response.content)
print(temp_chat_history.messages)
结果
根据之前的聊天记录，你提到过 **“我下午在打篮球”**。所以，你今天下午应该是在 **打篮球** 🏀。
如果之后有其他安排或者记错了，可以告诉我，我帮你记着！😊
[
HumanMessage(
content="我下午在打篮球", additional_kwargs={}, response_metadata={}),
AIMessage(
content="运动是一个不错选择", additional_kwargs={}, response_metadata={}),
HumanMessage(
content="我今天下午在干嘛?", additional_kwargs={}, response_metadata={}),
AIMessage(
content="根据之前的聊天记录，你提到过 **“我下午在打篮球”**。所以，你今天下午应该是在 **打篮球** 🏀。 \n\n如果之后有其他安排或者记错了，可以告诉我，我帮你记着！😊",
additional_kwargs={"refusal": None},
response_metadata={
"token_usage": {
"completion_tokens": 51,
"prompt_tokens": 42,
"total_tokens": 93,
"completion_tokens_details": None,
"prompt_tokens_details": None,
},
"model_name": "Pro/deepseek-ai/DeepSeek-V3",
"system_fingerprint": "",
"id": "019707f7c262a32b1a84a7c913ebb2e4",
"service_tier": None,
"finish_reason": "stop",
"logprobs": None,
},
id="run--2a3b9e51-b44c-4a23-8dce-047b1267502e-0",
usage_metadata={
"input_tokens": 42,
"output_tokens": 51,
"total_tokens": 93,
"input_token_details": {},
"output_token_details": {},
},
),
]
- 问最前提及的问题，无法知道答案
代码
...
response = chain_with_trimming.invoke(
{"input": "我叫什么名字?"},
{"configurable": {"session_id": "unused"}},
)
print(response.content)
结果
...
在你的对话历史中，没有提供过你的名字信息。作为AI助手，我无法知道你的真实姓名，除非你主动告诉我。如果你愿意分享，可以告诉我你的名字，我会记住并在之后 的对话中使用哦！
（根据当前对话记录，你只提到过“下午在打篮球”，没有提及任何姓名信息。）
五、总结记忆
1.作用：使用LLM来对历史对话记录做对话总结。目的减少使用token。
2.总结历史对话记录
- 创建一个summarize_messages()方法，总结temp_chat_history中的消息。
- 方法必须有chain_input参数
- 方法返回True为有总结，返回False为无总结。
- 方法内部其实就是一个LCEL链，主要操作就是总结所有的历史对话记录
- 过程：
- 先创建提示词模板，模板参数是所有的历史对话记录
- 创建链条，模型使用原始模型model
- 执行链条，输出总结内容
- 清理之前的所有记录，换成总结内容
- 总结过程
- 先使用RunnablePassthrough.assign(messages_trimmed=summarize_messages)将总结方法导入。
- 参数messages_trimmed为总结方法名称。
- 其实就是增加一个messages_trimmed的变量到链中，值为总结方法名称
- 再使用链，链接使用所有的历史对话记录
- 先使用RunnablePassthrough.assign(messages_trimmed=summarize_messages)将总结方法导入。
代码
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough
temp_chat_history = ChatMessageHistory()
temp_chat_history.add_user_message("我叫Jack，你好")
temp_chat_history.add_ai_message("你好")
temp_chat_history.add_user_message("我今天心情挺开心")
temp_chat_history.add_ai_message("很好啊！")
temp_chat_history.add_user_message("我下午在打篮球")
temp_chat_history.add_ai_message("运动是一个不错选择")
prompt = ChatPromptTemplate.from_messages([
("system","你是一个乐于助人的助手。尽力回答所有问题。提供的聊天历史包括与您交谈的用户的事实。",),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
])
model = ChatOpenAI(
model="Pro/deepseek-ai/DeepSeek-V3",
openai_api_key="************",
openai_api_base="https://api.siliconflow.cn/v1",
)
chain = prompt | model
chain_with_message_history = RunnableWithMessageHistory(
runnable=chain,
get_session_history=lambda session_id: temp_chat_history,
input_messages_key="input",
history_messages_key="chat_history",
)
def summarize_messages(chain_input):
stored_messages = temp_chat_history.messages
if len(stored_messages) == 0:
return False
summarization_prompt = ChatPromptTemplate.from_messages([
MessagesPlaceholder(variable_name="chat_history"),
("user","将上述聊天消息浓缩成一条摘要消息。尽可能包含具体细节。",),
])
summarization_chain = summarization_prompt | model
summary_message = summarization_chain.invoke(
{"chat_history": stored_messages}
)
temp_chat_history.clear()
temp_chat_history.add_message(summary_message) # 将总结内容以AI角色加到历史
return True
chain_with_summarization = (
RunnablePassthrough.assign(messages_summarized=summarize_messages)
| chain_with_message_history
)
response = chain_with_summarization.invoke(
{"input": "我的名字叫什么，下午在干嘛，心情如何"},
{"configurable": {"session_id": "unused"}},
)
print(response.content)
print("-" * 80)
print(temp_chat_history.messages)
结果
你的名字是Jack，下午你去打篮球了，心情非常愉快！运动确实是个很棒的选择，继续保持好状态哦！😊🏀
--------------------------------------------------------------------------------
[
AIMessage(
content="【聊天摘要】 \n用户Jack表示今天心情愉快，下午进行了打篮球的活动，认为运动是个不错的选择。",
additional_kwargs={"refusal": None},
response_metadata={
"token_usage": {
"completion_tokens": 23,
"prompt_tokens": 48,
"total_tokens": 71,
"completion_tokens_details": None,
"prompt_tokens_details": None,
},
"model_name": "Pro/deepseek-ai/DeepSeek-V3",
"system_fingerprint": "",
"id": "019832d716528a1e57a72f083a1466c4",
"service_tier": None,
"finish_reason": "stop",
"logprobs": None,
},
id="run--ed6ccf0d-bd3d-4aa6-840e-7b9e0a436991-0",
usage_metadata={
"input_tokens": 48,
"output_tokens": 23,
"total_tokens": 71,
"input_token_details": {},
"output_token_details": {},
},
),
HumanMessage(
content="我的名字叫什么，下午在干嘛，心情如何",
additional_kwargs={},
response_metadata={},
),
AIMessage(
content="你的名字是Jack，下午你去打篮球了，心情非常愉快！运动确实是个很棒的选择，继续保持好状态哦！ 😊🏀",
additional_kwargs={"refusal": None},
response_metadata={
"token_usage": {
"completion_tokens": 30,
"prompt_tokens": 58,
"total_tokens": 88,
"completion_tokens_details": None,
"prompt_tokens_details": None,
},
"model_name": "Pro/deepseek-ai/DeepSeek-V3",
"system_fingerprint": "",
"id": "019832d75408843db28af5fa11952722",
"service_tier": None,
"finish_reason": "stop",
"logprobs": None,
},
id="run--cc6a1ec8-7b3d-4193-86e7-e4186188c415-0",
usage_metadata={
"input_tokens": 58,
"output_tokens": 30,
"total_tokens": 88,
"input_token_details": {},
"output_token_details": {},
},
),
]
3.截取历史对话，部分总结历史
- 当历史记录大于10条时，会截取最旧的6条作为总结，剩余的不变。
- 汇总成一个背景，以system角色保存
- 会议对话形式来交互
代码
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough
from langchain_core.messages import SystemMessage
temp_chat_history = ChatMessageHistory()
prompt = ChatPromptTemplate.from_messages([
("system","你是一个专业的助手。尽力回答所有问题。提供的聊天历史包括与您交谈的用户的基本信息和对话内容。",),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
])
model = ChatOpenAI(
model="doubao-seed-1-6-flash-250615",
openai_api_key="************",
openai_api_base="https://ark.cn-beijing.volces.com/api/v3",
)
chain = prompt | model
chain_with_message_history = RunnableWithMessageHistory(
runnable=chain,
get_session_history=lambda session_id: temp_chat_history,
input_messages_key="input",
history_messages_key="chat_history",
)
def summarize_messages(chain_input):
stored_messages = temp_chat_history.messages
if len(stored_messages) == 0:
return False
summarization_prompt = ChatPromptTemplate.from_messages([
MessagesPlaceholder(variable_name="chat_history"),
("user","""请将上述聊天历史总结成一条结构化的背景消息，必须包含以下信息：
【用户基本信息】（必须保留）：
- 姓名、身份、职业等个人信息
- 重要的个人偏好和特征
【关键事件和时间】（重要）：
- 提到的具体时间、地点
- 重要的事件和活动
【对话要点】（简要）：
- 主要讨论的话题
- 用户的需求和问题
请确保用户的姓名等基本信息在每次汇总中都被明确保留，即使看起来不重要也要包含。
总结长度控制在150字以内。""",),
])
summarization_chain = summarization_prompt | model
msg_long = 10
msg_cut = -4
stored_messages = temp_chat_history.messages
if len(stored_messages) < msg_long: # 小于10条不处理
return False
# 保留最近的4条消息，总结其余的
recent_messages = stored_messages[msg_cut:]
messages_to_summarize = stored_messages[:msg_cut]
# 总结较早的消息
summarization_chain = summarization_prompt | model
summary_message = summarization_chain.invoke(
{"chat_history": messages_to_summarize}
)
# 清空并重新构建历史
temp_chat_history.clear()
summary_system_msg = SystemMessage(
content=summary_message.content
)
temp_chat_history.add_message(summary_system_msg)
for msg in recent_messages:
temp_chat_history.add_message(msg)
return True
chain_with_summarization = (
RunnablePassthrough.assign(messages_summarized=summarize_messages)
| chain_with_message_history
)
def main():
while True:
try:
user_input = input("\n用户: ").strip()
if user_input.lower() in ['quit', 'exit', '退出', 'q']:
print("再见！")
break
if not user_input:
continue
# 调用AI模型
response = chain_with_summarization.stream(
{"input": user_input},
{"configurable": {"session_id": "unused"}},
)
print(f"\nAI: ", end="", flush=True)
for chunk in response:
print(chunk.content, end="", flush=True)
# print(f"\n[当前对话历史: {len(temp_chat_history.messages)} 条消息]")
# print(temp_chat_history.messages)
except KeyboardInterrupt:
print("\n\n程序被用户中断，再见！")
break
except Exception as e:
print(f"\n发生错误: {e}")
print("请重试...")
if __name__ == "__main__":
main()
总结
本文整理了 LangChain 中消息管理与聊天历史存储的常见做法，包括使用 RunnableWithMessageHistory 管理内存会话、通过 session_id / user_id / conversation_id 区分不同会话、使用 Redis 实现持久化保存，以及通过消息裁剪和总结记忆减少上下文 token 消耗。实际项目中，可以根据业务规模选择内存、Redis、文件、SQL 或 MongoDB 等不同存储方式，并结合裁剪或总结策略控制历史消息长度。
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
12
0- 0
已为社区贡献8条内容
所有评论(0)