3.2 路径推理与上下文扩展 — GraphRAG知识图谱增强

本节导读：本节深入讲解GraphRAG系统中的路径推理技术，通过分析知识图谱中的实体关系路径，实现上下文扩展和深度语义推理。

学习目标

• 理解路径推理的核心原理和算法基础
• 掌握多跳推理和上下文扩展技术
• 学会实现基于路径的语义相似度计算
• 了解图结构化查询语言的使用

核心概念

路径推理与上下文扩展是GraphRAG系统的高级功能，通过利用知识图谱中的路径信息进行深度语义推理。

路径推理（Path Reasoning）

基于知识图谱中的实体关系路径，通过图遍历和路径分析技术，发现实体间的深层语义关联。

多跳推理（Multi-hop Reasoning）

通过多步实体关系跳转，实现复杂语义问题的逐步推理和解答。

上下文扩展（Context Expansion）

基于检索到的核心实体，通过路径分析扩展相关实体和关系，为后续生成提供更丰富的上下文信息。

环境准备 / 前置知识

基础环境配置

# 安装必要的Python库
pip install networkx neo4j py2neo rdflib
pip install numpy scipy scikit-learn
​

核心依赖库

• NetworkX：图结构和算法库
• Neo4j：图数据库
• Py2neo：Neo4j Python客户端
• RDFLib：RDF数据处理库

数据集准备

import networkx as nx

# 创建有向图
G = nx.DiGraph()

# 添加实体节点
G.add_node("张三", type="person", age=25, occupation="软件工程师")
G.add_node("李四", type="person", age=30, occupation="数据科学家")
G.add_node("阿里巴巴", type="company", industry="电商")
G.add_node("腾讯", type="company", industry="互联网")

# 添加关系边
G.add_edge("张三", "李四", relation="同事", company="阿里巴巴")
G.add_edge("李四", "腾讯", relation="跳槽", year=2020)
G.add_edge("张三", "阿里巴巴", relation="工作", department="技术部")
​

分步实战

步骤 1：基础路径搜索算法

广度优先搜索（BFS）实现

from collections import deque

def find_paths_bfs(graph, start_node, end_node, max_hops=3):
    """使用BFS查找所有路径"""
    paths = []
    queue = deque([(start_node, [start_node])])
    
    while queue:
        current_node, path = queue.popleft()
        
        if len(path) > max_hops + 1:
            continue
            
        if current_node == end_node and len(path) > 1:
            paths.append(path)
            continue
            
        for neighbor, edge_data in graph[current_node].items():
            if neighbor not in path:
                new_path = path + [neighbor]
                queue.append((neighbor, new_path))
    
    return paths

# 使用示例
paths = find_paths_bfs(G, "张三", "腾讯")
print(f"从张三到腾讯的路径: {paths}")
​

深度优先搜索（DFS）实现

def find_paths_dfs(graph, start_node, end_node, max_hops=3):
    """使用DFS查找所有路径"""
    paths = []
    
    def dfs(current, path, depth):
        if depth > max_hops:
            return
            
        if current == end_node and len(path) > 1:
            paths.append(path.copy())
            return
            
        for neighbor, edge_data in graph[current].items():
            if neighbor not in path:
                path.append(neighbor)
                dfs(neighbor, path, depth + 1)
                path.pop()
    
    dfs(start_node, [start_node], 0)
    return paths

# 使用示例
paths = find_paths_dfs(G, "张三", "腾讯")
print(f"从张三到腾讯的路径: {paths}")
​

步骤 2：路径权重计算和排序

def calculate_path_scores(graph, paths):
    """计算路径权重和排序"""
    scored_paths = []
    
    for path in paths:
        score = 0
        relations = []
        
        for i in range(len(path) - 1):
            u, v = path[i], path[i + 1]
            edge_data = graph[u][v]
            
            # 根据关系类型计算权重
            relation = edge_data.get('relation', '未知')
            relations.append(relation)
            
            # 关系权重
            relation_weights = {
                '同事': 1.0,
                '跳槽': 0.8,
                '工作': 0.9,
                '朋友': 0.7,
                '合作': 0.85
            }
            score += relation_weights.get(relation, 0.5)
        
        # 路径长度惩罚
        length_penalty = 1.0 / len(path)
        score *= length_penalty
        
        scored_paths.append({
            'path': path,
            'relations': relations,
            'score': score,
            'length': len(path)
        })
    
    # 按分数排序
    scored_paths.sort(key=lambda x: x['score'], reverse=True)
    return scored_paths

