3.3 混合检索策略优化 — GraphRAG知识图谱增强

本节导读：本节深入探讨GraphRAG系统中的混合检索策略优化方法，通过结合图结构检索和传统文本检索的优势，构建更精准、更高效的检索增强系统。

学习目标

• 理解混合检索的核心概念和优势
• 掌握图检索与文本检索的融合策略
• 学会实现多模态检索的权重调整
• 了解动态检索机制的设计与实现
• 掌握检索效果评估与优化方法

核心概念

混合检索策略优化是GraphRAG系统的核心技术之一，通过结合不同检索方法的优点，实现更全面、更精准的信息检索。

混合检索（Hybrid Search）

结合图结构检索和传统文本检索的优势，提供更全面的检索能力和更好的用户体验。

多模态检索（Multi-modal Search）

整合文本、图、向量等多种模态的检索能力，满足不同类型的信息需求。

动态检索（Dynamic Search）

根据查询特征和用户反馈动态调整检索策略，实现自适应的检索优化。

权重融合（Weight Fusion）

通过加权融合不同检索方法的结果，实现最优的检索性能。

环境准备 / 前置知识

基础环境配置

# 安装必要的Python库
pip install networkx neo4j py2neo rdflib
pip install numpy scipy scikit-learn
pip install transformers torch sentence-transformers
pip install rank-bm25 faiss-cpu
​

核心依赖库

• NetworkX：图结构和算法库
• Neo4j：图数据库
• Py2neo：Neo4j Python客户端
• RDFLib：RDF数据处理库
• Transformers：预训练模型库
• Sentence-Transformers：句子嵌入库
• FAISS：向量相似度搜索库
• Rank-BM25：BM25文本检索库

数据集准备

import networkx as nx
from sentence_transformers import SentenceTransformer
import numpy as np
from rank_bm25 import BM25Okapi
import faiss

# 创建知识图谱
G = nx.DiGraph()

# 添加实体节点
entities = [
    {"id": "张三", "type": "person", "name": "张三", "age": 25, "occupation": "软件工程师"},
    {"id": "李四", "type": "person", "name": "李四", "age": 30, "occupation": "数据科学家"},
    {"id": "阿里巴巴", "type": "company", "name": "阿里巴巴", "industry": "电商"},
    {"id": "腾讯", "type": "company", "name": "腾讯", "industry": "互联网"},
    {"id": "华为", "type": "company", "name": "华为", "industry": "通信"},
]

for entity in entities:
    G.add_node(entity["id"], **{k: v for k, v in entity.items() if k != "id"})

# 添加关系边
relationships = [
    ("张三", "李四", {"relation": "同事", "company": "阿里巴巴"}),
    ("李四", "腾讯", {"relation": "跳槽", "year": 2020}),
    ("张三", "阿里巴巴", {"relation": "工作", "department": "技术部"}),
    ("张三", "华为", {"relation": "合作", "project": "5G技术"}),
    ("阿里巴巴", "腾讯", {"relation": "竞争", "domain": "电商"}),
]

for u, v, data in relationships:
    G.add_edge(u, v, **data)

# 准备文本数据
documents = [
    "张三是阿里巴巴的软件工程师，负责技术架构设计",
    "李四是数据科学家，在2020年跳槽到腾讯",
    "阿里巴巴是中国最大的电商平台之一",
    "腾讯是中国领先的互联网公司",
    "华为是全球领先的通信设备供应商",
    "张三参与了华为的5G技术合作项目",
    "阿里巴巴和腾讯在电商领域存在竞争关系",
    "李四专注于机器学习和人工智能领域",
]

# 初始化模型
sentence_model = SentenceTransformer('all-MiniLM-L6-v2')
document_embeddings = sentence_model.encode(documents)

# 创建FAISS索引
dimension = document_embeddings.shape[1]
faiss_index = faiss.IndexFlatL2(dimension)
faiss_index.add(document_embeddings.astype('float32'))
​

分步实战

步骤 1：混合检索框架设计

