背景
前边在介绍RAG系统时提到了文本分割(或分段)的作用和重要性。也提到了分段后所带来的一些问题,比如由于分段导致检索出来的TOP-n的结果可能未包含完整的答案。
- 粒度太大可能导致检索不精准,粒度太小可能导致信息不全面
- 问题的答案可能跨越两个片段
一、改进方案
1.1按一定粒度,部分重叠式的切割文本,使上下文更完整
英文文档的重叠式文本实现。
from nltk.tokenize import sent_tokenizedef split_text(paragraphs, chunk_size=300, overlap_size=100):'''按指定 chunk_size 和 overlap_size 交叠割文本'''sentences = [s.strip() for p in paragraphs for s in sent_tokenize(p)]chunks = []i = 0while i < len(sentences):chunk = sentences[i]overlap = ''prev_len = 0prev = i - 1# 向前计算重叠部分while prev >= 0 and len(sentences[prev])+len(overlap) <= overlap_size:overlap = sentences[prev] + ' ' + overlapprev -= 1chunk = overlap+chunknext = i + 1# 向后计算当前chunkwhile next < len(sentences) and len(sentences[next])+len(chunk) <= chunk_size:chunk = chunk + ' ' + sentences[next]next += 1chunks.append(chunk)i = nextreturn chunks
中文sent_tokenize实现:
import re
import jieba
import nltk
from nltk.corpus import stopwords# nltk.download('stopwords')def to_keywords(input_string):"""将句子转成检索关键词序列"""# 按搜索引擎模式分词word_tokens = jieba.cut_for_search(input_string)# 加载停用词表stop_words = set(stopwords.words('chinese'))# 去除停用词filtered_sentence = [w for w in word_tokens if not w in stop_words]return ' '.join(filtered_sentence)def sent_tokenize(input_string):"""按标点断句"""# 按标点切分sentences = re.split(r'(?<=[。!?;?!])', input_string)# 去掉空字符串return [sentence for sentence in sentences if sentence.strip()]if "__main__" == __name__:# 测试关键词提取print(to_keywords("小明硕士毕业于中国科学院计算所,后在日本京都大学深造"))# 测试断句print(sent_tokenize("这是,第一句。这是第二句吗?是的!啊"))
测试示例:
chunks = split_text(paragraphs, 300, 100)# 创建一个向量数据库对象
vector_db = MyVectorDBConnector("demo_text_split", get_embeddings)
# 向向量数据库中添加文档
vector_db.add_documents(chunks)
# 创建一个RAG机器人
bot = RAG_Bot(vector_db,llm_api=get_completion
)# user_query = "llama 2有商用许可协议吗"
user_query="llama 2 chat有多少参数"search_results = vector_db.search(user_query, 2)
for doc in search_results['documents'][0]:print(doc+"\n")response = bot.chat(user_query)
print("====回复====")
print(response)
1.2 检索后排序
解决问题:有时最合适的答案不一定排在检索的最前面,截断后会导致返回结果不完全准确。
方案:
- 检索时招回一部分文本
- 通过一个排序模型对 query 和 document 重新打分排序
安装模型
pip install sentence_transformers
使用模型
from sentence_transformers import CrossEncoder# model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2', max_length=512) # 英文,模型较小
model = CrossEncoder('BAAI/bge-reranker-large', max_length=512) # 多语言,国产,模型较大
测试打分
user_query = "how safe is llama 2"
# user_query = "llama 2安全性如何"
scores = model.predict([(user_query, doc)for doc in search_results['documents'][0]])
# 按得分排序
sorted_list = sorted(zip(scores, search_results['documents'][0]), key=lambda x: x[0], reverse=True)
for score, doc in sorted_list:print(f"{score}\t{doc}\n")
1.3 混合检索(Hybrid Search)
在实际生产中,传统的关键字检索(稀疏表示)与向量检索(稠密表示)各有优劣。
举个具体例子,比如文档中包含很长的专有名词,关键字检索往往更精准而向量检索容易引入概念混淆。
# 背景说明:在医学中“小细胞肺癌”和“非小细胞肺癌”是两种不同的癌症query = "非小细胞肺癌的患者"documents = ["玛丽患有肺癌,癌细胞已转移","刘某肺癌I期","张某经诊断为非小细胞肺癌III期","小细胞肺癌是肺癌的一种"
]query_vec = get_embeddings([query])[0]
doc_vecs = get_embeddings(documents)print("Cosine distance:")
for vec in doc_vecs:print(cos_sim(query_vec, vec))
所以,有时候我们需要结合不同的检索算法,来达到比单一检索算法更优的效果。这就是混合检索。
混合检索的核心是,综合文档 𝑑 在不同检索算法下的排序名次(rank),为其生成最终排序。
一个最常用的算法叫 Reciprocal Rank Fusion(RRF)
𝑟𝑟𝑓(𝑑)=∑𝑎∈𝐴1𝑘+𝑟𝑎𝑛𝑘𝑎(𝑑)
其中 𝐴 表示所有使用的检索算法的集合,𝑟𝑎𝑛𝑘𝑎(𝑑) 表示使用算法 𝑎 检索时,文档 𝑑 的排序,𝑘 是个常数。
很多向量数据库都支持混合检索,比如 Weaviate、Pinecone 等。也可以根据上述原理自己实现。
手写混合检索示例:
- 基于关键字检索的排序
import timeclass MyEsConnector:def __init__(self, es_client, index_name, keyword_fn):self.es_client = es_clientself.index_name = index_nameself.keyword_fn = keyword_fndef add_documents(self, documents):'''文档灌库'''if self.es_client.indices.exists(index=self.index_name):self.es_client.indices.delete(index=self.index_name)self.es_client.indices.create(index=self.index_name)actions = [{"_index": self.index_name,"_source": {"keywords": self.keyword_fn(doc),"text": doc,"id": f"doc_{i}"}}for i, doc in enumerate(documents)]helpers.bulk(self.es_client, actions)time.sleep(1)def search(self, query_string, top_n=3):'''检索'''search_query = {"match": {"keywords": self.keyword_fn(query_string)}}res = self.es_client.search(index=self.index_name, query=search_query, size=top_n)return {hit["_source"]["id"]: {"text": hit["_source"]["text"],"rank": i,}for i, hit in enumerate(res["hits"]["hits"])}
- 基于向量检索的排序
# 创建向量数据库连接器
vecdb_connector = MyVectorDBConnector("demo_vec_rrf", get_embeddings)# 文档灌库
vecdb_connector.add_documents(documents)# 向量检索
vector_search_results = {"doc_"+str(documents.index(doc)): {"text": doc,"rank": i}for i, doc in enumerate(vecdb_connector.search(query, 3)["documents"][0])
} # 把结果转成跟上面关键字检索结果一样的格式print(json.dumps(vector_search_results, indent=4, ensure_ascii=False))
- 基于 RRF 的融合排序
def rrf(ranks, k=1):ret = {}# 遍历每次的排序结果for rank in ranks:# 遍历排序中每个元素for id, val in rank.items():if id not in ret:ret[id] = {"score": 0, "text": val["text"]}# 计算 RRF 得分ret[id]["score"] += 1.0/(k+val["rank"])# 按 RRF 得分排序,并返回return dict(sorted(ret.items(), key=lambda item: item[1]["score"], reverse=True))
import json# 融合两次检索的排序结果
reranked = rrf([keyword_search_results, vector_search_results])print(json.dumps(reranked, indent=4, ensure_ascii=False))
1.4 RAG-Fusion
RAG-Fusion 就是利用了 RRF 的原理来提升检索的准确性。
原始项目(一段非常简短的演示代码):https://github.com/Raudaschl/rag-fusion