在使用大模型(如 GPT-3.5/4o)时,你是否遇到过这样的尴尬场景:

  • 你告诉 AI“我叫张三,是一名 Python 开发者”,聊了几句后问 “我叫什么名字?”,AI 回答 “抱歉,我不记得了”;
  • 你和 AI 讨论一个复杂问题,需要多次补充信息,但每次都要重复之前的上下文;
  • 你开发的对话机器人,用户每次对话都要重新介绍自己的需求。

这些问题的根源在于大模型本身没有记忆—— 默认情况下,每次调用都是独立的,模型无法记住之前的对话内容。而 LangChain 的Memory 组件正是为解决这一痛点而生,它能让大模型拥有 “长期记忆”,实现连贯的多轮对话。


一、为什么需要 Memory?

1. 大模型的 “健忘” 本质

大模型的训练数据是静态的,且每次 API 调用都是无状态的 —— 模型无法在不同调用之间保留上下文信息。这意味着:

  • 多轮对话时,你需要手动将之前的对话历史传递给模型;
  • 对话越长,传递的上下文越多,API 调用成本越高;
  • 上下文长度超过模型的上下文窗口时,会被截断,导致模型 “失忆”。

2. Memory 的核心价值

LangChain 的 Memory 组件通过管理对话历史,让大模型实现连贯的多轮对话,带来三大优势:

  • 提升用户体验:用户无需重复输入上下文,对话更自然流畅;
  • 降低开发成本:无需手动管理对话历史,LangChain 自动处理;
  • 支持复杂任务:可以处理需要多轮交互的复杂任务(如客服对话、代码调试、创意协作)。

二、LangChain 核心 Memory 组件详解

LangChain 提供了多种 Memory 组件,适用于不同场景。以下是开发中最常用的 6 种:

1. ConversationBufferMemory:完整对话历史存储

适用场景:短对话场景,需要完整保留所有对话历史(如客服对话、简单问答)。核心原理:将所有对话历史存储在内存中,每次调用模型时将完整历史传递给模型。

代码示例:简单对话机器人
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI

# 1. 初始化Memory
memory = ConversationBufferMemory()

# 2. 创建对话链
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True  # 开启调试模式,查看传递给模型的上下文
)

# 3. 多轮对话
conversation.invoke({"input": "你好,我叫张三,是一名Python开发者。"})
conversation.invoke({"input": "我最近在学习LangChain,它是一个用于构建LLM应用的框架。"})
response = conversation.invoke({"input": "我叫什么名字?我的职业是什么?"})
print(response["response"])
输出结果:
我叫张三,是一名Python开发者。
关键说明:
  • verbose=True可以查看传递给模型的上下文,包含所有之前的对话历史;
  • 适合短对话场景,对话过长时会导致上下文窗口溢出;
  • 可以通过return_messages=True参数,将对话历史以 Message 对象返回,便于后续处理。

2. ConversationSummaryMemory:自动总结对话历史

适用场景:长对话场景,需要保留对话核心信息但又不想传递完整历史(如长时间客服对话、复杂问题讨论)。核心原理:自动总结对话历史,将总结后的内容传递给模型,减少上下文长度。

代码示例:长对话总结
from langchain.memory import ConversationSummaryMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI

# 1. 初始化Summary Memory(需要LLM用于总结)
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)
memory = ConversationSummaryMemory(llm=llm)

# 2. 创建对话链
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 3. 多轮长对话
conversation.invoke({"input": "我最近在学习LangChain,它是一个用于构建LLM应用的框架。"})
conversation.invoke({"input": "我已经掌握了基础链的使用,接下来想学习智能体(Agent)。"})
conversation.invoke({"input": "智能体可以调用外部工具,比如搜索引擎、计算器等。"})
conversation.invoke({"input": "我想开发一个能调用搜索引擎的智能助手,帮我获取实时信息。"})

# 4. 查看总结后的对话历史
print("总结后的对话历史:")
print(memory.load_memory_variables({})["history"])
输出结果:
用户最近在学习LangChain,这是一个用于构建LLM应用的框架。用户已经掌握了基础链的使用,接下来想学习智能体(Agent),智能体可以调用外部工具,比如搜索引擎、计算器等。用户想开发一个能调用搜索引擎的智能助手,帮他获取实时信息。
关键说明:
  • 总结后的对话历史长度远短于原始对话,节省上下文窗口空间;
  • 适合长对话场景,平衡上下文保留与 API 调用成本;
  • 可以通过summary_prompt参数自定义总结提示词,调整总结风格和详细程度。

3. ConversationBufferWindowMemory:最近 N 轮对话存储

适用场景:需要保留最近几轮对话,同时避免上下文过长的场景(如大多数日常对话)。核心原理:仅存储最近 N 轮对话,平衡上下文保留与性能。

代码示例:保留最近 2 轮对话
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI

# 1. 初始化Window Memory,保留最近2轮对话
memory = ConversationBufferWindowMemory(k=2)

