构建图形RAG应用：利用知识图谱和AI增强医学期刊洞察-CSDN博客

本文链接：https://blog.csdn.net/python12345_/article/details/146884716

什么是闭包？

闭包是 JavaScript 中一个非常重要的概念，它允许函数访问并操作其词法作用域之外的变量。换句话说，闭包是一个函数以及其周围状态（词法环境）的捆绑。

词法作用域

在理解闭包之前，我们需要先了解词法作用域。词法作用域是指在代码编写时就确定的作用域，它由代码在源代码中的位置决定。例如：

function outerFunction() {
let outerVariable = 'Hello';

functioninnerFunction() {
    console.log(outerVariable); // 访问 outerVariable
  }

innerFunction();
}

outerFunction(); // 输出 "Hello"

在这个例子中，innerFunction 可以访问 outerFunction 中定义的 outerVariable，因为 innerFunction 在 outerFunction 的词法作用域内。

闭包的特性

闭包具有以下几个关键特性：

• 访问外部变量： 闭包可以访问其定义时所在的词法作用域中的变量。
• 数据隐藏： 闭包可以用于创建私有变量，这些变量只能通过闭包内部的函数访问。
• 状态保持： 闭包可以记住其定义时所在的作用域，即使该作用域已经执行完毕。

闭包的例子

让我们通过一些例子来更好地理解闭包。

function createCounter() {
let count = 0;

returnfunction() {
    count++;
    console.log(count);
  }
}

const counter = createCounter();
counter(); // 输出 1
counter(); // 输出 2
counter(); // 输出 3

在这个例子中，createCounter 函数返回了一个匿名函数。这个匿名函数可以访问 createCounter 函数中的 count 变量。每次调用 counter() 时，count 都会递增，并且 counter 函数会记住 count 的当前值。

另一个例子

function outer() {
let name = 'Outer';

functioninner() {
    console.log(name);
  }

return inner;
}

let myFunc = outer();
myFunc(); // 输出 "Outer"

在这个例子中，inner 函数形成了一个闭包，它可以访问 outer 函数中的 name 变量。即使 outer 函数已经执行完毕，inner 函数仍然可以访问 name 变量。

闭包的应用场景

闭包在 JavaScript 中有很多应用场景，以下是其中几个常见的例子：

• 模块化： 闭包可以用于创建模块，将数据和方法封装起来，实现数据的隐藏和保护。
• 事件处理程序： 闭包可以用于在事件处理程序中访问外部变量。
• 回调函数： 闭包可以用于在回调函数中访问外部变量。
• 实现私有变量： 闭包可以用于创建私有变量，这些变量只能通过闭包内部的函数访问。

模块化

function createModule() {
let privateVariable = '私有变量';

functionpublicMethod() {
    console.log(privateVariable);
  }

return {
    publicMethod: publicMethod
  };
}

const myModule = createModule();
myModule.publicMethod(); // 输出 "私有变量"
// console.log(myModule.privateVariable); // 报错，无法访问私有变量

在这个例子中，privateVariable 是一个私有变量，只能通过 publicMethod 访问。

事件处理程序

function attachEventHandlers() {
  let button = document.getElementById('myButton');
  let message = 'Hello, World!';

  button.addEventListener('click', function() {
    alert(message);
  });
}

在这个例子中，事件处理程序可以访问 message 变量。

回调函数

function fetchData(url, callback) {
// 模拟异步请求
setTimeout(function() {
    let data = '数据';
    callback(data);
  }, 1000);
}

functionprocessData(data) {
console.log('处理后的数据：' + data);
}

fetchData('https://example.com/api', processData);

在这个例子中，callback 函数可以访问 fetchData 函数中的变量。

闭包的优缺点

优点

• 数据隐藏： 闭包可以隐藏变量，避免全局污染。
• 状态保持： 闭包可以保持变量的状态，实现一些特殊的功能。
• 模块化： 闭包可以用于实现模块化，提高代码的可维护性。

