前言

在构建AI智能体（Agent）的过程中，我踩过最深的一个坑就是——大模型是个“金鱼脑” 。无论你上一轮跟它聊了什么，下一轮它可能就忘得一干二净。

这不是大模型的“态度问题”，而是它的底层设计决定的：所有大模型本质上都是“无状态”的——它对每一次请求的处理都是独立的，不会主动留存上一轮对话的任何信息。想让大模型“记住”上下文，本质上需要做两件事：记录历史（将用户提问和模型回答存储在外部载体中），以及在发起新请求时将历史对话拼接到提示词中。

对于简单的对话机器人，这个问题可能还好解决。但当你的Agent需要执行复杂长链任务——比如自动完成一份市场调研报告、协调多个子Agent完成供应链调度——记忆机制的缺失就会成为致命的短板。任务执行到一半，Agent忘了之前收集了什么信息；跨会话交互时，Agent认不出这是“老熟人”。

本文将深入剖析LangChain生态中的记忆机制设计，从短期记忆的多种实现策略到长期记忆的架构方案，结合源码解读，帮你彻底搞懂如何让AI Agent告别“金鱼脑”。

一、记忆的本质：从“无状态”到“有记忆”

在深入框架之前，我们先建立一个核心认知：所谓“记忆”，本质是通过外部系统存储对话历史，再将其动态拼接至提示词中。

LangChain通过Memory组件将这个“记录-存储-拼接”的过程进行了标准化封装。整个记忆系统由两个核心抽象层组成：

抽象层 职责
BaseMemory 定义“如何管理记忆”——保存、加载、清空等行为接口
ChatMessageHistory 定义“记忆存放在哪里”——支持内存、Redis、数据库等不同后端

理解这两个层次后，就能轻松掌握不同类型记忆机制的设计逻辑。

二、短期记忆：让Agent“记住刚才说了什么”

短期记忆（Short-term Memory）是会话级别的记忆，负责在单次会话中维持对话上下文。LangChain提供了多种短期记忆实现，适用于不同场景。

2.1 ConversationBufferMemory：全量记忆

最基础的实现，直接将完整对话历史以列表形式存储。

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

memory = ConversationBufferMemory()
chain = ConversationChain(llm=ChatOpenAI(), memory=memory)

chain.run("你好，我叫张三")
chain.run("我喜欢喝拿铁")
chain.run("你记得我叫什么吗？")  # ✅ 能记住

原理：每次交互后，将用户输入和模型输出追加到内存列表中；调用时通过load_memory_variables返回完整历史。

适用场景：短对话（如客服单轮咨询）。优点是完全保留细节，缺点是随对话长度增加会导致Prompt过长，超出模型上下文窗口。

2.2 ConversationBufferWindowMemory：滑动窗口记忆

带“窗口限制”的缓冲记忆，只保留最近的k轮对话。

from langchain.memory import ConversationBufferWindowMemory

memory = ConversationBufferWindowMemory(k=3)  # 只保留最近3条消息
chain = ConversationChain(llm=ChatOpenAI(), memory=memory)

原理：维护一个固定长度的滑动窗口，超过k轮后自动丢弃最早的对话。

适用场景：中等长度对话（如多轮问答），平衡上下文相关性和长度。

2.3 ConversationTokenBufferMemory：按Token限制记忆

按Token数量限制记忆，确保总对话历史的Token数不超过模型窗口。

from langchain.memory import ConversationTokenBufferMemory

memory = ConversationTokenBufferMemory(llm=ChatOpenAI(), max_token_limit=2000)

原理：通过Tokenizer计算每轮对话的Token数，累计超过阈值时从最早的对话开始删除。

适用场景：对Token成本敏感或模型窗口固定的场景（如API调用）。

2.4 ConversationSummaryMemory：摘要压缩记忆

对对话历史进行“摘要压缩”，用摘要代替原始对话。

from langchain.memory import ConversationSummaryMemory

memory = ConversationSummaryMemory(llm=ChatOpenAI())
chain = ConversationChain(llm=ChatOpenAI(), memory=memory)

原理：每轮交互后调用LLM对历史对话生成摘要，后续仅将摘要传入Prompt。

适用场景：长对话（如持续几小时的咨询），避免原始历史过长。缺点是可能丢失细节（摘要精度依赖LLM）。

三、源码剖析：BaseMemory核心接口设计

了解了各种记忆类型后，我们深入源码看看LangChain是如何设计这套记忆系统的。

3.1 BaseMemory抽象基类

BaseMemory是所有记忆类的抽象基类，定义了记忆模块必须实现的核心接口：

from abc import ABC, abstractmethod
from typing import Any

class BaseMemory(Serializable, ABC):
    """记忆的抽象基类，定义了所有记忆类型必须实现的核心接口"""
    
    @property
    @abstractmethod
    def memory_variables(self) -> list[str]:
        """返回此记忆组件管理的变量名，例如：["history", "summary"]"""
        pass
    
    @abstractmethod
    def load_memory_variables(self, inputs: dict[str, Any]) -> dict[str, Any]:
        """
        加载记忆变量，根据输入获取相关的历史内容
        返回的字典会被添加到提示词的上下文中
        """
        pass
    
    @abstractmethod
    def save_context(self, inputs: dict[str, Any], outputs: dict[str, str]) -> None:
        """
        保存当前对话上下文
        inputs: 用户的输入，如 {"input": "你好"}
        outputs: 模型的输出，如 {"output": "你好！"}
        """
        pass
    
    def clear(self) -> None:
        """清除所有记忆内容"""
        pass

