【个人主页：玄同765】

大语言模型（LLM）开发工程师｜中国传媒大学·数字媒体技术（智能交互与游戏设计）

深耕领域：大语言模型开发 / RAG知识库 / AI Agent落地 / 模型微调

技术栈：Python / LangChain/RAG（Dify+Redis+Milvus）| SQL/NumPy | FastAPI+Docker ️

工程能力：专注模型工程化部署、知识库构建与优化，擅长全流程解决方案 

     

「让AI交互更智能，让技术落地更高效」

欢迎技术探讨/项目合作！ 关注我，解锁大模型与智能交互的无限可能！

前置准备（必看，避免运行报错）

在开始实践本文代码前，需安装完整依赖并配置环境变量，确保所有功能正常运行：

# 核心依赖：LangChain核心、OpenAI集成、环境变量管理
pip install langchain langchain-core langchain-openai python-dotenv
# 社区扩展依赖：包含实体记忆、FAISS向量存储
pip install langchain-community
# FAISS向量存储底层库（CPU版本，推荐测试/生产环境；GPU版本需安装faiss-gpu并匹配CUDA）
pip install faiss-cpu

配置环境变量：在项目根目录创建.env文件，填入 OpenAI API 密钥（用于模型调用、嵌入生成）：

OPENAI_API_KEY=your_openai_api_key_here

引言

在构建对话机器人、智能客服、个人助手等大模型应用时，“记住用户之前说过的内容” 是核心需求 —— 比如用户先提到 “我是一名 Python 后端开发者”，后续问 “推荐一些进阶学习资源”，模型需要结合身份给出精准建议。LangChain 的 Memory 模块正是为解决这一问题而生，v1.0 + 版本对其进行了全面升级：所有 Memory 实现均兼容 Runnable 接口，支持异步调用、会话隔离、持久化存储，彻底解决了旧版本中记忆与工作流耦合度高、扩展性差的问题。

一、Memory 模块核心设计

LangChain v1.0 + 中，所有「聊天场景」的 Memory 实现均继承自BaseChatMemory抽象类（非聊天场景的记忆基于更基础的BaseMemory抽象类），必须实现以下 3 个核心方法：

方法	功能描述
load_memory_variables(inputs: dict) -> dict	加载对话历史，返回包含记忆变量的字典（如{"history": [HumanMessage, AIMessage]}）
save_context(inputs: dict, outputs: dict) -> None	保存当前对话轮次的上下文（用户输入 + 模型输出）
clear() -> None	清除所有对话历史

在 LCEL 体系中，Memory 通常通过RunnableWithMessageHistory与 Chain 集成，自动处理记忆的加载、保存与会话隔离，无需手动调用上述方法，这是 v1.0 + 版本的核心用法。

关键参数对应关系说明（新手必看）：

• Memory的memory_key：记忆变量的名称，用于标识返回的历史数据。
• Prompt的MessagesPlaceholder（或字符串占位符）的variable_name：必须与memory_key一致，用于接收历史数据。
• RunnableWithMessageHistory的history_messages_key：必须与上述两者一致，用于传递历史数据；input_messages_key指定用户输入的字典 key。

二、核心 Memory 类型全解析

1. 基础全量记忆：ConversationBufferMemory

适用场景：短对话场景（如单轮咨询、简单闲聊），需要完整保留所有对话历史。核心特性：

• 将所有对话消息以列表形式存储在内存中；
• 支持返回ChatMessage对象或字符串格式的历史；
• 轻量、无额外计算开销，适合上下文长度较短的场景。

关键参数说明：

• return_messages=True：返回List[BaseMessage]（HumanMessage/AIMessage），适配MessagesPlaceholder占位符；
• return_messages=False（默认）：返回字符串拼接的历史，适配{history}字符串占位符。

from langchain_core.memory import ConversationBufferMemory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 1. 初始化基础组件
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手，严格根据对话历史回答问题。"),
    MessagesPlaceholder(variable_name="history"),  # 与memory_key一致
    ("human", "{question}"),
])
llm = ChatOpenAI(model="gpt-4o-mini")
output_parser = StrOutputParser()
base_chain = prompt | llm | output_parser