# 使用示例
paths = find_paths_bfs(G, "张三", "腾讯")
scored_paths = calculate_path_scores(G, paths)
for i, path_info in enumerate(scored_paths[:3]):
    print(f"路径 {i+1}: {' -> '.join(path_info['path'])}")
    print(f"  关系链: {' -> '.join(path_info['relations'])}")
    print(f"  权重分数: {path_info['score']:.3f}")
    print()
​

步骤 3：多跳推理实现

class PathReasoner:
    def __init__(self, graph):
        self.graph = graph
        self.path_cache = {}
    
    def find_reasoning_paths(self, start_entity, target_entity, max_hops=5):
        """多跳推理：查找两个实体之间的推理路径"""
        cache_key = f"{start_entity}->{target_entity}"
        
        if cache_key in self.path_cache:
            return self.path_cache[cache_key]
        
        # 查找所有可能路径
        paths = self._find_all_paths(start_entity, target_entity, max_hops)
        
        # 过滤和排序路径
        filtered_paths = self._filter_reasonable_paths(paths)
        
        # 生成推理解释
        reasoning_results = []
        for path_info in filtered_paths:
            reasoning = self._generate_reasoning_explanation(path_info)
            reasoning_results.append(reasoning)
        
        result = {
            'start_entity': start_entity,
            'target_entity': target_entity,
            'paths_found': len(paths),
            'reasonable_paths': len(filtered_paths),
            'reasoning_results': reasoning_results[:5]
        }
        
        self.path_cache[cache_key] = result
        return result
    
    def _find_all_paths(self, start, target, max_hops):
        """查找所有可能的路径"""
        paths = []
        queue = [(start, [start])]
        
        while queue:
            current, path = queue.pop(0)
            
            if len(path) > max_hops + 1:
                continue
                
            if current == target and len(path) > 1:
                paths.append(path)
                continue
                
            for neighbor in self.graph.neighbors(current):
                if neighbor not in path:
                    new_path = path + [neighbor]
                    queue.append((neighbor, new_path))
        
        return paths
    
    def _filter_reasonable_paths(self, paths):
        """过滤合理的路径"""
        reasonable_paths = []
        
        for path in paths:
            # 检查路径长度合理性
            if len(path) <= 6:  # 最多5跳
                if self._is_path_reasonable(path):
                    reasonable_paths.append(path)
        
        return reasonable_paths
    
    def _is_path_reasonable(self, path):
        """检查路径是否合理"""
        # 避免重复实体
        return len(set(path)) == len(path)
    
    def _generate_reasoning_explanation(self, path_info):
        """生成推理解释"""
        path = path_info['path']
        relations = []
        
        for i in range(len(path) - 1):
            u, v = path[i], path[i + 1]
            edge_data = self.graph[u][v]
            relations.append(edge_data.get('relation', '关联'))
        
        explanation = {
            'path': path,
            'relations': relations,
            'explanation': self._natural_language_explanation(path, relations)
        }
        
        return explanation
    
    def _natural_language_explanation(self, path, relations):
        """生成自然语言解释"""
        if len(path) == 2:
            return f"直接关系：{path[0]} 通过 {relations[0]} 关系连接到 {path[1]}"
        elif len(path) == 3:
            return f"两跳关系：{path[0]} 通过 {relations[0]} 连接到 {path[1]}，{path[1]} 通过 {relations[1]} 连接到 {path[2]}"
        else:
            return f"多跳关系：{' -> '.join(path)}，关系链：{' -> '.join(relations)}"

# 使用示例
reasoner = PathReasoner(G)
result = reasoner.find_reasoning_paths("张三", "腾讯")

print(f"推理结果：张三 -> 腾讯")
print(f"发现路径数：{result['paths_found']}")
print(f"合理路径数：{result['reasonable_paths']}")
print("\n推理解释：")
for i, reasoning in enumerate(result['reasoning_results']):
    print(f"{i+1}. {reasoning['explanation']}")
​

步骤 4：上下文扩展实现

