乐于分享
好东西不私藏
夜雨聆风 > > 办公文件 > Embedder 接口 + 缓存层源码:一个把 key 撑大 3 倍的实现(第71篇-E57)

Embedder 接口 + 缓存层源码:一个把 key 撑大 3 倍的实现(第71篇-E57)

当前时间： 2026-08-08 13:55:00 分类：办公文件 评论(0)
Embedder 接口 + 缓存层源码:一个把 key 撑大 3 倍的实现(第71篇-E57)

系列「企业级 AI Agent 实现拆解」E57 篇，Part 13 RAG 篇第六章。上一篇 结尾我说发现了 HashGenerator 里有点问题。这篇验证完了，结论比预想的严重：**用 sha256 「哈希」一段 108 字节的中文，得到的 Redis key 是 316 字节。**而且换 md5、换 sha512，都救不了。

读完这篇你会知道

  • Embedder
     接口只有一个方法，那 token 用量从哪拿
  • 缓存层三个接口怎么分工：Embedder 包装 / Cacher 存取 / Generator 造 key
  • 实测：5 条文本只有 2 条发给上游，缓存怎么削调用
  • HashGenerator 的 key 比原文还长
    ，根因是 hash.Hash.Sum(b) 的语义被用反了
  • 两种不一致的失败模式：Get 失败炸业务，Set 失败静默吞
  • 逐条 Get 没有 MGET：100 片入库 = 100 次 Redis 往返
  • 三处「包一层」就能修的地方，和一处包不了的

一、Embedder 接口：一个方法，token 用量在别处

type Embedder interface {    EmbedStrings(ctx context.Context, texts []string, opts ...Option) ([][]float64, error)}

返回值里只有向量。没有 token 用量，没有耗时，没有模型信息。

那这些信息去哪了？在 callback 里：

// components/embedding/callback_extra.gotype CallbackOutput struct {    Embeddings [][]float64    Config     *Config      // Model / EncodingFormat    TokenUsage *TokenUsage  // PromptTokens / TotalTokens    Extra      map[string]any}

想统计 embedding 花了多少 token，只能挂 callback handler，返回值里拿不到。

这是个刻意的取舍：接口签名保持最小，附加信息走旁路。好处是接口稳定——加一个统计字段不用改所有实现；代价是想拿数据得多写一段 handler。

第 67 篇《Document 组件源码》讲过 Conv*CallbackOutput 会串台，这里同样要先看 info.Component == components.ComponentOfEmbedding 再转换。

调用时能改的参数也只有一个：

type Options struct {    Model *string}

维度、编码格式、超时，全在构造时定死。这一点后面会咬人，第七节说。

二、缓存层的三个接口

eino-ext/components/embedding/cache 是官方的缓存包装。三个接口分工很干净：

接口
方法
管什么
cache.EmbedderEmbedStrings
包装层：查缓存、调上游、回填
cache.CacherGet
 / Set
存取：Redis、内存、随你
cache.GeneratorGenerate造 key
：文本 + 模型 → 缓存键
type Cacher interface {    Set(ctx context.Context, key string, value []float64, expire time.Duration) error    Get(ctx context.Context, key string) ([]float64, bool, error)}type Generator interface {    Generate(ctx context.Context, text string, opt GeneratorOption) string}

Generator 独立成接口是对的——key 策略是个真的会变的东西：要不要哈希、要不要带命名空间、多租户要不要隔离，每个项目不一样。

包装层本身实现了 embedding.Embedder，所以能无缝插在任何位置：

var _ embedding.Embedder = (*Embedder)(nil)emb, err := cache.NewEmbedder(realEmbedder,    cache.WithCacher(redisCacher),    cache.WithGenerator(cache.NewSimpleGenerator()),    cache.WithExpiration(time.Hour),)

**又是「包一层」。**第 70 篇《Embedding 选型》的 countingEmbedder 是我自己包的，这个是官方包的，套路完全一样——因为 Embedder 只有一个方法，装饰它的成本接近于零。

两个必填项，缺了直接报错：

if e.cacher == nil {    return nil, ErrCacherRequired}if e.generator == nil {    return nil, ErrGeneratorRequired}