# 2. 创建对话链
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 3. 多轮对话
conversation.invoke({"input": "你好,我叫张三。"})  # 轮1
conversation.invoke({"input": "我是一名Python开发者。"})  # 轮2
conversation.invoke({"input": "我最近在学习LangChain。"})  # 轮3
response = conversation.invoke({"input": "我叫什么名字?"})  # 轮4,此时轮1的对话已被遗忘
print(response["response"])
输出结果:
抱歉,我不记得你叫什么名字了。你能再告诉我一次吗?
关键说明:
  • k=2表示保留最近 2 轮对话,轮 4 时轮 1 的对话已被移除;
  • 适合大多数日常对话场景,既保留必要上下文,又避免过长;
  • 可以通过return_messages=True参数,将对话历史以 Message 对象返回。

4. ConversationKGMemory:基于知识图谱的对话记忆

适用场景:需要理解对话中的实体和关系的复杂场景(如知识问答、客户需求分析)。核心原理:将对话中的实体和关系提取到知识图谱中,存储结构化的对话信息。

代码示例:知识图谱记忆
from langchain.memory import ConversationKGMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI

# 1. 初始化KG Memory
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)
memory = ConversationKGMemory(llm=llm)

# 2. 创建对话链
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 3. 对话中包含实体和关系
conversation.invoke({"input": "我叫张三,我在阿里巴巴工作,职位是高级工程师。"})
conversation.invoke({"input": "我负责开发基于LangChain的AI应用。"})
response = conversation.invoke({"input": "张三在哪里工作?他的职位是什么?"})
print(response["response"])
输出结果:
张三在阿里巴巴工作,职位是高级工程师。
关键说明:
  • KG Memory 会自动提取对话中的实体(张三、阿里巴巴)和关系(工作于、职位是);
  • 适合需要理解实体和关系的复杂对话场景;
  • 可以通过kg参数自定义知识图谱存储(默认使用 LangChain 内置的知识图谱)。

5. VectorStoreRetrieverMemory:基于向量检索的记忆

适用场景:需要在大量对话历史中检索相关信息的场景(如客服历史对话分析、用户偏好记忆)。核心原理:将对话历史存储到向量数据库中,通过向量检索找到与当前问题最相关的对话内容。

代码示例:向量检索记忆
from langchain.memory import VectorStoreRetrieverMemory
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI

# 1. 创建向量存储
embeddings = OpenAIEmbeddings()
vector_store = FAISS.from_texts([""], embeddings)  # 初始化空向量存储
retriever = vector_store.as_retriever(search_kwargs={"k": 1})

# 2. 初始化VectorStoreRetrieverMemory
memory = VectorStoreRetrieverMemory(retriever=retriever)

# 3. 创建对话链
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 4. 对话与检索
conversation.invoke({"input": "我叫张三,喜欢打篮球和编程。"})
conversation.invoke({"input": "我最近在学习LangChain。"})
response = conversation.invoke({"input": "张三有什么爱好?"})
print(response["response"])
输出结果:
张三喜欢打篮球和编程。
关键说明:
  • 向量检索记忆会自动将对话历史存储到向量数据库中,检索时找到最相关的对话内容;
  • 适合大量对话历史的场景,避免传递完整历史;
  • 可以结合 FAISS、Chroma 等向量数据库使用。

6. CombinedMemory:组合多种 Memory

适用场景:需要同时使用多种 Memory 的复杂场景(如既需要保留最近对话,又需要总结长对话)。核心原理:将多种 Memory 组合使用,发挥各自的优势。

代码示例:组合 BufferWindow 和 Summary Memory
from langchain.memory import ConversationBufferWindowMemory, ConversationSummaryMemory, CombinedMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI

# 1. 初始化两种Memory
window_memory = ConversationBufferWindowMemory(k=2, memory_key="recent_history")
summary_memory = ConversationSummaryMemory(llm=ChatOpenAI(), memory_key="summary_history")

# 2. 组合Memory
memory = CombinedMemory(memories=[window_memory, summary_memory])

# 3. 创建对话链,需要指定输入键
conversation = ConversationChain(
    llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7),
    memory=memory,
    verbose=True,
    input_key="input"
)

# 4. 多轮对话
conversation.invoke({"input": "你好,我叫张三,是一名Python开发者。"})
conversation.invoke({"input": "我最近在学习LangChain,它是一个用于构建LLM应用的框架。"})
conversation.invoke({"input": "我已经掌握了基础链的使用,接下来想学习智能体。"})
response = conversation.invoke({"input": "我叫什么名字?我在学习什么?"})
print(response["response"])
输出结果:
你叫张三,你在学习LangChain,这是一个用于构建LLM应用的框架。你已经掌握了基础链的使用,接下来想学习智能体。
关键说明:
  • 组合 Memory 可以同时保留最近对话和总结历史,兼顾上下文保留和性能;
  • 适合复杂对话场景,需要根据需求选择合适的 Memory 组合;
  • 注意为每个 Memory 指定唯一的memory_key,避免冲突。

三、实战案例:构建智能客服机器人

需求分析

