LangChain RunnableWithMessageHistory 实践：让 LCEL 链自动管理聊天历史

在普通 LCEL 链里，聊天历史最开始往往是手动处理的。

调用前先读历史，把历史放进 Prompt；模型输出后，再把用户输入和模型回答写回历史。这个流程并不复杂，但每条链都手写一遍，很快就会出现重复代码、漏写历史、session 串线、输出字段对不上等问题。

RunnableWithMessageHistory 解决的就是这个问题：它把一个普通 Runnable 包起来，让链路在调用前自动加载历史，在调用后自动写回历史。

它适合普通聊天链、轻量问答、LCEL 应用；但如果你已经在用 create_agent 构建复杂 Agent，短期记忆主线应该优先考虑 checkpointer。

它解决了什么重复代码

一个带历史的聊天链通常有四步：

根据 session_id 读取历史
  -> 把历史填入 Prompt 的 MessagesPlaceholder
  -> 调用模型生成回答
  -> 把本轮 HumanMessage 和 AIMessage 写回历史

如果手动写，业务代码里会混进很多记忆读写逻辑：

history = get_session_history(session_id)
messages = history.messages + [HumanMessage(query)]
response = model.invoke(messages)
history.add_user_message(query)
history.add_ai_message(response.content)

RunnableWithMessageHistory 把这段流程变成包装器。你只需要告诉它：

• 原始 runnable 是什么。
• 怎么根据 session 找到消息历史。
• 输入里哪个字段是用户消息。
• Prompt 里哪个字段接收历史消息。
• 输出里哪个字段要写回历史。

核心参数

RunnableWithMessageHistory 常用参数有几个。

参数	作用
runnable	被包装的 LCEL 链或 Runnable
get_session_history	根据 session 配置返回 BaseChatMessageHistory
input_messages_key	输入中哪个字段代表本轮用户消息
output_messages_key	输出中哪个字段代表模型消息，输出是 dict 时常用
history_messages_key	Prompt 中哪个字段用于接收历史消息
history_factory_config	自定义历史工厂需要的配置字段

最常见的是：

RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="query",
    history_messages_key="history",
)

调用时要通过 config["configurable"] 传入 session_id：

chain.invoke(
    {"query": "我叫什么？"},
    config={"configurable": {"session_id": "session-001"}},
)

注意：session_id 不是 Prompt 输入，不会直接给模型看。它是运行配置，用来定位消息历史。

最小示例

下面是一个文件持久化版本的示例。

from langchain_community.chat_message_histories import FileChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI


store: dict[str, BaseChatMessageHistory] = {}


def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = FileChatMessageHistory(
            f"./storage/chat_history_{session_id}.txt"
        )

    return store[session_id]


prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个严谨、克制、准确的技术助手。"),
        MessagesPlaceholder("history"),
        ("human", "{query}"),
    ]
)

llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
chain = prompt | llm | StrOutputParser()

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="query",
    history_messages_key="history",
)

answer = chain_with_history.invoke(
    {"query": "我叫 Alice，正在做 Flask 项目。"},
    config={"configurable": {"session_id": "user-42:chat-1001"}},
)

answer = chain_with_history.invoke(
    {"query": "我刚才说我在做什么？"},
    config={"configurable": {"session_id": "user-42:chat-1001"}},
)

这段代码里，原始 chain 不关心历史读写。历史加载和保存都由包装器处理。

运行流程

invoke / stream 输入
  -> 读取 config.configurable.session_id
  -> 调用 get_session_history(session_id)
  -> 加载 history.messages
  -> 写入 history_messages_key 对应字段
  -> 调用原始 Runnable
  -> 解析输入消息和输出消息
  -> 追加到 BaseChatMessageHistory
  -> 返回原始 Runnable 输出

内部可以理解成三段。

第一段是进入链路前加载历史。包装器根据 session_id 找到 message_history，然后把历史消息插入到 Prompt 所需字段中。

第二段是调用原始 Runnable。这个 Runnable 可以是 prompt | model | parser，也可以是更复杂的 LCEL 链。

第三段是链路结束后写回历史。包装器会从输入中提取用户消息，从输出中提取模型消息，然后调用 hist.add_messages(...) 追加到历史里。

output_messages_key 什么时候需要

如果链路输出是字符串，可以不配置 output_messages_key。

chain = prompt | llm | StrOutputParser()

如果链路输出是 dict，就需要告诉包装器哪个字段是模型回答。

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="query",
    output_messages_key="answer",
    history_messages_key="history",
)

否则包装器不知道应该把哪个输出写成 AIMessage。

多参数 session 配置

生产里只用一个 session_id 往往不够。更稳妥的是用 tenant_id、user_id、conversation_id 一起定位历史。

可以通过 history_factory_config 声明多个配置字段：

from langchain_core.runnables import ConfigurableFieldSpec


def get_session_history(
    tenant_id: str,
    user_id: str,
    conversation_id: str,
) -> BaseChatMessageHistory:
    return PostgresChatMessageHistory(
        tenant_id=tenant_id,
        user_id=user_id,
        conversation_id=conversation_id,
    )


chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="query",
    history_messages_key="history",
    history_factory_config=[
        ConfigurableFieldSpec(
            id="tenant_id",
            annotation=str,
            name="Tenant ID",
            description="租户 ID",
            is_shared=True,
        ),
        ConfigurableFieldSpec(
            id="user_id",
            annotation=str,
            name="User ID",
            description="用户 ID",
            is_shared=True,
        ),
        ConfigurableFieldSpec(
            id="conversation_id",
            annotation=str,
            name="Conversation ID",
            description="会话 ID",
            is_shared=True,
        ),
    ],
)

