LangChain - Memory
目录+
一句话理解
大模型是无状态的——每次调用都是一次"失忆"，它根本不记得你上一句说了什么。所谓"多轮对话"的连贯感,其实是每次都把之前的聊天记录一起重新喂给模型装出来的。
Memory(记忆)就是负责"存历史 + 喂回去"这件事的机制。
看个最直白的例子,不带记忆时:
你: 我叫云帆。
AI: 你好,云帆!
你: 我叫什么名字?
AI: 抱歉,我不知道你的名字。 ← 它真的忘了
带上记忆后,第二轮发给模型的其实是:
[历史] Human: 我叫云帆。
AI: 你好,云帆!
[当前] Human: 我叫什么名字?
→ AI: 你叫云帆。 ← 因为历史被一起送了过去
模型本身没变,变的是我们每次都把对话历史拼进了输入。Memory 就是把这个"拼历史"的过程自动化、并管好"存哪、存多少、怎么取"。
经典记忆策略:存多少、怎么存
历史越攒越长,但模型的上下文窗口和 Token 成本有限,不可能无脑全塞。于是有了几种经典取舍策略:
| 策略 | 做法 | 适用场景 |
|---|---|---|
| Buffer(缓冲) | 全量保存历史,超出长度就截断 | 对话短、要完整上下文 |
| Window(窗口) | 只保留最近 K 轮对话 | 只关心近期、控制 Token |
| Summary(摘要) | 用 LLM 把旧历史压成一段摘要,只留摘要 | 长对话、需要长期主线 |
| 向量记忆 | 历史存入向量库,按当前问题检索式召回相关片段 | 超长历史、按需取相关内容 |
这几种策略对应旧版 LangChain 里的
ConversationBufferMemory、ConversationBufferWindowMemory、ConversationSummaryMemory等类。v1.0 已不再推荐这套Memory类(它们被移到了langchain-classic),取而代之的是下面两种更现代的做法:① 用RunnableWithMessageHistory给链加历史;② 在 LangGraph 里用 Checkpointer。但"存多少、怎么压"的思路是相通的,理解策略比记类名更重要。
新版方案:ChatMessageHistory + RunnableWithMessageHistory
新版 LangChain 把记忆拆成了两个职责清晰的组件:
ChatMessageHistory:一个消息仓库,负责把HumanMessage/AIMessage一条条存起来、读出来。RunnableWithMessageHistory:一个包装器,套在你已有的链外面,自动完成"调用前取历史、调用后存新消息"——你不用手写拼接代码。
三个关键拼图
要让一条普通的 prompt | llm | parser 链拥有记忆,需要三块拼图配合:
① MessagesPlaceholder 在 Prompt 里"挖一个坑",留给历史消息
② ChatMessageHistory 真正存历史的仓库(每个会话一个)
③ RunnableWithMessageHistory 把①和②接起来,自动取存
①MessagesPlaceholder:Prompt 模板里的占位符,运行时这个位置会被填入整段历史消息列表:
prompt_template = ChatPromptTemplate.from_messages([
("system", "你是一个聊天助手,用 {language} 回答所有的问题"),
MessagesPlaceholder(variable_name="history"), # ← 历史消息从这里注入
("human", "{input}"),
])
完整示例
from langchain.messages import HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from models import get_lc_model_client
client = get_lc_model_client()
prompt_template = ChatPromptTemplate.from_messages([
("system", "你是一个聊天助手,用 {language} 回答所有的问题"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}"),
])
chain = prompt_template | client | StrOutputParser()
# ② 仓库:按 session_id 隔离每个用户/会话的历史
store = {}
def get_session(session_id: str) -> ChatMessageHistory:
if session_id not in store:
store[session_id] = ChatMessageHistory()
return store[session_id]
# ③ 包装:把链和"取历史的函数"接起来
chatbot_with_his = RunnableWithMessageHistory(
chain,
get_session,
input_messages_key="input", # 输入里哪个键是"本轮用户消息"
history_messages_key="history", # 对应 MessagesPlaceholder 的变量名
)
config = {"configurable": {"session_id": "yunfang_chinese"}}
print(chatbot_with_his.invoke(
{"input": [HumanMessage(content="你好,我是云帆。")], "language": "中文"},
config=config,
))
print(chatbot_with_his.invoke(
{"input": [HumanMessage(content="请问我的名字是什么?")], "language": "中文"},
config=config,
))
# 第二轮能答出"云帆",因为历史被自动拼了进去
要点串一下:
- 调用时通过
config传session_id,RunnableWithMessageHistory据此调get_session拿到对应仓库; - 调用前,把仓库里的历史填进
history占位符;调用后,自动把本轮的input和模型回复追加回仓库; input_messages_key和history_messages_key必须和 Prompt 里的变量名对上,否则取存找不到位置。
ChatMessageHistory 的常用接口
上面 RunnableWithMessageHistory 是"自动取存",但底层那个仓库 ChatMessageHistory(实现自 BaseChatMessageHistory 接口)其实暴露了一组手动读写历史的方法。无论是 RedisChatMessageHistory、SQLChatMessageHistory 还是自定义实现,接口都是这一套——所以单独拎出来记一下,调试、迁移、自定义存储时都用得上。
| 接口 | 作用 |
|---|---|
messages | 属性,读出当前仓库里的全部历史消息(list[BaseMessage]) |
add_user_message(msg) | 追加一条用户消息(传 str 或 HumanMessage) |
add_ai_message(msg) | 追加一条 AI 回复(传 str 或 AIMessage) |
add_message(msg) | 追加任意一条消息(更通用,要传 BaseMessage 对象) |
add_messages(msgs) | 一次性追加多条消息 |
clear() | 清空该会话的全部历史 |
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain.messages import HumanMessage, AIMessage
history = ChatMessageHistory()
# 写入:两种风格等价
history.add_user_message("你好,我是云帆。") # 便捷方法,直接传字符串
history.add_ai_message("你好,云帆!")
history.add_message(HumanMessage(content="我叫什么?")) # 通用方法,传消息对象
history.add_messages([ # 批量追加
AIMessage(content="你叫云帆。"),
])
# 读取
for m in history.messages:
print(type(m).__name__, m.content)
# 清空
history.clear()
上表都是同步方法;每个方法还有对应的异步版本(
aget_messages()、aadd_messages()、aclear()等),用法一致,只是await调用。自定义存储时,最少实现messages、add_messages、clear三样即可接入RunnableWithMessageHistory。
多会话隔离:session_id
store 是个普通字典,session_id 是它的键——不同的 session_id 拿到不同的 ChatMessageHistory,历史天然隔离,互不串台:
config_a = {"configurable": {"session_id": "user_A"}}
config_b = {"configurable": {"session_id": "user_B"}}
# user_A 和 user_B 的对话各存各的,A 记不住 B 说过的话
上面用内存字典
store只是演示,进程一停历史就没了。生产环境应换成持久化实现,如RedisChatMessageHistory、SQLChatMessageHistory等,把历史落到 Redis / 数据库,重启也不丢。
LangGraph 时代:Checkpointer
在 v1.0 主推的 LangGraph / create_agent 体系里,记忆又换了一套更强的机制——Checkpointer(检查点)。
它不再只存"消息列表",而是在每一步自动保存 Agent 的整个状态(消息、工具调用、中间变量……),用 thread_id 标识一个会话线程;下次传同一个 thread_id,就能加载之前的完整状态接着跑。
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
agent = create_agent(
model=llm,
tools=tools,
system_prompt="你是人工智能助手。",
checkpointer=InMemorySaver(), # ← 短期记忆:线程级状态持久化
)
# 同一个 thread_id = 同一段连续对话
config = {"configurable": {"thread_id": "user_1"}}
agent.invoke({"messages": [{"role": "user", "content": "我叫云帆"}]}, config=config)
agent.invoke({"messages": [{"role": "user", "content": "我叫什么?"}]}, config=config)
相比 RunnableWithMessageHistory,Checkpointer 的优势在于:
- 存的是整个状态而非仅消息,天然支持工具调用、分支、循环等复杂 Agent 流程;
- 自带"时光倒流"(Time Travel)能力,可回到任意历史检查点重放、调试;
- 换持久化后端(如
SqliteSaver、PostgresSaver)即可落地存储。
一句话对应关系:旧版
Memory类 → 新版RunnableWithMessageHistory(给链加历史)→ LangGraphCheckpointer(给 Agent 加状态)。越往后越通用、越强。
短期记忆 vs 长期记忆
上面讲的(无论是消息历史还是 Checkpointer)本质都是短期记忆——只在一段会话/线程内有效,服务的是"这次聊天的上下文连贯"。
而长期记忆指的是跨会话留存的知识:用户的长期偏好、画像、历史结论等。它通常不靠 Checkpointer,而是落到外部存储里,在需要时检索回来——实现上常和 [RAG](/ai/langchain/l-a-n-g-c-h-a-i-n-gai-shu#7.2 RAG（检索增强生成）) / 向量库重叠(把记忆当成可检索的文档)。
| 短期记忆 | 长期记忆 | |
|---|---|---|
| 范围 | 单次会话 / 线程内 | 跨会话、长期留存 |
| 存什么 | 本轮对话的消息、状态 | 用户偏好、画像、沉淀知识 |
| 典型实现 | RunnableWithMessageHistory / Checkpointer | 向量库 + 检索(RAG 思路) |
| 解决什么 | "记得我们刚才聊了啥" | "记得我是谁、我喜欢什么" |
小结
| 要点 | 一句话 |
|---|---|
| 为什么要 Memory | 大模型无状态,多轮连贯靠"每次重新喂历史"实现 |
| 核心策略 | Buffer(全量)/ Window(近 K 轮)/ Summary(摘要)/ 向量(检索召回) |
| 新版给链加记忆 | MessagesPlaceholder 挖坑 + ChatMessageHistory 存 + RunnableWithMessageHistory 自动取存 |
| 会话隔离 | 用 session_id(或 LangGraph 的 thread_id)区分,互不串台 |
| LangGraph 方案 | Checkpointer 存整个状态,更强、支持时光倒流 |
| 短期 vs 长期 | 短期=会话内上下文;长期=跨会话知识,常用 RAG 实现 |
相关笔记:LangChain 概述LangChain 概述 1. 什么是 LangChain LangChain 是一个用于开发由大型语言模型（LLM）驱动的应用程序的开源框架。它的定位是"LLM 应用的中间件 / 胶水层"——抽象了 LLM 调用、链式组合、工具集成、记忆管理、检索增强等通用能力，让开发者不必从零拼装，把 LLM、工具、数据和业务逻辑像搭积木一样组合成应用。 我们希望大模型应用不仅仅是聊天，更能从已有的数据库或文件中提取信息并执行具体操作。LangChain 正是为此而生：它允许开发者将通义千问、DeepSeek、OpenAI 等大语言模型与外部系统和数据源结合，完成更复杂的操作。 也可以借助 Dify、RAGFlow 之类的低代码平台，但自由度受限，对复杂业务适配欠缺；也可以完全自研裸调 SDK，但费时费力。LangChain 提供了现成组件，同时保留了足够的自定义自由度。 创始人**：Harrison Chase（2022 年开源） 语言**：Python / JavaScript（TypeScript） 官网**：https://python.langchain.c · LangChain - LCEL 高级特性与组件LangChain - LCEL 高级特性与组件 LCEL（LangChain Expression Language）用 | 把各个 Runnable 串成链。除了 prompt | llm | parser 这种基础用法，还有几个常用组件能把自定义函数、并行分支、数据透传等能力接进链里。 | 组件 | 作用 | | --- | --- | | RunnableLambda | 把普通 Python 函数包装成 Runnable，接进链里 | | RunnableParallel | 多个分支接收同一份输入，并行执行，结果汇总成 dict | | RunnablePassthrough | 把输入原样传递下去，或用 assign 增强后再传 | RunnableLambda —— 把自定义函数加入链 RunnableLambda 把任意普通 Python 函数（自定义函数）包装成一个 Runnable，从而能用 | 接进链里。 案例：在 LCEL 链中用 RunnableLambda 对 LLM 的输出统计字数。 from langchain_core.runnables import · Naive RAG 概述Naive RAG 讲解RAG，全称 Retrieval-Augmented Generation（检索增强生成），是过去两年大模型应用落地里最重要的工程方案之一。 如果把大模型比作一个"很聪明但记忆固定的学生"，那么 RAG 做的事情就是：先去外部资料里找答案，再带着资料回答问题。 而 Naive RAG，可以理解为最基础、最原始、也是最容易落地的一代 RAG 方案。它的结构并不复杂，但已经足够解决很多实际问题，比如企业知识库问答、内部文档检索、客服机器人、私有数据接入大模型等。 一、为什么需要 RAG 大语言模型能力很强，但存在三类核心局限。 1. 时效性有限 模型知识来自训练数据，而训练数据有截止时间。最新政策、财报、内部 SOP、个人知识库文档——这些如果没被放进训练数据，模型就无法直接回答。 2. 知识覆盖不完整 即使训练了海量语料，也不可能覆盖所有垂直领域和每家企业的内部资料。医疗、法律、金融等高门槛领域，以及内部文档、合同、产品手册、私有接口说明——模型可能"懂一点"，但不够深、不够准。 3. 幻觉问题 大模型在不知道答案时，仍可能生成一个看起来很像答案的内容——语气自信，内