2026-05-14AIAgent记忆系统RAG架构

AI 智能体如何记住一切:长期记忆系统的工程实践

做 AI 智能体(Agent)的人都踩过同一个坑:模型在单次对话里聪明得吓人,但一旦跨 session(第二天再问),它就彻底失忆了。"我记得上次我们讨论过..."——这句话对没有记忆系统的 Agent 来说,永远是空话。

biluo·11112 words

前言:为什么记忆是 Agent 的生死线

做 AI 智能体(Agent)的人都踩过同一个坑:模型在单次对话里聪明得吓人,但一旦跨 session(第二天再问),它就彻底失忆了。"我记得上次我们讨论过..."——这句话对没有记忆系统的 Agent 来说,永远是空话。

记忆系统的设计,往往是 AI Agent 体验的分水岭。一个好的记忆系统,让 Agent 像一个不断成长的专家;一个差的设计,让 Agent 每秒都在重复自己。

本文从工程角度,拆解 AI Agent 长期记忆系统的核心组件、设计模式与避坑指南。

一、记忆的分层模型

在聊实现之前,先对齐"记忆"在 Agent 系统中的层次。

1.1 三层记忆架构

大多数成熟系统都遵循这个分层:

层级 容量 访问速度 典型实现 生命周期
**工作记忆(Working Memory)** ~128K tokens O(1) LLM Context Window 单次会话
**短期记忆(Short-Term)** 几十到几百条 O(1) 消息历史、Session Store 分钟~天
**长期记忆(Long-Term)** 无限 O(log n) ~ O(1) Vector DB、Graph DB、LMDB 永久

关键理解:上下文窗口不是用来存储记忆的,是用来"思考"的。把记忆塞进 context = 把书架塞进脑子里,思考空间没了。

1.2 各层的核心职责

工作记忆 —— 当前的推理空间,存储当前任务相关的所有信息。你的 prompt template、few-shot examples、当前对话内容,都在这里。

短期记忆 —— 最近发生的事件,比如"用户今天问了哪几个问题"、"今天跑了哪些任务"。Session 级别的状态存储。

长期记忆 —— 跨越 session 的知识沉淀。用户偏好、项目背景、领域知识、过往经验。

二、长期记忆的核心技术:向量检索

长期记忆的主流实现是向量数据库 + RAG(Retrieval-Augmented Generation)。这个模式有坑,也有大量的工程权衡。

2.1 Embedding 模型的选择

记忆的质量,直接由 Embedding 模型决定。常见选项:

`python

# 主流 Embedding 模型对比(2026年)

EMBEDDING_MODELS = {

"text-embedding-3-large": { # OpenAI

"dims": 3072,

"context_window": 8192,

"price_per_1k": 0.13,

"strength": "通用场景表现稳定,中文支持好",

},

"bge-m3": { # 北京智源开源

"dims": 1024,

"context_window": 8192,

"price_per_1k": 0, # 自托管免费

"strength": "多语言、极高检索精度,CPU 可跑",

},

"voyage-code-2": { # Voyage AI

"dims": 1024,

"context_window": 16000,

"price_per_1k": 0.12,

"strength": "代码理解极强,适合技术文档记忆",

},

"chorus-base": { # 月之暗面开源

"dims": 2048,

"context_window": 32768,

"price_per_1k": 0,

"strength": "长上下文友好,中文优化",

}

}

`

选型建议:代码类记忆用 Voyage Code 2;通用中文场景用 BGE-M3 自托管;如果记忆文本经常很长(>4K tokens),选 Chorus-base。

2.2 Chunking 策略:被严重低估的设计决策

很多人以为记忆系统只要"把文本向量化存进去"就行了。实际上 Chunking 策略 对检索质量影响极大。

`python

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 常见 Chunking 策略对比

# ❌ 固定长度(Bad)

# 问题:语义截断严重,上下文碎片化

chunk_size = 500

chunk_overlap = 50

# ✅ 递归字符分割(Good)

# 按段落、句子、单词层级拆分,减少语义截断

splitter = RecursiveCharacterTextSplitter(

separators=["\n\n", "\n", "。", " ", ""],

chunk_size=800,

chunk_overlap=100, # 重叠保持上下文连续性

length_function=lambda x: len(x) // 2, # 按 token 估算

)

# ✅ 按语义分割(Better)

# 用模型判断自然段落边界,适合结构化文档

# 可以用貔貅(pict-char)或自己做段落检测

# ✅ Metadata-enriched Chunking(Best for Agent)

# 每个 chunk 带上来源、时间、重要性标签

chunks = [

{

"text": "用户偏好:喜欢用 Python,不喜欢 Java",

"metadata": {

"source": "user_preference_v1",

"created_at": "2026-01-15",

"importance": "high", # 重要偏好标记

"category": "coding_preference"

}

},

{

"text": "项目架构:微服务,5个核心服务",

"metadata": {

"source": "project_doc_2026q1",

"created_at": "2026-03-01",

"importance": "medium",

"category": "architecture"

}

}

]

`

