使用RAG构建高质量知识库（四）- 数据检索

咱们的知识库，经过清洗、分块、向量化后，已经存入向量数据库中了。那么如何检索和召回这些知识？

今天就来聊聊RAG的检索策略。

检索前优化

如果想要达到比较好的检索效果，首先需要对检索的query进行优化。常见的优化策略如下：

原始问题改写

通常在检索知识库之前，会需要调用大模型，结合上下文对用户的问题进行改写和优化。

# 原始对话
Q: 请给我推荐一门编程语言
A: python
Q: 我该如何开始？
----
# 说明
如果直接使用"我该如何开始"去检索知识库
显然效果是很不好的

# 基于上下文进行问题改写
Q: 我应该如何开始学习python?

# 说明
根据上下文改写后的问题，具备了比较好的语义完整度。

Multi-Query

将单个查询拓展为多个相关的问题进行查询，从而丰富上下文内容的多样性和覆盖范围。

# 用户问题
Q: 我想学习PYTHON

# 提示词
你是一个AI语言模型助手。
你的任务是针对给定的用户问题生成五个不同版本的表述，以便从向量数据库中检索相关文档。
通过对用户问题生成多种角度的表述，你的目标是帮助用户克服基于距离的相似性搜索的一些局限性。
将这些替代问题用换行符分隔开。
原始问题：{{ question }}

# 改写后的问题
Q1: 如何高效掌握Python编程？
Q2: 学习Python的最佳路径是什么？
Q3: Python入门教程推荐有哪些？
Q4: 从零开始学Python应该怎么做？
Q5: Python编程技能提升方法有哪些？
-----
使用改写后的5个问题，分别检索知识库
得到5组不同的检索结果

最终再通过大模型把5种结果进行筛选合并
生成最佳答案

Sub-Question

通过把一个复杂的问题，拆解成若干个简单的子问题，分步进行检索，最终回答用户的问题。

# 用户原始问题
Q: Coze和dify有啥区别？

# 改写后的问题
Q1: Coze的基本介绍
A1: Coze 是字节跳动旗下...

Q2: Dify基本介绍
A2: Dify 是一个开源的LLM应用平台...

Q3: Coze和Dify有啥区别？
A3: 基于上面的两个检索结果生成最佳答案

以上只列举了比较常见的优化策略，实际上检索词的优化会很大程度上决定后续检索的质量。

实际生产场景需要根据情况进行定向优化， 以达到最佳的检索效果。

知识检索

经过优化后的检索词，已经具备较好检索效果了。接下来就是要通过检索词把对应的知识片段找出来。

关键词检索

这个是传统的检索策略，使用检索词对分段原文或者分段标签进行模糊匹配查询。

优点是能够精准匹配关键词，缺点是无法理解同义词、近义词。

# 知识原文
北京故宫中存放了超过180万件珍贵文物。
# 关键词
北京、故宫、文物

# 用户问题
Q: 故宫中有多少件文物？
# 匹配了关键词 "故宫"和"文物"
# 能够召回原文片段
A: 北京故宫中存放了超过180万件珍贵文物。

# 但问题如果变成
Q:紫禁城中有多少件宝贝？
# 这时候由于无法匹配关键词，召回就失败了
A: 不知道

语义检索

语义检索就是通过把检索词和待检索文章都转换成密集向量后再进行检索。

语义上越接近的词，在向量数据库中的距离就越接近

不了解的同学也可以去看上一篇：

使用RAG构建高质量知识库（三）- 数据嵌入

核心代码实现，这里用到了openai-sdk、Milvus向量数据库、阿里text-embedding-v4嵌入模型等。

# Step1: 检索词向量化
constsearch_vector =awaitembedding(query);
// 调用Embedding Model进行向量化
constembedding =async(text) => {
constresponse =awaitopenai.embeddings.create({
 model:"text-embedding-v4",
 input: text,
 });
returnresponse.data[0].embedding;
};

# Step2: 密集向量检索
constvector_res =awaitmilvusService.search(search_vector,3);
# 查询向量数据库
asyncsearch(vector, limit =10) {
 returnawaitthis.client.search({
   collection_name:'test',
   vector: vector,
   limit: limit,
   output_fields: ['id','block_id','name'],
   metric_type:'L2',
   params: {nprobe:10}
  });
}

检索结果如下：

# 检索词
`The founding time of Snow Beer`

# 检索词向量化
[
  0.03072373941540718, -0.010284383781254292, -0.006375404074788094,
 -0.06936439871788025, 0.0009934211848303676, -0.04058944061398506,
 ...924more items
]

