Eino的基础组件（五）文档向量化与索引召回

🧭 写在前面

上一篇文章我们聊了 Eino 的文档加载、解析和转换：Loader 负责把文档读进来，Parser 负责把不同格式转成统一结构，Transformer 负责把长文档切成更适合处理的片段。

但在 RAG 应用里，文档切好只是第一步。真正让知识库能回答问题，还要继续解决三个关键问题：

如何把文本片段转换成向量：让语义相近的内容在向量空间中距离更近
如何把文档和向量存入索引：让系统后续可以快速查找
如何根据用户问题召回相关文档：把最相关的上下文交给模型生成答案

这三个问题分别对应 Eino 的三个基础组件：Embedding、Indexer、Retriever。

如果把 RAG 比作一个图书馆，Embedding 就像给每段文字打上语义坐标，Indexer 像把图书和坐标录入检索系统，Retriever 则像根据读者的问题把最相关的书页找出来。理解这三个组件之后，RAG 的主干链路就基本闭环了。

🧠 语义检索的核心思路

传统关键词搜索看的是“字面是否匹配”，比如用户问“怎么处理登录过期”，系统会优先找包含“登录”“过期”的文档。这个方式简单直接，但问题也明显：如果文档里写的是“token 失效后重新认证”，关键词搜索可能就抓不到。

向量检索解决的是“语义是否相近”。它会把文本转换成一组浮点数，也就是向量：

1
2
"登录过期怎么处理"        → [0.12, -0.38, 0.76, ...]
"token 失效后重新认证"    → [0.10, -0.35, 0.72, ...]

两个文本表达的意思越接近，向量之间的距离通常也越近。这样即使用户问题和文档没有完全相同的关键词，也能通过语义相似度召回相关内容。

在 Eino 中，这条链路通常可以拆成两段：

1
2
3
4
5
入库阶段：
Document 片段 → Embedding 生成向量 → Indexer 存储文档和向量

查询阶段：
用户问题 → Embedding 生成查询向量 → Retriever 从索引中召回 Document

注意这里的 Embedding 会被两边共用：入库时给文档生成向量，查询时给用户问题生成向量。同一个知识库中，文档向量和查询向量必须使用兼容的模型与维度，否则相似度比较就失去了意义。

🧩 三个组件的职责边界

先把组件边界讲清楚，后面看接口和代码会轻松很多。

组件	输入	输出	核心职责
`Embedding`	`[]string`	`[][]float64`	把文本转换成向量
`Indexer`	`[]*schema.Document`	`[]string`	存储文档并建立索引
`Retriever`	`query string`	`[]*schema.Document`	根据问题召回相关文档

这三个组件不是互相替代的关系，而是前后协作的关系：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
Loader / Parser / Transformer
        ↓
[]*schema.Document
        ↓
Indexer ── 使用 Embedding 生成文档向量
        ↓
向量数据库 / 搜索引擎 / 其他索引后端
        ↑
Retriever ── 使用 Embedding 生成查询向量
        ↑
用户问题

这个分层的好处是很明显的：

想换 Embedding 模型，只要替换 Embedder 实现
想从 Milvus 换到 Elasticsearch，只要替换 Indexer 和 Retriever 实现
想调整召回数量和阈值，可以通过 RetrieverOption 控制
编排层只依赖接口，不需要知道底层后端的细节

📐 Embedding 组件

Embedding 是一个用于将文本转换为向量表示的组件，是文档向量化的入口。它不关心文本从哪里来，也不关心向量要存到哪里。它的主要作用是将文本内容映射到向量空间，使得语义相似的文本在向量空间中的距离较近。这个组件在以下场景中发挥重要作用：

🔤 接口定义

Eino 中的 Embedding 接口非常克制：

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

这个接口有几个关键点：

功能：将一组文本转换为向量表示
参数：
- ctx：上下文对象，用于传递请求级别的信息，同时也用于传递 Callback Manager
- texts：待转换的文本列表
- opts：转换选项，用于配置转换行为
返回值：
- [][]float64：文本对应的向量表示列表，每个向量的维度由具体的实现决定
- error：转换过程中的错误信息