class HybridRetrievalSystem:
    def __init__(self, graph, documents, sentence_model, faiss_index):
        self.graph = graph
        self.documents = documents
        self.sentence_model = sentence_model
        self.faiss_index = faiss_index
        
        # 初始化各检索器
        self.graph_retriever = GraphRetriever(graph)
        self.text_retriever = TextRetriever(documents, sentence_model, faiss_index)
        self.semantic_retriever = SemanticRetriever(graph, sentence_model)
        
        # 检索权重
        self.weights = {
            'graph': 0.3,
            'text': 0.4,
            'semantic': 0.3
        }
        
        # 结果缓存
        self.cache = {}
    
    def search(self, query, top_k=10, strategy='weighted'):
        """
        混合检索主方法
        :param query: 查询字符串
        :param top_k: 返回结果数量
        :param strategy: 融合策略 (weighted, rank_fusion, score_norm)
        :return: 混合检索结果
        """
        cache_key = f"{hash(query)}_{top_k}_{strategy}"
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        # 并行执行多种检索
        graph_results = self.graph_retriever.search(query, top_k)
        text_results = self.text_retriever.search(query, top_k)
        semantic_results = self.semantic_retriever.search(query, top_k)
        
        # 根据策略融合结果
        if strategy == 'weighted':
            final_results = self._weighted_fusion(
                graph_results, text_results, semantic_results, top_k
            )
        elif strategy == 'rank_fusion':
            final_results = self._rank_fusion(
                graph_results, text_results, semantic_results, top_k
            )
        elif strategy == 'score_norm':
            final_results = self._score_norm_fusion(
                graph_results, text_results, semantic_results, top_k
            )
        else:
            raise ValueError(f"Unknown strategy: {strategy}")
        
        # 缓存结果
        self.cache[cache_key] = final_results
        return final_results
    
    def adjust_weights(self, graph_weight, text_weight, semantic_weight):
        """调整检索权重"""
        total = graph_weight + text_weight + semantic_weight
        self.weights = {
            'graph': graph_weight / total,
            'text': text_weight / total,
            'semantic': semantic_weight / total
        }
    
    def _weighted_fusion(self, graph_results, text_results, semantic_results, top_k):
        """加权融合策略"""
        # 合并所有结果
        all_results = {}
        
        # 添加图检索结果
        for result in graph_results:
            entity_id = result['entity_id']
            score = result['score'] * self.weights['graph']
            if entity_id in all_results:
                all_results[entity_id]['score'] += score
                all_results[entity_id]['sources'].append('graph')
            else:
                all_results[entity_id] = {
                    'score': score,
                    'entity_id': entity_id,
                    'entity_name': result['entity_name'],
                    'sources': ['graph']
                }
        
        # 添加文本检索结果
        for result in text_results:
            doc_id = result['doc_id']
            score = result['score'] * self.weights['text']
            if doc_id in all_results:
                all_results[doc_id]['score'] += score
                all_results[doc_id]['sources'].append('text')
            else:
                all_results[doc_id] = {
                    'score': score,
                    'doc_id': doc_id,
                    'doc_content': result['doc_content'],
                    'sources': ['text']
                }
        
        # 添加语义检索结果
        for result in semantic_results:
            entity_id = result['entity_id']
            score = result['score'] * self.weights['semantic']
            if entity_id in all_results:
                all_results[entity_id]['score'] += score
                all_results[entity_id]['sources'].append('semantic')
            else:
                all_results[entity_id] = {
                    'score': score,
                    'entity_id': entity_id,
                    'entity_name': result['entity_name'],
                    'sources': ['semantic']
                }
        
        # 排序并返回top_k结果
        sorted_results = sorted(all_results.values(), key=lambda x: x['score'], reverse=True)
        return sorted_results[:top_k]
    
    def _rank_fusion(self, graph_results, text_results, semantic_results, top_k):
        """排序融合策略"""
        # 为每种检索方法分配排名分数
        rank_scores = {}
        
        # 图检索排名
        for i, result in enumerate(graph_results):
            entity_id = result['entity_id']
            rank_score = 1.0 / (i + 1)
            if entity_id in rank_scores:
                rank_scores[entity_id] += rank_score * self.weights['graph']
            else:
                rank_scores[entity_id] = rank_score * self.weights['graph']
        
        # 文本检索排名
        for i, result in enumerate(text_results):
            doc_id = result['doc_id']
            rank_score = 1.0 / (i + 1)
            if doc_id in rank_scores:
                rank_scores[doc_id] += rank_score * self.weights['text']
            else:
                rank_scores[doc_id] = rank_score * self.weights['text']
        
        # 语义检索排名
        for i, result in enumerate(semantic_results):
            entity_id = result['entity_id']
            rank_score = 1.0 / (i + 1)
            if entity_id in rank_scores:
                rank_scores[entity_id] += rank_score * self.weights['semantic']
            else:
                rank_scores[entity_id] = rank_score * self.weights['semantic']
        
        # 排序并返回top_k结果
        sorted_results = sorted(rank_scores.items(), key=lambda x: x[1], reverse=True)
        return [{'entity_id': k, 'score': v} for k, v in sorted_results[:top_k]]
    
    def _score_norm_fusion(self, graph_results, text_results, semantic_results, top_k):
        """分数归一化融合策略"""
        # 归一化每种检索方法的分数
        def normalize_scores(results):
            if not results:
                return []
            scores = [r['score'] for r in results]
            max_score = max(scores)
            min_score = min(scores)
            if max_score == min_score:
                return [r for r in results]
            return [
                {
                    **r,
                    'normalized_score': (r['score'] - min_score) / (max_score - min_score)
                }
                for r in results
            ]
        
        # 归一化各方法分数
        norm_graph = normalize_scores(graph_results)
        norm_text = normalize_scores(text_results)
        norm_semantic = normalize_scores(semantic_results)
        
        # 融合归一化分数
        all_results = {}
        
        for result in norm_graph:
            entity_id = result['entity_id']
            score = result['normalized_score'] * self.weights['graph']
            if entity_id in all_results:
                all_results[entity_id] += score
            else:
                all_results[entity_id] = score
        
        for result in norm_text:
            doc_id = result['doc_id']
            score = result['normalized_score'] * self.weights['text']
            if doc_id in all_results:
                all_results[doc_id] += score
            else:
                all_results[doc_id] = score
        
        for result in norm_semantic:
            entity_id = result['entity_id']
            score = result['normalized_score'] * self.weights['semantic']
            if entity_id in all_results:
                all_results[entity_id] += score
            else:
                all_results[entity_id] = score
        
        # 排序并返回top_k结果
        sorted_results = sorted(all_results.items(), key=lambda x: x[1], reverse=True)
        return [{'entity_id': k, 'score': v} for k, v in sorted_results[:top_k]]

