本文深入分析HiRAG的工作机制，该机制融合了前沿的GraphRAG论文思想，尤其关注其在知识图谱问答中的应用。
Github:https://github.com/hhy-huang/HiRAG论文：HiRAG: Retrieval-Augmented Generation with Hierarchical Knowledge.
项目结构
对于不熟悉GraphRAG的读者，建议先行了解LightRAG或微软的GraphRAG。
知识图谱构建
graph TD subgraph "知识图谱构建" B1[文档预处理] --> B2[文本分块] B2 --> B3[实体提取] B3 --> B4[关系抽取] B4 --> B5[图谱构建] B5 --> B6[社区检测] B6 --> B7[社区报告生成] end如图所示，HiRAG的知识图谱构建流程涵盖了实体关系抽取，这部分与LightRAG中的方法相似。其独特之处在于，HiRAG会对抽取的实体进行层级聚类，具体步骤如下：
graph TD A[基础实体集合] --> B[Layer 0 聚类] B --> C[高斯混合模型聚类] C --> D[形成实体簇] D --> E[LLM语义总结] E --> F[生成元属性实体] F --> G{停止条件判断} G -->|继续| H[Layer N+1] G -->|结束| I[层次化实体结构] H --> C聚类的停止条件通过稀疏度进行判断，稀疏度用于衡量聚类结果的分散程度。
# 计算聚类稀疏度cluster_sizes = Counter(np.concatenate(clusters))cluster_sparsity = 1 - sum([x * (x - 1) for x in cluster_sizes.values()]) / (len(nodes) * (len(nodes) - 1))# 稀疏度变化率cluster_sparsity_change_rate = abs(cluster_sparsity - pre_cluster_sparsity) / (pre_cluster_sparsity + 1e-8)# 停止条件判断if cluster_sparsity >= threshold: # 稀疏度≥98% logging.info(f"停止聚类，稀疏度达到{cluster_sparsity}") break if cluster_sparsity_change_rate <= thredshold_change_rate: # 变化率≤5% logging.info(f"停止聚类，变化率仅{cluster_sparsity_change_rate}") break稀疏度公式:
稀疏度 = 1 - Σ(ni × (ni-1)) / (N × (N-1))其中:- ni: 第i个簇的大小- N: 总实体数- 分子: 簇内实体对数之和- 分母: 总实体对数稀疏度
可视化示例
情况1：完全聚集 (稀疏度 = 0%)┌─────────────────┐│ ●●●●●● (6个实体) │ ← 1个大簇└─────────────────┘情况2：完全分散 (稀疏度 = 100%) │●│ │●│ │●│ │●│ │●│ │●│ ← 6个独立簇情况3：部分聚集 (稀疏度 = 60%)┌─────┐ ┌─────┐│ ●●● │ │ ●●● │ ← 2个簇，每个3个实体└─────┘ └─────┘稀疏度范围
- 0% (最小值): 所有实体在同一个簇中，完全聚集
- 100% (最大值): 每个实体独立成簇，完全分散
- 中间值: 部分聚集状态
详细计算过程
公式推导
第1步：理解实体对的概念
实体对 = 两个实体之间的关系总实体对数 = N × (N-1) = 从N个实体中选2个的组合数×2第2步：理解簇内实体对
簇内实体对 = 同一个簇中实体之间的配对对于大小为ni的簇，实体对数 = ni × (ni-1)第3步：计算聚集度
聚集度 = (簇内实体对数) / (总实体对数)表示：有多少比例的实体对是在同一个簇内的第4步：计算稀疏度
稀疏度 = 1 - 聚集度表示：有多少比例的实体对是跨簇分散的计算示例
示例A：[4, 2] 分布
实体总数: N = 6簇分布: [4个实体的簇, 2个实体的簇]# 步骤1：计算各簇内实体对数簇1内对数 = 4 × (4-1) = 12簇2内对数 = 2 × (2-1) = 2总簇内对数 = 12 + 2 = 14# 步骤2：计算总实体对数 总对数 = 6 × (6-1) = 30# 步骤3：计算稀疏度稀疏度 = 1 - 14/30 = 0.533 (53.3%)示例B：[2, 2, 2] 分布
实体总数: N = 6 簇分布: [2个实体的簇, 2个实体的簇, 2个实体的簇]# 步骤1：计算各簇内实体对数每簇内对数 = 2 × (2-1) = 2总簇内对数 = 2 + 2 + 2 = 6# 步骤2：计算总实体对数总对数 = 6 × (6-1) = 30# 步骤3：计算稀疏度 稀疏度 = 1 - 6/30 = 0.8 (80%)对比分析:
- [4,2] 分布的稀疏度 = 53.3%
- [2,2,2] 分布的稀疏度 = 80%
在HiRAG中的作用
1. 停止条件判断
# 稀疏度阈值检查if cluster_sparsity >= threshold: # 默认 threshold = 0.98 logging.info(f"停止聚类，稀疏度达到{cluster_sparsity}") break逻辑：当稀疏度≥98%时，说明实体已经非常分散，继续聚类意义不大。
2. 变化率监控
# 稀疏度变化率change_rate = abs(current_sparsity - previous_sparsity) / (previous_sparsity + 1e-8)if change_rate <= 0.05: # 变化率≤5% logging.info(f"停止聚类，改进微小：{change_rate}") break逻辑：当稀疏度变化很小时，说明聚类已经收敛。
稀疏度的意义
信息理论角度
熵的概念：
- 低稀疏度 = 低熵 = 高度有序 = 信息集中
- 高稀疏度 = 高熵 = 高度无序 = 信息分散
在知识图谱中：
- 低稀疏度：实体高度相关，形成紧密的知识簇
- 高稀疏度：实体相互独立，缺乏明显的关联模式
聚类优化角度
理想的聚类过程：
Layer 0: 低稀疏度 (实体相互关联) ↓ 聚类抽象Layer 1: 中等稀疏度 (形成有意义的簇) ↓ 继续抽象Layer N: 高稀疏度 (高度抽象的概念)稀疏度增长趋势：
sparsity_trend = [0.1, 0.3, 0.5, 0.7, 0.85, 0.95, 0.98]# ↑层级越高，稀疏度越大，概念越抽象实际应用示例
科技领域实体聚类
Layer 0(稀疏度 = 15%):
实体: [OpenAI, ChatGPT, GPT-4, Google, Bard, PaLM, Meta, LLaMA, ...]大部分实体都有关联，稀疏度较低Layer 1(稀疏度 = 45%):
抽象后: [AI_COMPANIES, LANGUAGE_MODELS, TECH_PRODUCTS, ...]形成了明显的类别，但类别间仍有关联Layer 2(稀疏度 = 78%):
进一步抽象: [TECHNOLOGY_SECTOR, AI_INDUSTRY, ...]高度抽象的概念，相互独立性较强Layer 3(稀疏度 = 95%):
最终抽象: [INNOVATION_ECOSYSTEM]极少数高度抽象的概念HiRAG的实体层级聚类旨在挖掘高度抽象概念的实体与关系，并将其与先前提取的实体、关系一同插入知识图谱，以丰富图谱数据。值得注意的是，这些高度抽象的实体与关系不作为后续社区检索的目标。
完成实体与关系抽取后，HiRAG将进行社区发现。
社区发现
HiRAG中基于Leiden的社区发现。
流程如下
graph TD A[实体关系图谱] --> B[图聚类层] B --> C[Leiden社区检测] C --> E[图社区结构] E --> G[社区报告生成]社区检测的层次结构
1. 图拓扑层次
# Leiden算法生成的层次结构leiden_hierarchy = { 0: { # Level 0: 基础社区 "cluster_0": ["实体A", "实体B", "实体C"], "cluster_1": ["实体D", "实体E", "实体F"], "cluster_2": ["实体G", "实体H"] }, 1: { # Level 1: 合并社区 "cluster_0": ["cluster_0", "cluster_1"], # 包含Level 0的多个社区 "cluster_1": ["cluster_2"] }}2. 语义抽象层次
# 实体聚类生成的层次结构semantic_hierarchy = { 0: [ # Layer 0: 基础实体 {"entity_name": "OPENAI", "type": "organization"}, {"entity_name": "CHATGPT", "type": "product"}, {"entity_name": "GPT4", "type": "technology"} ], 1: [ # Layer 1: 元实体 {"entity_name": "AI_COMPANY", "type": "organization"}, {"entity_name": "AI_PRODUCTS", "type": "concept"} ], 2: [ # Layer 2: 高层概念 {"entity_name": "AI_ECOSYSTEM", "type": "concept"} ]}理想情况下，社区发现完成后所形成的高层概念实体，应与先前通过实体层级聚类生成的实体相符。
社区报告生成
报告结构
@dataclassclass CommunityReport: title: str # 社区名称 summary: str # 执行摘要 rating: float # 影响评级 (0-10) rating_explanation: str # 评级说明 findings: List[dict] # 详细发现生成流程
async def generate_community_report(community_reports, graph_storage, global_config): """为每个社区生成详细报告""" communities = await graph_storage.community_schema() for community_id, community_data in communities.items(): # 1. 准备社区信息 entities_info = [] for node in community_data["nodes"]: node_data = await graph_storage.get_node(node) entities_info.append({ "name": node, "type": node_data.get("entity_type", "unknown"), "description": node_data.get("description", "") }) # 2. 收集关系信息 relationships_info = [] for edge in community_data["edges"]: edge_data = await graph_storage.get_edge(edge[0], edge[1]) relationships_info.append({ "source": edge[0], "target": edge[1], "description": edge_data.get("description", ""), "strength": edge_data.get("weight", 1.0) }) # 3. 构建报告上下文 report_context = format_community_context(entities_info, relationships_info) # 4. LLM生成报告 llm_func = global_config["best_model_func"] report_prompt = COMMUNITY_REPORT_PROMPT.format(input_text=report_context) report_json_str = await llm_func( report_prompt, **global_config.get("special_community_report_llm_kwargs", {}) ) # 5. 解析和存储 try: report_json = json.loads(report_json_str) community_report = { **community_data, "report_string": report_json_str, "report_json": report_json } await community_reports.upsert({community_id: community_report}) except json.JSONDecodeError: logger.error(f"社区 {community_id} 报告解析失败")报告示例
{ "title": "AI Technology Ecosystem", "summary": "This community represents the artificial intelligence technology ecosystem, including major companies, products, and technologies that are shaping the AI landscape.", "rating": 8.5, "rating_explanation": "High impact due to significant influence on technology industry and society.", "findings": [ { "summary": "Market Leaders in AI", "explanation": "OpenAI, Google, and Meta are the primary organizations driving AI innovation through their large language models and AI platforms." }, { "summary": "Product Innovation", "explanation": "ChatGPT, Bard, and similar AI products are transforming how users interact with AI technology." } ]}通过图聚类和社区检测，HiRAG构建了具备严密逻辑和语义连贯性的层次化知识结构。
问答
数据类型组合: 本地知识 + 全局知识 + 桥接知识 + 原始文档
graph LR A[查询] --> B[实体检索] B --> C[社区报告] B --> D[实体关系] B --> E[桥接路径] B --> F[原始文档] C --> G[多层次上下文] D --> G E --> G F --> G桥接知识是HiRAG的关键创新点，其作用是发现跨社区的隐式连接和推理链路，从而解决传统RAG在处理复杂推理问题时的局限性。
什么是推理路径？
graph TB subgraph "社区A：AI技术" A1[机器学习] A2[深度学习] A3[神经网络] end subgraph "社区B：医疗应用" B1[医学影像] B2[疾病诊断] B3[临床决策] end subgraph "桥接实体" C1[计算机视觉] C2[图像识别] end A2 --> C1 C1 --> C2 C2 --> B1 B1 --> B2 style C1 fill:#ff9999 style C2 fill:#ff9999桥接知识的作用：
- 连接不同领域: 发现看似无关领域间的潜在联系
- 支持复杂推理: 构建多跳推理链条
- 增强理解深度: 提供更全面的知识背景
- 发现隐含关系: 揭示间接但重要的关系
HiRAG的实际实现基于社区间关键实体的连接路径发现：
# 来源：hirag/_op.py 中 _build_hierarchical_query_context 函数def find_path_with_required_nodes(graph, source, target, required_nodes): """在必经节点间构建连接路径""" # inital final path final_path = [] # 起点设置为当前节点 current_node = source # 遍历必经节点 for next_node in required_nodes: # 找到从当前节点到下一个必经节点的最短路径 try: sub_path = nx.shortest_path(graph, source=current_node, target=next_node) except nx.NetworkXNoPath: # raise ValueError(f"No path between {current_node} and {next_node}.") final_path.extend([next_node]) current_node = next_node continue # 合并路径（避免重复添加当前节点） if final_path: final_path.extend(sub_path[1:]) # 从第二个节点开始添加，避免重复 else: final_path.extend(sub_path) # 更新当前节点为下一个必经节点 current_node = next_node # 最后，从最后一个必经节点到目标节点的路径 try: sub_path = nx.shortest_path(graph, source=current_node, target=target) final_path.extend(sub_path[1:]) # 从第二个节点开始添加，避免重复 except nx.NetworkXNoPath: # raise ValueError(f"No path between {current_node} and {target}.") final_path.extend([target]) return final_path# Step 2.1: 从不同社区选择关键实体key_entities = []max_entity_num = query_param.top_mif use_communities: for c in use_communities: cur_community_key_entities = [] community_entities = c['nodes'] # find the top-k entities in this community cur_community_key_entities.extend( [e for e in overall_node_datas if e['entity_name'] in community_entities][:max_entity_num] ) key_entities.append(cur_community_key_entities)else: key_entities = [overall_node_datas[:max_entity_num]]# Step 2.2: 去重和准备关键实体列表key_entities = [[e['entity_name'] for e in k] for k in key_entities]key_entities = list(set([k for kk in key_entities for k in kk]))# Step 2.3: 寻找关键实体间的最短路径try: path = find_path_with_required_nodes( knowledge_graph_inst._graph, key_entities[0], # 源节点 key_entities[-1], # 目标节点 key_entities[1:-1] # 必经的中间节点 ) path = list(set(path)) # 去重 # Step 2.4: 获取路径实体的完整信息 path_datas = await asyncio.gather( *[knowledge_graph_inst.get_node(r) for r in path] ) path_degrees = await asyncio.gather( *[knowledge_graph_inst.node_degree(r) for r in path] ) path_datas = [ {**n, "entity_name": k, "rank": d} for k, n, d in zip(path, path_datas, path_degrees) if n is not None ] except ValueError as e: print(f"路径发现失败: {e}") path_datas = [] path = []HiRAG的推理路径并非简单的“最小连通图”，而是一个精心设计的多社区桥接网络。该网络通过必经的关键节点确保推理的完整性和跨域性，是传统RAG难以实现的核心创新。
graph TD subgraph "简单最短路径" A1[深度学习] --> B1[医学图像] --> C1[癌症诊断] end subgraph "HiRAG推理路径" A2[深度学习] --> D1[神经网络架构] D1 --> B2[卷积神经网络] B2 --> D2[图像处理技术] D2 --> C2[医学影像分析] C2 --> D3[影像诊断方法] D3 --> E2[癌症诊断] end style B2 fill:#ffeb3b style C2 fill:#ffeb3b style A2 fill:#4caf50 style E2 fill:#f44336信息流转图
flowchart LR subgraph "知识构建阶段" A[原始文档] --> B[文本分块] B --> C[实体提取] D[关系抽取] D --> E[图谱构建] E --> F[社区检测] F --> G[报告生成] end subgraph "查询阶段" H[用户查询] --> I[实体检索] I --> J[社区映射] J --> K1[本地知识] J --> K2[全局知识] K3[桥接知识] K1 --> L[上下文整合] K2 --> L K3 --> L L --> M[LLM生成] M --> N[最终答案] end E --> I G --> K2
