第 8 章：记忆架构

记忆系统让 Agent 从"每次都是陌生人"变成"越用越懂你"。但别指望它记住一切——向量相似度不是精确匹配，召回率和准确率需要取舍。

你让 Agent 帮你研究了一周的技术方案。

每天都在聊，讨论了几十个细节。终于选定了方案。

一周后你回来问："上次我们选的方案是哪个？"

它回答："抱歉，我不知道你在说什么。"

你傻眼了。

一周的讨论，全忘了？这不是在逗我吗？

这不是 Agent 的问题，是没有记忆系统。

上一章讲的上下文窗口管理，解决的是"单次对话内"的记忆。但用户关了浏览器、换了设备、过了几天再来，那些上下文就没了。

真实场景中，用户会：

• 中断后继续昨天的话题
• 希望 Agent 记住自己的偏好
• 不想每次都重复解释背景

没有记忆系统的 Agent 就像失忆症患者——每次见面都要重新自我介绍。

8.1 记忆的类型

不是所有记忆都一样。我把它分成四类：

工作记忆和会话记忆，上一章的上下文窗口管理已经覆盖了。

这章重点讲长期记忆和语义记忆——如何让 Agent 跨会话记住东西。

8.2 为什么传统数据库不够？

你可能会想：直接用 MySQL/PostgreSQL 存历史记录不就行了？

问题是：用户不会精确查询。

用户问"React 组件怎么优化？"，你存的可能是"前端渲染性能提升方案"。

字符串完全不同，但语义相关。

-- 传统查询：找不到
SELECT * FROM memories WHERE content LIKE '%React组件优化%';

-- 用户真正需要的：语义相关的历史
-- "前端渲染性能" ≈ "React组件优化"

这就是向量数据库的价值：按语义相似度检索，而不是精确匹配。

8.3 存储层设计

Shannon 的记忆系统用了三种存储，各司其职：

为什么是三种？

PostgreSQL：需要 ACID 事务的数据。

• 执行历史：哪个 Agent 做了什么、成功还是失败
• 审计日志：谁在什么时候做了什么操作
• 用户偏好：显式保存的配置

Redis：热数据缓存，毫秒级访问。

• 活跃会话：减少数据库查询
• Token 预算：实时追踪，需要原子操作
• 速率限制：高频读写

Qdrant：向量相似度搜索。

• 语义记忆：找"意思相近"的历史
• 压缩摘要：上一章讲的摘要存储
• 文档分块：RAG 场景的知识库

为什么用 Qdrant？

向量数据库有很多选择：Pinecone、Milvus、Weaviate、Qdrant...

Shannon 选了 Qdrant，主要因为：

• 开源自托管：不依赖 SaaS，数据在自己手里
• 性能好：Rust 实现，单机百万级向量
• API 简洁：学习成本低

但这不是唯一选择。如果你用 Pinecone 或 Milvus，核心概念是一样的。

另一种思路：本地文件存储

上面讲的都是服务器端存储。但 2025 年以来，另一种模式在开发者工具圈越来越流行：直接用本地文件做持久化。

代表案例：

• Claude Code：用 CLAUDE.md 文件存项目级记忆，用 ~/.claude/ 目录存用户级偏好
• Cursor：用 .cursor/ 目录存项目上下文和规则
• Windsurf：类似的本地文件方案

这种方式的优势很直接：

• 零部署：不需要数据库服务
• 版本控制友好：可以直接 git commit
• 用户可读可编辑：Markdown 文件，人类友好
• 隐私：数据不离开本地

但也有明显局限：

• 不适合多设备同步：除非用云存储
• 语义检索弱：文件内容只能关键词匹配，没有向量相似度
• 规模受限：文件太大会影响读取速度

Shannon 选择了服务器端存储方案，因为目标场景是多租户、多设备、需要语义检索的企业级应用。

但如果你做的是单机开发者工具，本地文件方案值得考虑——简单、透明、用户可控。

8.4 语义检索怎么做？

核心流程：