缺点

• 内存泄漏： 闭包可能会导致内存泄漏，如果闭包中引用的变量没有被释放，就会一直占用内存。
• 性能问题： 闭包可能会影响性能，因为闭包会占用额外的内存。

总结

闭包是 JavaScript 中一个非常重要的概念，它允许函数访问并操作其词法作用域之外的变量。闭包具有访问外部变量、数据隐藏和状态保持等特性，在模块化、事件处理程序和回调函数等场景中都有广泛的应用。在使用闭包时，需要注意内存泄漏和性能问题。

使用知识图谱和 AI 检索、筛选和总结医学期刊文章

该应用程序和笔记本的随附代码在此处。

知识图谱 (KG) 和大型语言模型 (LLM) 是天作之合。我之前的文章更详细地讨论了这两种技术的互补性，但简而言之就是，“LLM 的一些主要弱点，即它们是黑盒模型并且难以处理事实知识，恰恰是 KG 的最大优势。KG 本质上是事实的集合，并且是完全可解释的。”

本文主要介绍如何构建一个简单的 Graph RAG 应用程序。什么是 RAG？RAG，即检索增强生成，是关于检索相关信息以增强发送给 LLM 的提示，LLM 会生成响应。Graph RAG 是使用知识图谱作为检索部分一部分的 RAG。如果您从未听说过 Graph RAG，或者想复习一下，我建议观看此视频。

基本思路是，与其将您的提示直接发送给未在您的数据上训练的 LLM，不如用 LLM 回答您的提示所需的相关信息来补充您的提示。我经常使用的例子是将一份职位描述和我的简历复制到 ChatGPT 中来写求职信。如果我向它提供我的简历和我申请的职位描述，LLM 就能对我的提示“写一封求职信”提供更相关的回复。由于知识图谱旨在存储知识，因此它们是存储内部数据并用其他上下文信息补充 LLM 提示的完美方式，从而提高响应的准确性和上下文理解能力。

这项技术有许多应用，例如客户服务机器人、药物发现、生命科学领域的自动化监管报告生成、人力资源部门的人才招聘和管理、法律研究和写作以及财富顾问助手。由于其广泛的适用性和提高 LLM 工具性能的潜力，Graph RAG（我将在此处使用该术语）的受欢迎程度一直在飙升。这是一个根据 Google 搜索显示随时间推移的兴趣的图表。

null

来源：https://trends.google.com/

Graph RAG 的搜索兴趣激增，甚至超过了知识图谱和检索增强生成等术语。请注意，Google Trends 衡量的是相对搜索兴趣，而不是绝对搜索次数。2024 年 7 月 Graph RAG 搜索量激增恰逢微软宣布其 GraphRAG 应用程序将在 GitHub 上提供的那一周。

然而，围绕 Graph RAG 的兴奋不仅仅局限于微软。三星于 2024 年 7 月收购了知识图谱公司 RDFox。宣布该收购的这篇文章并没有明确提及 Graph RAG，但在 2024 年 11 月发表的福布斯这篇文章中，一位三星发言人表示：“我们计划开发知识图谱技术，这是个性化 AI 的主要技术之一，并与生成的 AI 有机地连接，以支持特定于用户的服务。”

2024 年 10 月，领先的图数据库公司 Ontotext 和语义 Web 公司 PoolParty（知识图谱策划平台）的制造商合并，成立了 Graphwise。根据新闻稿，此次合并旨在“将 Graph RAG 作为一个类别进行普及”。

虽然围绕 Graph RAG 的一些炒作可能来自于对聊天机器人和生成式 AI 的更广泛的兴奋，但它反映了知识图谱在解决复杂、现实世界问题方面的应用方式的真正演变。一个例子是 LinkedIn 应用 Graph RAG 来改进其客户服务技术支持。由于该工具能够检索相关数据（如先前解决的类似工单或问题）来馈送给 LLM，因此响应更准确，平均解决时间从 40 小时缩短到 15 小时。

