ChatGPT Memory 进阶实战：个人 vs 团队 / 隐私 GDPR / 评测指标

前 3 篇 Memory 文章覆盖了『是什么 / 怎么架构 / 怎么用』。本文是『Memory 在 production 的最后一块拼图』：个人 vs 团队差异、合规边界、评测指标、版本控制与删除。

1. 个人 vs 团队 Memory

个人 Memory 与 团队 Memory 本质不同：

三种共享模式：

1. 严格分离 — 个人 Memory 只 user_id 维度；团队 Memory 走 team_id + user_id 复合 key。默认配置。
2. 共享池 — 团队 Memory 所有人可见，但每条 memory 带创建者 user_id（可追溯）。适合文档协作 / 共享笔记。
3. 渐进共享 — 默认个人 Memory，标记 shareable=true 的 memory 才进团队池。折中方案。

推荐：模式 1（默认严格分离）+ 模式 3（重要 memory 标记后共享）。

class MemoryType(Enum):
    PERSONAL = "personal"
    TEAM_SHARED = "team_shared"

class Memory:
    id: int
    user_id: str                    # always set
    team_id: str | None             # set only for TEAM_SHARED
    shareable: bool = False
    memory_text: str
    embedding: vector(3072)
    confidence: float
    version: int = 1
    deleted: bool = False
    created_at: timestamp

# Query: personal memory
memories = db.query("""
    SELECT * FROM memories
    WHERE user_id = :user_id AND team_id IS NULL AND deleted = false
    ORDER BY version DESC LIMIT 10
""", user_id="alice")

# Query: team memory (including personal memories shared to this team)
memories = db.query("""
    SELECT * FROM memories
    WHERE (user_id = :user_id OR team_id = :team_id)
      AND deleted = false
    ORDER BY version DESC LIMIT 10
""", user_id="alice", team_id="team_42")

2. 隐私与合规

GDPR 第 17 条（被遗忘权）

要求：用户可以要求删除其所有个人数据，组织必须：

• 立即删除（30 天内内）
• 包括所有副本 + 缓存 + backup
• 提供删除证明

Memory 系统实现：

def gdpr_delete(user_id: str) -> dict:
    """GDPR Article 17 deletion - cascade across all stores"""
    start = time.time()

    # 1. Delete from vector store
    embedding_ids = db.query("SELECT id FROM memories WHERE user_id = ?", user_id)
    vector_store.delete(ids=[m.id for m in embedding_ids])

    # 2. Delete from primary DB
    count = db.execute("""
        DELETE FROM memories
        WHERE user_id = ?
          AND (deleted = false OR deleted = true)
    """, user_id).rowcount

    # 3. Delete from caches
    cache.delete(f"user:{user_id}:memory_top_k")
    cache.delete(f"user:{user_id}:memory_summary")

    # 4. Mark audit log (keep for compliance per EU AI Act)
    audit_log.record(
        event="gdpr_delete",
        user_id=user_id,
        deleted_count=count,
        timestamp=datetime.utcnow(),
    )

    # 5. Notify user
    send_email(user_id, subject="Your memory data has been deleted", body=...)

    return {
        "status": "deleted",
        "user_id": user_id,
        "deleted_count": count,
        "duration_ms": (time.time() - start) * 1000,
    }

CCPA（加州消费者隐私法）

类似 GDPR，但有 opt-out 概念（用户可以选择不让数据出售）。Memory 系统需要支持：

• 用户导出所有 Memory 数据（JSON 格式）
• 用户标记 Memory 为『不出售』

欧盟 AI Act

对 AI 系统的 Memory / 训练数据审计要求：

• 高风险类别（医疗 / 金融 / 教育）必须审计训练数据来源
• 用户有权知道『AI 为什么这么回答』——可能涉及 Memory 召回链路
• AI Act 第 13 条：训练数据要可追溯

3. Memory 评测指标

生产环境 Memory 必须监控 3 个核心指标：

def evaluate_memory_precision(top_k_memories, ground_truth_useful):
    """LLM-as-judge evaluates each recalled memory"""
    useful = 0
    for mem in top_k_memories:
        judge = judge_memory_useful(mem, ground_truth_useful)
        if judge["useful"]:
            useful += 1
    return useful / len(top_k_memories)

