RunnableWithMessageHistory 完全指南上基础与存储实战本文是 LangChain Expression Language (LCEL) 系列的开篇。我们从「为什么需要对话历史」出发拆解RunnableWithMessageHistory的设计哲学、核心机制与生产级存储方案帮助你从「能跑」走向「能扛」。官方源码参考langchain_core/runnables/history.py一、为什么需要对话历史1.1 无状态 LLM 的局限大型语言模型LLM本质上是无状态的每次 API 调用都是独立的模型不会「记住」你上一句说了什么。fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o-mini)# 需配置 OPENAI_API_KEYllm.invoke(我叫张三)# 模型回应你好张三llm.invoke(我叫什么名字)# 模型回应我不知道你叫什么第二句完全丢失了第一句的上下文。单轮问答无所谓多轮对话则不可接受。1.2 手动管理历史的痛点早期开发者通常手动维护消息列表fromlangchain_core.messagesimportHumanMessage,AIMessage messages[HumanMessage(我叫张三),AIMessage(你好张三很高兴认识你),HumanMessage(我叫什么名字),]# 每次都要手动加载 → 拼接 → 调用 → 保存 ...痛点具体表现代码冗余每个对话接口都要写加载、拼接、保存三段逻辑状态混乱Session 隔离容易出错A 用户看到 B 用户的对话存储耦合业务代码里直接写 Redis/DB 连接难以替换边界情况并发写入、消息截断、异常回滚全靠手写1.3 RunnableWithMessageHistory声明式 vs 命令式LangChain 的解法是RunnableWithMessageHistory——一个声明式的历史管理包装器。命令式你告诉计算机「怎么做」加载历史 → 拼接消息 → 调用模型 → 保存历史声明式你告诉计算机「要什么」这个 Runnable 需要对话历史按 Session ID 隔离用 Redis 存储框架处理「怎么做」runnable_with_historyRunnableWithMessageHistory(chain,# 你的核心逻辑get_session_history,# 历史存储工厂函数input_messages_keyinput,history_messages_keyhistory,)这种设计将历史存储与业务逻辑解耦存储后端可以从内存换到 Redis、PostgreSQL而业务代码基本不动。二、核心概念图解2.1 包装器模式┌─────────────────────────────────────────────┐ │ RunnableWithMessageHistory │ │ ┌─────────────────────────────────────┐ │ │ │ 你的核心 Chain/Runnable │ │ │ │ (PromptTemplate | LLM | OutputParser) │ │ │ └─────────────────────────────────────┘ │ │ ▲ │ │ ┌─────────────────┼─────────────────────┐ │ │ │ 注入历史到 dict │ 保存本轮新消息 │ │ │ │ (load history) │ (save history) │ │ │ └─────────────────┴─────────────────────┘ │ │ ▲ │ │ ┌─────────────────┴─────────────────────┐ │ │ │ BaseChatMessageHistory 实现 │ │ │ │ (InMemory / Redis / Postgres ...) │ │ │ └─────────────────────────────────────┘ │ └─────────────────────────────────────────────┘对调用者而言它就是一个普通的Runnable内部自动完成历史的加载、注入和持久化。2.2 Session ID 与多维隔离Session ID 是历史隔离的标识。工厂函数get_session_history根据传入的 ID 返回对应的存储实例store{}defget_session_history(session_id:str)-InMemoryChatMessageHistory:ifsession_idnotinstore:store[session_id]InMemoryChatMessageHistory()returnstore[session_id]关键规则相同session_id→ 共享同一份历史不同session_id→ 历史完全隔离session_id由调用方在config中传入不由模型生成需要user_id conversation_id等多维隔离时使用history_factory_config详见官方示例。2.3 两种工作模式重要使用ChatPromptTemplate时务必先分清这两种模式不要混用模式 Adict 分离推荐模式 B消息列表合并典型 Chainprompt | llm直接llm配置指定input_messages_keyhistory_messages_key两者均不指定Prompt需要MessagesPlaceholder不需要 Prompt 模板invoke 输入{input: 你好}你好或[HumanMessage(...)]历史注入方式写入 dict 的history键与当前输入合并为消息列表本文后续示例均基于模式 A——也是 LangChain 官方 docstring 中的推荐写法。2.4 模式 A 的数据流用户输入 session_id │ ▼ ┌─────────────────┐ │ 1. 接收输入 │ invoke({input: 我叫什么}, │ │ config{configurable: {session_id: user_123}}) └────────┬────────┘ │ ▼ ┌─────────────────┐ │ 2. 加载历史 │ 从存储读取 → 注入 dict[history] │ (自动) │ [HumanMessage(我叫张三), AIMessage(你好张三...)] └────────┬────────┘ │ ▼ ┌─────────────────┐ │ 3. Prompt 渲染 │ MessagesPlaceholder 展开 history │ (自动) │ {input} 渲染当前问题 └────────┬────────┘ → [System, Human, AI, Human(当前)] │ ▼ ┌─────────────────┐ │ 4. 执行 LLM │ └────────┬────────┘ │ ▼ ┌─────────────────┐ │ 5. 保存历史 │ 仅追加本轮 Human AI不重复写入旧历史 │ (自动) │ └─────────────────┘你需要关注的步骤 1 的输入构造、Prompt 模板设计、以及get_session_history的实现。步骤 2、3、5 由框架自动完成。三、最小可运行示例内存版3.1 完整可复制的代码fromlangchain_core.promptsimportChatPromptTemplate,MessagesPlaceholderfromlangchain_core.runnables.historyimportRunnableWithMessageHistoryfromlangchain_core.chat_historyimportInMemoryChatMessageHistoryfromlangchain_openaiimportChatOpenAI# 1. 构建核心 Chain promptChatPromptTemplate.from_messages([(system,你是一个乐于助人的 AI 助手。),MessagesPlaceholder(variable_namehistory),# 历史消息渲染位置(human,{input}),])llmChatOpenAI(modelgpt-4o-mini)chainprompt|llm# 2. 定义历史存储工厂 store{}# 内存存储仅用于开发测试defget_session_history(session_id:str)-InMemoryChatMessageHistory:ifsession_idnotinstore:store[session_id]InMemoryChatMessageHistory()returnstore[session_id]# 3. 包装为带历史的 Runnable runnable_with_historyRunnableWithMessageHistory(chain,get_session_history,input_messages_keyinput,# 输入 dict 中「当前用户消息」的 keyhistory_messages_keyhistory,# Prompt 中 MessagesPlaceholder 的变量名)# 4. 调用 config{configurable:{session_id:demo_session_001}}response1runnable_with_history.invoke({input:我叫张三},configconfig)print(response1.content)response2runnable_with_history.invoke({input:我叫什么名字},configconfig)print(response2.content)# 输出你叫张三3.2 关键配置input_messages_key / history_messages_key这两个参数建立输入 dict、Prompt 模板和历史存储三者的映射输入 dict Prompt 模板 ┌─────────────┐ ┌─────────────────────────────┐ │ input │ ───────────────► │ (human, {input}) │ ← input_messages_key │ context │ ──RAG 直传──► │ (human, ...{context}) │ └─────────────┘ │ MessagesPlaceholder │ ← history_messages_key │ (variable_namehistory) │ └─────────────────────────────┘参数作用对应 Prompt 中的input_messages_key输入 dict 中当前用户消息的 key(human, {input})里的inputhistory_messages_key历史消息注入到 dict 的 keyMessagesPlaceholder(variable_namehistory)里的history常见配置# 标准对话RunnableWithMessageHistory(chain,get_session_history,input_messages_keyinput,history_messages_keyhistory)# invoke: {input: 你好}# RAG 多输入context 等额外字段直传给 Prompt不参与历史管理RunnableWithMessageHistory(chain,get_session_history,input_messages_keyquestion,history_messages_keychat_history)# invoke: {question: ..., context: 检索到的文档...}若 chain 返回 dict 而非AIMessage还需指定output_messages_key否则保存历史时可能出错。3.3 RunnableWithMessageHistory 与 MessagesPlaceholder 如何协作这是初学者最常见的疑问两者会不会重复插入历史不会。它们分工不同是官方推荐的组合写法源码示例组件职责RunnableWithMessageHistory存储层加载/保存历史将历史写入输入 dict 的history键MessagesPlaceholder模板层告诉 Prompt 在哪个位置渲染history里的消息第二轮 invoke 时框架构造的 dict 如下{input:我叫什么名字,# 仅当前轮history:[HumanMessage(我叫张三),AIMessage(...)]# 仅历史轮次}MessagesPlaceholder(history)负责把history读出来渲染它不做加载、不做保存。去掉它历史虽然被加载了却进不了发给 LLM 的消息列表多轮对话会失效。保存时框架只追加本轮的 Human AI 消息不会把旧历史重复写入存储——源码在「模式 B」下才有去重逻辑模式 A 天然不会重复。会导致重复的错误用法invoke 时手动传入history框架又自动注入一遍混用模式 A/B不设history_messages_keyPrompt 里又用MessagesPlaceholder在{input}里手动拼接了历史文本3.4 常见报错排查报错 1InputValueError: input_messages_key not in input dict原因配置了input_messages_keyinput但 invoke 传入的 dict 没有该 key。# 错误runnable_with_history.invoke({question:你好},configconfig)# 正确对齐 keyrunnable_with_history.invoke({input:你好},configconfig)# 或修改配置RunnableWithMessageHistory(...,input_messages_keyquestion)报错 2模板变量history未找到原因history_messages_keyhistory但 Prompt 里MessagesPlaceholder的variable_name不一致。# 错误MessagesPlaceholder(variable_namechat_history)RunnableWithMessageHistory(...,history_messages_keyhistory)# 正确两者保持一致MessagesPlaceholder(variable_namechat_history)RunnableWithMessageHistory(...,history_messages_keychat_history)报错 3缺少session_id原因未在 config 中传入 session 标识。# 错误runnable_with_history.invoke({input:你好})# 正确runnable_with_history.invoke({input:你好},config{configurable:{session_id:demo_session_001}},)报错 4ValueError: Expected a list of messages...原因MessagesPlaceholder收到的不是BaseMessage列表。解决确保get_session_history返回的是BaseChatMessageHistory实现如InMemoryChatMessageHistory而不是原始字符串列表。四、Prompt 模板设计规范4.1 模式 A 必须预留 MessagesPlaceholder在history_messages_keyChatPromptTemplate模式下MessagesPlaceholder是必需的# ✅ 正确promptChatPromptTemplate.from_messages([(system,你是一个乐于助人的 AI 助手。),MessagesPlaceholder(variable_namehistory),(human,{input}),])# ❌ 错误历史加载了但无法渲染到 Prompt 中promptChatPromptTemplate.from_messages([(system,你是一个乐于助人的 AI 助手。),(human,{input}),])4.2 history 变量名约定场景推荐变量名说明通用对话history最常用多轮 RAGchat_history与context检索文档区分角色扮演conversation_history强调对话历史一致性原则history_messages_key必须与MessagesPlaceholder的variable_name完全一致。4.3 System Message 的位置# ✅ 推荐System → 历史 → 当前输入promptChatPromptTemplate.from_messages([(system,你是专业助手。当前时间{datetime}),MessagesPlaceholder(variable_namehistory),(human,{input}),])# ❌ 不推荐System 放在历史之后权重可能被稀释promptChatPromptTemplate.from_messages([MessagesPlaceholder(variable_namehistory),(system,你是专业助手。),(human,{input}),])五、生产级存储集成内存版仅用于开发测试。生产环境需要持久化、共享、高可用的存储。以下四种方案只需替换get_session_historyRunnableWithMessageHistory的配置保持不变。5.1 Redis多实例共享会话适用场景高并发 Web 应用多实例部署需共享会话状态。pipinstalllangchain-redis redisfromlangchain_redisimportRedisChatMessageHistoryimportredis poolredis.ConnectionPool(hostlocalhost,port6379,db0,max_connections50,socket_keepaliveTrue,socket_connect_timeout5,socket_timeout5,health_check_interval30,)defget_session_history(session_id:str)-RedisChatMessageHistory:returnRedisChatMessageHistory(session_idsession_id,redis_clientredis.Redis(connection_poolpool),ttl3600*24*7,# 7 天过期)优化项说明连接池复用 TCP 连接避免每次请求建连TTL防止 Redis 内存无限增长Key 前缀langchain-redis默认key_prefixchat:可按需自定义5.2 PostgreSQL结构化持久化适用场景长期保存、审计、备份迁移。pipinstalllangchain-community sqlalchemy psycopg2-binarySQLChatMessageHistory会自动建表默认 schema 为列名类型说明idINTEGER PK自增主键session_idTEXT会话标识messageTEXTBaseMessage的 JSON 字符串一般无需手写 DDL。如需自定义 schema如加created_at通过custom_message_converter扩展。fromlangchain_community.chat_message_historiesimportSQLChatMessageHistoryfromsqlalchemyimportcreate_engine enginecreate_engine(postgresqlpsycopg2://user:passlocalhost/dbname,pool_size10,max_overflow20,pool_pre_pingTrue,)defget_session_history(session_id:str)-SQLChatMessageHistory:returnSQLChatMessageHistory(session_idsession_id,connectionengine,table_namemessage_store,)备份与迁移historyget_session_history(user_123)messageshistory.messages# List[BaseMessage]new_historyRedisChatMessageHistory(session_iduser_123,redis_urlredis://localhost:6379)formsginmessages:new_history.add_message(msg)5.3 MongoDB适用场景已有 MongoDB 基础设施或需要灵活文档结构。pipinstalllangchain-mongodb pymongoMongoDBChatMessageHistory默认按session_id存储文档结构类似{SessionId:user123_001,History:[{type:human,data:{content:我叫张三,type:human,additional_kwargs:{}}},{type:ai,data:{content:你好张三,type:ai,additional_kwargs:{}}}]}fromlangchain_mongodbimportMongoDBChatMessageHistoryfrompymongoimportMongoClient clientMongoClient(mongodb://localhost:27017/)defget_session_history(session_id:str)-MongoDBChatMessageHistory:returnMongoDBChatMessageHistory(connection_stringmongodb://localhost:27017/,session_idsession_id,database_namechat_db,collection_namechat_histories,clientclient,# 复用连接避免每次创建)集合策略适用场景单集合 索引大多数场景推荐按租户分集合超大规模多租户便于归档按时间分集合日志型数据定期冷热分离5.4 Upstash RedisServerless适用场景Vercel、Netlify、AWS Lambda 等无长连接环境。Serverless 的痛点实例无状态、函数执行完即销毁传统 Redis TCP 长连接不可靠。Upstash 提供 HTTP REST API天然适配。pipinstalllangchain-community upstash-redisfromlangchain_community.chat_message_historiesimportUpstashRedisChatMessageHistorydefget_session_history(session_id:str)-UpstashRedisChatMessageHistory:returnUpstashRedisChatMessageHistory(session_idsession_id,urlhttps://your-instance.upstash.io,tokenyour-token,ttl3600,)优化项原因HTTP 而非 TCP无连接状态适合 Serverless短 TTL控制冷启动时的历史加载量HTTP Keep-Alive单实例内复用 HTTP 连接六、上篇总结 下篇预告6.1 数据流一图流用户输入 ──► 加载历史 ──► Prompt 渲染 ──► 执行 LLM ──► 保存本轮消息 {input} session_id history 生成回复 Human AI ──► dict {input}6.2 核心要点要点内容设计哲学声明式历史管理存储与逻辑解耦两种模式模式 Adict MessagesPlaceholdervs 模式 B消息列表直传 LLM不要混用协作关系RunnableWithMessageHistory注入 dictMessagesPlaceholder渲染 Prompt不重复关键配置input_messages_keyhistory_messages_key必须与 Prompt 对齐Sessionconfig{configurable: {session_id: ...}}必传开发存储InMemoryChatMessageHistory仅用于本地调试生产存储Redis / PostgreSQL / MongoDB / Upstash只换get_session_history6.3 下篇预告流式输出stream()/astream()与历史管理的配合消息截断Token 超限时的滑动窗口、摘要压缩自定义实现继承BaseChatMessageHistory审计日志、敏感词过滤性能优化异步存储、批量写入、缓存策略附录依赖与版本包用途langchain-core≥ 0.2.0核心 Runnable 与InMemoryChatMessageHistorylangchain-openai示例 LLMlangchain-redisRedis 存储langchain-communitySQL / Upstash 存储langchain-mongodbMongoDB 存储代码示例基于 LangChain 0.2 LCEL API。旧版ConversationChain写法已废弃请以 官方文档 为准。