如果输入是三个文本片段，输出也应该是三个向量：

1
2
3
texts[0] → vectors[0]
texts[1] → vectors[1]
texts[2] → vectors[2]

这一点在实现自定义 Embedder 时很重要，不能打乱顺序，否则文档内容和向量会错位。

Embedding 组件使用 EmbeddingOption 来定义可选参数，下方是抽象出的公共 option。每个具体的实现可以定义自己的特定 Option，通过 WrapEmbeddingImplSpecificOptFn 函数包装成统一的 EmbeddingOption 类型。

1
2
3
4
type Options struct {
    // Model 是用于生成向量的模型名称
    Model *string
}

🧪 独立使用示例

下面是一个简化示例，展示如何使用 OpenAI 兼容的 Embedding 实现。真实项目中要根据所选供应商配置 APIKey、Model、BaseURL 等参数。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import (
    "context"
    "log"

    "github.com/cloudwego/eino-ext/components/embedding/openai"
)

func embedTexts(ctx context.Context, apiKey string) error {
    defaultDim := 1536

    embedder, err := openai.NewEmbedder(ctx, &openai.EmbeddingConfig{
        APIKey:     apiKey,
        Model:      "text-embedding-3-small",
        Dimensions: &defaultDim,
    })
    if err != nil {
        return err
    }

    texts := []string{
        "Eino 是一个 Go 语言的大模型应用开发框架",
        "Retriever 负责从知识库中召回相关文档",
    }

    vectors, err := embedder.EmbedStrings(ctx, texts)
    if err != nil {
        return err
    }

    log.Printf("texts=%d, vectors=%d, dim=%d", len(texts), len(vectors), len(vectors[0]))
    return nil
}

这段代码的重点不是某个具体模型，而是调用方式：业务代码面向 embedding.Embedder 接口写，底层可以换成 OpenAI、ARK、Ollama、Qianfan、dashscope 等实现。

🧷 使用注意点

Embedding 在 RAG 里看起来只是“调一下模型”，但它会直接影响召回质量：

模型要统一：入库和查询阶段尽量使用同一个 Embedding 模型
维度要匹配：向量数据库字段维度必须和模型输出维度一致
文本要适中：过短片段语义不足，过长片段容易稀释重点
批量要控制：批量太小效率低，批量太大容易触发供应商限流
错误要可追踪：记录失败文本、批次、模型名，方便补偿重试

我个人会把 Embedding 看成 RAG 的“语义压缩器”。它压缩得好，后面的索引和召回才有发挥空间；它压缩得乱，后面调 TopK、阈值、Prompt 都只是补救。

🗂️ Indexer 组件

Indexer 是一个用于存储和索引文档的组件。它的主要作用是将文档及其向量表示存储到后端存储系统中，并提供高效的检索能力。它通常会和向量数据库、搜索引擎或自定义存储后端打交道。

🧱 接口定义

Eino 的 Indexer 接口如下：

1
2
3
type Indexer interface {
    Store(ctx context.Context, docs []*schema.Document, opts ...Option) (ids []string, err error)
}

这个接口表达了一个非常清楚的意图：给我一批 Document，我负责存储并返回入库成功的 ID。

功能：存储文档并建立索引
参数：
- ctx：上下文对象，用于传递请求级别的信息，同时也用于传递 Callback Manager
- docs：待存储的文档列表
- opts：存储选项，用于配置存储行为
返回值：
- ids：存储成功的文档 ID 列表
- error：存储过程中的错误信息

常见的入库动作包括：

提取 Document.Content
使用 Embedding 生成文档向量
把内容、向量和 MetaData 写入后端
返回后端生成或确认的文档 ID

Indexer 的通用选项里有两个非常常用的配置：

