AI Agent记忆系统设计完全教程

从零构建具备长期记忆能力的智能Agent，掌握记忆架构设计的核心原理与工程实践

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

大语言模型（LLM）本质上是无状态的——每次API调用都是独立的，模型不会"记住"之前的对话。但一个真正有用的AI Agent必须能够：

• 记住用户偏好：用户说"我喜欢简洁风格的回答"，下次就不要啰嗦
• 积累任务经验：上次执行某类任务失败了，这次应该避免同样的错误
• 维护长期关系：跨越多天、多周的对话，保持上下文连贯
• 学习领域知识：在与用户的交互中不断积累专业知识

这就是记忆系统存在的意义。它让Agent从"一次性工具"进化为"持续学习的助手"。

二、记忆系统的认知科学基础

人类记忆并非单一系统，而是由多个子系统协同工作。认知科学将记忆分为以下几类，这也是我们设计Agent记忆系统的理论基础：

这个分类不是学术游戏——每种记忆类型对应不同的存储策略、检索算法和更新机制。

三、记忆系统总体架构

一个完整的Agent记忆系统通常包含以下层次：

┌─────────────────────────────────────────────┐
│              Agent推理引擎                    │
├─────────────────────────────────────────────┤
│  工作记忆 (Working Memory)                    │
│  - 当前任务上下文                              │
│  - 中间推理结果                               │
│  - 工具调用状态                               │
├─────────────────────────────────────────────┤
│  记忆检索层 (Retrieval Layer)                  │
│  - 相关性评分                                 │
│  - 时间衰减                                  │
│  - 重要性加权                                 │
├──────┬──────────┬───────────┬────────────────┤
│短期记忆│  情景记忆  │  语义记忆   │  程序性记忆     │
│(对话流) │ (事件序列) │ (知识图谱)  │ (技能/工具)     │
├──────┴──────────┴───────────┴────────────────┤
│              持久化存储层                       │
│  向量数据库 | 关系数据库 | 图数据库 | 文件系统     │
└─────────────────────────────────────────────┘

四、短期记忆：对话上下文管理

4.1 基础实现：滑动窗口

最简单的短期记忆策略——保留最近N轮对话：

from typing import List, Dict
from dataclasses import dataclass, field
from datetime import datetime


@dataclass
class Message:
    role: str          # "user", "assistant", "system"
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    token_count: int = 0


class SlidingWindowMemory:
    """滑动窗口短期记忆管理器"""

    def __init__(self, max_tokens: int = 4096, max_turns: int = 20):
        self.max_tokens = max_tokens
        self.max_turns = max_turns
        self.messages: List[Message] = []

    def add_message(self, role: str, content: str, token_count: int = 0):
        msg = Message(role=role, content=content, token_count=token_count)
        self.messages.append(msg)
        self._trim()

    def _trim(self):
        """裁剪到token和轮次限制内"""
        # 先按轮次裁剪
        if len(self.messages) > self.max_turns * 2:
            self.messages = self.messages[-(self.max_turns * 2):]

        # 再按token裁剪
        total_tokens = sum(m.token_count for m in self.messages)
        while total_tokens > self.max_tokens and len(self.messages) > 2:
            removed = self.messages.pop(0)
            total_tokens -= removed.token_count

    def get_context(self) -> List[Dict[str, str]]:
        return [{"role": m.role, "content": m.content} for m in self.messages]


# 使用示例
memory = SlidingWindowMemory(max_tokens=4000, max_turns=10)
memory.add_message("user", "帮我写一个Python排序算法")
memory.add_message("assistant", "好的，以下是快速排序实现...")
memory.add_message("user", "能改成归并排序吗？")

context = memory.get_context()  # 返回最近的对话历史

4.2 进阶：重要性感知的滑动窗口

简单滑动窗口的问题是：重要信息可能被丢弃。改进方案是为每条消息打重要性分数：

import hashlib


class ImportanceAwareMemory(SlidingWindowMemory):
    """带重要性评分的滑动窗口记忆"""

    # 关键词权重映射
    KEYWORD_WEIGHTS = {
        "记住": 2.0, "重要": 2.0, "密码": 3.0, "偏好": 1.8,
        "总是": 1.5, "永远": 1.5, "不要": 1.3, "喜欢": 1.5,
        "用户名": 2.0, "地址": 1.8, "配置": 1.5,
    }

    def calculate_importance(self, message: Message) -> float:
        """计算消息重要性分数 (0-1)"""
        score = 0.5  # 基础分

        # 1. 关键词加分
        for keyword, weight in self.KEYWORD_WEIGHTS.items():
            if keyword in message.content:
                score += 0.1 * weight

        # 2. 用户消息比助手消息更重要
        if message.role == "user":
            score += 0.1

        # 3. 较长的回复通常包含更多信息
        if len(message.content) > 200:
            score += 0.1

        # 4. 包含数字/代码的消息通常更具体
        if any(c.isdigit() for c in message.content):
            score += 0.05

        return min(score, 1.0)

    def _trim(self):
        """按重要性裁剪，保留高重要性消息"""
        if sum(m.token_count for m in self.messages) <= self.max_tokens:
            return

        # 计算每条消息的重要性
        scored = [(self.calculate_importance(m), i, m)
                  for i, m in enumerate(self.messages)]

        # 始终保留最近2轮对话
        recent_count = 4
        recent = self.messages[-recent_count:]
        older = self.messages[:-recent_count]

        # 按重要性排序旧消息，保留高分的
        older_scored = [(self.calculate_importance(m), i, m)
                        for i, m in enumerate(older)]
        older_scored.sort(key=lambda x: x[0], reverse=True)

        # 贪心选择：从最重要的开始保留，直到token限制
        selected = []
        remaining_tokens = self.max_tokens - sum(m.token_count for m in recent)

        for score, _, msg in older_scored:
            if msg.token_count <= remaining_tokens:
                selected.append(msg)
                remaining_tokens -= msg.token_count

        # 按原始顺序排列
        selected.sort(key=lambda m: m.timestamp)
        self.messages = selected + recent