# 2. 初始化全量记忆
memory = ConversationBufferMemory(
    return_messages=True,  # 返回ChatMessage对象，适配MessagesPlaceholder
    memory_key="history"  # 与Prompt中的variable_name、后续history_messages_key一致
)

# 3. 会话存储（生产环境可替换为Redis/数据库）
session_store = {}
def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
    if session_id not in session_store:
        session_store[session_id] = InMemoryChatMessageHistory()
    return session_store[session_id]

# 4. 集成Memory到Chain
chain_with_memory = RunnableWithMessageHistory(
    base_chain,
    get_session_history=get_session_history,
    input_messages_key="question",  # 用户输入的key
    history_messages_key="history"  # 对话历史的key，与上述配置一致
)

# 5. 多轮对话测试
response1 = chain_with_memory.invoke(
    {"question": "你好，我叫张三，是一名Python后端开发者。"},
    config={"configurable": {"session_id": "user_zhangsan"}}
)
print("AI回复1：", response1)

response2 = chain_with_memory.invoke(
    {"question": "我之前提到过我的名字和职业是什么？"},
    config={"configurable": {"session_id": "user_zhangsan"}}
)
print("AI回复2：", response2)

2. 窗口记忆：ConversationBufferWindowMemory

适用场景：中等长度对话场景，需要限制记忆的轮次，避免上下文过长导致 token 消耗过高或模型注意力分散。核心特性：

• 仅保留最近 N 轮对话，超出部分自动丢弃；
• 平衡上下文相关性与资源消耗，适合大多数对话场景；
• 支持自定义窗口大小（k参数）。

代码示例（仅展示记忆初始化，后续集成与上一致）：

from langchain_core.memory import ConversationBufferWindowMemory

# 初始化窗口记忆，仅保留最近2轮对话
memory = ConversationBufferWindowMemory(
    k=2,  # 窗口大小，保留最近2轮对话（用户提问+AI回复为1轮）
    return_messages=True,
    memory_key="history"
)

3. 令牌限制记忆：ConversationTokenBufferMemory

适用场景：需要精确控制对话历史 token 数量的场景，比按轮次限制更精准，适合 token 预算有限的生产环境。核心特性：

• 按 token 数量限制对话历史长度，而非轮次；
• 支持自定义 token 计数模型（如 OpenAI 的gpt-4o-mini）；
• 当历史超过 token 限制时，自动删除最早的消息。

关键说明：

• max_token_limit仅限制对话历史的总 token 数，不包含系统 Prompt、当前用户输入；
• token 计数依赖传入 LLM 对应的 tokenizer（如 OpenAI 模型使用tiktoken），不同模型计数规则略有差异。

代码示例：

from langchain_core.memory import ConversationTokenBufferMemory
from langchain_openai import ChatOpenAI

# 初始化LLM（用于token计数）
llm = ChatOpenAI(model="gpt-4o-mini")

# 初始化令牌限制记忆，最多保留1000个token的历史
memory = ConversationTokenBufferMemory(
    llm=llm,  # 用于计数token的LLM
    max_token_limit=1000,  # 最大token数量
    return_messages=True,
    memory_key="history"
)

4. 摘要记忆：ConversationSummaryMemory

适用场景：超长多轮对话场景（如客服咨询、复杂项目讨论），需要大幅压缩对话历史的 token 占用。核心特性：

• 通过 LLM 将历史对话摘要为一段简洁文本，替代原始对话列表；
• 显著减少上下文长度，降低 token 消耗；
• 支持自定义摘要提示词，优化摘要质量。

关键说明：

• return_messages=True时，摘要会被封装为单个AIMessage返回，适配MessagesPlaceholder；
• 聊天模型（ChatOpenAI）推荐使用ChatPromptTemplate自定义摘要提示词，提升摘要质量。

代码示例：

from langchain_core.memory import ConversationSummaryMemory
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_openai import ChatOpenAI

