在构建RAG系统时很多开发者会遇到检索结果不准确、上下文信息不足、复杂查询处理困难等问题。NirDiamant的RAG_Techniques仓库汇集了42可运行的notebook教程从基础到前沿全面覆盖RAG技术栈为开发者提供了实用的解决方案。本文将深入解析该仓库的核心技术体系帮助读者系统掌握RAG的进阶应用。1. RAG技术概述与NirDiamant仓库介绍1.1 RAG技术基础概念检索增强生成Retrieval-Augmented GenerationRAG是一种结合信息检索与生成式AI的技术范式。其核心思想是在大语言模型生成答案前先从知识库中检索相关文档作为上下文从而生成更准确、更具事实依据的响应。传统RAG系统的基本流程包括文档加载→文本分块→向量化存储→相似性检索→上下文增强生成。然而这种基础架构在处理复杂查询、多模态数据、长文档等场景时存在明显局限性这正是高级RAG技术需要解决的问题。1.2 NirDiamant/RAG_Techniques仓库价值该GitHub仓库目前获得28.6k星标包含42个以上可运行的notebook教程涵盖了从基础到前沿的完整RAG技术栈。仓库按技术类别系统组织每个技术都有详细的理论说明、代码实现和参考资源。仓库的技术体系分为六大类别基础RAG技术、查询增强、上下文丰富、高级检索方法、迭代自适应技术、评估与高级架构。这种分类方式帮助开发者根据具体需求快速定位合适的技术方案。2. 基础RAG技术深度解析2.1 基础RAG实现模式基础RAG技术为整个技术栈的基石主要包括简单的文档检索和问答系统。以下是一个基于LangChain的基础RAG实现示例from langchain.document_loaders import TextLoader from langchain.text_splitter import CharacterTextSplitter from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma from langchain.chains import RetrievalQA from langchain.llms import OpenAI # 文档加载与处理 loader TextLoader(document.txt) documents loader.load() # 文本分块 text_splitter CharacterTextSplitter(chunk_size1000, chunk_overlap200) texts text_splitter.split_documents(documents) # 向量化存储 embeddings OpenAIEmbeddings() vectorstore Chroma.from_documents(texts, embeddings) # 检索增强生成 qa_chain RetrievalQA.from_chain_type( llmOpenAI(), chain_typestuff, retrievervectorstore.as_retriever() ) response qa_chain.run(你的查询问题)这种基础实现虽然简单但为后续高级技术提供了可扩展的框架。在实际项目中需要根据具体需求调整分块大小、重叠区间等参数。2.2 分块策略优化分块策略直接影响检索效果。仓库中提供了多种分块方法的对比固定大小分块是最简单的方法但可能割裂语义完整性text_splitter CharacterTextSplitter( chunk_size512, chunk_overlap50 )语义分块基于内容主题进行划分保持语义连贯性from langchain_experimental.text_splitter import SemanticChunker from langchain.embeddings import OpenAIEmbeddings embeddings OpenAIEmbeddings() text_splitter SemanticChunker(embeddings)命题分块将文本转化为事实性陈述特别适合知识提取场景# 生成命题式分块 proposition_prompt 请从以下文本中提取关键事实命题 {text} 返回格式每个命题单独一行保持客观事实性。 # 使用LLM生成命题 propositions llm.generate_propositions(text)2.3 可靠RAG技术可靠RAG在基础版本上增加了验证和精炼机制确保检索信息的准确性和相关性。关键技术包括相关性验证对检索到的文档进行相关性评分过滤低质量结果来源高亮明确标识答案所依据的文档片段多重验证通过多个来源交叉验证关键信息3. 查询增强技术3.1 查询转换与重写查询转换技术通过改写用户原始查询来提高检索效果。主要方法包括查询重写使用LLM重新表述查询使其更符合文档表述方式rewrite_prompt 请将以下用户查询重写为更适合文档检索的版本 原始查询{query} 考虑因素同义词替换、专业术语标准化、查询意图明确化。 rewritten_query llm.invoke(rewrite_prompt)退步提示Step-back Prompting生成更一般的查询来获取背景信息step_back_prompt 基于以下具体查询生成一个更一般性的背景查询 具体查询{query} 背景查询应该帮助理解相关概念和上下文。 子查询分解将复杂查询拆分为多个简单子查询decomposition_prompt 将以下复杂查询分解为2-3个简单的子查询 复杂查询{complex_query} 每个子查询应该能够独立检索相关信息。 3.2 HyDE假设文档嵌入HyDE技术通过生成假设性答案文档来改善检索对齐度hyde_prompt 基于以下查询生成一个假设性的答案文档 查询{query} 假设文档应该包含查询可能涉及的关键信息和表述方式。 hypothetical_document llm.invoke(hyde_prompt) # 使用假设文档进行嵌入和检索 hypothetical_embedding embeddings.embed_query(hypothetical_document)3.3 HyPE假设提示嵌入HyPE技术在索引阶段预计算假设性问题将检索任务转化为问题-问题匹配# 在索引阶段为每个文档块生成多个假设性问题 hype_prompt 为以下文档块生成3个可能被问及的问题 文档内容{chunk_text} 问题应该覆盖文档的关键信息点。 hypothetical_questions [] for chunk in document_chunks: questions llm.generate_questions(chunk) # 将问题与原始块一起存储 vectorstore.add_texts(questions, metadata{original_chunk: chunk})HyPE相比HyDE的优势在于不需要在查询时进行LLM调用减少了延迟和计算成本。4. 上下文丰富与内容增强4.1 上下文块头信息为每个文本块添加文档级和章节级的上下文头信息显著提升检索准确性def create_contextual_header(chunk, document_metadata, section_info): header f 文档标题{document_metadata[title]} 章节{section_info[section_title]} 前一节主题{section_info[previous_topic]} 后一节主题{section_info[next_topic]} 文档类型{document_metadata[doc_type]} 创建时间{document_metadata[create_date]} 内容块 return header chunk # 应用上下文头 enhanced_chunks [] for chunk in text_chunks: enhanced_chunk create_contextual_header(chunk, doc_meta, section_meta) enhanced_chunks.append(enhanced_chunk)4.2 相关片段提取动态构建多块文本片段提供更完整的上下文信息def extract_relevant_segments(retrieved_chunks, query, window_size2): 基于检索到的块扩展获取相邻块形成完整片段 segments [] for chunk in retrieved_chunks: chunk_index chunk.metadata[index] start_idx max(0, chunk_index - window_size) end_idx chunk_index window_size 1 segment_chunks original_chunks[start_idx:end_idx] segment_text \n.join([chunk.text for chunk in segment_chunks]) segments.append(segment_text) return segments4.3 语义分块技术基于语义连贯性而非固定大小进行分块from langchain_experimental.text_splitter import SemanticChunker def semantic_chunking(text, embeddings, breakpoint_threshold0.3): 基于语义相似度断点进行分块 splitter SemanticChunker( embeddings, breakpoint_thresholdbreakpoint_threshold ) return splitter.split_text(text)4.4 上下文压缩使用LLM压缩检索到的信息保留查询相关内容compression_prompt 压缩以下文本只保留与查询相关的关键信息 查询{query} 文本{retrieved_text} 要求保持事实准确性删除无关内容突出相关信息。 def compress_context(retrieved_text, query): compressed llm.invoke(compression_prompt) return compressed5. 高级检索方法5.1 融合检索结合关键词搜索和向量搜索的优势from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain.vectorstores import Chroma def create_fusion_retriever(documents): # 关键词检索器 bm25_retriever BM25Retriever.from_documents(documents) bm25_retriever.k 5 # 向量检索器 vector_retriever Chroma.from_documents(documents, embeddings).as_retriever( search_kwargs{k: 5} ) # 融合检索器 ensemble_retriever EnsembleRetriever( retrievers[bm25_retriever, vector_retriever], weights[0.4, 0.6] ) return ensemble_retriever5.2 智能重排序应用高级评分机制改进检索结果的相关性排序from sentence_transformers import CrossEncoder class Reranker: def __init__(self, model_namecross-encoder/ms-marco-MiniLM-L-6-v2): self.cross_encoder CrossEncoder(model_name) def rerank(self, query, documents, top_k3): pairs [(query, doc.text) for doc in documents] scores self.cross_encoder.predict(pairs) # 结合原始分数和重排序分数 ranked_results [] for doc, score in zip(documents, scores): final_score 0.7 * doc.score 0.3 * score ranked_results.append((doc, final_score)) ranked_results.sort(keylambda x: x[1], reverseTrue) return ranked_results[:top_k]5.3 多维度过滤应用多种过滤技术提升结果质量def multifaceted_filtering(documents, query, filters): 应用多维度过滤 filtered_docs documents # 元数据过滤 if metadata_filters in filters: for key, value in filters[metadata_filters].items(): filtered_docs [doc for doc in filtered_docs if doc.metadata.get(key) value] # 相似度阈值过滤 if similarity_threshold in filters: filtered_docs [doc for doc in filtered_docs if doc.score filters[similarity_threshold]] # 内容关键词过滤 if required_keywords in filters: filtered_docs [doc for doc in filtered_docs if any(keyword in doc.text.lower() for keyword in filters[required_keywords])] return filtered_docs5.4 分层索引创建多层级系统实现高效信息导航class HierarchicalIndex: def __init__(self): self.summary_index {} # 文档摘要级索引 self.detail_index {} # 详细内容级索引 def build_index(self, documents): for doc in documents: # 生成文档摘要 summary self.generate_summary(doc.text) self.summary_index[doc.id] { summary: summary, metadata: doc.metadata } # 处理详细内容 chunks self.chunk_document(doc.text) for i, chunk in enumerate(chunks): chunk_id f{doc.id}_chunk_{i} self.detail_index[chunk_id] { content: chunk, doc_id: doc.id, chunk_index: i } def search(self, query, levelsummary): if level summary: return self.search_summaries(query) else: return self.search_details(query)6. 迭代式与自适应技术6.1 反馈循环检索基于用户交互学习并改进后续检索class FeedbackRetrieval: def __init__(self, base_retriever): self.retriever base_retriever self.feedback_log [] def log_feedback(self, query, retrieved_docs, user_ratings): 记录用户反馈 feedback_entry { query: query, retrieved_docs: [doc.id for doc in retrieved_docs], ratings: user_ratings, timestamp: datetime.now() } self.feedback_log.append(feedback_entry) def adapt_retrieval(self, query): 基于历史反馈调整检索策略 # 分析相似查询的成功模式 similar_successful self.analyze_successful_patterns(query) if similar_successful: # 应用学习到的最佳参数 optimized_params self.derive_optimized_parameters(similar_successful) return self.retriever.retrieve(query, **optimized_params) else: return self.retriever.retrieve(query)6.2 自适应检索根据查询类型和用户上下文动态调整检索策略class AdaptiveRetriever: def __init__(self): self.query_classifier QueryClassifier() self.retrieval_strategies { factual: FactualRetrievalStrategy(), analytical: AnalyticalRetrievalStrategy(), procedural: ProceduralRetrievalStrategy() } def retrieve(self, query, user_contextNone): # 分类查询类型 query_type self.query_classifier.classify(query) # 选择检索策略 strategy self.retrieval_strategies.get(query_type, DefaultRetrievalStrategy()) # 应用用户上下文偏好 if user_context: strategy.adapt_to_context(user_context) return strategy.execute(query)7. RAG系统评估方法7.1 端到端评估框架构建完整的RAG系统评估流水线from ragas import evaluate from ragas.metrics import ( faithfulness, answer_relevancy, context_precision, context_recall ) def evaluate_rag_pipeline(qa_pipeline, test_dataset): 综合评估RAG管道性能 results [] for test_case in test_dataset: # 运行管道 response qa_pipeline(test_case[question]) # 计算各项指标 result { question: test_case[question], expected_answer: test_case[answer], actual_answer: response[answer], retrieved_context: response[context], faithfulness: calculate_faithfulness(response), answer_relevancy: calculate_relevancy(test_case[question], response[answer]), context_precision: calculate_context_precision(response), context_recall: calculate_context_recall(test_case, response) } results.append(result) return results7.2 基于LLM的评估使用大语言模型作为评估者class LLMEvaluator: def __init__(self, llm): self.llm llm def evaluate_answer_quality(self, question, answer, context): evaluation_prompt 评估以下问答的质量 问题{question} 参考答案{expected_answer} 实际答案{actual_answer} 提供上下文{context} 请从以下维度评分1-5分 - 事实准确性 - 答案相关性 - 完整性 - 清晰度 并提供改进建议。 response self.llm.invoke(evaluation_prompt) return self.parse_evaluation_response(response)8. 高级架构与技术8.1 图RAG技术将知识图谱与向量检索相结合import networkx as nx class GraphRAG: def __init__(self): self.knowledge_graph nx.Graph() self.vector_store None def build_graph_from_documents(self, documents): 从文档构建知识图谱 for doc in documents: # 提取实体和关系 entities self.extract_entities(doc.text) relationships self.extract_relationships(doc.text) # 添加到图谱 for entity in entities: self.knowledge_graph.add_node(entity, document_iddoc.id) for rel in relationships: self.knowledge_graph.add_edge(rel[source], rel[target], relationshiprel[type]) def graph_enhanced_retrieval(self, query): 图增强的检索 # 向量检索获取初始结果 initial_results self.vector_store.similarity_search(query, k5) # 图扩展获取相关实体 related_entities self.expand_via_graph(initial_results) # 基于相关实体进行二次检索 expanded_results self.retrieve_by_entities(related_entities) return initial_results expanded_results8.2 自RAGSelf-RAG实现检索决策自适应的RAG系统class SelfRAG: def __init__(self, llm, retriever): self.llm llm self.retriever retriever def process_query(self, query): # 步骤1检索决策 need_retrieval self.decide_retrieval_need(query) if need_retrieval: # 步骤2文档检索 documents self.retriever.retrieve(query) # 步骤3相关性评估 relevant_docs self.evaluate_relevance(query, documents) # 步骤4响应生成 response self.generate_with_context(query, relevant_docs) # 步骤5支持评估 support_score self.assess_support(response, relevant_docs) else: # 直接生成 response self.llm.generate(query) support_score 1.0 # 无检索支持 return { response: response, support_score: support_score, retrieval_used: need_retrieval }8.3 多模态RAG扩展RAG处理多种数据类型class MultimodalRAG: def __init__(self): self.text_processor TextProcessor() self.image_processor ImageProcessor() self.audio_processor AudioProcessor() def process_multimodal_documents(self, documents): 处理多模态文档 multimodal_embeddings [] for doc in documents: if doc.type text: embedding self.text_processor.embed(doc.content) elif doc.type image: # 图像描述生成 caption self.image_processor.caption(doc.content) embedding self.text_processor.embed(caption) elif doc.type audio: # 语音转文本 transcript self.audio_process