五、长期记忆：向量数据库驱动的持久记忆

5.1 向量记忆的基本原理

长期记忆的核心挑战是语义检索——不是关键词匹配，而是理解"意思相近"的内容。向量嵌入（Embedding）是解决这个问题的关键技术。

工作流程：

1. 将记忆内容通过Embedding模型转为向量
2. 存储到向量数据库
3. 查询时将查询转为向量，计算相似度
4. 返回最相关的记忆片段

import numpy as np
from typing import List, Tuple, Optional
from dataclasses import dataclass
import json
import time


@dataclass
class MemoryItem:
    """一条记忆单元"""
    id: str
    content: str
    embedding: List[float]
    metadata: dict  # 时间戳、来源、标签等
    importance: float = 0.5
    access_count: int = 0
    last_accessed: float = 0.0


class VectorMemoryStore:
    """基于向量的长期记忆存储"""

    def __init__(self, embedding_dim: int = 1536):
        self.embedding_dim = embedding_dim
        self.memories: List[MemoryItem] = []
        self._index_dirty = True
        self._vectors_matrix: Optional[np.ndarray] = None

    def _get_embedding(self, text: str) -> List[float]:
        """获取文本的向量嵌入（示意，实际调用Embedding API）"""
        # 实际项目中调用 OpenAI / 本地模型
        # from openai import OpenAI
        # client = OpenAI()
        # resp = client.embeddings.create(input=text, model="text-embedding-3-small")
        # return resp.data[0].embedding

        # 模拟：返回随机向量（仅用于演示）
        np.random.seed(hash(text) % 2**32)
        return np.random.randn(self.embedding_dim).tolist()

    def add_memory(self, content: str, metadata: dict = None,
                   importance: float = 0.5) -> str:
        """添加一条记忆"""
        memory_id = f"mem_{int(time.time()*1000)}_{hash(content) % 10000}"
        embedding = self._get_embedding(content)

        item = MemoryItem(
            id=memory_id,
            content=content,
            embedding=embedding,
            metadata=metadata or {},
            importance=importance,
            last_accessed=time.time()
        )

        self.memories.append(item)
        self._index_dirty = True
        return memory_id

    def search(self, query: str, top_k: int = 5,
               time_decay: float = 0.01,
               min_importance: float = 0.0) -> List[Tuple[MemoryItem, float]]:
        """搜索相关记忆，综合考虑语义相似度、时间衰减、重要性"""
        if not self.memories:
            return []

        query_embedding = np.array(self._get_embedding(query))

        # 构建向量矩阵（缓存优化）
        if self._index_dirty or self._vectors_matrix is None:
            self._vectors_matrix = np.array([m.embedding for m in self.memories])
            self._index_dirty = False

        # 余弦相似度
        norms = np.linalg.norm(self._vectors_matrix, axis=1) * np.linalg.norm(query_embedding)
        norms = np.where(norms == 0, 1, norms)
        similarities = np.dot(self._vectors_matrix, query_embedding) / norms

        # 综合评分
        now = time.time()
        results = []
        for i, mem in enumerate(self.memories):
            if mem.importance < min_importance:
                continue

            # 时间衰减因子：越久远的记忆分数越低
            hours_since_access = (now - mem.last_accessed) / 3600
            decay = np.exp(-time_decay * hours_since_access)

            # 综合分数 = 相似度 * 0.6 + 重要性 * 0.2 + 时间衰减 * 0.2
            final_score = (
                similarities[i] * 0.6 +
                mem.importance * 0.2 +
                decay * 0.2
            )
            results.append((mem, float(final_score)))

        # 按分数排序
        results.sort(key=lambda x: x[1], reverse=True)

        # 更新访问记录
        for mem, _ in results[:top_k]:
            mem.access_count += 1
            mem.last_accessed = now

        return results[:top_k]


# 使用示例
store = VectorMemoryStore()

# 添加记忆
store.add_memory("用户喜欢使用Python进行数据分析", {"type": "preference"}, importance=0.8)
store.add_memory("用户的项目使用FastAPI框架", {"type": "project"}, importance=0.7)
store.add_memory("上次讨论了机器学习模型部署方案", {"type": "conversation"}, importance=0.6)
store.add_memory("用户提到他的团队有5个人", {"type": "personal"}, importance=0.5)

# 检索相关记忆
results = store.search("数据分析用什么工具好？", top_k=3)
for mem, score in results:
    print(f"[{score:.3f}] {mem.content}")

5.2 使用ChromaDB实现生产级向量记忆

上面是手动实现，生产环境推荐使用ChromaDB：

import chromadb
from chromadb.config import Settings


class ChromaMemoryStore:
    """基于ChromaDB的长期记忆存储"""

    def __init__(self, collection_name: str = "agent_memory",
                 persist_directory: str = "./memory_db"):
        self.client = chromadb.Client(Settings(
            chroma_db_impl="duckdb+parquet",
            persist_directory=persist_directory,
            anonymized_telemetry=False
        ))
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}
        )

    def add(self, content: str, memory_id: str = None,
            metadata: dict = None) -> str:
        import uuid
        mid = memory_id or str(uuid.uuid4())
        self.collection.add(
            documents=[content],
            ids=[mid],
            metadatas=[metadata or {}]
        )
        return mid

    def search(self, query: str, top_k: int = 5,
               where: dict = None) -> list:
        kwargs = {
            "query_texts": [query],
            "n_results": top_k
        }
        if where:
            kwargs["where"] = where

        results = self.collection.query(**kwargs)

        return [
            {"id": id_, "content": doc, "metadata": meta, "distance": dist}
            for id_, doc, meta, dist in zip(
                results["ids"][0],
                results["documents"][0],
                results["metadatas"][0],
                results["distances"][0]
            )
        ]

    def delete(self, memory_id: str):
        self.collection.delete(ids=[memory_id])

    def update(self, memory_id: str, content: str = None,
               metadata: dict = None):
        kwargs = {"ids": [memory_id]}
        if content:
            kwargs["documents"] = [content]
        if metadata:
            kwargs["metadatas"] = [metadata]
        self.collection.update(**kwargs)