这套接口设计的精妙之处在于：load_memory_variables和save_context构成了记忆的读写闭环，而memory_variables则告诉Chain哪些变量需要被注入到Prompt中。

3.2 BaseChatMemory：聊天记忆的具体实现

BaseChatMemory继承自BaseMemory，提供了聊天记忆的通用实现：

class BaseChatMemory(BaseMemory, ABC):
    """聊天记忆的基础类，提供了聊天记忆的通用实现"""
    
    chat_memory: BaseChatMessageHistory = Field(
        default_factory=InMemoryChatMessageHistory
    )
    """消息历史存储对象，可替换为Redis、数据库等不同后端"""
    
    output_key: Optional[str] = None
    input_key: Optional[str] = None
    return_messages: bool = False
    
    def _get_input(self, inputs: dict[str, Any]) -> str:
        """从输入中获取用户消息"""
        if self.input_key is not None:
            return inputs[self.input_key]
        # ...

这里的关键设计是依赖注入：chat_memory字段允许开发者注入不同的存储后端实现（内存、Redis、数据库等），而不需要修改记忆逻辑本身。这就是“如何管理记忆”与“记忆存放在哪里”的解耦。

四、长期记忆：让Agent“记住昨天的事”

短期记忆解决了“会话内”的问题，但Agent还需要“跨会话”的记忆能力——长期记忆（Long-term Memory） 。

4.1 长期记忆的核心挑战

实现长期记忆面临三大技术挑战：

  1. 上下文窗口限制：无法将所有历史都塞进Prompt
  2. 记忆数据的有效组织：如何结构化存储海量信息
  3. 长期与短期记忆的平衡：两者如何协同工作

4.2 VectorStoreRetrieverMemory：向量检索记忆

LangChain提供了VectorStoreRetrieverMemory，通过向量数据库实现海量对话的持久化记忆和语义检索：

from langchain.memory import VectorStoreRetrieverMemory
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 初始化向量数据库
vectorstore = Chroma(
    embedding_function=OpenAIEmbeddings(),
    collection_name="agent_memory"
)

# 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 创建向量记忆
memory = VectorStoreRetrieverMemory(retriever=retriever)

# 保存记忆（自动向量化存储）
memory.save_context(
    {"input": "我是张三，住在北京"},
    {"output": "好的，记住了"}
)

# 加载相关记忆（语义检索）
vars = memory.load_memory_variables(
    {"input": "我在哪个城市？"}
)
# 返回：{"history": "用户之前说住在北京"}

核心原理：VectorStoreRetrieverMemory将记忆存储在VectorDB中，每次调用时通过语义检索召回最相关的top-K条记忆。这种方式将RAG的思想应用到了记忆管理中——不是简单地把所有历史塞进去，而是按需检索最相关的内容。

4.3 LangGraph的长期记忆Store

在LangGraph中，长期记忆通过Store机制实现：

from langgraph.store.memory import InMemoryStore

# 创建长期记忆存储
store = InMemoryStore()

# 按命名空间存储记忆
store.put(
    ("user_profile", "user_123"),
    {"name": "张三", "preferences": {"theme": "dark", "language": "zh"}}
)

# 检索记忆
memories = store.search(("user_profile", "user_123"))

LangGraph的Store支持自定义命名空间，记忆可以在不同会话线程之间共享。这意味着：用户今天和Agent聊了偏好设置，明天打开新会话，Agent依然能“记住”这些信息。

4.4 长期记忆的存储选型

在实际生产环境中，长期记忆的存储有多种选择：

存储方案 适用场景
关系型数据库（PostgreSQL） 结构化用户档案、历史订单等
NoSQL（MongoDB） 灵活的对话记录、非结构化数据
向量数据库（Chroma/Pinecone） 语义检索、相似性召回
Redis 高频访问的短期缓存

五、混合记忆架构：短期+长期协同工作

生产级的Agent系统通常需要混合记忆架构——短期记忆负责会话内的上下文连贯性，长期记忆负责跨会话的知识积累。

5.1 架构设计

┌─────────────────────────────────────────────────────┐
│                    Agent 应用层                      │
├─────────────────────────────────────────────────────┤
│                                                     │
│  ┌──────────────┐         ┌──────────────┐        │
│  │  短期记忆层   │         │  长期记忆层   │        │
│  │              │         │              │        │
│  │ BufferWindow │◄────────│ VectorStore  │        │
│  │ Summary      │         │ PostgreSQL   │        │
│  │ TokenBuffer  │         │ Redis        │        │
│  └──────────────┘         └──────────────┘        │
│         │                        │                  │
│         ▼                        ▼                  │
│  ┌──────────────────────────────────────┐          │
│  │         ChatMessageHistory           │          │
│  │    (可插拔存储后端：内存/Redis/DB)    │          │
│  └──────────────────────────────────────┘          │
└─────────────────────────────────────────────────────┘

