AGENT

AI Agent 记忆终极指南

我们不能只是把无限的历史信息塞进模型的上下文窗口。我们需要构建复杂的记忆系统。本文是解决这个问题的权威指南。

admin

Dec 16, 2025 • 21 min read

你有没有和聊天机器人交谈过，当你提到你的名字时，两个问题后它又问你名字是什么？这种令人沮丧的经历是一种“数字失忆症”，突显了构建真正智能AI代理的最重大挑战之一。核心问题是，大型语言模型（LLM）本质上是无状态的。它们不会“记住”过去的互动。每次你发送一条消息，模型都会将其作为全新的事件来处理，没有历史。

像ChatGPT这样的应用程序中的记忆幻觉只是幻觉。应用层巧妙地绕过了这个限制，通过每一轮对话都提醒模型对话历史。这种不断的“提醒”是上下文管理的核心，这是任何AI工程师的关键技能。但当对话变得太长时会发生什么？如果你需要AI记住一小时前的一个关键事实，或者上周提到的一个偏好？

这就是真正的工程开始的地方。我们不能只是把无限的历史信息塞进模型中。我们需要构建复杂的记忆系统。本文是解决这个问题的权威指南。我们将从上下文窗口的基础约束出发，探索到代理记忆的前沿。我们将探讨理论，深入实践代码，并展示一个完整的、智能的聊天记忆系统，你可以自己构建和实验。最后，你将拥有知识，将你的AI应用从健忘的工具转变为真正具有上下文意识的智能伙伴。

1、基础：解构LLM的上下文窗口

在我们可以构建记忆系统之前，我们必须首先了解战场。在LLM的世界中，战场是上下文窗口。它是决定我们关于记忆管理的每一个决策的最重要的限制。

1.1 什么是上下文窗口？

将LLM的上下文窗口视为它的“工作记忆”。它是模型可以“看到”并处理的文本的有限数量，以生成响应。你发送给模型的一切——定义其角色的系统提示、你的当前查询、聊天历史以及你想让它分析的任何文档——都必须符合这个窗口。如果总内容超过这个限制，模型就无法处理，导致错误，或者更糟糕的是，无声截断，关键信息丢失。

这个限制类似于人类的工作记忆。你可以同时在脑海中保持几个想法来解决问题，但你不能保持整个图书馆。同样，LLM有强大的但有限的即时上下文容量。

1.2 上下文的货币：标记

上下文窗口的大小不是用单词或字符来衡量的，而是用标记（token）来衡量的。标记是模型处理的基本文本单位。它可以是一个完整的单词，比如“apple”，也可以是单词的一部分，如“ing”或“pre”。例如，短语“LLM memory is complex”可能会被分解为标记：

[“LLM”, “ memory”, “ is”, “ complex”]。

理解并计算标记是不可协商的，对于记忆管理来说至关重要。它使我们能够精确地知道我们的对话历史消耗了多少“空间”，以及何时即将超出模型的限制。对于像OpenAI这样的提供者，tiktoken库是准确计算给定文本的标记数的行业标准。

以下是使用tiktoken来计算简单文本和更复杂的聊天消息结构的标记数的方法：

import tiktoken  

# 获取模型的编码  
encoding = tiktoken.encoding_for_model("gpt-4")  

# 计算简单文本的标记数  
text = "Hello, how are you today?"  
token_count = len(encoding.encode(text))  
print(f"标记数: {token_count}")  

# 对于聊天消息，必须考虑元数据  
def count_chat_tokens(messages, model="gpt-4"):  
    """  
    计算聊天消息列表中的标记数。  
    """  
    encoding = tiktoken.encoding_for_model(model)  
    tokens = 0  
    for message in messages:  
        tokens += 4  # 每条消息遵循 <|start|>{role/name}\n{content}<|end|>\n  
        for key, value in message.items():  
            tokens += len(encoding.encode(value))  
    tokens += 2  # 每个回复都以 <|start|>assistant 开始  
    return tokens  

# 聊天消息示例用法  
chat_messages = [{"role": "user", "content": "Hello, how are you today?"}]  
chat_token_count = count_chat_tokens(chat_messages)  
print(f"聊天标记数: {chat_token_count}")

1.3 上下文军备竞赛及其权衡