用户问题: "React 组件怎么优化？"
            │
            ▼
    生成问题的 Embedding 向量
            │
            ▼
    在 Qdrant 中搜索相似向量
            │
            ▼
    返回: "之前讨论过的前端性能优化方案..."

Embedding 是什么？

把文本变成一串数字（向量）。语义相近的文本，向量也相近。

# 两个语义相近的句子
text1 = "React 组件性能优化"
text2 = "前端渲染效率提升"

vec1 = embedding_model.encode(text1)  # [0.12, 0.45, -0.23, ...]
vec2 = embedding_model.encode(text2)  # [0.11, 0.43, -0.21, ...]

# 计算余弦相似度
similarity = cosine_sim(vec1, vec2)  # 0.92 - 很相似

向量维度通常是 768 或 1536。OpenAI 的 text-embedding-3-small 是 1536 维。

Embedding 模型选择

⚠️ 时效性提示 (2026-01): 模型定价频繁变化，请查阅各厂商官网获取最新价格。

Shannon 默认用 text-embedding-3-small，平衡成本和效果。

相似度阈值

检索时要设阈值，太高找不到，太低返回垃圾：

// 太高，几乎匹配不到
Threshold: 0.95

// 太低，返回无关内容
Threshold: 0.3

// 实测比较平衡
Threshold: 0.7

我测试过，0.7 是个不错的起点。具体场景可以调：

• 精确场景（代码搜索）：调高到 0.8
• 宽泛场景（创意探索）：调低到 0.6

8.5 分层记忆检索

单纯的语义检索还不够。实际中需要融合多种策略：

用户问: "继续上次的讨论"
              │
              ├── Recent (最近 5 条)
              │   └── 保持对话连贯
              │
              ├── Semantic (语义相关 3 条)
              │   └── 找历史中相关的
              │
              └── Summary (摘要 2 条)
                  └── 快速了解长期上下文

为什么要分层？

• Recent：用户说"刚才那个"，需要最近的对话
• Semantic：用户问相关话题，需要历史中相关的
• Summary：长对话的压缩摘要，快速建立上下文

三层融合，去重后返回。

实现参考

Shannon 的 FetchHierarchicalMemory 函数：

func FetchHierarchicalMemory(ctx context.Context, in Input) (Result, error) {
    result := Result{Items: []Item{}, Sources: map[string]int{}}
    seen := make(map[string]bool)  // 去重用

    // 1. 时间维度：最近 N 条
    if in.RecentTopK > 0 {
        recent := FetchSessionMemory(ctx, in.SessionID, in.RecentTopK)
        for _, item := range recent {
            item["_source"] = "recent"  // 标记来源
            result.Items = append(result.Items, item)
            seen[item.ID] = true
        }
    }

    // 2. 语义维度：相关 N 条
    if in.SemanticTopK > 0 && in.Query != "" {
        semantic := FetchSemanticMemory(ctx, in.Query, in.SemanticTopK)
        for _, item := range semantic {
            if !seen[item.ID] {  // 去重
                item["_source"] = "semantic"
                result.Items = append(result.Items, item)
                seen[item.ID] = true
            }
        }
    }

    // 3. 压缩维度：历史摘要
    if in.SummaryTopK > 0 {
        summaries := FetchSummaries(ctx, in.Query, in.SummaryTopK)
        for _, item := range summaries {
            item["_source"] = "summary"
            result.Items = append(result.Items, item)
        }
    }

    // 4. 限制总数，防止上下文爆炸
    maxTotal := 10
    if len(result.Items) > maxTotal {
        result.Items = result.Items[:maxTotal]
    }

    return result, nil
}

_source 标记很重要。后续处理时，可以根据来源决定优先级：

• Recent 来源的信息，用户明确提到时优先使用
• Semantic 来源的信息，作为背景参考
• Summary 来源的信息，帮助建立长期上下文

8.6 记忆存储：去重和分块

存记忆不是简单的"把所有东西都存进去"。有三个关键问题：

问题 1：重复内容

用户问了类似的问题，不应该存多份。

// 检查是否重复（95% 相似度阈值）
const duplicateThreshold = 0.95

