🧠 AI Agent 记忆工程：构建可靠的长期记忆系统

发布日期：2026年5月29日 | 阅读时间：约15分钟 | 难度：中级~高级

一、为什么 Agent 需要记忆？

想象一下：你花30分钟向 Agent 解释了你的项目架构、编码规范和技术选型偏好，第二天打开新对话，一切归零。这就是当前大多数 Agent 的现状——有状态的工具，无记忆的伙伴。

记忆对 Agent 的重要性体现在三个层面：

2025-2026年，随着 Agent 被用于越来越复杂的任务（代码开发、数据分析、项目管理），记忆已经从”锦上添花”变成了”不可或缺”。

二、记忆类型学：Agent 的六种记忆

借鉴认知科学的记忆分类，Agent 的记忆系统可以分为六种类型：

2.1 工作记忆（Working Memory）

当前对话的上下文窗口。容量有限（受模型 context length 约束），速度最快，断电即失。对应：LLM 的 system prompt + 对话历史。

2.2 语义记忆（Semantic Memory）

关于世界的事实性知识。”Python 3.12 引入了 PEP 701″、”我们的项目使用 PostgreSQL”。通常存储在向量数据库中，通过语义检索获取。

2.3 情景记忆（Episodic Memory）

Agent 经历过的事件和交互历史。”上次用户让我重构 auth 模块时，他偏好使用 JWT 而不是 session”。包含时间、参与者、结果等元信息。

2.4 程序记忆（Procedural Memory）

Agent 学会的技能和操作流程。”当用户说’部署’时，先运行测试，再构建 Docker 镜像，最后推送到 ECR”。可以表示为可复用的工具或工作流模板。

2.5 情感记忆（Affective Memory）

用户的情绪反馈和偏好。”用户对冗长的解释表现出不耐烦”、”用户喜欢简洁的代码注释”。用于调整 Agent 的沟通风格。

2.6 元记忆（Meta Memory）

Agent 对自身记忆状态的认知。”我知道我不知道这个项目的数据库 schema”、”我对这个领域的信息可能已经过时”。用于触发主动的信息获取行为。

三、记忆系统架构设计

3.1 分层记忆架构

┌─────────────────────────────────────────────┐
│                 Agent Core                   │
│         (LLM + Reasoning Engine)             │
└─────────────┬──────────────────┬────────────┘
              │                  │
    ┌─────────▼────────┐ ┌──────▼──────────┐
    │   Memory Router   │ │  Consolidation  │
    │  (记忆路由/调度)   │ │   Engine        │
    └─┬────┬────┬────┬─┘ │  (记忆巩固引擎)  │
      │    │    │    │   └─────────────────┘
      ▼    ▼    ▼    ▼
   ┌────┐┌────┐┌────┐┌────┐
   │Work││Sema││Epi││Proc│
   │ing ││nti││sod││edur│
   │    ││c   ││ic  ││al  │
   └────┘└────┘└────┘└────┘
   Context  Vector  Graph  Skill
   Window   DB      DB     Store

3.2 核心组件

# memory_system/core.py
from dataclasses import dataclass, field
from typing import Any, Optional
from datetime import datetime
from enum import Enum

class MemoryType(Enum):
    SEMANTIC = "semantic"      # 语义记忆
    EPISODIC = "episodic"      # 情景记忆
    PROCEDURAL = "procedural"  # 程序记忆
    AFFECTIVE = "affective"    # 情感记忆