六、情景记忆：事件序列与时间线

情景记忆记录的是"发生了什么"——按时间顺序组织的事件序列。它对Agent理解上下文、追踪任务进度至关重要。

from datetime import datetime, timedelta
from typing import List, Optional
from enum import Enum
import json


class EventType(Enum):
    CONVERSATION = "conversation"
    TASK_START = "task_start"
    TASK_COMPLETE = "task_complete"
    TOOL_CALL = "tool_call"
    ERROR = "error"
    USER_FEEDBACK = "user_feedback"
    DECISION = "decision"


class EpisodicMemory:
    """情景记忆管理器 - 按时间线记录事件序列"""

    def __init__(self, max_episodes: int = 1000):
        self.episodes: List[dict] = []
        self.max_episodes = max_episodes
        self.current_session_id = None

    def start_session(self, session_id: str):
        self.current_session_id = session_id
        self.add_event(EventType.CONVERSATION, "新会话开始",
                       {"session_id": session_id})

    def add_event(self, event_type: EventType, summary: str,
                  details: dict = None, importance: float = 0.5):
        event = {
            "id": f"evt_{len(self.episodes)}",
            "session_id": self.current_session_id,
            "type": event_type.value,
            "summary": summary,
            "details": details or {},
            "importance": importance,
            "timestamp": datetime.now().isoformat(),
        }
        self.episodes.append(event)

        # 裁剪旧事件
        if len(self.episodes) > self.max_episodes:
            self.episodes = self.episodes[-self.max_episodes:]

    def get_timeline(self, hours: int = 24,
                     event_types: List[EventType] = None) -> List[dict]:
        """获取指定时间范围内的事件时间线"""
        cutoff = datetime.now() - timedelta(hours=hours)
        cutoff_str = cutoff.isoformat()

        filtered = [e for e in self.episodes if e["timestamp"] >= cutoff_str]

        if event_types:
            type_values = {t.value for t in event_types}
            filtered = [e for e in filtered if e["type"] in type_values]

        return filtered

    def get_session_summary(self, session_id: str = None) -> str:
        """生成会话摘要"""
        sid = session_id or self.current_session_id
        events = [e for e in self.episodes if e["session_id"] == sid]

        if not events:
            return "无事件记录"

        # 按类型统计
        type_counts = {}
        for e in events:
            type_counts[e["type"]] = type_counts.get(e["type"], 0) + 1

        summary_parts = []
        summary_parts.append(f"会话包含 {len(events)} 个事件")

        if "task_complete" in type_counts:
            summary_parts.append(f"完成 {type_counts['task_complete']} 个任务")
        if "error" in type_counts:
            summary_parts.append(f"遇到 {type_counts['error']} 个错误")
        if "tool_call" in type_counts:
            summary_parts.append(f"调用了 {type_counts['tool_call']} 次工具")

        # 提取重要事件
        important = [e for e in events if e["importance"] > 0.7]
        if important:
            summary_parts.append("重要事件：" + "；".join(
                e["summary"] for e in important[:5]
            ))

        return "。".join(summary_parts) + "。"

    def find_similar_episodes(self, description: str,
                              top_k: int = 5) -> List[dict]:
        """基于描述查找相似的历史情景（简化版）"""
        # 实际项目中用向量相似度搜索
        keywords = set(description.lower().split())
        scored = []
        for ep in self.episodes:
            ep_words = set(ep["summary"].lower().split())
            overlap = len(keywords & ep_words)
            if overlap > 0:
                scored.append((overlap, ep))

        scored.sort(key=lambda x: x[0], reverse=True)
        return [ep for _, ep in scored[:top_k]]


# 使用示例
episodic = EpisodicMemory()
episodic.start_session("session_001")

episodic.add_event(EventType.TASK_START, "开始处理用户的数据分析请求",
                   {"task": "data_analysis"}, importance=0.7)
episodic.add_event(EventType.TOOL_CALL, "调用pandas读取CSV文件",
                   {"tool": "pandas", "file": "sales.csv"}, importance=0.5)
episodic.add_event(EventType.ERROR, "CSV文件编码错误，尝试UTF-8",
                   {"error": "UnicodeDecodeError"}, importance=0.8)
episodic.add_event(EventType.DECISION, "切换为GBK编码成功读取",
                   {"encoding": "gbk"}, importance=0.6)
episodic.add_event(EventType.TASK_COMPLETE, "数据分析完成，生成报告",
                   {"output": "report.html"}, importance=0.9)

# 获取会话摘要
print(episodic.get_session_summary())

# 获取最近6小时的事件
timeline = episodic.get_timeline(hours=6)

七、语义记忆：知识图谱

语义记忆存储的是结构化的知识——实体之间的关系、概念的定义、因果关系等。

from typing import Dict, Set, Tuple, List
from collections import defaultdict
import json


