前言
近期一直在做知识图谱问答，参考了不少前沿的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) 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%)
对比分析:
在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"}
]
}
理想情况下，社区发现执行完成后，高层概念的实体，应该是先前基于实体层级聚类生成的实体。
社区报告生成
报告结构
@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构建出了既有严密逻辑又有语义连贯性的层次化知识结构
问答
数据类型组合: 本地知识 + 全局知识 + 桥接知识 + 原始文档
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_m
if 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[实体提取]
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
粤ICP备14082021号
免费POC，
零成本试错