1
2
3
4
5
6
type Options struct {
    // SubIndexes 是要建立索引的子索引列表
    SubIndexes []string   
    // Embedding 是用于生成文档向量的组件
    Embedding embedding.Embedder
}

SubIndexes：用于把文档写入一个或多个子索引，适合多租户、知识库分区等场景
Embedding：用于在入库阶段生成文档向量

可以通过以下方式设置选项

1
2
3
4
// 设置子索引
WithSubIndexes(subIndexes []string) Option
// 设置向量生成组件
WithEmbedding(emb embedding.Embedder) Option

📦 文档入库骨架

下面用一个偏工程化的骨架展示 Indexer 的使用位置。为了避免绑定具体向量数据库，这里把 buildIndexer 留成项目自己的初始化函数。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import (
    "context"
    "fmt"

    "github.com/cloudwego/eino/components/embedding"
    "github.com/cloudwego/eino/components/indexer"
    "github.com/cloudwego/eino/schema"
)

func storeDocuments(
    ctx context.Context,
    idx indexer.Indexer,
    emb embedding.Embedder,
    docs []*schema.Document,
) ([]string, error) {
    if len(docs) == 0 {
        return nil, nil
    }

    ids, err := idx.Store(ctx, docs,
        indexer.WithEmbedding(emb),
        indexer.WithSubIndexes([]string{"product_docs"}),
    )
    if err != nil {
        return nil, fmt.Errorf("store documents failed: %w", err)
    }

    return ids, nil
}

这段代码里最值得注意的是：Embedding 没有直接散落在业务逻辑里，而是作为 indexer.WithEmbedding(emb) 交给 Indexer。这样不同后端可以自行决定如何处理向量字段、批量写入和索引构建。

🧾 MetaData 的价值

在 RAG 中，Content 决定能不能召回，MetaData 决定召回后能不能用好。

我一般会在入库前把这些信息放进 MetaData：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
doc := &schema.Document{
    ID:      "auth-guide#chunk-003",
    Content: "当 access token 过期后，客户端应该使用 refresh token 换取新的 access token。",
    MetaData: map[string]any{
        "source":       "auth-guide.md",
        "category":     "登录鉴权",
        "chunk_index":  3,
        "parent_id":    "auth-guide",
        "section":      "Token 刷新",
        "updated_at":   "2026-08-25",
    },
}

这些元信息在后续会发挥很多作用：

召回结果展示来源，提升回答可信度
根据 category、tenant_id、version 做过滤
通过 parent_id 找回更完整的上下文
排查“为什么召回了这段文档”
做增量更新和删除时定位原文档

很多 RAG 系统一开始只存 Content，后面做权限、版本、引用来源时会非常痛苦。我的建议是：MetaData 宁可一开始设计得稍微完整一点，也不要等上线后再补历史数据。

🔍 Retriever 组件

Retriever 是查询阶段的核心。是一个用于从各种数据源检索文档的组件。它的主要作用是根据用户的查询（query）从文档库中检索出最相关的文档

🎯 接口定义

Retriever 的接口也很简洁：

1
2
3
type Retriever interface {
    Retrieve(ctx context.Context, query string, opts ...Option) ([]*schema.Document, error)
}

功能：根据查询检索相关文档
参数：
- ctx：上下文对象，用于传递请求级别的信息，同时也用于传递 Callback Manager
- query：查询字符串
- opts：检索选项，用于配置检索行为
返回值：
- []*schema.Document：检索到的文档列表
- error：检索过程中的错误信息

常用选项包括：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
type Options struct {
    // Index 是检索器使用的索引，不同检索器中的索引可能有不同含义
    Index *string
    
    // SubIndex 是检索器使用的子索引，不同检索器中的子索引可能有不同含义
    SubIndex *string
    
    // TopK 是检索的文档数量上限
    TopK *int
    
    // ScoreThreshold 是文档相似度的阈值，例如 0.5 表示文档的相似度分数必须大于 0.5
    ScoreThreshold *float64
    
    // Embedding 是用于生成查询向量的组件
    Embedding embedding.Embedder
    
    // DSLInfo 是用于检索的 DSL 信息，仅在 viking 类型的检索器中使用
    DSLInfo map[string]interface{}
}