近年来，我们看到了一场“上下文军备竞赛”，模型提供者发布了越来越多的上下文窗口版本。从最初的几千个标记扩展到了数十万，现在甚至达到了数百万。这种扩展打开了新的可能性，但并没有消除对记忆管理的需求。

更大的窗口只是延迟了问题，并引入了显著的权衡：

成本: API定价通常基于处理的标记数量。每次调用发送大量上下文可能会变得非常昂贵。
延迟: 模型需要处理的标记越多，生成响应所需的时间就越长。
性能下降: 研究表明，一些模型会遇到“中间丢失”的问题，即它们对非常长的上下文中心的信息关注较少。在整个窗口内的回忆质量并不总是完美的。

下表提供了当今一些领先模型的上下文窗口的快照。

最终，核心挑战是从适应信息到智能管理信息，以保持高信噪比，控制成本，并确保高质量的响应。

2、层次结构：从人类认知中学习

为了构建有效的AI记忆，我们可以借鉴人类认知架构，它使用多层记忆层次结构。

短期记忆（STM）: 这是当前对话的即时工作区。它的容量有限（如最后5-9次交互），持续时间只有几秒或几分钟。在LLM中，这直接通过上下文窗口实现。
长期记忆（LTM）: 提供跨不同会话的持久存储。它的容量几乎是无限的，并通过外部数据库或向量存储实现。
工作记忆: 这是信息被积极操作的地方。对于LLM，它是当前上下文窗口和从长期存储中检索的记忆的结合。

理解这一层次结构有助于我们设计更复杂的系统，而不仅仅依赖于单一的、庞大的记忆存储。

3、“记忆阶梯” ：从基本到高级策略

我们现在将攀登“记忆阶梯”，从简单的缓冲技术开始，逐步推进到智能摘要，然后探索稳健的混合记忆架构。

3.1 简单缓冲和修剪

我们的旅程从最简单的做法开始：缓冲。这种方法是聊天机器人的“Hello, World!”，教我们状态管理的基本机制。

手动消息传递

给LLM记忆的最简单方法是手动收集对话历史并在每次新查询时将其传回给模型。在像LangChain这样的框架中，提示模板中的MessagesPlaceholder充当变量，将被先前消息列表填充。

这里有一个概念性的代码示例，演示这个基本原理：

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage  
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder  
from langchain_openai import ChatOpenAI  

llm = ChatOpenAI(model="gpt-4o")  

prompt = ChatPromptTemplate.from_messages()  

chain = prompt | llm  

# 手动历史管理  
chat_history = [  
    HumanMessage(content="Hi, I'm Alice."),  
    AIMessage(content="Hello Alice! How can I help you?"),  
]  

# 新查询被添加到历史记录中以进行调用  
response = chain.invoke({  
    "messages": chat_history +  
})  

print(response.content) # 模型正确回答 "Your name is Alice."

这对于短对话来说完全有效，但随着chat_history列表的增长，我们不可避免地会达到上下文窗口限制。

滑动窗口技术

为了避免溢出，最常见的解决方案是滑动窗口技术。我们只需保留历史记录中的最后k条消息。虽然简单，但其主要缺点是 “上下文悬崖”，其中重要信息被推离窗口并永远遗忘。

LangChain为此提供了trim_messages辅助函数：

from langchain_core.messages import trim_messages, HumanMessage, AIMessage  

# 这个trimmer将保留最近的消息，最多2条。  
trimmer = trim_messages(  
    strategy="last",  
    max_tokens=2,  
    token_counter=len # 一个简单的计数器，每条消息为1个标记  
)  

messages = [  
    HumanMessage(content="Hi!"),  
    AIMessage(content="Hello!"),  
    HumanMessage(content="How are you?"),  
    AIMessage(content="I'm good, thanks!")  
]  

trimmed = trimmer.invoke(messages)  
# 结果: [HumanMessage(content='How are you?'), AIMessage(content="I'm good, thanks!")]  
print(trimmed)

LangChain实战：ConversationBufferMemory

像LangChain这样的框架提供了这种模式的抽象。ConversationBufferMemory组件自动存储和提供对话历史。这是在**Intelligent_Chat_Memory (Basic)**项目中展示的方法。

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

