LangChain 记忆模块快速上手

本章介绍

本章聚焦LangChain框架的核心能力之一——记忆模块，从LLM记忆机制的底层逻辑出发，逐步讲解常见记忆模式、手动模拟记忆功能、LangChain原生记忆组件的使用与源码解析，最终实现带记忆功能的聊天机器人。同时拓展Chain组件与RunnableWithMessageHistory工具的使用，帮助开发者掌握记忆模块的工程化落地技巧。

一、LLM实现记忆功能的核心思路与常见记忆模式

1.1 核心实现思路

LLM本身不具备上下文记忆能力，仅能根据当前输入生成输出。实现记忆功能的核心逻辑是：

存储历史对话：通过专门的记忆模块保存用户（Human）与AI的历史交互信息；
注入上下文：在每次新请求时，将历史对话信息插入到Prompt的预留占位符（如{chat_history}）中；
实时更新记忆：新的对话生成后，同步更新记忆模块中的历史记录。

核心流程：

用户提问 → 记忆模块提取历史对话 → 拼接Prompt（系统指令+历史对话+新提问） → LLM生成响应 → 记忆模块存储新对话 → 返回结果

1.2 常见记忆模式

根据历史对话的处理方式，LangChain支持7种主流记忆模式，适用于不同场景：

记忆模式	核心逻辑	优点	缺点
缓冲记忆（ConversationBufferMemory）	存储所有历史对话，完整注入Prompt	1. 无损记忆，细节不丢失；2. 实现简单，兼容性强	1. Token消耗大，响应慢、成本高；2. 易触发LLM上下文长度限制
缓冲窗口记忆（ConversationBufferWindowMemory）	仅保留最近k轮对话（k为窗口大小）	1. 控制Token使用量；2. 性能优异，适配小模型	1. 会遗忘早期对话；2. 长对话仍可能超Token限制
令牌缓冲记忆（ConversationTokenBufferMemory）	按Token数限制记忆长度，超限时删除旧对话	1. 精准匹配LLM上下文限制；2. 适配各类模型	1. 遗忘早期对话；2. 需额外计算Token数
摘要总结记忆（ConversationSummaryMemory）	不存储完整对话，仅保存对话摘要（由LLM生成）	1. 支持长期对话记忆；2. Token消耗低，增长慢	1. 细节丢失（模糊记忆）；2. 依赖摘要LLM能力，增加额外成本
摘要缓冲混合记忆（ConversationSummaryBufferMemory）	短期对话保留完整内容，超Token限制后将早期对话总结为摘要	1. 短期精准记忆+长期模糊记忆；2. Token消耗平衡	1. 长期记忆仍有细节丢失；2. 依赖摘要LLM能力
向量存储库记忆（VectorStoreRetrieverMemory）	将对话存储到向量数据库，查询时匹配最相关的k条记录	1. 支持无限长度记忆；2. 细节保留优于摘要模式	1. 需额外部署向量数据库+Embedding服务；2. 记忆效果依赖检索精度
实体记忆（ConversationEntityMemory）	提取对话中的实体（人物、地点、事件）及描述，单独存储	1. 精准记忆关键实体信息；2. 支持实体查询	1. 预设Prompt笨重，Token消耗大；2. 对LLM要求高，实用度有限

二、手动模拟记忆功能（ChatGPT/Playground）

通过手动拼接Prompt模拟摘要缓冲混合记忆，加深对记忆逻辑的理解，步骤如下：

2.1 模拟准备

工具：ChatGPT官网（无痕模式）、OpenAI Tokenizer（统计Token数）；
配置：设置最大Token限制为300，超过则触发摘要总结。

2.2 模拟流程

（1）第一轮对话

