前言
近期一直在做知识图谱问答，参考了不少前沿的GraphRAG论文思想，这篇文章，主要深入分析下HiRAG是如何工作的。
项目结构
对GraphRAG不熟悉的同学，可以先看看LightRAG或者 微软的GraphRAG
知识图谱构建
graph TD
subgraph "知识图谱构建"
B1[文档预处理] --> B2[文本分块]
B2 --> B3[实体提取]
B3 --> B4[关系抽取]
B4 --> B5[图谱构建]
B5 --> B6[社区检测]
B6 --> B7[社区报告生成]
end
整个知识图谱构建如上，其中，实体关系抽取与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)forxincluster_sizes.values()]) / (len(nodes) * (len(nodes) -1))
# 稀疏度变化率
cluster_sparsity_change_rate = abs(cluster_sparsity - pre_cluster_sparsity) / (pre_cluster_sparsity +1e-8)
# 停止条件判断
ifcluster_sparsity >= threshold: # 稀疏度≥98%
logging.info(f"停止聚类，稀疏度达到{cluster_sparsity}")
break
ifcluster_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. 停止条件判断
# 稀疏度阈值检查
ifcluster_sparsity >= threshold: # 默认 threshold = 0.98
logging.info(f"停止聚类，稀疏度达到{cluster_sparsity}")
break
逻辑：当稀疏度≥98%时，说明实体已经非常分散，继续聚类意义不大。
2. 变化率监控
# 稀疏度变化率
change_rate = abs(current_sparsity - previous_sparsity) / (previous_sparsity +1e-8)
ifchange_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"}
]
}
理想情况下，社区发现执行完成后，高层概念的实体，应该是先前基于实体层级聚类生成的实体。
社区报告生成
报告结构
@dataclass
classCommunityReport:
title: str # 社区名称
summary: str # 执行摘要
rating: float # 影响评级 (0-10)
rating_explanation: str# 评级说明
findings: List[dict] # 详细发现
生成流程
asyncdefgenerate_community_report(community_reports, graph_storage, global_config):
"""为每个社区生成详细报告"""
communities =awaitgraph_storage.community_schema()
forcommunity_id, community_dataincommunities.items():
# 1. 准备社区信息
entities_info = []
fornodeincommunity_data["nodes"]:
node_data =awaitgraph_storage.get_node(node)
entities_info.append({
"name": node,
"type": node_data.get("entity_type","unknown"),
"description": node_data.get("description","")
})
# 2. 收集关系信息
relationships_info = []
foredgeincommunity_data["edges"]:
edge_data =awaitgraph_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 =awaitllm_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
}
awaitcommunity_reports.upsert({community_id: community_report})
exceptjson.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":"
roduct 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 函数
deffind_path_with_required_nodes(graph, source, target, required_nodes):
"""在必经节点间构建连接路径"""
# inital final path
final_path = []
# 起点设置为当前节点
current_node = source
# 遍历必经节点
fornext_nodeinrequired_nodes:
# 找到从当前节点到下一个必经节点的最短路径
try:
sub_path = nx.shortest_path(graph, source=current_node, target=next_node)
exceptnx.NetworkXNoPath:
# raise ValueError(f"No path between {current_node} and {next_node}.")
final_path.extend([next_node])
current_node = next_node
continue
# 合并路径（避免重复添加当前节点）
iffinal_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:]) # 从第二个节点开始添加，避免重复
exceptnx.NetworkXNoPath:
# raise ValueError(f"No path between {current_node} and {target}.")
final_path.extend([target])
returnfinal_path
# Step 2.1: 从不同社区选择关键实体
key_entities = []
max_entity_num = query_param.top_m
ifuse_communities:
forcinuse_communities:
cur_community_key_entities = []
community_entities = c['nodes']
# find the top-k entities in this community
cur_community_key_entities.extend(
[eforeinoverall_node_datasife['entity_name']incommunity_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']foreink]forkinkey_entities]
key_entities = list(set([kforkkinkey_entitiesforkinkk]))
# 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 =awaitasyncio.gather(
*[knowledge_graph_inst.get_node(r)forrinpath]
)
path_degrees =awaitasyncio.gather(
*[knowledge_graph_inst.node_degree(r)forrinpath]
)
path_datas = [
{**n,"entity_name": k,"rank": d}
fork, n, dinzip(path, path_datas, path_degrees)
ifnisnotNone
]
exceptValueErrorase:
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[实体提取]
C --> D[关系抽取]
D --> E[图谱构建]
E --> F[社区检测]
F --> G[报告生成]
end
subgraph "查询阶段"
H[用户查询] --> I[实体检索]
I --> J[社区映射]
J --> K1[本地知识]
J --> K2[全局知识]
J --> K3[桥接知识]
K1 --> L[上下文整合]
K2 --> L
K3 --> L
L --> M[LLM生成]
M --> N[最终答案]
end
E --> I
G --> K2