如何使用知识图谱和向量数据库实现 Graph RAG -分步教程（下篇）

from rdflib importGraph,URIRef
from rdflib.namespaceimport 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)!=0else0
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(" ","_"))
returnURIRef(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 notin 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

# 按相似度排序，获取前 5 篇文章
    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)

# 合并标题和摘要的函数
def combine_abstracts(top_3_articles):
    combined_text ="".join(
[f"Title: {data['title']} Abstract: {data['abstract']}"for article_uri, data in top_3_articles]
)
return combined_text

# 合并前三篇文章的摘要
combined_text = combine_abstracts(top_3_articles)
print(combined_text)

import openai

# 设置你的 OpenAI API 密钥
api_key ="YOUR API KEY"
openai.api_key = api_key

def generate_summary(combined_text):
    response = openai.Completion.create(
        model="gpt-3.5-turbo-instruct",
        prompt=f"Summarize the key information here in bullet points. Make it understandable to someone without a medical degree:\n\n{combined_text}",
        max_tokens=1000,
        temperature=0.3
)

# 获取原始文本输出
    raw_summary = response.choices[0].text.strip()

# 将文本按行拆分并清理空白字符
    lines = raw_summary.split('\n')
    lines =[line.strip()for line in lines if line.strip()]

# 将行重新合并并保持换行
    formatted_summary ='\n'.join(lines)

return formatted_summary

# 生成并打印摘要
summary = generate_summary(combined_text)
print(summary)

-一名14岁男孩在前上颌发现了一个牙龈肿瘤，已被切除并通过光学和电子显微镜研究
-该肿瘤由肌上皮细胞组成，并且呈现恶性特征
-电子显微镜显示该肿瘤起源于唾液腺
-这是唯一一个确认存在恶性特征的肌上皮瘤案例
-进行了一项可行性研究，旨在提高在高发区域对口腔癌和癌前病变的早期检测
-烟草商参与了分发传单，邀请吸烟者接受全科医生的免费检查
-研究包括了93名患者，27%被转诊到专科医生
-63.6%的被转诊患者实际上看了专家，15.3%被确认有癌前病变
-一项研究发现口腔鳞状细胞癌（OSCC）中，胚胎致死异常视觉蛋白HuR的表达与环氧化酶 COX-2的表达相关
-胶质胞质中的HuR表达与 COX-2表达、淋巴结及远处转移相关
-抑制HuR表达导致口腔癌细胞中 COX-2表达减少。

# 创建有效 URI 的函数
def create_valid_uri(base_uri, text):
if pd.isna(text):
returnNone
# 对文本进行编码，使其适用于 URI
    sanitized_text = urllib.parse.quote(text.strip().replace(' ','_').replace('"','').replace('<','').replace('>','').replace("'","_"))
returnURIRef(f"{base_uri}/{sanitized_text}")

# 为 DataFrame 添加一列新的文章 URI
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 column type: {df['Title'].dtype}")
logging.info(f"abstractText column type: {df['abstractText'].dtype}")
logging.info(f"meshMajor column type: {df['meshMajor'].dtype}")
logging.info(f"Article_URI column type: {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)
)
exceptExceptionas e:
            logging.error(f"Error processing row {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))

from rdflib importGraph,Namespace,URIRef,Literal
from rdflib.namespaceimport 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.w3.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"Article URI: {uri}")
print(f"Title: {details['title']}")
print(f"Date Published: {details['date_published']}")
print()

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))

# 假设 response 是包含你文章的数据结构
article_uris =[URIRef(article["article_URI"])for article in response["data"]["Get"]["Articles_with_abstracts_and_URIs"]]

from rdflib importURIRef

# 使用文章 URI 过滤器构造 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.

# 过滤器，只包含 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 notin 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"Title: {data['title']}")
print(f"Abstract: {data['abstract']}")
print("MeSH Terms:")
for mesh_term in data['meshTerms']:
print(f"  - {mesh_term}")
print()

from rdflib importGraph,Namespace,URIRef,Literal
from rdflib.namespaceimport 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 notin 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"Article URI: {uri}")
print(f"Title: {details['title']}")
print(f"Abstract: {details['abstract']}")
print(f"Date Published: {details['date_published']}")
print(f"Access: {details['access']}")
print()

response =(
    client.query
.get("Articles_with_abstracts_and_URIs",["title","abstractText",'article_URI','meshMajor'])
.with_near_text({"concepts":["therapies for mouth neoplasms"]})
.with_limit(3)
.with_generate(grouped_task="Summarize the key information here in bullet points. Make it understandable to someone without a medical degree.")
.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("Filtered articles:")
for article in filtered_articles:
print(f"Title: {article['title']}")
print(f"URI: {article['article_URI']}")
print(f"Abstract: {article['abstractText']}")
print(f"MeshMajor: {article['meshMajor']}")
print("---")

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("Extracted article URIs:")
for uri in article_uris:
print(uri)

from rdflib importURIRef

# 使用文章 URI 过滤器构造 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.

# 过滤器，只包含 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 notin 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"Title: {data['title']}")
print(f"Abstract: {data['abstract']}")
print("MeSH Terms:")
for mesh_term in data['meshTerms']:
print(f"  - {mesh_term}")
print()

# 过滤响应
filtered_articles = filter_articles_by_uri(response, matching_articles)

# 将标题和摘要合并成一段文本的函数
def combine_abstracts(filtered_articles):
    combined_text ="\n\n".join(
[f"Title: {article['title']}\nAbstract: {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 波长组的治疗效果较好
-恶性间皮瘤的牙龈转移非常罕见，且生存率较低
-一项病例研究显示，患者出现牙龈肿块作为恶性间皮瘤多器官复发的首个征兆，强调了对所有新生长病变进行活检的重要性，即使在不常见的解剖部位也应如此。