腾讯开源 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 是原始数据湖，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

Agent Memory 并不是万能解。以下场景适合引入 Agent Memory：

1. 多轮复杂任务（客服对话、项目管理、数据分析）：每次会话跨度长、涉及多个决策点
2. 高价值用户（VIP客户、专业用户）：个性化服务带来的收益大于记忆系统运维成本
3. 知识密集型场景（法律咨询、医疗问诊）：用户的背景信息直接影响服务质量
4. 跨 Agent 协作：多个 Agent 需要共享同一个用户的上下文（多 Agent 协作场景）

以下场景不适合引入 Agent Memory：

1. 简单问答型 Agent（FAQ 机器人）：每次问题独立，记忆投入产出比太低
2. 高频短交互（搜索辅助、翻译工具）：用户期望的是即时响应，记忆引入的延迟不可接受
3. 数据敏感场景（金融交易、医疗记录）：合规审计复杂度急剧上升
4. 初创期 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 技术博客、腾讯云官方技术解读。所有架构分析和代码示例均为基于公开信息的二次创作。