什么是GraphRAG?
你如何看待GraphRAG?从你的角度来看,GraphRAG意味着什么?如果你可以同时拥有一个标准的RAG和一个GraphRAG的组合包,只需切换查询即可,那会怎样?
事实上,目前并没有一个具体且普遍接受的GraphRAG定义——至少目前还没有。基于我的经验、文献监测以及与许多人的交流,我会估计(并向Steven D. Levitt 致歉,我知道这不是呈现统计数据的正确方式):
就我个人而言,我并不完全信服前两种定义,下面我将解释原因。
首先,我必须说,我认为微软的GraphRAG是一个非常酷的想法。在大约五年左右的时间内,它很可能会被广泛采用,甚至成为GraphRAG方法中的主流选择。然而,今天,对于大型工业应用来说,它仍然过于昂贵且不切实际。现实是,大多数公司缺乏采用这种方法的时间、预算和信心。相反,他们更可能选择标准的“普通”向量数据库,这在当前的限制条件下更为可行。信心——因为事实上,并没有成千上万的GraphRAG在生产中的例子(可能就是因为上述原因)。
在我看来,文本到Cypher或文本到SPARQL的技术是微软GraphRAG的一个很好的替代方案(尽管它们也可以一起使用),并且我已经看到了一些其应用的出色例子。然而,也存在一些缺点。首先,它需要大量的昂贵LLM调用来生成查询。其次,在你和你的知识库之间总是存在一层不确定性——你依赖于你如何构思提示、所选模型如何有效地执行并构建Cypher或SPARQL查询。此外,额外的处理步骤增加了响应时间,而更高的实施复杂性也增加了挑战。总之,这种技术对于某些应用来说是非常有前景和强大的,但其适用性取决于具体的用例。
效率优化的困境
作为一名顾问和生成式人工智能解决方案开发者,我的目标是在任何规模上提供GraphRAG服务——从小型实现到大型企业级解决方案。
规模扩大往往伴随着权衡,特别是在准确性或效率方面。然而,如果一个较低复杂度和成本效益的解决方案仍然能提供令人满意的结果,那么它就值得保留在工具箱中,对吗?
考虑到这一点,所提出的方法是利用图的力量进行RAG(检索增强生成),而不会为图的创建本身产生高额成本。挑战在于构建和维护一个有用的图,同时最小化对LLM的依赖——或者理想情况下,通过使用小型、本地的LLM,而不是对大型云端模型进行昂贵的API调用。
固定实体架构
核心思想是构建一个分层图:
在两种情况下,我都展示了一种不依赖大型语言模型(LLM)来创建图表的方法。然而,这种方法的一个主要挑战是构建本体层。考虑以下事实:
由于这些限制,我开始探索消除对固定本体层需求的方法。
为何使用分层图表?
Neo4j允许对单个内部标签进行向量索引。如果节点具有不同的标签,则需要为每个标签构建单独的索引——这在执行向量搜索时并不总是实用的。
当然,在某些情况下,拥有大量节点类型是有意义的,例如,当需要严格的本体区分/过滤时。然而,在我的情况中,到目前为止这并没有必要。通常选择两到三层是一个合理的数量。因此,解决标签索引限制的一种变通方法是,为同一层内的所有节点分配相同的内部标签,同时将实际标签、名称和元数据存储为节点属性。
自然语言处理(NLP)的力量
如何不依赖自己的大脑或万亿参数的大型语言模型从文本中提取信息?这就是经典自然语言处理(NLP)可以成为有价值工具的地方。
值得一提的是,当我开始寻找GPT-3.5时代之前和之后最佳的NLP库和模型时,我感到震惊。其中许多——如果不是全部(如果我错了,请纠正我,并分享任何好的链接或想法!)——已不再受支持、更新或维护。它们仿佛被遗弃,几乎被遗忘,这真的很遗憾,因为它们蕴含着巨大的潜力。
尽管如此,出于现实世界的行业需求和实际限制,我决定迎接挑战,探索一种基于NLP的方法。我的目标是构建一个能够增强标准向量数据库性能的图表。
GraphRAG及其潜在应用
在深入探讨NLP驱动的图表在检索增强生成(RAG)中的实现并讨论结果之前,我想首先阐述我对不同类型的GraphRAG及其应用的看法。
当我提到微软GraphRAG时,我不仅包括微软研究院发表的原始方法,还包括自那时以来出现的各种更轻便的改编版本。
这些方法通常涉及:
虽然存在不同的实现方式,但基本原理是相同的:使用大型语言模型从文本中构建知识图表。
下面的信息图(图1)代表了我从行业角度出发,关于何时以及为何在RAG系统中使用不同类型的基于图表的向量搜索的观点。
首先,如果你面临选择使用图表还是标准向量数据库的决定,值得四处看看——有一些关于何时选择一种而非另一种的指南。
一旦你决定采用GraphRAG解决方案,这张信息图就适用了。在这里,我突出了在构建你的图表之前需要考虑的关键因素。
1. 数据量——你的知识库中存在多少数据?
2. 预算限制——你构建图表的预算有多有限?
3. 本体可用性:
这些因素极大地影响了你的GraphRAG解决方案的设计、可行性和效率。
图1
一旦你回答了三个关键问题——数据量、预算限制和本体可用性——你就可以确定适合你的用例的GraphRAG方法。
需要注意的是,图1并没有涵盖所有可能的场景。一些混合方法也是可行的,而且技术之间的界限并不是严格固定的。
然而,我观察到以下趋势:数据越多,你就越需要仔细评估你的投资。如果你有足够的预算并且需要非常高的准确性,微软的解决方案是一个不错的选择。
但是,如果预算限制是一个问题(这几乎总是如此),你可能需要在准确性上做出妥协,并选择几乎不含大型语言模型(LLM)的解决方案。在这种情况下,最好的方法是建立一个本体层,并构建一个固定实体架构图。
如果你难以定义本体、对数据的理解不够深入,或者面临数据复杂性高的问题,我建议你构建一个基于自然语言处理(NLP)的图。在接下来的部分中,我将演示如何实现这一点。
释放NLP的力量
现在,让我们卷起袖子,以一块巧克力的成本(包括所涉及的电费)来构建一个图。
技术设置
对于这个项目,我使用了:
基于NLP的图方法
如引言中所述,基于NLP的图是从固定实体架构中派生出来的,有一个关键区别——我去掉了本体层。
这意味着图将包括:
通过利用NLP而不是重度依赖大型语言模型(LLM)的处理方式,这种方法显著降低了成本。
数据预处理管道
数据预处理管道遵循以下关键步骤:
下面是一个代码示例,用于在Neo4j数据库中填充文档层。
def add_chunks_to_db(chunks, doc_name):
prev_node_id = None
for i, chunk in enumerate(chunks):
# Escape single quotes in the chunk content
escaped_chunk = chunk.replace("'", "\\'")
# Create the chunk node
query = f'''
MERGE (d:Document {{
chunkID: "{f"chunk_{i}"}",
docID: "{doc_name.replace("'", "\\'")}",
full_text: '{escaped_chunk}',
embeddings: {embeddings.embed_documents(chunk).tolist()}}}
)
RETURN elementId(d) as id
'''
result = run_query(query)
chunk_node_id = result[0]['id']
# If this is not the first chunk, create a NEXT relationship to the previous chunk
if prev_node_id is not None:
query = f'''
MATCH (c1:Document), (c2:Document)
WHERE elementId(c1) = $prev_node_id AND elementId(c2) = $chunk_node_id
MERGE (c1)-[:NEXT]->(c2)
MERGE (c2)-[:PREV]->(c1)
'''
run_query(driver, query)
prev_node_id = chunk_node_id
注意,我在这里构建的是文档链。我将每个文档的块添加进去,并用双向边连接起来:一条边称为NEXT,指向下一个块;另一条边称为PREV,指向前一个块。因此,我得到的图看起来像这样(见图2):
图2
在这里,你可以看到我已添加到图中的660个PDF中的4个。链从chunk_0开始,到chunk_n结束。
有了这一层,你可以轻松地在其上应用你的第一个向量和文本索引,例如:
query = '''
CREATE VECTOR INDEX vector_index_document
IF NOT EXISTS
FOR (d:Document)
ON (d.embeddings)
OPTIONS {indexConfig: {
`vector.dimensions`: 768,
`vector.similarity_function`: 'cosine'
}}
'''
而文本索引则为:
query = '''
CREATE FULLTEXT INDEX text_index_document FOR (n:Document) ON EACH [n.full_text]
'''
现在,人们可以将这个图作为标准向量数据库来使用。你只需要执行以下操作:
def pure_rag(query):
my_query_emb = emb.embed_query(query)
query = f"""
CALL db.index.vector.queryNodes('vector_index_document', 10, $user_query_emb)
YIELD node AS vectorNode, score as vectorScore
WITH vectorNode, vectorScore
ORDER BY vectorScore DESC
RETURN elementId(vectorNode), vectorNode.docID, vectorNode.full_text as document_text, vectorScore
LIMIT 10
"""
params = {'my_query': my_query, 'user_query_emb': my_query_emb.tolist()}
results = run_query(query, params)
return pd.DataFrame(data=results)
就这样!让我们在NVIDIA数据集上试试,并根据其中包含的数据查询一些内容。
图3
我使用的LLM是NVIDIA NIM模型“meta/llama-3.3–70b-instruct”,来自Try NVIDIA NIM API。注意,我并没有编写任何复杂的提示,只是传递了用户问题和检索到的前10个段落。
然而,我们构建这个图并不仅仅是为了实现标准向量数据库的功能,对吧?让我们充分利用它!
释放图的力量
图为数据增添了语义推理能力。即使没有传统的RDF世界的语义推理,图通过连接实体,也有助于更深入地理解数据。此外,我在之前的文章中提出过一个假设,即总是存在某种搜索不对称性,这种不对称性可以起到一定的作用。这种搜索不对称性也被称为幅度敏感性。点积受向量幅度的影响,这意味着如果比较的向量幅度差异显著,点积可能无法可靠地表示相似性。
在创建了包含文档层的图之后,我们需要一种方法来为文本块创建“粘合剂”。我们没有本体,而且我们的假设相当天真,尽管不幸的是很现实:我们拥有大量数据,我们并不完全清楚这些数据是关于什么的,但我们希望从中提取最大价值。我们的目标是构建一个词汇图,充分利用GraphRAG的所有优势,同时在这个过程中不花费太多钱。
我建议为此利用NLP技术。首先,让我们从每个文本块中提取单词、二元组和三元组。我使用了名为sparkNLP的NLP库,它允许你利用本地GPU的力量来处理大量文档。下面是我用于单词提取的代码片段。
from pyspark.sql import SparkSession
from sparknlp.base import *
from sparknlp.annotator import *
from sparknlp import DocumentAssembler, Finisher
import sparknlp
# Initialize Spark session
spark = sparknlp.start()
# Sample data
# Create DataFrame from the list of documents
data = spark.createDataFrame([(i, doc) for i, doc in enumerate(documents)], ["id", "text"])
# Document Assembler
document_assembler = DocumentAssembler() \
.setInputCol("text") \
.setOutputCol("document")
# Tokenizer
tokenizer = Tokenizer() \
.setInputCols(["document"]) \
.setOutputCol("token")
# NGram Generator for bigrams
bigram_generator = NGramGenerator() \
.setInputCols(["token"]) \
.setOutputCol("bigrams") \
.setN(2)
# NGram Generator for trigrams
trigram_generator = NGramGenerator() \
.setInputCols(["token"]) \
.setOutputCol("trigrams") \
.setN(3)
# Finisher to convert annotations to string
finisher = Finisher() \
.setInputCols(["bigrams", "trigrams"]) \
.setOutputCols(["finished_bigrams", "finished_trigrams"]) \
.setCleanAnnotations(False)
# Pipeline
pipeline = Pipeline(stages=[
document_assembler,
tokenizer,
bigram_generator,
trigram_generator,
finisher
])
# Fit and transform the data
model = pipeline.fit(data)
result = model.transform(data)
# Show the results
pandas_df = result.select("text", "finished_bigrams", "finished_trigrams").toPandas()
# Stop the Spark session
spark.stop()
创建完单词实体后,你可以将它们添加到图中,并与提取它们的文本块建立连接。这种方法简单且稳健,你可以再次在这一层应用我之前演示过的两个索引。这样,我们就创建了第二层,所有节点都被标记为“Token”。我在label属性中包含了“token”、“bigram”和“trigram”标签,同时将单词本身作为name属性,并关联了嵌入。以下示例展示了创建单词节点的Cypher查询,以及构建相应向量索引的查询:
# create token node
query = """MERGE (t:Token {label: "Token",
name: $token,
embeddings: $token_embeddings
}) RETURN elementId(t) as token_node_id"""
同样也为二元组和三元组执行此操作。
接下来,创建索引:
# create vector index on token embeddings
query = '''CREATE VECTOR INDEX vector_index_token IF NOT EXISTS
FOR (n:Token)
ON (n.embeddings)
OPTIONS {indexConfig: {
`vector.dimensions`: 768,
`vector.similarity_function`: 'cosine'
}}
图4展示了一个已创建的二元组节点的示例。请注意,包含单词、二元组和三元组的整个层都有一个内部标签“Token”,这使得可以一次性将所有节点的向量索引应用上。
图4
图5
到目前为止一切顺利:我们有了在不同文档中部分共享的单词,这在一定程度上使所有内容都相互连接。然而,遗憾的是,但也在意料之中,最初的RAG尝试并没有比单纯的RAG取得更好的结果。
为了充分释放图的潜力,我们需要利用上下文、逻辑和语义将实体相互连接起来。这里的挑战是:我们不想依赖GPT或其他拥有数万亿参数的大型模型。我们的图中已经有超过26.2万个节点,使用如此大型的模型对于我们“巧克力棒”级别的预算来说,将是一种过度消耗,确实如此。
三元组
有许多优秀的开源模型可用。然而,三元组提取可能是一项具有挑战性的任务。最佳方法是使用较小的变换器模型,并针对这一特定任务进行微调。更好的方法是自己进行微调,但在此次演示中,我使用了Hugging Face的预训练模型。bew/t5_sentence_to_triplet_xl模型是在FLAN-t5-xl的XL版本上进行微调的。该模型的大小约为GPT-4的六百分之一,因此可以轻松安装在我的电脑上,没有任何问题。该模型专门用于从文本中提取三元组。据模型所有者Brian Williams称,该模型还不够完美,是的,结果并不总是像我所希望的那样准确,但我们的目标并不是追求最高的准确性——只要以最低的成本达到非常好的准确性就足够了。
我提取了文本块,并将它们传递给模型。模型创建了许多三元组,这些三元组随后被映射到Token节点,使得图中的总边数超过65万条。
图6
以下是三元组映射的一小段代码:
def process_triplet(triplet):
subject, predicate, object_ = triplet
subject_emb = embed_query_on_gpu(subject)
predicate_emb = embed_query_on_gpu(predicate)
object_emb = embed_query_on_gpu(object_)
params = {'subject_emb': subject_emb.tolist(),
'predicate_emb': predicate_emb.tolist(),
'object_emb': object_emb.tolist(),
'subject': subject,
'predicate': predicate,
'object': object_}
similarSubjects_query = """
CALL () {
// Search for the subject duplicates
CALL db.index.vector.queryNodes('vector_index_token', 10, $subject_emb)
YIELD node AS vectorNode, score as vectorScore
WITH vectorNode, vectorScore
WHERE vectorScore >= 0.96
RETURN collect(vectorNode) AS similarSubjects
}
WITH similarSubjects
OPTIONAL MATCH (n:Token {name: toLower($subject)})
WITH similarSubjects + CASE WHEN n IS NULL THEN [] ELSE [n] END AS allSubjects
UNWIND allSubjects AS subject
RETURN collect(subject) AS similarSubjects
"""
similarSubjects = run_query(similarSubjects_query, params)[0]['similarSubjects']
similarPredicates_query = """
CALL () {
// Search for the predicate duplicates
CALL db.index.vector.queryNodes('vector_index_token', 10, $predicate_emb)
YIELD node AS vectorNode, score as vectorScore
WITH vectorNode, vectorScore
WHERE vectorScore >= 0.96
RETURN collect(vectorNode) AS similarPredicates
}
WITH similarPredicates
OPTIONAL MATCH (n:Token {name: toLower($predicate)})
WITH similarPredicates + CASE WHEN n IS NULL THEN [] ELSE [n] END AS allPredicates
UNWIND allPredicates AS predicate
RETURN collect(predicate) AS similarPredicates
"""
similarPredicates = run_query(similarPredicates_query, params)[0]['similarPredicates']
similarObjects_query = """
CALL () {
// Search for the object duplicates
CALL db.index.vector.queryNodes('vector_index_token', 10, $object_emb)
YIELD node AS vectorNode, score as vectorScore
WITH vectorNode, vectorScore
WHERE vectorScore >= 0.96
RETURN collect(vectorNode) AS similarObjects
}
WITH similarObjects
OPTIONAL MATCH (n:Token {name: toLower($object)})
WITH similarObjects + CASE WHEN n IS NULL THEN [] ELSE [n] END AS allObjects
UNWIND allObjects AS object
RETURN collect(object) AS similarObjects
"""
similarObjects = run_query(similarObjects_query, params)[0]['similarObjects']
query = """
UNWIND $similarSubjects AS subject
UNWIND $similarPredicates AS predicate
UNWIND $similarObjects AS object
WITH subject.name AS subjectName, predicate.name AS predicateName, object.name AS objectName, subject, predicate, object
MERGE (subjectNode:Token {name: toLower(subjectName)})
ON CREATE SET subjectNode.embeddings = $subject_emb, subjectNode.triplet_part = 'subject'
ON MATCH SET subjectNode.triplet_part = 'subject'
//MERGE (predicateNode:Token {name: toLower(predicateName)})
//ON CREATE SET predicateNode.embeddings = $predicate_emb, predicateNode.triplet_part = 'predicate'
//ON MATCH SET predicateNode.triplet_part = 'predicate'
MERGE (objectNode:Token {name: toLower(objectName)})
ON CREATE SET objectNode.embeddings = $object_emb, objectNode.triplet_part = 'object'
ON MATCH SET objectNode.triplet_part = 'object'
MERGE (subjectNode)-[r:predicate {name: toLower(predicateName)}]->(objectNode)
ON CREATE SET r.label = 'triplet', r.embeddings = $predicate_emb
ON MATCH SET r.label = 'triplet'
RETURN subjectName AS subject, predicateName AS predicate, objectName AS object
"""
final_params = {
'similarSubjects': similarSubjects,
'similarPredicates': similarPredicates,
'similarObjects': similarObjects,
'subject_emb': subject_emb.tolist(),
'predicate_emb': predicate_emb.tolist(),
'object_emb': object_emb.tolist()
}
results = run_query(query, final_params)
print(f"Processed triplet: {triplet}")
return results
基于NLP的GraphRAG
图7展示了混合RAG/GraphRAG方法在同一问题上的结果,该问题仅使用文档层检索(代表纯RAG,见图3)提出。答案更加全面,提供了对数据更深入的见解。
请注意,我没有执行任何实体解析或实体链接,这肯定是接下来的步骤,并且很可能会提高性能。此外,对于两次检索测试,我都传递了恰好10段检索到的文本。GraphRAG耗时几乎是RAG的两倍。虽然我们牺牲了一些延迟,但我们获得了更高的答案准确性。
图7
下面是使用三元组关系的检索函数。
def triplets_driven_retrieval(my_query):
my_query_emb = emb.embed_query(my_query)
query = """
CALL db.index.vector.queryNodes('vector_index_token', 300, $user_query_emb)
YIELD node AS token, score AS tokenScore
CALL (token, tokenScore) {
MATCH (token)
WHERE token.triplet_part IS NOT NULL
OPTIONAL MATCH (token)-[:predicate]->(object)
OPTIONAL MATCH (object)-[:predicate]->(subject)
OPTIONAL MATCH (subject)-[:CONTAINS]->(doc:Document)
RETURN DISTINCT doc, tokenScore as score, 1 AS isTripletPath
ORDER BY tokenScore DESC
LIMIT 200
UNION
MATCH (token)
WHERE token.triplet_part IS NULL
MATCH (token)-[:CONTAINS]-(doc:Document)
RETURN DISTINCT doc, tokenScore as score, 2 AS isTripletPath
ORDER BY tokenScore DESC
LIMIT 200
}
RETURN DISTINCT doc.full_text AS document_text, score, isTripletPath
ORDER BY score DESC
LIMIT 100
UNION
CALL () {
CALL db.index.vector.queryNodes('vector_index_document', 10, $user_query_emb)
YIELD node AS doc, score as vectorScore
WITH doc, vectorScore
ORDER BY vectorScore DESC
RETURN DISTINCT doc,
vectorScore AS score, 3 AS isTripletPath
ORDER BY vectorScore DESC
LIMIT 10
}
RETURN DISTINCT doc.full_text AS document_text, score, isTripletPath
ORDER BY score DESC
LIMIT 10
"""
params = {'user_query_emb': my_query_emb.tolist()}
results = run_query(query, params)
df = pd.DataFrame(data=results)
return df
你可以通过调整查询逻辑,以最佳方式遍历你的图。但让我们来看看上面展示的GraphRAG Cypher查询在做什么。这个查询分为几个步骤构建。首先,我们使用向量索引在用户查询和Token节点之间进行匹配。我们检查token是否有一个名为triplet_part的属性(这是从生成的三元组映射过来的token)。当我们遍历三元组并到达主语节点时,我们获取所有指向它的宾语节点,并选择附加在这些节点上的所有文档块,对搜索进行排序和限制。如果token没有三元组对,我们只需遍历到其所属的块。在查询的第二部分,我们执行标准的RAG搜索,并使用向量索引选择文档。
我相信这个查询可以进一步优化。另外值得一提的是,我还使用了spaCy的命名实体提取功能,提取了像ORG、DATE等token分类标签(见标题图像中的红色节点)。然而,结果并不太好,所以我坚持使用了两层架构。
看看针对简单用户问题“系统中提到了哪些公司?”(图8)的Cypher查询子图是什么样子,这很有趣。
图8
结果由两个截然不同的部分组成:一部分是从标准RAG部分检索到的、大多不连贯的文本块,另一部分是一组节点,其中包含主要的“主语”三元组节点,在本例中为“公司”。这种表示方法在检索查询的优化阶段非常有帮助。
结论
总之,本文提出了一种基于NLP的方法来构建知识图,该方法为RAG应用执行混合RAG/GraphRAG,而不严重依赖LLM。该方法涉及分层图,无需包含固定本体。
初步结果表明,使用这种混合检索方法回答的问题能提供更全面、更有见地的答案,为在大规模通用人工智能项目中进行进一步探索和潜在实施提供了可能性。