class SemanticMemory:
    """基于图结构的语义记忆"""

    def __init__(self):
        # 邻接表: entity -> {relation -> {target_entities}}
        self.graph: Dict[str, Dict[str, Set[str]]] = defaultdict(
            lambda: defaultdict(set)
        )
        # 实体属性
        self.entity_attrs: Dict[str, dict] = defaultdict(dict)
        # 反向索引: 用于快速查找
        self._reverse_index: Dict[str, Set[str]] = defaultdict(set)

    def add_relation(self, subject: str, relation: str, obj: str):
        """添加三元组关系：主体 -关系-> 客体"""
        self.graph[subject][relation].add(obj)
        self._reverse_index[obj].add(subject)
        self._reverse_index[subject].add(obj)

    def add_entity(self, entity: str, attributes: dict = None):
        """添加实体及其属性"""
        if attributes:
            self.entity_attrs[entity].update(attributes)

    def query_relations(self, subject: str, relation: str = None) -> dict:
        """查询实体的关系"""
        if subject not in self.graph:
            return {}

        if relation:
            return {relation: list(self.graph[subject].get(relation, set()))}

        return {rel: list(targets)
                for rel, targets in self.graph[subject].items()}

    def find_path(self, start: str, end: str,
                  max_depth: int = 3) -> List[List[Tuple[str, str, str]]]:
        """查找两个实体之间的关系路径（BFS）"""
        if start == end:
            return [[(start, "self", start)]]

        visited = {start}
        queue = [[(start, "", start)]]  # 路径列表

        for _ in range(max_depth):
            next_queue = []
            for path in queue:
                current = path[-1][0]  # 当前节点

                for relation, targets in self.graph.get(current, {}).items():
                    for target in targets:
                        if target == end:
                            return [path + [(current, relation, target)]]

                        if target not in visited:
                            visited.add(target)
                            next_queue.append(
                                path + [(current, relation, target)]
                            )

            queue = next_queue
            if not queue:
                break

        return []  # 未找到路径

    def get_subgraph(self, entity: str, depth: int = 2) -> dict:
        """获取实体的局部子图"""
        subgraph = {"nodes": set(), "edges": []}
        frontier = {entity}

        for _ in range(depth):
            next_frontier = set()
            for node in frontier:
                if node in subgraph["nodes"]:
                    continue
                subgraph["nodes"].add(node)

                for relation, targets in self.graph.get(node, {}).items():
                    for target in targets:
                        subgraph["edges"].append((node, relation, target))
                        next_frontier.add(target)

            frontier = next_frontier - subgraph["nodes"]

        subgraph["nodes"] = list(subgraph["nodes"])
        return subgraph

    def to_prompt_context(self, entity: str, max_relations: int = 10) -> str:
        """将实体相关知识转为可注入prompt的文本"""
        parts = []

        # 实体属性
        if entity in self.entity_attrs:
            attrs = self.entity_attrs[entity]
            parts.append(f"关于{entity}的信息：" + ", ".join(
                f"{k}是{v}" for k, v in attrs.items()
            ))

        # 关系
        relations = self.query_relations(entity)
        count = 0
        for rel, targets in relations.items():
            for target in targets:
                parts.append(f"{entity} {rel} {target}")
                count += 1
                if count >= max_relations:
                    break

        return "\n".join(parts) if parts else f"没有关于{entity}的记录"


# 使用示例
semantic = SemanticMemory()

# 构建知识图谱
semantic.add_relation("Python", "是一种", "编程语言")
semantic.add_relation("Python", "常用于", "数据分析")
semantic.add_relation("Python", "常用于", "机器学习")
semantic.add_relation("FastAPI", "是框架", "Python")
semantic.add_relation("FastAPI", "依赖", "Pydantic")
semantic.add_relation("用户A", "使用", "Python")
semantic.add_relation("用户A", "负责", "数据平台项目")
semantic.add_relation("数据平台项目", "使用", "FastAPI")
semantic.add_relation("数据平台项目", "部署在", "阿里云")

# 添加实体属性
semantic.add_entity("用户A", {"角色": "后端工程师", "团队": "数据平台"})

# 查询关系
print(semantic.query_relations("Python"))
# {'是一种': ['编程语言'], '常用于': ['数据分析', '机器学习']}

# 查找路径
path = semantic.find_path("用户A", "阿里云")
for step in path:
    print(f"  {step[0]} --[{step[1]}]--> {step[2]}")

# 生成prompt上下文
context = semantic.to_prompt_context("用户A")
print(context)

八、工作记忆：任务上下文管理

工作记忆是Agent当前正在处理的信息，类似于人脑的"前台"：

from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum
import time


class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    PAUSED = "paused"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class TaskContext:
    """单个任务的工作记忆"""
    task_id: str
    description: str
    status: TaskStatus = TaskStatus.PENDING
    goal: str = ""
    constraints: List[str] = field(default_factory=list)
    intermediate_results: dict = field(default_factory=dict)
    variables: dict = field(default_factory=dict)
    history: List[dict] = field(default_factory=list)
    created_at: float = field(default_factory=time.time)
    deadline: Optional[float] = None

    def add_step(self, action: str, result: Any, success: bool = True):
        self.history.append({
            "action": action,
            "result": str(result)[:500],  # 截断过长结果
            "success": success,
            "timestamp": time.time()
        })

    def set_variable(self, key: str, value: Any):
        self.variables[key] = value

    def get_variable(self, key: str, default: Any = None) -> Any:
        return self.variables.get(key, default)


class WorkingMemory:
    """工作记忆管理器 - 管理当前任务栈"""

    def __init__(self):
        self.task_stack: List[TaskContext] = []
        self.global_context: dict = {}  # 跨任务的全局上下文

    def create_task(self, task_id: str, description: str,
                    goal: str = "", constraints: List[str] = None) -> TaskContext:
        task = TaskContext(
            task_id=task_id,
            description=description,
            goal=goal,
            constraints=constraints or []
        )
        task.status = TaskStatus.IN_PROGRESS
        self.task_stack.append(task)
        return task

    @property
    def current_task(self) -> Optional[TaskContext]:
        return self.task_stack[-1] if self.task_stack else None

    def complete_current_task(self, result: Any = None):
        if self.current_task:
            if result is not None:
                self.current_task.intermediate_results["final"] = result
            self.current_task.status = TaskStatus.COMPLETED

    def push_subtask(self, task_id: str, description: str) -> TaskContext:
        """创建子任务（压栈）"""
        return self.create_task(task_id, description)

    def pop_subtask(self):
        """完成子任务（出栈）"""
        if len(self.task_stack) > 1:
            self.complete_current_task()
            return self.task_stack.pop()
        return None

    def to_prompt_context(self) -> str:
        """将当前工作记忆转为prompt上下文"""
        parts = []

        # 当前任务
        task = self.current_task
        if task:
            parts.append(f"当前任务：{task.description}")
            if task.goal:
                parts.append(f"目标：{task.goal}")
            if task.constraints:
                parts.append("约束条件：" + "；".join(task.constraints))

            # 已完成的步骤
            if task.history:
                parts.append("已完成步骤：")
                for step in task.history[-5:]:  # 最近5步
                    status = "✓" if step["success"] else "✗"
                    parts.append(f"  {status} {step['action']}")

            # 中间结果
            if task.intermediate_results:
                parts.append("中间结果：")
                for k, v in task.intermediate_results.items():
                    parts.append(f"  {k}: {str(v)[:200]}")

        return "\n".join(parts)

