Agent 记忆系统设计：让智能体记得住、想得起、用得上-拓冰建站

Agent 记忆系统设计：让智能体记得住、想得起、用得上

一、多轮对话里的记忆黑洞

Agent 最常见的失败模式是什么？不是推理错误，不是工具调用失败，而是把上一轮说过的事忘得一干二净。用户告诉它"我的数据库在东京区"，三句话后它又去查 global endpoint。这种体验比报错更糟糕——因为用户会直接丧失信任。

传统方案是往 prompt 里塞聊天历史。但 token 成本按上下文长度线性增长，GPT-4 的 128K 窗口跑长对话时，单次请求轻松破 $0.1。更隐蔽的问题是"迷失在中间"（Lost-in-the-Middle）：模型对长文本中间部分的注意力衰减，导致早期记忆照样丢失。

一个工程上可行的解法是分层记忆架构：工作记忆（短程）、情节记忆（中程）、语义记忆（长程）三层协同。

flowchart TD A[用户输入] --> B[工作记忆缓冲区] B --> C{触发归档?} C -->|摘要阈值达| D[情节记忆提取器] C -->|未触发| E[直接注入Prompt] D --> F{冲突检测} F -->|新信息| G[语义记忆写入] F -->|冲突| H[记忆合并/覆盖] G --> I[向量数据库] H --> I I --> J[语义检索] E --> K[LLM推理] J --> K

二、分层记忆的工程实现

工作记忆（Working Memory）

就是当前对话的完整上下文。不做压缩，但限制长度（建议最近 10 轮）。关键设计点是滑动窗口 + 优先级标记——不是所有历史轮次都同等重要。

情节记忆（Episodic Memory）

会话级别的摘要归档。核心思路：在对话达到 N 轮或 token 超过阈值时，异步触发摘要提取。这里有一个容易被忽略的细节——摘要粒度要与检索场景对齐。如果按整段对话摘要，检索时只能返回一大坨，不够细。应该按"语义块"拆分，每块标注时间戳和重要性权重。

语义记忆（Semantic Memory）

跨会话的持久化知识。存储到向量数据库（Milvus/Qdrant/Pinecone），检索时做 hybrid search：向量相似度 + 关键词 BM25，取交集。元数据里必须带时间衰减因子——两周前的偏好权重应低于昨天的。

三、生产级记忆写入代码

import asyncio import time from dataclasses import dataclass, field from typing import Optional import numpy as np from openai import AsyncOpenAI @dataclass class MemoryEntry: content: str embedding: Optional[list[float]] = None timestamp: float = field(default_factory=time.time) importance: float = 1.0 session_id: str = "" ttl_days: int = 30 class SemanticMemoryStore: """语义记忆存储层。 设计决策： - 使用异步客户端，避免阻塞 Agent 主循环 - 时间衰减嵌入在检索阶段计算，而非写入阶段，减少存储耦合 - 重要性评分由 LLM 生成，而非规则引擎——规则跟不上场景变化 """ def __init__(self, vector_dim: int = 1536, decay_halflife: int = 14): self._client = AsyncOpenAI() self._store: list[MemoryEntry] = [] self._vector_dim = vector_dim self._decay_halflife = decay_halflife async def embed(self, text: str) -> list[float]: """文本转向量，带重试和超时保护""" for attempt in range(3): try: resp = await asyncio.wait_for( self._client.embeddings.create( model="text-embedding-3-small", input=text ), timeout=5.0, ) return resp.data[0].embedding except asyncio.TimeoutError: if attempt == 2: raise await asyncio.sleep(1 * (attempt + 1)) except Exception: if attempt == 2: raise await asyncio.sleep(2**attempt) raise RuntimeError("unreachable") async def write( self, content: str, session_id: str, importance: float = 1.0, ttl_days: int = 30, ) -> None: """异步写入记忆，不阻塞 Agent 推理链路""" embedding = await self.embed(content) entry = MemoryEntry( content=content, embedding=embedding, session_id=session_id, importance=importance, ttl_days=ttl_days, ) self._store.append(entry) def _time_decay(self, entry: MemoryEntry, now: float) -> float: """指数衰减：半衰期默认 14 天""" age_days = (now - entry.timestamp) / 86400.0 return np.exp(-np.log(2) * age_days / self._decay_halflife) async def search(self, query: str, session_id: str, top_k: int = 5) -> list[MemoryEntry]: """向量检索 + 时间衰减 + 重要性加权""" q_embed = await self.embed(query) now = time.time() scored: list[tuple[float, MemoryEntry]] = [] for entry in self._store: if entry.session_id != session_id: continue if now - entry.timestamp > entry.ttl_days * 86400: continue if entry.embedding is None: continue similarity = np.dot(q_embed, entry.embedding) decay = self._time_decay(entry, now) score = similarity * decay * entry.importance scored.append((score, entry)) scored.sort(key=lambda x: x[0], reverse=True) return [entry for _, entry in scored[:top_k]]

注释要点：write不阻塞推理链路是刻意的——记忆归档优先级低于实时响应。时序解码用指数模型而非线性，因为人类记忆遗忘曲线是指数级。