图片由作者提供
实现检索增强生成（RAG）、语义搜索和推荐的分步教程本教程的配套代码在这里:(here.)
我的上一篇博客文章是关于如何在企业级实现知识图谱（KGs）和大型语言模型（LLMs）的。在这篇文章中，我介绍了当前KGs和LLMs交互的两种方式：LLMs作为构建KGs的工具；以及KGs作为输入进入LLM或GenAI应用程序。下图展示了这两种集成方式以及人们如何将它们结合起来使用。
图片由作者提供
在这篇文章中，我将重点介绍知识图谱（KGs）和大型语言模型（LLMs）结合使用的一种流行方式：使用知识图谱的检索增强生成（RAG），有时也称为Graph RAG，GraphRAG，GRAG，或语义RAG。检索增强生成（RAG）是指从相关信息中检索并增强发送给大型语言模型（LLM）的提示，LLM将生成响应。其核心思想是，与其直接将提示发送给未经过你数据训练的LLM，不如用相关的信息来补充提示，使LLM能够更准确地回答你的问题。我在上一篇文章中举的例子是将工作描述和我的简历复制到ChatGPT中，以撰写求职信。如果我给LLM提供我的简历和我申请的工作描述，它将能够更准确地回答我的提示：“写一封求职信”。由于知识图谱是为存储知识而设计的，它们是存储内部数据和补充LLM提示以提供额外上下文的完美方式，从而提高响应的准确性和上下文理解。
重要的是，我认为这一点经常被误解，RAG 和使用知识图谱（KG）的 RAG（Graph RAG）是结合技术的方法论，而不是一种产品或技术本身。没有人发明、拥有或垄断 Graph RAG。大多数人可以看到这两种技术结合在一起的潜力，而且有更多和更多的研究证明了它们结合的好处。
一般来说，有三种方式可以使用知识图谱（KG）来进行检索增强生成（RAG）的检索部分：
- 基于向量的检索: 将您的知识图谱（KG）向量化并存储在向量数据库中。如果您再将自然语言提示向量化，就可以在向量数据库中找到与提示最相似的向量。由于这些向量对应于图中的实体，因此可以根据自然语言提示返回图中最相关的实体。注意，您可以在没有图的情况下进行基于向量的检索。实际上，这是最初实现RAG的方式，有时称为Baseline RAG。您会将SQL数据库或内容向量化，并在查询时检索它们。
- 提示到查询的检索: 使用大型语言模型（LLM）为您生成SPARQL或Cypher查询，使用该查询对您的KG进行查询，然后使用返回的结果来增强您的提示。
- 混合（向量 + SPARQL）: 您可以以各种有趣的方式结合这两种方法。在本教程中，我将演示一些结合这些方法的方法。我将主要专注于使用向量化进行初始检索，然后使用SPARQL查询来细化结果。
然而，将向量数据库和知识图谱（KGs）结合用于搜索、相似性分析和RAG的方法有很多。这只是为了突出每种方法的优缺点以及它们结合使用的优点的一个示例。在这里，我使用向量化进行初始检索，然后使用SPARQL进行过滤的方法并不独特。我曾在其他地方看到过类似实现。一个我听说的例子是来自一家大型家具制造商的人。他说，向量数据库可能会向购买沙发的人推荐除尘刷，但知识图谱会理解材料、属性和关系，并确保不会向购买皮质沙发的人推荐除尘刷。
在这篇教程中我将：
- 将数据集向量化存入向量数据库，以测试语义搜索、相似性搜索和RAG (基于向量的检索)
- 将数据转换为知识图谱以测试语义搜索、相似性搜索和RAG (从提示到查询的检索，虽然实际上是更像查询检索，因为我直接使用SPARQL而不是让大语言模型将自然语言提示转换为SPARQL查询)
- 将带有知识图谱标签和URI的数据集向量化存入向量数据库（我将这称为“向量化知识图谱”），并测试语义搜索、相似性和RAG (混合型)
目标是说明KGs和向量数据库在这方面的差异，并展示它们可以如何协同工作。以下是向量数据库和知识图谱如何共同执行高级查询的高层次概述。
图片由作者提供
如果你不想继续阅读，这里有个“ tl;dr ”:
- 向量数据库可以很好地运行语义搜索、相似性计算和一些基本形式的RAG，但有一些限制。第一个限制是，我使用的数据包含期刊文章的摘要，即包含大量的非结构化文本。向量化模型主要是在非结构化数据上训练的，因此在给定与实体相关的文本块时表现良好。
- 话虽如此，将数据导入向量数据库并准备好查询的开销非常小。如果你的数据集中包含一些非结构化数据，你可以在15分钟内完成向量化并开始搜索。
- 不出所料，仅使用向量数据库的一个最大缺点是缺乏可解释性。响应可能有三个好的结果和一个没有意义的结果，而无法知道为什么会有那个第四结果。
- 向量数据库返回无关内容的可能性对于搜索和相似性来说是一个麻烦，但对于RAG来说是一个大问题。如果你用四篇文章来增强提示，其中一篇是完全不相关的主题，那么LLM的响应将具有误导性。这通常被称为“上下文中毒”。
- 上下文中毒尤其危险的是，响应不一定在事实上不准确，也不是基于不准确的数据，只是使用了错误的数据来回答你的问题。我在本教程中找到的一个例子是提示“口腔肿瘤的治疗方法”。检索到的一篇文章是关于直肠癌治疗研究的，被发送给LLM进行总结。我不是医生，但我肯定直肠不是口腔的一部分。LLM准确地总结了研究和不同治疗方案对口腔和直肠癌的影响，但并不总是提到癌症类型。因此，用户会不知不觉地阅读LLM描述直肠癌的不同治疗方案，而他们要求的是口腔癌的治疗方法。
- KG进行语义搜索和相似性搜索的能力取决于你的元数据质量和元数据连接的受控词汇表的质量。在本教程中的示例数据集中，所有期刊文章都已使用主题术语标记。这些术语属于美国国立卫生研究院的医学主题词表（MeSH）的丰富受控词汇表。因此，我们可以轻松地进行语义搜索和相似性搜索。
- 将KG直接向量化到向量数据库以作为RAG的知识库使用，可能有一些好处，但我没有在本教程中这样做。我只是将数据以表格格式向量化，并为每篇文章添加了一个URI列，以便将向量连接回KG。
- 使用KG进行语义搜索、相似性和RAG的一个最大优势在于可解释性。你可以始终解释为什么返回某些结果：它们被标记为某些概念或具有某些元数据属性。
- KG的另一个我没有预料到的好处有时被称为“增强数据丰富”或“图作为专家”——你可以使用KG来扩展或细化你的搜索词。例如，你可以找到相似的术语、更具体的术语或与你的搜索词以特定方式相关的术语，以扩展或细化你的查询。例如，我可能会从搜索“口腔癌”开始，但根据我的KG术语和关系，将搜索细化为“牙龈肿瘤和腭部肿瘤”。
- 使用KG的一个最大障碍是需要构建KG。话虽如此，有许多方法可以使用LLM来加速KG的构建（如上图1所示）。
- 仅使用KG的一个缺点是，你需要编写SPARQL查询来完成所有操作。因此，上述提示到查询检索的流行性。
- 使用Jaccard相似性在知识图谱中找到相似文章的结果很差。如果没有指定，KG返回具有诸如“老年”、“男性”和“人类”等重叠标签的文章，这些标签可能不如“治疗选择”或“口腔肿瘤”相关。
- 我面临的另一个问题是，Jaccard相似性运行了很长时间（大约30分钟）。我不知道是否有更好的方法（欢迎建议），但我猜测这可能是因为在一篇文章和9,999其他文章之间找到重叠标签非常计算密集。
- 由于我在本教程中使用的示例提示是像“总结这些文章”这样简单的东西——LLM响应的准确性（无论是基于向量的还是基于KG的检索方法）在很大程度上取决于检索而不是生成。我的意思是，只要你给LLM相关的上下文，它几乎不可能在像“总结”这样简单的提示上出错。当然，如果我们的提示更复杂，情况就会不同。
- 使用向量数据库进行初始搜索，然后使用KG进行过滤提供了最佳结果。这在某种程度上是显而易见的——你不会过滤以获得更差的结果。但这就是重点：不是KG本身提高了结果，而是KG为你提供了控制输出以优化结果的能力。
- 使用KG过滤结果可以提高响应的准确性和相关性，也可以根据提示的编写者自定义结果。例如，我们可能希望使用相似性搜索来找到类似的文章推荐给用户，但我们只希望推荐用户可以访问的文章。KG允许在查询时进行访问控制。
- KG还可以帮助减少上下文中毒的可能性。在上述RAG示例中，我们可以在向量数据库中搜索“口腔肿瘤的治疗方法”，然后过滤仅标记为口腔肿瘤（或相关概念）的文章。
- 在本教程中，我只专注于一个简单的实现，我们将提示直接发送到向量数据库，然后使用图过滤结果。有更好的方法可以做到这一点。例如，你可以从提示中提取与你的受控词汇表对齐的实体，并使用图丰富它们（使用同义词和更具体的术语）；你可以将提示解析为语义块并分别发送到向量数据库；你可以在向量化之前将RDF数据转换为文本，以便语言模型更好地理解它，等等。这些是未来博客文章的主题。
下面的图展示了整体计划。我们希望将期刊文章的摘要和标题向量化并存储到向量数据库中，以便运行不同的查询：语义搜索、相似性搜索以及 RAG 的简单版本。对于语义搜索，我们将测试一个术语如“口腔肿瘤”——向量数据库应返回与该主题相关的文章。对于相似性搜索，我们将使用给定文章的 ID 来查找其在向量空间中的最近邻，即与该文章最相似的文章。最后，向量数据库允许我们使用 RAG 的一种形式，即我们可以用一篇文章来补充提示，例如，“请像向没有医学学位的人解释一样解释这个内容”。
图片由作者提供
我决定使用来自PubMed库的这个数据集（许可CC0: 公共领域）：这个数据集，它包含了50,000篇研究文章。该数据集包括文章的标题、摘要，以及用于元数据标签的字段。这些标签来自医学主题词表（MeSH）控制词汇表。为了本教程的这一部分，我们仅使用摘要和标题。这是因为我们试图将向量数据库与知识图谱进行比较，而向量数据库的优势在于能够“理解”缺乏丰富元数据的非结构化数据。我仅使用了数据集中的前10,000行，只是为了使计算更快。
这里 是 Weaviate 的官方快速入门教程。我还发现 这篇文章 在入门时很有帮助。
from weaviate.util import generate_uuid5
import weaviate
import json
import pandas as pd
# 读取 pubmed 数据
df = pd.read_csv("PubMed Multi Label Text Classification Dataset Processed.csv")然后我们可以连接到我们的Weaviate集群：
client = weaviate.Client(
url = "XXX", # 用你的Weaviate端点替换
auth_client_secret=weaviate.auth.AuthApiKey(api_key="XXX"), # 用你的Weaviate实例API密钥替换
additional_headers = {
"X-OpenAI-Api-Key": "XXX" # 用你的推理API密钥替换
}
)在我们将数据向量化并存入向量数据库之前，我们必须定义模式。在这里，我们定义想要向量化的csv中的哪些列。如前所述，为了本教程的目的，我们首先只想向量化标题和摘要列。
class_obj = {
# 类定义
"class": "articles",
# 属性定义
"properties": [
{
"name": "title",
"dataType": ["text"],
},
{
"name": "abstractText",
"dataType": ["text"],
},
],
# 指定向量化器
"vectorizer": "text2vec-openai",
# 模块设置
"moduleConfig": {
"text2vec-openai": {
"vectorizeClassName": True,
"model": "ada",
"modelVersion": "002",
"type": "text"
},
"qna-openai": {
"model": "gpt-3.5-turbo-instruct"
},
"generative-openai": {
"model": "gpt-3.5-turbo"
}
},
}然后我们将此模式推送到我们的Weaviate集群：
client.schema.create_class(class_obj)你可以通过直接查看你的 Weaviate 集群来检查这是否成功。
现在我们已经建立了模式，可以将所有数据写入向量数据库了。
import logging
import numpy as np
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
# 将无穷大值替换为 NaN，然后填充 NaN 值
df.replace([np.inf, -np.inf], np.nan, inplace=True)
df.fillna('', inplace=True)
# 将列转换为字符串类型
df['Title'] = df['Title'].astype(str)
df['abstractText'] = df['abstractText'].astype(str)
# 记录数据类型
logging.info(f"Title 列类型: {df['Title'].dtype}")
logging.info(f"abstractText 列类型: {df['abstractText'].dtype}")
with client.batch(
batch_size=10, # 指定批处理大小
num_workers=2, # 并行化处理
) as batch:
for index, row in df.iterrows():
try:
question_object = {
"title": row.Title,
"abstractText": row.abstractText,
}
batch.add_data_object(
question_object,
class_name="articles",
uuid=generate_uuid5(question_object)
)
except Exception as e:
logging.error(f"处理第 {index} 行时出错: {e}")为了检查数据是否已进入集群，你可以运行以下命令：
client.query.aggregate("articles").with_meta_count().do()不知道为什么，只有9997行被向量化了。 ¯\(ツ)/¯
使用向量数据库进行语义搜索当我们谈到向量数据库中的语义时，意味着这些术语是通过大型语言模型（LLMs）API 被向量化到向量空间中的，该API经过大量非结构化内容的训练。这意味着向量会考虑术语的上下文。例如，如果在训练数据中，术语马克·吐温经常出现在术语塞缪尔·克莱门斯附近，那么这两个术语的向量在向量空间中应该很接近。同样，如果术语口腔癌与术语口腔肿瘤在训练数据中经常一起出现，我们期望关于口腔癌的文章的向量在向量空间中接近关于口腔肿瘤的文章的向量。
你可以通过运行一个简单的查询来检查是否成功：
response = (
client.query
.get("articles", ["title","abstractText"])
.with_additional(["id"])
.with_near_text({"concepts": ["Mouth Neoplasms"]})
.with_limit(10)
.do()
)
print(json.dumps(response, indent=4))以下是结果：
- Article 1: “Gingival metastasis as first sign of multiorgan dissemination of epithelioid malignant mesothelioma.” 这篇文章是关于对患有恶性间皮瘤（一种肺癌）的患者进行的研究，这种癌症已经扩散到他们的牙龈。研究测试了不同治疗方法（化疗、去皮术和放疗）对癌症的效果。这篇文章似乎是一个合适的返回结果——它关于牙龈肿瘤，是口腔肿瘤的一个子集。
- Article 2: “Myoepithelioma of minor salivary gland origin. Light and electron microscopical study.” 这篇文章是关于从一个14岁男孩的牙龈中移除的肿瘤，该肿瘤已经扩散到上颌的一部分，并且由起源于唾液腺的细胞组成。这篇文章似乎也是一个合适的返回结果——它关于从一个男孩的口腔中移除的肿瘤。
- Article 3: “Metastatic neuroblastoma in the mandible. Report of a case.” 这篇文章是一个关于一个5岁男孩下颌癌的案例研究。这篇文章是关于癌症，但技术上不是口腔癌——下颌肿瘤（下颌的肿瘤）不属于口腔肿瘤的子集。
这就是我们所说的语义搜索——这些文章的标题或摘要中都没有提到“口腔”这个词。第一篇文章是关于牙龈（牙龈）肿瘤，这是口腔肿瘤的一个子集。第二篇文章是关于起源于受试者唾液腺的牙龈肿瘤，这也是口腔肿瘤的一个子集。第三篇文章是关于下颌肿瘤——根据MeSH词汇表，这不是口腔肿瘤的子集。然而，向量数据库知道下颌与口腔很接近。
使用向量数据库进行相似性搜索我们也可以使用向量数据库来查找相似的文章。我选择了上面使用“口腔肿瘤”查询返回的一篇文章，标题为：“牙龈转移作为上皮样恶性间皮瘤多器官扩散的首个迹象。” 使用该文章的ID，我可以查询向量数据库以查找所有相似的实体：
response = (
client.query
.get("articles", ["title", "abstractText"])
.with_near_object({
"id": "a7690f03-66b9-5d17-b765-8c6eb21f99c8" # 给定文章的ID
})
.with_limit(10)
.with_additional(["distance"])
.do()
)
print(json.dumps(response, indent=2))结果按照相似度顺序排列。相似度是通过向量空间中的距离来计算的。如你所见，排名最靠前的结果是关于牙龈的文章——这篇文章与自身最为相似。
其他文章包括：
- 第四条：“针对烟草使用者进行口腔恶性病变筛查的可行性研究。”这一条是关于口腔癌的，但重点是如何让吸烟者参加筛查，而不是讨论治疗方法。
- 第五条: “对于老年人，胸膜剥脱和剥除术治疗恶性胸膜间皮瘤是一种有效且安全的减瘤手术。”这一条是关于一项研究，探讨在老年人中使用胸膜剥脱和剥除术（一种从肺部移除癌症的手术）治疗胸膜间皮瘤（肺部癌症）的效果。所以，这与治疗间皮瘤有关，但与牙龈肿瘤无关。
- 第三条（上面提到的）: “下颌转移性神经母细胞瘤。一例报告。”这一条是关于那个5岁男孩下颌癌症的案例。这与癌症有关，但严格来说不是口腔癌，而且这并不是关于治疗结果的文章，与牙龈肿瘤的文章不同。
所有这些文章，可以说都与我们最初的牙龈文章类似。很难评估它们有多相似，因此也很难评估相似性搜索的效果如何，因为这很大程度上取决于用户对“相似”的定义。如果你对其他关于间皮瘤治疗的文章感兴趣，而第一篇文章只是讨论了它如何扩散到牙龈，那么这种情况与你无关？那么在这种情况下，Article 5 最为相似。或者你是否对减少任何类型的口腔癌症感兴趣，无论是通过治疗还是预防？那么在这种情况下，Article 4 最为相似。向量数据库的一个缺点是它是黑盒子——我们不知道为什么这些文章被返回。
基于向量数据库的检索增强生成（RAG）这里是如何使用向量数据库检索结果，然后将这些结果发送给大型语言模型进行摘要的一个示例——RAG的应用。
response = (
client.query
.get("articles", ["title", "abstractText"])
.with_near_text({"concepts": ["Gingival metastasis as first sign of multiorgan dissemination of epithelioid malignant mesothelioma"]})
.with_generate(single_prompt="请像向没有医学学位的人解释这篇文章 {title} 一样解释。")
.with_limit(1)
.do()
)
print(json.dumps(response, indent=4))你可以看到下面的回复：
"当然！这篇文章讲述了一个患有称为上皮样恶性间皮瘤的癌症的病例。这种癌症通常起源于肺部或腹部的内层。然而，在这个案例中，癌症首次扩散到身体其他部位的迹象出现在牙龈（牙龈）上。这种情况被称为牙龈转移。\n\n转移意味着癌细胞已经从原发肿瘤扩散到身体的其他部位。在这个案例中，癌症在扩散到其他器官之前就已经扩散到了牙龈。这种情况很重要，因为它表明癌症已经处于晚期，并且在被发现之前就已经扩散到了多个器官。\n\n总的来说，这篇文章强调了早期发现和监测癌症的重要性，以及癌症可能扩散到身体意想不到部位的可能性。"
我对这个回复感到失望。摘要清楚地解释了这项研究跟踪了13名患有转移性恶性间皮瘤并接受不同治疗的患者及其结果。而RAG的输出却将文章描述为关于“一个人”，完全没有提到这项研究。
与其只总结一篇文章，我们尝试总结几篇文章。在接下来的例子中，我们使用与上面相同的搜索词（口腔肿瘤），然后将排名前三的文章连同提示“用项目符号列出此处的关键信息。使其易于没有医学学位的人理解”发送给大型语言模型（LLM）。
response = (
client.query
.get(collection_name, ["title", "abstractText"])
.with_near_text({"concepts": ["Mouth Neoplasms"]})
.with_limit(3)
.with_generate(grouped_task="用项目符号列出此处的关键信息。使其易于没有医学学位的人理解。")
.do()
)
print(response["data"]["Get"]["Articles"][0]["_additional"]["generate"]["groupedResult"])以下是结果：
- 口腔转移性恶性间皮瘤较为罕见，颌骨中的病例比软组织中的多
- 这种癌症的平均生存率为9-12个月
- 对13名接受新辅助化疗和手术的患者的研究显示，中位生存期为11个月
- 一名患者首次出现多器官复发的间皮瘤表现为牙龈肿块
- 对于有间皮瘤病史的患者，对新出现的病变进行活检，即使在不常见的部位，也很重要
- 来自小唾液腺的肌上皮瘤可以表现出恶性潜能的特征
- 颌骨中的转移性神经母细胞瘤非常罕见，儿童患者可能出现骨溶解性下颌缺陷和乳牙松动这比之前的回复更让我满意——它提到了在Article 1中进行的研究、治疗方法和结果。倒数第二个项目符号是关于文章 “小唾液腺源性肌上皮瘤。光镜和电镜研究” 的，似乎是一个准确的一句话描述。最后一个项目符号是关于上述的Article 3，同样似乎是一个准确的一句话描述。
步骤2：使用知识图谱进行数据检索这里是我们如何使用知识图谱进行语义搜索、相似性搜索和RAG的高层次概述：
图片由作者提供
使用知识图谱检索数据的第一步是将数据转换为RDF格式。下面的代码创建了所有数据类型的类和属性，并填充了文章和MeSH术语的实例。我还创建了发布日期和访问级别属性，并用随机值填充它们，仅作为演示。
from rdflib import Graph, RDF, RDFS, Namespace, URIRef, Literal
from rdflib.namespace import SKOS, XSD
import pandas as pd
import urllib.parse
import random
from datetime import datetime, timedelta
# 创建一个新的RDF图
g = Graph()
# 定义命名空间
schema = Namespace('http://schema.org/')
ex = Namespace('http://example.org/')
prefixes = {
'schema': schema,
'ex': ex,
'skos': SKOS,
'xsd': XSD
}
for p, ns in prefixes.items():
g.bind(p, ns)
# 定义类和属性
Article = URIRef(ex.Article)
MeSHTerm = URIRef(ex.MeSHTerm)
g.add((Article, RDF.type, RDFS.Class))
g.add((MeSHTerm, RDF.type, RDFS.Class))
title = URIRef(schema.name)
abstract = URIRef(schema.description)
date_published = URIRef(schema.datePublished)
access = URIRef(ex.access)
g.add((title, RDF.type, RDF.Property))
g.add((abstract, RDF.type, RDF.Property))
g.add((date_published, RDF.type, RDF.Property))
g.add((access, RDF.type, RDF.Property))
# 解析MeSH术语的函数
def parse_mesh_terms(mesh_list):
if pd.isna(mesh_list):
return []
return [term.strip().replace(' ', '_') for term in mesh_list.strip("[]'").split(',')]
# 创建有效URI的函数
def create_valid_uri(base_uri, text):
if pd.isna(text):
return None
sanitized_text = urllib.parse.quote(text.strip().replace(' ', '_').replace('"', '').replace('<', '').replace('>', '').replace("'", "_"))
return URIRef(f"{base_uri}/{sanitized_text}")
# 生成过去五年内随机日期的函数
def generate_random_date():
start_date = datetime.now() - timedelta(days=5*365)
random_days = random.randint(0, 5*365)
return start_date + timedelta(days=random_days)
# 生成1到10之间随机访问值的函数
def generate_random_access():
return random.randint(1, 10)
# 在这里加载你的DataFrame
# df = pd.read_csv('your_data.csv')
# 遍历DataFrame中的每一行并创建RDF三元组
for index, row in df.iterrows():
article_uri = create_valid_uri("http://example.org/article", row['Title'])
if article_uri is None:
continue
# 添加文章实例
g.add((article_uri, RDF.type, Article))
g.add((article_uri, title, Literal(row['Title'], datatype=XSD.string)))
g.add((article_uri, abstract, Literal(row['abstractText'], datatype=XSD.string)))
# 添加随机日期和访问级别
random_date = generate_random_date()
random_access = generate_random_access()
g.add((article_uri, date_published, Literal(random_date.date(), datatype=XSD.date)))
g.add((article_uri, access, Literal(random_access, datatype=XSD.integer)))
# 添加MeSH术语
mesh_terms = parse_mesh_terms(row['meshMajor'])
for term in mesh_terms:
term_uri = create_valid_uri("http://example.org/mesh", term)
if term_uri is None:
continue
# 添加MeSH术语实例
g.add((term_uri, RDF.type, MeSHTerm))
g.add((term_uri, RDFS.label, Literal(term.replace('_', ' '), datatype=XSD.string)))
# 将文章链接到MeSH术语
g.add((article_uri, schema.about, term_uri))
# 将图序列化到文件（可选）
g.serialize(destination='ontology.ttl', format='turtle')现在我们可以测试语义搜索。但在知识图谱的上下文中，语义这个词稍有不同。在知识图谱中，我们依赖于文档关联的标签及其在MeSH分类法中的关系来实现语义。例如，一篇文章可能关于唾液腺中的癌症（唾液腺肿瘤），但仍可能被标记为口腔肿瘤。
与其查询所有标记为“口腔肿瘤”的文章，我们还将查找任何比“口腔肿瘤”更具体的概念。MeSH词汇表不仅包含术语的定义，还包含诸如更广泛和更具体的关系。
from SPARQLWrapper import SPARQLWrapper, JSON
def get_concept_triples_for_term(term):
sparql = SPARQLWrapper("https://id.nlm.nih.gov/mesh/sparql")
query = f"""
PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX meshv: <http://id.nlm.nih.gov/mesh/vocab#>
PREFIX mesh: <http://id.nlm.nih.gov/mesh/>
SELECT ?subject ?p ?pLabel ?o ?oLabel
FROM <http://id.nlm.nih.gov/mesh>
WHERE {{
?subject rdfs:label "{term}"@en .
?subject ?p ?o .
FILTER(CONTAINS(STR(?p), "concept"))
OPTIONAL {{ ?p rdfs:label ?pLabel . }}
OPTIONAL {{ ?o rdfs:label ?oLabel . }}
}}
"""
sparql.setQuery(query)
sparql.setReturnFormat(JSON)
results = sparql.query().convert()
triples = set() # 使用集合来避免重复条目
for result in results["results"]["bindings"]:
obj_label = result.get("oLabel", {}).get("value", "No label")
triples.add(obj_label)
# 将术语本身添加到列表中
triples.add(term)
return list(triples) # 转换回列表以便更容易处理
def get_narrower_concepts_for_term(term):
sparql = SPARQLWrapper("https://id.nlm.nih.gov/mesh/sparql")
query = f"""
PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX meshv: <http://id.nlm.nih.gov/mesh/vocab#>
PREFIX mesh: <http://id.nlm.nih.gov/mesh/>
SELECT ?narrowerConcept ?narrowerConceptLabel
WHERE {{
?broaderConcept rdfs:label "{term}"@en .
?narrowerConcept meshv:broaderDescriptor ?broaderConcept .
?narrowerConcept rdfs:label ?narrowerConceptLabel .
}}
"""
sparql.setQuery(query)
sparql.setReturnFormat(JSON)
results = sparql.query().convert()
concepts = set() # 使用集合来避免重复条目
for result in results["results"]["bindings"]:
subject_label = result.get("narrowerConceptLabel", {}).get("value", "No label")
concepts.add(subject_label)
return list(concepts) # 转换回列表以便更容易处理
def get_all_narrower_concepts(term, depth=2, current_depth=1):
# 创建一个字典来存储术语及其更具体的概念
all_concepts = {}
# 初始获取主要术语
narrower_concepts = get_narrower_concepts_for_term(term)
all_concepts[term] = narrower_concepts
# 如果当前深度小于所需深度，递归获取更具体的概念
if current_depth < depth:
for concept in narrower_concepts:
# 递归调用以获取当前概念的更具体概念
child_concepts = get_all_narrower_concepts(concept, depth, current_depth + 1)
all_concepts.update(child_concepts)
return all_concepts
# 获取替代名称和更具体的概念
term = "Mouth Neoplasms"
alternative_names = get_concept_triples_for_term(term)
all_concepts = get_all_narrower_concepts(term, depth=2) # 根据需要调整深度
# 输出替代名称
print("替代名称:", alternative_names)
print()
# 输出更具体的概念
for broader, narrower in all_concepts.items():
print(f"更广泛的概念: {broader}")
print(f"更具体的概念: {narrower}")
print("---")以下是一切与口腔肿瘤相关的替代名称和更具体的概念。
图片由作者提供
我们将这转换为一个术语的扁平列表：
def flatten_concepts(concepts_dict):
flat_list = []
def recurse_terms(term_dict):
for term, narrower_terms in term_dict.items():
flat_list.append(term)
if narrower_terms:
recurse_terms(dict.fromkeys(narrower_terms, [])) # 使用空字典进行递归
recurse_terms(concepts_dict)
return flat_list
# 扁平化概念字典
flat_list = flatten_concepts(all_concepts)然后我们将术语转换为MeSH URI，以便我们可以将其纳入我们的SPARQL查询中：
# 将MeSH术语转换为URI
def convert_to_mesh_uri(term):
formatted_term = term.replace(" ", "_").replace(",", "_").replace("-", "_")
return URIRef(f"http://example.org/mesh/_{formatted_term}_")
# 将术语转换为URI
mesh_terms = [convert_to_mesh_uri(term) for term in flat_list]然后我们编写一个SPARQL查询来查找所有标记为“口腔肿瘤”、“口腔癌”或任何更具体术语的文章：
from rdflib import URIRef
query = """
PREFIX schema: <http://schema.org/>
PREFIX ex: <http://example.org/>
SELECT ?article ?title ?abstract ?datePublished ?access ?meshTerm
WHERE {
?article a ex:Article ;
schema:name ?title ;
schema:description ?abstract ;
schema:datePublished ?datePublished ;
ex:access ?access ;
schema:about ?meshTerm .
?meshTerm a ex:MeSHTerm .
}
"""
# 一个字典来存储文章及其相关的MeSH术语
article_data = {}
# 对每个MeSH术语运行查询
for mesh_term in mesh_terms:
results = g.query(query, initBindings={'meshTerm': mesh_term})
# 处理结果
for row in results:
article_uri = row['article']
if article_uri not in article_data:
article_data[article_uri] = {
'title': row['title'],
'abstract': row['abstract'],
'datePublished': row['datePublished'],
'access': row['access'],
'meshTerms': set()
}
# 将MeSH术语添加到该文章的集合中
article_data[article_uri]['meshTerms'].add(str(row['meshTerm']))
# 按匹配的MeSH术语数量对文章进行排序
ranked_articles = sorted(
article_data.items(),
key=lambda item: len(item[1]['meshTerms']),
reverse=True
)
# 获取排名前三的文章
top_3_articles = ranked_articles[:3]
# 输出结果
for article_uri, data in top_3_articles:
print(f"标题：{data['title']}")
print("MeSH术语：")
for mesh_term in data['meshTerms']:
print(f" - {mesh_term}")
print()返回的文章是：
- 第二条（来自上方）: “小唾液腺来源的肌上皮瘤。光镜和电子显微镜研究。”
- 第四条（来自上方）: “针对烟草使用者在口腔内恶性病变筛查的可行性研究。”
- 第六条: “胚胎致死异常视觉样蛋白HuR与环氧化酶-2在口腔鳞状细胞癌中的关联。” 这篇文章是一篇研究，旨在确定一种名为HuR的蛋白质的存在是否与环氧化酶-2的水平升高有关。环氧化酶-2在癌症的发展和癌细胞的扩散中起着重要作用。具体来说，这项研究专注于一种口腔癌，即口腔鳞状细胞癌。
这些结果与我们从向量数据库中得到的结果并不相距甚远。每篇文章都是关于口腔肿瘤的。知识图谱方法的好处在于，我们能够获得解释性——确切地知道为什么选择了这些文章。文章2被标记为“牙龈肿瘤”和“唾液腺肿瘤”。文章4和文章6都被标记为“口腔肿瘤”。由于文章2被标记了两个与我们的搜索词匹配的术语，因此它被排在了最高位置。
使用知识图谱进行相似性搜索与其使用向量空间来查找相似的文章，我们可以依赖文章关联的标签。使用标签来计算相似度有多种方法，但在这个例子中，我将使用一种常见的方法：Jaccard 相似度。我们将再次使用牙龈文章来进行不同方法的比较。
from rdflib import Graph, URIRef
from rdflib.namespace import RDF, RDFS, Namespace, SKOS
import urllib.parse
# 定义命名空间
schema = Namespace('http://schema.org/')
ex = Namespace('http://example.org/')
rdfs = Namespace('http://www.w3.org/2000/01/rdf-schema#')
# 计算 Jaccard 相似度并返回重叠项的函数
def jaccard_similarity(set1, set2):
intersection = set1.intersection(set2)
union = set1.union(set2)
similarity = len(intersection) / len(union) if len(union) != 0 else 0
return similarity, intersection
# 加载 RDF 图
g = Graph()
g.parse('ontology.ttl', format='turtle')
def get_article_uri(title):
# 将标题转换为 URI 安全字符串
safe_title = urllib.parse.quote(title.replace(" ", "_"))
return URIRef(f"http://example.org/article/{safe_title}")
def get_mesh_terms(article_uri):
query = """
PREFIX schema: <http://schema.org/>
PREFIX ex: <http://example.org/>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
SELECT ?meshTerm
WHERE {
?article schema:about ?meshTerm .
?meshTerm a ex:MeSHTerm .
FILTER (?article = <""" + str(article_uri) + """>)
}
"""
results = g.query(query)
mesh_terms = {str(row['meshTerm']) for row in results}
return mesh_terms
def find_similar_articles(title):
article_uri = get_article_uri(title)
mesh_terms_given_article = get_mesh_terms(article_uri)
# 查询所有文章及其 MeSH 术语
query = """
PREFIX schema: <http://schema.org/>
PREFIX ex: <http://example.org/>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
SELECT ?article ?meshTerm
WHERE {
?article a ex:Article ;
schema:about ?meshTerm .
?meshTerm a ex:MeSHTerm .
}
"""
results = g.query(query)
mesh_terms_other_articles = {}
for row in results:
article = str(row['article'])
mesh_term = str(row['meshTerm'])
if article not in mesh_terms_other_articles:
mesh_terms_other_articles[article] = set()
mesh_terms_other_articles[article].add(mesh_term)
# 计算 Jaccard 相似度
similarities = {}
overlapping_terms = {}
for article, mesh_terms in mesh_terms_other_articles.items():
if article != str(article_uri):
similarity, overlap = jaccard_similarity(mesh_terms_given_article, mesh_terms)
similarities[article] = similarity
overlapping_terms[article] = overlap
# 按相似度排序并获取前 15 篇
top_similar_articles = sorted(similarities.items(), key=lambda x: x[1], reverse=True)[:15]
# 打印结果
print(f"与 '{title}' 最相似的前 15 篇文章：")
for article, similarity in top_similar_articles:
print(f"文章 URI: {article}")
print(f"Jaccard 相似度: {similarity:.4f}")
print(f"重叠 MeSH 术语: {overlapping_terms[article]}")
print()
# 示例用法
article_title = "Gingival metastasis as first sign of multiorgan dissemination of epithelioid malignant mesothelioma."
find_similar_articles(article_title)结果如下。由于我们再次在牙龈文章上进行搜索，因此该文章是最相似的文章，这是我们可以预期的结果。其他结果为：
-
第7条: “股外侧肌钙化性肌腱炎。三例报告。” 这篇文章讨论的是股外侧肌（大腿肌肉）中的钙化性肌腱炎（肌腱中的钙沉积）。这与口腔肿瘤无关。
-
重叠术语: 断层摄影术、老年、男性、人类、X射线计算机断层扫描
-
第8条: “前列腺癌患者前列腺特异性抗原水平的雄激素剥夺治疗最佳持续时间。” 这篇文章讨论的是前列腺癌患者应接受特定治疗（雄激素剥夺疗法）的时长。这与癌症治疗（放射治疗）有关，但不是口腔癌的治疗。
-
重叠术语: 放射治疗、老年、男性、人类、辅助治疗
-
第9条: CT扫描大脑半球不对称性：预测失语症恢复的指标. 这篇文章讨论的是大脑左右半球的差异（大脑半球不对称性）如何预测中风后失语症患者的恢复情况。
- 重叠术语: 断层摄影术、老年、男性、人类、X射线计算机断层扫描
这种方法的优点在于，由于我们在这里计算相似度的方式，我们可以看到为什么其他文章是相似的——我们确切地看到哪些术语是重叠的，即哪些术语在牙龈文章和每个比较中是共通的。
解释性的缺点在于，从之前的成果来看，这些似乎并不是最相似的文章。它们都包含了三个共同的术语（Aged、Male 和 Humans），这些术语可能远没有 Treatment Options 或 Mouth Neoplasms 那样相关。你可以根据术语在整个语料库中的出现频率重新计算相似度——即使用词频-逆文档频率（TF-IDF）——这可能会改善结果。你还可以在进行相似度计算时选择对你来说最相关的标记术语，以更好地控制结果。
使用Jaccard相似性在知识图谱的术语上计算相似性的最大缺点是计算成本——运行这一计算花了大约30分钟。
使用知识图谱的RAG我们也可以仅使用知识图谱来实现检索增强生成（RAG）。我们已经有一份关于口腔肿瘤的文章列表，这些文章是上面语义搜索的结果。为了实现RAG，我们只需将这些文章发送给大型语言模型（LLM），并要求它对结果进行总结。
首先我们将每篇文章的标题和摘要合并成一个大的文本块，称为 combined_text:
# 合并摘要的函数
def combine_abstracts(top_3_articles):
combined_text = "".join(
[f"标题: {data['title']} 摘要: {data['abstract']}" for article_uri, data in top_3_articles]
)
return combined_text
# 合并前3篇文章的摘要
combined_text = combine_abstracts(top_3_articles)
print(combined_text)我们然后设置一个客户端，以便可以直接将此文本发送到大型语言模型：
import openai
# 设置你的 OpenAI API 密钥
api_key = "YOUR API KEY"
openai.api_key = api_key然后我们将上下文和提示传递给LLM：
def 生成摘要(组合文本):
response = openai.Completion.create(
model="gpt-3.5-turbo-instruct",
prompt=f"用项目符号列出关键信息。使其对没有医学学位的人也能理解：\n\n{组合文本}",
max_tokens=1000,
temperature=0.3
)
# 获取原始文本输出
原始摘要 = response.choices[0].text.strip()
# 将文本拆分为行并清理空白字符
行 = 原始摘要.split('\n')
行 = [line.strip() for line in 行 if line.strip()]
# 将行重新组合在一起，使用实际的换行符
格式化摘要 = '\n'.join(行)
return 格式化摘要
# 生成并打印摘要
摘要 = 生成摘要(组合文本)
print(摘要)结果如下所示：
- 一名14岁的男孩在上颌前部发现了一个牙龈肿瘤，该肿瘤被切除并通过光镜和电子显微镜进行了研究
- 肿瘤由肌上皮细胞组成，并且看起来是恶性的
- 电子显微镜显示肿瘤起源于唾液腺
- 这是唯一一例被确认具有恶性特征的肌上皮瘤
- 一项可行性研究在高发病率地区进行，旨在提高早期口腔癌和癌前病变的检测
- 烟草销售商参与分发传单，邀请吸烟者接受全科医生的免费检查
- 研究中共有93名患者，27%的患者被转诊给专科医生
- 27%中，63.6%的患者实际上见到了专科医生，其中15.3%被确认患有癌前病变
- 一项研究发现，口腔鳞状细胞癌（OSCC）中蛋白质HuR的表达增加与酶COX-2的表达之间存在相关性
- HuR的胞质表达与OSCC中COX-2的表达以及淋巴结和远处转移相关
- 抑制HuR的表达导致口腔癌细胞中COX-2表达的减少结果看起来不错，即它很好地总结了从语义搜索返回的三篇文章。仅使用知识图谱（KG）的RAG应用的质量取决于您的KG检索相关文档的能力。如本例所示，如果您的提示足够简单，例如，“总结此处的关键信息”，那么难点在于检索（为大语言模型提供正确的文章作为上下文），而不是生成响应。
步骤 3：使用向量化知识图谱测试数据检索现在我们想合力合作。我们将为数据库中的每篇文章添加一个 URI，然后在 Weaviate 中创建一个新的集合，在该集合中我们将对文章的名称、摘要、相关的 MeSH 术语以及 URI 进行向量化处理。URI 是文章的唯一标识符，也是我们连接回知识图谱的一种方式。
首先我们在数据中添加一个新的URI列：
# 创建有效URI的函数
def create_valid_uri(base_uri, text):
if pd.isna(text):
return None
# 对文本进行编码以用于URI
sanitized_text = urllib.parse.quote(text.strip().replace(' ', '_').replace('"', '').replace('<', '').replace('>', '').replace("'", "_"))
return URIRef(f"{base_uri}/{sanitized_text}")
# 为文章URI向DataFrame添加一个新的列
df['Article_URI'] = df['Title'].apply(lambda title: create_valid_uri("http://example.org/article", title)) 我们现在为新集合创建一个新的模式，并添加额外的字段：
class_obj = {
# 类定义
"class": "articles_with_abstracts_and_URIs",
# 属性定义
"properties": [
{
"name": "title",
"dataType": ["text"],
},
{
"name": "abstractText",
"dataType": ["text"],
},
{
"name": "meshMajor",
"dataType": ["text"],
},
{
"name": "Article_URI",
"dataType": ["text"],
},
],
# 指定向量化器
"vectorizer": "text2vec-openai",
# 模块设置
"moduleConfig": {
"text2vec-openai": {
"vectorizeClassName": True,
"model": "ada",
"modelVersion": "002",
"type": "text"
},
"qna-openai": {
"model": "gpt-3.5-turbo-instruct"
},
"generative-openai": {
"model": "gpt-3.5-turbo"
}
},
}将该模式推送到向量数据库：
client.schema.create_class(class_obj)现在我们将数据向量化到新的集合中：
import logging
import numpy as np
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
# 用 NaN 替换无穷大值，然后填充 NaN 值
df.replace([np.inf, -np.inf], np.nan, inplace=True)
df.fillna('', inplace=True)
# 将列转换为字符串类型
df['Title'] = df['Title'].astype(str)
df['abstractText'] = df['abstractText'].astype(str)
df['meshMajor'] = df['meshMajor'].astype(str)
df['Article_URI'] = df['Article_URI'].astype(str)
# 记录数据类型
logging.info(f"Title 列类型: {df['Title'].dtype}")
logging.info(f"abstractText 列类型: {df['abstractText'].dtype}")
logging.info(f"meshMajor 列类型: {df['meshMajor'].dtype}")
logging.info(f"Article_URI 列类型: {df['Article_URI'].dtype}")
with client.batch(
batch_size=10, # 指定批处理大小
num_workers=2, # 并行化处理
) as batch:
for index, row in df.iterrows():
try:
question_object = {
"title": row.Title,
"abstractText": row.abstractText,
"meshMajor": row.meshMajor,
"article_URI": row.Article_URI,
}
batch.add_data_object(
question_object,
class_name="articles_with_abstracts_and_URIs",
uuid=generate_uuid5(question_object)
)
except Exception as e:
logging.error(f"处理第 {index} 行时出错: {e}")现在我们可以在向量数据库上进行语义搜索，就像之前一样，但具有更多的可解释性和对结果的控制。
response = (
client.query
.get("articles_with_abstracts_and_URIs", ["title","abstractText","meshMajor","article_URI"])
.with_additional(["id"])
.with_near_text({"concepts": ["mouth neoplasms"]})
.with_limit(10)
.do()
)
print(json.dumps(response, indent=4))结果为：
- Article 1: "牙龈转移作为上皮样恶性间皮瘤多器官扩散的首个迹象。"
- Article 10: "一位老年男性中心面部血管中心性淋巴瘤的诊断挑战。" 这篇文章讲述了诊断一位男性鼻癌的困难。
- Article 11: "下颌假癌性增生。" 这篇文章对我来说非常难理解，但我认为它讲述的是假癌性增生看起来像癌症（因此名字中有“pseudo”），但实际上是非癌性的。虽然它确实与下颌有关，但它被标记为MeSH术语“口腔肿瘤”。
很难说这些结果是否比单独的知识图谱或向量数据库更好。理论上，结果应该会更好，因为每篇文章相关的MeSH术语现在也进行了向量化。然而，我们并没有真正对知识图谱进行向量化。例如，MeSH术语之间的关系并没有包含在向量数据库中。
拥有 MeSH 术语向量化的优点在于，我们可以立即获得一些解释性信息——例如，文章 11 也标记了口腔肿瘤。但是，将向量数据库连接到知识图谱真正酷的地方在于，我们可以从知识图谱中应用任何我们想要的过滤器。还记得我们之前如何将出版日期添加为数据字段吗？我们现在可以对此进行过滤。假设我们想找到 2020 年 5 月 1 日之后发表的关于口腔肿瘤的文章：
from rdflib import Graph, Namespace, URIRef, Literal
from rdflib.namespace import RDF, RDFS, XSD
# 定义命名空间
schema = Namespace('http://schema.org/')
ex = Namespace('http://example.org/')
rdfs = Namespace('http://www.w3.org/2000/01/rdf-schema#')
xsd = Namespace('http://www.w3.org/2001/XMLSchema#')
def get_articles_after_date(graph, article_uris, date_cutoff):
# 创建一个字典来存储每个 URI 的结果
results_dict = {}
# 使用文章 URI 列表和日期过滤器定义 SPARQL 查询
uris_str = " ".join(f"<{uri}>" for uri in article_uris)
query = f"""
PREFIX schema: <http://schema.org/>
PREFIX ex: <http://example.org/>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX xsd: <http://www.ww.org/2001/XMLSchema#>
SELECT ?article ?title ?datePublished
WHERE {{
VALUES ?article {{ {uris_str} }}
?article a ex:Article ;
schema:name ?title ;
schema:datePublished ?datePublished .
FILTER (?datePublished > "{date_cutoff}"^^xsd:date)
}}
"""
# 执行查询
results = graph.query(query)
# 提取每篇文章的详细信息
for row in results:
article_uri = str(row['article'])
results_dict[article_uri] = {
'title': str(row['title']),
'date_published': str(row['datePublished'])
}
return results_dict
date_cutoff = "2023-01-01"
articles_after_date = get_articles_after_date(g, article_uris, date_cutoff)
# 输出结果
for uri, details in articles_after_date.items():
print(f"文章 URI: {uri}")
print(f"标题: {details['title']}")
print(f"发表日期: {details['date_published']}")
print()原来的查询返回了十个结果（我们设置了最多十个），但其中只有六个是在2023年1月1日之后发布的。详见以下结果：
图片由作者提供
使用向量化知识图谱进行相似性搜索我们可以像之前对我们的牙龈文章（文章1）所做的那样，在这个新集合上运行相似性搜索：
response = (
client.query
.get("articles_with_abstracts_and_URIs", ["title","abstractText","meshMajor","article_URI"])
.with_near_object({
"id": "37b695c4-5b80-5f44-a710-e84abb46bc22"
})
.with_limit(50)
.with_additional(["distance"])
.do()
)
print(json.dumps(response, indent=2))结果如下：
- Article 3: "下颌转移性神经母细胞瘤。一例报告。"
- Article 4: “针对烟草使用者筛查口腔恶性病变的可行性研究。”
- Article 12: “弥漫性肺内恶性间皮瘤伪装成间质性肺病：一种间皮瘤的特殊变体。” 这篇文章讲述了五名男性患者，他们患有一种形式的间皮瘤，这种间皮瘤看起来很像另一种肺病：间质性肺病。
由于我们已经对 MeSH 进行了向量化处理，可以看到每篇文章相关的标签。其中一些标签虽然在某些方面可能相似，但并不是关于口腔肿瘤的。假设我们想要找到与我们的牙龈文章相似但具体关于口腔肿瘤的文章，我们现在可以将之前在知识图谱中使用的 SPARQL 过滤与这些结果结合起来。
MeSH 的 URI 已经保存了，但对于向量搜索返回的 50 篇文章，还需要获取它们的 URI：
# 假设 response 是包含文章的数据结构
article_uris = [URIRef(article["article_URI"]) for article in response["data"]["Get"]["Articles_with_abstracts_and_URIs"]]现在我们可以根据标签对结果进行排序，就像之前使用知识图谱进行语义搜索时那样。
from rdflib import URIRef
# 构造带有 FILTER 的 SPARQL 查询，用于文章 URI
query = """
PREFIX schema: <http://schema.org/>
PREFIX ex: <http://example.org/>
SELECT ?article ?title ?abstract ?datePublished ?access ?meshTerm
WHERE {
?article a ex:Article ;
schema:name ?title ;
schema:description ?abstract ;
schema:datePublished ?datePublished ;
ex:access ?access ;
schema:about ?meshTerm .
?meshTerm a ex:MeSHTerm .
# 过滤以仅包含 URI 列表中的文章
FILTER (?article IN (%s))
}
"""
# 将 URIRef 列表转换为适合 SPARQL 的字符串
article_uris_string = ", ".join([f"<{str(uri)}>" for uri in article_uris])
# 将文章 URI 插入查询中
query = query % article_uris_string
# 用于存储文章及其相关 MeSH 术语的字典
article_data = {}
# 对每个 MeSH 术语运行查询
for mesh_term in mesh_terms:
results = g.query(query, initBindings={'meshTerm': mesh_term})
# 处理结果
for row in results:
article_uri = row['article']
if article_uri not in article_data:
article_data[article_uri] = {
'title': row['title'],
'abstract': row['abstract'],
'datePublished': row['datePublished'],
'access': row['access'],
'meshTerms': set()
}
# 将 MeSH 术语添加到该文章的集合中
article_data[article_uri]['meshTerms'].add(str(row['meshTerm']))
# 根据匹配的 MeSH 术语数量对文章进行排序
ranked_articles = sorted(
article_data.items(),
key=lambda item: len(item[1]['meshTerms']),
reverse=True
)
# 输出结果
for article_uri, data in ranked_articles:
print(f"标题: {data['title']}")
print(f"摘要: {data['abstract']}")
print("MeSH 术语:")
for mesh_term in data['meshTerms']:
print(f" - {mesh_term}")
print()在向量数据库最初返回的50篇文章中，只有5篇被标记为“口腔肿瘤”或相关概念。
- Article 2: “小唾液腺来源的肌上皮瘤。光镜和电镜研究。” 标签：牙龈肿瘤，唾液腺肿瘤
- Article 4: “针对烟草使用者筛查口腔恶性病变的可行性研究。” 标签：口腔肿瘤
- Article 13: “起源于牙龈沟的表皮样癌。” 本文描述了一例牙龈癌（牙龈肿瘤）。标签：牙龈肿瘤
- Article 1: “牙龈转移作为上皮样恶性间皮瘤多器官播散的首个迹象。” 标签：牙龈肿瘤
- Article 14: “腮腺结节转移的CT和MRI影像学表现。” 本文讨论腮腺（大唾液腺）中的肿瘤。标签：腮腺肿瘤
最后，假设我们希望将这些相似的文章推荐给用户，但只推荐该用户有权访问的文章。假设我们知道该用户只能访问标记为访问级别3、5和7的文章。我们可以在知识图谱中使用类似的SPARQL查询来应用过滤：
from rdflib import Graph, Namespace, URIRef, Literal
from rdflib.namespace import RDF, RDFS, XSD, SKOS
# 假设您的 RDF 图（g）已经加载
# 定义命名空间
schema = Namespace('http://schema.org/')
ex = Namespace('http://example.org/')
rdfs = Namespace('http://www.w3.org/2000/01/rdf-schema#')
def filter_articles_by_access(graph, article_uris, access_values):
# 构造带有动态 VALUES 子句的 SPARQL 查询
uris_str = " ".join(f"<{uri}>" for uri in article_uris)
query = f"""
PREFIX schema: <http://schema.org/>
PREFIX ex: <http://example.org/>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
SELECT ?article ?title ?abstract ?datePublished ?access ?meshTermLabel
WHERE {{
VALUES ?article {{ {uris_str} }}
?article a ex:Article ;
schema:name ?title ;
schema:description ?abstract ;
schema:datePublished ?datePublished ;
ex:access ?access ;
schema:about ?meshTerm .
?meshTerm rdfs:label ?meshTermLabel .
FILTER (?access IN ({", ".join(map(str, access_values))}))
}}
"""
# 执行查询
results = graph.query(query)
# 提取每篇文章的详细信息
results_dict = {}
for row in results:
article_uri = str(row['article'])
if article_uri not in results_dict:
results_dict[article_uri] = {
'title': str(row['title']),
'abstract': str(row['abstract']),
'date_published': str(row['datePublished']),
'access': str(row['access']),
'mesh_terms': []
}
results_dict[article_uri]['mesh_terms'].append(str(row['meshTermLabel']))
return results_dict
access_values = [3,5,7]
filtered_articles = filter_articles_by_access(g, ranked_article_uris, access_values)
# 输出结果
for uri, details in filtered_articles.items():
print(f"文章 URI: {uri}")
print(f"标题: {details['title']}")
print(f"摘要: {details['abstract']}")
print(f"发布日期: {details['date_published']}")
print(f"访问级别: {details['access']}")
print()有一个文章用户无法访问。剩下的四篇文章是：
- Article 2: “小唾液腺来源的肌上皮瘤。光镜和电镜研究。” 标签：牙龈肿瘤，唾液腺肿瘤。访问级别：5
- Article 4: “针对烟草使用者筛查口腔恶性病变的可行性研究。” 标签：口腔肿瘤。访问级别：7
- Article 1: “**牙龈转移瘤作为上皮样恶性间皮瘤多器官播散的首个迹象。” 标签：牙龈肿瘤。访问级别：3
- Article 14: “腮腺结节转移：CT和MRI影像学表现。” 本文讨论腮腺，主要唾液腺中的肿瘤。标签：腮腺肿瘤。访问级别：3
最后，让我们看看在将向量数据库与知识图谱结合后，RAG是如何工作的。作为提醒，你可以直接将RAG与向量数据库结合，并将其发送到LLM以获取生成的响应：
response = (
client.query
.get("Articles_with_abstracts_and_URIs", ["title", "abstractText", 'article_URI', 'meshMajor'])
.with_near_text({"concepts": ["用于口腔肿瘤的疗法"]})
.with_limit(3)
.with_generate(grouped_task="用项目符号列出关键信息。使其易于没有医学学位的人理解。")
.do()
)
print(response["data"]["Get"]["Articles_with_abstracts_and_URIs"][0]["_additional"]["generate"]["groupedResult"])在这个例子中，我使用了搜索词“治疗口腔肿瘤的疗法”，并使用了相同的提示：“用项目符号列出这里的关键信息。使其对没有医学学位的人也能理解。”我们只返回前三篇文章来生成这个回复。以下是结果：
- 口腔转移性恶性间皮瘤较为罕见，平均生存期为9-12个月。
- 2012年8月至2013年9月，对13名患者使用了新辅助化疗和根治性胸膜剥脱术，随后进行了放疗。
- 2014年1月，11名患者仍然存活，中位生存期为11个月，而8名患者复发，2名患者分别在手术后8个月和9个月死亡。
- 一名68岁的男性患者出现牙龈肿块，最终被诊断为转移性恶性间皮瘤，导致多器官复发。
- 对于有间皮瘤病史的新生长病变，即使在不常见的部位，活检也很重要。
- 对于局部晚期直肠癌，新辅助放化疗可能有效，但一些患者可能反应不佳。
- 遗传改变可能与直肠癌对新辅助治疗的敏感性或耐药性有关。
- 在直肠癌肿瘤中发现了1p、8p、17p和18q的丢失，以及1q和13q的增加。
- 特定染色体区域的改变与对新辅助治疗的反应有关。
- 肿瘤细胞的细胞遗传学谱型可能会影响直肠癌放化疗的反应。
- 对于鼻咽癌，强度调制放射治疗在局部控制和总体生存率方面取得了良好的长期效果。
- 急性毒性反应包括黏膜炎、皮炎和口干，大多数患者经历了0-2级的毒性反应。
- 晚期毒性反应主要是口干，随着时间的推移有所改善。
- 远处转移仍然是治疗失败的主要原因，强调了需要更有效的全身治疗。 # 提取文章的URI
article_uris = [article["article_URI"] for article in response["data"]["Get"]["Articles_with_abstracts_and_URIs"]]
# 定义一个函数，用于过滤出给定URI的文章
def filter_articles_by_uri(response, article_uris):
filtered_articles = []
articles = response['data']['Get']['Articles_with_abstracts_and_URIs']
for article in articles:
if article['article_URI'] in article_uris:
filtered_articles.append(article)
return filtered_articles
# 过滤响应
filtered_articles = filter_articles_by_uri(response, article_uris)
# 输出过滤后的文章
print("过滤后的文章:")
for article in filtered_articles:
print(f"标题: {article['title']}")
print(f"URI: {article['article_URI']}")
print(f"摘要: {article['abstractText']}")
print(f"MeSH主要主题: {article['meshMajor']}")
print("---")有趣的是，第一篇文章是关于牙龈肿瘤的，这是口腔肿瘤的一个子集，但第二篇文章是关于直肠癌的，第三篇是关于鼻咽癌的。它们都是关于癌症的治疗方法，但并不是我搜索的那种癌症。令人担忧的是，提示是“口腔肿瘤的治疗方法”，而结果却包含了其他类型癌症治疗方法的信息。这种情况有时被称为“上下文中毒”——不相关或误导的信息被注入到提示中，导致大语言模型生成误导性的回复。
我们可以使用知识图谱来解决上下文中毒问题。以下是一个图表，展示了向量数据库和知识图谱如何协同工作以实现更好的RAG（检索增强生成）实施：
图片由作者提供
首先，我们使用向量数据库运行语义搜索，使用相同的提示：口腔癌症的治疗方法。这次我将限制增加到20篇文章，因为我们接下来会筛选掉一些。
response = (
client.query
.get("articles_with_abstracts_and_URIs", ["title", "abstractText", "meshMajor", "article_URI"])
.with_additional(["id"])
.with_near_text({"concepts": ["therapies for mouth neoplasms"]})
.with_limit(20)
.do()
)
# 提取文章URI
article_uris = [article["article_URI"] for article in response["data"]["Get"]["Articles_with_abstracts_and_URIs"]]
# 打印提取的文章URI
print("提取的文章URI:")
for uri in article_uris:
print(uri)接下来我们使用与之前相同的排序技术，使用口腔肿瘤相关概念：
from rdflib import URIRef
# 构造带有 FILTER 的 SPARQL 查询，用于文章 URI
query = """
PREFIX schema: <http://schema.org/>
PREFIX ex: <http://example.org/>
SELECT ?article ?title ?abstract ?datePublished ?access ?meshTerm
WHERE {
?article a ex:Article ;
schema:name ?title ;
schema:description ?abstract ;
schema:datePublished ?datePublished ;
ex:access ?access ;
schema:about ?meshTerm .
?meshTerm a ex:MeSHTerm .
# 过滤仅包含列表中的 URI 的文章
FILTER (?article IN (%s))
}
"""
# 将 URIRef 列表转换为适用于 SPARQL 的字符串
article_uris_string = ", ".join([f"<{str(uri)}>" for uri in article_uris])
# 将文章 URI 插入查询中
query = query % article_uris_string
# 用于存储文章及其相关 MeSH 术语的字典
article_data = {}
# 对每个 MeSH 术语运行查询
for mesh_term in mesh_terms:
results = g.query(query, initBindings={'meshTerm': mesh_term})
# 处理结果
for row in results:
article_uri = row['article']
if article_uri not in article_data:
article_data[article_uri] = {
'title': row['title'],
'abstract': row['abstract'],
'datePublished': row['datePublished'],
'access': row['access'],
'meshTerms': set()
}
# 将 MeSH 术语添加到该文章的集合中
article_data[article_uri]['meshTerms'].add(str(row['meshTerm']))
# 按匹配的 MeSH 术语数量对文章进行排序
ranked_articles = sorted(
article_data.items(),
key=lambda item: len(item[1]['meshTerms']),
reverse=True
)
# 输出结果
for article_uri, data in ranked_articles:
print(f"标题: {data['title']}")
print(f"摘要: {data['abstract']}")
print("MeSH 术语:")
for mesh_term in data['meshTerms']:
print(f" - {mesh_term}")
print()只有三个文章被标记为以下口腔肿瘤术语之一：
-
Article 4: “针对烟草使用者的口腔恶性病变筛查的可行性研究。” 标签：口腔肿瘤。
-
Article 15: “光敏剂介导的光动力疗法治疗化学诱导的前恶性病变和大鼠腭粘膜鳞状细胞癌。” 本文讨论了一种实验性癌症疗法（光动力疗法）用于治疗大鼠的腭癌。标签：腭肿瘤。
- Article 1: “牙龈转移作为多器官播散性上皮样恶性间皮瘤的首个迹象。” 标签：牙龈肿瘤。
让我们将这些内容发送到LLM，看看结果是否有所改善：
# 过滤响应
filtered_articles = filter_articles_by_uri(response, matching_articles)
# 将标题和摘要组合成一段文本的函数
def combine_abstracts(filtered_articles):
combined_text = "\n\n".join(
[f"标题: {article['title']}\n摘要: {article['abstractText']}" for article in filtered_articles]
)
return combined_text
# 将过滤后的文章的摘要组合在一起
combined_text = combine_abstracts(filtered_articles)
# 生成并打印摘要
summary = generate_summary(combined_text)
print(summary)以下是结果：
- 口腔癌很常见，通常在晚期才被发现
- 一项可行性研究在高风险地区进行，旨在提高口腔癌和癌前病变的早期检测
- 烟草销售商参与分发传单，鼓励吸烟者接受全科医生的免费检查
- 研究中共有93名患者参与，其中27%的患者被转介给专科医生
- 被转介的患者中有63.6%实际见到了专科医生，其中15.3%被诊断出患有癌前病变
- 光动力疗法（PDT）在实验中被用于治疗由化学诱导的癌前病变和腭粘膜鳞状细胞癌的实验性癌症疗法，研究对象为大鼠
- PDT使用了Photofrin和两种不同的激活波长，514.5 nm组的效果更好
- 来自恶性间皮瘤的牙龈转移极为罕见，生存率低
- 一例研究显示，一名患者牙龈肿块是多器官恶性间皮瘤复发的首个迹象，强调了对所有新病变进行活检的重要性，即使在不常见的解剖部位也是如此。我们确实可以看到改进——这些结果不是关于直肠癌或鼻咽部肿瘤。这看起来是对所选三篇文章的一个相对准确的总结，这些文章是关于口腔肿瘤的治疗。
结论总体而言，向量数据库非常适合快速实现搜索、相似度（推荐）和RAG（检索增强生成）应用。所需准备工作很少。如果你的数据集中包含结构化数据和非结构化数据，例如在这个期刊文章的例子中，向量数据库可以很好地工作。如果没有文章摘要作为数据集的一部分，例如，这种方法的效果将大打折扣。
知识图谱（KGs）在准确性与控制方面表现出色。如果你希望确保输入到搜索应用程序中的数据是“正确”的，而这里的“正确”是指根据你的需求所定义的标准，那么你将需要一个知识图谱。知识图谱在搜索和相似性分析方面表现良好，但它们能满足你需求的程度将取决于你的元数据的丰富程度以及标签的质量。标签的质量也可能因应用场景的不同而有所不同——如果你正在构建的是推荐引擎而非搜索引擎，那么你构建和应用分类法的方式可能会有所不同。
使用知识图谱（KG）来过滤向量数据库中的结果可以得到最佳效果。这并不令人惊讶——我使用知识图谱来过滤掉我认为不相关或有误导性的结果，因此当然结果会更好，根据我的判断。但重点在于：知识图谱本身不一定能直接改善结果，而是它为你提供了控制输出以优化结果的能力。