记忆与对话历史:让"金鱼记忆"的模型记住你说过什么
Day15 的 Agent 能干活了,但它每次被调用都是一张白纸——LLM 本身不记得上一轮说过什么。今天看 LangChain 的经典解法:chat_history.py 里的 BaseChatMessageHistory(一本可读可写的"记事本")+ runnables/history.py 里的 RunnableWithMessageHistory(一个包装器:调用前把记事本内容塞进输入,调用后把新对话补写回去)。弄懂它,你也就明白了所有"多轮对话记忆"的通用套路。
BaseChatMessageHistory 就是病历本(能翻看 messages、能补写 add_messages),RunnableWithMessageHistory 就是诊室门口的护士:你报出病历号(session_id),她先去档案室取出你的病历、连同你今天的主诉一起递给医生;医生看完,她再把"今日问诊记录"补写进病历、放回档案室。类比二:get_session_history 这个参数像档案室的取档规则——护士不关心病历存在铁皮柜(内存 dict)还是云端系统(Redis/数据库),她只管"拿号换本子",存哪儿由你自定义。痛点:模型是"金鱼记忆",历史谁来管
invoke 都是独立的 HTTP 请求,模型没有任何跨请求状态。要多轮对话,就得每次把之前的消息一起发过去。可如果让业务代码自己干,就得到处写"读历史 → 拼消息 → 调模型 → 存历史"的样板代码,还要处理多用户隔离(张三的历史不能混进李四的对话)。BaseChatMessageHistory(存储接口,你可以实现成内存/文件/Redis);②什么时候读、什么时候写——RunnableWithMessageHistory(时机包装器:invoke 前读、invoke 后写)。业务链本身完全不知道"记忆"这回事——它还是那条 prompt | model,只是被护士"包了一层"。这就是装饰器模式在 LCEL 里的用法。BaseChatMessageHistory:一本可读可写的记事本
先看存储接口 BaseChatMessageHistory(libs/core/langchain_core/chat_history.py:22)。接口非常小:一个"读"(messages 属性)、一个"写"(add_messages)、一个"清空"(clear):
# libs/core/langchain_core/chat_history.py:169
def add_messages(self, messages: Sequence[BaseMessage]) -> None:
"""Add a list of messages. ..."""
for message in messages:
self.add_message(message) # 默认逐条写;子类可覆写成"批量写"更高效
# libs/core/langchain_core/chat_history.py:202
class InMemoryChatMessageHistory(BaseChatMessageHistory, BaseModel):
"""In memory implementation of chat message history."""
messages: list[BaseMessage] = Field(default_factory=list) # ★存储就是一个 list
def add_message(self, message: BaseMessage) -> None:
self.messages.append(message) # 写 = append
def clear(self) -> None:
self.messages = [] # 清空 = 换个空 list
messages 属性"读病历"。基类注释明确说:读取可能涉及底层存储 IO(chat_history.py:92),所以是"预期有延迟"的操作——Redis 实现就是在这里发 GET。add_messages(批量)★官方推荐用这个批量接口而不是单条的 add_message/add_user_message——文档原话是"避免对持久层的多次往返"(chat_history.py:44-48)。一轮对话产生"人类消息+AI 回复"两条,批量一次写完。aget_messages/aadd_messages每个方法都有 async 双胞胎,默认实现是 run_in_executor 把同步版扔进线程池(chat_history.py:99-110)。子类可以覆写成真异步。InMemoryChatMessageHistory官方自带的"铁皮柜"实现:存储就是内存 list。开发调试用它,生产换 Redis/数据库实现——接口不变,护士(L03)完全无感。RunnableWithMessageHistory:护士是怎么"包"住链的
主角登场:RunnableWithMessageHistory(libs/core/langchain_core/runnables/history.py:39)。它的 __init__(history.py:250)把你的链组装成一条"带前后钩子"的新链:
# libs/core/langchain_core/runnables/history.py:332(__init__ 内部,节选)
history_chain: Runnable[Any, Any] = RunnableLambda(
self._enter_history, self._aenter_history # ① 前置步骤:进场读历史
).with_config(run_name="load_history")
messages_key = history_messages_key or input_messages_key
if messages_key:
history_chain = RunnablePassthrough.assign( # ② 读到的历史塞进输入 dict 的指定 key
**{messages_key: history_chain}
).with_config(run_name="insert_history")
runnable_sync = runnable.with_listeners(on_end=self._exit_history) # ③ 后置钩子:结束时写历史
runnable_async = runnable.with_alisteners(on_end=self._aexit_history)
bound = (
history_chain # ④ 新链 = 读历史 | 你的链(带写历史钩子)
| RunnableLambda(_call_runnable_sync, _call_runnable_async)
.with_config(run_name="check_sync_or_async")
).with_config(run_name="RunnableWithMessageHistory")
RunnableLambda(_enter_history)把"读历史"本身做成一个 Runnable 步骤(Day03 学的 LCEL 积木思想:一切皆 Runnable)。L05 看它内部。RunnablePassthrough.assign(...)★Day18 会细讲的积木在这里实战了:输入 dict 原样透传,只把 history 这个 key 换成"读出来的历史消息"。这样 prompt 里的 MessagesPlaceholder("history") 就有料可填。with_listeners(on_end=...)★"写历史"不是拼在链尾的步骤,而是挂在你的链上的结束监听器——链跑完(拿到 Run 记录)自动触发 _exit_history。好处:它拿到的是完整的输入/输出记录,且不改变链的输出类型。bound = ... | ...最终产物 bound 就是"读历史 → 原链"两截管道。RunnableWithMessageHistory 继承自 RunnableBindingBase,对外表现得和原链一模一样——这就是"包一层"的技术含义。RunnableLambda、RunnablePassthrough.assign、with_listeners、| 管道)拼出来的。框架的高级功能,是用框架自己的基础能力搭的——这也是读源码最爽的时刻:一切都似曾相识。session_id 怎么变成"你那本"历史
调用时你传 config={"configurable": {"session_id": "foo"}},护士怎么据此取到"foo 的病历"?答案在 _merge_configs(libs/core/langchain_core/runnables/history.py:581)——它在每次调用合并 config 时被触发:
# libs/core/langchain_core/runnables/history.py:581
def _merge_configs(self, *configs: RunnableConfig | None) -> RunnableConfig:
config = super()._merge_configs(*configs)
expected_keys = [field_spec.id for field_spec in self.history_factory_config]
configurable = config.get("configurable", {})
missing_keys = set(expected_keys) - set(configurable.keys())
if missing_keys and parameter_names:
msg = (f"Missing keys {sorted(missing_keys)} in config['configurable'] ...")
raise ValueError(msg) # 没带 session_id?直接报错教你怎么传
if len(expected_keys) == 1:
message_history = self.get_session_history( # ★单 key:位置参数调工厂
configurable[expected_keys[0]] # get_session_history("foo")
)
else:
message_history = self.get_session_history( # 多 key:按名字传
**{key: configurable[key] for key in expected_keys} # (user_id=..., conversation_id=...)
)
config["configurable"]["message_history"] = message_history # ★取到的"病历本"塞回 config
return config
history_factory_config默认就一个字段 session_id(history.py:362-371);也可以自定义成 user_id + conversation_id 两把钥匙(类 docstring 里就有完整示例)。get_session_history("foo")★这就是你构造时传入的那个工厂函数——"档案室取档规则"。内存版通常是 store.setdefault(session_id, InMemoryChatMessageHistory()):没有这本就新建一本。塞回 config取到的 message_history 对象被放进 config["configurable"]["message_history"]。config 会沿管道传给每个步骤(Day04 学过),所以 L05 的进场/退场钩子都能从 config 里摸到同一本病历。缺 key 报错忘传 session_id 时的报错信息里直接附了正确写法示例——防呆做得很贴心,也是新手最常撞的错。session_id(病历号)进来 → 工厂函数换出病历本 → 病历本挂在 config 上随管道流动。多用户隔离就这么实现的:不同号取不同本,谁也不串。进场读历史 / 退场写历史
最后看两个钩子本体。进场 _enter_history(libs/core/langchain_core/runnables/history.py:521)+ 退场 _exit_history(history.py:547):
# libs/core/langchain_core/runnables/history.py:521
def _enter_history(self, value: Any, config: RunnableConfig) -> list[BaseMessage]:
hist: BaseChatMessageHistory = config["configurable"]["message_history"] # 从 config 摸出病历本
messages = hist.messages.copy() # ① 翻出全部历史
if not self.history_messages_key:
input_val = (value if not self.input_messages_key
else value[self.input_messages_key])
messages += self._get_input_messages(input_val) # ② 没单独 history key 时,历史+本轮输入拼一起
return messages
# libs/core/langchain_core/runnables/history.py:547
def _exit_history(self, run: Run, config: RunnableConfig) -> None:
hist: BaseChatMessageHistory = config["configurable"]["message_history"]
inputs = load(run.inputs, allowed_objects="messages")
input_messages = self._get_input_messages(inputs) # ③ 从 Run 记录还原本轮输入
if not self.history_messages_key:
historic_messages = config["configurable"]["message_history"].messages
input_messages = input_messages[len(historic_messages):] # ④ ★去掉开头的历史,防止重复入库
output_val = load(run.outputs, allowed_objects="messages")
output_messages = self._get_output_messages(output_val)
hist.add_messages(input_messages + output_messages) # ⑤ 本轮"问+答"批量补写进病历
hist.messages.copy()进场第一件事:把病历整本翻出来(copy 防止后面误改)。这里就是 L02 说的"可能有 IO 延迟"的读。_get_input_messages宽容的输入归一化(history.py:452):字符串 → 包成 HumanMessage;单条消息 → 包成 list;dict → 按 key 取。所以你传 "你好" 或 {"question": "你好"} 都行。run: Run★退场钩子的入参是 Run——Day19 要讲的"追踪记录"。with_listeners 底层靠回调系统拿到本次运行的完整输入输出,记忆功能其实是搭在回调/追踪基础设施上的(明天 Day17 正好讲它)。[len(historic_messages):]★防重灵魂切片:不设 history_messages_key 时,进场已把"历史+本轮"拼成一串喂给链,Run 记录的输入里就包含历史;写回前必须把开头的历史裁掉,否则病历每轮翻倍膨胀。add_messages(输入+输出)批量写:本轮的人类消息 + AI 回复一次写入——正是 L02 强调的"用批量接口省 IO 往返"。_enter_history 是整本全塞——聊 200 轮就塞 400+ 条消息,迟早超过模型上下文窗口。源码这层不做裁剪,需要你自己在链里加 trim_messages(Day06 讲过消息工具)或换带窗口/摘要策略的 History 实现。"记得多"和"塞得下"是一对永恒矛盾。全景图 + 一次对话的真实值
prompt(含 MessagesPlaceholder("history")) | ChatAnthropic,包装:RunnableWithMessageHistory(chain, get_by_session_id, input_messages_key="question", history_messages_key="history")。第 1 轮:
invoke({"ability": "math", "question": "What does cosine mean?"}, config={"configurable": {"session_id": "foo"}}) → 工厂新建 store["foo"](空本)→ history 占位填入 [] → 模型解释余弦 → 退场写入 2 条,store["foo"].messages == [HumanMessage("What does cosine mean?"), AIMessage("Cosine is ...")]。第 2 轮:
invoke({"ability": "math", "question": "What's its inverse"}, config=同上) → 进场翻出那 2 条填进 history → 模型看得见上文,知道 "its" 指余弦 → 答反余弦 → 病历变 4 条。换成 session_id: "bar" 则从空本开始——互不干扰。👶 小白:input_messages_key、history_messages_key、output_messages_key 三个 key 晕了,到底啥时候要传?
👨🏫 老师:看你的链吃什么、吐什么。链吃 dict(如 {"question": ...})→ 要告诉护士"本轮输入在哪个 key",传 input_messages_key="question";prompt 里有单独的历史占位 → 传 history_messages_key="history",护士就把历史塞那个 key、而不是和输入混拼;链吐 dict → 传 output_messages_key 告诉护士去哪捞回复。如果链直接吃消息列表、吐消息,三个都不用传。判断依据全在 _get_input_messages/_get_output_messages(history.py:452/490)那几个 if 里。
它被标记废弃了?+ 今日小结
读 __init__ 时你可能注意到开头有段醒目的代码(libs/core/langchain_core/runnables/history.py:324):
# libs/core/langchain_core/runnables/history.py:324
warn_deprecated(
since="1.3.3",
message=(
"RunnableWithMessageHistory is deprecated. "
"Use LangGraph's built-in persistence instead." # 官方指路:用 LangGraph 的持久化
),
removal="2.0.0",
)
thread_id 持久化,是记忆的超集。但今天没白学:①存量项目里它仍大量存在;②BaseChatMessageHistory 抽象没废,各种数据库集成还在用;③"进场读 / 退场写 / 按 key 隔离"这套思路,正是 LangGraph checkpointer 在更大尺度上重演的东西。Day20 收官会再讲这层生态关系。👶 小白:那"记忆"就只有"存聊天记录"这一种吗?
👨🏫 老师:这只是最基础的"逐字病历"。进阶玩法都建立在同一接口上:窗口记忆(只回填最近 N 条)、摘要记忆(旧对话让 LLM 压缩成一段摘要)、向量记忆(历史入向量库,按当前问题检索相关片段——其实就是把 Day09-12 的 RAG 用在自己的对话上)。接口统一的好处:这些策略都能无缝替换。
🧠 今天你应该能回答
- LLM 为什么"没记性"?(每次调用无跨请求状态,历史必须随请求重发)
- 记忆被拆成哪两个正交抽象?(存储
BaseChatMessageHistory+ 时机RunnableWithMessageHistory) session_id怎么变成具体那本历史?(_merge_configs调get_session_history工厂,结果挂到 config 上流动)- 历史什么时候读、什么时候写?(
_enter_history前置步骤读;with_listeners(on_end=_exit_history)结束后写) - 为什么写回前要切片
[len(historic):]?(防止把已在历史里的消息重复入库) - 官方现在推荐用什么做记忆?(LangGraph 内置持久化/checkpointer,本类已标记 2.0 移除)
✋ 10 分钟动手
cd /Users/bitmart/work/codes/github/AI_WORK/langchain/libs/core/langchain_core
# 1. 记事本抽象:读/写/清空 + 内存实现
sed -n '22,50p' chat_history.py # BaseChatMessageHistory 与实现指南
sed -n '202,247p' chat_history.py # InMemoryChatMessageHistory
# 2. 护士的组装与钩子
sed -n '324,357p' runnables/history.py # 废弃警告 + history_chain 组装
sed -n '521,562p' runnables/history.py # _enter_history / _exit_history
sed -n '581,626p' runnables/history.py # _merge_configs:session_id → 病历本
# 3. 输入归一化的宽容度
sed -n '452,489p' runnables/history.py # _get_input_messages
_exit_history 靠 with_listeners 拿到 Run 记录——这背后是 LangChain 的神经系统:回调(Callbacks)。明天拆 CallbackManager 怎么在每个环节广播事件、stream/astream_events 的流式输出又是怎么从回调里"长"出来的。打字机效果的真相就在那里。