小白从零开始勇闯人工智能：LangChain中的Memory模块

啊巴矲

444人浏览 · 2026-03-17 20:40:39

啊巴矲 · 2026-03-17 20:40:39 发布

引言

人工智能越来越普及，比如 ChatGPT、通义千问等，它们能记住上下文，进行多轮流畅对话。但实际上，大语言模型本身并不具备记忆能力，每次调用都是独立的。要实现记忆，必须在外部存储对话历史，并在每次请求时把历史信息一起传给模型。LangChain 提供了一个强大的模块——Memory，专门用于存储和管理对话历史。

1、什么是 Memory？

在 LangChain 框架中，Memory 组件专门负责对话历史的存储与调用，其核心功能通过两种基本操作实现。首先是在读取阶段，系统会在处理用户当前输入之前，主动从 Memory 中调取历史消息，并将其与最新问题合并，共同构建出完整的提示模板。随后在写入阶段，当模型生成回复之后，系统会将本次完整的对话交互，包括用户的提问和 AI 的回答，一同记录回 Memory 中，为未来的对话积累上下文。

2、环境准备与基础调用

我们通过一个简单的例子，了解如何使用 OpenAI 客户端直接调用模型，并观察无记忆时的表现。

通过 OpenAI 客户端调用模型，首先询问“林俊杰是谁”，接着在第二个问题中直接提问“他有哪些代表作”

from openai import OpenAI

client = OpenAI(
    api_key="sk-0b717f29b6ee4852a2331cf1ffa30d4f",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
    model="qwen3.5-plus",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "林俊杰是谁？"},
    ],
)
print(completion.model_dump_json())

接着问第二个问题：

completion = client.chat.completions.create(
    model="qwen3.5-plus",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "他有哪些代表作？"},
    ],
)

由于没有传递历史消息，第二次调用时模型并不知道“他”指的是谁，这就是无记忆的缺陷。

3、初识 ConversationBufferMemory

ConversationBufferMemory 以列表形式存储所有对话。我们手动保存两轮对话：

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(return_messages=True)  # 以消息列表形式返回
memory.save_context({"input": "我的名字是cc"}, {"output": "你好，cc"})
memory.save_context({"input": "我在xxx"}, {"output": "好的，我记住了"})
memory.load_memory_variables({})  # 查看历史

4、将 Memory 集成到链中

为了让模型自动使用记忆，我们需要构建一个包含 Memory 的链。首先定义一个提示模板，其中用 MessagesPlaceholder 留出历史消息的位置：

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个AI助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{user_input}")
])

然后创建模型和链：

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="qwen-plus",
                   openai_api_key="sk-...",
                   openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1")
chain = prompt | model

手动读取 Memory 并调用链：

user_input = "你知道我的名字吗？"
history = memory.load_memory_variables({})["history"]
result = chain.invoke({"user_input": user_input, "history": history})
memory.save_context({"input": user_input}, {"output": result.content})

这样模型就能根据历史回答你的名字了。

5、使用 ConversationChain 简化操作

ConversationChain 封装了 Memory 和 LLM，自动处理历史读取和写入，大大简化代码。

在 LangChain 中使用 ConversationChain 结合 ConversationBufferMemory实现带记忆的对话。首先实例化模型和内存，并将它们传入 ConversationChain，此时链内部默认的提示模板已包含一个名为 "history"的占位符，用于动态插入历史消息。当调用 chain.invoke 时，链会自动从 memory 中加载历史并与当前输入合并，实现上下文连贯的多轮对话。此外，还可以通过自定义 ChatPromptTemplate 来替换默认模板，从而调整系统提示或对话风格，同时仍保留历史记录的自动填充功能。

from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

model = ChatOpenAI(model="qwen-plus", ...)
memory = ConversationBufferMemory(return_messages=True)
chain = ConversationChain(llm=model, memory=memory)

chain.invoke({"input": "你好，我的名字是cc"})
chain.invoke({"input": "我告诉过你我的名字，是什么？"})

ConversationChain 默认的提示模板包含了历史占位符，你也可以自定义：

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个热情的助手，喜欢用积极的语气回答问题。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])
chain = ConversationChain(llm=model, memory=memory, prompt=prompt)

6、不同类型的 Memory

