HiRAG问答流程深入分析

---

前言

近期一直在做知识图谱问答，参考了不少前沿的GraphRAG论文思想，这篇文章，主要深入分析下HiRAG是如何工作的。

Github: https://github.com/hhy-huang/HiRAG
论文：HiRAG: Retrieval-Augmented Generation with Hierarchical Knowledge.

项目结构

对GraphRAG不熟悉的同学，可以先看看 LightRAG 或者 微软的 GraphRAG

知识图谱构建

整个知识图谱构建如上，其中，实体关系抽取与LightRAG中的实体关系抽取，是没什么区别的，不一样的是，在HiRAG中，会对抽取出来的实体，进行实体层级聚类，步骤如下：

通过稀疏度（用来衡量聚类结果的分散程度）作为聚类的停止条件

# 计算聚类稀疏度
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}")breakif 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.98logging.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的社区发现

流程如下

社区检测的层次结构

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"}]
}

理想情况下，社区发现执行完成后，高层概念的实体，应该是先前基于实体层级聚类生成的实体。

社区报告生成

报告结构

@dataclass
class 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构建出了既有严密逻辑又有语义连贯性的层次化知识结构

问答

数据类型组合: 本地知识 + 全局知识 + 桥接知识 + 原始文档

桥接知识是HiRAG的关键创新，用于发现跨社区的隐式连接和推理链路。它解决了传统RAG无法处理的复杂推理问题。

什么是推理路径？

桥接知识的作用：

• 连接不同领域: 发现看似无关领域间的潜在联系
• 支持复杂推理: 构建多跳推理链条
• 增强理解深度: 提供更全面的知识背景
• 发现隐含关系: 揭示间接但重要的关系

HiRAG的实际实现基于社区间关键实体的连接路径发现：

# 来源：hirag/_op.py 中 _build_hierarchical_query_context 函数def find_path_with_required_nodes(graph, source, target, required_nodes):"""在必经节点间构建连接路径"""# inital final pathfinal_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_nodecontinue# 合并路径（避免重复添加当前节点）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 communitycur_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无法实现的核心创新！

信息流转图