掘金 人工智能 6小时前
LangChain 设计原理分析⁶ | Memory 系统设计:如何构建上下文感知的链
index_new5.html
../../../zaker_core/zaker_tpl_static/wap/tpl_guoji1.html

 

LangChain的Memory系统是构建支持上下文记忆和多轮对话智能链条的关键。它解决了大型语言模型(LLM)默认无状态的限制,通过记录、拼接和管理历史对话,并将相关信息注入Prompt,从而维持对话的连续性和上下文理解能力。文章详细介绍了ConversationBufferMemory、ConversationBufferWindowMemory、ConversationSummaryMemory和ConversationSummaryBufferMemory等多种Memory类型,阐述了它们各自的功能、适用场景以及如何通过RunnableWithMessageHistory高效地实现记忆管理,并提供了丰富的示例代码,为开发者提供了构建更智能、更具交互性的AI应用的实用指南。

💡 Memory系统是LLM实现多轮对话的关键:LLM本身是无状态的,每次交互仅基于当前输入。Memory系统通过记录、拼接历史对话,并将关键信息注入Prompt,使LLM能够理解并响应上下文,从而实现连贯的多轮对话。

📊 多样化的Memory类型满足不同场景需求:文章介绍了四种主要的Memory类型:ConversationBufferMemory(完整历史)、ConversationBufferWindowMemory(滑动窗口)、ConversationSummaryMemory(自动摘要)和ConversationSummaryBufferMemory(摘要+原文混合)。用户可以根据对话长度、上下文窗口限制和性能需求选择最适合的Memory策略。

🛠️ RunnableWithMessageHistory提供便捷的记忆管理:LangChain的RunnableWithMessageHistory类能够无缝封装任何Runnable,自动处理历史消息的注入和记录,简化了开发者的记忆管理工作。它只需要配置会话历史的获取方式和消息键名,即可实现记忆功能的自动化。

🚀 灵活的Memory扩展与实际部署:除了内置的Memory类型,LangChain还支持自定义Memory实现,允许开发者将消息持久化到数据库、Redis等存储中,以适应实际部署和大规模应用的需求。通过与LangGraph等框架结合,可以构建高度可扩展和多用户的智能对话系统。

本文深入介绍 LangChain 的 Memory 系统机制,包括 ConversationMemory 类型、历史对话拼接与自动总结功能,助你搭建支持上下文记忆与多轮对话的智能链条。


一、为什么需要 Memory 系统?


二、Memory 类型概览

类型功能适用场景
ConversationBufferMemory存储完整对话历史短会话或上下文窗口足够
ConversationBufferWindowMemory仅保留最后 k 条消息对话较长时防止上下文膨胀
ConversationSummaryMemory使用 LLM 自动生成 summary,拼接到 prompt对话长、希望压缩上下文时
ConversationSummaryBufferMemory综合 buffer + summarization 多策略混合平衡记忆与上下文长度管理

这些 Memory 类型虽已逐步迁移至 LangGraph,但源码支持仍可使用,适合当前链式记忆机制。


三、如何使用 RunnableWithMessageHistory

LangChain 提供 RunnableWithMessageHistory 封装任何 Runnable,实现会话记忆管理:

langchain_core.runnables.history.RunnableWithMessageHistory

实现逻辑简要(How it works):

