高级RAG模式2026：超越简单向量检索的架构演进

问题: "对比GPT-4和Claude在代码生成任务上的表现"

基础RAG: 一次检索"GPT-4 Claude 代码生成对比" → 可能找不到直接的对比文档
高级RAG: 
  检索1: "GPT-4 代码生成性能"
  检索2: "Claude 代码生成性能"  
  检索3: "GPT-4 vs Claude benchmark"
  → 综合多个来源生成对比

问题: "那位发明了Transformer的作者后来创立了哪家公司？"

跳1: 检索"Transformer论文作者" → 得到作者列表（包括Ashish Vaswani等）
跳2: 检索"Ashish Vaswani 创立公司" → 得到Adept AI
答案: Adept AI

class MultiHopRetriever:
    def __init__(self, retriever, llm, max_hops=3):
        self.retriever = retriever
        self.llm = llm
        self.max_hops = max_hops
    
    async def retrieve(self, question):
        accumulated_context = []
        
        for hop in range(self.max_hops):
            # 1. 根据当前上下文生成下一步检索查询
            if hop == 0:
                query = question
            else:
                query = await self.generate_next_query(
                    question, accumulated_context
                )
            
            if query is None:  # 不需要更多检索
                break
            
            # 2. 执行检索
            results = await self.retriever.search(query, top_k=5)
            
            # 3. 过滤和评估结果
            relevant = await self.filter_relevant(results, question, accumulated_context)
            accumulated_context.extend(relevant)
            
            # 4. 检查是否已有足够信息回答
            if await self.has_enough_info(question, accumulated_context):
                break
        
        return accumulated_context
    
    async def generate_next_query(self, question, context):
        """基于当前上下文生成下一步查询"""
        prompt = f"""
        原始问题: {question}
        已知信息: {context}
        
        还需要什么信息才能回答问题？请生成一个搜索查询。
        如果已经有足够信息，返回"DONE"。
        """
        response = await self.llm.call(prompt)
        if "DONE" in response:
            return None
        return response.strip()

class AdaptiveRetriever:
    def __init__(self):
        self.strategies = {
            "no_retrieval": NoRetrievalStrategy(),
            "single_retrieval": SingleRetrievalStrategy(),
            "multi_hop": MultiHopStrategy(),
            "graph_search": GraphSearchStrategy(),
            "hybrid_search": HybridSearchStrategy()
        }
        self.classifier = QueryClassifier()
    
    async def retrieve(self, question):
        # 1. 分类查询
        query_type = await self.classifier.classify(question)
        
        # 2. 选择检索策略
        strategy = self.strategies[query_type.strategy]
        
        # 3. 执行检索
        results = await strategy.retrieve(question)
        
        return results

class QueryClassifier:
    async def classify(self, question):
        """分类查询类型"""
        prompt = f"""
        问题: {question}
        
        请分类:
        1. 是否需要外部知识？(yes/no)
        2. 需要几轮检索？(0/1/multiple)
        3. 检索类型？(vector/graph/hybrid)
        4. 复杂度？(simple/moderate/complex)
        """
        result = await self.llm.call(prompt)
        return QueryType(
            strategy=result["strategy"],
            complexity=result["complexity"]
        )

class QueryDecomposer:
    async def decompose(self, question):
        """将复杂问题分解为子问题"""
        prompt = f"""
        将以下复杂问题分解为更简单的子问题:
        
        问题: {question}
        
        要求:
        1. 每个子问题应该可以独立检索
        2. 子问题的答案组合后应该能回答原始问题
        3. 子问题之间应该不重叠
        """
        sub_questions = await self.llm.call(prompt)
        return sub_questions

class QueryRewriter:
    async def rewrite(self, question, context=None):
        """改写查询以提高检索效果"""
        # 1. 扩展缩写
        expanded = await self.expand_abbreviations(question)
        
        # 2. 添加同义词
        with_synonyms = await self.add_synonyms(expanded)
        
        # 3. 基于上下文消歧
        if context:
            disambiguated = await self.disambiguate(with_synonyms, context)
        else:
            disambiguated = with_synonyms
        
        # 4. 生成多个变体
        variants = await self.generate_variants(disambiguated)
        
        return variants

class ResultReasoner:
    async def reason(self, question, retrieved_docs):
        """对检索结果进行推理整合"""
        prompt = f"""
        问题: {question}
        
        检索到的文档:
        {self.format_docs(retrieved_docs)}
        
        请进行以下推理:
        1. 哪些文档直接回答了问题？
        2. 哪些文档提供了间接支持？
        3. 文档之间是否有矛盾？
        4. 基于这些文档，答案是什么？
        5. 答案的置信度是多少？
        6. 还缺少什么信息？
        """
        reasoning = await self.llm.call(prompt)
        return reasoning