这篇文章将介绍一个 Graph RAG 在实践中如何工作的相当简单的例子，但我认为它具有说明性。最终结果是一个非技术用户可以与之交互的应用程序。就像我上一篇文章一样，我将使用一个由来自 PubMed 的医学期刊文章组成的数据集。这个想法是，这是一个医学领域的人可以用来进行文献综述的应用程序。但是，相同的原则可以应用于许多用例，这就是 Graph RAG 如此令人兴奋的原因。

该应用程序的结构以及这篇文章如下：

第零步是准备数据。我将在下面解释详细信息，但总目标是将原始数据向量化，并将其单独转换为 RDF 图。只要我们在向量化之前将 URI 与文章相关联，我们就可以在文章图和文章向量空间中导航。然后，我们可以：

\1. 搜索文章： 使用向量数据库的功能，根据搜索词对相关文章进行初步搜索。我将使用向量相似度来检索与搜索词向量最相似的文章。
\2. 优化术语： 探索医学主题词 (MeSH) 生物医学词汇以选择用于筛选步骤 1 中的文章的术语。此受控词汇表包含医学术语、备用名称、更窄的概念，

MeSH 术语：深度解析

\1. **简介：**MeSH（医学主题词）是由美国国家医学图书馆（NLM）开发的一个综合性受控词汇表。它用于索引、编目和搜索生物医学和健康相关信息。MeSH 术语提供了一种标准化的方式来描述文章、书籍和其他资源的内容，从而更容易检索相关信息。
\2. MeSH 术语的主要特征：
- • 层级结构： MeSH 术语以层级结构组织，允许进行广泛和具体的搜索。
- • 受控词汇表： 使用受控词汇表确保索引和搜索的一致性。
- • 特异性： MeSH 术语提供高度的特异性，能够精确检索信息。
- • 以及许多其他属性和关系。
\3. 筛选和总结： 使用 MeSH 术语筛选文章，以避免“上下文污染”。然后将剩余的文章与附加提示一起发送给 LLM，例如“用要点概括”。

在我们开始之前，关于此应用和教程的一些说明：

• 这种设置专门使用知识图谱作为元数据。之所以可行，完全是因为我的数据集中每篇文章都已使用属于丰富受控词汇表的术语进行了标记。我使用图谱来表示结构和语义，使用向量数据库进行基于相似度的检索，确保每种技术都用于其最擅长的领域。向量相似度可以告诉我们“食道癌”在语义上与“口腔癌”相似，但知识图谱可以告诉我们“食道癌”和“口腔癌”之间关系的详细信息。
• 我用于此应用程序的数据是来自 PubMed 的医学期刊文章的集合（更多数据见下文）。我选择此数据集是因为它具有结构化（表格）的特点，但也包含每篇文章的摘要文本，并且因为它已经使用与公认的受控词汇表（MeSH）对齐的主题术语进行了标记。由于这些是医学文章，我将此应用程序命名为“医学知识图谱 RAG”。但是，这种相同的结构可以应用于任何领域，并且不特定于医学领域。
• 我希望本教程和应用程序演示的是，您可以通过在检索步骤中加入知识图谱来提高 RAG 应用程序在准确性和可解释性方面的结果。我将展示知识图谱如何通过两种方式提高 RAG 应用程序的准确性：通过为用户提供一种过滤上下文的方法，以确保 LLM 仅获得最相关的信息；以及通过使用由领域专家维护和策划的具有密集关系的特定于领域的受控词汇表来执行过滤。
• 本教程和应用程序没有直接展示知识图谱可以增强 RAG 应用程序的另外两种重要方式：治理、访问控制和法规遵从性；以及效率和可扩展性。对于治理，知识图谱可以做的不仅仅是过滤相关性以提高准确性——它们可以执行数据治理策略。例如，如果用户无权访问某些内容，则可以将其从 RAG 管道中排除。在效率和可扩展性方面，知识图谱可以帮助确保 RAG 应用程序不会束之高阁。虽然创建一个令人印象深刻的一次性 RAG 应用程序（这实际上是本教程的目的）很容易，但许多公司都在努力应对缺乏内聚框架、结构或平台的、大量互不关联的 POC。这意味着许多应用程序无法长期存在。由知识图谱提供支持的元数据层可以打破数据孤岛，为有效构建、扩展和维护 RAG 应用程序提供所需的基础。使用 MeSH 等丰富的受控词汇表作为这些文章的元数据标签，可以确保此知识图谱 RAG 应用程序可以与其他系统集成，并降低其成为孤岛的风险。

