2024 年 4 月,微软发表了一篇论文《From Local to Global: A Graph RAG Approach to Query-Focused Summarization》。在这篇论文中,微软的研究人员提出了一种从文本构建并增强知识图的方法,来解决 Baseline RAG 系统在全局理解上的缺陷,例如:
-
Baseline RAG 系统在面对需要从多源信息中抽取并综合分析的情况时,会遭遇显著的障碍。具体来说,当回答一个复杂问题涉及到通过识别和利用不同信息片段之间的共享属性,来构建新的、综合性见解时,Baseline RAG 无法有效连接这些“点”,导致信息整合上的不足。
-
在要求 Baseline RAG 对大规模的数据集合或是单篇幅巨大的文档进行全面而深入的理解时,它的表现会显得较为逊色。这通常是因为它在处理大量数据时,难以有效地捕捉和理解那些被浓缩于其中的关键语义概念,从而影响了整体的理解质量。
2024 年 7 月,微软在 Github 开源了基于此种方法构建的 GraphRAG 项目代码,本文将展开介绍此 GraphRAG 项目的构建过程。并且尝试使用 Deepseek-chat 模型和 ZhipuAI 的 embedding 模型构建一个 GraphRAG 系统,使用 Neo4j 图数据库进行数据的可视化。
有兴趣的同学也可以直接访问对应的代码仓库和文档,自己动手试一试。
GraphRAG 介绍
索引过程
GraphRAG 系统的索引分为三个阶段 数据预处理、从文本构建图数据 和 从图数据构建社区结构,具体过程可以参照下图中紫色的部分。
阶段一:数据预处理
---
title: Compose TextUnits
---
flowchart LR
doc1[Document 1] --> c1[Chunk 1]
c1 --> e1
e1 --> tu1[TextUnit 1]
doc1 --> c2[Chunk 2]
c2 --> e2
e2 --> tu2[TextUnit 1]
doc2[Document 2] --> c3[Chunk 3]
c3 --> e3
e3 --> tu3[TextUnit 3]
doc2 --> c4[Chunk 4]
c4 --> e4
e4 --> tu4[TextUnit 4]
在开始构图数据前,GraphRAG 系统需要对原始数据进行一系列预处理,即切块和文本嵌入。
在预处理过程中,Chunk 大小是一个比较重要的参数。由于预处理后的文本单元(TextUnit)将在后续环节中提取图数据,使用较小的 Chunk 可以带来更高的保真度,但也会拖慢处理速度。
阶段二:构建图数据
---
title: Graph Extraction
---
flowchart LR
tu[TextUnit] --> ge[Graph Extraction] --> gs[Graph Summarization]
tu --> ce[Claim Extraction]
1. 图元素抽取
在这个阶段,GraphRAG 将驱动大模型对每个文本单元进行分析,从中抽取我们的图基础元素:实体、关系和主张(Entities, Relationships, Claims)。实体与关系通过我们的“Graph Extraction”操作一次性完成提取,而主张则通过“Graph Extraction”操作来获取。
-
Entities(实体) :实体指的是现实世界中的具体事物或者概念,它们可以是人、地点、组织、事件、抽象概念等。例如,在句子“奥巴马出生于夏威夷。”中,“奥巴马”和“夏威夷”是实体。在 GraphRAG 中一个 Entity 包含了实体名称、实体类型和实体描述。
-
Relationships(关系) :关系描述了实体之间的联系。例如,在上述句子中,“出生于”就是一种关系,它连接了实体“奥巴马”和“夏威夷”。在 GraphRAG 中一个 Relationships 包含了关系两端的实体名称和关系描述。
-
Claims(主张) :主张通常指的是关于实体及其关系的陈述,它包含了一个或多个实体和关系的信息。例如:“奥巴马是美国第44任总统”,就是关于实体“奥巴马”的一个声明。
例如,使用 GraphRAG 对余华的小说《活着》进行知识图谱的抽取后,我们可以绘制出如下的实体关系图,其中共包含 602 个实体和 1095 段关系:
2. 图元素总结
在抽取过程中,不同的 TextUnit 可能包含了关于同一实体的不同描述。例如随着小说《活着》的剧情发展,福贵会经历不同的事件。因此一个实体可能存在多个不同的描述。种情况下,我们不仅需要识别和提取实体,还要整合关于同一实体的多方面信息,确保我们的实体描述尽可能完整和准确。
在 Graph Summarization 过程中,GraphRAG 将驱动 LLM 将描述列表转化为一个简短的摘要,并确保每个实体和关系都有且仅有一段简洁的的描述信息。
阶段三:构建社区结构
---
title: Graph Augmentation
---
flowchart LR
cd[Leiden Hierarchical Community Detection] --> ge[Node2Vec Graph Embedding] --> sc[Generate Community Reports] --> ss[Summarize Community Reports] --> ce[Community Embedding]
1. 构建社区结构
为了获得对数据的整体性理解,GraphRAG 递归地使用莱顿算法对知识图谱进行聚类,并产生不同层次的社区(Community)结构。
Community 结构是指在一个更大的网络中,节点(如 GraphRAG 中的实体)倾向于形成密集的局部集群,这些集群被称为 Community。Community 内部的节点之间有较多的边(联系),而 Community 之间的边则相对较少。换句话说,Community 结构体现了网络中节点的聚类特性,这些节点基于某种相似性或功能上的关联性而聚集在一起。
如下图:圆圈代表实体节点,大小与其度数(即与之相连的边的数量)成正比。节点颜色代表实体群落,表示了在全部节点中的社区分布结构。
2. 图嵌入
GraphRAG 使用 Node2Vec 算法生成图的矢量表示。这将使系统能够理解图的隐式结构,并提供一个额外的向量空间,以便在查询阶段搜索相关概念。
3. 生成社区报告/摘要
在此阶段,GraphRAG 会为不同层级的社区建立报告(Report),而这些报告代表了对原始数据不同层次的理解。形象来说,这些社区构成了一个树状结构,更接近根的社区报告通常覆盖了更大的范围,代表了更加全局和概括性的理解。而越接近枝叶的社区通常覆盖面越窄,理解也越微观。
此外,这种报告也可以作为理解数据全局结构的一种方式,可以通过摘要找到有用的社区,然后下钻到包含更详细信息的子社区中。
4. 社区嵌入
最后,我们为此阶段创建的内容执行文本嵌入,以便于在查询阶段搜索相关内容。
由此我们就完成了 GraphRAG 中索引的过程。
Query 过程
GraphRAG 提供了两种不同的查询模式。
Global Query 用于回答更加全局性的问题,例如“小说《活着》的主题是什么”,而 Local Query 则用于回答更加具体的问题。
1. 全局查询 Global Query
全局搜索功能是GraphRAG系统的核心优势之一,它解决了传统RAG模型在处理需要跨数据集汇总信息的查询时的局限性。
传统RAG模型依赖于数据集中语义相似文本内容的向量搜索,对于诸如“小说《活着》的主题是什么”这类需要整合数据集信息才能回答的查询,往往表现不佳,因为查询本身无法直接指向正确信息。
具体而言,Global Query 的过程如图所示:
---
title: Global Search Dataflow
---
%%{ init: { 'flowchart': { 'curve': 'step' } } }%%
flowchart LR
uq[User Query] --- .1
ch1[Conversation History] --- .1
subgraph RIR
direction TB
ri1[Rated Intermediate<br/>Response 1]~~~ri2[Rated Intermediate<br/>Response 2] -."{1..N}".-rin[Rated Intermediate<br/>Response N]
end
.1--Shuffled Community<br/>Report Batch 1-->RIR
.1--Shuffled Community<br/>Report Batch 2-->RIR---.2
.1--Shuffled Community<br/>Report Batch N-->RIR
.2--Ranking +<br/>Filtering-->agr[Aggregated Intermediate<br/>Responses]-->res[Response]
classDef green fill:#26B653,stroke:#333,stroke-width:2px,color:#fff;
classDef turquoise fill:#19CCD3,stroke:#333,stroke-width:2px,color:#fff;
classDef rose fill:#DD8694,stroke:#333,stroke-width:2px,color:#fff;
classDef orange fill:#F19914,stroke:#333,stroke-width:2px,color:#fff;
classDef purple fill:#B356CD,stroke:#333,stroke-width:2px,color:#fff;
classDef invisible fill:#fff,stroke:#fff,stroke-width:0px,color:#fff, width:0px;
class uq,ch1 turquoise;
class ri1,ri2,rin rose;
class agr orange;
class res purple;
class .1,.2 invisible;
Global Query 方法使用从社区层次结构指定层级中收集的报告作为上下文数据,以类似Map - Reduce
的方式生成响应。在 Map
步骤中,社区报告被分割成文本块,每个文本块用于生成中间响应,其中每个点都有一个数值评级。在 Reduce
步骤中,从中间响应中挑选出最重要的点并进行聚合,最终形成用于生成最终响应的上下文。
这种方法的直观理解是:越宏观的问题需要越宏观的视角和信息来回答。
社区层次结构中的每一层代表着不同粒度的信息集合。高层级可能包含更为概括和宏观的观点,而低层级则提供具体和细节化的信息。对于宏观问题,系统倾向于选择较高层级的数据,以捕捉整体趋势和主要观点。
通过 Map - Reduce
过程,系统可以选择对于原始问题的重要性或相关性较高的信息,并进一步提炼和汇总,形成一个精炼的、综合的上下文。这个上下文包含了对原始问题最相关的见解和数据,是生成最终响应的基础。
2. 本地查询 Local Query
本地查询则基于更加微观的视角,结合知识图谱中的结构化数据与原始文档中的非结构化数据,来增强检索和生成过程中的上下文。因此,Local Query 别适合回答那些需要理解输入文档中特定实体的问题,比如询问“洋甘菊有哪些治疗特性?”这样的问题。
具体而言,Local Query 的过程如图所示:
---
title: Local Search Dataflow
---
%%{ init: { 'flowchart': { 'curve': 'step' } } }%%
flowchart LR
uq[User Query] ---.1
ch1[Conversation<br/>History]---.1
.1--Entity<br/>Description<br/>Embedding--> ee[Extracted Entities]
ee[Extracted Entities] ---.2--Entity-Text<br/>Unit Mapping--> ctu[Candidate<br/>Text Units]--Ranking + <br/>Filtering -->ptu[Prioritized<br/>Text Units]---.3
.2--Entity-Report<br/>Mapping--> ccr[Candidate<br/>Community Reports]--Ranking + <br/>Filtering -->pcr[Prioritized<br/>Community Reports]---.3
.2--Entity-Entity<br/>Relationships--> ce[Candidate<br/>Entities]--Ranking + <br/>Filtering -->pe[Prioritized<br/>Entities]---.3
.2--Entity-Entity<br/>Relationships--> cr[Candidate<br/>Relationships]--Ranking + <br/>Filtering -->pr[Prioritized<br/>Relationships]---.3
.2--Entity-Covariate<br/>Mappings--> cc[Candidate<br/>Covariates]--Ranking + <br/>Filtering -->pc[Prioritized<br/>Covariates]---.3
ch1 -->ch2[Conversation History]---.3
.3-->res[Response]
classDef green fill:#26B653,stroke:#333,stroke-width:2px,color:#fff;
classDef turquoise fill:#19CCD3,stroke:#333,stroke-width:2px,color:#fff;
classDef rose fill:#DD8694,stroke:#333,stroke-width:2px,color:#fff;
classDef orange fill:#F19914,stroke:#333,stroke-width:2px,color:#fff;
classDef purple fill:#B356CD,stroke:#333,stroke-width:2px,color:#fff;
classDef invisible fill:#fff,stroke:#fff,stroke-width:0px,color:#fff, width:0px;
class uq,ch1 turquoise
class ee green
class ctu,ccr,ce,cr,cc rose
class ptu,pcr,pe,pr,pc,ch2 orange
class res purple
class .1,.2,.3 invisible
首先,系统将依据原始提问,从知识图谱中识别出一组与用户输入语义相关的实体。
然后,利用这些实体作为查询条件,在知识图谱或相关数据库中进行检索,找到与这些实体直接相关的内容,包含:TextUnit、社区报告、实体、关系或协变量(如主张)。
检索的结果经过过滤和重排序后,选择高质量的数据源,并将其整合进一个预定义大小的上下文窗口。这个窗口内的信息用于构建响应,确保回答内容既精确又详尽。
那么代价是什么呢?
虽然 Graph 带来了更优秀的效果,但由于索引过程完全使用 LLM 来完成,这将带来巨大的 Token 开销,耗时也非常久。
例如,我尝试对 12 万字的小说进行索引时产生了数百万 Token 的调用量。这远比构建索引仅需调用 embedding 模型的传统 RAG 系统昂贵得多。
而在 Query 阶段,一次响应的耗时与 Token 用量也非常大。例如,在一次 Global Query 中,响应耗时接近 1 分钟,且消耗了 65083 个 Token。
本地搭建 GraphRAG 系统
下面是一个在本地运行 GraphRAG 系统实例的简单介绍。其中,由于原项目调用了 OpenAI 的 API 接口,国内不易访问且 API 价格昂贵,因此在下面的案例中使用 Deepseek 的大模型 API 和 ZhipuAI 的 embedding 模型 API 进行替代。
大家也可以替换为其他兼容 OpenAI 格式的大模型接口,但考虑到 GraphRAG 巨大的 Token 消耗量,这里建议大家选用 Deepseek 的 llm,毕竟价格可太香了。
Install GraphRAG
- 使用 pip 安装 GraphRAG
pip install graphrag
- 创建 GraphRAG 目录,并将数据样本存放于项目目录下的
input
文件夹内.
mkdir -p ./ragtest/input
- 初始化 GraphRAG 项目
python -m graphrag.index --init --root ./ragtest
- 为了使用 Deepseek 和 ZhipuAI 的模型替代 OpenAI 的模型,需要修改一些参数。打开项目目录下的
settings.yaml
,并参考以下内容进行修改:
# 修改 llm 模型配置
llm:
api_key: <此处填入你的 Deepseek API KEY>
type: openai_chat
model: deepseek-chat
# 由于 Deepseek 不支持 OpenAI SDK 中 `response_format = { "type": "json_object" }` ,此处需改为 False
model_supports_json: False
api_base: https://api.deepseek.com/v1
# 修改 embedding 模型配置
embeddings:
llm:
api_key: <此处填入你的 ZhipuAI API KEY>
model: embedding-2
api_base: https://open.bigmodel.cn/api/paas/v4
索引数据
运行索引过程需要一些时间,这与进行索引的数据量、使用的模型和 Chunk 大小有关。
python -m graphrag.index --root ./ragtest
当索引运行完成后,我们在项目文件夹中应该会看到一个名为 ./ragtest/output/<timestamp>/artifacts
的新文件夹,其中包含一系列 parquet
文件。
进行查询
- 使用 Global Query 模式进行查询
python -m graphrag.query --root ./ragtest --method global "<在此输入你的问题>"
- 使用 Local Query 模式进行查询
python -m graphrag.query --root ./ragtest --method local "<在此输入你的问题>"
大家也可以参考 GraphRAG 官方提供的 Examples Notebook,以获得查询过程的更详细的信息:github.com/microsoft/g…
使用 Neo4j 进行数据可视化
在索引过程中,GraphRAG 从原始资料中抽取出了知识图谱。我们可以使用图数据库,例如 Neo4j 来对知识图谱进行可视化展示。
在开始前,请确保您在本地有一个可运行的 Neo4j 图数据库。
- 将
parquet
文件转换为csv
文件
GraphRAG 提取出的知识图谱保存在 ./ragtest/output/<timestamp>/artifacts
目录下的一系列 parquet
文件中,为了将其导入到 Neo4j 数据库中,我们需要将其转换为 csv
文件。
为了完成这项工作,我们可以运行以下代码:
import os
import pandas as pd
# 指定输入和输出文件夹路径
input_folder = './ragtest/output/<timestamp>/artifacts'
output_folder = '<请输入 neo4j 项目目录下的 import 文件夹路径>'
def convert_parquet_to_csv(input_folder, output_folder):
# 检查输出文件夹是否存在,如果不存在则创建
if not os.path.exists(output_folder):
os.makedirs(output_folder)
# 遍历输入文件夹中的所有文件
for filename in os.listdir(input_folder):
if filename.endswith(".parquet"):
# 构建完整的文件路径
parquet_file = os.path.join(input_folder, filename)
csv_file = os.path.join(output_folder, filename.replace(".parquet", ".csv"))
# 读取 Parquet 文件
df = pd.read_parquet(parquet_file)
# 将数据写入 CSV 文件
df.to_csv(csv_file, index=False)
print(f"Converted {parquet_file} to {csv_file}")
# 调用函数进行转换
convert_parquet_to_csv(input_folder, output_folder)
- 将
csv
导入 Neo4j 数据库,我们可以将以下Cypher
代码拷贝至 Neo4j 中并执行:
// 1. Import Documents
LOAD CSV WITH HEADERS FROM 'file:///create_final_documents.csv' AS row
CREATE (d:Document {
id: row.id,
title: row.title,
raw_content: row.raw_content,
text_unit_ids: row.text_unit_ids
});
// 2. Import Text Units
LOAD CSV WITH HEADERS FROM 'file:///create_final_text_units.csv' AS row
CREATE (t:TextUnit {
id: row.id,
text: row.text,
n_tokens: toFloat(row.n_tokens),
document_ids: row.document_ids,
entity_ids: row.entity_ids,
relationship_ids: row.relationship_ids
});
// 3. Import Entities
LOAD CSV WITH HEADERS FROM 'file:///create_final_entities.csv' AS row
CREATE (e:Entity {
id: row.id,
name: row.name,
type: row.type,
description: row.description,
human_readable_id: toInteger(row.human_readable_id),
text_unit_ids: row.text_unit_ids
});
// 4. Import Relationships
LOAD CSV WITH HEADERS FROM 'file:///create_final_relationships.csv' AS row
CREATE (r:Relationship {
source: row.source,
target: row.target,
weight: toFloat(row.weight),
description: row.description,
id: row.id,
human_readable_id: row.human_readable_id,
source_degree: toInteger(row.source_degree),
target_degree: toInteger(row.target_degree),
rank: toInteger(row.rank),
text_unit_ids: row.text_unit_ids
});
// 5. Import Nodes
LOAD CSV WITH HEADERS FROM 'file:///create_final_nodes.csv' AS row
CREATE (n:Node {
id: row.id,
level: toInteger(row.level),
title: row.title,
type: row.type,
description: row.description,
source_id: row.source_id,
community: row.community,
degree: toInteger(row.degree),
human_readable_id: toInteger(row.human_readable_id),
size: toInteger(row.size),
entity_type: row.entity_type,
top_level_node_id: row.top_level_node_id,
x: toInteger(row.x),
y: toInteger(row.y)
});
// 6. Import Communities
LOAD CSV WITH HEADERS FROM 'file:///create_final_communities.csv' AS row
CREATE (c:Community {
id: row.id,
title: row.title,
level: toInteger(row.level),
raw_community: row.raw_community,
relationship_ids: row.relationship_ids,
text_unit_ids: row.text_unit_ids
});
// 7. Import Community Reports
LOAD CSV WITH HEADERS FROM 'file:///create_final_community_reports.csv' AS row
CREATE (cr:CommunityReport {
id: row.id,
community: row.community,
full_content: row.full_content,
level: toInteger(row.level),
rank: toFloat(row.rank),
title: row.title,
rank_explanation: row.rank_explanation,
summary: row.summary,
findings: row.findings,
full_content_json: row.full_content_json
});
// 8. Create indexes for better performance
CREATE INDEX FOR (d:Document) ON (d.id);
CREATE INDEX FOR (t:TextUnit) ON (t.id);
CREATE INDEX FOR (e:Entity) ON (e.id);
CREATE INDEX FOR (r:Relationship) ON (r.id);
CREATE INDEX FOR (n:Node) ON (n.id);
CREATE INDEX FOR (c:Community) ON (c.id);
CREATE INDEX FOR (cr:CommunityReport) ON (cr.id);
// 9. Create relationships after all nodes are imported
MATCH (d:Document)
UNWIND split(d.text_unit_ids, ',') AS textUnitId
MATCH (t:TextUnit {id: trim(textUnitId)})
CREATE (d)-[:HAS_TEXT_UNIT]->(t);
MATCH (t:TextUnit)
UNWIND split(t.document_ids, ',') AS docId
MATCH (d:Document {id: trim(docId)})
CREATE (t)-[:BELONGS_TO]->(d);
MATCH (t:TextUnit)
UNWIND split(t.entity_ids, ',') AS entityId
MATCH (e:Entity {id: trim(entityId)})
CREATE (t)-[:HAS_ENTITY]->(e);
MATCH (t:TextUnit)
UNWIND split(t.relationship_ids, ',') AS relId
MATCH (r:Relationship {id: trim(relId)})
CREATE (t)-[:HAS_RELATIONSHIP]->(r);
MATCH (e:Entity)
UNWIND split(e.text_unit_ids, ',') AS textUnitId
MATCH (t:TextUnit {id: trim(textUnitId)})
CREATE (e)-[:MENTIONED_IN]->(t);
MATCH (r:Relationship)
MATCH (source:Entity {name: r.source})
MATCH (target:Entity {name: r.target})
CREATE (source)-[:RELATES_TO]->(target);
MATCH (r:Relationship)
UNWIND split(r.text_unit_ids, ',') AS textUnitId
MATCH (t:TextUnit {id: trim(textUnitId)})
CREATE (r)-[:MENTIONED_IN]->(t);
MATCH (c:Community)
UNWIND split(c.relationship_ids, ',') AS relId
MATCH (r:Relationship {id: trim(relId)})
CREATE (c)-[:HAS_RELATIONSHIP]->(r);
MATCH (c:Community)
UNWIND split(c.text_unit_ids, ',') AS textUnitId
MATCH (t:TextUnit {id: trim(textUnitId)})
CREATE (c)-[:HAS_TEXT_UNIT]->(t);
MATCH (cr:CommunityReport)
MATCH (c:Community {id: cr.community})
CREATE (cr)-[:REPORTS_ON]->(c);
- 使用
Cypher
语言进行查询,例如我希望查询与“福贵”相关的节点,可以运行如下Cypher
代码:
MATCH (e:Entity)-[r:RELATES_TO]->(otherNode)
WHERE e.name CONTAINS '福贵'
RETURN e, r, otherNode
LIMIT 50