# 初始化LLM（用于生成摘要）
llm = ChatOpenAI(model="gpt-4o-mini")

# 方案1：文本模型专属摘要Prompt（适配OpenAI Completion API）
summary_prompt = PromptTemplate.from_template(
    "请将以下对话历史摘要为一段简洁的文本，保留关键信息：\n{history}"
)

# 方案2：聊天模型专属摘要Prompt（推荐，适配ChatOpenAI）
summary_chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名对话摘要助手，需将以下对话历史压缩为一段简洁文本，保留所有核心信息（用户身份、关键诉求、解决方案）。"),
    ("human", "{history}")
])

# 初始化摘要记忆
memory = ConversationSummaryMemory(
    llm=llm,  # 用于生成摘要的LLM
    summary_prompt=summary_chat_prompt,  # 自定义摘要提示词（二选一即可）
    return_messages=True,
    memory_key="history"
)

5. 实体跟踪记忆：ConversationEntityMemory

适用场景：客服系统、CRM 辅助工具、实体跟踪场景，比如用户提到的 “订单编号 12345”“收件地址北京朝阳区”。核心特性：

• 通过 LLM 自动提取对话中的实体及其属性，并维护实体知识库；
• 后续对话可直接引用实体信息，无需重复输入；
• 支持自定义实体提取提示词，提升实体识别准确率。

关键说明：

• 该类位于langchain-community，需额外安装对应依赖；
• 推荐使用OpenAI文本补全模型（如gpt-3.5-turbo-instruct），Prompt 写法更简洁，聊天模型需适配专属消息格式。

代码示例：

from langchain_community.memory import ConversationEntityMemory
from langchain_openai import OpenAI, ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

# 初始化LLM（二选一，推荐文本补全模型）
# 方案1：文本补全模型（适配实体提取，Prompt简洁）
llm = OpenAI(model="gpt-3.5-turbo-instruct")
# 方案2：聊天模型（需自定义实体提取Prompt，否则效果不佳）
# llm = ChatOpenAI(model="gpt-4o-mini")

# 初始化实体记忆
memory = ConversationEntityMemory(
    llm=llm,
    return_messages=True,
    memory_key="history"
)

# 保存对话到记忆
memory.save_context(
    inputs={"question": "我的订单编号是12345，状态是什么？"},
    outputs={"answer": "你的订单12345已发货，预计明天送达。"}
)

# 加载记忆时，会包含实体信息（history：对话历史，entities：实体知识库）
print(memory.load_memory_variables({}))

6. 向量检索记忆：VectorStoreRetrieverMemory

适用场景：超长篇对话或需要关联历史中特定片段的场景（如技术支持、法律咨询），比如用户问 “之前说的那个 API 的参数怎么用？”，需要精准检索到对应的历史片段。核心特性：

• 将对话历史存入向量数据库，当用户发起新查询时，通过向量检索找到最相关的历史片段；
• 无需加载全部历史，仅传递最相关的片段，减少 token 消耗；
• 支持自定义检索参数（如返回数量、相似度阈值）。

关键说明：

• 向量数据默认存储在进程内存中，服务重启后丢失，需手动实现持久化；
• load_memory_variables传入的question会作为向量检索查询语句，匹配最相关的历史。

代码示例（含持久化方案，适配 LangChain v1.0+）：

from langchain_core.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from dotenv import load_dotenv

load_dotenv()

# 1. 初始化向量存储与检索器
embeddings = OpenAIEmbeddings()
# 纯空初始化向量库（符合官方规范）
vector_store = FAISS.from_texts([], embeddings)
# 配置检索参数：返回最相关的1条历史
retriever = vector_store.as_retriever(search_kwargs={"k": 1})

# 2. 初始化向量检索记忆
memory = VectorStoreRetrieverMemory(
    retriever=retriever,
    memory_key="relevant_history",  # 记忆变量名
    return_messages=True
)

