RAG召回质量翻倍的两个核心技术：我是这样解决 - 链载Ai

最近在优化公司的知识问答系统时，遇到了一个让人头疼的问题：明明知识库里有相关内容，但LLM总是回答"我不知道"或者答非所问。经过深入分析发现，问题出在RAG的召回环节——检索到的文档片段要么不够相关，要么上下文支离破碎。

经过一番折腾，我找到了两个特别有效的解决方案：索引扩展和Small-to-Big策略。今天就来分享一下这两个技术的原理和具体实现，希望能帮到有同样困扰的朋友。

问题的根源：单一检索的局限性

传统RAG系统通常只用一种方式检索：把查询和文档都转成向量，然后计算相似度。这种方法简单直接，但问题不少：

为了解决这些问题，我开始研究多路召回和层级索引的方案。

解决方案一：索引扩展——"多条腿走路"

索引扩展的核心思想是：既然单一检索有局限，那就用多种方式检索，然后把结果融合起来。就像投资要分散风险一样，检索也要分散"召回风险"。

技术架构

具体实现

importspacyfromsklearn.feature_extraction.textimportTfidfVectorizerimportjiebaimportreclassDiscreteIndexer:def__init__(self):#加载中文NER模型self.nlp=spacy.load("zh_core_web_sm")self.tfidf=TfidfVectorizer(max_features=1000,stop_words='english')defextract_keywords(self,text,top_k=10):"""提取关键词"""#使用jieba分词words=jieba.analyse.extract_tags(text,topK=top_k,withWeight=True)return[wordforword,weightinwords]defextract_entities(self,text):"""提取命名实体"""doc=self.nlp(text)entities=[]forentindoc.ents:entities.append({'text':ent.text,'label':ent.label_,'start':ent.start_char,'end':ent.end_char})returnentitiesdefbuild_discrete_index(self,documents):"""构建离散索引"""index=[]fori,docinenumerate(documents):keywords=self.extract_keywords(doc['text'])entities=self.extract_entities(doc['text'])index.append({'doc_id':doc['id'],'text':doc['text'],'keywords':keywords,'entities':[e['text']foreinentities],'entity_details':entities})returnindex

fromsentence_transformersimportSentenceTransformerimportnumpyasnpfromtypingimportList,DictclassMultiVectorIndexer:def__init__(self):#加载多个embedding模型self.models={'bge':SentenceTransformer('BAAI/bge-large-zh-v1.5'),'text2vec':SentenceTransformer('shibing624/text2vec-base-chinese'),'multilingual':SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')}defencode_documents(self,documentsist[str])->Dict[str,np.ndarray]:"""使用多个模型编码文档"""embeddings={}formodel_name,modelinself.models.items():print(f"正在使用{model_name}编码文档...")embeddings[model_name]=model.encode(documents)returnembeddingsdefsearch_single_model(self,query:str,model_name:str,embeddings:np.ndarray,top_k:int=5):"""在单个模型的向量空间中搜索"""query_embedding=self.models[model_name].encode([query])#计算余弦相似度similarities=np.dot(embeddings,query_embedding.T).flatten()similarities=similarities/(np.linalg.norm(embeddings,axis=1)*np.linalg.norm(query_embedding))#获取top_k结果top_indices=np.argsort(similarities)[::-1][:top_k]results=[(idx,similarities[idx])foridxintop_indices]returnresults

