HiRAG问答流程深入分析

前言

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

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

整个知识图谱构建如上，其中，实体关系抽取与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"}
    ]
}

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

社区报告生成

报告结构

@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