关键经验:对于 Agent 记忆系统,我强烈建议在每个 chunk 的 metadata 里加 importance 字段(high/medium/low)。检索时做加权——重要记忆被召回的概率要高得多。

2.3 向量索引与检索

`python

# 典型的 Agent 记忆检索流程

async def retrieve_memory(query: str, user_id: str, top_k: int = 5):

# 1. 查询向量化

query_embedding = await embedding_model.embed(query)

# 2. 混合检索:向量 + 关键词

results = await vector_db.search(

collection=f"memory_{user_id}",

vector=query_embedding,

top_k=top_k * 2, # 多取一些,后面要过滤

filters={

"importance": {"$in": ["high", "medium"]},

# 时间衰减:优先近期记忆,但保留老记忆的召回机会

"created_at": {"$gte": "2025-01-01"}

}

)

# 3. Rerank:让更相关的排在前面

reranked = await reranker.rerank(

query=query,

documents=[r["text"] for r in results],

top_k=top_k

)

# 4. 构建记忆上下文

memory_context = "\n".join([

f"[{r['metadata']['category']}] {r['text']}"

for r in reranked

])

return memory_context

`

三、Graph Memory:超越向量检索

向量检索擅长"相似性匹配",但不擅长关系推理。比如"找出所有和这个项目相关的人,以及这些人之间的工作关系"——这类问题需要图数据库。

3.1 什么时候用 Graph Memory

`

向量检索 → "这个记忆和当前问题语义相似吗?"(What)

Graph查询 → "这个记忆和哪个实体相关?关联路径是什么?"(Who/Why/How)

典型场景:

✅ "用户之前问过哪些类似问题" → 向量检索

✅ "哪些人和这个项目有关联" → Graph

✅ "用户的知识图谱中有没有这个概念" → Graph

✅ "最近学到的新知识如何影响已有记忆" → 两者结合

`

3.2 一个简单的 Graph Memory 实现

`python

from dataclasses import dataclass, field

from typing import Dict, List, Set

from datetime import datetime

import json

@dataclass

class MemoryNode:

id: str

content: str

node_type: str # "person" | "concept" | "event" | "project"

properties: Dict = field(default_factory=dict)

created_at: datetime = field(default_factory=datetime.now)

importance: str = "medium"

@dataclass

class MemoryEdge:

from_node: str

to_node: str

relation: str # "works_on" | "knows" | "part_of" | "learned_from"

class GraphMemory:

def __init__(self, storage_path: str = "./graph_memory.json"):

self.nodes: Dict[str, MemoryNode] = {}

self.edges: List[MemoryEdge] = []

self.storage_path = storage_path

def add_memory(self, node: MemoryNode, connections: List[tuple] = None):

self.nodes[node.id] = node

if connections:

for target_id, relation in connections:

self.edges.append(MemoryEdge(node.id, target_id, relation))

self._save()

def query_related(self, node_id: str, depth: int = 2) -> List[MemoryNode]:

"""找出和某个节点相关的所有记忆(可指定深度)"""

visited = set()

queue = [(node_id, 0)]

result = []

while queue:

current, d = queue.pop(0)

if current in visited or d > depth:

continue

visited.add(current)

if current in self.nodes:

result.append(self.nodes[current])

# 找邻居

for edge in self.edges:

if edge.from_node == current and edge.to_node not in visited:

queue.append((edge.to_node, d + 1))

elif edge.to_node == current and edge.from_node not in visited:

queue.append((edge.from_node, d + 1))

return result

def _save(self):

with open(self.storage_path, 'w') as f:

json.dump({

"nodes": {k: vars(v) for k, v in self.nodes.items()},

"edges": [vars(e) for e in self.edges]

}, f, ensure_ascii=False, indent=2)

`

3.3 混合架构:向量 + Graph 的实战组合

`

┌─────────────────┐

│ User Query │

└────────┬────────┘

┌──────────────┴──────────────┐

▼ ▼

┌────────────────┐ ┌──────────────────┐

│ Vector Store │ │ Graph Store │

│ (Pinecone/ │ │ (Neo4j/本地JSON) │

│ Qdrant/Milvus)│ │ │

└───────┬────────┘ └────────┬─────────┘

│ │

│ "语义相关" │ "实体关系"

▼ ▼

┌─────────────────────────────────────────┐

│ Memory Fuser │

│ (合并检索结果,做去重和重要性加权) │

└────────────────────┬────────────────────┘

┌───────────────────────┐

│ LLM Context 构建 │

│ (记忆 + 当前任务) │

└───────────────────────┘

`

实际项目中,我推荐先用向量做快速召回,用 Graph 做精确补全。Graph 不需要每次都查——可以每天跑一次关系构建,然后向量检索结果里附带已经构建好的关系上下文。

四、记忆的衰减与更新策略