class RunnableWithMessageHistory(Runnable):    def __init__(        self,        runnable: Runnable,  # 被包裹的链或模型        get_session_history: Callable,  # 根据 session_id 获取 MessageHistory 对象        input_messages_key: str = ...,        history_messages_key: str = ...    ):        ...    def invoke(self, input: dict, config: RunnableConfig) -> Any:        # 1. 从 config 中获取 session_id        session_id = config.configurable.get("session_id")        # 2. 调用 get_session_history(session_id) 获取消息记录对象        history = self.get_session_history(session_id)        # 3. 从 history 读取历史消息 → 注入 input 中指定的 key        input[self.history_messages_key] = history.messages        # 4. 执行原始 runnable        output = self.runnable.invoke(input, config)        # 5. 将当前对话的输入/输出消息记录到 history 中        history.add_messages([HumanMessage(...), AIMessage(...)])        return output

作用(What it does):

import osfrom typing import Dictfrom langchain_core.runnables import RunnableWithMessageHistoryfrom langchain_core.chat_history import BaseChatMessageHistoryfrom langchain_community.chat_message_histories import ChatMessageHistoryfrom langchain.prompts import ChatPromptTemplatefrom langchain_openai import ChatOpenAI# 模拟一个内存中的历史记录存储(实际部署可替换为 Redis 等)session_store: Dict[str, ChatMessageHistory] = {}def get_session_history(session_id: str) -> BaseChatMessageHistory:    if session_id not in session_store:        session_store[session_id] = ChatMessageHistory()    return session_store[session_id]# 创建 Prompt 模板prompt = ChatPromptTemplate.from_messages([    ("system", "你是一个非常专业且友好的助手。"),    ("user", "{question}"),])# 创建原始模型链llm = ChatOpenAI(    temperature=0.7,    model="glm-4.5",    openai_api_key=os.getenv("ZAI_API_KEY"),    openai_api_base="https://open.bigmodel.cn/api/paas/v4/")chain = prompt | llm# 包装为带记忆功能的链chain_with_history = RunnableWithMessageHistory(    chain,    get_session_history,    input_messages_key="question",  # 输入参数名    history_messages_key="history",  # prompt 接收历史的参数名)# 执行调用,自动注入历史并记录response = chain_with_history.invoke(    {"question": "帮我总结一下《实践论》的核心论点,纯文本输出。"},    config={"configurable": {"session_id": "user123"}})print(session_store)print(response.content)

此机制适合单用户/多线程管理,可替换持久化存储(如 Redis)用于实际部署。


当然可以。下面是对你提供的内容进行专业、全面的扩展说明,包括每种 Memory 类型的功能解析、使用场景完整示例代码


四、Memory 类型与历史拼接方法详解

LangChain 的 Memory 系统本质上是对历史消息的持久化与上下文拼接机制,其核心目标是:

1. ConversationBufferMemory:最简单的全量对话缓存器

✅ 适用于短对话或窗口足够大的场景

功能说明:
示例代码:
import osfrom langchain.chains.conversation.base import ConversationChainfrom langchain.memory import ConversationBufferMemoryfrom langchain_openai import ChatOpenAImemory = ConversationBufferMemory(return_messages=True)llm = ChatOpenAI(    temperature=0.7,    model="glm-4.5",    openai_api_key=os.getenv("ZAI_API_KEY"),    openai_api_base="https://open.bigmodel.cn/api/paas/v4/")chain = ConversationChain(    llm=llm,    memory=memory,    verbose=False)print(chain.invoke(    {"input": "请用一句话回答(不要超过30个汉字):苏联的存在时间区间是什么?"})["response"])print(chain.invoke({"input": "我刚才问的问题是什么?"})["response"])


2. ConversationBufferWindowMemory:滑动窗口对话历史

✅ 适用于需要历史但受限于 Prompt 长度的中长对话

功能说明:
示例代码:
from langchain.memory import ConversationBufferWindowMemorymemory = ConversationBufferWindowMemory(    k=3,  # 仅保留最近3轮消息    return_messages=True)

搭配使用方式与 ConversationBufferMemory 相同。


3. ConversationSummaryMemory:自动摘要上下文

✅ 适用于历史很长但上下文窗口受限场景

功能说明:
示例代码:
import osfrom langchain.chains.conversation.base import ConversationChainfrom langchain.memory import ConversationSummaryMemoryfrom langchain_openai import ChatOpenAIllm = ChatOpenAI(    temperature=0.7,    model="glm-4.5",    openai_api_key=os.getenv("ZAI_API_KEY"),    openai_api_base="https://open.bigmodel.cn/api/paas/v4/")memory = ConversationSummaryMemory(    llm=llm,  # 用于生成摘要的 LLM    return_messages=True,    max_token_limit=300  # 控制 summary 长度)chain = ConversationChain(    llm=llm,    memory=memory,    verbose=False)response = chain.invoke(dict(input="我最近很焦虑,可能是因为工作压力大。"))print(response['response'])print("\n" + "*" * 100 + "\n")response = chain.invoke(dict(input="你有什么放松建议吗?"))print(response['response'])


4. ConversationSummaryBufferMemory:摘要 + 滑窗的混合策略

✅ 适用于既想缩短上下文,又希望保留部分原文历史

功能说明:
示例代码:
from langchain.memory import ConversationSummaryBufferMemorymemory = ConversationSummaryBufferMemory(    llm=ChatOpenAI(),    max_token_limit=500,    return_messages=True)

此方案对长时间交互特别友好,适合实际产品部署。


5. 自定义 Memory:继承 BaseMemory / BaseChatMessageHistory

如果你希望将消息保存到数据库、Redis、MongoDB 等,也可以通过实现自定义 Memory 类来替换默认行为:

from langchain_core.chat_history import BaseChatMessageHistoryclass MyCustomMemory(BaseChatMessageHistory):    def __init__(self):        self.messages = []    def add_message(self, message): self.messages.append(message)    def add_messages(self, messages): self.messages.extend(messages)    def clear(self): self.messages.clear()

你可以将其结合到 RunnableWithMessageHistory 或自定义链中。


✅ 小结对比

Memory 类型优点适用场景
ConversationBufferMemory实现简单、保留全部对话原型、短对话
ConversationBufferWindowMemory控制窗口大小、性能更好中长对话
ConversationSummaryMemory自动压缩语义、节省 token超长上下文、摘要理解
ConversationSummaryBufferMemory兼顾原文与摘要,灵活性最高高质量对话系统

总结


接下来,我们将解析 LangChain Agent 的决策流程、工具调用与动作推理机制,构建智能 Agent 执行体系。

Fish AI Reader

Fish AI Reader

AI辅助创作,多种专业模板,深度分析,高质量内容生成。从观点提取到深度思考,FishAI为您提供全方位的创作支持。新版本引入自定义参数,让您的创作更加个性化和精准。

FishAI

FishAI

鱼阅,AI 时代的下一个智能信息助手,助你摆脱信息焦虑

联系邮箱 441953276@qq.com

相关标签

LangChain Memory系统 LLM 多轮对话 上下文记忆
相关文章