九、记忆检索策略

9.1 多维评分模型

好的记忆检索不是简单的相似度匹配，需要综合多个维度：

import math
from datetime import datetime


class MemoryScorer:
    """多维记忆评分器"""

    def __init__(self, recency_weight=0.3, relevance_weight=0.4,
                 importance_weight=0.2, frequency_weight=0.1):
        self.weights = {
            "recency": recency_weight,
            "relevance": relevance_weight,
            "importance": importance_weight,
            "frequency": frequency_weight,
        }

    def score_recency(self, created_at: float, half_life_hours: float = 24) -> float:
        """基于时间衰减的分数（指数衰减）"""
        hours_ago = (time.time() - created_at) / 3600
        return math.exp(-0.693 * hours_ago / half_life_hours)

    def score_relevance(self, query_embedding: list,
                        memory_embedding: list) -> float:
        """基于向量相似度的相关性分数"""
        import numpy as np
        q = np.array(query_embedding)
        m = np.array(memory_embedding)
        dot = np.dot(q, m)
        norm = np.linalg.norm(q) * np.linalg.norm(m)
        return float(dot / norm) if norm > 0 else 0.0

    def score_importance(self, memory: dict) -> float:
        """基于内容特征的重要性分数"""
        return memory.get("importance", 0.5)

    def score_frequency(self, access_count: int) -> float:
        """基于访问频率的分数（对数归一化）"""
        return min(1.0, math.log1p(access_count) / 5.0)

    def final_score(self, scores: dict) -> float:
        """加权综合分数"""
        return sum(
            scores[dim] * weight
            for dim, weight in self.weights.items()
        )

9.2 RAG风格的记忆检索

将记忆检索与RAG（检索增强生成）结合：

class RAGMemoryRetriever:
    """RAG风格的记忆检索器"""

    def __init__(self, vector_store, episodic_memory, semantic_memory):
        self.vector_store = vector_store
        self.episodic = episodic_memory
        self.semantic = semantic_memory

    def retrieve(self, query: str, top_k: int = 5) -> str:
        """综合检索各类型记忆，生成注入prompt的上下文"""
        context_parts = []

        # 1. 向量记忆检索
        vector_results = self.vector_store.search(query, top_k=top_k)
        if vector_results:
            context_parts.append("=== 相关记忆 ===")
            for mem, score in vector_results:
                context_parts.append(f"[相关度:{score:.2f}] {mem.content}")

        # 2. 情景记忆检索（最近相关事件）
        episodes = self.episodic.find_similar_episodes(query, top_k=3)
        if episodes:
            context_parts.append("\n=== 历史事件 ===")
            for ep in episodes:
                context_parts.append(
                    f"[{ep['timestamp'][:16]}] {ep['summary']}"
                )

        # 3. 语义记忆（提取查询中的实体）
        entities = self._extract_entities(query)
        for entity in entities:
            entity_context = self.semantic.to_prompt_context(entity)
            if entity_context:
                context_parts.append(f"\n=== 关于{entity}的知识 ===")
                context_parts.append(entity_context)

        return "\n".join(context_parts)

    def _extract_entities(self, text: str) -> list:
        """简单实体提取（实际项目用NER模型）"""
        # 已知实体列表中查找
        known = set()
        for entity in self.semantic.graph.keys():
            if entity in text:
                known.add(entity)
        return list(known)

十、记忆压缩与摘要

随着记忆积累，存储和检索成本都会增加。记忆压缩是必要的优化手段：

class MemoryCompressor:
    """记忆压缩器 - 通过摘要减少存储量"""

    def __init__(self, llm_call=None):
        # llm_call: function(prompt: str) -> str
        self.llm_call = llm_call or self._default_summarize

    def _default_summarize(self, text: str) -> str:
        """默认摘要（截断）"""
        return text[:200] + "..." if len(text) > 200 else text

    def compress_conversation(self, messages: list,
                              target_length: int = 500) -> str:
        """压缩对话历史为摘要"""
        conversation = "\n".join(
            f"{m['role']}: {m['content']}" for m in messages
        )

        prompt = f"""请将以下对话压缩为{target_length}字以内的摘要，保留关键信息、
决策、结论和待办事项：

{conversation}

摘要："""

        return self.llm_call(prompt)

    def merge_memories(self, memories: list) -> dict:
        """合并多条相似记忆为一条"""
        contents = [m.content for m in memories]
        merged_content = self.llm_call(
            "请合并以下相似记忆为一条综合记忆，保留所有重要信息：\n" +
            "\n---\n".join(contents)
        )

        # 取最高重要性和最新时间
        max_importance = max(m.importance for m in memories)
        latest_time = max(m.last_accessed for m in memories)

        return {
            "content": merged_content,
            "importance": max_importance,
            "last_accessed": latest_time,
            "merged_from": [m.id for m in memories],
            "access_count": sum(m.access_count for m in memories),
        }

    def periodic_cleanup(self, store, threshold: float = 0.8):
        """定期清理：合并相似、删除低价值记忆"""
        # 找出相似记忆对
        all_memories = store.memories
        to_merge = []
        merged_ids = set()

        for i, m1 in enumerate(all_memories):
            if m1.id in merged_ids:
                continue
            group = [m1]
            for j, m2 in enumerate(all_memories[i+1:], i+1):
                if m2.id in merged_ids:
                    continue
                similarity = self._compute_similarity(m1, m2)
                if similarity > threshold:
                    group.append(m2)
                    merged_ids.add(m2.id)

            if len(group) > 1:
                to_merge.append(group)

        # 执行合并
        for group in to_merge:
            merged = self.merge_memories(group)
            # 删除旧的，添加新的
            for mem in group:
                store.delete_memory(mem.id)
            store.add_memory(merged["content"], merged.get("metadata", {}),
                             merged["importance"])

