AI Agent记忆系统实现指南：从短期记忆到终身记忆的完整架构

AI Agent开发中最常被忽视的组件就是记忆系统。没有记忆的Agent每次对话都是”初次见面”——它不会记得用户偏好、不会记笔记、不会从错误中学习。本文将带你构建一个工业级的AI Agent记忆系统。

记忆系统的四层架构

一个完整的Agent记忆系统应该包含四个层级，从即时到永久：

用户输入 → [短期记忆] → [工作记忆] → [长期记忆] → [核心记忆]
                  ↕              ↕             ↕
            对话缓冲区    知识向量库    压缩摘要存储

第一层：短期记忆（对话上下文）

最简单也最基础的记忆层。本质上就是LLM的上下文窗口：

class ShortTermMemory:
    def __init__(self, max_tokens=128000):
        self.messages = []
        self.max_tokens = max_tokens
    
    def add(self, role, content):
        self.messages.append({"role": role, "content": content})
        self._trim_context()
    
    def _trim_context(self):
        """超出上下文窗口时，从最早的对话开始丢弃"""
        while self._count_tokens() > self.max_tokens:
            # 保留 system prompt
            if len(self.messages) > 1:
                self.messages.pop(1)  # 移除最早的对话（保留system）

注意：不要占用整个上下文窗口——建议为其他记忆层保留10-20%的token空间。

第二层：工作记忆（Agent的”便签本”）

工作记忆是Agent在处理当前任务时使用的临时存储区域。它和短期记忆的区别在于：工作记忆是有结构的，而短期记忆是对话流。

class WorkingMemory:
    def __init__(self):
        self.current_task = None
        self.subtasks = []
        self.observations = []  # Agent执行过程中的"发现"
        self.pending_decisions = []
    
    def add_observation(self, observation):
        self.observations.append({
            "time": datetime.now(),
            "content": observation,
            "importance": self._score_importance(observation)
        })
    
    def _score_importance(self, text):
        """用一个小模型评估信息重要性（0-1）"""
        return importance_model.predict(text)
    
    def summarize(self):
        """将工作记忆压缩成一段摘要"""
        # 只保留高重要性观察
        important = [o for o in self.observations if o["importance"] > 0.7]
        return self._summarize_to_text(important[-10:])  # 最多保留10个

第三层：长期记忆（知识向量库）

这是Agent的”大海马体”。使用向量数据库存储用户的偏好、历史对话的关键信息：

import chromadb

class LongTermMemory:
    def __init__(self, collection_name="agent_memory"):
        self.client = chromadb.Client()
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}
        )
    
    def remember(self, user_id, content, metadata=None):
        # 先检查是否已有相似记忆（去重）
        similar = self.collection.query(
            query_texts=[content],
            n_results=1,
            where={"user_id": user_id}
        )
        if similar and similar['distances'][0][0] < 0.1:
            return  # 太相似了，跳过
        self.collection.add(
            documents=[content],
            metadatas=[{"user_id": user_id, "timestamp": str(datetime.now()), **(metadata or {})}],
            ids=[f"mem_{uuid.uuid4()}"]
        )
    
    def recall(self, user_id, query, top_k=5):
        results = self.collection.query(
            query_texts=[query],
            n_results=top_k,
            where={"user_id": user_id}
        )
        return results['documents'][0]

关键技巧：记忆的去重和衰减是长期记忆系统最容易被忽视的优化点。建议在每次 recall 时，对超过30天且未被访问的记忆进行降权。

第四层：核心记忆（压缩摘要）

Agent还需要一个”人生故事”——一段压缩的自传，包含所有重要的不变信息。类似于记忆压缩：

class CoreMemory:
    def __init__(self, llm):
        self.llm = llm
        self.core = {
            "user_profile": "",
            "learned_patterns": "",
            "key_decisions": []
        }
    
    def update(self, session_events, last_core):
        prompt = f"""
        当前核心记忆：{last_core}
        
        本次对话中的新事件：{session_events}
        
        请更新核心记忆摘要（200字以内）。
        保留：用户不变的偏好、重要决策、学到的关键模式。
        丢弃：一次性的对话细节、临时状态。
        """
        self.core = self.llm.generate(prompt)

整合：完整的记忆Agent

class MemoricAgent:
    def __init__(self, llm):
        self.stm = ShortTermMemory()
        self.wm = WorkingMemory()
        self.ltm = LongTermMemory()
        self.cm = CoreMemory(llm)
    
    def run(self, user_input, user_id="default"):
        # 1. 从长期记忆中召回相关记忆
        past_memories = self.ltm.recall(user_id, user_input)
        
        # 2. 构建带记忆的 prompt
        system_prompt = f"""
        用户核心档案：{self.cm.core}
        相关历史记录：{past_memories}
        当前工作记忆：{self.wm.summarize()}
        """
        
        # 3. 执行Agent逻辑
        response = self.llm.generate(system_prompt, self.stm.messages)
        
        # 4. 更新记忆
        self.wm.add_observation(f"用户问了：{user_input}")
        self.ltm.remember(user_id, f"对话：{user_input} → {response}")
        
        # 5. 每10次对话压缩一次核心记忆
        if len(self.stm.messages) % 20 == 0:
            self.cm.push_to_long_term(self.stm)
        
        return response

性能与优化建议

1. 异步记忆写入：使用消息队列处理记忆写入，不阻塞主Agent流程
2. 记忆优先级：为每段记忆打分，定期剪枝低优先级记忆
3. 遗忘曲线：模拟艾宾浩斯遗忘曲线，超过遗忘时间点还未被Recall的记忆自动降权

通过上述四层架构，你的AI Agent将拥有接近人类”记住→理解→应用”的记忆能力。这是构建真正有用Agent的基础能力。