人类提问：你是由OpenAI开发的聊天机器人，请帮助用户回答问题。摘要：- 消息列表：- 用户的提问是“你好，我是慕小课，喜欢唱跳rap打篮球，你是谁？你喜欢什么呢？”
AI响应：你好，慕小课！我是ChatGPT，一个由OpenAI训练的大型语言模型。很高兴认识你！我没有像人类一样的个人喜好，但我喜欢帮助人们解决问题、回答问题和分享知识。你提到喜欢唱跳rap和打篮球，感觉你生活非常丰富多彩！你有特别喜欢的歌手或者篮球队吗？
记忆存储：消息列表添加本轮对话，摘要仍为空。

（2）第二轮对话

人类提问：你是由OpenAI开发的聊天机器人，请帮助用户回答问题。摘要：- 消息列表：[第一轮对话] 用户的提问是“你能用一段话快速解释下LLM是什么吗？”
AI响应：大型语言模型（LLM）是一种通过大量文本数据训练的人工智能模型，能够理解和生成自然语言文本。它应用于写作辅助、语言翻译、对话系统和问答系统等自然语言处理任务，表现出高水平的语言理解和生成能力。
记忆存储：消息列表添加第二轮对话，总Token数超过300，触发摘要总结。

（3）触发摘要总结

总结Prompt：你是一个强大的聊天机器人，请根据用户提供的谈话内容，总结内容，并将其添加到先前提供的摘要中，返回一个新的摘要。
当前对话内容：[第一轮对话]
生成摘要：慕小课介绍自己喜欢唱跳rap和打篮球，并问AI的喜好。AI自我介绍为ChatGPT，由OpenAI训练，喜欢帮助人们解决问题，还问慕小课是否有特别喜欢的歌手或篮球队。
更新记忆：消息列表保留第二轮对话，早期对话替换为摘要。

（4）第三轮对话（验证记忆）

人类提问：你是由OpenAI开发的聊天机器人，请帮助用户回答问题。摘要：[上述生成的摘要] 消息列表：[第二轮对话] 用户的提问是“我叫什么名字？”
AI响应：根据消息列表中的内容，你的名字是慕小课。（成功回忆早期对话中的实体信息）

三、Python+OpenAI SDK原生实现记忆功能

基于LangChain的FileChatMessageHistory组件，实现本地文件持久化的记忆功能，支持对话关闭后重新读取历史记录：

3.1 实现代码

python

import dotenv
from langchain_community.chat_message_histories import FileChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage
from openai import OpenAI

dotenv.load_dotenv()

# 1. 初始化OpenAI客户端和本地文件记忆（持久化存储到./memory.txt）
client = OpenAI()
chat_history = FileChatMessageHistory("./memory.txt")

# 2. 循环对话逻辑
while True:
    # 3. 获取用户输入
    query = input("Human: ")
    if query == "q":
        exit(0)
    
    # 4. 构建包含历史对话的Prompt
    system_prompt = (
        "你是由OpenAI开发的ChatGPT聊天机器人，可以根据对应的上下文回复用户信息，上下文里存放的是人类和您的交互。\n\n"
        f"<context>\n{chat_history.messages}\n</context>\n\n"
    )
    
    # 5. 发起流式请求
    print("AI: ", end="", flush=True)
    response = client.chat.completions.create(
        model='gpt-3.5-turbo-16k',
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": query}
        ],
        stream=True
    )
    
    # 6. 处理流式响应并存储对话
    ai_content = ""
    for chunk in response:
        content = chunk.choices[0].delta.content
        if content is None:
            break
        ai_content += content
        print(content, end="", flush=True)
    
    # 7. 保存本轮对话到本地文件
    chat_history.add_messages([HumanMessage(query), AIMessage(ai_content)])
    print("")

3.2 核心逻辑

记忆存储：FileChatMessageHistory将对话以消息对象形式存储到本地文件，支持add_messages（添加）、clear（清空）等操作；
上下文注入：每次请求时，将chat_history.messages（历史对话列表）嵌入到System Prompt中；
持久化：关闭程序后，文件保留对话记录，下次运行可直接读取。

四、LangChain记忆组件核心原理与源码解析

4.1 核心基类