静态记忆有个致命问题:知识会过时。用户的偏好、项目状态、联系人信息都在变。记忆系统必须支持更新和衰减

4.1 基于时间的衰减模型

`python

from datetime import datetime, timedelta

import math

def compute_memory_weight(created_at: datetime, importance: str, now: datetime = None) -> float:

"""计算记忆的当前权重,核心思想:越老的记忆权重越低,重要记忆衰减更慢"""

now = now or datetime.now()

days_old = (now - created_at).days

base_decay = {

"high": 0.995, # 重要记忆每天衰减 0.5%

"medium": 0.980, # 普通记忆每天衰减 2%

"low": 0.950, # 不重要记忆快速衰减

}

decay_rate = base_decay.get(importance, 0.98)

# 指数衰减,但最低不低于 0.1

weight = max(0.1, math.pow(decay_rate, days_old))

return weight

`

4.2 主动更新 vs 被动覆盖

两种记忆更新模式:

被动覆盖:新记忆直接覆盖旧记忆。简单,但容易丢失重要的历史上下文。

主动合并(推荐):

`python

async def update_memory(new_fact: str, entity_id: str, memory_store):

"""记忆更新时做语义合并,而非简单覆盖"""

# 1. 找出同一实体的旧记忆

old_memories = await memory_store.query_by_entity(entity_id)

if not old_memories:

# 没有旧记忆,直接创建

await memory_store.create(new_fact, entity_id)

return

# 2. 检查新旧记忆是否冲突

conflicting = check_conflict(old_memories, new_fact)

if conflicting:

# 冲突时,保留两者的加权平均(新记忆权重更高)

merged = merge_with_conflict_resolution(old_memories, new_fact, new_weight=0.6)

await memory_store.update(entity_id, merged)

else:

# 不冲突,做增量更新

await memory_store.append(entity_id, new_fact)

`

五、生产级别的工程细节

5.1 记忆去重:同一事实不要存两次

`python

import hashlib

def fact_fingerprint(content: str) -> str:

"""对记忆内容做指纹,避免重复存储"""

# 归一化:去除多余空格、转小写、排序关键词

normalized = " ".join(sorted(content.lower().split()))

return hashlib.md5(normalized.encode()).hexdigest()[:16]

`

5.2 记忆压缩:上下文满了怎么办

当记忆积累到一定规模,检索回来的 context 可能超出 LLM 窗口。这是常见问题,解决方案:

1. Summary 压缩:对同一实体的多条记忆做摘要,存一条高层次的总结

2. 重要性过滤:只召回 importance=high 的记忆

3. 时间窗口:限制只检索最近 N 天的记忆

4. 分层索引:高频记忆 → 中频 → 低频,分层管理

5.3 隐私与安全:记忆系统不是法外之地

最后但最重要的一点:记忆系统存储的是用户隐私数据

  • 敏感记忆要加密存储(AES-256)
  • 对记忆的访问要有权限控制
  • 支持"删除记忆"操作(GDPR 要求)
  • 定期做记忆审计:哪些记忆还在?有没有过时敏感信息?

`python

class EncryptedMemoryStore:

def __init__(self, key: bytes):

from cryptography.fernet import Fernet

self.cipher = Fernet(key)

async def store(self, memory_id: str, content: str):

encrypted = self.cipher.encrypt(content.encode())

await self.db.put(memory_id, encrypted)

async def retrieve(self, memory_id: str) -> str:

encrypted = await self.db.get(memory_id)

if not encrypted:

return ""

return self.cipher.decrypt(encrypted).decode()

async def delete(self, memory_id: str):

# 真正删除,而非软删除

await self.db.delete(memory_id)

`

六、实战建议:我的记忆系统配置

基于过去一年在多个 Agent 项目中的实践,以下是我的推荐配置:

`

Embedding: BGE-M3(自托管,中文好,免费)

向量数据库: Qdrant(轻量,Rust 实现,支持混合检索)

图数据库: 低频场景用 JSON 文件,高频用 Neo4j

Chunk 大小: 800 tokens,overlap 100

检索策略: 向量 top_k=10 + importance 过滤 → rerank → top_k=5

衰减策略: importance=high 衰减率 0.995/天,low 衰减率 0.95/天

`

这套配置在我自己的 Agent 系统里运行良好,内存占用稳定,检索延迟 <50ms。

结语

记忆系统是 AI Agent 最重要的基础设施之一,却往往被当作"后期再加"的模块。早期设计好记忆的层次、更新策略和存储结构,后期能省下大量重建成本。

核心原则只有三条:

1. 分层清晰:工作/短期/长期各有分工,不要混用

2. 检索为王:记忆存了找不出来等于没存,Chunking 策略和索引设计是关键

3. 持续衰减:记忆会过时,没有衰减机制的系统的终态是垃圾堆

做好了记忆系统,你的 Agent 才真正从"每次都是新的对话"进化到"持续学习的数字专家"。