class ContextExpander:
    def __init__(self, graph, reasoner):
        self.graph = graph
        self.reasoner = reasoner
        self.context_cache = {}
    
    def expand_context(self, seed_entities, max_depth=2, max_entities=20):
        """上下文扩展：从种子实体开始扩展相关实体和关系"""
        context = {
            'seed_entities': seed_entities,
            'expanded_entities': set(seed_entities),
            'relations': [],
            'entity_properties': {},
            'paths_to_seeds': {}
        }
        
        # 扩展实体
        self._expand_entities_recursive(seed_entities, context, max_depth, max_entities)
        
        # 收集实体属性
        self._collect_entity_properties(context)
        
        # 收集路径信息
        self._collect_paths_to_seeds(context, seed_entities)
        
        return context
    
    def _expand_entities_recursive(self, entities, context, depth, max_entities):
        """递归扩展实体"""
        if depth <= 0 or len(context['expanded_entities']) >= max_entities:
            return
        
        for entity in entities:
            if entity in context['expanded_entities']:
                continue
                
            # 获取直接相关的实体
            related_entities = self._get_related_entities(entity)
            
            for related_entity, relation_info in related_entities:
                if related_entity not in context['expanded_entities']:
                    context['expanded_entities'].add(related_entity)
                    context['relations'].append({
                        'from': entity,
                        'to': related_entity,
                        'relation': relation_info['relation'],
                        'attributes': relation_info
                    })
                    
                    # 递归扩展
                    self._expand_entities_recursive([related_entity], context, depth - 1, max_entities)
    
    def _get_related_entities(self, entity):
        """获取与指定实体相关的其他实体"""
        related_entities = []
        
        # 出边关系
        for neighbor in self.graph.neighbors(entity):
            edge_data = self.graph[entity][neighbor]
            related_entities.append((neighbor, edge_data))
        
        # 入边关系
        for predecessor in self.graph.predecessors(entity):
            edge_data = self.graph[predecessor][entity]
            related_entities.append((predecessor, edge_data))
        
        return related_entities
    
    def _collect_entity_properties(self, context):
        """收集实体属性"""
        for entity in context['expanded_entities']:
            if entity in self.graph.nodes:
                node_data = self.graph.nodes[entity]
                context['entity_properties'][entity] = node_data
    
    def _collect_paths_to_seeds(self, context, seed_entities):
        """收集到种子的路径"""
        for entity in context['expanded_entities']:
            if entity not in seed_entities:
                paths_to_seeds = []
                for seed in seed_entities:
                    result = self.reasoner.find_reasoning_paths(entity, seed, max_hops=3)
                    if result['reasonable_paths'] > 0:
                        paths_to_seeds.append({
                            'seed': seed,
                            'paths_count': result['reasonable_paths'],
                            'shortest_path_length': min(len(r['path']) for r in result['reasoning_results'])
                        })
                context['paths_to_seeds'][entity] = paths_to_seeds
    
    def get_context_summary(self, context):
        """获取上下文摘要"""
        return {
            'total_entities': len(context['expanded_entities']),
            'total_relations': len(context['relations']),
            'entity_properties_count': len(context['entity_properties']),
            'coverage_percentage': (len(context['expanded_entities']) / 30) * 100
        }

# 使用示例
expander = ContextExpander(G, reasoner)

# 从种子实体开始扩展
seed_entities = ["张三", "阿里巴巴"]
context = expander.expand_context(seed_entities, max_depth=2, max_entities=15)

print(f"上下文扩展结果：")
print(f"种子实体：{context['seed_entities']}")
print(f"扩展实体数：{len(context['expanded_entities'])}")
print(f"关系数：{len(context['relations'])}")
print(f"实体属性数：{len(context['entity_properties'])}")

summary = expander.get_context_summary(context)
print(f"\n上下文摘要：")
print(f"总实体数：{summary['total_entities']}")
print(f"总关系数：{summary['total_relations']}")
print(f"覆盖率：{summary['coverage_percentage']:.1f}%")
​

步骤 5：Neo4j图数据库集成

from neo4j import GraphDatabase