LangChain的记忆组件基于两大抽象基类设计，所有具体记忆组件均继承自这两个基类：

（1）BaseChatMessageHistory（消息历史管理基类）

作用：管理对话消息的存储与操作，是记忆功能的基础；
核心方法：
- add_message/add_messages：添加单条/多条消息；
- add_user_message/add_ai_message：快捷添加用户/AI消息；
- messages：获取消息列表；
- clear：清空消息历史；
内置实现：InMemoryChatMessageHistory（内存存储）、FileChatMessageHistory（文件存储）；
第三方集成：支持Redis、Postgres、MongoDB等数据库存储（需导入langchain_community包）。

（2）BaseChatMemory（聊天记忆基类）

作用：封装记忆的加载、存储逻辑，对接Chain组件；
核心属性：
- chat_memory：关联BaseChatMessageHistory实例，负责消息存储；
- input_key/output_key：指定输入/输出的键名；
- return_messages：是否返回消息对象（默认返回字符串）；
核心方法：
- load_memory_variables：加载记忆变量，供Prompt使用；
- save_context：存储本轮对话上下文（用户输入+AI输出）；
- clear：清空记忆。

4.2 BaseChatMemory运行流程（源码核心）

python

class BaseChatMemory(BaseMemory, ABC):
    chat_memory: BaseChatMessageHistory = Field(default_factory=InMemoryChatMessageHistory)
    output_key: Optional[str] = None
    input_key: Optional[str] = None
    return_messages: bool = False

    def _get_input_output(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> Tuple[str, str]:
        # 提取用户输入和AI输出
        if self.input_key is None:
            prompt_input_key = get_prompt_input_key(inputs, self.memory_variables)
        else:
            prompt_input_key = self.input_key
        
        if self.output_key is None:
            # 自动推断输出键
            if len(outputs) == 1:
                output_key = list(outputs.keys())[0]
            elif "output" in outputs:
                output_key = "output"
            else:
                raise ValueError("请明确设置output_key")
        else:
            output_key = self.output_key
        
        return inputs[prompt_input_key], outputs[output_key]

    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
        # 存储对话上下文到chat_memory
        input_str, output_str = self._get_input_output(inputs, outputs)
        self.chat_memory.add_messages([HumanMessage(content=input_str), AIMessage(content=output_str)])

    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        # 加载记忆变量，供Prompt使用
        messages = self.chat_memory.messages
        if self.return_messages:
            return {self.memory_key: messages}
        else:
            # 转换为字符串格式
            return {self.memory_key: get_buffer_string(messages)}

五、LangChain核心记忆组件使用示例

5.1 缓冲窗口记忆（ConversationBufferWindowMemory）

仅保留最近k轮对话，适用于短期记忆场景：

python

import dotenv
from operator import itemgetter
from langchain.memory import ConversationBufferWindowMemory
from langchain_community.chat_message_histories import FileChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

# 1. 初始化记忆组件（k=2表示保留最近2轮对话）
memory = ConversationBufferWindowMemory(
    input_key="query",
    return_messages=True,  # 返回消息对象（而非字符串）
    k=2,
    chat_memory=FileChatMessageHistory("./window_memory.txt")  # 持久化到文件
)

# 2. 构建Prompt（包含历史对话占位符）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是OpenAI开发的聊天机器人，请帮助用户解决问题"),
    MessagesPlaceholder("history"),  # 历史对话占位符
    ("human", "{query}")
])

# 3. 构建链应用（LCEL表达式）
llm = ChatOpenAI(model="gpt-3.5-turbo-16k")
chain = (
    RunnablePassthrough.assign(
        history=RunnableLambda(lambda x: memory.load_memory_variables(x)["history"])
    )
    | prompt
    | llm
    | StrOutputParser()
)