classEnsembleRetriever:def__init__(self,discrete_indexer,multi_vector_indexer):self.discrete_indexer=discrete_indexerself.multi_vector_indexer=multi_vector_indexerdefreciprocal_rank_fusion(self,ranked_listsist[List],k:int=60):"""RRF融合算法"""#收集所有候选文档all_docs=set()forranked_listinranked_lists:fordoc_id,_inranked_list:all_docs.add(doc_id)#计算RRF分数rrf_scores={}fordoc_idinall_docs:score=0forranked_listinranked_lists:#找到文档在当前排序列表中的位置rank=Nonefori,(candidate_id,_)inenumerate(ranked_list):ifcandidate_id==doc_id:rank=i+1#排名从1开始breakifrankisnotNone:score+=1/(k+rank)rrf_scores[doc_id]=score#按分数排序sorted_results=sorted(rrf_scores.items(),key=lambdax:x[1],reverse=True)returnsorted_resultsdefsearch(self,query:str,all_embeddingsict,discrete_indexist,documentsist,top_k:int=10):"""综合检索"""all_results=[]#1.离散检索discrete_results=self._discrete_search(query,discrete_index,top_k)all_results.append(discrete_results)#2.多向量检索formodel_name,embeddingsinall_embeddings.items():vector_results=self.multi_vector_indexer.search_single_model(query,model_name,embeddings,top_k)all_results.append(vector_results)#3.RRF融合final_results=self.reciprocal_rank_fusion(all_results)returnfinal_results[:top_k]def_discrete_search(self,query:str,discrete_indexist,top_k:int):"""离散索引检索"""query_keywords=self.discrete_indexer.extract_keywords(query)query_entities=[e['text']foreinself.discrete_indexer.extract_entities(query)]scores=[]fori,doc_metainenumerate(discrete_index):score=0#关键词匹配分数keyword_overlap=len(set(query_keywords)&set(doc_meta['keywords']))score+=keyword_overlap*2#实体匹配分数entity_overlap=len(set(query_entities)&set(doc_meta['entities']))score+=entity_overlap*3scores.append((i,score))#按分数排序scores.sort(key=lambdax:x[1],reverse=True)returnscores[:top_k]

解决方案二：Small-to-Big——"先找点，再扩面"

Small-to-Big策略解决的是另一个问题：检索到的文档片段太短，上下文不完整。这个策略的思路是：用小粒度的内容（摘要、关键句）建立索引来快速定位，然后返回大粒度的完整上下文。

实现原理

代码实现

fromtransformersimportpipeline,AutoTokenizer,AutoModelimporttorchclassSmallToBigIndexer:def__init__(self):#初始化摘要模型self.summarizer=pipeline("summarization",model="facebook/bart-large-cnn")defcreate_summary(self,text:str,max_length:int=150)->str:"""生成文档摘要"""iflen(text)<100:returntexttry:summary=self.summarizer(text,max_length=max_length,min_length=30,do_sample=False)returnsummary[0]['summary_text']exceptExceptionase:#如果摘要失败，返回前几句sentences=text.split('。')[:3]return'。'.join(sentences)+'。'defextract_key_sentences(self,text:str,num_sentences:int=3)->List[str]:"""提取关键句子"""sentences=text.split('。')sentences=[s.strip()forsinsentencesiflen(s.strip())>10]iflen(sentences)<=num_sentences:returnsentences#简单的关键句提取：选择包含更多实体和关键词的句子sentence_scores=[]forsentenceinsentences:score=0#长度因子score+=len(sentence)*0.1#位置因子（开头和结尾的句子更重要）score+=10ifsentenceinsentences[:2]else0score+=5ifsentenceinsentences[-2:]else0sentence_scores.append((sentence,score))#按分数排序sentence_scores.sort(key=lambdax:x[1],reverse=True)return[s[0]forsinsentence_scores[:num_sentences]]defbuild_small_to_big_index(self,documentsist[Dict])->Dict:"""构建Small-to-Big索引"""small_index=[]big_storage={}fordocindocuments:doc_id=doc['id']text=doc['text']#将长文档分割成大的chunksbig_chunks=self._split_into_big_chunks(text)fori,big_chunkinenumerate(big_chunks):big_chunk_id=f"{doc_id}_chunk_{i}"#存储大chunkbig_storage[big_chunk_id]={'text':big_chunk,'doc_id':doc_id,'chunk_index':i}#创建小的索引内容summary=self.create_summary(big_chunk)key_sentences=self.extract_key_sentences(big_chunk)#添加到小索引small_index.append({'small_content':summary,'content_type':'summary','big_chunk_id':big_chunk_id})forsentenceinkey_sentences:small_index.append({'small_content':sentence,'content_type':'key_sentence','big_chunk_id':big_chunk_id})return{'small_index':small_index,'big_storage':big_storage}def_split_into_big_chunks(self,text:str,chunk_size:int=1000,overlap:int=100)->List[str]:"""将文本分割成大的chunks"""chunks=[]start=0whilestart<len(text):end=start+chunk_size#尝试在句号处分割ifend<len(text):last_period=text.rfind('。',start,end)iflast_period>start:end=last_period+1chunk=text[start:end]ifchunk.strip():chunks.append(chunk.strip())start=end-overlapreturnchunks