def evaluate_memory_recall(top_k_memories, all_useful_memories):
    """How many of all useful memories were recalled"""
    top_k_ids = {m.id for m in top_k_memories}
    useful_ids = {m.id for m in all_useful_memories}
    return len(top_k_ids & useful_ids) / len(useful_ids)

def evaluate_memory_drift(user_id, k=100):
    """Drift = memory lost recall quality over 30 days"""
    # Sample 100 memories that user_id wrote 30+ days ago
    old_memories = db.query("""
        SELECT * FROM memories
        WHERE user_id = ? AND created_at < now() - interval '30 days'
        LIMIT ?
    """, user_id, k)

    drift_count = 0
    for mem in old_memories:
        # Re-query with current embedding model and check if still top-K for same query
        # (you need original query — keep it in memory metadata)
        still_recalled = recall_test(mem.text, mem.metadata.original_query)
        if not still_recalled:
            drift_count += 1

    return drift_count / len(old_memories)

每月跑一次评测，输出 dashboard：

Month | precision | recall | drift
2026-08 | 0.86 | 0.74 | 7%
2026-09 | 0.84 | 0.72 | 9%   ← drift 接近阈值
2026-10 | 0.81 | 0.70 | 12% ← drift 告警！需要查索引

4. Memory 版本控制

每次写 memory 都打版本戳，方便回滚：

CREATE TABLE memories (
  id BIGSERIAL PRIMARY KEY,
  user_id TEXT NOT NULL,
  memory_text TEXT NOT NULL,
  embedding vector(3072),
  memory_type TEXT,  -- 'personal' / 'team_shared'
  confidence FLOAT,
  version INTEGER DEFAULT 1,
  deleted BOOLEAN DEFAULT false,
  created_at TIMESTAMPTZ DEFAULT now(),
  updated_at TIMESTAMPTZ DEFAULT now()
);

-- 索引（按 user_id + version 排序）
CREATE INDEX ON memories (user_id, version DESC);
CREATE INDEX ON memories (user_id, deleted, version DESC);

写时：

def write_memory(user_id, memory_text, memory_type="personal"):
    """Soft-delete old version + insert new version"""
    with db.transaction():
        # 1. Mark existing memory as superseded
        db.execute("""
            UPDATE memories
            SET deleted = true, updated_at = now()
            WHERE user_id = ? AND memory_text_hash = ? AND deleted = false
        """, user_id, hash(memory_text))

        # 2. Insert new version
        new_version = db.query("""
            SELECT COALESCE(MAX(version), 0) + 1
            FROM memories WHERE user_id = ?
        """, user_id)[0]

        embedding = embed(memory_text)
        db.execute("""
            INSERT INTO memories (user_id, memory_text, embedding, memory_type, version)
            VALUES (?, ?, ?, ?, ?)
        """, user_id, memory_text, embedding, memory_type, new_version)

回滚：

def rollback_memory(memory_id):
    """Mark current as deleted + restore previous version"""
    with db.transaction():
        mem = db.query("SELECT * FROM memories WHERE id = ?", memory_id)
        # Find all versions with same content hash, keep the highest non-deleted version
        previous_version = db.query("""
            SELECT * FROM memories
            WHERE user_id = ? AND memory_text_hash = ?
              AND version < ? AND deleted = false
            ORDER BY version DESC LIMIT 1
        """, mem.user_id, mem.text_hash, mem.version)

        if previous_version:
            db.execute("UPDATE memories SET deleted = false WHERE id = ?", previous_version.id)

5. Memory 删除机制

用户主动撤回（UI Settings → Memory → Delete All）

def user_delete_memory(user_id):
    """User-initiated deletion via ChatGPT UI"""
    # Delete all memories for user
    db.execute("DELETE FROM memories WHERE user_id = ?", user_id)
    vector_store.delete(where={"user_id": user_id})
    cache.delete(f"user:{user_id}:memory_*")
    return {"status": "deleted", "user_id": user_id}

GDPR 请求（API / 邮件）