similar, _ := vdb.Search(ctx, queryEmbedding, 1, duplicateThreshold)
if len(similar) > 0 && similar[0].Score > duplicateThreshold {
    // 跳过，已经有类似的了
    return
}

问题 2：长回答分块

一个回答可能很长。直接存一个大向量，检索效果不好——因为长文本的 embedding 会"稀释"语义。

需要分块存储，检索时再聚合：

// 长文本分块存储
if len(answer) > chunkThreshold {  // 比如 2000 tokens
    chunks := chunker.ChunkText(answer, ChunkConfig{
        MaxTokens:    2000,
        OverlapTokens: 200,  // 块之间有重叠，保持上下文
    })

    // 批量生成 embedding（一次 API 调用，省钱）
    embeddings := svc.GenerateBatchEmbeddings(ctx, chunks)

    // 每个块单独存储，但共享 qa_id
    qaID := uuid.New().String()
    for i, chunk := range chunks {
        payload := map[string]interface{}{
            "query":       query,
            "chunk_text":  chunk.Text,
            "qa_id":       qaID,           // 用于聚合
            "chunk_index": i,
            "chunk_count": len(chunks),
        }
        vdb.Upsert(ctx, embeddings[i], payload)
    }
}

检索时，根据 qa_id 把同一个回答的块聚合回来：

// 检索后聚合
results := vdb.Search(ctx, queryVec, 10, 0.7)

// 按 qa_id 分组
grouped := make(map[string][]Result)
for _, r := range results {
    qaID := r.Payload["qa_id"].(string)
    grouped[qaID] = append(grouped[qaID], r)
}

// 重建完整回答
for qaID, chunks := range grouped {
    // 按 chunk_index 排序
    sort.Slice(chunks, func(i, j int) bool {
        return chunks[i].Payload["chunk_index"].(int) < chunks[j].Payload["chunk_index"].(int)
    })
    // 拼接
    fullAnswer := ""
    for _, chunk := range chunks {
        fullAnswer += chunk.Payload["chunk_text"].(string)
    }
}

问题 3：低价值内容

不是所有内容都值得存：

// 跳过低价值内容
if len(answer) < 50 {
    return  // 太短，没信息量
}

if containsError(answer) {
    return  // 错误消息，不要污染记忆
}

if isSmallTalk(query) {
    return  // 闲聊，没有记忆价值
}

8.7 Agent 级别隔离

多 Agent 场景下，每个 Agent 可以有自己的记忆空间。

type FetchAgentMemoryInput struct {
    SessionID string
    AgentID   string  // 关键：Agent 级别隔离
    TopK      int
}

func FetchAgentMemory(ctx context.Context, in Input) (Result, error) {
    // 按 session_id + agent_id 过滤
    filter := map[string]interface{}{
        "session_id": in.SessionID,
        "agent_id":   in.AgentID,
    }
    items := vdb.SearchWithFilter(ctx, queryVec, filter, in.TopK)
    return Result{Items: items}, nil
}

场景：研究 Agent 和代码 Agent 各有自己的记忆，互不干扰。

研究 Agent 记住的是"用户对竞品的看法"，代码 Agent 记住的是"用户的编码偏好"。

如果混在一起，代码 Agent 可能会在写代码时提起竞品分析——这很奇怪。

8.8 策略学习（高级）

更高级的记忆不只是存"问答对"，还能学习"什么方法有效"。

分解模式记忆

记住成功的任务分解方式：

type DecompositionMemory struct {
    QueryPattern string    // "优化 API 性能"
    Subtasks     []string  // 分解成的子任务
    Strategy     string    // "parallel" or "sequential"
    SuccessRate  float64   // 这个分解方式的成功率
    UsageCount   int       // 被使用了多少次
}

下次遇到类似任务，可以复用成功的分解模式：

func (advisor *DecompositionAdvisor) Suggest(query string) Suggestion {
    // 找历史中类似的成功分解
    for _, prev := range advisor.Memory.DecompositionHistory {
        if similar(query, prev.QueryPattern) > 0.8 && prev.SuccessRate > 0.9 {
            return Suggestion{
                Subtasks:   prev.Subtasks,
                Strategy:   prev.Strategy,
                Confidence: prev.SuccessRate,
                Reason:     "Based on similar successful task",
            }
        }
    }
    return Suggestion{Confidence: 0}  // 没找到可复用的
}