class IterativeRefinementRetriever:
    async def retrieve(self, question, max_iterations=3):
        query = question
        all_results = []
        
        for i in range(max_iterations):
            # 1. 检索
            results = await self.retriever.search(query, top_k=10)
            
            # 2. 评估结果质量
            quality = await self.assess_quality(results, question)
            
            if quality > 0.8:  # 质量足够好
                all_results = results
                break
            
            # 3. 改进查询
            query = await self.improve_query(query, results, question)
            
            # 4. 积累好的结果
            good_results = [r for r in results if r.relevance > 0.7]
            all_results.extend(good_results)
        
        # 5. 去重和重排
        all_results = self.deduplicate(all_results)
        all_results = await self.rerank(all_results, question)
        
        return all_results[:5]
    
    async def improve_query(self, old_query, results, question):
        """基于检索结果改进查询"""
        prompt = f"""
        原始问题: {question}
        当前查询: {old_query}
        检索结果质量: 不够好
        
        检索到的部分结果:
        {self.format_results(results[:3])}
        
        请改进搜索查询，使其更精确地找到回答问题所需的信息。
        """
        return await self.llm.call(prompt)

class HybridRetriever:
    def __init__(self, vector_retriever, keyword_retriever):
        self.vector_retriever = vector_retriever
        self.keyword_retriever = keyword_retriever
    
    async def search(self, query, top_k=10):
        # 1. 向量检索
        vector_results = await self.vector_retriever.search(query, top_k=top_k*2)
        
        # 2. 关键词检索
        keyword_results = await self.keyword_retriever.search(query, top_k=top_k*2)
        
        # 3. 融合结果
        fused = self.reciprocal_rank_fusion(vector_results, keyword_results)
        
        return fused[:top_k]
    
    def reciprocal_rank_fusion(self, *result_lists, k=60):
        """倒数排名融合"""
        scores = {}
        for results in result_lists:
            for rank, doc in enumerate(results):
                if doc.id not in scores:
                    scores[doc.id] = 0
                scores[doc.id] += 1 / (k + rank)
        
        # 按融合分数排序
        sorted_docs = sorted(scores.items(), key=lambda x: -x[1])
        return [doc for doc, score in sorted_docs]

class GraphVectorRetriever:
    async def search(self, query, top_k=10):
        # 1. 向量检索找到种子文档
        seed_docs = await self.vector_retriever.search(query, top_k=3)
        
        # 2. 图谱检索扩展相关文档
        expanded_docs = []
        for seed in seed_docs:
            # 沿知识图谱的边扩展
            neighbors = await self.graph_store.get_neighbors(
                seed.entity_id, 
                max_hops=2,
                relation_types=["related_to", "part_of", "authored_by"]
            )
            expanded_docs.extend(neighbors)
        
        # 3. 融合和重排
        all_docs = seed_docs + expanded_docs
        all_docs = self.deduplicate(all_docs)
        reranked = await self.reranker.rerank(all_docs, query)
        
        return reranked[:top_k]

class SelfRAG:
    async def generate(self, question):
        # 1. 判断是否需要检索
        need_retrieval = await self.judge_need_retrieval(question)
        
        if not need_retrieval:
            return await self.llm.generate(question)
        
        # 2. 生成检索查询
        query = await self.generate_query(question)
        
        # 3. 检索
        docs = await self.retriever.search(query)
        
        # 4. 评估每个文档的相关性
        relevant_docs = []
        for doc in docs:
            is_relevant = await self.judge_relevance(question, doc)
            if is_relevant:
                relevant_docs.append(doc)
        
        # 5. 基于相关文档生成
        if relevant_docs:
            answer = await self.generate_with_docs(question, relevant_docs)
            
            # 6. 验证答案是否被文档支持
            is_supported = await self.verify_support(answer, relevant_docs)
            
            if is_supported:
                return answer
            else:
                # 答案不被支持，重新检索或生成
                return await self.retry(question, relevant_docs)
        else:
            # 没有找到相关文档
            return await self.handle_no_relevant_docs(question)

class RAGEvaluator:
    async def evaluate(self, rag_system, test_cases):
        """评估RAG系统"""
        metrics = {
            "retrieval_precision": [],
            "retrieval_recall": [],
            "answer_accuracy": [],
            "answer_faithfulness": [],
            "answer_completeness": [],
            "latency": []
        }
        
        for case in test_cases:
            # 运行RAG
            start = time.time()
            result = await rag_system.query(case["question"])
            elapsed = time.time() - start
            
            # 评估检索
            retrieval_metrics = self.evaluate_retrieval(
                result.retrieved_docs, case.relevant_docs
            )
            metrics["retrieval_precision"].append(retrieval_metrics["precision"])
            metrics["retrieval_recall"].append(retrieval_metrics["recall"])
            
            # 评估答案
            answer_metrics = await self.evaluate_answer(
                result.answer, case.expected_answer, result.retrieved_docs
            )
            metrics["answer_accuracy"].append(answer_metrics["accuracy"])
            metrics["answer_faithfulness"].append(answer_metrics["faithfulness"])
            metrics["answer_completeness"].append(answer_metrics["completeness"])
            
            metrics["latency"].append(elapsed)
        
        return {k: np.mean(v) for k, v in metrics.items()}