# 3. 保存对话到向量存储
memory.save_context(
    inputs={"question": "FastAPI中如何定义POST接口？"},
    outputs={"answer": "可以用@app.post('/path')装饰器，结合Pydantic模型做参数校验。"}
)

# 4. 向量库持久化（保存到本地，避免重启丢失）
vector_store.save_local("faiss_memory_index")

# 5. 从本地加载向量库（生产环境需注意安全风险）
loaded_vector_store = FAISS.load_local(
    "faiss_memory_index",
    embeddings,
    allow_dangerous_deserialization=True  # 允许反序列化，生产环境推荐加密存储
)
loaded_retriever = loaded_vector_store.as_retriever(search_kwargs={"k": 1})
loaded_memory = VectorStoreRetrieverMemory(
    retriever=loaded_retriever,
    memory_key="relevant_history",
    return_messages=True
)

# 6. 检索相关历史（传入question作为检索查询语句）
print(loaded_memory.load_memory_variables({"question": "FastAPI的POST接口怎么写？"}))

7. 组合记忆：MultiMemory

适用场景：需要同时使用多种记忆类型的复杂场景，比如同时保留全量历史和实体信息，或同时使用摘要记忆和向量检索记忆。核心特性：

• 支持组合多个 Memory 实例；
• 加载记忆时会合并所有记忆的输出；
• 灵活适配复杂业务需求。

关键注意点：

• 多个 Memory 的memory_key不能重复，否则后续 Memory 结果会覆盖前面的；
• 所有 Memory 需保持一致的return_messages配置，避免格式冲突。

代码示例：

from langchain_core.memory import MultiMemory, ConversationBufferMemory
from langchain_community.memory import ConversationEntityMemory
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

# 1. 初始化LLM（用于实体记忆）
llm = ChatOpenAI(model="gpt-4o-mini")

# 2. 初始化两种不同记忆（memory_key不重复）
buffer_memory = ConversationBufferMemory(
    return_messages=True,
    memory_key="full_history"  # 全量对话历史
)
entity_memory = ConversationEntityMemory(
    llm=llm,
    return_messages=True,
    memory_key="entity_info"  # 实体信息（不与full_history重复）
)

# 3. 组合记忆
multi_memory = MultiMemory(memories=[buffer_memory, entity_memory])

# 4. 测试保存对话
multi_memory.save_context(
    inputs={"question": "我叫李四，订单67890还没到。"},
    outputs={"answer": "李四你好，订单67890正在派送中，预计1小时内送达。"}
)

# 5. 加载合并后的记忆结果
print(multi_memory.load_memory_variables({}))

8. 自定义记忆：继承 BaseChatMemory

适用场景：内置 Memory 无法满足业务专属需求的场景，比如需要过滤特定类型的对话历史、与内部知识库集成等。核心特性：

• 完全自定义记忆的加载、保存和清除逻辑；
• 可实现复杂的业务规则和性能优化；
• 兼容 LCEL 体系，可与其他组件无缝集成。

关键说明：

• 基于 Pydantic v2 构建，需开启arbitrary_types_allowed支持自定义类型字段；
• 需实现load_memory_variables、save_context、clear三个核心方法。

代码示例（含完整使用流程，适配 LangChain v1.0+）：

from langchain_core.memory import BaseChatMemory
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from typing import Dict, List
from dotenv import load_dotenv

load_dotenv()

class CustomBusinessMemory(BaseChatMemory):
    """自定义业务记忆：仅保留与订单相关的对话历史，过滤无关内容"""
    chat_memory: BaseChatMessageHistory
    memory_key: str = "history"
    
    # 补充Pydantic配置，支持BaseChatMessageHistory自定义类型
    class Config:
        arbitrary_types_allowed = True

    def load_memory_variables(self, inputs: Dict) -> Dict:
        """加载记忆：过滤出包含「订单」关键词的对话消息"""
        filtered_history = [
            msg for msg in self.chat_memory.messages
            if "订单" in msg.content
        ]
        return {self.memory_key: filtered_history}

    def save_context(self, inputs: Dict, outputs: Dict) -> None:
        """保存记忆：仅存储与订单相关的对话"""
        question = inputs.get("question", "")
        answer = outputs.get("answer", "")
        if "订单" in question or "订单" in answer:
            self.chat_memory.add_user_message(question)
            self.chat_memory.add_ai_message(answer)

    def clear(self) -> None:
        """清除所有记忆"""
        self.chat_memory.clear()

