Day 16 / 共 20 天 · 阶段4 工具与 Agent

记忆与对话历史:让"金鱼记忆"的模型记住你说过什么

Day15 的 Agent 能干活了,但它每次被调用都是一张白纸——LLM 本身不记得上一轮说过什么。今天看 LangChain 的经典解法:chat_history.py 里的 BaseChatMessageHistory(一本可读可写的"记事本")+ runnables/history.py 里的 RunnableWithMessageHistory(一个包装器:调用前把记事本内容塞进输入,调用后把新对话补写回去)。弄懂它,你也就明白了所有"多轮对话记忆"的通用套路。

📍 你在 20 天里的位置(阶段4:工具与 Agent · D13-16)
S1 全景/LCEL S2 模型/消息 S3 数据/RAG D13 Tool D14 工具调用 D15 Agent D16 记忆/历史 D17 回调/流式 S5 收官
💡 先用两个类比兜住今天 类比一:LLM 像一位接诊量巨大的医生——他医术高明,但根本记不住每个病人,全靠病历本BaseChatMessageHistory 就是病历本(能翻看 messages、能补写 add_messages),RunnableWithMessageHistory 就是诊室门口的护士:你报出病历号(session_id),她先去档案室取出你的病历、连同你今天的主诉一起递给医生;医生看完,她再把"今日问诊记录"补写进病历、放回档案室。类比二:get_session_history 这个参数像档案室的取档规则——护士不关心病历存在铁皮柜(内存 dict)还是云端系统(Redis/数据库),她只管"拿号换本子",存哪儿由你自定义。
L01

痛点:模型是"金鱼记忆",历史谁来管

🤔 痛点你问模型"什么是余弦?",它答得头头是道;你接着问"那它的反函数呢?"——它一脸茫然:"什么的反函数?"。因为每次 invoke 都是独立的 HTTP 请求,模型没有任何跨请求状态。要多轮对话,就得每次把之前的消息一起发过去。可如果让业务代码自己干,就得到处写"读历史 → 拼消息 → 调模型 → 存历史"的样板代码,还要处理多用户隔离(张三的历史不能混进李四的对话)。
💡 本质:读写历史是"横切关注点",用包装器切出去LangChain 的做法是把"记忆"拆成两个正交的抽象:①存哪儿、怎么存——BaseChatMessageHistory(存储接口,你可以实现成内存/文件/Redis);②什么时候读、什么时候写——RunnableWithMessageHistory(时机包装器:invoke 前读、invoke 后写)。业务链本身完全不知道"记忆"这回事——它还是那条 prompt | model,只是被护士"包了一层"。这就是装饰器模式在 LCEL 里的用法。
L02

BaseChatMessageHistory:一本可读可写的记事本

先看存储接口 BaseChatMessageHistorylibs/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)完全无感。
大白话这个接口小到不能再小:能翻看、能补写、能撕掉,就是一本记事本。它故意不管"记事本按什么规则发给谁"——那是下一节护士的活。
L03

RunnableWithMessageHistory:护士是怎么"包"住链的

主角登场:RunnableWithMessageHistorylibs/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,对外表现得和原链一模一样——这就是"包一层"的技术含义。
💡 本质:用 LCEL 自己实现 LangChain 的功能注意这段源码没有任何"特殊机关"——它就是用你前 15 天学过的积木(RunnableLambdaRunnablePassthrough.assignwith_listeners| 管道)拼出来的。框架的高级功能,是用框架自己的基础能力搭的——这也是读源码最爽的时刻:一切都似曾相识。
L04

session_id 怎么变成"你那本"历史

调用时你传 config={"configurable": {"session_id": "foo"}},护士怎么据此取到"foo 的病历"?答案在 _merge_configslibs/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_idhistory.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 上随管道流动。多用户隔离就这么实现的:不同号取不同本,谁也不串。
L05

进场读历史 / 退场写历史

最后看两个钩子本体。进场 _enter_historylibs/core/langchain_core/runnables/history.py:521)+ 退场 _exit_historyhistory.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 实现。"记得多"和"塞得下"是一对永恒矛盾。
L06

全景图 + 一次对话的真实值

RunnableWithMessageHistory:一次 invoke 的读写时序 invoke({"question": "反函数呢?"}, config={session_id: "foo"}) _merge_configs get_session_history("foo") → 病历本 _enter_history 读历史 assign 塞进 {history} 占位 你的链 prompt | model 正常执行 (完全不知道"记忆"这回事) on_end → _exit_history 本轮问+答 add_messages 写回 病历本 store["foo"].messages Human("什么是余弦?") / AI("余弦是...") Human("反函数呢?") / AI("反余弦是...") 下一轮 invoke(同一个 session_id) _enter_history 会把这 4 条全翻出来 → 模型"记得"你聊过余弦
图注:护士(包装器)在链前读、链后写;你的链本体对"记忆"零感知。换 session_id 就换病历本,天然多用户隔离。
📝 真实值:两轮对话跑一遍(源码 docstring 里的原例) 链: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_messageshistory.py:452/490)那几个 if 里。

L07

它被标记废弃了?+ 今日小结

__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",
)
💡 取舍:为什么官方把"记忆"让给了 LangGraph?包装器方案只能管"一进一出"的链;而 Agent 时代的状态远不止聊天记录(还有工具结果、中间步骤、断点续跑)。LangGraph 的 checkpointer 把整个图状态thread_id 持久化,是记忆的超集。但今天没白学:存量项目里它仍大量存在;BaseChatMessageHistory 抽象没废,各种数据库集成还在用;"进场读 / 退场写 / 按 key 隔离"这套思路,正是 LangGraph checkpointer 在更大尺度上重演的东西。Day20 收官会再讲这层生态关系。

👶 小白:那"记忆"就只有"存聊天记录"这一种吗?

👨‍🏫 老师:这只是最基础的"逐字病历"。进阶玩法都建立在同一接口上:窗口记忆(只回填最近 N 条)、摘要记忆(旧对话让 LLM 压缩成一段摘要)、向量记忆(历史入向量库,按当前问题检索相关片段——其实就是把 Day09-12 的 RAG 用在自己的对话上)。接口统一的好处:这些策略都能无缝替换。

🧠 今天你应该能回答

  • LLM 为什么"没记性"?(每次调用无跨请求状态,历史必须随请求重发)
  • 记忆被拆成哪两个正交抽象?(存储 BaseChatMessageHistory + 时机 RunnableWithMessageHistory
  • session_id 怎么变成具体那本历史?(_merge_configsget_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
明日预告 · Day 17:今天 _exit_historywith_listeners 拿到 Run 记录——这背后是 LangChain 的神经系统:回调(Callbacks)。明天拆 CallbackManager 怎么在每个环节广播事件、stream/astream_events 的流式输出又是怎么从回调里"长"出来的。打字机效果的真相就在那里。
← Day 15 Agent Day 17 · 回调与流式 →