class Importance(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4

@dataclass
class Memory:
    """统一记忆单元"""
    id: str
    type: MemoryType
    content: str                          # 记忆内容
    embedding: list[float] = field(default_factory=list)  # 向量表示
    metadata: dict = field(default_factory=dict)           # 元信息
    importance: Importance = Importance.MEDIUM
    created_at: datetime = field(default_factory=datetime.utcnow)
    last_accessed: datetime = field(default_factory=datetime.utcnow)
    access_count: int = 0                 # 访问次数（用于巩固决策）
    ttl: Optional[int] = None             # 过期时间（秒），None=永久
    
    @property
    def is_expired(self) -> bool:
        if self.ttl is None:
            return False
        elapsed = (datetime.utcnow() - self.created_at).total_seconds()
        return elapsed > self.ttl
    
    def touch(self):
        """更新访问记录"""
        self.last_accessed = datetime.utcnow()
        self.access_count += 1


@dataclass
class MemoryQuery:
    """记忆查询"""
    text: str
    types: list[MemoryType] = field(
        default_factory=lambda: list(MemoryType)
    )
    top_k: int = 5
    min_score: float = 0.6
    time_range: Optional[tuple[datetime, datetime]] = None
    filters: dict = field(default_factory=dict)

3.3 记忆路由策略

# memory_system/router.py
class MemoryRouter:
    """根据查询意图自动路由到对应的记忆存储"""
    
    def __init__(self, stores: dict[MemoryType, MemoryStore]):
        self.stores = stores
        self._intent_classifier = None  # 可选：用轻量模型做意图分类
    
    async def store(self, memory: Memory) -> str:
        """存储记忆到对应存储"""
        store = self.stores.get(memory.type)
        if not store:
            raise ValueError(f"未注册 {memory.type} 的存储后端")
        return await store.save(memory)
    
    async def retrieve(self, query: MemoryQuery) -> list[Memory]:
        """从多个存储中检索并合并结果"""
        results = []
        for mem_type in query.types:
            store = self.stores.get(mem_type)
            if store:
                memories = await store.search(query)
                results.extend(memories)
        
        # 按综合得分排序（语义相似度 + 重要性 + 时效性）
        results.sort(key=lambda m: self._score(m, query), reverse=True)
        return results[:query.top_k]
    
    def _score(self, memory: Memory, query: MemoryQuery) -> float:
        """综合评分：语义相似度(0.5) + 重要性(0.2) + 时效性(0.2) + 访问频率(0.1)"""
        semantic_score = cosine_similarity(
            memory.embedding, 
            embed(query.text)
        )
        importance_score = memory.importance.value / 4.0
        
        # 时效性：越近越好，指数衰减
        days_old = (datetime.utcnow() - memory.created_at).days
        recency_score = math.exp(-0.01 * days_old)
        
        # 访问频率：被引用越多越重要
        frequency_score = min(memory.access_count / 10.0, 1.0)
        
        return (
            0.5 * semantic_score +
            0.2 * importance_score +
            0.2 * recency_score +
            0.1 * frequency_score
        )

四、语义记忆：向量检索实战

语义记忆是 Agent 记忆系统中最成熟的部分，核心是嵌入 + 向量检索。但生产环境和玩具 demo 之间有几个关键差距。

4.1 嵌入策略选择

4.2 分块策略：被忽视的关键

# memory_system/chunking.py
from typing import Callable

class SmartChunker:
    """智能分块：保持语义完整性"""
    
    def __init__(self, chunk_size: int = 512, overlap: int = 64):
        self.chunk_size = chunk_size
        self.overlap = overlap
    
    def chunk_document(self, text: str, metadata: dict) -> list[dict]:
        """按语义边界分块，而非机械截断"""
        # 1. 按段落分割
        paragraphs = text.split('\n\n')
        
        chunks = []
        current_chunk = ""
        current_meta = {**metadata, "chunk_index": 0}
        
        for para in paragraphs:
            # 如果当前段落加上已有内容不超过限制
            if len(current_chunk) + len(para) <= self.chunk_size:
                current_chunk += para + "\n\n"
            else:
                # 保存当前块
                if current_chunk.strip():
                    chunks.append({
                        "text": current_chunk.strip(),
                        "metadata": {**current_meta, "length": len(current_chunk)}
                    })
                # 新块，保留 overlap
                if current_chunk and self.overlap > 0:
                    overlap_text = current_chunk[-self.overlap:]
                    current_chunk = overlap_text + para + "\n\n"
                else:
                    current_chunk = para + "\n\n"
                current_meta["chunk_index"] += 1
        
        # 最后一块
        if current_chunk.strip():
            chunks.append({
                "text": current_chunk.strip(),
                "metadata": {**current_meta, "length": len(current_chunk)}
            })
        
        return chunks
    
    def chunk_conversation(self, messages: list[dict], window_size: int = 6) -> list[dict]:
        """对话分块：滑动窗口，保持上下文连贯"""
        chunks = []
        for i in range(0, len(messages), window_size // 2):
            window = messages[i:i + window_size]
            # 生成摘要作为块的文本表示
            summary = self._summarize_window(window)
            chunks.append({
                "text": summary,
                "metadata": {
                    "type": "conversation_summary",
                    "start_msg": i,
                    "end_msg": min(i + window_size, len(messages)),
                    "participants": list(set(m.get("role") for m in window)),
                    "timestamp": window[0].get("timestamp", "")
                }
            })
        return chunks
    
    def _summarize_window(self, messages: list[dict]) -> str:
        """用轻量方式生成对话窗口摘要（实际项目中可调用 LLM）"""
        parts = []
        for m in messages:
            role = m.get("role", "unknown")
            content = m.get("content", "")[:200]
            parts.append(f"[{role}]: {content}")
        return "\n".join(parts)

4.3 混合检索：向量 + 关键词

纯向量检索有一个致命弱点：对精确匹配不敏感。当用户查询包含 ID、版本号、专有名词时，BM25 关键词检索往往更准确。

# memory_system/hybrid_search.py
import numpy as np
from rank_bm25 import BM25Okapi

class HybridRetriever:
    """混合检索：向量语义检索 + BM25 关键词检索 + RRF 融合"""
    
    def __init__(self, vector_store, alpha: float = 0.6):
        """
        alpha: 向量检索权重，1-alpha 为 BM25 权重
        """
        self.vector_store = vector_store
        self.alpha = alpha
        self.bm25 = None
        self.corpus = []
    
    def index(self, documents: list[dict]):
        """构建双重索引"""
        texts = [doc["text"] for doc in documents]
        self.corpus = documents
        
        # 1. 向量索引
        embeddings = [embed(text) for text in texts]
        self.vector_store.add(embeddings, documents)
        
        # 2. BM25 索引
        tokenized = [self._tokenize(text) for text in texts]
        self.bm25 = BM25Okapi(tokenized)
    
    def search(self, query: str, top_k: int = 10) -> list[dict]:
        """混合检索"""
        # 向量检索
        query_embedding = embed(query)
        vector_results = self.vector_store.search(query_embedding, top_k * 2)
        vector_scores = {r["id"]: r["score"] for r in vector_results}
        
        # BM25 检索
        tokenized_query = self._tokenize(query)
        bm25_scores_raw = self.bm25.get_scores(tokenized_query)
        bm25_top_indices = np.argsort(bm25_scores_raw)[::-1][:top_k * 2]
        bm25_max = bm25_scores_raw.max() or 1.0
        bm25_scores = {
            self.corpus[i]["id"]: bm25_scores_raw[i] / bm25_max 
            for i in bm25_top_indices
        }
        
        # RRF (Reciprocal Rank Fusion)
        all_ids = set(vector_scores.keys()) | set(bm25_scores.keys())
        rrf_scores = {}
        for doc_id in all_ids:
            v_rank = self._get_rank(doc_id, vector_results)
            b_rank = self._get_rank_bm25(doc_id, bm25_top_indices)
            rrf_scores[doc_id] = (
                self.alpha * (1.0 / (60 + v_rank + 1)) +
                (1 - self.alpha) * (1.0 / (60 + b_rank + 1))
            )
        
        # 排序返回
        sorted_ids = sorted(rrf_scores, key=rrf_scores.get, reverse=True)[:top_k]
        id_to_doc = {doc["id"]: doc for doc in self.corpus}
        return [
            {**id_to_doc[doc_id], "hybrid_score": rrf_scores[doc_id]}
            for doc_id in sorted_ids
            if doc_id in id_to_doc
        ]
    
    def _tokenize(self, text: str) -> list[str]:
        """简单分词，生产环境建议用 jieba（中文）或 spaCy"""
        return text.lower().split()
    
    def _get_rank(self, doc_id, results) -> int:
        for i, r in enumerate(results):
            if r["id"] == doc_id:
                return i
        return 999
    
    def _get_rank_bm25(self, doc_id, top_indices) -> int:
        for rank, idx in enumerate(top_indices):
            if self.corpus[idx]["id"] == doc_id:
                return rank
        return 999

五、情景记忆：经验回放系统

情景记忆是 Agent 的”自传”——它记录了 Agent 与用户交互的具体事件。这是实现个性化和持续学习的关键。

5.1 情景记忆的数据模型

# memory_system/episodic.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class Episode:
    """情景记忆单元：记录一次交互经验"""
    id: str
    title: str                    # 事件标题（自动生成）
    summary: str                  # 事件摘要
    conversation: list[dict]      # 原始对话（可选，可压缩存储）
    outcome: str                  # 结果：success / failure / partial
    lessons_learned: list[str]    # 从中学到的教训
    user_feedback: Optional[str]  # 用户反馈
    timestamp: datetime
    tags: list[str]               # 标签，便于检索
    related_episodes: list[str]   # 关联事件ID
    
    def to_memory(self) -> Memory:
        """转换为通用记忆格式"""
        return Memory(
            id=self.id,
            type=MemoryType.EPISODIC,
            content=f"{self.title}\n\n{self.summary}\n\n教训: {'; '.join(self.lessons_learned)}",
            importance=Importance.HIGH if self.outcome == "failure" else Importance.MEDIUM,
            metadata={
                "outcome": self.outcome,
                "tags": self.tags,
                "timestamp": self.timestamp.isoformat(),
                "lessons": self.lessons_learned,
            }
        )


class ExperienceReplayBuffer:
    """经验回放缓冲区：从历史交互中提取和存储经验"""
    
    def __init__(self, llm_client, max_buffer: int = 1000):
        self.llm = llm_client
        self.buffer: list[Episode] = []
        self.max_buffer = max_buffer
    
    async def extract_episode(
        self, 
        conversation: list[dict],
        user_feedback: Optional[str] = None
    ) -> Episode:
        """从对话中提取情景记忆"""
        # 使用 LLM 生成结构化摘要
        extraction_prompt = f"""
分析以下对话，提取关键信息：

对话内容：
{format_conversation(conversation)}

用户反馈：{user_feedback or "无"}

请输出 JSON：
{{
    "title": "简短的事件标题（10字以内）",
    "summary": "事件摘要（100字以内）",
    "outcome": "success | failure | partial",
    "lessons_learned": ["教训1", "教训2"],
    "tags": ["标签1", "标签2"]
}}
"""
        result = await self.llm.complete(extraction_prompt, response_format="json")
        
        episode = Episode(
            id=generate_id(),
            title=result["title"],
            summary=result["summary"],
            conversation=conversation if result["outcome"] == "failure" else [],  # 失败才存原始对话
            outcome=result["outcome"],
            lessons_learned=result["lessons_learned"],
            user_feedback=user_feedback,
            timestamp=datetime.utcnow(),
            tags=result["tags"],
            related_episodes=[]
        )
        
        self.buffer.append(episode)
        
        # 缓冲区满了，淘汰低价值记忆
        if len(self.buffer) > self.max_buffer:
            self._evict()
        
        return episode
    
    def _evict(self):
        """淘汰策略：保留失败经验和高价值成功经验"""
        # 失败经验永远保留
        failures = [e for e in self.buffer if e.outcome == "failure"]
        successes = [e for e in self.buffer if e.outcome != "failure"]
        
        # 成功经验按标签多样性和时效性排序
        successes.sort(
            key=lambda e: (len(e.lessons_learned), e.timestamp),
            reverse=True
        )
        
        # 保留所有失败 + 最有价值的成功
        keep_count = self.max_buffer - len(failures)
        self.buffer = failures + successes[:max(keep_count, 0)]

5.2 经验检索与复用

class ExperienceRetriever:
    """基于当前上下文检索相关历史经验"""
    
    def __init__(self, episode_store):
        self.store = episode_store
    
    async def get_relevant_experiences(
        self, 
        current_task: str,
        top_k: int = 3
    ) -> list[Episode]:
        """检索与当前任务相关的历史经验"""
        # 1. 语义检索
        candidates = await self.store.semantic_search(current_task, top_k=10)
        
        # 2. 过滤：只返回有实际教训的
        candidates = [e for e in candidates if e.lessons_learned]
        
        # 3. 排序：失败经验优先（更有学习价值）
        candidates.sort(
            key=lambda e: (
                1 if e.outcome == "failure" else 0,
                e.timestamp
            ),
            reverse=True
        )
        
        return candidates[:top_k]
    
    def format_for_prompt(self, episodes: list[Episode]) -> str:
        """将经验格式化为 prompt 注入"""
        if not episodes:
            return ""
        
        lines = ["## 📚 相关历史经验（请参考避免重复犯错）\n"]
        for ep in episodes:
            outcome_emoji = "❌" if ep.outcome == "failure" else "✅"
            lines.append(f"### {outcome_emoji} {ep.title}")
            lines.append(f"**时间**: {ep.timestamp.strftime('%Y-%m-%d')}")
            lines.append(f"**摘要**: {ep.summary}")
            if ep.lessons_learned:
                lines.append("**教训**:")
                for lesson in ep.lessons_learned:
                    lines.append(f"  - {lesson}")
            lines.append("")
        
        return "\n".join(lines)

六、程序记忆：技能自动积累

程序记忆是 Agent 的”肌肉记忆”——它记录了 Agent 学会的操作流程和技能。这是 Agent 能力自动成长的核心机制。

6.1 技能表示

# memory_system/procedural.py
from pydantic import BaseModel

class Skill(BaseModel):
    """可复用的技能单元"""
    name: str
    description: str
    steps: list[str]              # 操作步骤
    preconditions: list[str]      # 前置条件
    postconditions: list[str]     # 完成条件
    success_rate: float = 0.0     # 历史成功率
    usage_count: int = 0          # 使用次数
    created_from: list[str]       # 从哪些情景记忆中学到的
    tags: list[str] = []
    
    def to_tool_schema(self) -> dict:
        """转换为 LLM 可理解的 tool schema"""
        return {
            "name": self.name,
            "description": self.description,
            "inputSchema": {
                "type": "object",
                "properties": {
                    "context": {
                        "type": "string",
                        "description": f"执行上下文。前置条件: {'; '.join(self.preconditions)}"
                    }
                },
                "required": ["context"]
            }
        }

class SkillMiner:
    """从情景记忆中自动挖掘技能"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
    
    async def mine_skills(self, episodes: list[Episode]) -> list[Skill]:
        """从多个相关事件中提炼可复用的技能"""
        # 只分析成功的事件
        success_episodes = [e for e in episodes if e.outcome == "success"]
        if len(success_episodes) < 2:
            return []  # 至少需要2次成功经验才能提炼技能
        
        mining_prompt = f"""
分析以下成功经验，提炼出可复用的操作技能：

{format_episodes(success_episodes)}

请输出 JSON 数组，每个技能包含：
{{
    "name": "技能名称（动词+名词，如'部署后端服务'）",
    "description": "技能描述",
    "steps": ["步骤1", "步骤2", "步骤3"],
    "preconditions": ["前置条件1"],
    "postconditions": ["完成标志1"],
    "tags": ["标签"]
}}

只提炼在至少2个事件中重复出现的操作模式。
"""
        result = await self.llm.complete(mining_prompt, response_format="json")
        return [Skill(**s, created_from=[e.id for e in success_episodes]) for s in result]

七、记忆巩固：从短期到长期

人类大脑在睡眠期间会进行记忆巩固——将海马体中的短期记忆转移到皮层成为长期记忆。Agent 也需要类似的机制。

7.1 巩固策略

# memory_system/consolidation.py
class MemoryConsolidationEngine:
    """记忆巩固引擎：定期整理、压缩、淘汰记忆"""
    
    def __init__(self, router: MemoryRouter, llm_client):
        self.router = router
        self.llm = llm_client
    
    async def consolidate(self):
        """执行记忆巩固（建议每天低峰期运行一次）"""
        # 1. 去重：合并高度相似的记忆
        await self._deduplicate()
        
        # 2. 压缩：将细节丰富但访问频率低的记忆压缩为摘要
        await self._compress()
        
        # 3. 关联：发现并建立记忆之间的关联
        await self._link()
        
        # 4. 淘汰：删除过期和低价值的记忆
        await self._prune()
        
        # 5. 升华：从具体经验中提炼通用原则
        await self._abstract()
    
    async def _deduplicate(self, threshold: float = 0.92):
        """去重：合并相似度超过阈值的记忆"""
        all_memories = await self.router.stores[MemoryType.SEMANTIC].get_all()
        
        to_remove = set()
        for i, m1 in enumerate(all_memories):
            if m1.id in to_remove:
                continue
            for m2 in all_memories[i+1:]:
                if m2.id in to_remove:
                    continue
                sim = cosine_similarity(m1.embedding, m2.embedding)
                if sim > threshold:
                    # 保留更重要的，合并元数据
                    keeper, discard = (m1, m2) if m1.importance >= m2.importance else (m2, m1)
                    keeper.metadata.update(discard.metadata)
                    keeper.access_count += discard.access_count
                    to_remove.add(discard.id)
        
        for mid in to_remove:
            await self.router.stores[MemoryType.SEMANTIC].delete(mid)
    
    async def _compress(self):
        """压缩：将细节记忆转为摘要"""
        store = self.router.stores[MemoryType.EPISODIC]
        old_memories = await store.get_old(threshold_days=30)
        low_access = [m for m in old_memories if m.access_count < 3]
        
        for memory in low_access:
            # 生成压缩摘要
            compressed = await self.llm.complete(
                f"将以下记忆压缩为50字以内的精华摘要，保留关键教训：\n{memory.content}",
                max_tokens=100
            )
            memory.content = f"[摘要] {compressed}"
            memory.metadata["compressed"] = True
            await store.update(memory)
    
    async def _abstract(self):
        """升华：从具体经验中提炼通用原则"""
        store = self.router.stores[MemoryType.EPISODIC]
        recent_failures = await store.get_by_outcome("failure", limit=20)
        
        if len(recent_failures) < 3:
            return
        
        abstraction_prompt = f"""
从以下失败经验中提炼出通用的行为原则：

{format_memories(recent_failures)}

输出 3-5 条通用原则，格式：
[{{"principle": "原则描述", "evidence": ["支撑事件ID"]}}]
"""
        principles = await self.llm.complete(abstraction_prompt, response_format="json")
        
        for p in principles:
            memory = Memory(
                id=generate_id(),
                type=MemoryType.SEMANTIC,
                content=p["principle"],
                importance=Importance.HIGH,
                metadata={
                    "source": "abstraction",
                    "evidence": p["evidence"],
                    "type": "behavioral_principle"
                }
            )
            await self.router.store(memory)

八、总结与展望

8.1 关键要点

8.2 2026年记忆系统趋势

• 记忆即服务 (MaaS)：Mem0、Zep、LangMem 等专用记忆层正在标准化，Agent 框架直接集成即可
• 多 Agent 共享记忆：多个 Agent 共享同一记忆池，实现团队级知识积累
• 记忆安全：记忆注入攻击（通过精心构造的对话污染 Agent 记忆）成为新的安全研究方向
• 分层存储：热记忆（Redis）→ 温记忆（PostgreSQL）→ 冷记忆（对象存储），按访问频率自动迁移
• 记忆压缩：使用小模型对记忆进行持续压缩和摘要，降低存储和检索成本

记忆是智能的基石。一个没有记忆的 Agent 就像一个每天醒来都失忆的人——能思考，但不能成长。2026年，记忆工程正在成为 AI 系统架构中最关键的差异化因素。现在就开始为你的 Agent 构建记忆系统吧！

本文涉及的完整代码示例已整理为可运行的 Python 项目，支持 ChromaDB / Qdrant 向量后端，兼容 OpenAI / Ollama 嵌入模型。