混合RAG：图+向量检索的协同威力

问题: "爱因斯坦的博士导师是谁？"

向量检索: 搜索"爱因斯坦 博士 导师" → 
可能找到: "爱因斯坦在苏黎世联邦理工学院学习" (语义相关但缺关键信息)
可能找不到: "Alfred Kleiner是爱因斯坦的博士论文导师" (语义距离较远)

问题: "量子力学的哲学意义是什么？"

图检索: 需要遍历"量子力学"→"哲学意义"的边 →
但"哲学意义"是一个抽象概念，知识图谱中可能没有对应的节点

问题: "爱因斯坦的博士导师的研究领域是什么？"

混合检索:
1. 向量检索: "爱因斯坦 博士 导师" → 找到相关文档
2. 实体识别: 识别出"Alfred Kleiner"
3. 图检索: 查询"Alfred Kleiner" → "研究领域" → "实验物理学"
4. 向量检索: "Alfred Kleiner 实验物理学" → 找到详细描述
5. 综合答案: "爱因斯坦的博士导师Alfred Kleiner的研究领域是实验物理学..."

用户问题
    ↓
┌─────────────────┐
│  查询分析器      │ → 识别实体、关系、意图
└────────┬────────┘
         ↓
┌─────────────────────────────────────┐
│           混合检索引擎                │
├──────────┬──────────┬───────────────┤
│ 向量检索  │ 图谱检索  │  关键词检索    │
└──────────┴──────────┴───────────────┘
         ↓
┌─────────────────┐
│  结果融合器      │ → 排序、去重、互补
└────────┬────────┘
         ↓
┌─────────────────┐
│  推理生成器      │ → 基于融合结果生成答案
└─────────────────┘

class QueryAnalyzer:
    async def analyze(self, question):
        """分析查询，提取检索线索"""
        # 1. 实体识别
        entities = await self.ner.extract(question)
        
        # 2. 关系识别
        relations = await self.relation_extractor.extract(question, entities)
        
        # 3. 意图识别
        intent = await self.intent_classifier.classify(question)
        
        # 4. 生成不同检索策略的查询
        return {
            "entities": entities,
            "relations": relations,
            "intent": intent,
            "vector_query": await self.generate_vector_query(question, entities),
            "graph_query": self.generate_graph_query(entities, relations),
            "keyword_query": await self.generate_keyword_query(question)
        }

class KnowledgeGraphBuilder:
    async def build_from_documents(self, documents):
        """从文档构建知识图谱"""
        for doc in documents:
            # 1. 实体抽取
            entities = await self.entity_extractor.extract(doc.text)
            
            # 2. 关系抽取
            relations = await self.relation_extractor.extract(doc.text, entities)
            
            # 3. 添加到图谱
            for entity in entities:
                await self.graph.add_node(
                    id=entity.id,
                    label=entity.text,
                    type=entity.type,
                    properties={"source": doc.id, "embedding": entity.embedding}
                )
            
            for relation in relations:
                await self.graph.add_edge(
                    source=relation.subject,
                    target=relation.object,
                    label=relation.predicate,
                    properties={"confidence": relation.confidence}
                )
            
            # 4. 同时添加到向量库
            await self.vector_store.add(
                id=doc.id,
                text=doc.text,
                embedding=doc.embedding,
                metadata={"entities": [e.id for e in entities]}
            )

class HybridRetrievalEngine:
    def __init__(self, vector_store, graph_store, keyword_index):
        self.vector_store = vector_store
        self.graph_store = graph_store
        self.keyword_index = keyword_index
    
    async def retrieve(self, query_analysis, top_k=10):
        # 1. 并行执行三种检索
        vector_task = self.vector_search(query_analysis["vector_query"], top_k*2)
        graph_task = self.graph_search(query_analysis)
        keyword_task = self.keyword_search(query_analysis["keyword_query"], top_k*2)
        
        vector_results, graph_results, keyword_results = await asyncio.gather(
            vector_task, graph_task, keyword_task
        )
        
        # 2. 融合结果
        fused = self.fuse_results(vector_results, graph_results, keyword_results)
        
        # 3. 重排
        reranked = await self.rerank(fused, query_analysis)
        
        return reranked[:top_k]
    
    async def graph_search(self, query_analysis):
        """图谱检索"""
        results = []
        
        # 1. 实体匹配：在图谱中找到查询实体的对应节点
        for entity in query_analysis["entities"]:
            matched_nodes = await self.graph_store.find_nodes(
                label=entity.text,
                fuzzy=True
            )
            
            # 2. 关系遍历：沿着关系边扩展
            for node in matched_nodes:
                for relation in query_analysis["relations"]:
                    neighbors = await self.graph_store.traverse(
                        start=node,
                        edge_label=relation.predicate,
                        max_depth=2
                    )
                    results.extend(neighbors)
        
        # 3. 获取关联文档
        for result in results:
            if result.has_property("source"):
                doc = await self.document_store.get(result.source)
                result.content = doc.text
        
        return results
    
    def fuse_results(self, vector_results, graph_results, keyword_results):
        """融合三种检索结果"""
        fused = {}
        
        # 为每种检索结果分配权重
        weights = {"vector": 0.4, "graph": 0.4, "keyword": 0.2}
        
        for result_type, results, weight in [
            ("vector", vector_results, weights["vector"]),
            ("graph", graph_results, weights["graph"]),
            ("keyword", keyword_results, weights["keyword"])
        ]:
            for rank, result in enumerate(results):
                doc_id = result.id
                if doc_id not in fused:
                    fused[doc_id] = {
                        "result": result,
                        "score": 0,
                        "sources": []
                    }
                # 倒数排名融合，加上权重
                fused[doc_id]["score"] += weight * (1 / (60 + rank))
                fused[doc_id]["sources"].append(result_type)
        
        # 按融合分数排序
        sorted_results = sorted(fused.values(), key=lambda x: -x["score"])
        
        # 标记同时被多种检索命中的文档（置信度更高）
        for item in sorted_results:
            item["multi_source"] = len(item["sources"]) > 1
        
        return [item["result"] for item in sorted_results]