def gdpr_delete_request(user_id, request_id):
    """GDPR Article 17 - process formal deletion request"""
    # Audit log
    audit_log.record(
        event="gdpr_request_received",
        user_id=user_id,
        request_id=request_id,
    )

    # 30-day deadline (typically complete in 1 day)
    result = gdpr_delete(user_id)

    audit_log.record(
        event="gdpr_request_completed",
        user_id=user_id,
        request_id=request_id,
        result=result,
    )

    return result

自动过期（Retention）

RETENTION_RULES = {
    "medical": 7 * 365,    # 7 years
    "financial": 5 * 365,  # 5 years
    "general": 2 * 365,    # 2 years
    "eu_user": 0,          # immediate (GDPR)
}

def retention_sweep():
    """Daily cron to mark expired memories as deleted"""
    for category, days in RETENTION_RULES.items():
        if days == 0:
            continue  # eu_user handled via gdpr_delete

        cutoff = datetime.utcnow() - timedelta(days=days)
        expired = db.query("""
            UPDATE memories
            SET deleted = true, scheduled_at = now()
            WHERE category = ? AND created_at < ? AND deleted = false
            RETURNING id
        """, category, cutoff)

        audit_log.record(
            event="retention_expired",
            category=category,
            expired_count=len(expired),
        )

6. Memory 召回 LLM 摘要 vs 全量

Memory 摘要用 LLM 的 trade-off：

推荐：起步用 LLM 摘要（省 token）；用户报告『AI 记错了』时切混合模式；高敏感场景（医疗 / 法律）直接用全量原始。

常见问题

1. 团队 Memory 和个人 Memory 怎么共存？

三种模式：(1) 严格分离——个人 Memory 只 user_id 维度，团队 Memory 走 team_id + user_id 复合 key；(2) 共享池——团队 Memory 所有人可见，但每条 memory 带创建者 user_id；(3) 渐进共享——默认个人，标记 shareable=true 的 memory 才进团队池。推荐 (1) + (3)：默认严格分离，关键信息标记后共享。

2. GDPR 被遗忘权怎么处理？

三步：(1) 收到用户删除请求（API 或 UI）→ 立即触发；(2) 后端 bulk delete 该 user_id 的所有 memory（含个人 + 团队引用 + 关联 embedding）；(3) 删除 audit log（除合规要求保留外）+ 通知用户删除完成。技术实现：DELETE FROM memories WHERE user_id = ? + DELETE FROM memories WHERE json_contains(shared_with, ?) AND user_id != ?。注意：所有相关的缓存 / 副本都要清。

3. Memory precision 怎么测？

三步：(1) 准备 100+ 测试 query（含 happy + 边界）；(2) 跑 query → 召回 top-K memory；(3) 人工或 LLM-as-judge 评估每条召回是否真的有用（是 → 1，否 → 0）。precision = sum(有用) / K。LLM-as-judge 评 precision 与人工 correlation 通常 > 0.85。目标 > 0.8。

4. Memory drift 怎么监控？

指标：同一 user_id 在 T0 写入一条 memory，T+30 天召回同样的 query，检查召回的这条 memory 是否还在 top-K。drift = 1 - (T+30 还在 top-K 比例)。目标 < 10%/月。drift 高说明：(1) vector store 索引问题；(2) memory 写入太多导致新 memory 抢占；(3) 时间衰减逻辑没生效。建议：每月抽样 100 条已存在 memory 检查召回质量。

5. Memory 一定要用 LLM 摘要吗？

不必。Memory 摘要用 LLM 的目的是压缩 token + 提取结构化字段，但：(1) 摘要增加 latency + 5-10% 成本；(2) 摘要可能丢失细节；(3) 摘要 LLM 偶尔幻觉。生产可选模式：(1) 全量存原始消息（token 贵但完整）；(2) 全量存 LLM 摘要（token 便宜但细节丢）；(3) 混合——重要消息存原始，闲聊存摘要；(4) 用户可切换。建议：起步用 LLM 摘要（省 token），数据敏感时切全量。

下一步

• 想了解 Memory 架构？读 《ChatGPT Memory 架构解析：长上下文、向量检索与跨会话持久化》。
• 想了解 Memory 完全使用手册？读 《ChatGPT Memory 完全使用手册》。
• 想了解 Memory 高价值场景？读 《ChatGPT Memory 12 个高价值使用场景：从技术栈到家庭信息》。