classSmallToBigRetriever:def__init__(self,indexer,encoder):self.indexer=indexerself.encoder=encoderdefsearch(self,query:str,small_indexist,big_storageict,top_k:int=5)->List[Dict]:"""Small-to-Big检索"""#1.在小索引中检索small_results=self._search_small_index(query,small_index,top_k*2)#2.获取对应的大chunkIDsbig_chunk_ids=set()forresultinsmall_results:big_chunk_ids.add(result['big_chunk_id'])#3.从存储中获取大chunksretrieved_contexts=[]forbig_chunk_idinbig_chunk_ids:ifbig_chunk_idinbig_storage:big_chunk=big_storage[big_chunk_id]retrieved_contexts.append({'chunk_id':big_chunk_id,'text':big_chunk['text'],'doc_id':big_chunk['doc_id']})returnretrieved_contexts[:top_k]def_search_small_index(self,query:str,small_indexist,top_k:int)->List[Dict]:"""在小索引中搜索"""#将小索引内容编码small_texts=[item['small_content']foriteminsmall_index]embeddings=self.encoder.encode(small_texts)#查询编码query_embedding=self.encoder.encode([query])#计算相似度similarities=np.dot(embeddings,query_embedding.T).flatten()similarities=similarities/(np.linalg.norm(embeddings,axis=1)*np.linalg.norm(query_embedding))#获取top结果top_indices=np.argsort(similarities)[::-1][:top_k]results=[]foridxintop_indices:results.append({'small_content':small_index[idx]['small_content'],'content_type':small_index[idx]['content_type'],'big_chunk_id':small_index[idx]['big_chunk_id'],'similarity':similarities[idx]})returnresults

完整的使用示例

defmain():#准备测试数据documents=[{'id':'doc1','text':'深度学习是机器学习的一个分支，它基于人工神经网络进行学习和决策。深度学习模型通常包含多个隐层，能够学习数据的复杂模式。在图像识别、自然语言处理等领域都有广泛应用。目前主流的深度学习框架包括TensorFlow、PyTorch等。'},{'id':'doc2','text':'RAG（Retrieval-AugmentedGeneration）是一种结合检索和生成的技术。它先从知识库中检索相关信息，然后将检索结果作为上下文输入到生成模型中。这种方法可以让模型访问到更多的外部知识，提高回答的准确性。RAG特别适用于知识问答、文档摘要等任务。'}]query="什么是深度学习？"#1.索引扩展方法print("===索引扩展检索结果===")discrete_indexer=DiscreteIndexer()multi_vector_indexer=MultiVectorIndexer()ensemble_retriever=EnsembleRetriever(discrete_indexer,multi_vector_indexer)#构建索引discrete_index=discrete_indexer.build_discrete_index(documents)doc_texts=[doc['text']fordocindocuments]all_embeddings=multi_vector_indexer.encode_documents(doc_texts)#检索results=ensemble_retriever.search(query,all_embeddings,discrete_index,documents,top_k=3)fori,(doc_idx,score)inenumerate(results):print(f"结果{i+1}:文档{doc_idx},分数:{score:.4f}")print(f"内容:{documents[doc_idx]['text'][:100]}...")print()#2.Small-to-Big方法print("===Small-to-Big检索结果===")stb_indexer=SmallToBigIndexer()stb_retriever=SmallToBigRetriever(stb_indexer,multi_vector_indexer.models['bge'])#构建索引stb_data=stb_indexer.build_small_to_big_index(documents)#检索contexts=stb_retriever.search(query,stb_data['small_index'],stb_data['big_storage'],top_k=2)fori,contextinenumerate(contexts):print(f"上下文{i+1}:{context['chunk_id']}")print(f"内容:{context['text'][:200]}...")print()if__name__=="__main__":main()

实际效果对比

实施建议

总结

虽然实现相对复杂，但效果提升是实实在在的。特别是在企业级应用中，这种质量提升往往能带来用户体验的质的飞跃。