# 4. 循环对话
while True:
    query = input("Human: ")
    if query == "q":
        exit(0)
    chain_input = {"query": query}
    print("AI: ", end="", flush=True)
    response = chain.stream(chain_input)
    output = ""
    for chunk in response:
        output += chunk
        print(chunk, end="", flush=True)
    # 5. 保存上下文到记忆
    memory.save_context(chain_input, {"output": output})
    print("\nhistory:", memory.load_memory_variables({}))

5.2 摘要缓冲混合记忆（ConversationSummaryBufferMemory）

短期保留完整对话，超Token限制后总结早期对话：

python

import dotenv
from operator import itemgetter
from langchain.memory import ConversationSummaryBufferMemory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

# 1. 初始化记忆组件（max_token_limit=300）
memory = ConversationSummaryBufferMemory(
    input_key="query",
    return_messages=True,
    llm=ChatOpenAI(model="gpt-3.5-turbo-16k"),  # 用于生成摘要的LLM
    max_token_limit=300
)

# 2. 构建链应用
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是强大的聊天机器人，请根据上下文回复用户问题"),
    MessagesPlaceholder("history"),
    ("human", "{query}")
])

llm = ChatOpenAI(model="gpt-3.5-turbo-16k")
chain = (
    RunnablePassthrough.assign(
        history=RunnableLambda(lambda x: memory.load_memory_variables(x)["history"])
    )
    | prompt
    | llm
    | StrOutputParser()
)

# 3. 循环对话
while True:
    query = input("Human: ")
    if query == "q":
        exit(0)
    chain_input = {"query": query}
    print("AI: ", end="", flush=True)
    response = chain.stream(chain_input)
    output = ""
    for chunk in response:
        output += chunk
        print(chunk, end="", flush=True)
    memory.save_context(chain_input, {"output": output})
    print("\nhistory:", memory.load_memory_variables({}))

5.3 实体记忆（ConversationEntityMemory）

提取对话中的实体及描述，精准记忆关键信息：

python

import dotenv
from langchain.chains.conversation.base import ConversationChain
from langchain.memory import ConversationEntityMemory
from langchain.memory.prompt import ENTITY_MEMORY_CONVERSATION_TEMPLATE
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

# 1. 初始化LLM和记忆组件
llm = ChatOpenAI(model="gpt-4o", temperature=0)
memory = ConversationEntityMemory(llm=llm)

# 2. 构建对话链
chain = ConversationChain(
    llm=llm,
    prompt=ENTITY_MEMORY_CONVERSATION_TEMPLATE,
    memory=memory
)

# 3. 发起对话并提取实体
print(chain.invoke({"input": "你好，我是慕小课，最近正在学习LangChain"}))
print(chain.invoke({"input": "我最喜欢的编程语言是Python"}))
print(chain.invoke({"input": "我住在广州"}))

# 4. 查询实体记忆
entity_store = chain.memory.entity_store.store
print("\n实体记忆:", entity_store)

输出结果（实体记忆）

python

{
    "慕小课": "慕小课最近正在学习LangChain，最喜欢的编程语言是Python，住在广州",
    "LangChain": "LangChain是一个专注于构建和连接语言模型的项目",
    "Python": "Python是一门非常受欢迎且功能强大的编程语言，特别适合数据科学、人工智能、Web开发等领域",
    "广州": "广州是中国的第三大城市，不仅是重要的经济中心，也是充满活力的科技和创新之地，拥有丰富的历史和文化"
}

六、记忆组件持久化与第三方集成

LangChain的记忆组件本身不具备持久化能力，但可通过chat_memory关联第三方存储实现持久化，支持近50+种集成方案：

6.1 主流持久化方案

存储类型	组件类	核心特点
本地文件	FileChatMessageHistory	无需额外服务，适合开发/测试
Redis	RedisChatMessageHistory	高性能，支持分布式部署
Postgres	PostgresChatMessageHistory	关系型数据库，支持复杂查询
MongoDB	MongoDBChatMessageHistory	文档型数据库，适配非结构化数据
SQLite	SQLChatMessageHistory	轻量本地数据库，无需服务端