十一、记忆冲突解决

当新信息与旧记忆矛盾时，需要有明确的冲突解决策略：

from enum import Enum


class ConflictStrategy(Enum):
    NEWEST_WINS = "newest"      # 最新信息优先
    HIGHEST_IMPORTANCE = "importance"  # 高重要性优先
    MOST_ACCESSED = "frequent"  # 最常访问的优先
    MERGE = "merge"             # 尝试合并


class MemoryConflictResolver:
    """记忆冲突解决器"""

    def __init__(self, strategy: ConflictStrategy = ConflictStrategy.NEWEST_WINS):
        self.strategy = strategy
        self.conflict_log: List[dict] = []

    def detect_conflict(self, new_content: str,
                        existing_memories: list) -> list:
        """检测新内容是否与已有记忆冲突"""
        conflicts = []
        # 简化实现：基于关键词对比
        # 实际项目中用NLI（自然语言推理）模型判断矛盾

        negation_pairs = [
            ("喜欢", "不喜欢"), ("是", "不是"),
            ("能", "不能"), ("会", "不会"),
            ("要", "不要"), ("使用", "不使用"),
        ]

        for mem in existing_memories:
            for pos, neg in negation_pairs:
                if (pos in new_content and neg in mem.content) or \
                   (neg in new_content and pos in mem.content):
                    conflicts.append({
                        "memory": mem,
                        "conflict_type": "contradiction",
                        "detail": f"'{new_content}' 与 '{mem.content}' 矛盾"
                    })

        return conflicts

    def resolve(self, new_content: str, conflicts: list,
                new_importance: float = 0.5) -> dict:
        """解决冲突"""
        if not conflicts:
            return {"action": "add", "content": new_content}

        self.conflict_log.extend([{
            "timestamp": datetime.now().isoformat(),
            "new": new_content,
            "conflicting": c["detail"]
        } for c in conflicts])

        if self.strategy == ConflictStrategy.NEWEST_WINS:
            return {
                "action": "replace",
                "content": new_content,
                "replaced_ids": [c["memory"].id for c in conflicts]
            }

        elif self.strategy == ConflictStrategy.HIGHEST_IMPORTANCE:
            max_old_imp = max(c["memory"].importance for c in conflicts)
            if new_importance > max_old_imp:
                return {
                    "action": "replace",
                    "content": new_content,
                    "replaced_ids": [c["memory"].id for c in conflicts]
                }
            else:
                return {"action": "skip", "reason": "existing memory more important"}

        elif self.strategy == ConflictStrategy.MERGE:
            return {
                "action": "merge",
                "content": new_content,
                "merge_with": [c["memory"].id for c in conflicts]
            }

        return {"action": "add", "content": new_content}

十二、跨会话记忆持久化

12.1 完整的持久化方案

import sqlite3
import json
from pathlib import Path