第 0 步：准备数据

用于准备数据的代码位于 this 笔记本中。

如前所述，我再次决定使用来自 PubMed 存储库的 this 包含 50,000 篇研究文章的数据集（许可证 CC0：公共领域）。此数据集包含文章的标题、摘要以及元数据标签的字段。这些标签来自医学主题词 (MeSH) 受控词汇表术语库。PubMed 文章实际上只是关于文章的元数据——每篇文章都有摘要，但我们没有全文。数据已采用表格格式，并使用 MeSH 术语进行了标记。

我们可以直接将此表格数据集向量化。我们可以在向量化之前将其转换为图谱 (RDF)，但对于此应用程序，我没有这样做，而且我不知道它是否会对这种数据的最终结果有所帮助。关于向量化原始数据，最重要的是我们首先为每篇文章添加统一资源标识符 (URI)。URI 是用于导航 RDF 数据的唯一 ID，对于我们在图谱中来回切换向量和实体是必要的。此外，我们将在向量数据库中为 MeSH 术语创建一个单独的集合。这将允许用户搜索相关术语，而无需事先了解此受控词汇表。下面是我们准备数据的示意图。

null

图片由作者提供

我们在向量数据库中有两个集合要查询：文章和术语。我们还将数据表示为 RDF 格式的图谱。由于 MeSH 具有 API，我将直接查询 API 以获取术语的备用名称和更窄的概念。

在 Weaviate 中向量化数据

首先导入所需的软件包并设置 Weaviate 客户端：

import weaviate
from weaviate.utilimport generate_uuid5
from weaviate.classes.initimportAuth
import os
import json
import pandas as pd

client = weaviate.connect_to_weaviate_cloud(
    cluster_url="XXX",  # 替换为您的 WeaviateCloudURL
    auth_credentials=Auth.api_key("XXX"),  # 替换为您的 WeaviateCloud 密钥
    headers={'X-OpenAI-Api-key': "XXX"}  # 替换为您的 OpenAIAPI 密钥
)