6.2 Flask API集成记忆组件（持久化到文件）

实现带3轮记忆功能的聊天机器人API，对话历史存储到本地文件：

python

def debug(self, app_id: UUID):
    # 1. 校验输入
    req = CompletionReq()
    if not req.validate():
        return validate_error_json(req.errors)
    
    # 2. 初始化Prompt和记忆组件（k=3保留3轮对话）
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是强大的聊天机器人，能根据用户的提问回复对应的问题"),
        MessagesPlaceholder("history"),
        ("human", "{query}")
    ])
    memory = ConversationBufferWindowMemory(
        k=3,
        input_key="query",
        output_key="output",
        return_messages=True,
        chat_memory=FileChatMessageHistory("./storage/memory/chat_history.txt")
    )
    
    # 3. 构建链应用
    llm = ChatOpenAI(model="gpt-3.5-turbo-16k")
    chain = (
        RunnablePassthrough.assign(
            history=RunnableLambda(lambda x: memory.load_memory_variables(x)["history"])
        )
        | prompt
        | llm
        | StrOutputParser()
    )
    
    # 4. 调用链并存储上下文
    chain_input = {"query": req.query.data}
    content = chain.invoke(chain_input)
    memory.save_context(chain_input, {"output": content})
    
    return success_json({"content": content})

七、Chain组件与RunnableWithMessageHistory

7.1 Chain组件简介

Chain是LangChain中串联多个组件的核心工具，分为两种类型：

LCEL链（推荐）：基于LangChain Expression Language构建，支持|运算符拼接组件，简洁灵活；
遗产链：基于Chain类子类构建（如LLMChain、ConversationChain），功能冗余，未来将逐步淘汰。

基础LCEL链示例（文档对话链）

python

import dotenv
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

# 1. 构建Prompt和LLM
prompt = ChatPromptTemplate.from_messages([
    ("system", "根据<context>中的文档信息回复用户问题\n<context>{context}</context>"),
    ("human", "{query}")
])
llm = ChatOpenAI(model="gpt-3.5-turbo-16k")

# 2. 创建文档对话链
chain = create_stuff_documents_chain(llm=llm, prompt=prompt)

# 3. 传入文档和查询
documents = [
    Document(page_content="小明喜欢绿色，不喜欢黄色"),
    Document(page_content="小王喜欢粉色，也喜欢红色"),
    Document(page_content="小泽喜欢蓝色，更喜欢青色")
]
resp = chain.invoke({"query": "大家都喜欢什么颜色？", "context": documents})
print(resp)

7.2 RunnableWithMessageHistory（简化记忆链）

RunnableWithMessageHistory是LangChain提供的工具类，可自动处理记忆的加载与存储，简化链应用的记忆集成：

使用示例

python

import dotenv
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

dotenv.load_dotenv()

# 1. 定义会话存储（多用户支持）
store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = FileChatMessageHistory(f"chat_history_{session_id}.txt")
    return store[session_id]

# 2. 构建基础链
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是强大的聊天机器人，请根据用户需求回复问题"),
    MessagesPlaceholder("history"),
    ("human", "{query}")
])
llm = ChatOpenAI(model="gpt-3.5-turbo-16k")
chain = prompt | llm | StrOutputParser()

# 3. 包装记忆链
with_message_chain = RunnableWithMessageHistory(
    chain=chain,
    get_session_history=get_session_history,
    input_messages_key="query",
    history_messages_key="history"
)

# 4. 运行链（指定session_id区分用户）
while True:
    query = input("Human: ")
    if query == "q":
        exit(0)
    response = with_message_chain.stream(
        {"query": query},
        config={"configurable": {"session_id": "muxiaoke"}}  # 会话ID，区分不同用户
    )
    print("AI: ", end="", flush=True)
    for chunk in response:
        print(chunk, end="", flush=True)
    print("")

核心优势

自动记忆管理：无需手动调用load_memory_variables和save_context；
多用户支持：通过session_id区分不同用户的对话历史；
灵活集成：支持任意LCEL链，兼容各类记忆存储方案。