class PersistentMemoryStore:
    """基于SQLite的持久化记忆存储"""

    def __init__(self, db_path: str = "agent_memory.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.executescript("""
                CREATE TABLE IF NOT EXISTS memories (
                    id TEXT PRIMARY KEY,
                    content TEXT NOT NULL,
                    memory_type TEXT NOT NULL,
                    importance REAL DEFAULT 0.5,
                    access_count INTEGER DEFAULT 0,
                    last_accessed REAL,
                    created_at REAL,
                    metadata TEXT DEFAULT '{}',
                    embedding BLOB
                );

                CREATE TABLE IF NOT EXISTS episodes (
                    id TEXT PRIMARY KEY,
                    session_id TEXT,
                    event_type TEXT,
                    summary TEXT,
                    details TEXT DEFAULT '{}',
                    importance REAL DEFAULT 0.5,
                    timestamp TEXT
                );

                CREATE TABLE IF NOT EXISTS knowledge_graph (
                    subject TEXT,
                    relation TEXT,
                    object TEXT,
                    confidence REAL DEFAULT 1.0,
                    created_at REAL,
                    PRIMARY KEY (subject, relation, object)
                );

                CREATE INDEX IF NOT EXISTS idx_memories_type
                    ON memories(memory_type);
                CREATE INDEX IF NOT EXISTS idx_episodes_session
                    ON episodes(session_id);
                CREATE INDEX IF NOT EXISTS idx_episodes_timestamp
                    ON episodes(timestamp);
            """)

    def save_memory(self, memory_id: str, content: str,
                    memory_type: str, importance: float = 0.5,
                    metadata: dict = None):
        import time
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                INSERT OR REPLACE INTO memories
                (id, content, memory_type, importance, last_accessed,
                 created_at, metadata)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            """, (memory_id, content, memory_type, importance,
                  time.time(), time.time(),
                  json.dumps(metadata or {}, ensure_ascii=False)))

    def load_memories(self, memory_type: str = None,
                      min_importance: float = 0.0,
                      limit: int = 100) -> list:
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            query = "SELECT * FROM memories WHERE importance >= ?"
            params = [min_importance]

            if memory_type:
                query += " AND memory_type = ?"
                params.append(memory_type)

            query += " ORDER BY last_accessed DESC LIMIT ?"
            params.append(limit)

            rows = conn.execute(query, params).fetchall()
            return [dict(row) for row in rows]

    def save_episode(self, event_id: str, session_id: str,
                     event_type: str, summary: str,
                     details: dict = None, importance: float = 0.5):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                INSERT INTO episodes
                (id, session_id, event_type, summary, details,
                 importance, timestamp)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            """, (event_id, session_id, event_type, summary,
                  json.dumps(details or {}, ensure_ascii=False),
                  importance, datetime.now().isoformat()))

    def export_for_session(self, session_id: str = None) -> dict:
        """导出指定会话的全部记忆用于恢复"""
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row

            memories = conn.execute(
                "SELECT * FROM memories ORDER BY importance DESC LIMIT 50"
            ).fetchall()

            episodes_query = "SELECT * FROM episodes"
            params = []
            if session_id:
                episodes_query += " WHERE session_id = ?"
                params.append(session_id)
            episodes_query += " ORDER BY timestamp DESC LIMIT 100"

            episodes = conn.execute(episodes_query, params).fetchall()

            graph = conn.execute(
                "SELECT * FROM knowledge_graph"
            ).fetchall()

            return {
                "memories": [dict(r) for r in memories],
                "episodes": [dict(r) for r in episodes],
                "knowledge_graph": [dict(r) for r in graph],
            }

十三、开源方案对比

13.1 Mem0

Mem0是目前最流行的Agent记忆管理框架：

# 安装: pip install mem0ai

from mem0 import Memory

# 初始化
m = Memory()

# 添加记忆
m.add("我喜欢使用Python做数据分析", user_id="user_001")
m.add("我的项目部署在AWS上", user_id="user_001")
m.add("我偏好使用VS Code编辑器", user_id="user_001")

# 搜索记忆
results = m.search("用户用什么编程语言？", user_id="user_001")
for result in results["results"]:
    print(f"记忆: {result['memory']}, 分数: {result['score']:.3f}")

# 获取所有记忆
all_memories = m.get_all(user_id="user_001")

# 更新记忆
m.update("mem_id_here", "我喜欢使用Python和Go做开发")

# Mem0还支持自动从对话中提取记忆
messages = [
    {"role": "user", "content": "我最近在学习Rust，感觉很不错"},
    {"role": "assistant", "content": "Rust确实很棒，内存安全又高性能"},
]
m.add(messages, user_id="user_001")  # 自动提取关键信息

13.2 Zep

Zep专注于对话AI的记忆管理：

# Zep的特点：
# 1. 自动摘要和事实提取
# 2. 时间感知的检索
# 3. 支持多种向量数据库后端
# 4. 内置对话分析

# Zep的核心API
# - memory.add() - 添加对话历史
# - memory.search() - 语义搜索
# - memory.get_memory() - 获取增强上下文
# - user.add() - 添加用户
# - session.create() - 创建会话

13.3 方案对比

十四、记忆系统评估

如何衡量记忆系统的好坏？以下是一套评估框架：

class MemoryEvaluator:
    """记忆系统评估器"""

    def __init__(self, memory_system):
        self.memory = memory_system
        self.metrics = {}

    def evaluate_retrieval_accuracy(self, test_cases: list) -> float:
        """评估检索准确率"""
        correct = 0
        total = len(test_cases)

        for query, expected_ids in test_cases:
            results = self.memory.search(query, top_k=5)
            retrieved_ids = {r.id for r, _ in results}
            if set(expected_ids) & retrieved_ids:
                correct += 1

        accuracy = correct / total if total > 0 else 0
        self.metrics["retrieval_accuracy"] = accuracy
        return accuracy

    def evaluate_response_quality(self, test_questions: list) -> dict:
        """评估使用记忆后的回答质量"""
        scores = {
            "with_memory": [],
            "without_memory": []
        }

        for question, reference_answer in test_questions:
            # 有记忆的回答
            context = self.memory.retrieve(question)
            answer_with = self._generate_answer(question, context)
            score_with = self._compute_similarity(answer_with, reference_answer)
            scores["with_memory"].append(score_with)

            # 无记忆的回答
            answer_without = self._generate_answer(question, "")
            score_without = self._compute_similarity(answer_without, reference_answer)
            scores["without_memory"].append(score_without)

        improvement = (
            sum(scores["with_memory"]) / len(scores["with_memory"]) -
            sum(scores["without_memory"]) / len(scores["without_memory"])
        )

        self.metrics["response_improvement"] = improvement
        return {
            "avg_with_memory": sum(scores["with_memory"]) / len(scores["with_memory"]),
            "avg_without_memory": sum(scores["without_memory"]) / len(scores["without_memory"]),
            "improvement": improvement
        }

    def evaluate_memory_growth(self, time_period_days: int = 30) -> dict:
        """评估记忆增长和质量"""
        all_memories = self.memory.get_all()
        total = len(all_memories)
        by_type = {}
        by_importance = {"high": 0, "medium": 0, "low": 0}

        for mem in all_memories:
            mtype = mem.get("type", "unknown")
            by_type[mtype] = by_type.get(mtype, 0) + 1

            imp = mem.get("importance", 0.5)
            if imp > 0.7:
                by_importance["high"] += 1
            elif imp > 0.4:
                by_importance["medium"] += 1
            else:
                by_importance["low"] += 1

        return {
            "total_memories": total,
            "by_type": by_type,
            "by_importance": by_importance,
            "memory_per_day": total / max(time_period_days, 1)
        }

    def run_full_evaluation(self, test_data: dict) -> dict:
        """运行完整评估"""
        results = {}

        if "retrieval_cases" in test_data:
            results["retrieval"] = self.evaluate_retrieval_accuracy(
                test_data["retrieval_cases"]
            )

        if "quality_questions" in test_data:
            results["quality"] = self.evaluate_response_quality(
                test_data["quality_questions"]
            )

        results["growth"] = self.evaluate_memory_growth()
        return results

十五、实战：构建完整的Agent记忆系统

将上述所有组件整合为一个完整的记忆系统：

class AgentMemorySystem:
    """完整的Agent记忆系统"""

    def __init__(self, agent_id: str, config: dict = None):
        self.agent_id = agent_id
        config = config or {}

        # 初始化各记忆子系统
        self.short_term = SlidingWindowMemory(
            max_tokens=config.get("max_context_tokens", 4096),
            max_turns=config.get("max_turns", 20)
        )

        self.long_term = VectorMemoryStore(
            embedding_dim=config.get("embedding_dim", 1536)
        )

        self.episodic = EpisodicMemory(
            max_episodes=config.get("max_episodes", 1000)
        )

        self.semantic = SemanticMemory()

        self.working = WorkingMemory()

        self.persistence = PersistentMemoryStore(
            db_path=config.get("db_path", f"memory_{agent_id}.db")
        )

        self.compressor = MemoryCompressor()
        self.conflict_resolver = MemoryConflictResolver()

        # 恢复历史记忆
        self._restore_memories()

    def _restore_memories(self):
        """从持久化存储恢复记忆"""
        saved = self.persistence.export_for_session()
        for mem in saved.get("memories", []):
            self.long_term.add_memory(
                mem["content"],
                json.loads(mem.get("metadata", "{}")),
                mem.get("importance", 0.5)
            )

    def process_message(self, role: str, content: str) -> dict:
        """处理一条新消息，更新所有记忆子系统"""
        # 1. 添加到短期记忆
        self.short_term.add_message(role, content)

        # 2. 提取重要信息存入长期记忆
        importance = self.short_term.calculate_importance(
            Message(role=role, content=content)
        )
        if importance > 0.6:
            mem_id = self.long_term.add_memory(
                content, {"role": role}, importance
            )
            self.persistence.save_memory(
                mem_id, content, "conversation", importance
            )

        # 3. 记录情景
        self.episodic.add_event(
            EventType.CONVERSATION,
            f"{role}: {content[:100]}",
            {"role": role},
            importance
        )

        # 4. 提取知识三元组（简化版）
        self._extract_knowledge(content)

        return {
            "short_term_context": self.short_term.get_context(),
            "relevant_memories": self.long_term.search(content, top_k=3),
            "recent_episodes": self.episodic.get_timeline(hours=1),
        }

    def _extract_knowledge(self, text: str):
        """从文本中提取知识（简化版，实际用LLM提取）"""
        # 示例：检测"X是Y"模式
        patterns = [
            (r"(\w+)是(\w+)", "是一种"),
            (r"(\w+)使用(\w+)", "使用"),
            (r"(\w+)喜欢(\w+)", "喜欢"),
        ]
        import re
        for pattern, relation in patterns:
            matches = re.findall(pattern, text)
            for subj, obj in matches:
                if len(subj) > 1 and len(obj) > 1:
                    self.semantic.add_relation(subj, relation, obj)

    def get_enhanced_context(self, query: str) -> str:
        """获取增强上下文 - 综合所有记忆源"""
        parts = []

        # 短期记忆
        short_context = self.short_term.get_context()
        if short_context:
            parts.append("=== 近期对话 ===")
            for msg in short_context[-6:]:
                parts.append(f"{msg['role']}: {msg['content']}")

        # 长期记忆检索
        long_results = self.long_term.search(query, top_k=3)
        if long_results:
            parts.append("\n=== 相关记忆 ===")
            for mem, score in long_results:
                parts.append(f"[{score:.2f}] {mem.content}")

        # 情景记忆
        episodes = self.episodic.find_similar_episodes(query, top_k=3)
        if episodes:
            parts.append("\n=== 历史事件 ===")
            for ep in episodes:
                parts.append(f"[{ep['timestamp'][:16]}] {ep['summary']}")

        # 语义知识
        entities = self._extract_entities_from_query(query)
        for entity in entities:
            ctx = self.semantic.to_prompt_context(entity)
            if ctx:
                parts.append(f"\n=== {entity}的知识 ===")
                parts.append(ctx)

        # 工作记忆
        working_ctx = self.working.to_prompt_context()
        if working_ctx:
            parts.append(f"\n=== 当前任务 ===")
            parts.append(working_ctx)

        return "\n".join(parts)

    def _extract_entities_from_query(self, query: str) -> list:
        entities = []
        for entity in self.semantic.graph.keys():
            if entity in query:
                entities.append(entity)
        return entities

    def save_session(self):
        """保存当前会话的所有记忆"""
        # 保存长期记忆
        for mem in self.long_term.memories:
            self.persistence.save_memory(
                mem.id, mem.content, "long_term",
                mem.importance, mem.metadata
            )

        # 保存情景
        for ep in self.episodic.episodes:
            self.persistence.save_episode(
                ep["id"], ep.get("session_id"),
                ep["type"], ep["summary"],
                ep.get("details"), ep.get("importance", 0.5)
            )

十六、最佳实践与设计原则

16.1 设计原则

1. 分层存储：不同生命周期的数据用不同存储策略
2. 按需加载：不要一次性加载所有记忆，用检索过滤
3. 渐进遗忘：低价值记忆自然衰减，保持记忆库健康
4. 冲突透明：记忆更新时记录冲突日志，便于审计
5. 隐私优先：敏感信息加密存储，设置访问控制

16.2 常见陷阱

16.3 性能优化建议

• 向量索引：使用HNSW或IVF索引加速检索
• 缓存热点：高频访问的记忆缓存在内存中
• 异步写入：记忆持久化用异步队列
• 分片存储：按用户/会话分片，减小单次检索范围
• 批量嵌入：Embedding调用批量处理，减少API调用次数

十七、总结

一个优秀的Agent记忆系统需要：

1. 多类型记忆协同：短期、长期、情景、语义各司其职
2. 智能检索：不是简单的关键词匹配，而是多维综合评分
3. 生命周期管理：记忆从创建、更新到压缩、删除的完整流程
4. 持久化保障：跨会话、跨设备的记忆一致性
5. 持续评估：用数据驱动记忆系统的迭代优化

记忆系统是Agent从"工具"进化为"助手"的核心基础设施。随着大模型能力的提升，记忆系统的设计也会不断演进——但分层、检索、压缩、持久化这些基本原则不会过时。

本教程涵盖了AI Agent记忆系统设计的核心内容。建议读者从简单的滑动窗口+向量存储开始，逐步添加情景记忆、语义记忆等高级功能，根据实际需求迭代优化。