读入 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], 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/"):
    """
    通过将非单词字符替换为下划线并进行 URL 编码来为文章创建 URI。

    参数：
        title (str): 文章的标题。
        base_namespace (str): 文章 URI 的基本命名空间。

    返回：
        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)
    # 与 base_namespace 连接，不添加下划线
    uri = f"{base_namespace}{encoded_title}"
    return URIRef(uri)

为文章 URI 添加新列到 DataFrame

df['Article_URI'] = df['Title'].apply(lambda title: create_valid_uri("http://example.org/article", title))

我们还想创建一个包含用于标记文章的所有 MeSH 术语的 DataFrame。这将在我们稍后想要搜索相似的 MeSH 术语时有所帮助。

### 清理和解析 MeSH 术语的函数
defparse_mesh_terms(mesh_list):
    if pd.isna(mesh_list):
        return []
    return [
        term.strip().replace(' ', '_')
        for term in mesh_list.strip("[]'").split(',')
    ]

### 创建 MeSH 术语的有效 URI 的函数
defcreate_valid_uri(base_uri, text):
    if pd.isna(text):
        returnNone
    sanitized_text = urllib.parse.quote(
        text.strip()
        .replace(' ', '_')
        .replace('"', '')
        .replace('<', '')
        .replace('>', '')
        .replace("'", "_")
    )
    returnf"{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

#define the collection
articles = client.collections.create(
    name = "Article",
    vectorizer_config=Configure.Vectorizer.text2vec_openai(),  # If set to "none" you must always provide vectors yourself. Could be any other "text2vec-*" also.
    generative_config=Configure.Generative.openai(),  # Ensure the `generative-openai` module is used for generative queries
)

#add ojects
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 术语：

#define the collection
terms = client.collections.create(
    name = "term",
    vectorizer_config=Configure.Vectorizer.text2vec_openai(),  # If set to "none" you must always provide vectors yourself. Could be any other "text2vec-*" also.
    generative_config=Configure.Generative.openai(),  # Ensure the `generative-openai` module is used for generative queries
)

#add ojects
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 中的一个“文章”实体。然后，我们为每个文章提供标题、摘要和 MeSH 术语的属性。我们还将每个 MeSH 术语也转化为一个实体。此代码还为每篇文章添加了随机日期，用于名为 date published 的属性，以及 1 到 10 之间的随机数，用于名为 access 的属性。我们不会在此演示中使用这些属性。以下是我们将从数据中创建的图的直观表示。

null

作者提供的图片

以下是如何遍历 DataFrame 并将其转化为 RDF 数据：

from rdflib importGraph, RDF, RDFS, Namespace, URIRef, Literal
from rdflib.namespaceimportSKOS, XSD
import pandas as pd
import urllib.parse
import random
from datetime import datetime, timedelta
import re
from urllib.parseimport 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(',')]

增强的 convert_to_uri 函数

def convert_to_uri(term, base_namespace="http://example.org/mesh/"):
    """
    通过将空格和特殊字符替换为下划线，将 MeSH 术语转换为标准化的 URI，
    确保其以单个下划线开头和结尾，并对术语进行 URL 编码。

    参数：
        term (str): 要转换的 MeSH 术语。
        base_namespace (str): URI 的基本命名空间。

    返回：
        URIRef: 格式化的 URI。
    """
    if pd.isna(term):
        returnNone# 优雅地处理 NaN 或 None 术语

    # 步骤 1：去除现有的前导和尾随非单词字符（包括下划线）
    stripped_term = re.sub(r'^\W+|\W+$', '', term)

    # 步骤 2：将非单词字符替换为下划线（一个或多个）
    formatted_term = re.sub(r'\W+', '_', stripped_term)

    # 步骤 3：将多个连续下划线替换为单个下划线
    formatted_term = re.sub(r'_+', '_', formatted_term)

    # 步骤 4：对术语进行 URL 编码，以处理任何剩余的特殊字符
    encoded_term = quote(formatted_term)

    # 步骤 5：添加单个前导和尾随下划线
    term_with_underscores = f"_{encoded_term}_"

    # 步骤 6：与 base_namespace 串联，不添加额外的下划线
    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"):
    """
    通过将非单词字符替换为下划线并进行 URL 编码来为文章创建 URI。

    参数：
        title (str): 文章的标题。
        base_namespace (str): 文章 URI 的基本命名空间。

    返回：
        URIRef: 格式化的文章 URI。
    """
    if pd.isna(title):
        return None
    # 对要在 URI 中使用的文本进行编码
    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 isNone:
        continue

    # 添加 Article 实例
    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)))

    # 添加随机 datePublished 和 access
    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 isNone:
            continue

        # 添加 MeSH 术语实例
        g.add((term_uri, RDF.type, MeSHTerm))
        g.add((term_uri, RDFS.label, Literal(term.replace('_', ' '), datatype=XSD.string)))

        # 将 Article 链接到 MeSH 术语
        g.add((article_uri, schema.about, term_uri))

保存文件的路径

file_path = "/Workspace/PubMedGraph.ttl"

保存文件

g.serialize(destination=file_path, format='turtle')

print(f"File saved at {file_path}")

好的，现在我们有了数据的向量化版本，以及数据的图（RDF）版本。每个向量都有一个与其关联的 URI，该 URI 对应于 KG 中的一个实体，因此我们可以在数据格式之间来回切换。

构建一个应用程序

我决定使用 Streamlit 来构建此图 RAG 应用程序的界面。与上一篇博文类似，我保持了用户流程不变。

\1. 搜索文章： 首先，用户使用搜索词搜索文章。这完全依赖于向量数据库。用户的搜索词将发送到向量数据库，并返回向量空间中与该词最接近的十篇文章。
\2. 优化术语： 其次，用户决定使用哪些 MeSH 术语来过滤返回的结果。由于我们也对 MeSH 术语进行了向量化，因此我们可以让用户输入自然语言提示以获取最相关的 MeSH 术语。然后，我们允许用户扩展这些术语以查看其替代名称和更窄的概念。用户可以为他们的过滤条件选择任意数量的术语。
\3. 过滤和总结： 第三，用户将选定的术语作为过滤器应用于最初的十篇期刊文章。我们可以这样做，因为 PubMed 文章都标有 MeSH 术语。最后，我们允许用户输入一个额外的提示，将其与过滤后的期刊文章一起发送给 LLM。这是 RAG 应用程序的生成步骤。

让我们一次执行这些步骤。您可以在我的 GitHub 上查看完整的应用程序和代码，但以下是结构：

-- app.py (一个 python 文件，用于驱动应用程序并根据需要调用其他函数)
-- query_functions (一个包含带有查询的 python 文件的文件夹)
  -- rdf_queries.py (带有 RDF 查询的 python 文件)
  -- weaviate_queries.py (包含 weaviate 查询的 python 文件)
-- PubMedGraph.ttl (以 RDF 格式存储的 pubmed 数据，作为 ttl 文件存储)

搜索文章

首先，要实现的是 Weaviate 的向量相似度搜索。由于我们的文章已向量化，我们可以将搜索词发送到向量数据库并取回相似的文章。

null

图片由作者提供

在向量数据库中搜索相关期刊文章的主要函数位于 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)

## 在此处提取 URIs
            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

正如您所看到的，我在这里设置了十个结果的限制，只是为了简化操作，但您可以更改它。这只是使用 Weaviate 中的向量相似性搜索来返回相关结果。

应用程序中的最终结果如下所示：

null

作者的图片

作为演示，我将搜索术语“mouth cancer 的治疗方法”。正如您所看到的，返回了 10 篇文章，其中大多数是相关的。这展示了基于向量的检索的优点和缺点。

优点是，我们可以用最少的努力在我们的数据上构建语义搜索功能。正如您在上面看到的，我们所做的只是设置客户端并将数据发送到向量数据库。一旦我们的数据被向量化，我们就可以进行语义搜索、相似性搜索，甚至 RAG。我已将其中一些内容放在了这篇文章附带的笔记本中，但 Weaviate 的官方文档中还有更多内容。

正如我上面提到的，基于向量的检索的缺点是它们是黑盒的，并且难以处理事实知识。在我们的示例中，看起来大多数文章都与某种癌症的某种治疗方法或疗法有关。其中一些文章专门讨论口腔癌，有些文章讨论口腔癌的亚型，如牙龈癌（牙龈癌）和腭癌（腭癌）。但也有关于鼻咽癌（上喉癌）、下颌癌（下颌癌）和食道癌（食道癌）的文章。这些（上喉、下颌或食道）都不被认为是口腔癌。可以理解的是，一篇关于鼻咽肿瘤的特定癌症放射治疗的文章会被认为与“口腔癌的治疗方法”的提示相似，但如果您只寻找口腔癌的治疗方法，它可能并不相关。如果我们将这十篇文章直接插入到我们对 LLM 的提示中，并要求它“总结不同的治疗方案”，我们将获得不正确的信息。

RAG 的目的是为 LLM 提供一组非常具体的额外信息，以更好地回答您的问题——如果这些信息不正确或不相关，它可能会导致 LLM 给出误导性回复。这通常被称为“上下文中毒”。关于上下文中毒，特别危险的是，回复不一定在事实上不准确（LLM 可能会准确地总结我们提供给它的治疗方案），并且它不一定基于不准确的数据（据推测，期刊文章本身是准确的），它只是使用错误的数据来回答您的问题。在此示例中，用户可能正在阅读如何治疗错误的癌症，这似乎非常糟糕。

优化术语

KGs 可以通过优化向量数据库的结果来帮助提高响应的准确性并降低上下文中毒的可能性。下一步是选择我们要用来过滤文章的 MeSH 术语。首先，我们对向量数据库执行另一次向量相似性搜索，但在 Terms 集合上进行搜索。这是因为用户可能不熟悉 MeSH 受控词汇表。在上面的示例中，我搜索了“口腔癌的治疗方法”，但“口腔癌”不是 MeSH 中的一个术语——它们使用“口腔肿瘤”。我们希望用户能够开始探索 MeSH 术语，而无需事先了解它们——无论使用什么元数据标记内容，这都是一个好习惯。

null

作者的图片

获取相关 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

以下是它在应用程序中的样子：

null

作者提供的图片

正如您所看到的，我搜索了“mouth cancer”，并返回了最相似的术语。没有返回“mouth cancer”，因为它不是 MeSH 中的术语，但列表中有“Mouth Neoplasms”。

下一步是允许用户扩展返回的术语以查看替代名称和更窄的概念。这需要查询 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 等。最后的排名基于文章标记的术语数量。

以下是它在应用程序中的样子：

null

作者提供的图片

我可以扩展“Mouth Neoplasms”以查看替代名称，在这种情况下为“Cancer of Mouth”，以及所有更窄的概念。正如您所看到的，大多数更窄的概念都有自己的子概念，您也可以扩展它们。为了演示的目的，我将选择“Mouth Neoplasms”的所有子概念。

null

作者提供的图片

此步骤很重要，不仅因为它允许用户过滤搜索结果，还因为它使用户能够探索 MeSH 图本身并从中学习。例如，这将是用户了解鼻咽肿瘤不是口腔肿瘤的子集的地方。

筛选和总结

现在你已经获得了文章和筛选条件，你可以应用筛选并总结结果。在这里，我们将第一步中返回的 10 篇文章与精炼的 MeSH 术语列表结合起来。我们允许用户在将其发送给 LLM 之前向提示中添加额外的上下文。

null

图片由作者提供

我们进行这种筛选的方式是，我们需要从原始搜索中获取 10 篇文章的 URI。然后，我们可以查询我们的知识图谱，以了解哪些文章已被标记为相关的 MeSH 术语。此外，我们保存这些文章的摘要，以供下一步使用。这里是我们可以根据访问控制或其他用户控制的参数（如作者、文件类型、发布日期等）进行筛选的地方。我没有在本次应用中包含任何内容，但我确实添加了访问控制和发布日期的属性，以防我们稍后想在 UI 中添加它。

这是 app.py 中的代码：

if st.button("FilterArticles"):
            try:
                # Check if we have URIs from tab 1
                if"article_uris"in st.session_state and st.session_state.article_uris:
                    article_uris=st.session_state.article_uris

## Convert list of URIs into a string for the VALUES clause or FILTER
                    article_uris_string=", ".join([f"<{str(uri)}>"foruriinarticle_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}))
                    }}
                    """
                    # Insert the article URIs into the query
                    query=SPARQL_QUERY.format(article_uris=article_uris_string)
                else:
                    st.write("NoarticlesselectedfromTab1.")
                    st.stop()