# 自定义记忆使用示例
if __name__ == "__main__":
    # 1. 初始化聊天历史存储
    chat_history = InMemoryChatMessageHistory()
    
    # 2. 初始化自定义记忆
    custom_memory = CustomBusinessMemory(chat_memory=chat_history)
    
    # 3. 测试保存订单相关对话（会被正常存储）
    custom_memory.save_context(
        inputs={"question": "我的订单12345还没送达，请问怎么回事？"},
        outputs={"answer": "订单12345正在派送中，预计1小时内送达，可联系快递员138xxxxxxx查询。"}
    )
    
    # 4. 测试保存非订单对话（会被过滤，不存储）
    custom_memory.save_context(
        inputs={"question": "今天天气怎么样？"},
        outputs={"answer": "今天晴天，气温25度。"}
    )
    
    # 5. 加载记忆（仅返回订单相关对话）
    print("自定义记忆加载结果：")
    print(custom_memory.load_memory_variables({}))
    
    # 6. 清除记忆（可选）
    # custom_memory.clear()

三、Memory 选型指南：快速匹配业务需求

为了帮助开发者快速选型，下面整理了各类 Memory 的适用场景、优点、缺点及选型建议：

Memory 类型	适用场景	优点	缺点	选型建议
ConversationBufferMemory	短对话、简单闲聊	轻量、无额外计算开销、完整保留历史	上下文过长时 token 消耗高、模型响应慢	短对话场景优先选择
ConversationBufferWindowMemory	中等长度对话、需要限制历史轮次	平衡上下文相关性与资源消耗、配置简单	无法精确控制 token 数量、当重要信息超出窗口轮次时可能丢失重要信息	大多数对话场景优先选择
ConversationTokenBufferMemory	生产环境、token 预算有限的场景	精确控制 token 数量、避免超预算	需要依赖 LLM 进行 token 计数、有轻微计算开销	生产环境优先选择
ConversationSummaryMemory	超长多轮对话、复杂项目讨论	大幅压缩 token 消耗、保留关键信息	摘要生成有计算开销、可能丢失细节信息	长对话场景优先选择
ConversationEntityMemory	客服系统、CRM 辅助工具、实体跟踪场景	自动提取实体信息、支持实体引用	实体提取有计算开销、依赖 LLM 的实体识别能力	需要跟踪实体信息的场景优先选择
VectorStoreRetrieverMemory	超长篇对话、技术支持、法律咨询	精准检索相关历史、减少 token 消耗	分布式向量数据库有部署成本、检索有计算开销	超长对话、需要精准引用历史的场景优先选择
MultiMemory	复杂业务场景、需要同时使用多种记忆	灵活组合多种记忆、适配复杂需求	配置复杂、可能增加计算开销	复杂业务场景优先选择
自定义 Memory	业务专属需求、特殊规则场景	完全自定义逻辑、适配特殊需求	开发成本高、需要维护自定义代码	内置 Memory 无法满足需求时选择

四、生产级落地注意事项

1. 会话隔离与持久化存储