# 密集向量检索结果
[
 {
 score:0.46093612909317017,
 block_id:'94',
 name:'Below is a detailed introduction to Snow Beer:',
 id:'460215000564761333'
 },
 {
 score:0.517002284526825,
 block_id:'95',
 name:'1. Brand History & Development Origins: Snow Beer traces its roots to Shenyang Beer Factory, established in 1957.',
 id:'460215000564761323'
 },
 {
 score:0.790127158164978,
 block_id:'94',
 name:"Snow Beer is one of China's most renowned beer brands, owned by China Resources Snow Breweries (China) Co., Ltd. (CR Snow), a joint venture between the state-owned China Resources Group and the global beer giant SABMiller.",
 id:'460215000564761331'
 }
]

要说明的是采用L2欧氏距离作为度量单位，值越小，相似度越高。

其他的度量单位:

度量单位	说明	取值范围
L2(欧氏距离)	值越小表示相似度越高。	[0, ∞)
IP(内积)	值越大，表示相似度越高。	[-1, 1]
COSINE(余弦相似度)	数值越大，表示相似度越高。	[-1, 1]
JACCARD	值越小，表示相似度越高。	[0, 1]
MHJACCARD	根据 MinHash 签名位估算 Jaccard 相似度；距离越小 = 越相似	[0, 1]
HAMMING	值越小表示相似度越高。	[0，dim(向量)
BM25	根据术语频率、反转文档频率和文档规范化对相关性进行评分。	[0, ∞)

全文检索

全文检索通过BM25算法或者SPLADE预训练模型把文档转换为稀疏变量。通过TF词频、IDF逆文档频率等统计算法指标，“猜”出文档和检索词之间的相关性。

核心代码实现：

# query不需要向量化，所以也不消耗token
constquery ='The founding time of Snow Beer';
# 直接调用向量数据库进行查询
constsparse_res =awaitmilvusService.search(query,3);

asyncsearch(query, limit =10) {
 returnawaitthis.client.search({
   collection_name:'test',
   anns_field:'sparse_vector',
   limit: limit,
   data: [query],
   metric_type:'BM25',
   output_fields: ['id','block_id','name']
  }); 
}

检索结果：

# 检索词
'The founding time of Snow Beer'

# 稀疏向量检索结果:
[
 {
 score:4.728769302368164,
 block_id:'94',
 name:"Snow Beer is one of China's most renowned beer brands, owned by China Resources Snow Breweries (China) Co., Ltd. (CR Snow), a joint venture between the state-owned China Resources Group and the global beer giant SABMiller.",
 id:'460215000564761331'
 },
 {
 score:3.9664859771728516,
 block_id:'108',
 name:'Domestic Dominance: Snow has been the world’s top-selling single beer brand since 2006, with 2022 sales reaching ~11 billion liters (30% of China’s market share).',
 id:'460215000564761291'
 },
 {
 score:3.210528612136841,
 block_id:'100',
 name:'Brave the World: Flagship product for mass markets, known for its slogan "Snow Beer, Brave the World."',
 id:'460215000564761309'
 }
]

这里度量单位只能只用BM25，值越大，相关性越高。一般情况下BM25就可以替代传统的关键词检索来进行精确匹配了。当然也可以结合使用。

混合检索

简单来说就是分别使用全文检索和语义检索得到结果，再通过一定的策略进行结果聚合。

当然一般的向量数据库都有封装好的方法可以直接调用。

例如Milvus使用例子：

consthybrid_res =awaitmilvusService.search(search_vector, query,10);

asynchybrid_search(vector, query, limit =10) {
 returnawaitthis.client.hybridSearch({
   collection_name:'test',
   limit: limit,
   output_fields: ['id','block_id','name'],
   data: [{
     // 密集向量检索
     anns_field:'vector',
     data: [vector],
     metric_type:'L2',
     params: {nprobe:10}
    }, {
     // 稀疏向量检索
     anns_field:'sparse_vector',
     data: [query],
     metric_type:'BM25'
    }],
   // 排序策略
   rerank: {
     strategy:"rrf",// 使用RRF(Reciprocal Rank Fusion)策略
     params: {
      k:60,// RRF参数k，默认60
      },
    }
  })
}

混合检索后的结果通过RRF或者Weighted策略进行聚合。

那么RRF和Weight又是啥？

重排序

上面介绍了几种常见的检索策略。实际应用场景下，基本上都会使用多路召回分别使用各种策略检索得到结果。然后通过重排序进行优中选优。

有点像足球世界杯，先通过小组赛进行海选（各种策略分别进行海选召回），然后各个小组的优胜者再进行淘汰赛决出冠军（重排序）。

RRF（Reciprocal Rank Fusion）

这是一种仅使用每种结果的排名进行融合排序的算法。

一个例子：

你和你的朋友都在找一本丢失的书

每个朋友都有一个针对“书在哪里的猜测排名”

人员 书架 床下 桌上
朋友A 1 2 3
朋友B 3 1 2
朋友C 2 1 3

那么，针对每一个地方：

桌上：1/3 + 1/2 + 1/3 = 1.16
床下: 1/2 + 1/1 + 1/1 = 2.5
书架: 1/1 + 1/3 + 1/2 = 1.83

所以RRF算法在考虑了不同朋友的猜测后
得出书最可能在床下（2.5）

这种算法的好处是只考虑不同策略的排名，而完全不考虑分数，避免分数之间的差异化。

Weighted加权策略算法

这种算法通过给每个分路设置一个权重值，然后根据每个分路的算法，对打分进行归一化处理（不同分数的尺度可能是不同的），然后加权计算最终结果。

一个简单的算法例子：

/**
* 合并结果并应用加权排序
*@param{array}resultsList 各个搜索的结果列表
*@param{number[]}weights 权重数组
*@param{number}limit 最终返回结果数量
*@returns{object}格式化后的搜索结果
*/
functionmergeAndWeightResults(resultsList, weights, limit){
// 结果合并表 {id: {data: 文档数据, scores: [各搜索的分数]}}
constmerged = {};

// 归一化权重
constsumWeights = weights.reduce((sum, w) =>sum + w,0);
constnormalizedWeights = weights.map(w=>w / sumWeights);

// 收集所有结果并记录各搜索的分数
 resultsList.forEach((results, searchIndex) =>{
  results.results.forEach(item=>{
  if(!merged[item.id]) {
    merged[item.id] = {
    data: item,
    scores:newArray(resultsList.length).fill(0)
    };
   }
   merged[item.id].scores[searchIndex] = item.score;
  });
 });

// 计算加权分数并排序
constweightedResults =Object.values(merged).map(item=>{
 // 计算加权分数
 letweightedScore =0;
 for(leti =0; i < normalizedWeights.length; i++) {
   weightedScore += normalizedWeights[i] * normalizeScore(item.scores[i], i);
  }
 
 return{
   ...item.data,
   weightedScore,// 加权后的综合分数
  originalScores: item.scores// 保留原始分数供调试
  };
 });

// 按加权分数降序排序
 weightedResults.sort((a, b) =>b.weightedScore - a.weightedScore);

// 返回格式与Milvus原生搜索一致
return{
 status: {error_code:"Success",reason:""},
 results: weightedResults.slice(0, limit),
 recalls: resultsList.map(r=>r.recalls?.[0] ||0)
 };
}

/**
* 分数归一化处理（不同搜索的分数可能尺度不同）
*@param{number}score 原始分数
*@param{number}searchIndex 搜索索引
*@returns{number}归一化后的分数
*/
functionnormalizeScore(score, searchIndex){
// 实际应用中应根据不同搜索类型调整归一化方法
// 这里使用简单的sigmoid归一化示例
return1/ (1+Math.exp(-score));
}

Rerank Model

上面的两种算法都不消耗token，优点是简单速度快。但缺点都是其中不够精准。

所以现在主流的方案都是先通过RRF或者加权算法归并一轮，再调用Rerank Model进一步进行筛选。

咱们这里选用的是阿里的gte-rerank-v2模型进行演示。

直接上代码：

// rerank方法，直接使用https的方式
constrerank =async(query, documents) => {
 constresponse =awaitfetch('https://dashscope.aliyuncs.com/api/v1/services/rerank/text-rerank/text-rerank', {
  method:'OST',
  headers: {
   'Authorization':`Bearer${openai.apiKey}`,
   'Content-Type':'application/json'
   },
  body:JSON.stringify({
   model:'gte-rerank-v2',
   input: {
    query: query,
    documents: documents
    },
   parameters: {
    top_n:1
    }
   })
  });
 returnawaitresponse.json();
};
// 调用重排序算法
constrerankRes =awaitrerank(query, allResults.map(item=>item.name));
// 找到最优解
consttopIndex = rerankRes?.output?.results?.[0]?.index ??0;
console.log('重排序结果:', allResults[topIndex])

结语

以上介绍的都是一些常见的优化策略，掌握这些大多数的文档都已经能够很好的进行召回了。但是实际生产场景，依然会面临不小的挑战。

更希望大家掌握的是调优的思路，当面临实际复杂场景时，可以生成自己的一套优化策略，定向进行知识库的优化效果更加。