# 初始化模型  
llm = ChatOpenAI(model="gpt-4o")  

# 初始化记忆缓冲区  
# 这将存储消息并将它们注入到{history}或{chat_history}变量中  
memory = ConversationBufferMemory(memory_key="history", return_messages=True)  

# ConversationChain 自动使用记忆对象  
conversation = ConversationChain(  
    llm=llm,  
    memory=memory,  
    verbose=True # 设置为True以查看发送到LLM的完整提示  
)  

# 第一次交互  
conversation.predict(input="Hi, I'm Bob.")  
# 第二次交互  
conversation.predict(input="I live in New York.")  
# 第三次交互  
# 前两次交互的历史自动包含在提示中  
conversation.predict(input="What is my name?")

虽然ConversationBufferMemory简化了代码，但它不是长时间对话的可扩展解决方案。

3.2 通过摘要进行智能压缩

仅仅忘记旧消息是一种粗暴的解决方案。一种更复杂的策略是摘要：使用LLM将较旧的消息压缩成一个运行摘要。这代表了从管理上下文大小到管理上下文保真度的关键范式转变。

LangChain的ConversationSummaryBufferMemory

这是Advance_Intelligent_Chat_Memory (Advanced)项目的核心技术。LangChain的ConversationSummaryBufferMemory提供了一种混合解决方案，结合了最近消息的原始缓冲区和较旧消息的逐步更新摘要。

神奇之处在于max_token_limit触发器。系统监控原始消息缓冲区的标记计数。一旦超过限制，最旧的消息会被发送到LLM进行摘要，并由摘要替换。

以下是此逻辑的实现方式：

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

# 我们需要一个LLM来驱动摘要  
llm = ChatOpenAI(model="gpt-4o", temperature=0)  

# 初始化摘要缓冲内存  
# 一旦缓冲区超过1000个标记，它将开始摘要。  
memory = ConversationSummaryBufferMemory(  
    llm=llm,  
    max_token_limit=1000,  
    return_messages=True  
)  

# 使用此高级内存创建对话链  
conversation = ConversationChain(  
    llm=llm,  
    memory=memory,  
    verbose=True  
)  

# 运行一个长对话...  
conversation.predict(input="Hi, I'm Carol, a data scientist from London.")  
conversation.predict(input="I'm working on a project about LLM memory systems.")  
#...许多其他交互...  

# 超过1000个标记限制后，内存对象将包含早期对话的摘要和最近消息的缓冲区。  
print(memory.chat_memory.messages)

摘要的优缺点

优点: 有效地保留了很长对话中的上下文，防止“上下文悬崖”。
缺点: 会产生额外的成本和延迟用于摘要调用，并且在压缩过程中可能会丢失细微差别（保真度损失）。

3.3 构建外部大脑：混合记忆和检索

摘要保持了对话的“要点”，但在高保真事实方面表现不佳。为了解决这个问题，我们需要一个结合我们对话缓冲区和外部、可搜索的长期记忆存储的混合架构，就像人类记忆一样。

长期记忆：向量存储

实现长期记忆的最常见方法是使用向量存储。这个过程是检索增强生成（RAG）的基础，涉及将文本嵌入数值向量并存储在数据库中以进行语义搜索。这允许AI根据意义而不是关键词找到相关信息。

双阶段记忆架构

最强大的系统将这些方法结合成一个双阶段或混合记忆架构。

短期/工作记忆: 由ConversationSummaryBufferMemory管理，以保持流畅的逐轮上下文。
长期/情景记忆: 一个向量存储，作为重要事实或关键决策的永久、可搜索的档案。

这是一个语义记忆检索器和混合系统的概念实现：

import numpy as np  
from sentence_transformers import SentenceTransformer  
from datetime import datetime  

class SemanticMemoryRetriever:  
    def init(self):  
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')  
        self.memory_store =  
        self.embeddings =  

    def store_memory(self, message, context):  
        """存储带有语义嵌入的消息"""  
        embedding = self.encoder.encode(message)  
        self.memory_store.append({  
            'message': message,  
            'context': context,  
            'timestamp': datetime.now()  
        })  
        self.embeddings.append(embedding)  

    def retrieve_relevant_memories(self, query, top_k=5):  
        """检索与给定查询相关的记忆"""  
        query_embedding = self.encoder.encode(query)  

        # 计算余弦相似性  
        similarities = [np.dot(query_embedding, emb) / (np.linalg.norm(query_embedding) * np.linalg.norm(emb)) for emb in self.embeddings]  

        top_indices = np.argsort(similarities)[-top_k:]  
        return [self.memory_store[i] for i in top_indices]  

