知识图谱（KGs）和大语言模型（LLMs）简直是天作之合。我在之前的文章中更详细地讨论了这两种技术的互补性，但简而言之，“LLMs 的主要弱点之一在于它们是黑箱模型（即无法解释的模型），并且在处理事实性知识时存在困难，而这些恰恰是 KGs 的最大优势。知识图谱本质上是事实的集合，并且完全可解释。”
本文将介绍如何构建一个简单的GraphRAG 应用程序。什么是 RAG？RAG，即检索增强生成（Retrieval-Augmented Generation），是指通过检索相关信息来增强发送到 LLM 的提示词（prompt），从而生成响应。而GraphRAG 是指在检索部分中使用知识图谱的 RAG。如果你从未听说过GraphRAG，或者想要回顾一下，我建议你观看这个视频。
基本思路是，与其直接将提示词发送给未针对你的数据进行训练的 LLM，不如在提示词中补充相关信息，使 LLM 能够更准确地回答你的问题。我经常用的例子是，将一份职位描述和我的简历复制到 ChatGPT 中，让它帮我写一封求职信。如果我提供了我的简历和我要申请职位的描述，LLM 就能对我的提示词“帮我写一封求职信”给出更相关的回答。由于知识图谱是为了存储知识而设计的，因此它们是存储内部数据并为 LLM 提供额外上下文以提高响应准确性和语境理解的完美工具。
这种技术有许多应用场景，例如客户服务机器人、药物研发、生命科学领域的自动化法规报告生成、HR 的人才招聘和管理、法律研究与写作以及财富顾问助手。由于其广泛的适用性以及提升 LLM 工具性能的潜力，GraphRAG（本文将使用该术语）近年来人气飙升。以下是基于 Google 搜索的兴趣随时间变化的图表。
GraphRAG 的搜索兴趣激增，甚至超过了“知识图谱”和“检索增强生成”等术语。2024 年 7 月，微软宣布其 GraphRAG 应用程序将在 GitHub 上可用，这一消息与 Graph RAG 搜索兴趣的激增时间一致。
然而，围绕GraphRAG 的热潮并不仅限于微软。2024 年 7 月，三星收购了一家知识图谱公司 RDFox。虽然宣布收购的文章中没有明确提到GraphRAG，但在 2024 年 11 月 Forbes 发布的一篇文章中，三星的一位发言人表示：“我们计划开发知识图谱技术，这是一种个性化 AI 的主要技术，并与生成式 AI 有机结合，以支持用户特定的服务。”
2024 年 10 月，领先的图谱数据库公司 Ontotext 和语义网公司 PoolParty（知识图谱管理平台的开发者）合并成立了 Graphwise。根据新闻稿，这次合并的目标是“推动GraphRAG 作为一个类别的普及。”
虽然围绕GraphRAG 的部分热度可能源于围绕聊天机器人和生成式 AI 的更广泛兴奋，但它确实反映了知识图谱在解决复杂现实问题方面的应用发生了真正的演变。一个例子是 LinkedIn 应用GraphRAG 改进了其客户服务技术支持。由于该工具能够检索相关数据（例如之前解决过的类似问题或问题）并将其提供给 LLM，响应更加准确，平均解决时间从 40 小时降至 15 小时。
这篇文章将通过一个相当简单但我认为具有启发性的例子来展示GraphRAG 如何在实践中工作。最终结果是一个非技术用户可以交互的应用程序。与我上一篇文章类似，我将使用 PubMed 的医学期刊文章数据集。这个想法是，这是一个医疗领域的人可以用来进行文献综述的应用程序。然而，这些原则可以应用于许多用例，这也是GraphRAG 令人兴奋的原因。
该应用程序的结构以及本文的结构如下：
第零步是准备数据。我将在下面解释细节，但总体目标是将原始数据向量化，并将其单独转换为 RDF 图。只要我们在向量化之前将 URI 与文章关联起来，我们就可以在文章的图谱和向量空间之间导航。然后，我们可以：
1. 搜索文章： 使用向量数据库的强大功能，根据搜索词对相关文章进行初步搜索。我将使用向量相似性来检索与搜索词向量最相似的文章。
2. 优化术语： 探索医学主题词表（MeSH）生物医学词汇，以选择用于过滤第 1 步文章的术语。这种受控词汇包含医学术语、替代名称、更狭义的概念以及许多其他属性和关系。
3. 过滤与总结： 使用 MeSH 术语过滤文章以避免“上下文污染”（即由于不相关信息导致 LLM 的响应不准确）。然后将剩余的文章连同附加提示词（例如“用项目符号总结”）发送到 LLM。
在开始之前，关于此应用程序和教程的一些说明：
• 此设置仅将知识图谱用于元数据。这之所以可行，是因为我的数据集中每篇文章已经使用了一个丰富的受控词汇中的术语进行了标记。我使用图谱来提供结构和语义，使用向量数据库进行基于相似性的检索，确保每种技术都用于其最擅长的领域。向量相似性可以告诉我们“食道癌”在语义上与“口腔癌”相似，但知识图谱可以告诉我们“食道癌”和“口腔癌”之间关系的细节。
• 我为此应用程序使用的数据是来自 PubMed 的 50,000 篇医学期刊文章的集合（许可证 CC0：公共领域）。该数据集包含文章标题、摘要以及一个用于元数据标签的字段。这些标签来自医学主题词表（MeSH）受控词汇表。由于这些是医学文章，我将此应用程序称为“医学GraphRAG”。但这种结构可以应用于任何领域，并不特定于医学领域。
• 我希望本教程和应用程序能证明，通过在检索步骤中引入知识图谱，可以提高 RAG 应用程序的准确性和可解释性。我将展示知识图谱如何通过两种方式提高 RAG 应用程序的准确性：为用户提供过滤上下文的方法，确保 LLM 只接收最相关的信息；以及使用由领域专家维护和管理的密集关系的领域特定受控词汇进行过滤。
• 本教程和应用程序未直接展示的另一个重要方面是知识图谱如何通过治理、访问控制和法规遵从性以及效率和可扩展性增强 RAG 应用程序。在治理方面，知识图谱不仅可以过滤内容以提高准确性，还可以执行数据治理政策。例如，如果用户没有权限访问某些内容，则可以将该内容从其 RAG 管道中排除。在效率和可扩展性方面，知识图谱可以帮助确保 RAG 应用程序不会被搁置。尽管创建一个令人印象深刻的一次性 RAG 应用程序很容易（这正是本教程的目的），但许多公司在管理缺乏统一框架、结构或平台的孤立概念验证（POC）时遇到了困难。这意味着许多此类应用程序可能无法长期存活。通过知识图谱驱动的元数据层可以打破数据孤岛，为有效构建、扩展和维护 RAG 应用程序提供基础。使用像 MeSH 这样丰富的受控词汇对这些文章的元数据标签进行标记，是确保此GraphRAG 应用程序可以与其他系统集成并降低其成为孤岛风险的一种方式。
第 0 步：准备数据
准备数据的代码在这个笔记本中：https://github.com/SteveHedden/kg_llm/blob/main/graphRAGapp/VectorVsKG_updated.ipynb。
如前所述，我再次决定使用来自 PubMed 仓库的数据集中的 50,000 篇研究文章（许可证 CC0：公共领域）。该数据集包含文章的标题、摘要以及一个用于元数据标签的字段。这些标签来自医学主题词表（MeSH）受控词汇表。PubMed 文章实际上只是文章的元数据——每篇文章都有摘要，但我们没有完整的文本。数据已经是表格格式，并使用 MeSH 术语进行了标记。
我们可以直接向量化这个表格数据集。我们可以在向量化之前将其转换为图（RDF），但我没有为此应用程序这样做，我认为对于这种数据类型来说，这样做可能对最终结果帮助不大。向量化原始数据最重要的是，我们首先为每篇文章添加唯一资源标识符（URI）。URI 是导航 RDF 数据所需的唯一标识符，它是我们在向量和图谱中的实体之间来回切换的必要条件。此外，我们将在向量数据库中为 MeSH 术语创建一个单独的集合。这将允许用户在没有该受控词汇的先验知识的情况下搜索相关术语。以下是我们准备数据所做工作的示意图。
我们在向量数据库中有两个可查询的集合：文章和术语。我们还以 RDF 格式将数据表示为图谱。由于 MeSH 提供了一个 API，我将直接查询该 API 以获取术语的替代名称和更狭义的概念。
在 Weaviate 中向量化数据
首先导入所需的包并设置 Weaviate 客户端：
import weaviate
from weaviate.util import generate_uuid5
from weaviate.classes.init import Auth
import os
import json
import pandas as pd
client = weaviate.connect_to_weaviate_cloud(
cluster_url="XXX", # 替换为你的 Weaviate 云 URL
auth_credentials=Auth.api_key("XXX"), # 替换为你的 Weaviate 云密钥
headers={'X-OpenAI-Api-key': "XXX"} # 替换为你的 OpenAI API 密钥
)读取 PubMed 期刊文章数据。我在 Databricks 中运行此笔记本，因此你可能需要根据运行环境进行更改。目标是将数据加载到 pandas DataFrame 中。
df = spark.sql("SELECT * FROM workspace.default.pub_med_multi_label_text_classification_dataset_processed").toPandas()如果你在本地运行，只需执行以下代码：
df = pd.read_csv("PubMed Multi Label Text Classification Dataset Processed.csv")然后稍微清理一下数据：
import numpy as np
# 将无穷值替换为 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)现在我们需要为每篇文章创建一个 URI，并将其作为新列添加。这很重要，因为 URI 是我们可以将文章的向量表示与其知识图谱表示连接起来的方式。
import urllib.parse
from rdflib import Graph, RDF, RDFS, Namespace, URIRef, Literal
# 创建有效 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 的函数
def create_article_uri(title, base_namespace="http://example.org/article/"):
""" 创建文章的 URI，通过用下划线替换非单词字符并进行 URL 编码。
Args:
title (str): 文章标题。
base_namespace (str): 文章 URI 的基础命名空间。
Returns:
URIRef: 格式化的文章 URI。
"""
if pd.isna(title):
return None
# 用下划线替换非单词字符
sanitized_title = re.sub(r'\W+', '_', title.strip())
# 将多个下划线压缩为一个
sanitized_title = re.sub(r'_+', '_', sanitized_title)
# 对术语进行 URL 编码
encoded_title = quote(sanitized_title)
# 拼接基础命名空间，不添加下划线
uri = f"{base_namespace}{encoded_title}"
return URIRef(uri)
# 为 DataFrame 添加新列以存储文章的 URI
df['Article_URI'] = df['Title'].apply(lambda title: create_valid_uri("http://example.org/article", title))我们还希望创建一个包含所有用于标记文章的 MeSH 术语的 DataFrame。这将在稍后我们需要搜索类似的 MeSH 术语时派上用场。
# 清理和解析 MeSH 术语的函数
def parse_mesh_terms(mesh_list):
if pd.isna(mesh_list):
return []
return [
term.strip().replace(' ', '_')
for term in mesh_list.strip("[]'").split(',')
]
# 为 MeSH 术语创建有效 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 f"{base_uri}/{sanitized_text}"
# 提取并处理所有 MeSH 术语
all_mesh_terms = []
for mesh_list in df["meshMajor"]:
all_mesh_terms.extend(parse_mesh_terms(mesh_list))
# 去重术语
unique_mesh_terms = list(set(all_mesh_terms))
# 创建包含 MeSH 术语及其 URI 的 DataFrame
mesh_df = pd.DataFrame({
"meshTerm": unique_mesh_terms,
"URI": [create_valid_uri("http://example.org/mesh", term) for term in unique_mesh_terms]
})
# 显示 DataFrame
print(mesh_df)向量化文章的 DataFrame：
from weaviate.classes.config import Configure
# 定义集合
articles = client.collections.create(
name = "Article",
vectorizer_config=Configure.Vectorizer.text2vec_openai(), # 如果设置为“none”，则必须始终自行提供向量。也可以是其他 "text2vec-*"。
generative_config=Configure.Generative.openai(), # 确保使用 `generative-openai` 模块进行生成查询
)
# 添加对象
articles = client.collections.get("Article")
with articles.batch.dynamic() as batch:
for index, row in df.iterrows():
batch.add_object({
"title": row["Title"],
"abstractText": row["abstractText"],
"Article_URI": row["Article_URI"],
"meshMajor": row["meshMajor"],
})现在向量化 MeSH 术语：
# 定义集合
terms = client.collections.create(
name = "term",
vectorizer_config=Configure.Vectorizer.text2vec_openai(), # 如果设置为“none”，则必须始终自行提供向量。也可以是其他 "text2vec-*"。
generative_config=Configure.Generative.openai(), # 确保使用 `generative-openai` 模块进行生成查询
)
# 添加对象
terms = client.collections.get("term")
with terms.batch.dynamic() as batch:
for index, row in mesh_df.iterrows():
batch.add_object({
"meshTerm": row["meshTerm"],
"URI": row["URI"],
})此时，你可以直接针对向量化数据集运行语义搜索、相似性搜索和 RAG。我不会在这里详细介绍所有内容，但你可以查看我随附笔记本中的代码来实现这些功能。
将数据转换为知识图谱
我使用了上一篇文章中相同的代码来完成这一步。基本上，我们将数据中的每一行转换为知识图谱（KG）中的一个“文章（Article）”实体。然后，我们为每篇文章添加标题、摘要和 MeSH 术语等属性。同时，我们还将每个 MeSH 术语也转换为一个实体。此代码还为每篇文章添加了一个随机日期作为“发布日期（date published）”属性，以及一个 1 到 10 之间的随机数作为“访问权限（access）”属性。在本次演示中，我们不会使用这些属性。以下是从数据创建的图谱的可视化表示：
以下是如何遍历 DataFrame 并将其转换为 RDF 数据的代码：
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
import re
from urllib.parse import quote
# --- 初始化 ---
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() for term in mesh_list.strip("[]'").split(',')]
# 增强的 URI 转换函数
def convert_to_uri(term, base_namespace="http://example.org/mesh/"):
if pd.isna(term):
return None
stripped_term = re.sub(r'^\W+|\W+$', '', term)
formatted_term = re.sub(r'\W+', '_', stripped_term)
formatted_term = re.sub(r'_+', '_', formatted_term)
encoded_term = quote(formatted_term)
term_with_underscores = f"_{encoded_term}_"
uri = f"{base_namespace}{term_with_underscores}"
return URIRef(uri)
# 生成过去 5 年内随机日期的函数
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)
# 为文章创建有效 URI 的函数
def create_article_uri(title, base_namespace="http://example.org/article"):
if pd.isna(title):
return None
sanitized_text = urllib.parse.quote(title.strip().replace(' ', '_').replace('"', '').replace('<', '').replace('>', '').replace("'", "_"))
return URIRef(f"{base_namespace}/{sanitized_text}")
# 遍历 DataFrame 中的每一行并创建 RDF 三元组
for index, row in df.iterrows():
article_uri = create_article_uri(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 = convert_to_uri(term, base_namespace="http://example.org/mesh/")
if term_uri is None:
continue
g.add((term_uri, RDF.type, MeSHTerm))
g.add((term_uri, RDFS.label, Literal(term.replace('_', ' '), datatype=XSD.string)))
g.add((article_uri, schema.about, term_uri))
# 保存为文件
file_path = "/Workspace/PubMedGraph.ttl"
g.serialize(destination=file_path, format='turtle')
print(f"文件已保存至 {file_path}")现在，我们已经完成了数据的向量化版本和图谱（RDF）版本。每个向量都关联了一个 URI，该 URI 对应于知识图谱中的一个实体，因此我们可以在不同的数据格式之间自由切换。
构建应用程序
我决定使用 Streamlit 来构建这个GraphRAG 应用程序的界面。与上一篇文章类似，我保持了用户流程不变：
1. 搜索文章： 用户首先通过搜索词搜索文章。这完全依赖于向量数据库。用户的搜索词被发送到向量数据库，并返回与搜索词在向量空间中最接近的 10 篇文章。
2. 优化术语： 用户选择用于过滤返回结果的 MeSH 术语。由于我们也向量化了 MeSH 术语，用户可以输入自然语言提示来获取最相关的 MeSH 术语。然后，用户可以扩展这些术语以查看它们的替代名称和更狭义的概念。用户可以选择任意数量的术语作为过滤条件。
3. 过滤与总结： 用户将选定的术语应用为过滤器以筛选最初返回的 10 篇期刊文章。由于 PubMed 文章已用 MeSH 术语标记，我们可以轻松完成这一步。最后，用户可以输入一个附加提示词，将其与筛选后的期刊文章一起发送到 LLM。这是 RAG 应用程序的生成步骤。
搜索文章
首先，我们需要实现 Weaviate 的向量相似性搜索功能。由于我们的文章已被向量化，我们可以将搜索词发送到向量数据库并返回相似的文章。
在 app.py 中搜索相关期刊文章的主要函数如下：
# --- TAB 1: 搜索文章 ---
with tab_search:
st.header("搜索文章（向量查询）")
query_text = st.text_input("输入你的向量搜索词（例如：口腔肿瘤）：", key="vector_search")
if st.button("搜索文章", key="search_articles_btn"):
try:
client = initialize_weaviate_client()
article_results = query_weaviate_articles(client, query_text)
# 提取 URI
article_uris = [
result["properties"].get("article_URI")
for result in article_results
if result["properties"].get("article_URI")
]
# 将 article_uris 存储到会话状态
st.session_state.article_uris = article_uris
st.session_state.article_results = [
{
"Title": result["properties"].get("title", "N/A"),
"Abstract": (result["properties"].get("abstractText", "N/A")[:100] + "..."),
"Distance": result["distance"],
"MeSH Terms": ", ".join(
ast.literal_eval(result["properties"].get("meshMajor", "[]"))
if result["properties"].get("meshMajor")
else []
),
}
for result in article_results
]
client.close()
except Exception as e:
st.error(f"搜索文章时出错：{e}")
if st.session_state.article_results:
st.write("**文章搜索结果：**")
st.table(st.session_state.article_results)
else:
st.write("尚未找到文章。")此函数使用存储在 weaviate_queries 文件夹中的查询来初始化 Weaviate 客户端（initialize_weaviate_client）并搜索文章（query_weaviate_articles）。然后，我们将返回的文章以表格形式显示，包括标题、摘要、距离（与搜索词的相似度）、以及文章标记的 MeSH 术语。
在 weaviate_queries.py 中查询 Weaviate 的函数如下：
# 查询 Weaviate 文章的函数
def query_weaviate_articles(client, query_text, limit=10):
# 在 Article 集合上执行向量搜索
response = client.collections.get("Article").query.near_text(
query=query_text,
limit=limit,
return_metadata=MetadataQuery(distance=True)
)
# 解析响应结果
results = []
for obj in response.objects:
results.append({
"uuid": obj.uuid,
"properties": obj.properties,
"distance": obj.metadata.distance,
})
return results如你所见，我将返回结果限制为 10 条以简化操作，但你可以更改此限制。此函数仅使用 Weaviate 的向量相似性搜索来返回相关结果。
应用程序中的最终结果如下所示：
作为演示，我搜索了“治疗口腔癌的方法”。如图所示，返回了 10 篇文章，大多数都与搜索词相关。这展示了基于向量检索的优缺点。
优点在于，我们可以通过最小的努力在数据上构建语义搜索功能。如上所示，我们只需设置客户端并将数据发送到向量数据库。一旦数据被向量化，我们就可以执行语义搜索、相似性搜索，甚至是 RAG。我在本文附带的笔记本中包含了一些相关内容，但更多内容可以参考 Weaviate 的官方文档。
缺点在于，基于向量的检索是“黑箱”的，并且在处理事实性知识时存在困难。在我们的示例中，大多数文章确实与某种癌症治疗或疗法有关。有些文章是关于口腔癌的，有些则是关于口腔癌的子类型，例如牙龈癌（gingival cancer）或腭癌（palatal cancer）。但也有一些文章是关于鼻咽癌（nasopharyngeal cancer）、下颌癌（mandibular cancer）和食管癌（esophageal cancer）。这些（鼻咽、下颌或食管）都不被视为口腔癌。可以理解的是，一篇关于鼻咽肿瘤放射治疗的文章可能会被误认为与“治疗口腔癌的方法”相关，但如果你只想寻找治疗口腔癌的方法，可能会发现它并不相关。如果我们直接将这 10 篇文章插入提示词并让 LLM“总结不同的治疗方法”，我们可能会得到错误的信息。
RAG 的目的是为 LLM 提供一组非常具体的附加信息，以更好地回答你的问题——如果这些信息不正确或无关，就可能导致 LLM 的误导性回答。这通常被称为“上下文污染（context poisoning）”。上下文污染的特别危险之处在于，LLM 的回答不一定是事实上的错误（LLM 可能准确地总结了我们提供的治疗方法），也不一定基于不准确的数据（假设期刊文章本身是准确的），而是使用了错误的数据来回答你的问题。在这个例子中，用户可能会阅读到如何治疗错误类型癌症的信息，这显然是非常糟糕的。
优化术语
知识图谱（KGs）可以通过优化向量数据库的结果来提高响应的准确性并减少上下文污染（context poisoning，即由于不相关信息导致 LLM 生成误导性回答）的可能性。下一步是选择要用来过滤文章的 MeSH 术语。首先，我们对向量数据库中的“术语（Terms）”集合执行另一轮向量相似性搜索。这是因为用户可能对 MeSH 受控词汇不熟悉。在上面的示例中，我搜索了“therapies for mouth cancer”（口腔癌的治疗方法），但“mouth cancer”（口腔癌）并不是 MeSH 中的术语——它们使用的是“mouth neoplasms”（口腔肿瘤）。我们希望用户能够在没有先验知识的情况下开始探索 MeSH 术语——无论使用何种元数据标记内容，这都是一种良好的实践。
获取相关 MeSH 术语的函数与之前的 Weaviate 查询几乎相同，只需将“Article”替换为“term”即可：
# 查询 Weaviate 中 MeSH 术语的函数
def query_weaviate_terms(client, query_text, limit=10):
# 在 MeshTerm 集合上执行向量搜索
response = client.collections.get("term").query.near_text(
query=query_text,
limit=limit,
return_metadata=MetadataQuery(distance=True)
)
# 解析响应
results = []
for obj in response.objects:
results.append({
"uuid": obj.uuid,
"properties": obj.properties,
"distance": obj.metadata.distance,
})
return results在应用程序中的效果如下：
如图所示，我搜索了“mouth cancer”（口腔癌），并返回了最相似的术语。“mouth cancer” 并未被返回，因为这不是 MeSH 中的术语，但“mouth neoplasms”（口腔肿瘤）在列表中。
下一步是允许用户展开返回的术语，以查看替代名称和更狭义的概念（narrower concepts，即更具体的子级术语）。这需要查询 MeSH API。这是构建此应用程序时最棘手的部分之一，主要原因如下：Streamlit 要求每个元素都有唯一的 ID，但 MeSH 术语可能会重复——如果返回的某个概念是另一个概念的子节点，那么当你展开父节点时，就会出现子节点的重复。我认为我已经解决了大部分问题，应用程序应该可以正常运行，但在此阶段可能仍然会有一些错误。
以下是我们依赖的函数，这些函数存储在 rdf_queries.py 文件中。我们需要一个函数来获取术语的替代名称：
# 获取 MeSH 术语的替代名称和三元组
def get_concept_triples_for_term(term):
term = sanitize_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 . }}
}}
"""
try:
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(sanitize_term(obj_label)) # 在添加前清理术语
triples.add(sanitize_term(term)) # 确保包含清理后的术语本身
return list(triples)
except Exception as e:
print(f"Error fetching concept triples for term '{term}': {e}")
return []我们还需要获取某个术语的更狭义（子级）概念的函数。我实现了两个函数，一个用于获取术语的直接子节点，另一个是递归函数，用于返回给定深度的所有子节点。
# 获取 MeSH 术语的更狭义概念
def get_narrower_concepts_for_term(term):
term = sanitize_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 .
}}
"""
try:
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(sanitize_term(subject_label)) # 在添加前清理术语
return list(concepts)
except Exception as e:
print(f"Error fetching narrower concepts for term '{term}': {e}")
return []
# 递归函数：获取指定深度的所有更狭义概念
def get_all_narrower_concepts(term, depth=2, current_depth=1):
term = sanitize_term(term) # 清理输入术语
all_concepts = {}
try:
narrower_concepts = get_narrower_concepts_for_term(term)
all_concepts[sanitize_term(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)
except Exception as e:
print(f"Error fetching all narrower concepts for term '{term}': {e}")
return all_concepts在步骤 2 的另一个重要部分是允许用户选择术语并将其添加到“已选术语（Selected Terms）”列表中，这些术语会显示在屏幕左侧的侧边栏中。以下是一些可以改进此步骤的建议：
• 目前没有“清除所有”的功能，但可以通过清除缓存或刷新浏览器来解决。
• 没有“选择所有更狭义概念”的选项，这将非常有用。
• 没有为过滤添加规则的选项。目前，我们假设文章必须包含术语 A 或术语 B 或术语 C 等。最终的排名基于文章标记的术语数量。
以下是应用程序中的展示效果：
我可以展开“mouth neoplasms”（口腔肿瘤）以查看替代名称，例如“cancer of mouth”（口腔癌），以及所有更狭义的概念。如图所示，大多数更狭义的概念都有自己的子级，你也可以展开这些子级。在本次演示中，我选择了“mouth neoplasms”的所有子级。
此步骤不仅重要，因为它允许用户过滤搜索结果，还因为它为用户提供了探索 MeSH 图谱本身并从中学习的机会。例如，用户可以在这里了解到“nasopharyngeal neoplasms”（鼻咽肿瘤）并不是“mouth neoplasms”（口腔肿瘤）的子集。
过滤与总结
现在，我们已经有了文章和过滤术语，可以应用过滤器并总结结果。在这一步中，我们将第一步返回的原始 10 篇文章与优化后的 MeSH 术语列表结合起来。我们允许用户在将数据发送到 LLM 之前添加额外的上下文到提示词中。
我们通过以下方式实现过滤：首先获取第一步返回的 10 篇文章的 URI，然后查询知识图谱以确定哪些文章被标记为相关的 MeSH 术语。此外，我们保存这些文章的摘要，以便在下一步中使用。在这一阶段，我们还可以基于访问控制或其他用户控制的参数（例如作者、文件类型、发布日期等）进行过滤。尽管我没有在此应用程序中包含这些功能，但我在数据中添加了访问控制和发布日期属性，以备将来在 UI 中使用。
以下是 app.py 中的代码：
if st.button("Filter Articles"):
try:
# 检查是否有来自步骤 1 的 URI
if "article_uris" in st.session_state and st.session_state.article_uris:
article_uris = st.session_state.article_uris
# 将 URI 列表转换为字符串以用于 VALUES 子句或 FILTER
article_uris_string = ", ".join([f"<{str(uri)}>" for uri in article_uris])
SPARQL_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 .
FILTER (?article IN ({article_uris}))
}}
"""
# 将文章 URI 插入查询
query = SPARQL_QUERY.format(article_uris=article_uris_string)
else:
st.write("第 1 步中未选择任何文章。")
st.stop()
# 查询 RDF 并将结果保存到会话状态
top_articles = query_rdf(LOCAL_FILE_PATH, query, final_terms)
st.session_state.filtered_articles = top_articles
if top_articles:
# 合并顶级文章的摘要并保存到会话状态
def combine_abstracts(ranked_articles):
combined_text = " ".join(
[f"Title: {data['title']} Abstract: {data['abstract']}" for article_uri, data in ranked_articles]
)
return combined_text
st.session_state.combined_text = combine_abstracts(top_articles)
else:
st.write("未找到符合所选术语的文章。")
except Exception as e:
st.error(f"过滤文章时出错：{e}")此代码使用 rdf_queries.py 文件中的 query_rdf 函数。该函数如下：
# 使用 SPARQL 查询 RDF 的函数
def query_rdf(local_file_path, query, mesh_terms, base_namespace="http://example.org/mesh/"):
if not mesh_terms:
raise ValueError("MeSH 术语列表为空或无效。")
print("SPARQL Query:", query)
# 创建并解析 RDF 图
g = Graph()
g.parse(local_file_path, format="ttl")
article_data = {}
for term in mesh_terms:
# 将术语转换为有效的 URI
mesh_term_uri = convert_to_uri(term, base_namespace)
# 执行 SPARQL 查询并绑定初始变量
results = g.query(query, initBindings={'meshTerm': mesh_term_uri})
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()
}
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
)
return ranked_articles[:10]此函数还将 MeSH 术语转换为 URI，以便我们可以使用图谱进行过滤。在转换术语为 URI 时，请确保转换方式与其他函数保持一致。
以下是应用程序中的效果：
如图所示，我们在上一步中选择的两个 MeSH 术语显示在这里。如果我点击“Filter Articles”（过滤文章），它将使用步骤 2 中的过滤条件对原始 10 篇文章进行过滤。返回的文章将包括它们的完整摘要以及标记的 MeSH 术语（见下图）。
共返回了 5 篇文章。其中两篇标记为“mouth neoplasms”（口腔肿瘤），一篇标记为“gingival neoplasms”（牙龈肿瘤），两篇标记为“palatal neoplasms”（腭肿瘤）。
现在我们已经有了优化后的文章列表，可以用于生成响应。我们将这些文章发送到 LLM 来生成响应，同时可以在提示词中添加额外的上下文。我设置了一个默认提示词：“Summarize the key information here in bullet points. Make it understandable to someone without a medical degree.”（以项目符号总结这里的关键信息，使其易于非医学背景的人理解）。在本次演示中，我将提示词调整为反映我们的原始搜索词：
以下是生成的结果：
结果看起来更好，主要是因为我知道我们总结的文章大多是关于口腔癌治疗方法的。数据集不包含实际的期刊文章，仅包含摘要。因此，这些结果只是摘要的摘要。尽管如此，如果我们要构建一个真实的应用程序而不仅仅是演示，这是我们可以引入文章全文的步骤。或者，这也是用户/研究人员自己阅读这些文章而不是完全依赖 LLM 生成摘要的地方。
总结
本教程展示了如何结合向量数据库和知识图谱来显著增强 RAG 应用程序。通过利用向量相似性进行初始搜索，并使用结构化的知识图谱元数据进行过滤和组织，我们可以构建一个能够提供准确、可解释且特定领域结果的系统。引入 MeSH 这样的成熟受控词汇，突显了领域专业知识在元数据管理中的力量，确保检索步骤与应用程序的独特需求保持一致，同时保持与其他系统的互操作性。这种方法不仅限于医学领域——其原理可以应用于任何结构化数据与文本信息共存的领域。
本教程强调了充分利用每种技术的优势：向量数据库在基于相似性的检索中表现出色，而知识图谱在提供上下文、结构和语义方面具有优势。此外，扩展 RAG 应用程序需要一个元数据层来打破数据孤岛并执行治理策略。以领域特定元数据和稳健治理为基础的精心设计，是构建既准确又可扩展的 RAG 系统的关键。