## Query the RDF and save results in session state
                top_articles=query_rdf(LOCAL_FILE_PATH,query,final_terms)
                st.session_state.filtered_articles=top_articles

if top_articles:

## Combine abstracts from top articles and save in session state
                    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("Noarticlesfoundfortheselectedterms.")
            except Exception as e:
                st.error(f"Errorfiltering articles: {e}")

这使用了 rdf_queries.py 文件中的 query_rdf 函数。该函数如下所示：

### Function to query RDF using SPARQL
def query_rdf(local_file_path, query, mesh_terms, base_namespace="http://example.org/mesh/"):
    if not mesh_terms:
        raise ValueError("The list of MeSH terms is empty or invalid.")

print("SPARQL Query:", query)

## Create and parse the RDF graph
    g = Graph()
    g.parse(local_file_path, format="ttl")

article_data = {}

for term in mesh_terms:
        # Convert the term to a valid URI
        mesh_term_uri = convert_to_uri(term, base_namespace)
        #print("Term:", term, "URI:", mesh_term_uri)

## Perform SPARQL query with initBindings
        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']))
        #print("DEBUG article_data:", article_data)

按匹配 MeSH 术语的数量对文章进行排名

    ranked_articles = sorted(
        article_data.items(),
        key=lambda item: len(item[1]['meshTerms']),
        reverse=True
    )
    return ranked_articles[:10]