开发一个智能客服机器人,具备以下功能:

  1. 支持多轮对话,能记住用户的上下文信息;
  2. 能回答产品相关问题(基于知识库);
  3. 能记住用户的订单信息和偏好;
  4. 复杂问题自动转人工。

实现步骤

from langchain.memory import ConversationBufferMemory
from langchain.chains import RetrievalQAWithSourcesChain, ConversationChain
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import PyPDFLoader

# 1. 加载知识库(产品手册)
loader = PyPDFLoader("product_manual.pdf")
documents = loader.load()
embeddings = OpenAIEmbeddings()
vector_store = FAISS.from_documents(documents, embeddings)
retriever = vector_store.as_retriever(search_kwargs={"k": 2})

# 2. 初始化Memory
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

# 3. 创建对话链和检索链
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)
qa_chain = RetrievalQAWithSourcesChain.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    memory=memory,
    verbose=True
)

# 4. 客服对话逻辑
def customer_service_bot(user_input):
    # 简单判断是否需要转人工
    if "转人工" in user_input or "人工客服" in user_input:
        return "已为您转人工客服,请稍等..."
    
    # 调用检索链回答问题
    result = qa_chain.invoke({"question": user_input})
    return result["answer"]

# 5. 测试对话
print("智能客服机器人:您好,请问有什么可以帮您的?")
while True:
    user_input = input("您:")
    if user_input.lower() in ["再见", "退出"]:
        print("智能客服机器人:再见,欢迎再次光临!")
        break
    response = customer_service_bot(user_input)
    print(f"智能客服机器人:{response}")

案例说明

  • 机器人使用 ConversationBufferMemory 记住用户的上下文信息;
  • 使用 RetrievalQAWithSourcesChain 从知识库中检索产品相关信息;
  • 支持简单的意图识别,复杂问题自动转人工;
  • 对话自然流畅,用户无需重复输入上下文。

四、高级技巧:自定义 Memory 组件

对于复杂场景,LangChain 支持自定义 Memory 组件,只需继承BaseMemory类并实现load_memory_variablessave_contextclear方法。

示例:自定义文件存储 Memory

from langchain.memory import BaseMemory
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from typing import Dict, List
import json

class FileMemory(BaseMemory):
    def __init__(self, file_path: str = "memory.json"):
        self.file_path = file_path
        self.load_memory_from_file()

    def load_memory_from_file(self):
        try:
            with open(self.file_path, "r") as f:
                messages_data = json.load(f)
                self.chat_history = [
                    HumanMessage(content=msg["content"]) if msg["type"] == "human" else AIMessage(content=msg["content"])
                    for msg in messages_data
                ]
        except FileNotFoundError:
            self.chat_history = []

    def save_memory_to_file(self):
        messages_data = [
            {"type": "human", "content": msg.content} if isinstance(msg, HumanMessage) else {"type": "ai", "content": msg.content}
            for msg in self.chat_history
        ]
        with open(self.file_path, "w") as f:
            json.dump(messages_data, f)

    def load_memory_variables(self, inputs: Dict) -> Dict[str, List[BaseMessage]]:
        return {"chat_history": self.chat_history}

    def save_context(self, inputs: Dict, outputs: Dict) -> None:
        self.chat_history.append(HumanMessage(content=inputs["input"]))
        self.chat_history.append(AIMessage(content=outputs["response"]))
        self.save_memory_to_file()

    def clear(self) -> None:
        self.chat_history = []
        self.save_memory_to_file()

    @property
    def memory_variables(self) -> List[str]:
        return ["chat_history"]

# 使用自定义Memory
memory = FileMemory()
conversation = ConversationChain(
    llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7),
    memory=memory,
    verbose=True
)

conversation.invoke({"input": "你好,我叫张三。"})

五、最佳实践与注意事项

  1. 选择合适的 Memory:根据对话长度和复杂度选择 Memory,短对话用 ConversationBufferMemory,长对话用 ConversationSummaryMemory 或 VectorStoreRetrieverMemory;
  2. 控制上下文长度:避免传递过长的对话历史,超过模型上下文窗口会导致截断或 API 调用失败;
  3. 持久化 Memory:生产环境中,需要将 Memory 持久化到数据库或文件中,避免应用重启后丢失对话历史;
  4. 错误处理:处理对话历史过大导致的 API 调用失败,可通过截断历史或使用 SummaryMemory 解决;
  5. 结合 RetrievalQA:将 Memory 与 RetrievalQA 结合,实现既能记住用户上下文,又能回答知识库问题的智能应用。

六、总结

LangChain 的 Memory 组件是实现连贯多轮对话的核心工具,它解决了大模型 “健忘” 的痛点,让 AI 应用更自然、更智能。

在实际开发中,需要根据场景选择合适的 Memory 组件:

  • 短对话:ConversationBufferMemory;
  • 长对话:ConversationSummaryMemory 或 VectorStoreRetrieverMemory;
  • 复杂场景:CombinedMemory 或自定义 Memory。

掌握 Memory 组件的使用,是从 “简单对话机器人” 到 “智能 AI 助手” 的关键一步,让你的 AI 应用真正具备 “记忆” 能力,提供更优质的用户体验!

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