5.2 实践示例：混合记忆的组装

from langchain.memory import (
    ConversationBufferWindowMemory,
    VectorStoreRetrieverMemory
)
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

class HybridMemory:
    """混合记忆：短期窗口记忆 + 长期向量检索记忆"""
    
    def __init__(self, window_size: int = 10, top_k: int = 5):
        # 短期记忆：只保留最近10轮对话
        self.short_term = ConversationBufferWindowMemory(
            k=window_size,
            memory_key="recent_history"
        )
        
        # 长期记忆：向量检索
        vectorstore = Chroma(
            embedding_function=OpenAIEmbeddings(),
            collection_name="long_term_memory"
        )
        retriever = vectorstore.as_retriever(search_kwargs={"k": top_k})
        self.long_term = VectorStoreRetrieverMemory(
            retriever=retriever,
            memory_key="long_term_context"
        )
    
    def load_memory_variables(self, inputs: dict) -> dict:
        """加载混合记忆"""
        short = self.short_term.load_memory_variables(inputs)
        long = self.long_term.load_memory_variables(inputs)
        return {**short, **long}
    
    def save_context(self, inputs: dict, outputs: dict) -> None:
        """保存到两种记忆"""
        self.short_term.save_context(inputs, outputs)
        self.long_term.save_context(inputs, outputs)

5.3 LangGraph的Checkpointer：状态持久化

在LangGraph中，短期记忆通过Checkpointer（检查点保存器） 实现持久化：

from langgraph.checkpoint.sqlite import SqliteSaver

# 创建检查点保存器（持久化到SQLite）
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")

# 编译图时传入checkpointer
graph = builder.compile(checkpointer=checkpointer)

# 指定thread_id恢复会话
config = {"configurable": {"thread_id": "conversation_123"}}
result = graph.invoke(inputs, config=config)

Checkpointer的工作原理类似于游戏存档：每当图中的一个节点执行完毕后，LangGraph会自动把当前的State（包括消息历史、变量值）打一个快照并存入数据库。通过thread_id，可以随时恢复到任意时间点的状态。

六、LangChain vs LangGraph：记忆实现的差异

同为LangChain生态，LangChain和LangGraph在记忆实现上走出了不同路径：

维度 LangChain LangGraph
定位 通用大模型应用框架 面向Agent的专用框架
记忆方式 Memory组件（对话历史管理） State + Checkpointer（状态图管理）
短期记忆 各类Buffer/Window/Summary Memory State中的messages字段 + Checkpointer
长期记忆 VectorStoreRetrieverMemory Store（跨线程命名空间存储）
状态恢复 需手动管理 通过thread_id自动恢复

简单来说：LangChain的Memory是“附加的” ，你需要显式地将Memory组件挂载到Chain上；而LangGraph的记忆是“天然的” ，StateGraph本身就有状态，Checkpointer让这种状态可以随时保存和恢复。

七、总结与最佳实践

经过以上分析，我总结了几条在生产环境中落地记忆机制的实践经验：

7.1 选型建议

场景 推荐方案
简单问答机器人 ConversationBufferWindowMemory（k=5~10）
长对话客服系统 ConversationSummaryMemory 或 ConversationSummaryBufferMemory
Token敏感场景 ConversationTokenBufferMemory
需要跨会话记忆 VectorStoreRetrieverMemory + 向量数据库
复杂Agent任务 LangGraph + Checkpointer + Store

7.2 避坑指南

  1. 不要把所有历史都塞进去：大模型的上下文窗口有限，超出限制会触发截断或报错。
  2. 短期记忆和长期记忆要分工：短期记“刚才说了什么”，长期记“用户是谁、偏好什么”。
  3. 记忆与RAG要区分：RAG的本质是按需检索外部静态知识，天然无状态；而记忆系统强调跨会话的状态积累。
  4. 生产环境务必持久化：开发时用InMemoryChatMessageHistory没问题，上线后一定要换成Redis或数据库后端。

7.3 核心 Takeaways

  • 大模型是无状态的，“记忆”本质是外部存储+动态拼接
  • LangChain的BaseMemory定义了记忆的读写接口，ChatMessageHistory定义了存储后端
  • 短期记忆有四种主流实现：全量、窗口、Token限制、摘要
  • 长期记忆通过向量检索或LangGraph Store实现跨会话知识复用
  • 混合架构才是生产级Agent的正确打开方式

记忆机制是AI Agent从“玩具”走向“工具”的关键一步。希望这篇文章能帮你彻底搞懂LangChain生态中的记忆设计，让你的Agent真正告别“金鱼脑”。

# 人工智能 # microsoft
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

【粉丝福利社】OpenClaw深度解析：从架构原理到Agent工程实践

avatar AI Agent技术社区
cover

MCP与Skill的思考：AI Agent架构深度解析

avatar AI Agent技术社区
cover

Agent 效果不好，先别急着改 Prompt

avatar AI Agent技术社区