正如您所见，此函数还将 MeSH 术语转换为 URI，因此我们可以使用图进行过滤。在将术语转换为 URI 时要小心，并确保其与其他函数对齐。

它在应用程序中的显示效果如下：

null

作者提供的图片

正如您所见，我们从上一步中选择的两个 MeSH 术语就在这里。如果我单击“筛选文章”，它将使用我们在步骤 2 中的筛选条件来筛选最初的 10 篇文章。文章将返回其完整的摘要，以及它们标记的 MeSH 术语（见下图）。

null

作者提供的图片

返回了 5 篇文章。其中两篇被标记为“mouth neoplasms”（口腔肿瘤），一篇被标记为“gingival neoplasms”（牙龈肿瘤），两篇被标记为“palatal neoplasms”（腭肿瘤）。

现在我们有了一份经过精炼的文章列表，我们希望使用它来生成响应，我们可以进入最后一步。我们希望将这些文章发送给 LLM 以生成响应，但我们也可以在提示中添加其他上下文。我有一个默认提示，内容是“在此处以要点形式总结关键信息。让没有医学学位的人也能理解。”对于此演示，我将调整提示以反映我们的原始搜索词：

null

结果如下：

null

在我看来，结果更好，这主要是因为我知道我们正在总结的文章，大概是关于口腔癌的治疗方法。数据集不包含实际的期刊文章，只有摘要。因此，这些结果只是摘要的摘要。这可能有一些价值，但如果我们要构建一个真正的应用程序而不仅仅是一个演示，那么这就是我们可以整合文章全文的步骤。或者，这是用户/研究人员自己阅读这些文章的地方，而不是完全依赖 LLM 进行总结。

结论

本教程演示了如何结合向量数据库和知识图谱可以显著增强 RAG 应用程序。通过利用向量相似性进行初始搜索，并利用结构化知识图谱元数据进行过滤和组织，我们可以构建一个提供准确、可解释和特定领域结果的系统。MeSH（一个完善的受控词汇表）的集成突出了领域专业知识在策划元数据方面的强大功能，这确保了检索步骤与应用程序的独特需求保持一致，同时保持与其他系统的互操作性。这种方法不限于医学——其原则可以应用于结构化数据和文本信息共存的任何领域。

本教程强调了利用每种技术发挥其最佳作用的重要性。向量数据库擅长基于相似性的检索，而知识图谱则擅长提供上下文、结构和语义。此外，扩展 RAG 应用程序需要一个元数据层来打破数据孤岛并实施治理策略。深思熟虑的设计，植根于特定领域的元数据和强大的治理，是构建不仅准确而且可扩展的 RAG 系统的途径。