1、ConversationBufferMemory

最基础的记忆，存储所有历史消息。随着对话变长，Token 消耗也会不断增加。

memory = ConversationBufferMemory(return_messages=True)
chain = ConversationChain(llm=model, memory=memory)
chain.invoke({"input": "你好，我的名字是cc"})
chain.invoke({"input": "我的名字是什么？"})

2、ConversationBufferWindowMemory

只保留最近 k 轮对话，防止历史无限增长。

memory = ConversationBufferWindowMemory(k=1, return_messages=True)
chain = ConversationChain(llm=model, memory=memory)
chain.invoke({"input": "你好，我的名字是cc"})
chain.invoke({"input": "我是一个AI开发者"})
chain.invoke({"input": "我的名字是什么？"})  # 此时已忘记名字，因为窗口只有1轮

由于 k=1，只记住上一轮对话，所以第三轮忘记名字。

3、ConversationSummaryMemory

将历史对话总结成一段话存储，而不是保存每条消息，节省 Token。

memory = ConversationSummaryMemory(return_messages=True, llm=model)
chain = ConversationChain(llm=model, memory=memory)
chain.invoke({"input": "你好，我的名字是cc"})
chain.invoke({"input": "我是一个AI开发者，你呢？"})
chain.invoke({"input": "我的名字是什么？"})  # 可以从总结中提取

查看 Memory 中的内容，会发现是一个 SystemMessage，内容是对话总结。

4、ConversationSummaryBufferMemory

结合了缓冲区和总结：先保留最近对话的原始消息，当总 Token 数超过阈值时，将最早的消息总结成摘要。

memory = ConversationSummaryBufferMemory(llm=model, max_token_limit=100, return_messages=True)
chain = ConversationChain(llm=model, memory=memory)
chain.invoke({"input": "你好，我的名字是cc"})  
chain.invoke({"input":"我是一个程序员，你呢？"})
chain.invoke({"input":"我的名字是什么？我前面说过的"})

5、ConversationTokenBufferMemory

按 Token 数量限制记忆，超过指定 Token 数则丢弃最早的消息。

memory = ConversationTokenBufferMemory(llm=model, max_token_limit=200, return_messages=True)
chain = ConversationChain(llm=model, memory=memory)
chain.invoke({"input":"你好，我的名字是cc"})
chain.invoke({"input":"我是一个程序员，你呢？"})
chain.invoke({"input":"我的名字是什么？我前面说过的"})

7、综合应用：用 Streamlit 搭建聊天机器人

最后，我们将所学知识整合到一个完整的 Streamlit 应用中。这个应用会：在侧边栏输入 API Key，用ConversationBufferMemory 存储对话历史。显示聊天界面，每次输入都调用 get_chat_response 函数，并更新界面。

from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory

def get_chat_response(prompt, memory, openai_api_key):
    model = ChatOpenAI(model="qwen3.5-plus",
                       openai_api_key=openai_api_key,
                       openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1")
    chain = ConversationChain(llm=model, memory=memory)
    response = chain.invoke({"input": prompt})
    return response["response"]

import streamlit as st
from langchain.memory import ConversationBufferMemory
from scripts import get_chat_response

st.title("💬 克隆ChatGPT")

with st.sidebar:
    openai_api_key = st.text_input("请输入OpenAI API Key：", type="password")
    st.markdown("[获取OpenAI API key](https://platform.openai.com/account/api-keys)")

if "memory" not in st.session_state:
    st.session_state["memory"] = ConversationBufferMemory(return_messages=True)
    st.session_state["messages"] = [{"role": "ai", "content": "你好，我是你的AI助手，有什么可以帮你的吗？"}]

for message in st.session_state["messages"]:
    st.chat_message(message["role"]).write(message["content"])

prompt = st.chat_input()
if prompt:
    if not openai_api_key:
        st.info("请输入你的OpenAI API Key")
        st.stop()
    st.session_state["messages"].append({"role": "human", "content": prompt})
    st.chat_message("human").write(prompt)

    with st.spinner("AI正在思考中，请稍等..."):
        response = get_chat_response(prompt, st.session_state["memory"], openai_api_key)
    msg = {"role": "ai", "content": response}
    st.session_state["messages"].append(msg)
    st.chat_message("ai").write(response)