这些选项基本覆盖了生产检索里最常见的控制点：

Index：指定主索引
SubIndex：指定子索引或分区
TopK：最多召回多少条文档
ScoreThreshold：过滤相似度过低的文档
Embedding：把用户问题转换成查询向量
DSLInfo：给特定后端传递过滤 DSL

可以通过以下方式进行设置

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
// 设置索引
WithIndex(index string) Option

// 设置子索引
WithSubIndex(subIndex string) Option

// 设置检索文档数量上限
WithTopK(topK int) Option

// 设置相似度阈值
WithScoreThreshold(threshold float64) Option

// 设置向量生成组件
WithEmbedding(emb embedding.Embedder) Option

// 设置 DSL 信息（仅用于 viking 类型检索器）
WithDSLInfo(dsl map[string]any) Option

💡 召回示例

下面是一个独立使用 Retriever 的骨架：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import (
    "context"
    "fmt"

    "github.com/cloudwego/eino/components/embedding"
    "github.com/cloudwego/eino/components/retriever"
    "github.com/cloudwego/eino/schema"
)

func retrieveContext(
    ctx context.Context,
    r retriever.Retriever,
    emb embedding.Embedder,
    question string,
) ([]*schema.Document, error) {
    docs, err := r.Retrieve(ctx, question,
        retriever.WithEmbedding(emb),
        retriever.WithTopK(5),
        retriever.WithScoreThreshold(0.35),
        retriever.WithIndex("knowledge_base"),
        retriever.WithSubIndex("product_docs"),
    )
    if err != nil {
        return nil, fmt.Errorf("retrieve context failed: %w", err)
    }

    return docs, nil
}

这里的 TopK 和 ScoreThreshold 需要结合业务调参：

TopK 太小：可能漏掉关键上下文
TopK 太大：会给模型塞入噪声，增加 token 成本
阈值太高：召回结果可能为空
阈值太低：召回结果可能相关性不足

实战里我通常先用一个保守组合，比如 TopK=5、ScoreThreshold=0.3~0.5，再根据测试集逐步调整。

📊 召回结果的处理

Retriever 返回的是 []*schema.Document，但这并不意味着要把所有内容原样塞进 Prompt。更稳妥的做法是先做一次整理：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import (
    "fmt"
    "strings"

    "github.com/cloudwego/eino/schema"
)

func buildContext(docs []*schema.Document) string {
    var builder strings.Builder

    for i, doc := range docs {
        source, _ := doc.MetaData["source"].(string)
        section, _ := doc.MetaData["section"].(string)

        builder.WriteString(fmt.Sprintf("【资料 %d】\n", i+1))
        if source != "" {
            builder.WriteString(fmt.Sprintf("来源：%s\n", source))
        }
        if section != "" {
            builder.WriteString(fmt.Sprintf("章节：%s\n", section))
        }
        builder.WriteString(doc.Content)
        builder.WriteString("\n\n")
    }

    return builder.String()
}

这样做有三个好处：

模型能看到文档来源，回答时更容易引用依据
多段文档之间有清晰边界，不容易混在一起
后续做调试时，可以直接看到哪几段内容进入了 Prompt

🔄 入库与召回的完整链路

现在把上一篇的文档处理流程和这一篇的向量索引流程连起来。

🧵 入库流程

一个典型的知识库入库流程如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
本地文件 / 网页 / 对象存储
        ↓
Loader 加载原始内容
        ↓
Parser 解析为 Document
        ↓
Transformer 分割为 chunk
        ↓
Embedding 生成文档向量
        ↓
Indexer 写入向量数据库或搜索引擎

