腾讯开源 Agent Memory：四层渐进式记忆架构如何让 AI Agent 真正「记住」用户——从碎片对话到结构化知识的工程实践
一、背景：当 AI Agent 遇见「遗忘症」
过去两年，AI Agent 是整个 AI 产业最火热的赛道。从 OpenAI 的 ChatGPT Plugins 到 Anthropic 的 Claude Code，从微软的 Copilot Stack 到国内的扣子（Coze）、Dify，业界在「让大模型具备工具调用和任务执行能力」这件事上已经积累了相当成熟的方法论。然而，当我们在生产环境中真正部署 Agent 时，发现了一个比工具调用更棘手的问题——记忆丢失。
想象这样一个场景：用户 Alice 第一次使用公司的智能客服 Agent，说「我上个月买过你们的 Pro 套餐，还剩 15 天到期」。Agent 的回复无懈可击——续费提醒、套餐对比、升级优惠，一气呵成。第二天 Alice 再次打开对话：「续费」。Agent 一脸茫然：「您好，请问您想了解什么产品？」
这不是 Agent 不够智能，而是记忆断层——Agent 在每次新的会话中丢失了关于 Alice 的所有上下文信息。它记住了昨天的对话，但无法跨越会话边界积累知识。传统 RAG 方案可以检索静态文档，但无法捕捉用户的个性化偏好、历史交互模式、以及那些「模糊但重要」的非结构化线索。
腾讯云数据库团队开源的 TencentDB-Agent-Memory（以下简称 Agent Memory）正是为解决这一问题而生。这个项目在 GitHub 上已累计 18.1k Stars，周新增 8,046 Stars，位居 GitHub Trending 周榜第二位。本文将深度拆解其四层渐进式记忆架构、核心代码实现、以及在生产环境中的实战集成方案。
二、为什么现有的记忆方案都不够用
在深入 Agent Memory 的设计之前，我们需要先理解为什么它比现有的方案更有优势。
2.1 三种主流方案及其局限
方案一：全量塞进上下文窗口
这是最简单的思路——把用户所有历史对话记录都作为上下文传给大模型。简单是简单，但问题也最致命：
# 方案一：全量上下文
def get_full_context(user_id: str, max_tokens: int = 200000) -> str:
"""将用户所有历史对话塞进上下文窗口"""
history = db.fetch_all(f"SELECT role, content FROM messages WHERE user_id = {user_id}")
context = "\n".join([f"{m['role']}: {m['content']}" for m in history])
# 问题：Token 超出窗口上限，直接爆掉
return context[:max_tokens * 4] # 粗暴截断，信息丢失
以 GPT-4 Turbo 128K 上下文窗口为例，一个普通用户半年的对话记录轻松超过这个上限。即使没有超限，大量的无关历史也会稀释关键信息，导致模型在复杂推理时丢失焦点。学术界将这个问题称为「lost in the middle」——模型对长上下文中中间位置的信息记忆最弱。
方案二：向量数据库检索（Naive RAG）
目前最流行的方案。将历史对话切成块（Chunk），向量化后存入向量数据库，查询时做相似度检索：
# 方案二：向量检索式记忆
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
class NaiveVectorMemory:
def __init__(self):
self.vectorstore = Chroma(embedding_function=OpenAIEmbeddings())
def add_message(self, text: str, metadata: dict):
# 固定大小切块，无语义感知
chunks = text_chunking(text, chunk_size=500, overlap=50)
self.vectorstore.add_texts(chunks, metadatas=[metadata] * len(chunks))
def retrieve(self, query: str, top_k: int = 5) -> list[str]:
return self.vectorstore.similarity_search(query, k=top_k)
# 问题一：按query关键词匹配，无法理解「还剩15天」=「即将到期」
# 问题二：各chunks之间无关联，碎片化严重
# 问题三：所有记忆等权重，无法区分「重要事实」和「闲聊废话」
向量检索的局限在于它本质上是关键词匹配的高级版——它能找到语义相似的内容，但无法理解事件的时间线、因果关系和重要性层级。「用户上个月购买了 Pro 套餐」和「用户昨天抱怨加载速度慢」在向量空间中可能距离很远，但它们对理解用户当前意图同等重要。
方案三：纯 Prompt 工程（System Prompt Engineering）
通过精心设计的 System Prompt 让模型「假装」有记忆：
SYSTEM_PROMPT = """
你是一个贴心的私人助理。请记住以下关于用户的信息：
- 用户姓名：[从对话中提取]
- 用户偏好：[从对话中提取]
- 未完成的任务：[从对话中提取]
注意：上述信息可能已过时，请结合最新对话综合判断。
"""
这种方式在演示中看起来不错，但实则是沙上建塔——没有任何持久化存储，服务器重启、上下文窗口刷新、或切换到另一个 Agent 实例，所有「记忆」立即消失。而且，随着要记住的信息增多，System Prompt 本身也会膨胀，进一步挤压有效上下文空间。
2.2 Agent Memory 的设计哲学
腾讯 Agent Memory 团队在设计文档中提出了一个核心洞察：记忆不是一次性的检索操作，而是一个渐进式提炼的过程。
就像人类大脑处理记忆一样——我们不会记住每一次对话的完整原文，而是自动提取关键事实、形成概念、将相关记忆关联起来。大脑的长期记忆是经过压缩、抽象、关联后的知识结构，而非原始数据的堆叠。
基于这一认知，Agent Memory 提出了四层渐进式记忆架构：
| 层级 | 名称 | 保留内容 | 数据形态 |
|---|---|---|---|
| L0 | 原始对话层 | 完整对话原文 | 结构化 JSON |
| L1 | 原子记忆层 | 事实 + 约束 + 意图 | 结构化标签 |
| L2 | 场景记忆层 | 项目/任务级别的知识块 | 图谱节点 |
| L3 | 用户画像层 | 个性化偏好与行为模式 | Profile 对象 |
这四层之间的关系是：自底向上提炼，自顶向下检索——L0 是原始数据湖，L1-L3 是逐层抽象的知识金字塔；检索时则从 L3 到 L0 层层递进，确保记忆召回既精准又全面。
三、架构深度解析：从数据流到四层记忆
3.1 整体系统架构
Agent Memory 的架构分为四大组件：
┌─────────────────────────────────────────────────────────────┐
│ Agent Application │
│ (OpenClaw / Claude Code / Dify / Coze / 自研 Agent) │
└─────────────────┬───────────────────────────────────────────┘
│ 工具调用 (save_memory / recall_memory)
▼
┌─────────────────────────────────────────────────────────────┐
│ Memory Service Layer │
│ ┌─────────────┐ ┌─────────────┐ ┌──────────────────────┐ │
│ │ Ingestion │ │ Extraction │ │ Retrieval & Ranking │ │
│ │ Pipeline │ │ Engine │ │ Engine │ │
│ └─────────────┘ └─────────────┘ └──────────────────────┘ │
└─────────────────┬───────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Storage Layer (Tencent Cloud VectorDB) │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │
│ │ L0 Raw │ │ L1 Atomic│ │ L2 Scene │ │ L3 Persona │ │
│ │ Store │ │ Store │ │ Graph │ │ Profile DB │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────┘
3.2 L0：原始对话层——数据湖的设计
L0 层是最底层的存储，完整保留了每次对话的原始数据。它解决的是可审计性问题——当需要回溯用户说过的某句原话时，必须有完整的原始记录。
// L0 原始对话存储的数据结构
interface RawConversation {
session_id: string; // 会话 ID
user_id: string; // 用户 ID
timestamp: number; // Unix 时间戳（毫秒）
messages: ConversationMessage[];
metadata: ConversationMetadata;
}
interface ConversationMessage {
role: 'user' | 'assistant' | 'system';
content: string;
token_count: number; // 用于计算上下文窗口占用
attachments?: Attachment[]; // 支持附件
}
interface ConversationMetadata {
platform: string; // 来自哪个 Agent 平台
intent?: string; // 可选：意图分类标签
satisfaction?: number; // 可选：用户满意度评分
language: string; // 对话语言
}
存储选型上，L0 采用 Tencent Cloud VectorDB（原-postgres pgvector） 的 JSONB 列存储。原因有三：JSONB 支持灵活 schema，便于存储不同 Agent 平台的多样化数据结构；全文索引（GIN Index）可以快速做时间范围查询和关键词检索；与向量检索共用同一数据库，运维成本最低。
-- L0 表结构（PostgreSQL JSONB）
CREATE TABLE agent_memory_l0 (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
user_id VARCHAR(64) NOT NULL,
session_id VARCHAR(128) NOT NULL,
timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
messages JSONB NOT NULL, -- 完整对话原文
metadata JSONB, -- 元数据
message_count INTEGER, -- 本次对话轮次数
total_tokens INTEGER, -- 总 token 数（用于计费分析）
created_at TIMESTAMPTZ DEFAULT NOW()
);
-- 高频查询索引
CREATE INDEX idx_l0_user_time ON agent_memory_l0 (user_id, timestamp DESC);
CREATE INDEX idx_l0_session ON agent_memory_l0 (session_id);
CREATE INDEX idx_l0_messages_gin ON agent_memory_l0 USING GIN (messages);
3.3 L1：原子记忆层——从对话到事实
L1 是整个系统的智能核心。它通过 LLM 的结构化提取能力，将原始对话提炼为一组原子化的「记忆原子」——每一个原子代表一个独立的事实、约束或意图。
// L1 原子记忆的数据结构
interface AtomicMemory {
id: string; // 雪花 ID
user_id: string;
session_id: string; // 来源会话
source_snippet: string; // 原文摘录（用于溯源）
// 记忆内容
type: MemoryType; // 记忆类型
content: string; // 自然语言描述
entities: ExtractedEntity[]; // 提取的实体
// 元信息
importance: 1 | 2 | 3 | 4 | 5; // 重要性评分（LLM 评估）
ttl_days: number; // 自然过期时间（天）
verified: boolean; // 是否经过二次验证
created_at: number;
last_accessed: number;
access_count: number; // 访问频次（用于热度加权）
}
type MemoryType =
| 'fact' // 客观事实（用户买了什么、在哪里工作）
| 'preference' // 用户偏好（喜欢简洁界面、偏好中文客服）
| 'constraint' // 约束条件（预算不超500、必须在周五前完成）
| 'intent' // 当前意图（正在比较产品、准备下单）
| 'knowledge' // 背景知识（用户是金融行业从业者）
| 'relationship'; // 关系信息（用户的老板是XX、同事YY）
L1 提取的 Prompt 设计是这里最有技术含量的部分。Agent Memory 的提取器使用了一个精心设计的 few-shot prompt：
# L1 原子记忆提取器核心逻辑
EXTRACTION_PROMPT = """
你是一个记忆提取专家。请从用户的对话中提取所有值得长期记忆的信息。
## 提取规则
1. 只提取客观可验证的事实，不提取情绪感受
2. 每个记忆原子必须是一个独立的、不需要额外上下文就能理解的句子
3. 同一实体在不同轮次中出现的，以最后一次为准（更新而非新建）
4. 重要性评分标准：
5分：直接影响当前任务执行或长期决策（价格、截止日期、个人关键信息）
4分：显著影响交互体验（中度偏好、频繁使用的功能）
3分：一般性偏好或背景信息
2分：闲聊中的偶然提及
1分：明显无关的噪音信息
## 输出格式（JSON数组）
[
{{
"type": "fact|preference|constraint|intent|knowledge|relationship",
"content": "提取的记忆（中文，完整句子）",
"entities": [{{"name": "实体名", "type": "person|product|location|organization|event"}}],
"importance": 1-5,
"ttl_days": 数值（fact默认180天，preference默认365天，constraint默认30天，intent默认7天）
}}
]
## 示例
用户说：「我是做金融的，平时用 Python 比较多」
输出：[{{"type": "knowledge", "content": "用户从事金融行业", "entities": [{{"name": "金融", "type": "industry"}}], "importance": 3, "ttl_days": 365}}]
"""
def extract_atomic_memories(messages: list[dict]) -> list[AtomicMemory]:
"""从对话消息列表中提取原子记忆"""
# 1. 构建上下文：将最近的N条消息作为上下文窗口
recent_messages = messages[-10:] # 最近10轮对话
context = format_conversation(recent_messages)
# 2. 调用 LLM 进行结构化提取
response = llm.structured_output(
context,
schema=AtomicMemory.schema(), # Pydantic schema
prompt=EXTRACTION_PROMPT
)
# 3. 记忆去重：新提取的记忆与已有点记忆做相似度比对
deduplicated = []
for new_mem in response:
is_duplicate = check_similarity(
new_mem.content,
existing_memories[session_id],
threshold=0.85 # 相似度 > 85% 视为重复
)
if not is_duplicate:
deduplicated.append(new_mem)
else:
# 更新已有点记忆的 TTL 和重要性
existing_mem.update_importance(new_mem)
return deduplicated
去重机制是 L1 层最关键的性能优化——如果不进行去重，同一个事实（如「用户的公司叫 ABC」）会在每次相关对话后重复存储，不仅浪费存储空间，更会在检索时引入大量噪音。
3.4 L2：场景记忆层——知识图谱的构建
L1 的原子记忆是扁平的，但现实世界中的知识是有结构的。L2 层通过知识图谱将相关的原子记忆组织成「场景」——一个场景对应一个完整的任务、项目或话题。
// L2 场景记忆层 - 知识图谱节点
interface SceneNode {
id: string;
user_id: string;
// 场景基本信息
name: string; // 场景名称（如"Pro套餐续费决策"）
scene_type: SceneType;
status: 'active' | 'completed' | 'archived';
// 图谱关系
atomic_memory_ids: string[]; // 关联的 L1 记忆 ID
temporal_span: { // 场景覆盖的时间范围
start: number;
end: number;
};
// 图谱语义关系（定义节点之间的关联）
relations: SceneRelation[];
// 场景级别的摘要（用于快速检索）
summary: string;
key_decisions: string[]; // 关键决策点
open_issues: string[]; // 未解决的问题
created_at: number;
last_interaction: number;
}
type SceneType =
| 'task' // 具体任务（"完成季度报告"）
| 'project' // 项目（"部署新的推荐系统"）
| 'topic' // 话题（"对比三家云服务商"）
| 'relationship'; // 关系维护（"跟进客户反馈"）
// 场景间关系（跨场景知识关联）
interface SceneRelation {
target_scene_id: string;
relation_type: 'subtask_of' | 'related_to' | 'blocks' | 'supersedes';
weight: number; // 0-1，关联强度
}
L2 层的提取发生在会话结束后。当用户关闭一个会话或超过 30 分钟无交互时，触发 L2 场景提取流程：
def build_scene_from_memories(session_id: str) -> SceneNode:
"""从会话的 L1 原子记忆中构建场景节点"""
atomic_memories = memory_store.get_l1_memories(session_id)
# 1. 场景聚类：将相关记忆分组
# 使用embedding相似度 + 时间邻近性做层次聚类
clusters = hierarchical_clustering(
atomic_memories,
similarity_threshold=0.7,
time_window_hours=24
)
# 2. 为每个簇生成场景名称和摘要
for cluster in clusters:
scene_name = llm.generate(
f"为以下记忆簇生成一个简洁的场景名称（不超过20字）："
f"{format_memories(cluster)}"
)
scene_summary = llm.generate(
f"用一段话（100字以内）总结这个场景的核心内容："
f"{format_memories(cluster)}"
)
# 3. 识别关键决策和开放问题
key_decisions = extract_decisions(cluster)
open_issues = extract_open_questions(cluster)
# 4. 构建图谱关系：扫描其他场景，寻找关联
existing_scenes = memory_store.get_user_scenes(user_id)
for existing_scene in existing_scenes:
if has_cross_scene_relation(cluster, existing_scene):
scene.relations.append(SceneRelation(
target_scene_id=existing_scene.id,
relation_type=infer_relation_type(cluster, existing_scene),
weight=calculate_weight(cluster, existing_scene)
))
scene.save()
return scene
举一个具体例子来理解三层之间的关系：
- L0 原文：「我上个月买了你们的 Pro 套餐，当时是年付的，一年 2999，上周开始系统特别慢，体验很差」
- L1 原子记忆：
{type: "fact", content: "用户于一个月前购买了 Pro 年付套餐", importance: 5}{type: "fact", content: "Pro 年付套餐价格为 2999 元", importance: 4}{type: "preference", content: "用户对系统性能敏感", importance: 3}{type: "intent", content: "用户在抱怨系统性能问题", importance: 5}
- L2 场景：
{name: "Pro套餐性能问题投诉", scene_type: "task", status: "active", key_decisions: ["需要技术排查"], open_issues: ["系统慢的原因是否查明"]}
3.5 L3：用户画像层——个性化认知引擎
L3 是整个架构中最「AI Native」的层级。它不存储具体的事实，而是存储用户的行为模式、偏好规律和认知特征。这些信息来自于对 L1 和 L2 数据的长期统计和归纳：
interface UserPersona {
user_id: string;
// 语言与沟通偏好
communication_style: {
formality_level: 'formal' | 'casual' | 'mixed'; // 沟通正式程度
preferred_response_length: 'brief' | 'moderate' | 'detailed';
emoji_usage: 'high' | 'medium' | 'low';
language: string;
};
// 专业背景
professional_profile: {
industry: string[]; // 行业
role: string[]; // 职位
skill_levels: Record<string, 'expert' | 'proficient' | 'basic'>;
// 如 {"python": "expert", "kubernetes": "basic"}
};
// 交互偏好
interaction_patterns: {
preferred_session_length: 'short' | 'medium' | 'long';
follow_up_rate: number; // 跟进率（0-1）
complaint_keywords: string[]; // 历史上投诉过的关键词
success_indicators: string[]; // 表示满意的信号词
};
// 决策特征
decision_patterns: {
price_sensitivity: 'low' | 'medium' | 'high';
decision_makers: string[]; // 决策影响人
typical_response_time: string;// 典型响应时间
urgency_triggers: string[]; // 触发紧急感的关键词
};
// 知识状态
knowledge_gaps: string[]; // 已知的知识盲区（Agent 应主动填补）
recently_learned: string[]; // 最近教给用户的新知识（避免重复讲解）
last_updated: number;
confidence: number; // 画像置信度（基于足够多的样本）
}
L3 的生成是异步的、增量式的——每次会话结束后，L3 Profile 会根据新增的 L1/L2 数据进行微调更新：
def update_user_persona(user_id: str, new_scene: SceneNode,
new_memories: list[AtomicMemory]) -> UserPersona:
"""增量更新用户画像"""
persona = memory_store.get_or_create_persona(user_id)
# 1. 更新专业背景（基于新增事实）
for mem in new_memories:
if mem.type == 'knowledge' and mem.importance >= 4:
update_professional_profile(persona, mem)
# 2. 更新交互模式（基于会话元数据）
if new_scene.scene_type == 'task':
persona.interaction_patterns.follow_up_rate = (
persona.interaction_patterns.follow_up_rate * 0.9
+ (1.0 if has_follow_up else 0.0) * 0.1
)
# 3. 识别知识盲区（新场景中出现但用户不了解的概念）
new_concepts = extract_technical_terms(new_scene)
for concept in new_concepts:
if not persona.has_knowledge(concept) and user_asked_about_it():
persona.knowledge_gaps.append(concept)
# 4. 计算置信度（样本越多置信度越高）
persona.confidence = min(1.0, session_count / 10)
persona.save()
return persona
四、检索引擎：如何让记忆「召之即来」
四层存储解决了「如何存」的问题，但「如何高效召回」同样是工程难点。Agent Memory 的检索引擎采用了分层检索 + 多路召回 + 重排序的经典范式：
class MemoryRetrievalEngine:
def __init__(self, memory_store: MemoryStore):
self.store = memory_store
self.llm = get_llm()
def recall(self, user_id: str, current_context: dict,
top_k: int = 20) -> RetrievalResult:
"""
分层检索入口
"""
results: list[MemoryItem] = []
# === 第一路：L3 用户画像（最高优先级，始终返回）===
persona = self.store.get_persona(user_id)
results.append(MemoryItem(
layer='L3',
data=persona,
relevance=1.0,
reason='用户画像，决定交互风格和背景知识'
))
# === 第二路：L2 场景检索（相关性最高的活跃场景）===
current_intent = current_context.get('intent', '')
scene_candidates = self.store.search_scenes(
user_id=user_id,
query=current_intent,
filters={'status': 'active'},
top_k=3
)
results.extend(scene_candidates)
# === 第三路：L1 原子记忆检索 ===
# 三种检索策略并行
tasks = [
# 策略A：向量检索（语义相似）
self._vector_search(user_id, current_context, top_k=10),
# 策略B：关键词检索（精确匹配）
self._keyword_search(user_id, current_context, top_k=10),
# 策略C：时间邻近检索（最近会话的上下文）
self._temporal_search(user_id, current_context, window_days=7)
]
l1_results = asyncio.gather(*tasks)
# === 第四路：L0 原文检索（兜底，仅在需要溯源时触发）===
# 当 L1/L2 均无满意结果时，扩大到 L0
# === 多路召回合并 + 去重 ===
merged = self._merge_and_deduplicate(l1_results)
# === LLM 重排序 ===
reranked = self._llm_rerank(
merged,
query=current_context['original_query'],
top_k=top_k
)
return RetrievalResult(
memories=reranked,
layer_distribution={m.layer: len(m) for m in groupby(reranked)},
retrieval_latency_ms=measure_time()
)
def _llm_rerank(self, candidates: list[MemoryItem],
query: str, top_k: int) -> list[MemoryItem]:
"""
使用 LLM 对候选记忆进行重排序
这是检索质量的关键一步
"""
rerank_prompt = f"""
当前用户问题：「{query}」
以下是候选记忆列表，每条附带了该记忆与问题的初步相似度：
{candidates}
请根据以下标准对候选记忆进行重排序：
1. 与当前问题的话题相关性（最重要）
2. 记忆的重要性评分（同等相关性下，高分优先）
3. 时间的接近性（近期记忆权重略高）
4. 避免重复信息（相同语义的多条记忆合并为一条）
请输出重排序后的记忆ID列表（JSON数组）和每条记忆的最终评分。
"""
reranked = self.llm.structured_output(
rerank_prompt,
schema=RerankResult.schema()
)
return [c for c in candidates
if c.id in reranked.ordered_ids[:top_k]]
五、集成实战：从零到生产的完整代码
5.1 快速集成 Agent Memory
Agent Memory 提供了多种集成方式，以下是以 OpenClaw Agent 为例的完整集成代码：
// agent-memory-integration.ts
import {
AgentMemoryClient,
MemoryConfig,
RecallOptions
} from '@tencentcloud/agent-memory';
// 1. 初始化客户端
const memoryClient = new AgentMemoryClient({
// 腾讯云认证
secretId: process.env.TENCENT_SECRET_ID!,
secretKey: process.env.TENCENT_SECRET_KEY!,
// 或者使用 API Key（简化场景）
apiKey: process.env.AGENT_MEMORY_API_KEY!,
// 配置各层存储
vectorDbConfig: {
region: 'ap-guangzhou',
instanceId: process.env.VECTORDB_INSTANCE_ID,
},
// LLM 配置（支持 OpenAI / DeepSeek / 腾讯混元）
llmProvider: 'deepseek',
llmConfig: {
model: 'deepseek-chat',
apiKey: process.env.DEEPSEEK_API_KEY,
baseUrl: 'https://api.deepseek.com',
},
});
// 2. 注册记忆工具到 Agent
const memoryTools = [
{
name: 'save_memory',
description: '保存对话中的关键信息到长期记忆',
parameters: {
type: 'object',
properties: {
memory_type: {
type: 'string',
enum: ['fact', 'preference', 'constraint', 'intent', 'knowledge'],
description: '记忆类型'
},
content: {
type: 'string',
description: '要保存的具体记忆内容'
},
importance: {
type: 'integer',
minimum: 1, maximum: 5,
description: '重要性评分（1-5）'
}
},
required: ['memory_type', 'content']
}
},
{
name: 'recall_memory',
description: '检索用户的历史记忆和偏好',
parameters: {
type: 'object',
properties: {
query: {
type: 'string',
description: '当前想了解的用户信息或上下文'
},
layers: {
type: 'array',
items: { type: 'string', enum: ['L0', 'L1', 'L2', 'L3'] },
description: '要检索的记忆层级'
}
},
required: ['query']
}
}
];
// 3. 对话流程集成
async function agentWithMemory(userMessage: string, userId: string,
sessionId: string) {
// Step 1: 检索相关记忆
const recalled = await memoryClient.recall({
userId,
query: userMessage,
layers: ['L3', 'L2', 'L1'], // 默认不查 L0（L0 是兜底）
topK: 10,
includeSourceSnippet: true, // 显示记忆来源
});
// Step 2: 构建带记忆的上下文
const memoryContext = formatMemoryForLLM(recalled);
const systemPrompt = buildSystemPrompt(memoryContext);
// Step 3: 调用 LLM（DeepSeek / GPT-4）
const response = await llm.chat({
model: 'deepseek-chat',
messages: [
{ role: 'system', content: systemPrompt },
{ role: 'user', content: userMessage }
]
});
// Step 4: 关键信息自动保存（通过 LLM 判断哪些需要记住）
await autoExtractAndSave(sessionId, userMessage, response.content);
// Step 5: 会话结束时触发场景构建（异步，不阻塞响应）
if (isSessionEnding(userMessage)) {
backgroundTaskQueue.add({
task: 'build_scene',
sessionId,
userId,
priority: 'low' // 场景构建不紧急，异步处理
});
}
return response.content;
}
function buildSystemPrompt(memoryContext: MemoryContext): string {
return `【用户记忆】
${memoryContext.persona_summary}
【相关场景】
${memoryContext.active_scenes.map(s => `• ${s.name}: ${s.summary}`).join('\n')}
【关键事实】
${memoryContext.atomic_memories.map(m => `• [${m.type}] ${m.content}`).join('\n')}
【记忆说明】
以上信息来自用户的长期记忆，请结合这些背景提供个性化回复。
如果用户询问的问题与记忆中的信息相关，请直接使用这些信息。
如果记忆中有未解决的事项或开放问题，请主动跟进。`;
}
5.2 自动记忆提取器
手动调用 save_memory 工具依赖于 Agent 主动判断哪些信息值得记住，这不够可靠。更健壮的方案是使用自动提取器：
// auto-memory-extractor.ts
class AutoMemoryExtractor {
constructor(private client: AgentMemoryClient) {}
async extractAndSave(sessionId: string, messages: Message[]) {
// 1. 调用 LLM 批量提取记忆
const extractedMemories = await this.llmExtract(messages);
// 2. 去重检查
const existingMemories = await this.client.getSessionMemories(sessionId);
const newMemories = this.deduplicate(extractedMemories, existingMemories);
// 3. 批量保存（批量 API 减少网络开销）
if (newMemories.length > 0) {
await this.client.batchSaveMemory({
sessionId,
memories: newMemories.map(m => ({
type: m.type,
content: m.content,
importance: m.importance,
source_snippet: m.source_snippet,
entities: m.entities,
ttl_days: this.inferTTL(m.type)
}))
});
console.log(`[Memory] Saved ${newMemories.length} new memories for session ${sessionId}`);
}
return newMemories;
}
private async llmExtract(messages: Message[]): Promise<ExtractedMemory[]> {
const EXTRACTION_PROMPT = `
## 任务
从以下对话记录中提取所有值得长期记忆的信息。
## 对话记录
${messages.map(m => `${m.role}: ${m.content}`).join('\n\n')}
## 提取要求
- 事实类：具体的数据、日期、事件（保留原文中的具体数值）
- 偏好类：用户表达的功能/风格/质量偏好
- 约束类：时间限制、预算限制、特殊要求
- 意图类：用户当前想完成的目标
- 知识类：用户透露的背景信息（职业、行业、技能等）
## 重要性评分标准
- 5分：直接影响任务完成或长期决策
- 4分：对交互质量有显著影响
- 3分：一般性偏好或背景
- 2分：偶然提及
- 1分：噪音
请以 JSON 数组格式输出，示例：
[{"type":"fact","content":"用户公司名为ABC科技","importance":5,"entities":[]}]
`;
const response = await this.llm.structuredOutput(
EXTRACTION_PROMPT,
schema=ExtractedMemory.schema()
);
return response;
}
private inferTTL(type: MemoryType): number {
const TTL_MAP: Record<MemoryType, number> = {
'fact': 180, // 事实保留半年
'preference': 365, // 偏好保留一年
'constraint': 30, // 约束保留一个月
'intent': 7, // 意图保留一周
'knowledge': 365, // 背景知识保留一年
'relationship': 730 // 关系信息保留两年
};
return TTL_MAP[type];
}
}
5.3 会话结束时触发 L2 场景构建
// scene-builder.ts - 后台异步任务
async function buildSceneFromSession(sessionId: string) {
const startTime = Date.now();
// 1. 获取本次会话的所有 L1 原子记忆
const atomicMemories = await memoryClient.getMemories({
sessionId,
layer: 'L1',
sortBy: 'importance',
limit: 50
});
if (atomicMemories.length < 3) {
// 对话太短，跳过场景构建
return;
}
// 2. 生成场景名称和摘要
const sceneMeta = await generateSceneMetadata(atomicMemories);
// 3. 识别关键决策和开放问题
const { decisions, issues } = await analyzeSceneContent(atomicMemories);
// 4. 构建与其他场景的关联
const relations = await findCrossSceneRelations(
atomicMemories,
await memoryClient.getUserSceneGraph(sessionId)
);
// 5. 保存 L2 场景节点
const scene = await memoryClient.createScene({
sessionId,
name: sceneMeta.name,
scene_type: inferSceneType(atomicMemories),
status: determineSceneStatus(atomicMemories),
summary: sceneMeta.summary,
atomic_memory_ids: atomicMemories.map(m => m.id),
temporal_span: {
start: atomicMemories[0].created_at,
end: atomicMemories[atomicMemories.length - 1].created_at
},
relations,
key_decisions: decisions,
open_issues: issues
});
console.log(`[Scene] Built scene "${scene.name}" with ${relations.length} cross-scene relations in ${Date.now() - startTime}ms`);
}
六、性能优化：如何在生产环境中跑稳
6.1 存储层优化
Agent Memory 在生产部署中面临的核心挑战是向量检索延迟和提取成本控制：
# docker-compose.yml 关键配置
services:
agent-memory:
environment:
# LLM 批处理配置（降低 API 调用频率）
EXTRACTION_BATCH_SIZE: "10"
EXTRACTION_INTERVAL_SECONDS: "300" # 每5分钟批量提取一次
# 向量检索配置
VECTOR_SEARCH_TOP_K: "20"
RERANK_TOP_K: "10"
# 缓存配置（热点数据加速）
L3_CACHE_TTL_SECONDS: "3600"
L2_CACHE_TTL_SECONDS: "1800"
# 并发控制
MAX_CONCURRENT_EXTRACTIONS: "5"
MAX_CONCURRENT_RECALLS: "100"
deploy:
resources:
limits:
memory: 4G
reservations:
memory: 2G
6.2 记忆过期与治理
每个 L1 原子记忆都设置了 TTL（生存时间），但 TTL 到期并不意味着直接删除。Agent Memory 实现了智能衰减 + 分级归档策略：
class MemoryGovernance:
"""
记忆治理：自动过期、归档、和重要性重评估
每24小时运行一次（定时任务）
"""
def run_daily_maintenance(self):
# 1. 检查过期记忆
expired = self.store.get_expired_memories()
# 2. 访问频次高的记忆：延长 TTL
high_frequency = [m for m in expired if m.access_count > 5]
for mem in high_frequency:
mem.ttl_days *= 2 # 热度越高，寿命越长
mem.save()
# 3. 低频记忆：降级到归档存储
low_frequency = [m for m in expired if m.access_count <= 5]
self.archive_memories(low_frequency)
# 4. 重评估场景完整性：孤立的小场景合并或归档
self.consolidate_scenes()
# 5. 画像置信度校准
self.recalibrate_persona_confidence()
def archive_memories(self, memories: list[AtomicMemory]):
"""将低频记忆归档到冷存储"""
for mem in memories:
# 归档前生成摘要（保留语义，不保留原文）
summary = self.llm.generate(
f"用一句话概括以下记忆的核心内容（不超过30字）：{mem.content}"
)
self.cold_store.insert({
'original_id': mem.id,
'summary': summary,
'type': mem.type,
'archived_at': datetime.now(),
'original_content': mem.content # 可选：保留原文以供审计
})
self.store.delete(mem.id)
6.3 隐私合规：记忆的「遗忘权」
GDPR 和国内个人信息保护法都要求用户有权删除个人数据。Agent Memory 在架构层面支持完全删除：
# 隐私合规：用户数据删除
async def delete_user_all_memory(user_id: str, deletion_type: str = 'full'):
"""
支持多种删除级别：
- 'full': 删除所有层级的所有记忆
- 'l1_only': 只删除 L1/L2/L3，保留 L0（审计用）
- 'recent': 删除最近30天的记忆，保留历史
"""
if deletion_type == 'full':
# 逐层删除（L0 保留审计记录，metadata 清除）
self.store.delete_l3_profile(user_id)
self.store.delete_l2_scenes(user_id)
self.store.delete_l1_memories(user_id)
self.store.anonymize_l0_messages(user_id) # 保留结构，清除可识别信息
elif deletion_type == 'l1_only':
self.store.delete_l1_memories(user_id)
self.store.delete_l2_scenes(user_id)
self.store.delete_l3_profile(user_id)
# 后续 L0 可继续提炼，但无用户标识
# 记录删除审计日志（必须保留）
self.audit_log.record(
action='memory_deletion',
user_id=user_id,
deletion_type=deletion_type,
timestamp=datetime.now(),
performed_by='user_request' # 或 'gdpr_request', 'admin_action'
)
七、与其他方案的横向对比
| 维度 | Agent Memory | 纯向量数据库 | 全上下文塞窗口 | 外部知识库 |
|---|---|---|---|---|
| 记忆持久性 | ✅ 跨会话 | ✅ 跨会话 | ❌ 单会话 | ✅ 跨会话 |
| 记忆提炼能力 | ✅ 四层渐进 | ❌ 原始块 | ❌ 原始全文 | ❌ 原始块 |
| 个性化支持 | ✅ L3画像 | ❌ 无 | ❌ 无 | ❌ 无 |
| 时序感知 | ✅ 完整 | ⚠️ 弱 | ⚠️ 弱 | ⚠️ 弱 |
| 上下文窗口占用 | ✅ 低（提炼后） | ✅ 中 | ❌ 高 | ✅ 中 |
| 检索延迟 | ⚠️ 中（多路召回） | ✅ 低 | N/A | ✅ 低 |
| 运维复杂度 | ⚠️ 高 | ✅ 低 | ✅ 极低 | ✅ 低 |
| 隐私合规 | ✅ 原生支持 | ⚠️ 需额外处理 | ✅ 天然合规 | ⚠️ 需额外处理 |
八、实战建议：什么时候用 Agent Memory
Agent Memory 并不是万能解。以下场景适合引入 Agent Memory：
- 多轮复杂任务（客服对话、项目管理、数据分析）：每次会话跨度长、涉及多个决策点
- 高价值用户（VIP客户、专业用户）：个性化服务带来的收益大于记忆系统运维成本
- 知识密集型场景（法律咨询、医疗问诊）：用户的背景信息直接影响服务质量
- 跨 Agent 协作：多个 Agent 需要共享同一个用户的上下文（多 Agent 协作场景）
以下场景不适合引入 Agent Memory：
- 简单问答型 Agent（FAQ 机器人）：每次问题独立，记忆投入产出比太低
- 高频短交互（搜索辅助、翻译工具）：用户期望的是即时响应，记忆引入的延迟不可接受
- 数据敏感场景（金融交易、医疗记录）：合规审计复杂度急剧上升
- 初创期 MVP：先跑通核心价值，记忆系统是规模化的优化而非初期必须
九、总结与展望
腾讯 Agent Memory 的四层渐进式架构给 AI Agent 的记忆问题提供了一个系统性的工程解法。它没有试图用单一技术解决所有问题，而是通过数据湖（ L0）→ 知识提炼（L1）→ 图谱关联（L2）→ 画像抽象（L3） 的分层设计，让每层做最擅长的事：
- L0 解决可审计性：保留完整原文，满足合规要求
- L1 解决信息压缩：将噪音过滤，提取真正值得记住的事实
- L2 解决结构化：将碎片记忆组织成可理解的场景
- L3 解决个性化：从历史数据中归纳用户的行为模式
从工程角度看，这个架构最值得借鉴的设计理念是**「渐进式提炼 + 按需检索」**——不需要在每次对话时处理用户的全部历史，只在需要时精准召回最相关的那部分记忆。这与人类大脑的工作方式高度一致：我们不是记住了所有经历，而是记住了最重要的模式。
GitHub: https://github.com/TencentCloud/TencentDB-Agent-Memory
Star: 18.1k | Language: TypeScript | License: Apache 2.0
本文参考资料：GitHub 官方仓库文档、CSDN 技术博客、腾讯云官方技术解读。所有架构分析和代码示例均为基于公开信息的二次创作。