• 会话隔离：生产环境下必须为每个用户维护独立的对话历史，避免不同用户的历史混淆，可通过RunnableWithMessageHistory的session_id实现，推荐使用用户唯一标识（如用户 ID、手机号哈希值）作为session_id。
• 持久化存储：避免使用内存存储（InMemoryChatMessageHistory），服务重启后数据丢失，推荐以下内置方案：
  1. Redis 持久化（推荐高并发场景，支持 TTL 自动过期）：

     from langchain_community.chat_message_histories import RedisChatMessageHistory
     
     # 初始化Redis会话存储，TTL设置为86400秒（24小时）
     def get_redis_session_history(session_id: str) -> RedisChatMessageHistory:
         return RedisChatMessageHistory(
             session_id=session_id,
             redis_url="redis://localhost:6379",
             ttl=86400
         )
     

  2. 数据库持久化（支持复杂查询，如 PostgreSQL/SQLite）：

     from langchain_community.chat_message_histories import SQLChatMessageHistory
     
     # 初始化SQLite会话存储（生产环境推荐PostgreSQL）
     def get_sql_session_history(session_id: str) -> SQLChatMessageHistory:
         return SQLChatMessageHistory(
             session_id=session_id,
             connection_string="sqlite:///chat_history.db"
         )
     

2. 记忆清理策略

即使使用持久化存储，也需要设置记忆过期策略，避免历史数据无限累积导致存储成本过高、检索性能下降：

• TTL 过期策略：Redis 存储直接在初始化时设置ttl，自动清理过期会话，无需手动干预。
• 定期清理脚本：数据库存储编写定时脚本（如使用 Airflow、Linux Crontab），清理超过阈值的历史数据（如 30 天前、轮次超 100 轮）。
• 主动清理接口：提供管理员 API 接口，支持手动清理特定用户、时间范围的历史数据，应对特殊业务需求。

3. 性能优化

• 摘要与向量检索结合：超长对话场景，使用ConversationSummaryMemory提供全局上下文，VectorStoreRetrieverMemory提供精准细节，平衡 token 消耗与回答准确性。
• 异步操作：所有 Memory 均支持异步方法（如ainvoke、async_save_context），高并发场景优先使用异步调用提升系统吞吐量，示例：

  import asyncio
  from langchain_core.runnables.history import RunnableWithMessageHistory
  
  # 异步多轮对话示例
  async def async_chat_demo():
      chain_with_memory = RunnableWithMessageHistory(
          base_chain,
          get_session_history=get_redis_session_history,
          input_messages_key="question",
          history_messages_key="history"
      )
      response = await chain_with_memory.ainvoke(
          {"question": "我的订单67890进度如何？"},
          config={"configurable": {"session_id": "user_lisi"}}
      )
      print("AI异步回复：", response)
  
  # 运行异步函数
  asyncio.run(async_chat_demo())
  

• 缓存优化：对频繁访问的历史片段（如用户近期实体信息、高频问题回复）进行本地缓存（如使用cachetools），减少向量检索或摘要生成的计算开销。

4. 错误处理

• 记忆加载失败：添加 try-except 异常捕获逻辑，当 Redis / 数据库连接失败、向量库加载异常时，返回空历史或默认回复，避免服务崩溃。
• 实体识别错误：对ConversationEntityMemory提取的实体进行规则过滤（如订单编号格式校验）或人工复核，提升实体信息准确性。
• token 超限防护：生产环境中监控对话历史的 token 数量，当接近模型最大上下文长度时，主动触发摘要压缩或历史清理，避免模型调用报错。

五、总结

LangChain v1.0 + 的 Memory 模块通过与 LCEL 体系的深度集成，提供了从基础全量记忆到生产级向量检索记忆的全栈解决方案，核心价值在于：

1. 全面覆盖场景：8 大类 Memory 适配从短对话到超长对话、从简单闲聊到复杂实体跟踪的多数业务需求。
2. 生产级特性：支持会话隔离、持久化存储、异步操作、token 精确控制，为企业级应用提供稳定性和性能支撑。
3. 灵活扩展：支持组合多种记忆类型和自定义记忆，适配复杂业务场景，降低二次开发成本。
4. 易用性：与 LCEL 体系无缝集成，无需手动处理历史传递，提供可参考的实践代码，提升开发效率。

掌握各类 Memory 的适用场景和核心特性，是构建生产级大模型应用的核心基础。本文所有代码均基于 LangChain v1.0 + 规范进行梳理与实践，可作为对话机器人、客服系统、智能助手等实际场景的参考实现，开发者可根据自身业务环境、依赖版本与核心需求进行适配调整与优化。