Eino RAG Pipeline：生产级索引、检索与评测

核心摘要

生产级 RAG 是两条分别版本化的 Pipeline：离线链路把受治理的源文档转成可溯源索引，在线链路在生成前检索经过授权的证据。Eino v0.9.21 为加载、转换、Embedding、索引和检索提供稳定组件接口。可靠性来自保留 Provenance、固定索引兼容契约、分开评测 Retrieval 与 Generation，并在证据不足或无权访问时安全拒答。

目录

1. 两条 RAG Pipeline
2. Eino v0.9.21 组件契约
3. 设计索引契约
4. 文档切分与溯源
5. 在线检索与精排
6. 可运行的 Eino 参考实现
7. 权限与 Prompt Injection 边界
8. 评测与发布门禁
9. 故障定位
10. 常见问题
11. 一手资料

两条 RAG Pipeline

检索增强生成（Retrieval-Augmented Generation）把模型参数与可检索的外部证据结合。RAG 原始论文将其描述为参数记忆与非参数记忆的组合；生产系统还必须保证这些记忆是最新、已授权、可归因且可测量的。

离线链路负责语料正确性，在线链路负责查询正确性。把二者混在一个函数中，会显著增加重试、访问控制和回滚难度。

Eino v0.9.21 组件契约

Eino Core 定义了一组窄接口，而不是一个包办所有环节的 RAG 对象。

schema.Document 包含 ID、Content 和 MetaData；辅助方法会把 Retrieval Score 与 Sub-index 信息保存在 Metadata 中。Eino Retriever 文档要求结果按相关性排序；WithScoreThreshold 负责过滤，不会自动完成排序。

这些是核心契约。具体 Vector Database、Loader、Parser 和 Model Client 属于独立模块或应用实现。每个模块都要独立固定版本，不能从旧教程复制 Import Path 后假设它在当前 Revision 中仍存在。

设计索引契约

只有完整编码契约一致时，索引才兼容。

每个 Index Version 都应保存以下 Manifest：

即使新旧模型的向量维度相同，原地替换 Embedding Model 也不安全。应该创建新索引，运行 Shadow Evaluation，切换 Alias 或 Routing Key，并在新版本被验证前保留回滚能力。

写入语义

使用稳定的 Document ID 与 Chunk ID，让重复索引天然幂等。一种实用身份构造是：

source identity + source revision + structural location + normalized content hash

删除传播也是索引契约的一部分。源文档被删除或撤销权限后，不能无限期残留在检索索引中。

文档切分与溯源

Chunking 是需要评测的检索决策，不是一个固定 Token 数。

按结构边界切分通常比任意窗口更能保留语义。标题、段落、代码块、表格和列表组不应被破坏。只有评测集证明 Overlap 能恢复跨边界证据时才加入；Overlap 同时会增加重复候选、存储和 Prompt 成本。

每个 Chunk 至少携带：

• 不可变 Chunk ID 与父 Document ID
• Canonical Source URI 与 Source Revision
• Title 与 Structural Path
• 可用时的 Byte、Line、Page 或 Timestamp 位置
• Tenant 与对象级 Authorization 字段
• Language 与 Content Type
• Ingestion Timestamp 与 Index Version

Provenance 是引用、删除、故障分析和新鲜度判断的前提。入库时丢失的来源信息，不能靠最终答案补回来。

在线检索与精排

检索应拆成可独立测试的阶段：

1. 标准化 Query，但保留有意义的精确标识符。
2. 从可信身份解析 Tenant、User 与 Object Filter。
3. 召回较宽的候选集合。
4. 可选地融合词法与稠密排序。
5. 对有限候选进行 Rerank。
6. 去除近重复 Chunk。
7. 在 Token Budget 内构造带 Source ID 的上下文。
8. 证据不足时拒答，否则生成答案。

稠密、稀疏与混合检索

Dense Retrieval 擅长语义相似，但可能漏掉错误码、Product ID 或专有名称；Sparse Retrieval 保留词法匹配，但可能漏掉改写表达。Hybrid Retrieval 可以结合两者，但不保证必然更好。

Reciprocal Rank Fusion 是不依赖原始分数刻度的简单基线：

RRF(document) = sum(1 / (k + rank_in_list))

Fusion Constant 与 Candidate Depth 必须在冻结 Query Set 上选择。Corpus、Filter 或 Relevance Label 不同的系统不能直接比较。

