声明本文数据源于官方文档与官方实现重点参考 Indexer 使用说明 以及eino/components/indexer/interface.go、eino/components/indexer/option.go相关源码。为什么很多人会用 Indexer却没真正看懂 Store1. 它的用处它又被误会在哪它能干嘛它最适合哪些场景它不能直接干嘛2. Indexer 不是“向量库 insert 封装”3. Indexer 在 RAG 入库链路里到底站在哪4. 接口只有一个方法但 Store 这个词一点都不简单5. 公共 Option 真正控制的不是“几个小参数”4.1 SubIndexes 更像逻辑分区不只是一个字符串数组4.2 Embedding 是写入时临时挂接的能力不是 Indexer 本体6. 用一个 Milvus 最小例子把 Store 的写入链路看顺7. 为什么说 Indexer 不只服务向量数据库8. 放进 Chain / Graph 里你会发现 Indexer 也是正式组件9. Callback 和自定义实现10. 总结参考资料很多人做知识库时前面那几步通常都知道怎么搞文档切块文本做 embedding向量库建 collection看起来好像已经差不多了。可真到落库这一步问题马上就来了文档正文怎么写进去向量是谁来生成、什么时候生成元数据和来源信息放哪里多个知识库、多个业务空间怎么分区写写进去以后怎么保证后面真的能被检索到这时候你就会发现Indexer这层不是可有可无。说白了Indexer本质上就是“把文档写成以后能被检索的样子”的组件。它不是只管塞一段文本进去。它更像是把文档、向量、元数据、子索引这些东西一起整理好再送进可检索后端。这也是为什么它在知识库入库、语义搜索底库构建、多知识库分区写入这些场景里特别关键。但很多人第一次看到Indexer还是会下意识地把它理解成“哦这不就是调一下 Milvus、VikingDB 或 ES 的写入接口吗”问题也恰恰出在这儿。因为如果事情真这么简单那我们明明已经有Embedding向量库 SDK搜索引擎写入接口为什么 Eino 还要单独设计一个Indexer这篇文章想讲清楚的就是这件事。如果你前面刚看过我上一篇 Document Loader 的文章那篇讲的是文档怎么进入[]*schema.Document这套统一协议。这一篇刚好接上下一站当文档已经变成标准协议后它到底怎么被写进“可检索系统”1. 它的用处它又被误会在哪先别急着看接口。如果一上来我就讲 API这样虽可以记住函数名但不利于大家理解它能解决什么问题。Indexer更适合先从“它是拿来干嘛的”讲起。它能干嘛放到最常见的知识库链路里Indexer主要做的是这些事1. 把文档写进可检索后端不是只写正文。而是把schema.Document这套统一协议整理成后端真正能存、以后也真正能查的样子。2. 把向量和元数据一起落进去很多场景不是只有文本内容。你还得把向量、来源、chunk 编号、业务标签这些信息一起写进去不然后面检索和追踪都会很难受。3. 处理逻辑子索引或知识库分区你可能不是只有一个知识库。多租户、多业务空间、多资料域这些场景都要求写入时就把路由和隔离想清楚。它最适合哪些场景最常见的就是这几类知识库入库语义搜索底库构建多知识库 / 多租户分区写入它不能直接干嘛这块也必须提前讲清楚。否则你会误会它的功能。Indexer很重要但它不是万能层。它不负责切块。文档怎么拆 chunk不是它的职责。它不负责生成最终回答。那是 LLM 干的事。它不负责读侧召回。怎么把内容查出来是Retriever的边界。所以更准确地说Embedding是把文本变成向量。Indexer是把内容写成可检索对象。Retriever是再把这些东西查出来。这三个环节如果混在一起后面链路虽然也能跑就会造成边界模糊耦合严重的场景。2.Indexer不是“向量库 insert 封装”这里先把结论摆出来Indexer不是一个“帮你调 Milvus / VikingDB / ES SDK”的小工具。它是 Eino 在写入侧给出的统一组件协议。这层协议真正收口的是三件事文档输入统一成[]*schema.Document写入行为统一成Store(ctx, docs, opts...)写入结果统一成[]string形式的ids这就意味着它关心的是“写入侧边界”不是某个后端产品自己的调用细节。所以你看Indexer时最好先把几个常见误会排掉它不负责切块。文档怎么切成 chunk是Loader / Parser后面的预处理问题不是Indexer本体。它不负责读侧召回。相似度搜索、过滤、排序、返回 topK那是Retriever的边界。它不等于Embedding。向量可以在写入时生成但生成向量这件事本身仍然是另一层能力。也正因为这三件事被拆开了Eino 才能同时挂住 Milvus、VikingDB、ES、OpenSearch 这些看起来很不像一家人的后端实现。如果它只是“向量库 insert 封装”那 ES / OpenSearch 这两类实现就会显得很别扭。可官方偏偏把它们也归在Indexer下面这恰恰说明Indexer抽象的不是“向量库”而是“可检索后端的写入入口”。一套可插拔的接口3.Indexer在 RAG 入库链路里到底站在哪很多人理解 RAG 时脑子里只有一句话“文档切块做 embedding丢进向量库。”这当然没错但如果你在 Eino 里写组件你最好把链路拆得再清楚一点Source - Loader / Parser - []*schema.Document - 切块 / 清洗 - Embedding / Field Mapping - Indexer.Store - Retriever - ChatModel这里最关键的是中间那段。Loader / Parser负责把不同来源的内容收口成标准Document。Indexer负责把这些Document写进后端让它以后能被查出来。而Retriever则负责真正把它们读出来。也就是说Loader / Parser管“东西从哪来、怎么解释”Indexer管“怎么写进去”Retriever管“怎么查出来”很多人之所以会把Indexer理解歪就是因为把“写进去”和“以后怎么查”混成了一件事。可在工程里这两件事差得很远。写入时你关心的是文档 ID 怎么处理向量何时生成元数据写到哪些字段逻辑分区、子索引、批量写入怎么做召回时你关心的是query 该怎么向量化topK 怎么取filter 怎么写score 怎么解释如果这两层边界不拆开最后很容易变成“能跑但组件职责已经糊了”。4. 接口只有一个方法但Store这个词一点都不简单官方核心接口其实非常短typeIndexerinterface{Store(ctx context.Context,docs[]*schema.Document,opts...Option)(ids[]string,errerror)}很多人第一次看到这个接口会觉得信息量不大。可它真正想收口的边界恰恰都藏在这几个参数里。先看ctx。在 Eino 里ctx从来都不只是取消信号。官方文档已经明确写了它还承担Callback Manager的传递。这意味着Store不是一个藏在角落里的工具函数而是一段可以被编排、被观察、被追踪的正式运行时行为。再看docs []*schema.Document。这很关键。Indexer吃进去的不是某家向量库自己的 row也不是某个搜索引擎专属的字段结构而是统一文档协议。这件事的价值在上一篇Document Loader里其实已经埋下了文档一旦被标准化成schema.Document后面的写入端就终于可以和“来源差异”解耦。最后看返回值ids []string。这块很多人会想当然地把它理解成“就是把doc.ID原样回给你”。但实际上ids更准确的意思是后端最终确认写入成功的文档标识。它可能是直接沿用你传进来的Document.ID后端生成的新 ID一次批量 upsert 之后真正生效的主键集合所以Store这个词千万别按数据库里那种“我插一行你回一个自增主键”的直觉去理解。在 Eino 语境里一次Store里可能同时发生文档字段映射向量生成批量写入子索引分流回调触发错误上抛这已经明显不是一句“insert 一下”能说清的事了。5. 公共 Option 真正控制的不是“几个小参数”官方给Indexer的公共 option 很克制typeOptionsstruct{SubIndexes[]string// 子索引/子分区这批文档要写到哪些逻辑分组里Embedding embedding.Embedder// 向量模型写入前用它把文本转成向量}funcWithSubIndexes(subIndexes[]string)Option// 设置子索引/分区funcWithEmbedding(emb embedding.Embedder)Option// 设置本次写入使用的向量生成器字段不多但信息量不小。4.1SubIndexes更像逻辑分区不只是一个字符串数组很多人第一次看SubIndexes会把它当成“顺手多传几个名字”。可如果你把它放回知识库场景里就会发现它更像逻辑分区入口。比如同一套物理后端里你可能会按下面这些维度做隔离不同知识库不同租户不同业务空间不同文档域这时SubIndexes的作用就不是“多一个参数”这么简单了。它更接近在同一个Indexer抽象之下把文档路由到不同的逻辑子索引或子分区。所以我更愿意把它理解成写入侧的 namespace / partition 入口而不是一个普通切片字段。4.2Embedding是写入时临时挂接的能力不是Indexer本体WithEmbedding更值得多看两眼。它说明什么说明 Eino 允许你在Store这一跳里临时指定“这批文档怎么向量化”。也就是说向量生成可以是Indexer初始化时配置好的默认能力本次调用临时覆盖进去的 embedder这就把“写入协议”和“向量模型选择”拆开了。而且还有一个容易被忽略的点。VikingDB 示例里官方给的是后端内建 embedding 配置EmbeddingConfig:volc_vikingdb.EmbeddingConfig{UseBuiltin:true,ModelName:bge-m3,UseSparse:true,},这恰恰说明Indexer可以挂接 embedding但它本身不等于 embedding。有些实现会走外部 embedder有些实现会直接利用后端内建能力。如果想要了解更多可以打开看一下官方源码的行为细节。6. 用一个 Milvus 最小例子把Store的写入链路看顺Milvus 是一个向量数据库主要用于存储向量及其关联元数据并支持相似度检索。如果只讲概念还是容易飘。不如直接看一个最典型的组合外部EmbeddingMilvus 负责向量存储Indexer.Store统一完成写入packagemainimport(contextloggithub.com/cloudwego/eino/components/embeddinggithub.com/cloudwego/eino/schemagithub.com/cloudwego/eino-ext/components/indexer/milvus2github.com/milvus-io/milvus/client/v2/milvusclient)funcmain(){ctx:context.Background()// 这里假设 emb 已经提前初始化完成比如 千问 / OpenAI / Ark 等 embedding 组件varemb embedding.Embedder idx,err:milvus2.NewIndexer(ctx,milvus2.IndexerConfig{ClientConfig:milvusclient.ClientConfig{Address:addr,Username:username,Password:password,},Collection:kb_chunks,Dimension:1024,// 必须和 embedding 模型输出维度一致MetricType:milvus2.COSINE,IndexBuilder:milvus2.NewHNSWIndexBuilder().WithM(16).WithEfConstruction(200),Embedding:emb,})iferr!nil{log.Fatal(err)}docs:[]*schema.Document{{ID:chunk_001,Content:RAG 的第一步不是问模型而是先把文档变成可检索对象。,MetaData:map[string]any{source:rag_intro.md,chunk_no:1,},},}ids,err:idx.Store(ctx,docs)iferr!nil{log.Fatal(err)}log.Printf(stored ids%v,ids)}这段代码真正值得看的不是 Milvus 的参数怎么填而是职责分工业务层交给它的仍然是schema.Document向量生成能力通过Embedding挂进去Store统一把内容、向量、元数据写到后端返回的ids才是这次写入最终确认下来的结果也就是说业务层并没有直接面对“Milvus 的行结构”。它只是在说我有一批标准文档请把它们写成以后能被检索的样子。这才是Indexer抽象真正值钱的地方。7. 为什么说Indexer不只服务向量数据库如果你只看 Milvus 或 VikingDB很容易觉得Indexer就是“向量库接口”。可官方把 ES / OpenSearch 也放在Indexer下面这个信号其实非常强。来看 ES7 这种写法indexer,_:es7.NewIndexer(ctx,es7.IndexerConfig{Client:client,Index:kb_chunks,// 写入到 ES 的哪个索引// 把统一的 Document 转成 ES 里的字段结构DocumentToFields:func(ctx context.Context,doc*schema.Document)(map[string]es7.FieldValue,error){returnmap[string]es7.FieldValue{content:{Value:doc.Content,// 文档正文EmbedKey:content_vector,// 对 content 做向量化结果写到 content_vector 字段},source:{Value:doc.MetaData[source],// 普通元数据字段不做向量化},},nil},Embedding:emb,// 向量模型把指定字段文本转成向量})这段代码很能说明问题。这里的Indexer已经不是“往向量列里塞一个浮点数组”那么简单了而是在做两件事把Document映射成搜索引擎的字段结构决定哪些字段要向量化哪些字段按普通字段存储这说明Indexer抽象的是“检索后端的写入协议”不是“某一家向量数据库的专属写法”。换句话说Milvus / VikingDB 让你更容易看见vector。ES / OpenSearch 则提醒你别把Indexer只看成vector。它真正落的是Document - backend indexable representation这层转换。8. 放进 Chain / Graph 里你会发现 Indexer 也是正式组件很多人平时把Indexer单独调用一下就觉得这层已经懂了。其实不够。只有当你把它放进编排里才会更清楚它在 Eino 里的定位。// 在 Chain 中使用chain:compose.NewChain[[]*schema.Document,[]string]()chain.AppendIndexer(indexer)// 在 Graph 中使用graph:compose.NewGraph[[]*schema.Document,[]string]()graph.AddIndexerNode(indexer_node,indexer)这段代码表达的不是“语法还能这么写”。它真正表达的是Indexer从一开始就不是一个 helper。它和ChatModel、Tool、Retriever一样是能直接进入编排图的正式节点。这带来的工程价值非常实际你可以把文档加载、清洗、索引串成一条稳定流水线你可以通过compose.WithCallbacks统一观察整个入库过程你可以在更复杂的 Graph 里把不同写入策略拆成不同节点一旦你从“帮我写一下数据”切换到“它是编排节点”的视角Indexer的位置就完全不一样了。9.Callback和自定义实现到了生产环境中很多问题不是“能不能写进去”而是哪批文档写失败了哪一步失败的是 embedding 还是 backend 写入返回的ids和输入文档是否一一对应某次写入到底落到了哪个子索引这时候Callback的价值就出来了。官方给的回调输入输出很精妙typeCallbackInputstruct{Docs[]*schema.Document Extramap[string]any}typeCallbackOutputstruct{IDs[]stringExtramap[string]any}字段不多但刚好卡在写入侧最该观察的地方进来的是什么文档出去的是哪些 ID如果你自己实现一个Indexer真正该守住的顺序也很明确先收公共 option框架统一认的再收实现级 option 你自己需要的从ctx里拿 callback managerOnStart执行真实写入OnError / OnEnd一个更稳的骨架可以像下面这样写// MyIndexerOptions 是当前这个自定义 Indexer 的“实现级 option”。// 也就是只有 MyIndexer 自己认识和使用的参数。typeMyIndexerOptionsstruct{BatchSizeint// 批量写入时每批处理多少条MaxRetriesint// 写入失败时最多重试几次}// WithBatchSize 用来生成一个实现级 option。// 调用方可以在 Store(..., opts...) 时传入它覆盖默认批大小。funcWithBatchSize(sizeint)indexer.Option{returnindexer.WrapImplSpecificOptFn(func(o*MyIndexerOptions){o.BatchSizesize})}// Store 是 Indexer 对外暴露的统一写入入口。// 它做的不是“直接 insert”而是// 1. 收集通用 option// 2. 收集当前实现自己的 option// 3. 触发回调开始事件// 4. 执行真实写入// 5. 根据结果触发结束或错误回调func(i*MyIndexer)Store(ctx context.Context,docs[]*schema.Document,opts...indexer.Option)([]string,error){// 解析“公共 option”// 比如 SubIndexes、Embedding 这类所有 Indexer 都能理解的参数。commonOpts:indexer.GetCommonOptions(nil,opts...)// 解析“实现级 option”// 先给一个默认值再用调用方传进来的 opts 覆盖。implOpts:indexer.GetImplSpecificOptions(MyIndexerOptions{BatchSize:i.batchSize,},opts...)// 从 ctx 中拿到 callback manager。// 它负责记录这次 Store 的开始、结束和错误。cm:callbacks.ManagerFromContext(ctx)runInfo:callbacks.RunInfo{}// 通知回调系统这次写入开始了。// 这里把输入文档和一些额外上下文信息带进去便于日志、追踪和调试。ctxcm.OnStart(ctx,runInfo,indexer.CallbackInput{Docs:docs,Extra:map[string]any{sub_indexes:commonOpts.SubIndexes,batch_size:implOpts.BatchSize,},})// 执行真正的写入逻辑。// 这里会进入 doStore完成 embedding、字段映射、批量写入等动作。ids,err:i.doStore(ctx,docs,commonOpts,implOpts)iferr!nil{// 如果写入失败通知回调系统发生了错误。cm.OnError(ctx,runInfo,err)returnnil,err}// 如果写入成功通知回调系统结束并把最终写入成功的 IDs 带出去。cm.OnEnd(ctx,runInfo,indexer.CallbackOutput{IDs:ids,})returnids,nil}// doStore 是真正执行写入细节的地方。// Store 负责“流程控制”doStore 负责“实际干活”。func(i*MyIndexer)doStore(ctx context.Context,docs[]*schema.Document,commonOpts*indexer.Options,implOpts*MyIndexerOptions,)([]string,error){// 如果本次写入指定了 Embedding就先把文档内容转成向量。// 这样后续写入后端时就能把文本和向量一起存进去。ifcommonOpts.Embedding!nil{// 先提取所有文档的正文内容准备批量做 embedding。texts:make([]string,len(docs))forj,doc:rangedocs{texts[j]doc.Content}// 调用 embedding 模型把文本批量转成向量。vectors,err:commonOpts.Embedding.EmbedStrings(ctx,texts)iferr!nil{returnnil,err}// 把生成出来的向量挂回到每个 Document 上。forj,doc:rangedocs{doc.WithVector(vectors[j])}}// implOpts 里一般会继续参与下面的写入逻辑// 比如按 BatchSize 分批写、按 MaxRetries 做重试等。_implOpts// 这里继续做// - 批量写入// - 字段映射// - 分区/子索引路由// - 调用具体后端 SDK//// 最后返回后端确认写入成功的文档 ID 列表。return[]string{stored_doc_1},nil}这段骨架最重要的不是细节实现而是以下这几点公共 option 和实现级 option 分开处理callback 生命周期完整触发embedding 是“写入前可挂接能力”真正的 backend 写入逻辑被收敛在doStore这才是一个能进工程的Store形状。10. 总结用一句话总结Store的本质不是一次 insert而是“文档协议进入检索系统的统一写入入口”。其中有3点需要重视Indexer解决的是写入侧协议统一不是某家后端 SDK 的薄封装Store里可能同时发生字段映射、向量生成、分区路由、批量写入和回调触发ES / OpenSearch 的实现已经足够说明Indexer抽象的不是“向量库”而是“可检索后端”参考资料CloudWeGo Eino Indexer 使用说明CloudWeGo Eino components/indexer/interface.goCloudWeGo Eino components/indexer/option.go
AI 大模型落地系列｜Eino 组件核心篇：Indexer 背后，真正值得看懂的是 Store
声明本文数据源于官方文档与官方实现重点参考 Indexer 使用说明 以及eino/components/indexer/interface.go、eino/components/indexer/option.go相关源码。为什么很多人会用 Indexer却没真正看懂 Store1. 它的用处它又被误会在哪它能干嘛它最适合哪些场景它不能直接干嘛2. Indexer 不是“向量库 insert 封装”3. Indexer 在 RAG 入库链路里到底站在哪4. 接口只有一个方法但 Store 这个词一点都不简单5. 公共 Option 真正控制的不是“几个小参数”4.1 SubIndexes 更像逻辑分区不只是一个字符串数组4.2 Embedding 是写入时临时挂接的能力不是 Indexer 本体6. 用一个 Milvus 最小例子把 Store 的写入链路看顺7. 为什么说 Indexer 不只服务向量数据库8. 放进 Chain / Graph 里你会发现 Indexer 也是正式组件9. Callback 和自定义实现10. 总结参考资料很多人做知识库时前面那几步通常都知道怎么搞文档切块文本做 embedding向量库建 collection看起来好像已经差不多了。可真到落库这一步问题马上就来了文档正文怎么写进去向量是谁来生成、什么时候生成元数据和来源信息放哪里多个知识库、多个业务空间怎么分区写写进去以后怎么保证后面真的能被检索到这时候你就会发现Indexer这层不是可有可无。说白了Indexer本质上就是“把文档写成以后能被检索的样子”的组件。它不是只管塞一段文本进去。它更像是把文档、向量、元数据、子索引这些东西一起整理好再送进可检索后端。这也是为什么它在知识库入库、语义搜索底库构建、多知识库分区写入这些场景里特别关键。但很多人第一次看到Indexer还是会下意识地把它理解成“哦这不就是调一下 Milvus、VikingDB 或 ES 的写入接口吗”问题也恰恰出在这儿。因为如果事情真这么简单那我们明明已经有Embedding向量库 SDK搜索引擎写入接口为什么 Eino 还要单独设计一个Indexer这篇文章想讲清楚的就是这件事。如果你前面刚看过我上一篇 Document Loader 的文章那篇讲的是文档怎么进入[]*schema.Document这套统一协议。这一篇刚好接上下一站当文档已经变成标准协议后它到底怎么被写进“可检索系统”1. 它的用处它又被误会在哪先别急着看接口。如果一上来我就讲 API这样虽可以记住函数名但不利于大家理解它能解决什么问题。Indexer更适合先从“它是拿来干嘛的”讲起。它能干嘛放到最常见的知识库链路里Indexer主要做的是这些事1. 把文档写进可检索后端不是只写正文。而是把schema.Document这套统一协议整理成后端真正能存、以后也真正能查的样子。2. 把向量和元数据一起落进去很多场景不是只有文本内容。你还得把向量、来源、chunk 编号、业务标签这些信息一起写进去不然后面检索和追踪都会很难受。3. 处理逻辑子索引或知识库分区你可能不是只有一个知识库。多租户、多业务空间、多资料域这些场景都要求写入时就把路由和隔离想清楚。它最适合哪些场景最常见的就是这几类知识库入库语义搜索底库构建多知识库 / 多租户分区写入它不能直接干嘛这块也必须提前讲清楚。否则你会误会它的功能。Indexer很重要但它不是万能层。它不负责切块。文档怎么拆 chunk不是它的职责。它不负责生成最终回答。那是 LLM 干的事。它不负责读侧召回。怎么把内容查出来是Retriever的边界。所以更准确地说Embedding是把文本变成向量。Indexer是把内容写成可检索对象。Retriever是再把这些东西查出来。这三个环节如果混在一起后面链路虽然也能跑就会造成边界模糊耦合严重的场景。2.Indexer不是“向量库 insert 封装”这里先把结论摆出来Indexer不是一个“帮你调 Milvus / VikingDB / ES SDK”的小工具。它是 Eino 在写入侧给出的统一组件协议。这层协议真正收口的是三件事文档输入统一成[]*schema.Document写入行为统一成Store(ctx, docs, opts...)写入结果统一成[]string形式的ids这就意味着它关心的是“写入侧边界”不是某个后端产品自己的调用细节。所以你看Indexer时最好先把几个常见误会排掉它不负责切块。文档怎么切成 chunk是Loader / Parser后面的预处理问题不是Indexer本体。它不负责读侧召回。相似度搜索、过滤、排序、返回 topK那是Retriever的边界。它不等于Embedding。向量可以在写入时生成但生成向量这件事本身仍然是另一层能力。也正因为这三件事被拆开了Eino 才能同时挂住 Milvus、VikingDB、ES、OpenSearch 这些看起来很不像一家人的后端实现。如果它只是“向量库 insert 封装”那 ES / OpenSearch 这两类实现就会显得很别扭。可官方偏偏把它们也归在Indexer下面这恰恰说明Indexer抽象的不是“向量库”而是“可检索后端的写入入口”。一套可插拔的接口3.Indexer在 RAG 入库链路里到底站在哪很多人理解 RAG 时脑子里只有一句话“文档切块做 embedding丢进向量库。”这当然没错但如果你在 Eino 里写组件你最好把链路拆得再清楚一点Source - Loader / Parser - []*schema.Document - 切块 / 清洗 - Embedding / Field Mapping - Indexer.Store - Retriever - ChatModel这里最关键的是中间那段。Loader / Parser负责把不同来源的内容收口成标准Document。Indexer负责把这些Document写进后端让它以后能被查出来。而Retriever则负责真正把它们读出来。也就是说Loader / Parser管“东西从哪来、怎么解释”Indexer管“怎么写进去”Retriever管“怎么查出来”很多人之所以会把Indexer理解歪就是因为把“写进去”和“以后怎么查”混成了一件事。可在工程里这两件事差得很远。写入时你关心的是文档 ID 怎么处理向量何时生成元数据写到哪些字段逻辑分区、子索引、批量写入怎么做召回时你关心的是query 该怎么向量化topK 怎么取filter 怎么写score 怎么解释如果这两层边界不拆开最后很容易变成“能跑但组件职责已经糊了”。4. 接口只有一个方法但Store这个词一点都不简单官方核心接口其实非常短typeIndexerinterface{Store(ctx context.Context,docs[]*schema.Document,opts...Option)(ids[]string,errerror)}很多人第一次看到这个接口会觉得信息量不大。可它真正想收口的边界恰恰都藏在这几个参数里。先看ctx。在 Eino 里ctx从来都不只是取消信号。官方文档已经明确写了它还承担Callback Manager的传递。这意味着Store不是一个藏在角落里的工具函数而是一段可以被编排、被观察、被追踪的正式运行时行为。再看docs []*schema.Document。这很关键。Indexer吃进去的不是某家向量库自己的 row也不是某个搜索引擎专属的字段结构而是统一文档协议。这件事的价值在上一篇Document Loader里其实已经埋下了文档一旦被标准化成schema.Document后面的写入端就终于可以和“来源差异”解耦。最后看返回值ids []string。这块很多人会想当然地把它理解成“就是把doc.ID原样回给你”。但实际上ids更准确的意思是后端最终确认写入成功的文档标识。它可能是直接沿用你传进来的Document.ID后端生成的新 ID一次批量 upsert 之后真正生效的主键集合所以Store这个词千万别按数据库里那种“我插一行你回一个自增主键”的直觉去理解。在 Eino 语境里一次Store里可能同时发生文档字段映射向量生成批量写入子索引分流回调触发错误上抛这已经明显不是一句“insert 一下”能说清的事了。5. 公共 Option 真正控制的不是“几个小参数”官方给Indexer的公共 option 很克制typeOptionsstruct{SubIndexes[]string// 子索引/子分区这批文档要写到哪些逻辑分组里Embedding embedding.Embedder// 向量模型写入前用它把文本转成向量}funcWithSubIndexes(subIndexes[]string)Option// 设置子索引/分区funcWithEmbedding(emb embedding.Embedder)Option// 设置本次写入使用的向量生成器字段不多但信息量不小。4.1SubIndexes更像逻辑分区不只是一个字符串数组很多人第一次看SubIndexes会把它当成“顺手多传几个名字”。可如果你把它放回知识库场景里就会发现它更像逻辑分区入口。比如同一套物理后端里你可能会按下面这些维度做隔离不同知识库不同租户不同业务空间不同文档域这时SubIndexes的作用就不是“多一个参数”这么简单了。它更接近在同一个Indexer抽象之下把文档路由到不同的逻辑子索引或子分区。所以我更愿意把它理解成写入侧的 namespace / partition 入口而不是一个普通切片字段。4.2Embedding是写入时临时挂接的能力不是Indexer本体WithEmbedding更值得多看两眼。它说明什么说明 Eino 允许你在Store这一跳里临时指定“这批文档怎么向量化”。也就是说向量生成可以是Indexer初始化时配置好的默认能力本次调用临时覆盖进去的 embedder这就把“写入协议”和“向量模型选择”拆开了。而且还有一个容易被忽略的点。VikingDB 示例里官方给的是后端内建 embedding 配置EmbeddingConfig:volc_vikingdb.EmbeddingConfig{UseBuiltin:true,ModelName:bge-m3,UseSparse:true,},这恰恰说明Indexer可以挂接 embedding但它本身不等于 embedding。有些实现会走外部 embedder有些实现会直接利用后端内建能力。如果想要了解更多可以打开看一下官方源码的行为细节。6. 用一个 Milvus 最小例子把Store的写入链路看顺Milvus 是一个向量数据库主要用于存储向量及其关联元数据并支持相似度检索。如果只讲概念还是容易飘。不如直接看一个最典型的组合外部EmbeddingMilvus 负责向量存储Indexer.Store统一完成写入packagemainimport(contextloggithub.com/cloudwego/eino/components/embeddinggithub.com/cloudwego/eino/schemagithub.com/cloudwego/eino-ext/components/indexer/milvus2github.com/milvus-io/milvus/client/v2/milvusclient)funcmain(){ctx:context.Background()// 这里假设 emb 已经提前初始化完成比如 千问 / OpenAI / Ark 等 embedding 组件varemb embedding.Embedder idx,err:milvus2.NewIndexer(ctx,milvus2.IndexerConfig{ClientConfig:milvusclient.ClientConfig{Address:addr,Username:username,Password:password,},Collection:kb_chunks,Dimension:1024,// 必须和 embedding 模型输出维度一致MetricType:milvus2.COSINE,IndexBuilder:milvus2.NewHNSWIndexBuilder().WithM(16).WithEfConstruction(200),Embedding:emb,})iferr!nil{log.Fatal(err)}docs:[]*schema.Document{{ID:chunk_001,Content:RAG 的第一步不是问模型而是先把文档变成可检索对象。,MetaData:map[string]any{source:rag_intro.md,chunk_no:1,},},}ids,err:idx.Store(ctx,docs)iferr!nil{log.Fatal(err)}log.Printf(stored ids%v,ids)}这段代码真正值得看的不是 Milvus 的参数怎么填而是职责分工业务层交给它的仍然是schema.Document向量生成能力通过Embedding挂进去Store统一把内容、向量、元数据写到后端返回的ids才是这次写入最终确认下来的结果也就是说业务层并没有直接面对“Milvus 的行结构”。它只是在说我有一批标准文档请把它们写成以后能被检索的样子。这才是Indexer抽象真正值钱的地方。7. 为什么说Indexer不只服务向量数据库如果你只看 Milvus 或 VikingDB很容易觉得Indexer就是“向量库接口”。可官方把 ES / OpenSearch 也放在Indexer下面这个信号其实非常强。来看 ES7 这种写法indexer,_:es7.NewIndexer(ctx,es7.IndexerConfig{Client:client,Index:kb_chunks,// 写入到 ES 的哪个索引// 把统一的 Document 转成 ES 里的字段结构DocumentToFields:func(ctx context.Context,doc*schema.Document)(map[string]es7.FieldValue,error){returnmap[string]es7.FieldValue{content:{Value:doc.Content,// 文档正文EmbedKey:content_vector,// 对 content 做向量化结果写到 content_vector 字段},source:{Value:doc.MetaData[source],// 普通元数据字段不做向量化},},nil},Embedding:emb,// 向量模型把指定字段文本转成向量})这段代码很能说明问题。这里的Indexer已经不是“往向量列里塞一个浮点数组”那么简单了而是在做两件事把Document映射成搜索引擎的字段结构决定哪些字段要向量化哪些字段按普通字段存储这说明Indexer抽象的是“检索后端的写入协议”不是“某一家向量数据库的专属写法”。换句话说Milvus / VikingDB 让你更容易看见vector。ES / OpenSearch 则提醒你别把Indexer只看成vector。它真正落的是Document - backend indexable representation这层转换。8. 放进 Chain / Graph 里你会发现 Indexer 也是正式组件很多人平时把Indexer单独调用一下就觉得这层已经懂了。其实不够。只有当你把它放进编排里才会更清楚它在 Eino 里的定位。// 在 Chain 中使用chain:compose.NewChain[[]*schema.Document,[]string]()chain.AppendIndexer(indexer)// 在 Graph 中使用graph:compose.NewGraph[[]*schema.Document,[]string]()graph.AddIndexerNode(indexer_node,indexer)这段代码表达的不是“语法还能这么写”。它真正表达的是Indexer从一开始就不是一个 helper。它和ChatModel、Tool、Retriever一样是能直接进入编排图的正式节点。这带来的工程价值非常实际你可以把文档加载、清洗、索引串成一条稳定流水线你可以通过compose.WithCallbacks统一观察整个入库过程你可以在更复杂的 Graph 里把不同写入策略拆成不同节点一旦你从“帮我写一下数据”切换到“它是编排节点”的视角Indexer的位置就完全不一样了。9.Callback和自定义实现到了生产环境中很多问题不是“能不能写进去”而是哪批文档写失败了哪一步失败的是 embedding 还是 backend 写入返回的ids和输入文档是否一一对应某次写入到底落到了哪个子索引这时候Callback的价值就出来了。官方给的回调输入输出很精妙typeCallbackInputstruct{Docs[]*schema.Document Extramap[string]any}typeCallbackOutputstruct{IDs[]stringExtramap[string]any}字段不多但刚好卡在写入侧最该观察的地方进来的是什么文档出去的是哪些 ID如果你自己实现一个Indexer真正该守住的顺序也很明确先收公共 option框架统一认的再收实现级 option 你自己需要的从ctx里拿 callback managerOnStart执行真实写入OnError / OnEnd一个更稳的骨架可以像下面这样写// MyIndexerOptions 是当前这个自定义 Indexer 的“实现级 option”。// 也就是只有 MyIndexer 自己认识和使用的参数。typeMyIndexerOptionsstruct{BatchSizeint// 批量写入时每批处理多少条MaxRetriesint// 写入失败时最多重试几次}// WithBatchSize 用来生成一个实现级 option。// 调用方可以在 Store(..., opts...) 时传入它覆盖默认批大小。funcWithBatchSize(sizeint)indexer.Option{returnindexer.WrapImplSpecificOptFn(func(o*MyIndexerOptions){o.BatchSizesize})}// Store 是 Indexer 对外暴露的统一写入入口。// 它做的不是“直接 insert”而是// 1. 收集通用 option// 2. 收集当前实现自己的 option// 3. 触发回调开始事件// 4. 执行真实写入// 5. 根据结果触发结束或错误回调func(i*MyIndexer)Store(ctx context.Context,docs[]*schema.Document,opts...indexer.Option)([]string,error){// 解析“公共 option”// 比如 SubIndexes、Embedding 这类所有 Indexer 都能理解的参数。commonOpts:indexer.GetCommonOptions(nil,opts...)// 解析“实现级 option”// 先给一个默认值再用调用方传进来的 opts 覆盖。implOpts:indexer.GetImplSpecificOptions(MyIndexerOptions{BatchSize:i.batchSize,},opts...)// 从 ctx 中拿到 callback manager。// 它负责记录这次 Store 的开始、结束和错误。cm:callbacks.ManagerFromContext(ctx)runInfo:callbacks.RunInfo{}// 通知回调系统这次写入开始了。// 这里把输入文档和一些额外上下文信息带进去便于日志、追踪和调试。ctxcm.OnStart(ctx,runInfo,indexer.CallbackInput{Docs:docs,Extra:map[string]any{sub_indexes:commonOpts.SubIndexes,batch_size:implOpts.BatchSize,},})// 执行真正的写入逻辑。// 这里会进入 doStore完成 embedding、字段映射、批量写入等动作。ids,err:i.doStore(ctx,docs,commonOpts,implOpts)iferr!nil{// 如果写入失败通知回调系统发生了错误。cm.OnError(ctx,runInfo,err)returnnil,err}// 如果写入成功通知回调系统结束并把最终写入成功的 IDs 带出去。cm.OnEnd(ctx,runInfo,indexer.CallbackOutput{IDs:ids,})returnids,nil}// doStore 是真正执行写入细节的地方。// Store 负责“流程控制”doStore 负责“实际干活”。func(i*MyIndexer)doStore(ctx context.Context,docs[]*schema.Document,commonOpts*indexer.Options,implOpts*MyIndexerOptions,)([]string,error){// 如果本次写入指定了 Embedding就先把文档内容转成向量。// 这样后续写入后端时就能把文本和向量一起存进去。ifcommonOpts.Embedding!nil{// 先提取所有文档的正文内容准备批量做 embedding。texts:make([]string,len(docs))forj,doc:rangedocs{texts[j]doc.Content}// 调用 embedding 模型把文本批量转成向量。vectors,err:commonOpts.Embedding.EmbedStrings(ctx,texts)iferr!nil{returnnil,err}// 把生成出来的向量挂回到每个 Document 上。forj,doc:rangedocs{doc.WithVector(vectors[j])}}// implOpts 里一般会继续参与下面的写入逻辑// 比如按 BatchSize 分批写、按 MaxRetries 做重试等。_implOpts// 这里继续做// - 批量写入// - 字段映射// - 分区/子索引路由// - 调用具体后端 SDK//// 最后返回后端确认写入成功的文档 ID 列表。return[]string{stored_doc_1},nil}这段骨架最重要的不是细节实现而是以下这几点公共 option 和实现级 option 分开处理callback 生命周期完整触发embedding 是“写入前可挂接能力”真正的 backend 写入逻辑被收敛在doStore这才是一个能进工程的Store形状。10. 总结用一句话总结Store的本质不是一次 insert而是“文档协议进入检索系统的统一写入入口”。其中有3点需要重视Indexer解决的是写入侧协议统一不是某家后端 SDK 的薄封装Store里可能同时发生字段映射、向量生成、分区路由、批量写入和回调触发ES / OpenSearch 的实现已经足够说明Indexer抽象的不是“向量库”而是“可检索后端”参考资料CloudWeGo Eino Indexer 使用说明CloudWeGo Eino components/indexer/interface.goCloudWeGo Eino components/indexer/option.go