class Neo4jPathReasoner:
    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
    
    def close(self):
        self.driver.close()
    
    def create_knowledge_graph(self, G):
        """在Neo4j中创建知识图谱"""
        with self.driver.session() as session:
            # 删除现有数据
            session.run("MATCH (n) DETACH DELETE n")
            
            # 创建实体节点
            for node, data in G.nodes(data=True):
                node_type = data.get('type', 'entity')
                properties = {k: v for k, v in data.items() if k != 'type'}
                
                query = f"""
                CREATE (n:{node_type} {{id: $id, $properties}})
                """
                session.run(query, id=node, properties=properties)
            
            # 创建关系边
            for u, v, data in G.edges(data=True):
                relation = data.get('relation', 'related')
                properties = {k: v for k, v in data.items() if k != 'relation'}
                
                query = f"""
                MATCH (a), (b)
                WHERE a.id = $from_id AND b.id = $to_id
                CREATE (a)-[r:{relation} $properties]->(b)
                """
                session.run(query, from_id=u, to_id=v, properties=properties)
    
    def query_paths_cypher(self, start_entity, end_entity, max_hops=5):
        """使用Cypher查询路径"""
        query = f"""
        MATCH path = (start)-[*1..{max_hops}]-(end)
        WHERE start.id = $start_id AND end.id = $end_id
        RETURN path, length(path) as path_length
        ORDER BY path_length
        LIMIT 10
        """
        
        with self.driver.session() as session:
            result = session.run(query, start_id=start_entity, end_id=end_entity)
            return [record for record in result]

# 使用示例
neo4j_reasoner = Neo4jPathReasoner("bolt://localhost:7687", "neo4j", "password")

# 创建知识图谱（在实际应用中需要真实的数据库连接）
# neo4j_reasoner.create_knowledge_graph(G)

# 查询路径（示例）
# paths = neo4j_reasoner.query_paths_cypher("张三", "腾讯")
# for path_record in paths:
#     print(f"路径长度: {path_record['path_length']}")
#     print(f"路径: {path_record['path']}")

neo4j_reasoner.close()
​

完整示例：企业知识图谱推理系统

class EnterpriseKnowledgeGraphSystem:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.reasoner = PathReasoner(self.graph)
        self.expander = ContextExpander(self.graph, self.reasoner)
        
        # 加载企业知识图谱
        self._load_enterprise_knowledge()
    
    def _load_enterprise_knowledge(self):
        """加载企业知识图谱数据"""
        # 添加实体
        entities = [
            {"id": "员工-001", "type": "员工", "name": "张三", "department": "技术部", "position": "高级工程师"},
            {"id": "员工-002", "type": "员工", "name": "李四", "department": "产品部", "position": "产品经理"},
            {"id": "部门-001", "type": "部门", "name": "技术部", "parent": "公司总部"},
            {"id": "部门-002", "type": "部门", "name": "产品部", "parent": "公司总部"},
            {"id": "项目-001", "type": "项目", "name": "智能客服系统", "department": "技术部", "status": "进行中"},
            {"id": "技术-001", "type": "技术", "name": "机器学习", "domain": "AI"},
        ]
        
        for entity in entities:
            self.graph.add_node(entity["id"], **{k: v for k, v in entity.items() if k != "id"})
        
        # 添加关系
        relationships = [
            ("员工-001", "部门-001", {"relation": "属于", "start_date": "2020-01-01"}),
            ("员工-002", "部门-002", {"relation": "属于", "start_date": "2019-06-15"}),
            ("部门-001", "公司总部", {"relation": "隶属于"}),
            ("部门-002", "公司总部", {"relation": "隶属于"}),
            ("员工-001", "项目-001", {"relation": "参与", "role": "技术负责人"}),
            ("员工-001", "技术-001", {"relation": "掌握", "level": "精通"}),
        ]
        
        for u, v, data in relationships:
            self.graph.add_edge(u, v, **data)
    
    def find_expertise_chain(self, employee_id, target_technology):
        """查找从员工到目标技术的专业能力链"""
        result = self.reasoner.find_reasoning_paths(employee_id, target_technology, max_hops=4)
        
        if result['reasonable_paths'] > 0:
            print(f"从员工 {employee_id} 到技术 {target_technology} 的推理路径：")
            for i, reasoning in enumerate(result['reasoning_results'][:3]):
                print(f"{i+1}. {reasoning['explanation']}")
        els
​