默认过期时间 2 小时（expiration: time.Hour * 2）。

三、执行流程：未命中的才发给上游

func(e *Embedder) EmbedStrings(ctx context.Context, texts []string, opts ...embedding.Option) ([][]float64, error) {    var (        embeddingsByKey = make(map[int][]float64)        embeddingOpts   = embedding.GetCommonOptions(nil, opts...)        uncached        []int        uncachedTexts   []string    )    var generatorOpt GeneratorOption    if embeddingOpts.Model != nil {        generatorOpt.Model = *embeddingOpts.Model    }    // ① 逐条查缓存，未命中的记下来    for idx, text := range texts {        key := e.generator.Generate(ctx, text, generatorOpt)        emb, ok, err := e.cacher.Get(ctx, key)        if err != nil {            return nil, err                   // ← 注意这里        } else if ok {            embeddingsByKey[idx] = emb        } else {            uncached = append(uncached, idx)            uncachedTexts = append(uncachedTexts, text)        }    }    // ② 未命中的合成一批，一次调上游    if len(uncachedTexts) > 0 {        uncachedEmbeddings, err := e.embedder.EmbedStrings(ctx, uncachedTexts, opts...)        if err != nil {            return nil, err        }        // ③ 回填缓存        for i, idx := range uncached {            key := e.generator.Generate(ctx, texts[idx], generatorOpt)            if err := e.cacher.Set(ctx, key, uncachedEmbeddings[i], e.expiration); err != nil {                _ = err                       // ← 也注意这里            }            embeddingsByKey[idx] = uncachedEmbeddings[i]        }    }    // ④ 按原始下标还原顺序    result := make([][]float64, len(texts))    for i := range texts {        if emb, ok := embeddingsByKey[i]; ok {            result[i] = emb        } else {            result[i] = nil   // it seems that such a case should not happen        }    }    return result, nil}

第 ② 步是这个包装层最值钱的地方：未命中的文本被合成一批，只调上游一次，而不是逐条调。顺序靠 uncached 存的原始下标还原。

实测三轮：

第一轮（全未命中）：  上游调用 1 次，送了 3 条；缓存 Get 3 次 Set 3 次第二轮（全命中）：  上游调用 1 次，送了 3 条；缓存 Get 6 次 Set 3 次第三轮（3 老 + 2 新，部分命中）：  上游调用 2 次，送了 5 条；缓存 Get 11 次 Set 5 次  → 上游只收到 2 条（未命中的那两条），不是 5 条

第二轮上游调用数没涨（还是 1 次），说明全命中时一次上游都没调。第三轮传了 5 条，上游只多收到 2 条。

Generate 被调了两次（查的时候一次，回填的时候一次）。key 生成如果很贵，这里是双倍开销——下面会看到这个「贵」是真的。

四、坑一：逐条 Get，没有批量

注意上面实测里 Get 的次数：三轮累计 11 次。

因为查缓存是个 for 循环，一条一个 Get。而这是接口签名决定的：

Get(ctx context.Context, key string) ([]float64, bool, error)

单键。没有 MGet(keys []string)。

100 片文档入库 = 100 次 Redis 往返。

内网 RTT 按 0.5ms 算，100 次就是 50ms；跨可用区 2ms 的话就是 200ms。而这本来一个 MGET 就能搞定。

能不能在自己的 Cacher 实现里偷偷用 pipeline？**不能。**包装层是同步逐条调 Get 的，你的实现根本不知道后面还有多少个 key 要查，攒不起批。

要批量只有一条路：不用这个包装层，自己在业务侧写查缓存的逻辑。几十行的事，但得自己维护。

判断标准很简单：

  • 查询侧
    （一次一条用户提问）→ 用官方包装层，没问题
  • 建库侧
    （一次几百上千片）→ 自己写，用 MGET

五、坑二：HashGenerator 把 key 撑大 3 倍

这是本篇的重点。

官方提供两个 Generator：

type SimpleGenerator struct{}func(g *SimpleGenerator) Generate(_ context.Context, text string, opt GeneratorOption) string {    return fmt.Sprintf("%s-%s", text, opt.Model)     // 原文 + "-" + 模型名}type HashGenerator struct {    *SimpleGenerator    hasher hash.Hash}func(g *HashGenerator) Generate(ctx context.Context, text string, opt GeneratorOption) string {    plainText := g.SimpleGenerator.Generate(ctx, text, opt)    return fmt.Sprintf("%x", g.hasher.Sum([]byte(plainText)))}

SimpleGenerator 把原文当 key，长文本的 key 会很长——所以官方给了 HashGenerator，看起来是为了把 key 压成定长摘要。注释也是这么写的：

// Note: Because of the use of the [hash.Hash] algorithm, there is a probability that data// with different text and options will generate the same key. This is a trade-off// between uniqueness and performance.

「有碰撞概率，是唯一性和性能的取舍」——听起来很合理。

实测一下。输入是一句 108 字节的中文：

原文       108 字节SimpleGen  126 字节HashGen    316 字节   ← 期望是定长 64（sha256 hex）

哈希完变成 316 字节，比原文长了将近 2 倍。

看看 key 长什么样：

HashGen key: "e59198e5b7a5e585a5e8818ce6bba1e4b880e5b9b4e5908ee5bc80e5a78be4baab...              ...203520e5a4a9e380822d746578742d656d62656464696e672d7633              e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"

前面那一大段是原文的 hex 编码，尾巴 64 个字符是个固定值。验证：

HashGen key 是否以 hex(SimpleGen key) 开头？true去掉这段前缀后，剩余 64 字节：e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855空输入的 sha256    ：e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855

尾巴那 64 个字符，正好是「空输入的 sha256」。

根因：Sum 不是「计算哈希」

hash.Hash 的 Sum 方法签名是：

Sum(b []byte) []byte

它的语义是：**把当前累积状态的摘要 append 到 b 后面，返回拼接结果。**不是「计算 b 的哈希」。

所以 g.hasher.Sum([]byte(plainText)) 做的是：

plainText（原样） + digest(至今为止 Write 进去的所有数据)

而这个 hasher 从创建到使用一次 Write 都没调过，累积状态是空的，摘要恒为 sha256("")。

于是 key = hex(原文 + "-" + 模型名 + sha256(""))。原文一个字节都没被压缩，还因为 hex 编码翻了一倍，再加 32 字节固定尾巴。

正确用法是：

h.Reset()h.Write(data)return h.Sum(nil)      // ← 参数传 nil

三个连带后果

① 注释说反了。「有碰撞概率」——不存在。原文完整出现在 key 里，两个不同文本的 key 必然不同。它比 SimpleGenerator 更不可能碰撞，代价是 key 长了 2.5 倍。

② key 长度完全不定长：

不同长度文本的 HashGen key 长度：106 vs 700（定长哈希应该相等）

100 个字的文本，key 700 字节。1000 字的片，key 就是 6KB 级别。Redis key 本身要占内存、要走网络、还会出现在慢日志和监控里。

③ 换哈希算法救不了：

md5     len= 50  e5b9b4e581872d7633d41d8cd98f00b204e9800998ecf8427esha256  len= 82  e5b9b4e581872d7633e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855sha512  len=146  e5b9b4e581872d7633cf83e1357eefb8bdf1542850d66d8007d620e4050b5715dc...

三个算法的前半段完全相同（e5b9b4e581872d7633 = hex 的「年假-v3」），只有尾巴长度不同。换成 md5 只是让 key 短一点——因为空输入的 md5 比空输入的 sha256 短。

选 sha512 反而最长。

自己写一个，10 行

type sha256Generator struct{}func(g *sha256Generator) Generate(_ context.Context, text string, opt cache.GeneratorOption) string {    h := sha256.New()    h.Write([]byte(text))    h.Write([]byte{0})              // 分隔符：防止 "ab"+"c" 和 "a"+"bc" 撞    h.Write([]byte(opt.Model))    return hex.EncodeToString(h.Sum(nil))   // ← 传 nil}

三个要点：

  1. h.Sum(nil)
    ，不是 h.Sum(data)
  2. 每次 New 一个新 hasher
    ，别复用实例。hash.Hash 有内部状态且不是并发安全的——官方那个 HashGenerator 把 hasher 存成结构体字段，多 goroutine 共用一个实例本身就是设计味道（当前实现因为从不 Write 才侥幸没出事）
  3. 字段之间加分隔符
    ，避免边界歧义

key 变成稳定的 64 字符，跟文本长度无关。

六、坑三：两种失败模式不一致

回看流程里我标注的两处。

Get 失败 → 整个请求失败：

emb, ok, err := e.cacher.Get(ctx, key)if err != nil {    return nil, err          // 直接返回}

Set 失败 → 静默忽略：

if err := e.cacher.Set(ctx, key, uncachedEmbeddings[i], e.expiration); err != nil {    _ = err                  // 源码原文}

实测：

Cacher.Get 返回 error → EmbedStrings err = redis connection refused  上游被调用了 0 次 → 缓存挂了，业务直接失败（fail-closed）Cacher.Set 返回 error → EmbedStrings err = <nil>，结果 [[6 7 8 9]]  上游被调用了 1 次 → 写缓存失败被静默忽略，只是白算（fail-open）

Redis 一挂，你的检索功能整个不可用——即使上游 embedding 服务完全健康，上游一次都不会被调用。

Set 那边的处理是对的（写缓存失败无非是下次白算一遍）。Get 这边我认为反了：**缓存是加速器，不是依赖项。**读不到就该走上游。

包一层就能修，跟前面几处同一个套路：

type resilientCacher struct {    inner cache.Cacher}func(r *resilientCacher) Get(ctx context.Context, key string) ([]float64, bool, error) {    v, ok, err := r.inner.Get(ctx, key)    if err != nil {        // 缓存故障降级成「未命中」，让上游顶上        return nil, false, nil    }    return v, ok, nil}func(r *resilientCacher) Set(ctx context.Context, key string, v []float64, exp time.Duration) error {    return r.inner.Set(ctx, key, v, exp)}

生产上建议再加个计数器，把降级次数打成指标——不然缓存悄悄挂了三天没人知道，只是账单涨了。

七、key 里带 model：对了一半

这个设计是对的：

同一文本 三次请求（v2 / v3 / v2）→ 上游调用 2 次缓存里的 key：  "年假-v2"  "年假-v3"  → model 进了 key，v2 和 v3 各存一份，不会串味

第 66 篇《最简 RAG》强调过「存和查必须同一个模型」，key 里带模型名正好防住了换模型读到旧向量。第三次请求 v2 命中了缓存，所以三次只调了两次上游。

但只有 Model 进了 key。

var generatorOpt GeneratorOptionif embeddingOpts.Model != nil {    generatorOpt.Model = *embeddingOpts.Model}

GeneratorOption 结构体里就一个字段。而第 70 篇提到，ark 和 openai 的 Dimensions 是可配的——虽然目前是构造时配置（不同维度就是不同 Embedder 实例，各自的缓存自然分开），但如果你自己写的 Embedder 支持调用时改维度，那就有隐患了：

同一段文本、同一个模型、不同维度 → 同一个 key → 第二次请求拿到第一次那个维度的向量。

不会报错。1024 维的库里混进 512 维的向量，cosine 计算时循环 for i := range a 只走 512 位，算出来是个看着正常的数——静默的错误相似度。

自己实现 Generator 时，把所有影响向量结果的参数都拼进 key。

还有个小细节：不传 WithModel 时，GeneratorOption.Model 是空串：

不传 WithModel 时的 key：  "年假-"  ← Model 空，尾巴是个裸的 "-"

所以「不指定模型」和「指定了一个空名字的模型」共享缓存。实践中每次都显式传 embedding.WithModel(...) 更稳妥。

八、Redis Cacher 的实现细节

func NewCacher(rdb redis.UniversalClient, opts ...Option) *Cacher {    cacher := &Cacher{        rdb:    rdb,        prefix: "eino:",        codec:  defaultCodec,    }    // ...}func WithPrefix(prefix string) Option {    return optionFunc(func(c *Cacher) {        c.prefix = strings.TrimSuffix(prefix, ":") + ":"   // 冒号自动规范化    })}

默认前缀 eino:，WithPrefix 会先削掉尾部冒号再补一个——所以传 "myapp" 和 "myapp:" 结果一样。这个细节挺贴心。

redis.Nil 被正确地翻译成「未命中」而不是错误：

data, err := c.rdb.Get(ctx, c.prefix+key).Bytes()if err != nil {    if errors.Is(err, redis.Nil) {        return nil, false, nil        // 未命中，不是错误    }    return nil, false, err}

序列化用 sonic 存 JSON：

var defaultCodec codec = &sonicCodec{}func(*sonicCodec) Marshal(v any) ([]byte, error) {    return sonic.Marshal(v)}

**[]float64 存成 JSON 数组文本。**一个 float64 用 JSON 写出来平均十几个字符，而二进制只要 8 字节。1024 维的向量，JSON 大概是二进制的两倍多。

想换成紧凑的二进制格式？codec 是包内私有接口：

type codec interface {          // 小写，包外不可实现    Marshal(v any) ([]byte, error)    Unmarshal(data []byte, v any) error}

没有 WithCodec 选项。要换编码只能自己实现整个 cache.Cacher——好消息是那个接口只有两个方法，照着写不到 40 行。

九、这层缓存到底值不值

值得用的场景：

  • 查询侧
    ：热门问题反复被问，同一句话不用重复算。而且查询是一条一条来的，逐条 Get 的缺点不成立
  • 多次入库同一份文档
    ：调切片参数时反复重建索引，文本没变的片直接命中
  • 多租户共享语料
    ：公共知识库被多个租户各自建库

不值得用的场景：

  • 一次性全量建库
    ：每条文本都是新的，缓存全部未命中，纯粹白搭 N 次 Get 往返

默认过期时间 2 小时也要留意：

e := &Embedder{    embedder:   embedder,    expiration: time.Hour * 2,}

对查询侧够用，对「调参数反复重建索引」的场景太短——中午建的库，下午再建一次就全过期了。这种场景显式给个 cache.WithExpiration(7 * 24 * time.Hour)。

小结

  • Embedder 接口返回值只有向量
    ，token 用量、模型信息全在 callback 的 CallbackOutput 里
  • 缓存层三接口分工干净
    ：包装层管流程、Cacher 管存取、Generator 管造 key。又是「包一层」，因为接口只有一个方法
  • 未命中的文本会合成一批只调上游一次
    ：实测传 5 条，上游只收到 2 条
  • HashGenerator 的 key 比原文长 2.5 倍
    ：hash.Hash.Sum(b) 是「append 摘要到 b」，不是「哈希 b」。key = hex(原文) + sha256("")，换 md5/sha512 都只改尾巴
  • 正确写法是 h.Write(data) 后 h.Sum(nil)
    ，而且每次 New 新 hasher（hash.Hash 非并发安全）
  • 两种失败模式不一致
    ：Get 失败炸整个请求（Redis 挂 = 检索不可用），Set 失败静默吞。前者建议包一层降级成「未命中」
  • 只有 Model 进 key
    ：其他影响向量结果的参数（如可变维度）不进，有静默串味的风险
  • 逐条 Get 无 MGET
    ：接口签名决定的，建库侧建议绕过包装层自己批量
  • codec 是包内私有
    ，想换紧凑二进制编码只能自己实现 Cacher（两个方法，40 行）

下一篇（第 72 篇《pgvector 入门》）从「算向量」走到「存向量」：Docker 起一个 Qdrant，写入、查询、过滤，5 分钟跑通。

代码状态说明：本文五组验证（HashGenerator key 结构 / 缓存削减调用 / 两种失败模式 / model 进 key / 三种哈希算法对比）均在 eino v0.9.13 + eino-ext/components/embedding/cache（2026-07-24 版本）下真机运行，输出原样粘贴。验证用的是我自己写的内存 Cacher 和计数 Embedder，没有连真实 Redis，也没有调真实 embedding 服务——这几个结论都在包装层内部，不依赖外部服务。第八节 Redis Cacher 的内容为源码解读。