LangChain教程:InMemoryChatMessageHistory和RunnableWithMessageHistory临时记忆

介绍

这两个组件是 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),核心业务逻辑无需改动

原文出处:/show_1GW2r1dAVcdC.html