RunnableWithMessageHistory(弃用)

弃用(langchain-core ≥ 1.3.3,计划 2.0 移除):新项目用 LangGraph checkpointer(thread_id)持久化多轮;可选 Store 做跨 thread 记忆。本篇只解释旧包装器如何把 history 挂到 LCEL 上,便于读存量代码。

有了 ChatMessageHistory,还要在每次 chain.invoke 时手动 merge 历史、写回 AI 回复,样板代码重复。RunnableWithMessageHistory 包装任意 Runnable:自动从 session 读历史、拼输入、把新 messages 写回,API 层只需传 session_id

段末注释RunnableWithMessageHistory = 为 Runnable 自动挂载按 session 读写消息历史的包装器。


1. 一句话定位

维度 内容
角色 能力层 多轮链:session 感知 invoke
输入 → 输出 session_id 的 dict + 用户输入 → 链输出,并更新 history
典型调用入口 RunnableWithMessageHistory(chain, get_session_history, ...)
与 LangGraph 复杂记忆(摘要、跨 session)用 LangGraph store;简单聊天链用本篇

2. 实现逻辑

1
2
3
4
5
6
7
8
9
10
11
1. base_chain = prompt | model(输入含 messages 或 history placeholder)
2. def get_session_history(session_id): return store[session_id]
3. chain = RunnableWithMessageHistory(
base_chain,
get_session_history,
input_messages_key="input",
history_messages_key="history",
)
4. chain.invoke({"input": "你好"}, config={"configurable": {"session_id": "s1"}})
5. 内部:取 s1 的 history → 拼 messages → invoke → add AI 到 history
6. 同 session_id 第二次 invoke 可见上一轮

字段级变形

1
2
3
4
session s1 history: [HumanMessage("hi"), AIMessage("hello")]
+ input "天气"
→ 拼入 prompt
→ 新 AIMessage 追加到 s1 history

3. 原理说明

3.1 构造参数

RunnableWithMessageHistory(类,langchain_core.runnables.history
功能:在 invoke 前后自动读写 ChatMessageHistory,把多轮拼进 base chain。

参数 类型 默认值 最小维度
runnable Runnable 必填 须能吃 messages 或含 Placeholder 的 dict
get_session_history callable 必填 (session_id: str) -> BaseChatMessageHistory
input_messages_key str / None None 输入为 dict 时必填,如 "input"
history_messages_key str / None None MessagesPlaceholder 名一致
output_messages_key str / None None 链输出为 dict 时才需要
1
2
3
4
5
6
RunnableWithMessageHistory(
prompt | model,
get_session_history,
input_messages_key="input",
history_messages_key="history",
)

3.2 configurable session_id

session_id(config 键)
功能:隔离会话。路径:config["configurable"]["session_id"]
默认值:无,每次 invoke 必传
最小维度:非空 str;多租户可用 f"{tenant}:{user}"

1
chain.invoke({"input": "你好"}, config={"configurable": {"session_id": "u1"}})

3.3 get_session_history 工厂

get_session_history(session_id: str)(你提供的函数)
功能:按 id 返回同一 ChatMessageHistory 实例(或从 DB 加载)。
最小实现:内存 dict,未见过的 id 新建 InMemoryChatMessageHistory()

1
2
3
4
5
store = {}
def get_session_history(session_id: str):
"""输入会话键,输出该键对应的 ChatMessageHistory(没有则新建)。"""
store.setdefault(session_id, InMemoryChatMessageHistory())
return store[session_id]

3.4 与 create_agent

Agent 图自带 messages state;本包装更适合 无图prompt | model 多轮。两轮对话后 history 至少 4 条(human+ai ×2)。


4. 最小可运行示例

1
pip install -U langchain-openai
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from langchain_openai import ChatOpenAI
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

store = {}

def get_session_history(session_id: str):
"""按 session_id 返回同一份内存历史。输入为会话键,输出为 ChatMessageHistory。"""
if session_id not in store:
store[session_id] = InMemoryChatMessageHistory()
return store[session_id]

model = ChatOpenAI(
model="qwen3.5:9b",
api_key="ollama",
base_url="http://localhost:11434/v1",
temperature=0,
)
prompt = ChatPromptTemplate.from_messages([
("system", "根据对话历史回答。若用户问自己说过什么,原样复述。"),
MessagesPlaceholder("history"),
("human", "{input}"),
])
chain = RunnableWithMessageHistory(
prompt | model,
get_session_history,
input_messages_key="input",
history_messages_key="history",
)
cfg = {"configurable": {"session_id": "u1"}}

chain.invoke({"input": "请记住:我的工号是 A42。"}, config=cfg)
out = chain.invoke({"input": "我的工号是什么?"}, config=cfg)
print(out.content)
print(len(get_session_history("u1").messages))
# 预期形态:回复含 A42;history 至少 4 条(两轮 human+ai)

重要配置参数

参数(API 名) 类型 / 默认值 功能说明 作用与影响 参考起点 配置指导
session_id config["configurable"] 内 str,必传 告诉包装器读写哪一份 ChatMessageHistory 不传 KeyError 或落到空 session UUID 勿用可猜测顺序 id
input_messages_key str / None invoke 的 dict 里哪一个键是「本轮用户输入」 与 invoke 键不一致则读不到本轮话 "input" 输入为 dict 时必填
history_messages_key str / None 把历史注入 base chain 的哪个占位符 必须等于 MessagesPlaceholder 名,否则历史进不了模型 "history" 与模板变量逐字相同
get_session_history callable,必填 (session_id) -> 同一 ChatMessageHistory 实例 每次 new 则无法累积历史 返回 BaseChatMessageHistory 必须 per-id 单例或从 DB 加载
output_messages_key str,可选 链输出为 dict 时从哪个键取 AIMessage 写回历史 链直接出 AIMessage 可不设 "output" 输出非 AIMessage 时才调
历史截断 应用层,在工厂内 filter 加载后只保留最近若干条再给模型 控 token;截太狠丢指代 最近 10 条 在 get 后 filter

5. 易踩坑

  1. 忘记传 config.configurable.session_id:报 KeyError 或使用默认空 session。
  2. prompt 无 MessagesPlaceholder 却设 history_messages_key:历史从未进入模型。
  3. get_session_history 每次 new 新 InMemory:同 id 无法累积历史。

小结

  • RunnableWithMessageHistory(弃用)把 session_id → history 挂到链上;新项目用 LangGraph checkpointer。
  • MessagesPlaceholder + ChatPromptTemplate 配合。
  • session_id 走 RunnableConfig.configurable
  • 生产用持久化 history 后端,勿 InMemory 多进程。

参考链接

-------------本文结束感谢您的阅读-------------