# 使用示例
hybrid_system = HybridRetrievalSystem(G, documents, sentence_model, faiss_index)

# 调整权重
hybrid_system.adjust_weights(0.4, 0.3, 0.3)

# 执行混合检索
query = "张三的工作"
results = hybrid_system.search(query, top_k=5, strategy='weighted')

print(f"查询：{query}")
print(f"混合检索结果（加权融合）：")
for i, result in enumerate(results):
    print(f"{i+1}. {result}")
​

步骤 2：图检索器实现

class GraphRetriever:
    def __init__(self, graph):
        self.graph = graph
        self.entity_index = {}
        self._build_entity_index()
    
    def _build_entity_index(self):
        """构建实体索引"""
        for node_id, data in self.graph.nodes(data=True):
            # 构建实体名称和属性索引
            entity_name = data.get('name', node_id)
            entity_type = data.get('type', 'entity')
            
            # 名称索引
            if entity_name not in self.entity_index:
                self.entity_index[entity_name] = []
            self.entity_index[entity_name].append(node_id)
            
            # 类型索引
            if 'type' not in self.entity_index:
                self.entity_index['type'] = {}
            if entity_type not in self.entity_index['type']:
                self.entity_index['type'][entity_type] = []
            self.entity_index['type'][entity_type].append(node_id)
            
            # 属性索引
            for key, value in data.items():
                if key != 'name' and key != 'type':
                    if key not in self.entity_index:
                        self.entity_index[key] = {}
                    if value not in self.entity_index[key]:
                        self.entity_index[key][value] = []
                    self.entity_index[key][value].append(node_id)
    
    def search(self, query, top_k=10):
        """基于图的检索"""
        query_lower = query.lower()
        results = []
        
        # 1. 实体名称匹配
        if query_lower in self.entity_index:
            for node_id in self.entity_index[query_lower]:
                node_data = self.graph.nodes[node_id]
                results.append({
                    'entity_id': node_id,
                    'entity_name': node_data.get('name', node_id),
                    'entity_type': node_data.get('type', 'entity'),
                    'score': 1.0,  # 完全匹配
                    'match_type': 'name'
                })
        
        # 2. 实体类型匹配
        if 'type' in self.entity_index and query_lower in self.entity_index['type']:
            for node_id in self.entity_index['type'][query_lower]:
                node_data = self.graph.nodes[node_id]
                results.append({
                    'entity_id': node_id,
                    'entity_name': node_data.get('name', node_id),
                    'entity_type': node_data.get('type', 'entity'),
                    'score': 0.8,  # 类型匹配
                    'match_type': 'type'
                })
        
        # 3. 属性值匹配
        for key, value_map in self.entity_index.items():
            if key not in ['type']:  # 已经处理过类型
                for value, node_ids in value_map.items():
                    if query_lower in value.lower():
                        for node_id in node_ids:
                            node_data = self.graph.nodes[node_id]
                            results.append({
                                'entity_id': node_id,
                                'entity_name': node_data.get('name', node_id),
                                'entity_type': node_data.get('type', 'entity'),
                                'score': 0.6,  # 属性匹配
                                'match_type': f'attribute:{key}'
                            })
        
        # 4. 关系路径匹配
        path_results = self._search_by_path(query)
        results.extend(path_results)
        
        # 去重并排序
        unique_results = {}
        for result in results:
            entity_id = result['entity_id']
            if entity_id in unique_results:
                # 保留更高分数的结果
                if result['score'] > unique_results[entity_id]['score']:
                    unique_results[entity_id] = result
            else:
                unique_results[entity_id] = result
        
        # 排序并返回top_k结果
        sorted_results = sorted(unique_results.values(), key=lambda x: x['score'], reverse=True)
        return sorted_results[:top_k]
    
    def _search_by_path(self, query):
        """基于路径的搜索"""
        query_lower = query.lower()
        results = []
        
        # 查找包含查询字符串的关系
        for u, v, edge_data in self.graph.edges(data=True):
            relation = edge_data.get('relation', '')
            
            if query_lower in relation.lower():
                # 获取实体信息
                u_data = self.graph.nodes[u]
                v_data = self.graph.nodes[v]
                
                # 双向结果
                results.append({
                    'entity_id': u,
                    'entity_name': u_data.get('name', u),
                    'entity_type': u_data.get('type', 'entity'),
                    'score': 0.5,
                    'match_type': 'relation',
                    'relation': relation,
                    'related_entity': v
                })
                
                results.append({
                    'entity_id': v,
                    'entity_name': v_data.get('name', v),
                    'entity_type': v_data.get('type', 'entity'),
                    'score': 0.5,
                    'match_type': 'relation',
                    'relation': relation,
                    'related_entity': u
                })
        
        return results
    
    def find_entities_by_relation(self, entity_id, relation_type):
        """根据关系查找实体"""
        related_entities = []
        
        # 出边关系
        for neighbor, edge_data in self.graph[entity_id].items():
            if edge_data.get('relation') == relation_type:
                neighbor_data = self.graph.nodes[neighbor]
                related_entities.append({
                    'entity_id': neighbor,
                    'entity_name': neighbor_data.get('name', neighbor),
                    'entity_type': neighbor_data.get('type', 'entity'),
                    'relation_data': edge_data
                })
        
        # 入边关系
        for predecessor, edge_data in self.graph.predecessors(entity_id):
            if edge_data.get('relation') == relation_type:
​