失败模式识别

记住失败的模式，避免重蹈覆辙：

type FailurePattern struct {
    Pattern    string   // "rate_limit"
    Indicators []string // ["quickly", "urgent", "asap"]
    Mitigation string   // "考虑串行执行避免限流"
    Severity   int      // 1-5
}

func (advisor *DecompositionAdvisor) CheckRisks(query string) []Warning {
    warnings := []Warning{}
    for _, pattern := range advisor.Memory.FailurePatterns {
        if matches(query, pattern.Indicators) {
            warnings = append(warnings, Warning{
                Pattern:    pattern.Pattern,
                Mitigation: pattern.Mitigation,
                Severity:   pattern.Severity,
            })
        }
    }
    return warnings
}

这样 Agent 会越用越聪明——记住哪些方法有效，避免重复失败。

8.9 性能优化

MMR 多样性

纯相似度检索可能返回一堆重复的内容。用 MMR（Maximal Marginal Relevance）平衡相关性和多样性：

MMR = lambda * 相关性 - (1-lambda) * 与已选结果的最大相似度

lambda = 0.7: 偏向相关性（默认）
lambda = 0.5: 平衡
lambda = 0.3: 偏向多样性

实现：

// 获取 3 倍候选，再用 MMR 重排
poolSize := topK * 3
candidates := vdb.Search(ctx, vec, poolSize)
results := applyMMR(candidates, vec, topK, 0.7)  // lambda=0.7

效果：返回的结果既相关，又不重复。

批量 Embedding

分块存储时，一次 API 调用处理所有块：

// 差：N 个块 → N 次 API 调用
for _, chunk := range chunks {
    embedding := svc.GenerateEmbedding(ctx, chunk.Text)  // 慢
}

// 好：N 个块 → 1 次 API 调用
embeddings := svc.GenerateBatchEmbeddings(ctx, chunkTexts)  // 快 5x

性能提升：5x 更快，成本不变。

Payload 索引

在 Qdrant 中，对常用过滤字段建索引：

indexFields := []string{
    "session_id",
    "tenant_id",
    "agent_id",
    "timestamp",
}
for _, field := range indexFields {
    vdb.CreatePayloadIndex(ctx, collection, field, "keyword")
}

过滤性能提升 50-90%。

Embedding 缓存

同样的文本不要重复计算 embedding：

type EmbeddingCache struct {
    lru   *lru.Cache  // 内存 LRU，2048 条
    redis *redis.Client  // Redis 持久化
}

func (c *EmbeddingCache) Get(text string) ([]float32, bool) {
    key := hash(text)
    // 先查内存
    if vec, ok := c.lru.Get(key); ok {
        return vec.([]float32), true
    }
    // 再查 Redis
    if vec, err := c.redis.Get(ctx, key).Bytes(); err == nil {
        c.lru.Add(key, vec)  // 回填内存
        return vec, true
    }
    return nil, false
}

8.10 隐私保护

记忆系统存的是用户数据，隐私很重要。

PII 脱敏

存储前自动检测并移除敏感信息：

func redactPII(text string) string {
    // Email
    text = emailRe.ReplaceAllString(text, "[REDACTED_EMAIL]")
    // 电话
    text = phoneRe.ReplaceAllString(text, "[REDACTED_PHONE]")
    // 信用卡
    text = ccRe.ReplaceAllString(text, "[REDACTED_CC]")
    // SSN
    text = ssnRe.ReplaceAllString(text, "[REDACTED_SSN]")
    // IP 地址
    text = ipRe.ReplaceAllString(text, "[REDACTED_IP]")
    // API Key
    text = apiKeyRe.ReplaceAllString(text, "[REDACTED_API_KEY]")
    return text
}

Shannon 在压缩摘要和存储记忆前都会调用这个函数。

数据保留策略