调用：

chain_with_history.invoke(
    {"query": "继续刚才的问题。"},
    config={
        "configurable": {
            "tenant_id": "tenant-1",
            "user_id": "user-42",
            "conversation_id": "chat-1001",
        }
    },
)

这比把所有信息拼成一个字符串更容易做权限校验和审计。

源码里的关键动作

从源码流程看，RunnableWithMessageHistory 关键动作有三个。

第一，构造时会创建一个 history_chain。它负责调用 _enter_history 加载历史，并把历史插入到输入里。

第二，它会给原始 runnable 绑定监听器。原始链运行结束后，触发 _exit_history，把输入消息和输出消息写入 message_history。

第三，_merge_configs 会检查运行配置。如果缺少 session_id 或自定义配置字段，会直接报错，提示你应该通过 config["configurable"] 传入。

这也是为什么下面这种调用会失败：

chain_with_history.invoke({"query": "你好"})

正确写法是：

chain_with_history.invoke(
    {"query": "你好"},
    config={"configurable": {"session_id": "session-001"}},
)

和手动 Memory 写法的区别

旧写法通常是：

history = memory.load_memory_variables(inputs)
result = chain.invoke({**inputs, **history})
memory.save_context(inputs, {"output": result})

RunnableWithMessageHistory 把这件事标准化为 Runnable 包装器。

它的优势是：

• 读写历史逻辑不侵入业务链。
• 同一条链可以服务多个 session。
• 支持 invoke、stream、batch 等 Runnable 调用方式。
• 配合 BaseChatMessageHistory 可以接文件、Redis、Postgres 等存储。
• 不需要每次手动调用 save_context。

但它也不是完整 Agent 记忆系统。它主要服务普通 Runnable 的消息历史管理。

问题

RunnableWithMessageHistory 的主要问题，是它只自动化“消息历史读写”，不自动解决生产记忆治理。

常见风险包括：

• session_id 由前端随便传，导致越权读取历史。
• get_session_history 用全局 dict，服务重启后丢失。
• 历史无限增长，没有 token 裁剪和摘要。
• 输出是 dict 时忘记配置 output_messages_key，导致写回失败或写错。
• 工具消息、异常消息、流式中断消息没有一致处理。
• 把聊天历史当长期记忆，导致过期事实污染回答。

它能减少样板代码，但不能替你做权限、持久化、删除、摘要、审计和可观测性。

拓展

可以从三个方向拓展。

第一，持久化 history。把 FileChatMessageHistory 换成 Postgres、Redis、MongoDB 或自定义 BaseChatMessageHistory。

第二，增强上下文选择。在历史进入 Prompt 前做窗口裁剪、token 预算、摘要合并和敏感信息过滤。

第三，分离长期记忆。用户偏好、项目事实、实体信息不要只放在聊天历史里，而应该抽取到 Store 或业务数据库。

更完整的结构是：

RunnableWithMessageHistory
  -> 负责当前 session 消息读写
Context selector
  -> 负责选择本轮要进入 Prompt 的历史
Long-term memory extractor
  -> 负责抽取跨会话偏好和事实
Store
  -> 负责长期记忆持久化

实际生产是否使用

会使用，但主要用于普通 LCEL 链，不是复杂 Agent 的主记忆方案。

如果你的应用是一个普通聊天接口、RAG 问答链、客服助手，RunnableWithMessageHistory 很适合。它能让链路保持 Runnable 风格，同时统一处理多 session 消息历史。

如果你的应用是复杂 Agent，包含工具调用、长任务、状态恢复、人机协同、分支执行，那么生产里更推荐 checkpointer。因为 checkpointer 保存的是完整 Agent state，而不只是聊天消息。

现在是否抛弃

没有抛弃。

RunnableWithMessageHistory 仍然是当前 LangChain 里管理普通 Runnable 消息历史的有效方式。它和 BaseChatMessageHistory、InMemoryChatMessageHistory 属于仍然可用的核心抽象。

但它的边界要看清：它不是新版 Agent 短期记忆主线。新 Agent 更推荐 create_agent + checkpointer + thread_id。

最新生产如何实现

普通 LCEL 链可以这样实现：

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="query",
    output_messages_key="answer",
    history_messages_key="history",
)

生产里的 get_session_history 应该做权限校验和持久化：

def get_session_history(
    tenant_id: str,
    user_id: str,
    conversation_id: str,
) -> BaseChatMessageHistory:
    assert_can_access_conversation(tenant_id, user_id, conversation_id)

    return PostgresChatMessageHistory(
        tenant_id=tenant_id,
        user_id=user_id,
        conversation_id=conversation_id,
        pool=db_pool,
    )

复杂 Agent 使用 checkpointer：

from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver


with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()

    agent = create_agent(
        model="gpt-5.5",
        tools=[],
        checkpointer=checkpointer,
    )

生产还要补上：消息表索引、软删除、token 裁剪、摘要、敏感信息过滤、trace metadata、失败写入策略和保留周期。

总结

RunnableWithMessageHistory 的价值，是把“读历史、填 Prompt、写历史”这套重复流程包装成 Runnable 能力。它让普通 LCEL 链更干净，也让多 session 聊天更容易维护。

但它不是万能记忆系统。普通链路可以用它，复杂 Agent 要用 checkpointer，跨会话长期偏好和事实要用 Store 或业务数据库。

可以这样记：

• 普通聊天链：RunnableWithMessageHistory。
• 消息历史存储：BaseChatMessageHistory 实现。
• Agent 短期记忆：checkpointer。
• 长期记忆：Store。

参考：

• LangChain RunnableWithMessageHistory API (opens new window)
• LangChain Short-term memory (opens new window)