八、本章核心总结

记忆原理：LLM记忆的本质是“历史对话存储+Prompt注入”，LangChain通过封装组件简化这一流程；
记忆模式选择：短期精准记忆用缓冲窗口/令牌缓冲，长期记忆用摘要混合/向量存储，关键实体记忆用实体记忆；
组件使用：掌握ConversationBufferWindowMemory（常用）、ConversationSummaryBufferMemory（长期对话）的核心用法；
持久化：通过chat_memory关联文件/Redis/Postgres实现记忆持久化，适配生产环境；
简化开发：使用RunnableWithMessageHistory自动处理记忆加载与存储，减少重复代码；
工程化落地：结合Chain组件与API框架（如Flask），实现带记忆功能的企业级聊天机器人。

LangChain 记忆模块快速上手 ​

本章介绍 ​

一、LLM实现记忆功能的核心思路与常见记忆模式 ​

1.1 核心实现思路 ​

1.2 常见记忆模式 ​

二、手动模拟记忆功能（ChatGPT/Playground） ​

2.1 模拟准备 ​

2.2 模拟流程 ​

（1）第一轮对话 ​

（2）第二轮对话 ​

（3）触发摘要总结 ​

（4）第三轮对话（验证记忆） ​

三、Python+OpenAI SDK原生实现记忆功能 ​

3.1 实现代码 ​

3.2 核心逻辑 ​

四、LangChain记忆组件核心原理与源码解析 ​

4.1 核心基类 ​

（1）BaseChatMessageHistory（消息历史管理基类） ​

（2）BaseChatMemory（聊天记忆基类） ​

4.2 BaseChatMemory运行流程（源码核心） ​

五、LangChain核心记忆组件使用示例 ​

5.1 缓冲窗口记忆（ConversationBufferWindowMemory） ​

5.2 摘要缓冲混合记忆（ConversationSummaryBufferMemory） ​

5.3 实体记忆（ConversationEntityMemory） ​

输出结果（实体记忆） ​

六、记忆组件持久化与第三方集成 ​

6.1 主流持久化方案 ​

6.2 Flask API集成记忆组件（持久化到文件） ​

七、Chain组件与RunnableWithMessageHistory ​

7.1 Chain组件简介 ​

基础LCEL链示例（文档对话链） ​

7.2 RunnableWithMessageHistory（简化记忆链） ​

使用示例 ​

核心优势 ​

八、本章核心总结 ​

LangChain 记忆模块快速上手

本章介绍

一、LLM实现记忆功能的核心思路与常见记忆模式

1.1 核心实现思路

1.2 常见记忆模式

二、手动模拟记忆功能（ChatGPT/Playground）

2.1 模拟准备

2.2 模拟流程

（1）第一轮对话

（2）第二轮对话

（3）触发摘要总结

（4）第三轮对话（验证记忆）

三、Python+OpenAI SDK原生实现记忆功能

3.1 实现代码

3.2 核心逻辑

四、LangChain记忆组件核心原理与源码解析

4.1 核心基类

（1）BaseChatMessageHistory（消息历史管理基类）

（2）BaseChatMemory（聊天记忆基类）

4.2 BaseChatMemory运行流程（源码核心）

五、LangChain核心记忆组件使用示例

5.1 缓冲窗口记忆（ConversationBufferWindowMemory）

5.2 摘要缓冲混合记忆（ConversationSummaryBufferMemory）

5.3 实体记忆（ConversationEntityMemory）

输出结果（实体记忆）

六、记忆组件持久化与第三方集成

6.1 主流持久化方案

6.2 Flask API集成记忆组件（持久化到文件）

七、Chain组件与RunnableWithMessageHistory

7.1 Chain组件简介

基础LCEL链示例（文档对话链）

7.2 RunnableWithMessageHistory（简化记忆链）

使用示例

核心优势

八、本章核心总结