租户隔离

多租户场景下，不同租户的记忆绝对不能互相访问：

func (m *Manager) GetSession(ctx context.Context, sessionID string) (*Session, error) {
    session := m.loadFromCache(sessionID)

    // 租户隔离检查
    userCtx := authFromContext(ctx)
    if userCtx.TenantID != "" && session.TenantID != userCtx.TenantID {
        // 不泄露 Session 存在信息
        return nil, ErrSessionNotFound  // 不是 ErrUnauthorized
    }

    return session, nil
}

注意返回的是 ErrSessionNotFound 而不是 ErrUnauthorized——不泄露 Session 是否存在。

8.11 常见的坑

坑 1：Embedding 服务没配置

记忆功能静默降级，Agent 表现得"健忘"。

# 必须配置
OPENAI_API_KEY=sk-...

没有 OpenAI key，Shannon 的记忆功能会静默降级：不报错，但也不工作。

坑 2：相似度阈值不对

太高找不到，太低返回垃圾。从 0.7 开始调。

坑 3：不去重

相同内容重复出现，浪费上下文。用 qa_id 和相似度阈值去重。

坑 4：存低价值内容

错误消息、短回复污染记忆。自动跳过。

坑 5：忽略 _source 标记

Recent/Semantic/Summary 来源不同，优先级不同。要区分处理。

坑 6：分块太大或太小

// 太大：语义被稀释，检索效果差
ChunkConfig{MaxTokens: 8000}

// 太小：丢失上下文，碎片化
ChunkConfig{MaxTokens: 200}

// 推荐
ChunkConfig{MaxTokens: 2000, OverlapTokens: 200}

Shannon Lab（10 分钟上手）

本节帮你在 10 分钟内把本章概念对应到 Shannon 源码。

必读（1 个文件）

• docs/memory-system-architecture.md：重点看 "Storage Layers"、"Memory Types"、"Advanced Chunking System" 部分，理解三层存储和分块机制

选读深挖（2 个，按兴趣挑）

• activities/semantic_memory.go：看 FetchHierarchicalMemory 函数，理解分层检索逻辑和去重机制
• vectordb/client.go：看 Search、Upsert 方法，理解向量数据库的基本操作

练习

练习 1：设计记忆策略

你在做一个客服 Agent，需要记住：

• 用户的历史工单
• 用户的产品偏好
• 常见问题的解决方案

设计你的记忆架构：

• 用什么存储？
• 怎么检索？
• 怎么去重？

练习 2：源码阅读

读 semantic_memory.go，回答：

1. _source 字段有哪些可能的值？
2. 如果 Recent 和 Semantic 返回了同一条记录，会发生什么？
3. 为什么要限制总数为 10？

练习 3（进阶）：实现 MMR

手写一个 MMR 重排序函数：

func applyMMR(candidates []SearchResult, queryVec []float32, topK int, lambda float64) []SearchResult {
    // 你的实现
}

思考：

• 时间复杂度是多少？
• 能优化吗？

划重点

核心就一句话：记忆系统让 Agent 从"无状态"变成"有经验"。

要点：

1. 三层存储：PostgreSQL + Redis + Qdrant，各司其职
2. 语义检索：Embedding + 向量相似度，找"意思相近"的历史
3. 分层融合：Recent + Semantic + Summary，三层去重合并
4. 智能去重：95% 相似度阈值 + qa_id
5. 策略学习：记住成功模式，识别失败模式

但别过度期望：向量相似度不是精确匹配，召回率和准确率需要权衡。有时候"找不到"可能是阈值问题，有时候"找到的不对"可能是 embedding 模型的局限。

记忆系统解决了信息存储和检索。但多轮对话中，还有会话管理、状态追踪、隐私保护等问题。

下一章我们来聊多轮对话设计——如何在连续交互中保持连贯性。

下一章见。

延伸阅读

• Qdrant Vector Database - Qdrant 官方文档
• RAG Best Practices - Pinecone 的 RAG 指南
• MMR for Information Retrieval - MMR 原论文
• Shannon Memory System - Shannon 记忆系统文档