为了让示例更贴近 Eino 的组件边界，可以把“文档处理”和“索引写入”拆开看。前面 Loader、Transformer 负责得到 []*schema.Document，后面再接一个专门的索引 Chain：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
import (
    "github.com/cloudwego/eino/components/indexer"
    "github.com/cloudwego/eino/compose"
    "github.com/cloudwego/eino/schema"
)

func buildStoreChain(idx indexer.Indexer) *compose.Chain[[]*schema.Document, []string] {
    chain := compose.NewChain[[]*schema.Document, []string]()
    chain.AppendIndexer(idx)
    return chain
}

这段代码主要表达索引阶段的编排思路：输入是已经处理好的文档片段，输出是入库成功的文档 ID。真实项目中可以在 Compile 和 Invoke 时传入 Callback、Embedding、运行时配置等选项。

🛤️ 查询流程

查询阶段的链路更短：

1
2
3
4
5
6
7
8
9
用户问题
  ↓
Retriever 召回相关 Document
  ↓
整理上下文
  ↓
ChatTemplate 构造 Prompt
  ↓
ChatModel 生成回答

用 Chain 表达时，可以先把 Retriever 作为第一个节点：

1
2
3
4
5
func buildRetrieveChain(r retriever.Retriever) *compose.Chain[string, []*schema.Document] {
    chain := compose.NewChain[string, []*schema.Document]()
    chain.AppendRetriever(r)
    return chain
}

如果要做完整问答，可以继续接 Lambda 做上下文拼接，再接 ChatTemplate 和 ChatModel。当业务逻辑只是“召回 → 生成”，Chain 就够用；如果要加入多路召回、条件分支、重排序、降级策略，就更适合用 Graph。

🛠️ 实战：构建一个最小 RAG 骨架

下面用一个最小骨架把组件位置串起来。它不绑定具体数据库，重点是展示各组件怎么协作。

⚙️ 组件初始化

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
type RAGComponents struct {
    Embedder  embedding.Embedder
    Indexer   indexer.Indexer
    Retriever retriever.Retriever
}

func NewRAGComponents(ctx context.Context) (*RAGComponents, error) {
    // 1. 初始化 Embedding
    embedder, err := newEmbedder(ctx)
    if err != nil {
        return nil, err
    }

    // 2. 初始化 Indexer
    idx, err := newIndexer(ctx, embedder)
    if err != nil {
        return nil, err
    }

    // 3. 初始化 Retriever
    ret, err := newRetriever(ctx, embedder)
    if err != nil {
        return nil, err
    }

    return &RAGComponents{
        Embedder:  embedder,
        Indexer:   idx,
        Retriever: ret,
    }, nil
}

这里我会让 Indexer 和 Retriever 共享同一个 Embedder。这样可以减少“入库用 A 模型、查询用 B 模型”导致的向量空间不一致问题。

🧮 文档入库

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
func (c *RAGComponents) StoreChunks(ctx context.Context, chunks []*schema.Document) ([]string, error) {
    if len(chunks) == 0 {
        return nil, nil
    }

    ids, err := c.Indexer.Store(ctx, chunks,
        indexer.WithEmbedding(c.Embedder),
        indexer.WithSubIndexes([]string{"default"}),
    )
    if err != nil {
        return nil, fmt.Errorf("store chunks failed: %w", err)
    }

    return ids, nil
}

在生产环境中，这里通常还会加：

批量大小控制
失败重试
重复文档去重
入库进度记录
文档版本号

不要小看这些工程细节。RAG 的稳定性很多时候不是输在模型，而是输在“知识库更新不完整、重复入库、旧版本未删除”这些脏数据问题上。

📥 查询召回

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
func (c *RAGComponents) Retrieve(ctx context.Context, question string) ([]*schema.Document, error) {
    docs, err := c.Retriever.Retrieve(ctx, question,
        retriever.WithEmbedding(c.Embedder),
        retriever.WithTopK(5),
        retriever.WithScoreThreshold(0.35),
        retriever.WithSubIndex("default"),
    )
    if err != nil {
        return nil, fmt.Errorf("retrieve failed: %w", err)
    }

    return docs, nil
}