class HybridMemorySystem:  
    def init(self):  
        # 注意：这些将在实际应用中完全初始化  
        self.short_term = ConversationBufferMemory(max_length=10)   
        self.long_term = SemanticMemoryRetriever()  
        self.summary_buffer = ConversationSummaryBufferMemory(max_token_limit=2000)  

    def get_context(self, query):  
        """从多个记忆源中汇编上下文"""  
        # 从短期缓冲区获取最近的对话  
        recent_context = self.short_term.get_recent_messages()  

        # 通过语义搜索获取相关长期记忆  
        relevant_memories = self.long_term.retrieve_relevant_memories(query)  

        # 获取运行中的对话摘要  
        summary = self.summary_buffer.get_summary()  

        # 组合并优先考虑这些上下文以提供给LLM  
        return self.combine_contexts(recent_context, relevant_memories, summary)  

    def combine_contexts(self, recent, relevant, summary):  
        # 这个方法将包含格式化最终上下文字符串的逻辑  
        # 例如: f"Summary: {summary}\nRelevant Info: {relevant}\nRecent Chat: {recent}"  
        pass

这种分层设计更为强大，认识到对话流程和事实回忆需要不同的机制。

4、AI记忆的前沿：代理和结构化方法

我们现在到达了AI记忆的前沿，超越了简单的上下文注入，将记忆视为代理世界模型中的结构化、可查询的组件。

4.1 ReAct：记忆作为主动工具

ReAct（Reason + Act）框架改变了代理与其记忆之间的关系。访问记忆成为代理在推理循环中决定采取的显式动作。

流程是：观察 -> 思考 -> 动作 -> 观察 -> 思考 -> 最终答案。这使得记忆访问成为一个明确、可审计的步骤，让代理决定是否、何时和如何使用其记忆。

4.2 将历史浓缩为知识图谱

虽然向量存储可以找到语义相似的文本，但它们不了解实体之间的关系。基于图的记忆通过将信息解析为节点（实体）和边（关系）的结构化格式解决了这个问题。

而不是存储“Alice的朋友是Bob”，图会有：

节点: Alice, Bob
边: Alice — (IS_FRIEND_OF) → Bob

这解锁了复杂的多跳推理。像Graphiti和Mem0这样的框架正在引领这一方法，允许代理实时构建和查询动态知识图。

这是一个使用networkx的概念性上下文记忆图示例：

import networkx as nx  

class ContextualMemoryGraph:  
    def init(self):  
        self.memory_graph = nx.DiGraph()  
        self.embeddings = {} # 假设我们有获取嵌入的方法  

    def calculate_similarity(self, content1, content2):  
        # 占位符，用于真实的相似性函数（例如嵌入的余弦相似性）  
        return 0.8   

    def add_memory_node(self, memory_id, content, context):  
        """添加一个记忆节点并将其连接到上下文相似的节点"""  
        self.memory_graph.add_node(memory_id, content=content)  

        # 根据上下文相似性创建边  
        for existing_node_id, existing_node_attrs in self.memory_graph.nodes(data=True):  
            if memory_id!= existing_node_id:  
                similarity = self.calculate_similarity(content, existing_node_attrs['content'])  
                if similarity > 0.7:  
                    self.memory_graph.add_edge(memory_id, existing_node_id, weight=similarity)  

    def retrieve_connected_memories(self, query_memory_id, depth=2):  
        """检索指定图距离内的记忆"""  
        if query_memory_id not in self.memory_graph:  
            return  
        return list(nx.single_source_shortest_path_length(  
            self.memory_graph, query_memory_id, cutoff=depth  
        ).keys())

4.3 LangGraph：有状态的记忆管理

LangGraph通过提供checkpointer在编译图形时实现了复杂的、有状态的记忆管理，保存每个步骤后的对话状态，无需手动传递历史。