class CommunityDetector:
    async def detect_communities(self, graph):
        """使用Leiden算法检测社区"""
        communities = await self.leiden_algorithm(graph)
        
        # 为每个社区生成摘要
        for community in communities:
            summary = await self.summarize_community(community)
            community.summary = summary
        
        return communities
    
    async def summarize_community(self, community):
        """生成社区摘要"""
        entities_text = "\n".join([f"- {n.label} ({n.type})" for n in community.nodes])
        relations_text = "\n".join([f"- {e.source} {e.label} {e.target}" for e in community.edges])
        
        prompt = f"""
        以下是一组相关实体和关系：
        
        实体:
        {entities_text}
        
        关系:
        {relations_text}
        
        请用一段话总结这组实体和关系的主题。
        """
        
        return await self.llm.call(prompt)

全局层级: 整个图谱的摘要
  ↓
社区层级: 每个社区的摘要
  ↓
实体层级: 每个实体的描述

class GraphRAGQueryRouter:
    async def route(self, question):
        """根据问题类型路由到不同的检索策略"""
        # 1. 全局性问题（"这个领域的主要趋势是什么？"）
        if self.is_global_question(question):
            return await self.global_search(question)
        
        # 2. 局部性问题（"X和Y的关系是什么？"）
        elif self.is_local_question(question):
            return await self.local_search(question)
        
        # 3. 混合问题
        else:
            return await self.hybrid_search(question)
    
    async def global_search(self, question):
        """全局搜索：基于社区摘要"""
        # 遍历所有社区摘要，找到与问题相关的社区
        relevant_communities = await self.find_relevant_communities(question)
        
        # 基于相关社区的摘要生成答案
        answer = await self.llm.generate(
            question=question,
            context=[c.summary for c in relevant_communities]
        )
        
        return answer
    
    async def local_search(self, question):
        """局部搜索：基于实体和关系"""
        # 1. 找到问题中的实体
        entities = await self.extract_entities(question)
        
        # 2. 在图谱中找到这些实体
        nodes = []
        for entity in entities:
            matches = await self.graph_store.find_nodes(label=entity, fuzzy=True)
            nodes.extend(matches)
        
        # 3. 获取这些实体的邻居和关系
        subgraph = await self.graph_store.get_subgraph(nodes, max_depth=2)
        
        # 4. 基于子图生成答案
        answer = await self.llm.generate(
            question=question,
            context=self.format_subgraph(subgraph)
        )
        
        return answer

class DynamicGraphUpdater:
    async def update_with_new_document(self, document):
        """新文档到达时更新图谱"""
        # 1. 提取实体和关系
        entities = await self.entity_extractor.extract(document.text)
        relations = await self.relation_extractor.extract(document.text, entities)
        
        # 2. 实体对齐：将新实体与已有实体匹配
        for entity in entities:
            existing = await self.find_existing_entity(entity)
            if existing:
                # 合并实体
                await self.merge_entities(existing, entity)
            else:
                # 创建新实体
                await self.graph_store.add_node(entity)
        
        # 3. 添加新关系
        for relation in relations:
            await self.graph_store.add_edge(relation)
        
        # 4. 更新社区结构
        await self.community_detector.incremental_update()

class HybridRAGEvaluator:
    async def evaluate(self, test_cases):
        metrics = {
            "entity_accuracy": [],    # 实体识别准确率
            "relation_accuracy": [],  # 关系抽取准确率
            "retrieval_recall": [],   # 检索召回率
            "answer_accuracy": [],    # 答案准确率
            "reasoning_accuracy": [], # 推理准确率
            "latency": []             # 延迟
        }
        
        for case in test_cases:
            result = await self.system.query(case.question)
            
            # 评估
            metrics["entity_accuracy"].append(
                self.eval_entities(result.entities, case.expected_entities)
            )
            metrics["relation_accuracy"].append(
                self.eval_relations(result.relations, case.expected_relations)
            )
            metrics["answer_accuracy"].append(
                await self.eval_answer(result.answer, case.expected_answer)
            )
            # ...
        
        return {k: np.mean(v) for k, v in metrics.items()}