Agent 记忆系统深度解析

一、为什么 Agent 需要记忆系统

传统 LLM 的上下文窗口有限，每次对话都是"一次性"的——对话结束后模型就"忘记"了之前的内容。Agent 要完成复杂的长周期任务，必须拥有记忆能力。

无记忆的 LLM：        有记忆的 Agent：
用户→问→答→结束       用户→问→做→记→再问→回忆→再做→完成
<link rel="stylesheet" href="/assets/css/code-blocks.css?v=20261019_04">

1.1 记忆的三大功能

功能	说明	类比
保持上下文	跟踪当前任务的进展状态	工作中的便签纸
积累经验	从过去任务中学习	个人的笔记本
个性化	记住用户偏好和习惯	老朋友了解你

二、记忆的层级结构

2.1 感官记忆（Sensory Memory）

最短暂的记忆层，保存原始感知输入（毫秒级）。在 Agent 中对应原始的用户输入和系统事件的缓冲。

2.2 短期记忆（Short-Term Memory）

当前任务周期的上下文，对应 LLM 的 上下文窗口（Context Window）。

class ShortTermMemory:
    """基于上下文窗口的短期记忆"""

    def __init__(self, max_tokens: int = 128000):
        self.max_tokens = max_tokens
        self.messages = []
        self.token_count = 0

    def add(self, message: Dict):
        """添加消息到短期记忆"""
        tokens = estimate_tokens(message["content"])

        # 如果超限，丢弃最早的非系统消息
        while self.token_count + tokens > self.max_tokens:
            removed = self._pop_earliest_non_system()
            if removed:
                self.token_count -= estimate_tokens(removed["content"])
            else:
                break

        self.messages.append(message)
        self.token_count += tokens

    def _pop_earliest_non_system(self) -> Optional[Dict]:
        """移除最早的非 system 消息"""
        for i, msg in enumerate(self.messages):
            if msg["role"] != "system":
                return self.messages.pop(i)
        return None

    def get_context(self) -> List[Dict]:
        return self.messages

2.3 长期记忆（Long-Term Memory）

持久化存储，跨会话保留。通常使用向量数据库实现。

短期记忆：当前对话 ≈ LLM 上下文窗口
长期记忆：过往经验 ≈ 向量数据库检索
工作记忆：当前任务状态 ≈ 结构化 JSON

三、基于向量数据库的长期记忆

3.1 核心流程

文本 → Embedding 向量 → 存入向量数据库
查询 → Embedding 向量 → 相似度检索 → 返回相关片段

3.2 实现示例

import chromadb
from openai import OpenAI

class LongTermMemory:
    """基于向量数据库的长期记忆"""

    def __init__(self, collection_name: str = "agent_memory"):
        self.client = chromadb.Client()
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}
        )
        self.llm = OpenAI()

    def store(self, content: str, metadata: dict = None):
        """存储一条记忆"""
        embedding = self._get_embedding(content)
        doc_id = f"mem_{int(time.time())}_{random.randint(1000, 9999)}"

        self.collection.add(
            documents=[content],
            embeddings=[embedding],
            metadatas=[metadata or {"timestamp": time.time()}],
            ids=[doc_id]
        )

    def recall(self, query: str, k: int = 5) -> List[str]:
        """检索相关记忆"""
        embedding = self._get_embedding(query)
        results = self.collection.query(
            query_embeddings=[embedding],
            n_results=k
        )
        return results["documents"][0] if results["documents"] else []

    def _get_embedding(self, text: str) -> List[float]:
        response = self.llm.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding

    def forget(self, older_than: float):
        """清理过期记忆"""
        self.collection.delete(
            where={"timestamp": {"$lt": older_than}}
        )

四、记忆管理策略

4.1 记忆的重要性评分

不是所有信息都值得长期存储。引入重要性评分机制：

def calculate_importance(content: str, llm) -> float:
    """评估信息的重要性 (0.0 ~ 1.0)"""
    response = llm.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "评估下面这条信息对 Agent 长期帮助的重要性，"
                       "返回 0.0 到 1.0 之间的数字。"
                       "只返回数字，不要其他文字。"
        }, {
            "role": "user",
            "content": content
        }]
    )
    return float(response.choices[0].message.content)

高重要性（> 0.8）： 用户偏好、关键决策、错误教训
中等重要性（0.4 ~ 0.8）： 中间结果、工具使用经验
低重要性（< 0.4）： 临时状态、冗余信息 → 丢弃

4.2 记忆压缩与提炼

当短期记忆接近上限时，进行压缩：

def compress_memory(messages: List[Dict], llm) -> str:
    """压缩多条消息为一条摘要"""
    full_text = "\n".join(m["content"] for m in messages if m["content"])

    response = llm.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "将以下对话压缩为一段简洁的摘要，"
                       "保留所有关键信息（用户需求、已完成的步骤、"
                       "重要决策）。控制在 500 字以内。"
        }, {
            "role": "user",
            "content": full_text
        }]
    )
    return response.choices[0].message.content

4.3 分层检索策略

查询 → 先查短期记忆（精确匹配）
    ↓ 未找到
    查长期记忆（语义相似度）
    ↓ 未找到
    查外部知识库（RAG）

五、记忆与 RAG 的区别

维度	Agent 记忆	RAG（检索增强生成）
数据来源	自身交互历史	外部文档库
更新频率	每次交互后更新	定期批量更新
用途	保持上下文连贯性	补充领域知识
存储内容	决策、偏好、经验	文档、知识片段
检索方式	重要性 + 时效性加权	单纯语义相似度

六、实践建议

6.1 记忆系统设计检查清单

[ ] 短期记忆是否有容量上限保护？
[ ] 长期记忆是否设置了过期策略？
[ ] 是否实现了重要性评分机制？
[ ] 检索结果是否经过重排序？
[ ] 敏感信息是否做了脱敏处理？

6.2 常见陷阱

记忆污染：低质量信息冲淡有用记忆 → 严格评分阈值
检索噪声：检索到无关内容导致混乱 → 提高相似度阈值
记忆爆炸：无限增长的存储 → 定期归档和清理
隐私泄露：记忆中包含敏感数据 → 存储前脱敏

七、总结

记忆系统是 Agent 从"无状态工具"进化为"有经验助手"的关键组件。一个成熟的记忆系统需要：

分层架构：短期 + 长期 + 工作记忆
智能存储：基于重要性评分决定存储策略
高效检索：混合检索 + 重排序
生命周期管理：写入→检索→更新→遗忘的完整闭环

下一篇将深入讲解 Agent 的工具调用与 Function Calling。

目录