精排

Rerank 会对较小候选集合投入更多计算。在 Eino 中，它可以表示为 document.Transformer，也可以是应用级阶段；具体模型或后端 API 不属于核心 Retriever 接口。

BEIR 研究显示，不同检索方法的跨域表现差异明显，Rerank 可以提高 Zero-shot 质量，但计算成本更高。这个结论支持按代表性 Domain 评测，而不是声称某一种 Retriever 永远最佳。

可运行的 Eino 参考实现

下面的确定性程序使用内存词法分数实现 Eino 的 Indexer 与 Retriever 接口。它不是生产搜索引擎，而是一个可编译的契约测试，覆盖 ID 分配、Metadata 保留、相关性排序、TopK、ScoreThreshold 和取消。

package main

import (
	"context"
	"fmt"
	"log"
	"sort"
	"strconv"
	"strings"
	"sync"
	"unicode"

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

type memoryStore struct {
	mu   sync.RWMutex
	docs map[string]*schema.Document
}

func newMemoryStore() *memoryStore {
	return &memoryStore{docs: make(map[string]*schema.Document)}
}

func cloneDocument(doc *schema.Document) *schema.Document {
	metadata := make(map[string]any, len(doc.MetaData))
	for key, value := range doc.MetaData {
		metadata[key] = value
	}
	return &schema.Document{
		ID:       doc.ID,
		Content:  doc.Content,
		MetaData: metadata,
	}
}

func (s *memoryStore) Store(
	ctx context.Context,
	docs []*schema.Document,
	_ ...indexer.Option,
) ([]string, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	ids := make([]string, 0, len(docs))
	for i, doc := range docs {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		copy := cloneDocument(doc)
		if copy.ID == "" {
			copy.ID = "doc-" + strconv.Itoa(i+1)
		}
		s.docs[copy.ID] = copy
		ids = append(ids, copy.ID)
	}
	return ids, nil
}

func terms(text string) map[string]struct{} {
	result := make(map[string]struct{})
	for _, token := range strings.FieldsFunc(
		strings.ToLower(text),
		func(r rune) bool {
			return !unicode.IsLetter(r) && !unicode.IsDigit(r)
		},
	) {
		result[token] = struct{}{}
	}
	return result
}

func overlapScore(query, content string) float64 {
	queryTerms := terms(query)
	if len(queryTerms) == 0 {
		return 0
	}
	contentTerms := terms(content)
	matches := 0
	for term := range queryTerms {
		if _, ok := contentTerms[term]; ok {
			matches++
		}
	}
	return float64(matches) / float64(len(queryTerms))
}

func (s *memoryStore) Retrieve(
	ctx context.Context,
	query string,
	opts ...retriever.Option,
) ([]*schema.Document, error) {
	options := retriever.GetCommonOptions(nil, opts...)
	topK := 3
	if options.TopK != nil && *options.TopK > 0 {
		topK = *options.TopK
	}
	threshold := 0.0
	if options.ScoreThreshold != nil {
		threshold = *options.ScoreThreshold
	}

	s.mu.RLock()
	defer s.mu.RUnlock()

	results := make([]*schema.Document, 0, len(s.docs))
	for _, doc := range s.docs {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		score := overlapScore(query, doc.Content)
		if score >= threshold {
			results = append(results, cloneDocument(doc).WithScore(score))
		}
	}
	sort.Slice(results, func(i, j int) bool {
		return results[i].Score() > results[j].Score()
	})
	if len(results) > topK {
		results = results[:topK]
	}
	return results, nil
}

func main() {
	ctx := context.Background()
	store := newMemoryStore()

	_, err := store.Store(ctx, []*schema.Document{
		{
			ID:      "timeouts",
			Content: "Propagate one context deadline to model and tool calls.",
			MetaData: map[string]any{
				"source": "runbook.md",
				"tenant": "acme",
			},
		},
		{
			ID:      "retries",
			Content: "Retry only transient and idempotent operations.",
			MetaData: map[string]any{
				"source": "reliability.md",
				"tenant": "acme",
			},
		},
	})
	if err != nil {
		log.Fatal(err)
	}

	docs, err := store.Retrieve(
		ctx,
		"How should tool calls use a deadline?",
		retriever.WithTopK(2),
		retriever.WithScoreThreshold(0.2),
	)
	if err != nil {
		log.Fatal(err)
	}
	for _, doc := range docs {
		fmt.Printf(
			"id=%s score=%.2f source=%s\n",
			doc.ID,
			doc.Score(),
			doc.MetaData["source"],
		)
	}
	// Output: id=timeouts score=0.43 source=runbook.md
}

模块依赖固定为：

require github.com/cloudwego/eino v0.9.21

生产后端还应支持 Index 与 Sub-index Option，在打分前执行权限过滤，定义一致性行为，并批量执行网络操作。同一组 Fixture 可以复用于不同实现的 Contract Test。

权限与 Prompt Injection 边界

Retrieval 是数据访问操作，绝不能让模型决定适用哪个 Tenant 或 Object Filter。

正确顺序是：

authenticated identity
  -> server-side policy decision
  -> mandatory retrieval filters
  -> candidate documents
  -> content safety and injection handling
  -> model context

即使内容来自内部知识库，Document 也属于不可信输入。检索文本可能包含企图覆盖系统策略或外传数据的指令。必须保持“证据”和“指令”的边界，独立限制 Tool，并且绝不能根据 Corpus 中的文本授予访问权限。

Cache Key 必须包含所有会改变可见性或输出的字段：Tenant、Authorization Scope、Locale、Corpus/Index Version、Retrieval Configuration 与 Query Normalization Version。缺少策略维度的共享缓存会造成跨租户泄漏。

评测与发布门禁

先评测 Retrieval，再评测 Generation。否则流畅的模型可能掩盖薄弱检索，差答案也可能在没有证据时被错误归因给 Retriever。

检索集

每条 Query 至少保存：

• 相关 Document 与 Chunk ID
• 存在部分相关时的 Relevance Grade
• 预期 Tenant 与 Policy Filter
• Unanswerable 或 No-access 标签
• 重要精确标识符

按 Language、Domain、Document Age 与 Query Type 报告 Recall@K、MRR 和 nDCG。始终保留词法基线；BEIR 已说明为什么不能假设 Dense Retrieval 在所有 Domain 都占优。

端到端数据集

需要衡量：

• 答案正确性
• Citation Precision 与 Completeness
• 每个关键结论是否有证据支持
• 缺少证据时是否正确拒答
• 是否抵抗检索文档中的指令
• 延迟、模型 Token、Candidate Count 与 Reranker Cost

新的 Chunker、Embedding、Retriever 或 Reranker 必须先通过 Shadow Evaluation 与 Canary Check。条件允许时，一次只改变一个主要检索变量。

故障定位

使用受保护、有限留存的诊断存储保存 Retrieval Candidate 与 Score。排查 Bad Case 时不能只看最终答案。

常见问题

Eino RAG Pipeline 应该使用 Chain 还是 Graph？

固定线性流程使用 Chain；需要按 Query Type 分支、并行检索，或需要 Loop 与 State 时使用 Graph。两种方式都能复用相同组件接口。

必须使用向量数据库吗？

不必须。小型或高度结构化语料可能更适合 SQL、倒排索引或精确查找。本文示例刻意从词法实现开始，让契约和基线都清晰可见。

Chunk 应该多大？

不存在通用 Token 数。应在真实问题上比较结构化切分与固定窗口，再评估 Retrieval Quality、Duplicate Rate、Context Cost 与 Citation Precision。Tokenizer 与 Overlap 必须写入 Index Manifest。

什么时候应该增加 Rerank？

只有第一阶段召回率足够、但排序质量较差，并且质量收益能够覆盖延迟和成本时才增加。Reranker 无法找回第一阶段没有召回的文档。

总结

Eino 在 RAG 中的核心价值是拆分契约：Load、Transform、Embed、Index、Retrieve，然后 Generate。生产质量取决于流经这些契约的内容是否正确，包括版本化语料、完整 Provenance、强制权限、兼容 Embedding、可测 Retrieval 和安全灰度。先建立可复现基线，再让评测结果决定是否增加复杂度。

相关资源

• Eino 核心组件：ChatModel、Tool 与 Retriever
• RAG 与微调选型
• 混合检索术语
• Embedding 术语
• RAG 术语

一手资料

• Eino v0.9.21 Retriever API
• Eino v0.9.21 Indexer API
• Eino v0.9.21 Document API
• CloudWeGo Retriever 指南
• Retrieval-Augmented Generation 原始论文
• BEIR 异构检索基准论文