这是一个自动保存对话历史的完整图形：

from langgraph.graph import StateGraph, START  
from langgraph.graph.message import MessagesState  
from langgraph.checkpoint.memory import MemorySaver  
from langchain_core.messages import HumanMessage  
from langchain_openai import ChatOpenAI  

# 假设llm已经初始化：llm = ChatOpenAI(model="gpt-4o")  

# 1. 定义调用模型的函数  
def call_model(state: MessagesState):  
    """图中的一个节点，调用LLM。"""  
    response = llm.invoke(state["messages"])  
    return {"messages": [response]} # 将响应添加到状态  

# 2. 定义状态图  
workflow = StateGraph(MessagesState)  
workflow.add_node("model", call_model)  
workflow.add_edge(START, "model")  

# 3. 添加一个内存检查点  
memory = MemorySaver()  
app = workflow.compile(checkpointer=memory)  

# --- 示例调用 ---  
# 每次调用都链接到一个“thread_id”。LangGraph使用这个ID从检查点程序中检索正确的历史。  
thread_config = {"configurable": {"thread_id": "user_123"}}  

# 第一次交互  
app.invoke(  
    {"messages":},  
    config=thread_config  
)  

# 第二次交互（不需要手动传递历史）  
response = app.invoke(  
    {"messages":},  
    config=thread_config  
)  

# 最终的AIMessage在响应中将包含答案  
print(response['messages'][-1].content)

4.4 分块对于检索的重要性

任何基于检索的记忆系统质量在很大程度上取决于分块——信息是如何分解和存储的。朴素的分块会破坏上下文。先进的策略至关重要：

基于句子的分块: 在句子边界处分割文本，保留完整的思想。
递归分块: 使用段落和标题等分隔符对文档进行分层分解，以保持结构。
语义分块: 使用嵌入模型找到文本中的自然语义断点，分组相关的句子。

5、生产系统的最佳实践

在生产环境中实施记忆系统时，有几个最佳实践可以确保稳健性和效率。

5.1 记忆生命周期管理

实现适当的记忆生命周期管理，以处理更新和清理过时的记忆，防止无限增长并确保相关性。

from datetime import datetime, timedelta  

class MemoryLifecycleManager:  
    def init(self):  
        self.memory_store = {}  
        self.access_counts = {}  
        self.last_access = {}  

    def update_memory(self, memory_id, content):  
        """更新记忆并跟踪其访问"""  
        self.memory_store[memory_id] = content  
        self.access_counts[memory_id] = self.access_counts.get(memory_id, 0) + 1  
        self.last_access[memory_id] = datetime.now()  

    def cleanup_stale_memories(self, retention_days=30):  
        """删除最近未访问的记忆"""  
        cutoff_date = datetime.now() - timedelta(days=retention_days)  
        stale_memories = [  
            mid for mid, last_access in self.last_access.items()  
            if last_access < cutoff_date  
        ]  

        for memory_id in stale_memories:  
            del self.memory_store[memory_id]  
            del self.access_counts[memory_id]  
            del self.last_access[memory_id]

5.2 错误处理和后备方案

对记忆操作进行健壮的错误处理至关重要。在主要记忆存储失败时，实现后备方案。

import logging  

# 假设PrimaryMemoryStore和FallbackMemoryStore是定义的类  
# logger = logging.getLogger(__name__)  

class RobustMemorySystem:  
    def init(self):  
        self.primary_memory = PrimaryMemoryStore()  
        self.fallback_memory = FallbackMemoryStore()  

    def store_memory(self, memory_item):  
        """存储记忆并处理后备"""  
        try:  
            self.primary_memory.store(memory_item)  
        except Exception as e:  
            logging.warning(f"主记忆存储失败: {e}")  
            try:  
                self.fallback_memory.store(memory_item)  
            except Exception as fallback_error:  
                logging.error(f"后备记忆存储失败: {fallback_error}")  
                # 实施紧急存储或优雅降级  

    def retrieve_memory(self, query):  
        """检索记忆并处理后备"""  
        try:  
            return self.primary_memory.retrieve(query)  
        except Exception as e:  
            logging.warning(f"主记忆检索失败: {e}")  
            return self.fallback_memory.retrieve(query)