介绍
这两个组件是 LangChain 中实现对话记忆(上下文) 的核心:
InMemoryChatMessageHistory:内存级别的对话历史存储,将对话记录临时存在内存中,程序重启后丢失,适合测试/临时场景;RunnableWithMessageHistory:包装器,将普通的 Runnable 链(如 Prompt+LLM)包装为 带对话历史 的链,自动管理对话上下文的传入和存储。
用法
定义提示词模板类
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个温柔的小学数学老师,回答问题时用“亲”、“小朋友”,并且简短回答问题,回答问题不要用markdown格式"),
("placeholder", "{chat_history}"),
("human", "计算过程:{msg}")
])
提示词模板类和模型 链
model = ChatTongyi(model="qwen3-max")
chain = prompt | model
定义保存历史序列
chat_history = {}
用户对话ID
SESSION_ID = "user_小明" # 可理解为「用户ID」或「对话ID」
定义获取历史消息函数
根据用户对话ID获取历史消息
def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
"""
根据 session_id 获取对话历史:
- 若不存在,创建新的 InMemoryChatMessageHistory 实例
- 若存在,返回已有实例
"""
if session_id not in chat_history:
# 初始化内存对话历史容器
chat_history[session_id] = InMemoryChatMessageHistory()
return chat_history[session_id]
创建 RunnableWithMessageHistory 对象
chain_with_history = RunnableWithMessageHistory(
chain, # 提示词模板类、模型 链
get_session_history=get_session_history, # 告诉链如何获取对话历史
input_messages_key="msg", # 用户当前输入的变量名(对应Prompt中的{input})
history_messages_key="chat_history" # 对话历史的变量名(对应Prompt中的{chat_history})
)
对应关系如下图:
调用大模型
res = chain_with_history.stream({
"msg": question, # 传入问题
"chat_history": chat_history # 传入历史序列
},
# 关键:传入 config 指定 session_id,关联对话历史
config={
"configurable": {
"session_id": SESSION_ID
}
}
)
查看历史对话(可选)
print("\n===== 查看对话历史 =====")
history = get_session_history(SESSION_ID)
for msg in history.messages:
print(f"{msg.type.upper()}: {msg.content}")
print("\n===== 查看对话历史结束 =====\n\n\n")
传统方式
详见:LangChain教程:ChatPromptTemplate聊天大模型提示词模板类
使用新方式,可更加合理规范的方式管理历史消息
新方式-案例
from dotenv import load_dotenv
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_models.tongyi import ChatTongyi
# 加载环境变量(需配置 OPENAI_API_KEY)
load_dotenv()
# 1. 初始化 LLM 和 Prompt(关键:添加 MessagesPlaceholder 接收对话历史)
model = ChatTongyi(model="qwen3-max")
# Prompt 中通过 占位符接收对话历史
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个温柔的小学数学老师,回答问题时用“亲”、“小朋友”,并且简短回答问题,回答问题不要用markdown格式"),
("placeholder", "{chat_history}"),
("human", "计算过程:{msg}")
])
# 2. 构建基础链(Prompt + LLM)
chain = prompt | model
# 3. 定义「获取对话历史」的函数(给不同用户分配不同的历史容器)
# 用字典存储不同 session_id 对应的对话历史(模拟多用户)
chat_history = {}
def get_history(session_id: str) -> InMemoryChatMessageHistory:
"""
根据 session_id 获取对话历史:
- 若不存在,创建新的 InMemoryChatMessageHistory 实例
- 若存在,返回已有实例
"""
if session_id not in chat_history:
# 初始化内存对话历史容器
chat_history[session_id] = InMemoryChatMessageHistory()
return chat_history[session_id]
# 4. 用 RunnableWithMessageHistory 包装基础链
# 核心:关联「对话历史获取函数」和「历史变量名」
chain_his = RunnableWithMessageHistory(
chain,
get_session_history=get_history, # 告诉链如何获取对话历史
history_messages_key="chat_history", # 对话历史的变量名(对应Prompt中的{chat_history})
input_messages_key="msg", # 用户当前输入的变量名(对应Prompt中的{input})
)
# 用户对话ID(同一个 session_id 共享历史)
SESSION_ID = "user_小明" # 可理解为「用户ID」或「对话ID」
while True:
"""
第一次提问:3x+5=20,x 等于多少?
第二次提问:5x-8=72,x 等于多少?
第三次提问:把x=10代入上面公式,等于多少?
"""
question = input("请输入:")
# 调用stream向模型提问,返回 generator类型
res = chain_his.stream({
"msg": question, # 传入问题
"chat_history": chat_history # 传入历史序列
},
# 关键:传入 config 指定 session_id,关联对话历史
config={
"configurable": {
"session_id": SESSION_ID
}
}
)
print("\n----AI回答-------\n")
# 流式输出
for chunk in res:
print(chunk.content, end="", flush=True)
print("\n----AI回答结束-------\n")
# 查看内存中的对话历史(验证存储)
print("\n===== 查看对话历史 =====")
history = get_history(SESSION_ID)
for msg in history.messages:
print(f"{msg.type.upper()}: {msg.content}")
print("\n===== 查看对话历史结束 =====\n\n\n")
与传统方式对比
仅用 ChatPromptTemplate 只能 被动接收 你手动传入的历史,而搭配这两个组件能实现 自动管理 对话历史,大幅降低开发成本、提升代码健壮性。
| 特性 | 仅用 ChatPromptTemplate | InMemoryChatMessageHistory + RunnableWithMessageHistory |
|---|---|---|
| 历史管理方式 | 手动管理:需自己编写代码保存历史、拼接历史、注入到 Prompt 中,每一步都要手动控制 | 自动管理:链会自动读取对应用户的历史、注入到 Prompt、保存新的对话记录,无需手动干预 |
| 多用户隔离 | 需自己编写多用户历史隔离逻辑(如字典存储、session_id 映射),容易出错 | 内置多用户隔离:通过 session_id 自动为不同用户分配独立的历史容器,逻辑封装在底层 |
| 代码复杂度 | 高:需要手动处理「保存历史→读取历史→注入Prompt→更新历史」全流程 | 低:只需定义「获取历史的函数」,其余逻辑由 LangChain 封装,代码量减少 50%+ |
| 流式调用兼容 | 手动处理流式输出时,无法自动将流式回复保存到历史,需额外编写拼接逻辑 | 自动兼容流式调用:流式输出的每一段内容会自动整合并保存到历史中 |
| 扩展性 | 差:若要切换历史存储方式(如从内存到 Redis),需重写全部历史管理代码 | 好:只需修改 get_session_history 函数的返回值(如替换为 RedisChatMessageHistory),核心业务逻辑无需改动 |
