今天翻到一篇不错的技术分享，看完之后自己也琢磨了一下，把思路梳理记录下来。
文章结构：
引言：为什么 Agent 需要记忆系统（LLM 的无状态缺陷）
三层架构总览
Working Memory 深度解析
Episodic Memory（向量存储 + 召回）
Semantic Memory（Agentic RAG）
Memory Graph 融合检索
记忆一致性与冲突解决（冲突检测）
生产级工程实践（Embedding 选型、分块策略、可观测性）
技术演进路径
前沿进展（GraphRAG、MemoryOS、隐私保护）
总结
摘要
大语言模型本身是无状态的——每次推理独立进行，上下文窗口一旦关闭，所有"经历"灰飞烟灭。然而真正有价值的 AI Agent
必须具备跨会话的持续学习能力、对历史事件的精准回溯能力，以及对领域知识的深度利用能力。这里系统梳理 Agent
记忆系统的三层架构设计（Working Memory / Episodic Memory / Semantic
Memory），深入剖析每一层的实现机制、工程挑战与优化策略，并通过完整代码示例展示从朴素 RAG 到 Memory Graph
的演进路径，最后结合记忆压缩、遗忘机制、记忆一致性等生产级关键问题给出工程实践建议。
一、为什么 Agent 需要记忆系统
1.1 LLM 的天然缺陷：无状态与遗忘
当前主流大语言模型（GPT-4、Claude、Qwen 等）在架构上本质是一个"无状态函数"：输入 token 序列，输出 token 序列，推理结束后不保留任何内部状态。这意味着：
- 跨会话失忆：用户昨天告诉 Agent “我喜欢 Python，不喜欢 Java”，今天 Agent 完全不记得；
- 上下文窗口瓶颈：即便在单会话内，当对话长度超过模型上下文窗口（如 128K tokens），早期信息会被截断丢弃；
- 知识静态化：模型权重在训练后固化，无法实时吸收新知识，只能通过 RAG 或 Fine-tuning 补偿。
1.2 记忆能力对 Agent 的战略价值
从产品维度看，记忆能力直接决定 Agent 的"温度"——用户是否感受到被理解、被记住。从技术维度看，记忆系统是 Agent 从"单次问答工具"升级为"持续学习助理"的核心基础设施，也是 Multi-Agent 系统中知识共享、经验传递的关键通道。
1.3 人类记忆的启发
认知科学将人类记忆分为：感觉记忆（Sensory）、短期工作记忆（Working）、长期情节记忆（Episodic）、长期语义记忆（Semantic）、程序性记忆（Procedural）等层次。Agent 记忆系统的架构设计高度借鉴了这一体系，但在工程实现上有其独特的约束与创新点。
二、三层记忆架构总览
如上方架构图所示，主流 Agent 记忆系统可分为三个层次，每层在时效性、容量、访问代价上呈现梯度差异。
2.1 三层架构对比
维度Working MemoryEpisodic MemorySemantic Memory对应概念短期工作记忆情节（事件）记忆语义（知识）记忆存储介质Context Window向量数据库 / KV Store向量DB + 知识图谱时效范围单次会话内历史会话 / 事件日志长期稳定知识容量极小（受限于上下文窗口）中等（可扩展）大（TB 级可行）访问延迟极低（in-memory）中等（向量检索 ~10ms）中高（KG 查询 ~50ms+）更新频率实时（每轮对话）会话结束后批量写入低频（知识库更新）典型实现Prompt 拼接ChromaDB / PineconeNeo4j / Weaviate
2.2 信息流转路径
如上方生命周期流程图所示，一次完整的 Agent 推理过程涉及：
1. 用户输入触发感知层；
2. 并行从 Episodic DB 和 Semantic KB 检索相关记忆片段；
3. 将检索结果注入 Working Memory（Prompt 组装）；
4. LLM 基于完整上下文推理、执行工具调用；
5. 会话结束后将新产生的信息压缩、向量化，写回 Episodic Memory；
6. 定期知识提炼将高频稳定知识升华至 Semantic Memory。
三、Working Memory：上下文窗口的精细化管理
3.1 核心挑战：有限空间的最大化利用
Working Memory 对应 LLM 的 Context Window。以 Claude 3.5 为例，其 200K token 的窗口看似充裕，但一旦加入系统提示、工具描述、历史对话、检索内容，实际可用空间往往只剩 20%~40%。
Working Memory 管理的核心命题是：在有限的 token 预算内，放置信息价值密度最高的内容。
3.2 Prompt 结构标准化
一个生产级 Working Memory 的 Prompt 结构通常如下：
[System Prompt] int:
"""
艾宾浩斯遗忘曲线：删除超龄且低重要性的记忆
返回删除条数
"""
cutoff = (
datetime.utcnow() - timedelta(days=max_age_days)
).isoformat()
# 查询候选删除记忆（低重要性 + 超龄）
all_metas = self.collection.get(include=["metadatas", "ids"])
to_delete = []
for mid, meta in zip(all_metas["ids"], all_metas["metadatas"]):
ts = meta.get("timestamp", "")
importance = float(meta.get("importance", 0.5))
if ts < cutoff and importance < 0.4:
to_delete.append(mid)
if to_delete:
self.collection.delete(ids=to_delete)
return len(to_delete)
五、Semantic Memory：知识图谱与结构化知识的融合
5.1 语义记忆 vs 情节记忆的本质区别
情节记忆是"具体的事件"，语义记忆是"抽象的知识"。比如：
- 情节记忆：
"2024-06-10，用户说他在南京工作，偏好用 Python 做数据分析" - 语义记忆：
"Python 是一种解释型编程语言，适合数据科学领域"
5.2 从朴素 RAG 到 Agentic RAG
朴素 RAG（Naive RAG）的流程是：将文档切片 → 向量化 → 存储 → Query 时检索最相似片段 → 注入 Prompt。这个方案在简单问答场景效果不错，但在 Agent 场景下有明显短板：
- 单跳检索不足：繁琐问题需要多个知识片段的组合推理；
- Query 质量依赖用户输入：用户的自然语言往往不是最优检索词；
- 无法处理隐式依赖：知识点之间的关联关系未被捕获。
class AgenticRAG:
"""
Agentic RAG：Agent 自主决策检索策略
支持 Query 改写、多跳检索、自我反思
"""
def __init__(self, semantic_store, llm_client):
self.store = semantic_store
self.llm = llm_client
def query_rewrite(self, original_query: str) -> List[str]:
"""
将用户原始 Query 改写为多个更精准的检索词
"""
prompt = f"""
将以下用户问题改写为 3 个不同角度的检索查询（JSON 数组格式）：
原始问题：{original_query}
改写后的查询（只返回 JSON，不要其他内容）："""
resp = self.llm.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
)
import json
text = resp.choices[0].message.content.strip()
try:
queries = json.loads(text)
except Exception:
queries = [original_query]
return queries
def multi_hop_retrieve(
self, query: str, max_hops: int = 3
) -> List[str]:
"""
多跳检索：每跳结果作为下一跳的检索种子
"""
all_results = []
current_queries = self.query_rewrite(query)
for hop in range(max_hops):
new_queries = []
for q in current_queries:
results = self.store.similarity_search(q, k=3)
all_results.extend(results)
# 从检索结果中提取新的检索线索
if hop < max_hops - 1:
follow_up = self._extract_follow_up_queries(
q, results
)
new_queries.extend(follow_up)
current_queries = new_queries
if not current_queries:
break
# 去重，按相关性排序
seen = set()
unique_results = []
for r in all_results:
if r.page_content not in seen:
seen.add(r.page_content)
unique_results.append(r)
return unique_results[:10]
def _extract_follow_up_queries(
self, original_q: str, results: List
) -> List[str]:
"""从已检索内容提取新的检索方向"""
context = "\n".join(r.page_content for r in results[:2])
prompt = f"""
基于问题"{original_q}"和以下检索结果，
提取 1-2 个需要进一步检索的子问题（JSON 数组）：
{context}
子问题："""
resp = self.llm.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
)
try:
return json.loads(resp.choices[0].message.content.strip())
except Exception:
return []
def self_reflect(
self, query: str, retrieved_docs: List[str], answer: str
) -> dict:
"""
自我反思：评估答案是否充分，是否需要更多检索
"""
context = "\n".join(retrieved_docs[:5])
prompt = f"""
问题：{query}
检索到的资料：{context}
当前答案：{answer}
请评估：
1. 答案是否完整回答了问题？（是/否）
2. 是否还需要补充检索？（是/否）
3. 如果需要，指出缺失的信息点（一句话）
以 JSON 格式返回：{{"complete": bool, "need_more": bool, "missing": str}}"""
resp = self.llm.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
)
try:
return json.loads(resp.choices[0].message.content.strip())
except Exception:
return {"complete": True, "need_more": False, "missing": ""}
5.3 知识图谱：结构化关系的语义记忆
向量数据库擅长语义相似度检索，但对实体关系的表达能力较弱。知识图谱（Knowledge Graph，KG）则以图结构存储实体与关系，能够支持繁琐的关系推理。
以 Neo4j 为例，将 Agent 的 Semantic Memory 部分存储在知识图谱中：
from neo4j import GraphDatabase
class MemoryGraph:
"""
基于 Neo4j 的 Memory Graph：
存储实体、关系、属性，支持图遍历查询
"""
def __init__(self, uri: str, user: str, password: str):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def store_entity(self, entity_type: str, name: str, props: dict):
"""写入或更新一个实体节点"""
with self.driver.session() as session:
session.run(
f"""
MERGE (e:{entity_type} {{name: $name}})
SET e += $props
""",
name=name,
props=props,
)
def store_relation(
self,
from_entity: str,
relation: str,
to_entity: str,
props: dict = None,
):
"""写入两个实体之间的关系"""
with self.driver.session() as session:
session.run(
f"""
MATCH (a {{name: $from_name}})
MATCH (b {{name: $to_name}})
MERGE (a)-[r:{relation}]->(b)
SET r += $props
""",
from_name=from_entity,
to_name=to_entity,
props=props or {},
)
def query_neighbors(
self, entity_name: str, depth: int = 2
) -> List[dict]:
"""
查询某实体的 N 跳邻居：
用于向 Agent 提供关联知识上下文
"""
with self.driver.session() as session:
result = session.run(
f"""
MATCH (start {{name: $name}})-[*1..{depth}]-(neighbor)
RETURN DISTINCT neighbor.name AS name,
labels(neighbor) AS types,
neighbor AS props
LIMIT 20
""",
name=entity_name,
)
return [
{
"name": r["name"],
"types": r["types"],
}
for r in result
]
def extract_and_store(
self, text: str, llm_client
) -> None:
"""
从文本中提取实体和关系，自动写入知识图谱
（Information Extraction 管线）
"""
prompt = f"""
从以下文本中提取所有实体和关系，以 JSON 格式返回：
{{
"entities": [{{"name": "...", "type": "Person|Organization|Product|Concept"}}],
"relations": [{{"from": "...", "relation": "...", "to": "..."}}]
}}
文本：{text}
只返回 JSON，不要其他内容。"""
resp = llm_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
)
try:
data = json.loads(resp.choices[0].message.content.strip())
except Exception:
return
for entity in data.get("entities", []):
self.store_entity(
entity.get("type", "Entity"),
entity["name"],
{},
)
for rel in data.get("relations", []):
self.store_relation(
rel["from"],
rel["relation"].upper().replace(" ", "_"),
rel["to"],
)
六、Memory Graph：向量与图的融合检索### 6.1 融合检索的动机
单纯的向量检索善于捕捉语义相似性，但对实体关系的推理能力弱；单纯的图谱遍历精准但依赖实体识别质量，对语义模糊的 Query 泛化性差。Memory Graph 的核心思路是将两条检索路径并行执行，再通过融合策略取长补短。
如上图所示，融合后的结果通过 Reciprocal Rank Fusion（RRF） 或 Cross-encoder 重排序 产生最终排名，再注入 Agent 的 Working Memory。
6.2 代码示例：混合检索融合
from typing import List, Tuple
class MemoryGraphRetriever:
"""
Memory Graph 混合检索器：
向量检索 + 知识图谱 并行，RRF 融合
"""
def __init__(
self,
episodic_store: EpisodicMemoryStore,
memory_graph: MemoryGraph,
):
self.episodic = episodic_store
self.graph = memory_graph
def _reciprocal_rank_fusion(
self,
result_lists: List[List[Tuple[str, float]]],
k: int = 60,
) -> List[Tuple[str, float]]:
"""
Reciprocal Rank Fusion (RRF)：
多路结果融合，k 是平滑参数（通常取 60）
"""
scores: dict[str, float] = {}
for result_list in result_lists:
for rank, (doc_id, _) in enumerate(result_list):
scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank + 1)
sorted_results = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return sorted_results
def retrieve(
self,
query: str,
entities: List[str] = None,
top_k: int = 8,
) -> List[dict]:
"""
并行执行向量检索 + 图谱检索，RRF 融合
"""
import asyncio
# --- 路径 1：向量语义检索 ---
vector_results = self.episodic.recall(query, top_k=top_k * 2)
vector_ranked = [
(r["content"], r["score"]) for r in vector_results
]
# --- 路径 2：图谱关系检索 ---
graph_ranked = []
if entities:
for entity in entities:
neighbors = self.graph.query_neighbors(entity, depth=2)
# 将图谱结果转换为文本片段（简化示例）
for neighbor in neighbors:
desc = f"{entity} 关联到 {neighbor['name']} (类型: {','.join(neighbor['types'])})"
graph_ranked.append((desc, 0.7)) # 固定初始分
# --- RRF 融合 ---
fused = self._reciprocal_rank_fusion(
[vector_ranked, graph_ranked]
)
# 取 top_k 并组装返回结构
final_results = []
content_map = {r["content"]: r for r in vector_results}
for doc_id, score in fused[:top_k]:
base = content_map.get(doc_id, {})
final_results.append({
"content": doc_id,
"score": score,
"metadata": base.get("metadata", {}),
"source": "vector" if doc_id in content_map else "graph",
})
return final_results
七、记忆一致性与冲突解决
7.1 记忆矛盾问题
随着时间推移，Agent 的记忆库会产生语义冲突：
- 用户某天说"我在北京工作"，三个月后说"我已经搬到上海了"；
- 同一知识在不同版本的文档中有不同描述；
- LLM 总结产生的摘要可能丢失或曲解原始信息。
7.2 记忆冲突检测与解决策略
class MemoryConsistencyChecker:
"""
记忆一致性检测器：
写入新记忆前检测是否与已有记忆矛盾，并决策处理方式
"""
def __init__(self, episodic_store, llm_client):
self.store = episodic_store
self.llm = llm_client
def check_and_resolve(
self, new_memory: str, existing_memories: List[str]
) -> dict:
"""
返回：
- action: "insert" | "replace" | "merge" | "discard"
- reason: 原因说明
- merged: 合并后内容（若 action=="merge"）
"""
if not existing_memories:
return {"action": "insert", "reason": "无已有记忆"}
context = "\n".join(
f"[{i+1}] {m}" for i, m in enumerate(existing_memories)
)
prompt = f"""
新记忆：{new_memory}
已有相关记忆：
{context}
请判断新记忆与已有记忆的关系，返回 JSON：
{{
"conflict": true/false,
"action": "insert"|"replace"|"merge"|"discard",
"reason": "...",
"merged": "..." // 仅 action=merge 时填写
}}
判断标准：
- 新记忆是更新的事实 → replace（指定要替换的编号）
- 新记忆提供了补充信息 → merge
- 新记忆完全矛盾且旧的更可信 → discard
- 新记忆是独立新信息 → insert"""
resp = self.llm.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
)
try:
return json.loads(resp.choices[0].message.content.strip())
except Exception:
return {"action": "insert", "reason": "解析失败，默认插入"}
八、生产级工程实践：避坑指南
8.1 Embedding 模型的选择
Embedding 质量直接决定向量检索精度。生产建议：
- 中文场景：优先选择
text-embedding-3-large（OpenAI）或bge-large-zh（BAAI，开源）； - 成本敏感场景：
text-embedding-3-small在大多数任务上性价比极高； - 离线/私有部署：
bge-m3（支持多语言多粒度检索）或stella_en_400M_v5。
8.2 分块策略（Chunking）的工程细节
文档切片质量是 RAG 效果的基础，几个常被忽视的要点：
- 语义分块优于固定长度分块：使用句边界、段落边界切分，避免在句子中间截断；
- 上下文重叠：相邻 chunk 保留 20% 重叠，避免关键信息落在边界；
- 层次化分块：大 chunk 用于检索（提供更多上下文），小 chunk 用于精确定位；
- 元数据富化：在 chunk 中注入文档标题、章节信息，帮助模型理解局部内容的全局位置。
8.3 向量检索的精度优化
- Hybrid Search：向量检索 + BM25 关键词检索并行，RRF 融合，比纯向量检索通常提升 10%~20% 的 Recall；
- Reranking：检索后用 Cross-encoder（如
bge-reranker-v2-m3）对 top-50 结果重新打分，取 top-5，精度显著提升； - 查询扩展：用 LLM 对用户 Query 进行 HyDE（假设文档嵌入）或多角度改写，增强检索召回率。
8.4 记忆系统的可观测性
生产环境必须对记忆系统进行监控：
- 检索命中率：监控每次召回的平均分值，分值持续下降说明记忆质量在退化；
- 记忆库规模：监控向量数量增长，结合遗忘机制控制规模；
- 检索延迟：P99 延迟超过 100ms 需考虑缓存或索引优化；
- 幻觉率：定期抽样检查 Agent 输出是否与召回记忆内容一致。
九、从 RAG 到 Memory Graph：演进路径总结### 9.1 四个阶段的核心差异
阶段一（Naive RAG）：文档切块→向量化→相似度检索→塞入 Prompt。实现成本极低，但对繁琐推理无能为力，且没有跨会话记忆能力。适合：简单文档问答 MVP。
阶段二（Advanced RAG）：引入 Query 改写、混合检索（向量+BM25）、Cross-encoder 重排序。显著提升检索精度，但仍属于"单次无状态检索"。适合：知识库问答生产环境。
阶段三（Episodic RAG）：引入多层记忆架构，情节记忆支持跨会话召回，Working Memory 精细化管理。Agent 开始有了"记住你是谁"的能力。适合：有持续用户关系的 Agent 产品。
阶段四（Memory Graph）：向量检索与知识图谱融合，支持多跳关系推理，记忆自动提炼与知识图谱动态更新。这是当前学术前沿与工业探索的主战场，代表性工作包含 Microsoft 的 GraphRAG、Meta 的 MemoryOS 等。
十、前沿进展与未来方向
10.1 GraphRAG：微软的图增强 RAG
2024 年微软发布的 GraphRAG 将社区检测算法（Leiden 算法）应用于文档知识图谱构建，通过层次化社区摘要实现"全局语义理解"——这是传统向量 RAG 无法做到的。在"这批文档的整体主题是什么"类问题上，GraphRAG 相比 Naive RAG 有质的飞跃。
10.2 MemoryOS：分层记忆操作系统
MemoryOS（2025 年论文）将操作系统的页面置换算法（LRU、优先级队列）引入 Agent 记忆管理，构建了类 OS 的三级记忆层次（短期缓冲/中期整合/长期存储），并通过记忆热度动态调度，在多任务长对话场景下显著优于固定容量的 Episodic Memory。
10.3 记忆的个性化与隐私
随着记忆系统能力增强，隐私保护成为核心挑战：
- 差分隐私 Embedding：对用户私有信息的向量化引入噪声，防止原始内容被逆向还原；
- 记忆权限分层：区分个人记忆（仅本人可访问）与共享记忆（多 Agent 共享），设计细粒度访问控制；
- 主动遗忘机制：支持用户主动触发特定记忆的删除（GDPR 合规需求）。
10.4 持续学习与记忆的联合优化
长远来看，外部记忆系统与模型权重更新（持续学习/增量微调）的协同设计是终极方向：记忆系统捕捉高频稳定知识后，触发模型的增量微调，将知识"内化"进权重；而权重微调又能提升记忆检索的语义对齐精度——形成正向飞轮。
十一、总结
这里系统梳理了 AI Agent 记忆系统的三层架构，核心结论如下：
- Working Memory 是 Agent 的即时意识，关键工程任务是 token 预算管理和 Prompt 精细化组装；
- Episodic Memory 是 Agent 的个人经历档案，向量数据库 + 语义检索是主流实现，遗忘机制和重要性评估是生产必需；
- Semantic Memory 是 Agent 的知识基础，向量 RAG 解决语义检索，知识图谱解决关系推理，两者融合（Memory Graph）是当前前沿；
- 从 Naive RAG 到 Memory Graph 的演进，本质是 Agent 从"工具"向"具身认知主体"的升级——它开始有了记忆，有了经验，也开始真正地"认识"用户。
这篇笔记就先到这里，后面用到新的思路或者发现有问题再补充。
评论 (0)
暂无评论