目录
RAG 工程的全流程
一、知识的预处理
1、为知识切片预生成问题(Query-Answer 对齐)
2、对话知识沉淀(让系统越用越聪明)
3、知识库健康度审查(定期体检)
4、知识库版本管理(把知识当产品迭代)
二、高效召回
1、多查询召回
2、索引扩展
3、Rerank模型使用
1)BGE-Rerank
2)Cohere Rerank
4、双向改写:Query2Doc和Doc2Query
5、Small to big策略
RAG 工程的全流程
很多人谈 RAG,想到的第一个画面是"向量数据库 + 大模型"。但如果把 RAG 当作一个工程拆开看,它至少包含以下环节:
阶段 | 环节 | 说明 | 来源 |
|---|---|---|---|
① 知识接入与解析 | 文档解析(PDF/网页/表格/OCR)、清洗、格式统一 | 原始数据往往很"杂",解析质量决定后续一切 | 文档预处理:RAG多模态知识库 |
② 知识切片(Chunking) | 按固定长度 / 语义边界 / 文档结构切块 | 切片策略是检索质量的"最高杠杆" | 切片策略:RAG搭建本地知识库 |
③ 知识存储(Index) | Embedding 向量化 → 写入向量数据库(FAISS / ES / Milvus 等) | 相当于给知识"建索引",目的是高效检索 | Embedding模型和向量数据库的选择 |
④ 知识预处理 | 预生成问题、打标签、生成摘要、健康度审查、版本管理、权限标签 | 一次性投入,持续受益 | 📖 本文档(重点) |
⑤ 查询理解与改写 | Query Rewrite、多轮对话上下文补全 | 用户的问题未必是"好问题" | 提示词的改写:如何与大模型交互?从提示词工程到RAG |
⑥ 检索(Retrieve) | 向量检索 + 关键词检索(Hybrid 混合检索)、BM25、元数据过滤 | 生产级系统的标配 | 📖 本文档 |
⑦ 重排序(Rerank) | Cross-Encoder 对候选集精排 | 把"对的证据"排到最前 | 📖 本文档 |
⑧ 上下文组装 | 版本过滤、去重、token 预算裁剪、Small-to-Big | 检索之后、生成之前的"最后一道关" | 📖 本文档 |
⑨ 生成(Generate) | chunk + 用户问题 → LLM 生成回答 | 有了背景知识,回答才不会"幻觉" | |
⑩ 评估与运营 | 召回率/准确率评测、可观测性、对话知识沉淀、版本迭代、免责声明 | RAG 是持续运营的产品,不是一锤子买卖 | 📖 本文档 |
在企业落地的真实场景中,80%~90% 的 case 用 Native RAG(原生 RAGRAG的工作流程与目标)就能解决——即"切片 → 向量化 → 检索 → 生成"这条朴素的主链路。但如果想让系统从"能跑"走向"好用",从"一次性项目"走向"持续运营的产品",就必须把上面这十个环节当作一个整体工程来对待。部分内容我们已经写了,我把文档的链接贴出来了。还有部分高级内容,我将在本文介绍。
一、知识的预处理
1、为知识切片预生成问题(Query-Answer 对齐)
我们的问题一般是疑问句,而知识切片一般是陈述句。那么问题和知识转化成向量后在表达方向上有可能不完全一致,导致用户问题和知识切片的相似度不是很高。我们可以为通过AI为知识切片生成一些可能性的问题,然后将问题和问题之间来做个匹配,以此来提高用户问题和知识切片的匹配度。
思路:让大模型扮演"问答专家",对每个知识切片预生成若干用户可能提问的问题,检索时做问题与问题的匹配,而非问题与陈述句的匹配。
成本账:预处理是一次性成本,相当于给知识加"外挂标签",不需要随每次查询重复付费。建议用代码而非对话式工具来做——提示词固定后直接批量运行,更省 token。
def get_completion(prompt, model="qwen3.7-plus"): messages = [{"role": "user", "content": prompt}] response = client.chat.completions.create( model=model, messages=messages, temperature=0.7, ) return response.choices[0].message.content ....... def generate_questions_for_chunk(self, knowledge_chunk, num_questions=5): """为单个知识切片生成多样化问题""" instruction = """ 你是一个专业的问答系统专家。给定的知识内容能回答哪些多样化的问题,这些问题可以: 1. 使用不同的问法(直接问、间接问、对比问等) 2. 避免重复和相似的问题 3. 确保问题不超出知识内容范围 请返回JSON格式: { "questions": [ { "question": "问题内容", "question_type": "问题类型(直接问/间接问/对比问/条件问等)", "difficulty": "难度等级(简单/中等/困难)" } ] } """ prompt = f""" ### 指令 ### {instruction} ### 知识内容 ### {knowledge_chunk} ### 生成问题数量 ### {num_questions} ### 生成结果 ### """ response = get_completion(prompt, self.model) # 预处理响应,移除markdown代码块格式 response = preprocess_json_response(response) try: result = json.loads(response) return result.get('questions', []) except json.JSONDecodeError as e: print(f"JSON解析失败: {e}") print(f"AI返回内容: {response[:50]}...") # 如果JSON解析失败,返回简单的问题列表 return [{"question": f"关于{knowledge_chunk[:50]}...的问题", "question_type": "直接问", "keywords": [], "difficulty": "中等"}]测试及结果:
要是想生成更多样化的问题,修改提示词就可以了,如:
instruction = """ 你是一个专业的问答系统专家。请为给定的知识内容生成高度多样化的问题,确保: 1. 问题类型多样化:直接问、间接问、对比问、条件问、假设问、推理问等 2. 表达方式多样化:使用不同的句式、词汇、语气 3. 难度层次多样化:简单、中等、困难的问题都要有 4. 角度多样化:从不同角度和维度提问 5. 确保问题不超出知识内容范围 请返回JSON格式: { "questions": [ { "question": "问题内容", "question_type": "问题类型", "difficulty": "难度等级", "perspective": "提问角度", "is_answerable": "给出的知识能否回答该问题", "answer": "基于该知识的回答" } ] } """2、对话知识沉淀(让系统越用越聪明)
系统上线后,会有大量的用户对话,这些对话是非常有价值的,把这些用户对话收集起来,提取对话中用户反馈的问题,表达的不满,提出的新需求整理出来,作为系统的下一次的迭代方向。核心:
• 使用AI模型从对话中提取结构化知识
• 支持多种知识类型:事实、问题、流程、注意事项等
• 自动识别用户意图和对话摘要
def extract_knowledge_from_conversation(self, conversation): """从单次对话中提取知识""" instruction = """ 你是一个专业的知识提取专家。请从给定的对话中提取有价值的知识点,包括: 1. 事实性信息(地点、时间、价格、规则等) 2. 用户需求和偏好 3. 常见问题和解答 4. 操作流程和步骤 5. 注意事项和提醒 请返回JSON格式: { "extracted_knowledge": [ { "knowledge_type": "知识类型(事实/需求/问题/流程/注意)", "content": "知识内容", "confidence": "置信度(0-1)", "source": "来源(用户/AI/对话)", "keywords": ["关键词1", "关键词2"], "category": "分类" } ], "conversation_summary": "对话摘要", "user_intent": "用户意图" } """ prompt = f""" ### 指令 ### {instruction} ### 对话内容 ### {conversation} ### 提取结果 ### """ response = get_completion(prompt, self.model) # 预处理响应,移除markdown代码块格式 response = preprocess_json_response(response) try: result = json.loads(response) return result except json.JSONDecodeError as e: print(f"对话知识提取JSON解析失败: {e}") print(f"AI返回内容: {response[:200]}...") return { "extracted_knowledge": [], "conversation_summary": "无法解析对话", "user_intent": "未知" }测试:
批量提取知识则上for循环:
def batch_extract_knowledge(self, conversations): """批量提取知识""" all_knowledge = [] for i, conversation in enumerate(conversations): print(f"正在处理对话 {i+1}/{len(conversations)}...") result = self.extract_knowledge_from_conversation(conversation) all_knowledge.extend(result.get('extracted_knowledge', [])) # 更新频率统计 for knowledge in result.get('extracted_knowledge', []): key = f"{knowledge['knowledge_type']}:{knowledge['content'][:50]}" self.knowledge_frequency[key] += 1 return all_knowledge知识合并:
按照类型把问题分类(knowledge_by_type),然后再用大模型扮演"知识整理专家"去重合并(merge_knowledge_with_llm)
def merge_similar_knowledge(self, knowledge_list): """使用LLM合并相似的知识点,过滤掉需求和问题类型""" # 过滤掉需求和问题类型的知识,因为它们是临时的、个性化的 filtered_knowledge = [ knowledge for knowledge in knowledge_list if knowledge.get('knowledge_type') not in ['需求', '问题'] ] print(f"过滤前知识点数量: {len(knowledge_list)}") print(f"过滤后知识点数量: {len(filtered_knowledge)}") print(f"过滤掉的'需求'和'问题'类型知识点: {len(knowledge_list) - len(filtered_knowledge)}") # 按知识类型分组 knowledge_by_type = {} for knowledge in filtered_knowledge: knowledge_type = knowledge.get('knowledge_type', '其他') if knowledge_type not in knowledge_by_type: knowledge_by_type[knowledge_type] = [] knowledge_by_type[knowledge_type].append(knowledge) merged_knowledge = [] # 对每个知识类型分别进行LLM合并 for knowledge_type, knowledge_group in knowledge_by_type.items(): if len(knowledge_group) == 1: # 只有一个知识点,直接添加 merged_knowledge.append(knowledge_group[0]) else: # 多个知识点,使用LLM合并 merged = self.merge_knowledge_with_llm(knowledge_group, knowledge_type) merged_knowledge.append(merged) return merged_knowledgedef merge_knowledge_with_llm(self, knowledge_group, knowledge_type): """使用LLM合并同类型的知识组""" # 准备知识内容列表 knowledge_contents = [] all_keywords = set() all_sources = [] for i, knowledge in enumerate(knowledge_group, 1): content = knowledge.get('content', '') confidence = knowledge.get('confidence', 0.5) keywords = knowledge.get('keywords', []) source = knowledge.get('source', '') category = knowledge.get('category', '') knowledge_contents.append(f"{i}. 内容: {content}") knowledge_contents.append(f" 置信度: {confidence}") knowledge_contents.append(f" 分类: {category}") knowledge_contents.append(f" 来源: {source}") knowledge_contents.append(f" 关键词: {', '.join(keywords)}") knowledge_contents.append("") # 构建LLM合并提示 prompt = f""" 你是一个专业的知识整理专家。请将以下{knowledge_type}类型的知识点进行智能合并,生成一个更完整、准确的知识点。 ### 合并要求: 1. 保留所有重要信息,避免信息丢失 2. 消除重复内容,整合相似表述 3. 提高内容的准确性和完整性 4. 保持逻辑清晰,结构合理 5. 合并后的置信度取所有知识点中的最高值 ### 待合并的知识点: {chr(10).join(knowledge_contents)} ### 请返回JSON格式: {{ "knowledge_type": "{knowledge_type}", "content": "合并后的知识内容", "confidence": 最高置信度值, "keywords": ["合并后的关键词列表"], "category": "合并后的分类", "sources": ["所有来源"], "frequency": {len(knowledge_group)} }} ### 合并结果: """ response = get_completion(prompt, self.model) # 预处理响应 response = preprocess_json_response(response) try: result = json.loads(response) return result except json.JSONDecodeError as e: print(f"知识合并JSON解析失败: {e}") print(f"AI返回内容: {response[:200]}...")关键技巧——置信度(Confidence):复杂任务不要让 AI"直接干活",要让它自我打分(0~100)。经验法则:≥90 分可基本信任;80~85 分说明方向对但不完整,不能直接入库;70 分以下要人工重点复核。类比考试:估分 95 说明有把握,估分 80 说明答案有欠缺。
这个技巧可迁移到所有"让 AI 做复杂决策"的场景——比如让 AI 写代码时给出 A/B/C 多个方案并各自打分:95 分的方案直接采纳,70 分的"推荐方案"也要再调研。打分让人从"逐条审查"中解放出来,只处理低分样本。
测试:
3、知识库健康度审查(定期体检)
知识库上线后不是一劳永逸的。需要定期对整个知识库进行健康度检查,找出缺少的知识、过期的知识、冲突的知识,确保知识库的质量和可靠性。核心内容:
• 完整性检查:评估知识库是否覆盖用户的主要查询需求
• 时效性检查:识别过期或需要更新的知识内容
• 一致性检查:发现知识库中的冲突和矛盾信息
• 综合评分:提供量化的健康度评分和改进建议
前三个检查都是让大模型来扮演相应的角色:
# 检查缺少的知识---------------------------------------- instruction = """ 你是一个知识库完整性检查专家。请分析给定的测试查询和知识库内容,判断知识库中是否缺少相关的知识。 检查标准: 1. 查询是否能在知识库中找到相关答案 2. 知识是否完整、准确 3. 是否覆盖了用户的主要需求 4. 是否存在知识空白 请返回JSON格式: { "missing_knowledge": [ { "query": "测试查询", "missing_aspect": "缺少的知识方面", "importance": "重要性(高/中/低)", "suggested_content": "建议的知识内容", "category": "知识分类" } ], "coverage_score": "覆盖率评分(0-1)", "completeness_analysis": "完整性分析" } """ # 检查过期的知识---------------------------------------- instruction = """ 你是一个知识时效性检查专家。请分析给定的知识内容,判断是否存在过期或需要更新的信息。 检查标准: 1. 时间相关信息是否过期(年份、日期、时间范围) 2. 价格信息是否最新(价格、费用、票价等) 3. 政策规则是否更新(政策、规定、规则等) 4. 活动信息是否有效(活动、节日、特殊安排等) 5. 联系方式是否准确(电话、地址、网址等) 6. 技术信息是否过时(版本、技术标准等) 请返回JSON格式: { "outdated_knowledge": [ { "chunk_id": "知识切片ID", "content": "知识内容", "outdated_aspect": "过期方面", "severity": "严重程度(高/中/低)", "suggested_update": "建议更新内容", "last_verified": "最后验证时间" } ], "freshness_score": "新鲜度评分(0-1)", "update_recommendations": "更新建议" } """ # 检查冲突的知识---------------------------------------- instruction = """ 你是一个知识一致性检查专家。请分析给定的知识库,找出可能存在冲突或矛盾的信息。 检查标准: 1. 同一主题的不同说法(地点、名称、描述等) 2. 价格信息的差异(价格、费用、收费标准等) 3. 时间信息的不一致(营业时间、开放时间、活动时间等) 4. 规则政策的冲突(规定、政策、要求等) 5. 操作流程的差异(步骤、方法、流程等) 6. 联系方式的差异(地址、电话、网址等) 请返回JSON格式: { "conflicting_knowledge": [ { "conflict_type": "冲突类型", "chunk_ids": ["相关切片ID"], "conflicting_content": ["冲突内容"], "severity": "严重程度(高/中/低)", "resolution_suggestion": "解决建议" } ], "consistency_score": "一致性评分(0-1)", "conflict_analysis": "冲突分析" } """然后根据返回的内容生成健康度报告并打印出来:
示例系统的健康评分只有 0.57~0.6——说明大多数知识库都"带病运行"。审查只是提示问题在哪里,补充和修正则是下一步的运营动作。
4、知识库版本管理(把知识当产品迭代)
知识库像代码和 Prompt 模板一样,需要版本化(V1、V2……),核心功能:
• 版本创建:为知识库创建带描述和统计信息的版本
• 统计信息:记录知识切片数量、内容长度、分类分布等
• 版本比较:比较两个版本的差异和变化
向量数据库本身支持增删改查,增量更新、局部更新完全可行,不必每次全量重建索引。
现在的公司里用的是哪些知识库管理工具,有知道的吗?
二、高效召回
我们原来是怎么用代码在Faiss数据库召回内容的呢?回忆一下:
for ... all_vectors.append(vector) #向量 metadata_store.append(metadata) #metadata # 创建FAISS索引 text_index = faiss.IndexFlatL2(TEXT_EMBEDDING_DIM) #参数维度 text_index_map = faiss.IndexIDMap(text_index) if all_vectors: text_ids = [m["id"] for m in metadata_store] text_index_map.add_with_ids(np.array(text_vectors).astype('float32'), np.array(text_ids)) ...... # 获取查询的embedding query_vector = np.array([get_text_embedding(query)]).astype('float32') # 使用faiss进行检索 distances, indices = text_index_map.search(query_vector, k) for i, doc_id in enumerate(indices[0]): m = metadata[idx] content_preview = m['content'][:45].replace('\n', ' ') ...... # 取top-k文本结果 top_results = [r for r in results if r["metadata"]["type"] == "text"][:k]# 创建文本分割器,用于将长文本分割成小块 text_splitter = RecursiveCharacterTextSplitter( separators=["\n\n", "\n", ".", " ", ""], chunk_size=1000, chunk_overlap=200, length_function=len, ) # 分割文本 chunks = text_splitter.split_text(text) print(f"文本被分割成 {len(chunks)} 个块。") # 创建嵌入模型 embeddings = DashScopeEmbeddings( model="text-embedding-v1", dashscope_api_key=DASHSCOPE_API_KEY, ) # 从文本块创建知识库 knowledgeBase = FAISS.from_texts(chunks, embeddings) print("已从文本块创建知识库。") ...... query = "客户经理被投诉了,投诉一次扣多少分" if query: # 执行相似度搜索,找到与查询相关的文档 docs = knowledgeBase.similarity_search(query,k=10) #K是召回多少个片段之前的召回,我们想要回答质量更高,我们的召回数量K(一般5~10)可以设置的高一点。然后给到大模型,如果召回到不太相关的内容给到大模型是没关系的,只要能召回到关键的信息,大模型自己会做筛选,组织,推理等工作。
1、多查询召回
相似语义改写,就是使用大模型将用户查询改写成多个语义相近的查询,提升召回多样性。在LangChain旧版本中提供了MultiQueryRetriever支持多查询召回,新版本需要自己编写。我们编写的也是用大模型来帮我们来改写:
"""使用LLM生成多个查询变体""" prompt = f"""你是一个AI助手,负责生成多个不同视角的搜索查询。 给定一个用户问题,生成{num_queries}个不同但相关的查询,以帮助检索更全面的信息。 原始问题: {query} 请直接输出{num_queries}个查询,每行一个,不要编号和其他内容:""" response = llm.invoke(prompt)我问一个问题:客户经理的考核标准是什么,生成结果如下:
遍历这些问题,将向量库中匹配的内容去重后作为大模型的参考资料让大模型回答问题。
2、索引扩展
1)离散索引扩展(关键词的索引BM25):使用关键词抽取、实体识别等技术生成离散索引,与向量检索互补,提升召回准确性。
BM25是一种经典的文本检索算法,它是TF-IDF(词频-逆文档频率)的改进版本,通过更精细的词频饱和度和文档长度归一化,提升了检索的相关性排序效果。
计算方法:用了TF * IDF以后,我们再除上一个它的归一化的系数,文档的长度越长,分数越低。文档的长度越短,价值越明显。BM25的值越高。
实体识别:从文档中识别出命名实体(如人名、地点、组织等),作为离散索引的一部分,增强检索的精确性。
2)连续索引扩展(向量索引):结合多种向量模型(如OpenAI的Ada、智源的BGE)进行多路召回,取长补短。
3)混合索引召回:将BM25等离散索引与向量索引结合,通过Ensemble Retriever实现混合召回,提升召回多样性
# ----------------------------BM25--------------------------------- # 构建BM25索引,tokenize_chinese是用jieba分词器把chunk切分 tokenized_chunks = [tokenize_chinese(chunk) for chunk in chunks] self.bm25 = BM25Okapi(tokenized_chunks) ....... # BM25检索 tokenized_query = tokenize_chinese(query) # 用jieba分词器query分词 bm25_scores = self.bm25.get_scores(tokenized_query) #分词列表打分 # 归一化BM25分数,归一化就是分数统一缩放到[0, 1] 区间 max_bm25 = max(bm25_scores) if max(bm25_scores) > 0 else 1 #获取最大的bm25打分 bm25_scores_normalized = [s / max_bm25 for s in bm25_scores] #归一化计算 # ----------------------------向量--------------------------------- # 构建chunk到索引的映射 self.chunk_to_idx = {chunk: idx for idx, chunk in enumerate(chunks)} ....... # 向量检索 (获取更多结果用于融合) vector_results = self.vectorstore.similarity_search_with_score(query, k=len(self.chunks)) # 构建向量分数字典 (距离越小越好,转换为分数) vector_scores = {} max_distance = max(score for _, score in vector_results) if vector_results else 1 for doc, distance in vector_results: idx = self.chunk_to_idx.get(doc.page_content) if idx is not None: # 距离转分数: 1 - (distance / max_distance) vector_scores[idx] = 1 - (distance / max_distance) if max_distance > 0 else 0 # ----------------------------融合--------------------------------- hybrid_scores = [] for idx in range(len(self.chunks)): bm25_score = bm25_scores_normalized[idx] vector_score = vector_scores.get(idx, 0) # 加权融合 combined = self.alpha * vector_score + (1 - self.alpha) * bm25_score hybrid_scores.append((idx, combined)) # 排序并返回top-k hybrid_scores.sort(key=lambda x: x[1], reverse=True) top_k = hybrid_scores[:k] results = [] for idx, score in top_k: doc = Document(page_content=self.chunks[idx], metadata={"hybrid_score": score}) #将切片内容和分数转为LangChain Document 对象 results.append(doc) ...... # 大模型根据返回内容回答 context = "\n\n".join([doc.page_content for doc in docs]) prompt = f"""根据以下上下文回答问题: {context} 问题: {query}""" response = llm.invoke(prompt)离散索引其实指的就是关键词的索引,我们会把关键词提取出来,根据关键词来去做一些匹配得到关键词的分数。以上的代码用jieba来分词,实际上,抽取关键词有TF-IDF、TextRank等方法,实体识别有SpaCy、BERT-based NER等方法。
BM25检索比较精确,而向量检索是模糊检索,避免丢掉一些写法上不统一的内容。向量检索是一个粗筛,他在给我们带来兜底的情况下,也给我们带来了干扰。
根据公式:Score = α ×Vector + (1-α) ×BM25
•专业术语多的文档-> 降低alpha (偏BM25)
•用户问题口语化-> 提高alpha (偏向量)
•不确定时-> 从α = 0.5 开始,根据效果微调
3、Rerank模型使用
有时候我们经过粗筛,召回的数量很多,并且这些也并不都有价值,反而给我们的检索带来干扰。这时候我们可以使用重排序Rerank在帮我们做一个筛选。
重排序Rerank主要用于优化初步检索结果的排序,提高最终输出的相关性或准确性。BGE-Rerank和Cohere Rerank是 两种广泛使用的重排序模型,它们在检索增强生成(RAG)系统、搜索引擎优化和问答系统中表现优异。
1)BGE-Rerank
由北京智源人工智能研究院(BAAI)开源发布,属于FlagEmbedding项目的一部分。 基于Transformer的Cross-Encoder结构,直接计算查询(Query)与文档(Document)的交互相关性得分。 训练数据:支持多语言(中、英等),训练数据包括T2Ranking、MSMARCO、NLI等数据集。 提供bge-reranker-base和bge-reranker-large两个版本,后者在精度上更优。部署方式:可本地部署
在魔搭社区可以检索到bge-rerank的模型:
本地部署Rerank模型:
# model_name: ModelScope上的模型名称 # - BAAI/bge-reranker-base (轻量级,推荐) # - BAAI/bge-reranker-large (效果更好,但更慢) model_dir = snapshot_download(model_name, cache_dir=cache_dir) print(f"模型已下载到: {model_dir}") self.tokenizer = AutoTokenizer.from_pretrained(model_dir) self.model = AutoModelForSequenceClassification.from_pretrained(model_dir) self.model.eval() # 使用GPU如果可用 self.device = "cuda" if torch.cuda.is_available() else "cpu" self.model.to(self.device) print(f"Rerank模型已加载,使用设备: {self.device}")大家看下,Rerank模型的下载和加载和我们前面介绍的其他模型一样呢。以下是使用Rerank模型召回文档,刚好接上一个代码:混合检索后召回docs,再用rerank模型做一个二次排序
reranked_docs = rerank(query, docs, top_k=final_k) def rerank(self, query: str, documents: List[Document], top_k: int = None) -> List[Document]: if not documents: return [] # 构建query-document对 pairs = [[query, doc.page_content] for doc in documents] # 批量计算分数 with torch.no_grad(): inputs = self.tokenizer( pairs, padding=True, truncation=True, max_length=512, return_tensors="pt" ).to(self.device) # 打分的重点!! scores = self.model(**inputs).logits.squeeze(-1).cpu().tolist() # 如果只有一个文档,scores可能是float而不是list if isinstance(scores, float): scores = [scores] # 为每个文档添加rerank分数 scored_docs = list(zip(documents, scores)) scored_docs.sort(key=lambda x: x[1], reverse=True) # 更新metadata中的分数 results = [] for doc, score in scored_docs: doc.metadata["rerank_score"] = score results.append(doc) if top_k: results = results[:top_k] return resultsReRank属于轻量级的弱化版的大模型。他不能回答问题,但是只能给你打分,告诉两个之间的相关性。由于问题-答案对都要去计算分数,实际他的速度不会很快。
在BGE-Rerank模型中,相关性分数scores是一个未归一化的对数几率(logits)值,范围没有固定的上 限或下限(不像某些模型限制在0-1)。不过BGE-Rerank的分数通常落在以下范围:高相关性:3.0~10.0 ;中等相关性:0.0~3.0;低相关性/不相关:负数(如-5.0以下)
2)Cohere Rerank
由Cohere公司提供的商业API服务。 基于专有的深度学习模型,支持多语言(如rerank-multilingual-v3.0)。 训练数据:优化了语义匹配,特别适用于混合检索(如结合BM25和向量检索)后的结果优化。 使用方式:通过API调用,集成到LangChain、LlamaIndex等框架中。
优势:
• 简单易用,适合快速集成到现有系统。
• 在英文和多语言任务中表现优异,如提升Hit Rate(命中率)和MRR(平均倒数排名)。
import cohere co = cohere.Client(api_key="YOUR_API_KEY") query = "What is the capital of France?" docs = ["Paris is the capital of France.", "Berlin is the capital of Germany."] results = co.rerank(query=query, documents=docs, top_n=2, model='rerank-multilingual-v3.0') print(results) #Cohere Rerank的API返回的是归一化后的相关性分数(如0-1),更易解释。4、双向改写:Query2Doc和Doc2Query
核心是解决“用户怎么问”和“知识怎么写”不一致的问题。
Query2Doc:把用户问题先扩写成一段“预想答案”,再用这段内容去匹配知识库。
例:用户问“怎么提高模型训练效率”,先扩成“可从混合精度、分布式训练、数据预处理、学习率调整等方向优化”,更容易命中知识库。Doc2Query:建库时为每个知识片段预生成若干“它能回答什么问题”。
检索时不只匹配原文,也匹配这些预生成问题,提高召回率。
5、Small to big策略
一种高效的检索方法,特别适用于处理长文档或多文档场景。核心思想是通过小规模内容(如摘要、关键句 或段落)建立索引,并链接到大规模内容主体中。这种策略的优势在于能够快速定位相关的小规模内容,并通过链接获取更详细的上下文信息,从而提高检索效率和答案的逻辑连贯性。
Small-to-Big机制:
小规模内容检索:用户输入查询后,系统首先在小规模内容(如摘要、关键句或段落)中检索匹配的内容。小规模内容通常是通过摘要生成、关键句提取等技术从大规模内容中提取的,并建立索引。
链接到大规模内容:当小规模内容匹配到用户的查询后,系统会通过预定义的链接(如文档ID、URL 或指针)找到对应的大规模内容(如完整的文档、文章)。大规模内容包含更详细的上下文信息,为RAG 提供丰富的背景知识。
上下文补充:将大规模内容作为RAG 系统的上下文输入,结合用户查询和小规模内容,生成更准确和连贯的答案。