这个函数看起来简单，但已经包含 RAG 召回的核心动作：把问题变成查询向量，然后从索引中找出相似文档。

📈 召回质量怎么调

RAG 的难点不只是“能召回”，而是“召回得准”。常见问题和调整方向可以先按下面这个表排查。

问题	可能原因	调整方向
召回为空	阈值太高、索引为空、维度不匹配	降低阈值，检查入库数量和向量维度
召回很多无关内容	chunk 太大、阈值太低、知识库混杂	提高阈值，优化分块，增加元信息过滤
答案缺少关键细节	`TopK` 太小、分块切断上下文	增大 `TopK`，增加 overlap，召回父文档
召回结果重复	文档重复入库、chunk overlap 太大	加去重逻辑，控制 overlap
多租户串数据	缺少租户过滤	使用 `SubIndex` 或后端 DSL 过滤

我会优先从三个地方动手：

分块策略：先确保每个 chunk 本身语义完整
元信息过滤：用 category、tenant_id、version 缩小搜索范围
召回参数：再调整 TopK 和 ScoreThreshold

很多人一上来就调模型，其实更该先看数据。RAG 是数据工程味很重的系统，召回链路干净，模型才有稳定发挥的空间。

🚦 生产实践建议

🧯 入库侧建议

固定文档 ID 规则：比如 文件名#chunk-序号，方便更新和删除
保留父文档关系：通过 parent_id 找回完整上下文
记录文档版本：避免新旧内容混在同一个知识库里
批量写入索引：减少网络开销，提高入库效率
失败可补偿：记录失败文件和失败 chunk，支持重试

🛡️ 查询侧建议

先过滤再向量检索：能用租户、分类、权限缩小范围就先缩小
召回后去重：同一父文档的相邻 chunk 可以合并或压缩
保留得分信息：方便观察召回质量和调参
上下文不要贪多：相关性不足的内容会干扰模型
空召回要降级：明确告诉用户“知识库未找到相关资料”，不要硬编

📡 可观测性建议

Embedding、Indexer、Retriever 都应该接入 Callback 或统一日志。至少记录：

Embedding 的输入条数、耗时、模型名
Indexer 的入库数量、成功 ID、失败原因
Retriever 的 query、TopK、阈值、召回数量
每次回答实际进入 Prompt 的文档 ID 和来源

有了这些数据，后面排查“为什么答错”才不会变成玄学。

🔗 和上一篇文章的连接

到这里，Eino 文档类组件的主线就连起来了：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
Document Loader
    负责从文件、网页、对象存储中加载原始内容

Document Parser
    负责把 PDF、HTML、Markdown 等格式解析成 Document

Document Transformer
    负责切分、过滤、清洗 Document

Embedding
    负责把文本内容转换成语义向量

Indexer
    负责把 Document 和向量写入索引后端

Retriever
    负责根据用户问题召回相关 Document

这套组件拆分得很细，但组合起来并不复杂。Eino 的思路不是给你一个“全自动黑盒 RAG”，而是把每个关键环节抽象成可替换、可编排、可观测的组件。这样你既能快速搭出原型，也能在生产环境里逐步优化每个环节。

📝 总结

这一篇主要梳理了 Eino 中 Embedding、Indexer、Retriever 三个组件：

Embedding：把文本转换成向量，是语义检索的基础
Indexer：把文档和向量写入后端，建立可检索索引
Retriever：根据用户问题召回相关文档，给模型提供上下文

它们共同构成了 RAG 的核心闭环：

1
文档处理 → 向量化 → 建索引 → 查询召回 → 构造上下文 → 模型回答

在实际项目中，不要只关注“代码能不能跑通”，更要关注数据质量、分块策略、元信息设计、召回参数和可观测性。RAG 的上限由模型决定，但下限往往由数据链路决定。

下一步如果继续往下做，就可以把 Retriever 接到 ChatTemplate 和 ChatModel，构建一个完整的知识库问答应用。

参考资料：