知识图谱构建与应用

@prefix ex: <http://example.org/> .
@prefix foaf: <http://xmlns.com/foaf/0.1/> .

ex:Einstein rdf:type ex:Scientist ;
            foaf:name "Albert Einstein" ;
            ex:birthPlace ex:Germany ;
            ex:developed ex:TheoryOfRelativity .

ex:TheoryOfRelativity rdf:type ex:ScientificTheory ;
                      ex:published "1905" .

# 查询爱因斯坦的所有理论
SELECT ?theory ?name
WHERE {
  ?einstein foaf:name "Albert Einstein" .
  ?einstein ex:developed ?theory .
  ?theory foaf:name ?name .
}

# 查询出生在德国的科学家的理论
SELECT ?scientist ?theory
WHERE {
  ?scientist rdf:type ex:Scientist ;
             ex:birthPlace ex:Germany ;
             ex:developed ?theory .
}

# 两跳推理：德国科学家的理论影响了哪些领域
SELECT DISTINCT ?field
WHERE {
  ?scientist ex:birthPlace ex:Germany ;
             ex:developed ?theory .
  ?theory ex:influences ?field .
}

# 使用 spaCy 进行 NER
import spacy

nlp = spacy.load("zh_core_web_trf")  # 中文 transformer 模型

text = "爱因斯坦出生于德国，后来开发了相对论，对美国物理学产生了重要影响。"
doc = nlp(text)

for ent in doc.ents:
    print(f"{ent.text} -> {ent.label_}")

# 输出:
# 爱因斯坦 -> PERSON
# 德国 -> GPE
# 相对论 -> WORK_OF_ART
# 美国 -> GPE

import re

# 基于模板的关系抽取
def extract_relations(text, entities):
    relations = []
    
    # 模板：X 出生于 Y
    pattern = r"(.+?)出生于(.+?)[，。]"
    for match in re.finditer(pattern, text):
        subj = match.group(1).strip()
        obj = match.group(2).strip()
        if subj in entities and obj in entities:
            relations.append((subj, "birthPlace", obj))
    
    # 模板：X 开发了 Y
    pattern = r"(.+?)开发了(.+?)[，。]"
    for match in re.finditer(pattern, text):
        subj = match.group(1).strip()
        obj = match.group(2).strip()
        if subj in entities and obj in entities:
            relations.append((subj, "developed", obj))
    
    return relations

# BERT  fine-tuning 进行关系分类
from transformers import BertTokenizer, BertForSequenceClassification
import torch

tokenizer = BertTokenizer.from_pretrained("bert-base-chinese")
model = BertForSequenceClassification.from_pretrained(
    "bert-base-chinese",
    num_labels=len(relation_types)
)

# 输入: [CLS] 爱因斯坦 [SEP] 德国 [SEP]
# 标签: birthPlace

% Datalog 风格规则
% 规则：如果 X 出生在 Y，Y 的首都是 Z，那么 X 的国籍是 Y
nationality(X, Y) :- birthPlace(X, Y), country(Y).

% 传递闭包
influences(X, Z) :- influences(X, Y), influences(Y, Z).

# 使用 PyTorch Geometric 或 DGL-KE 训练嵌入
# TransE 模型：v_h + v_r ≈ v_t（头实体 + 关系 ≈ 尾实体）

import torch
import torch.nn as nn

class TransE(nn.Module):
    def __init__(self, num_entities, num_relations, dim):
        super().__init__()
        self.entity_embeddings = nn.Embedding(num_entities, dim)
        self.relation_embeddings = nn.Embedding(num_relations, dim)
        
    def score(self, heads, relations, tails):
        h = self.entity_embeddings(heads)
        r = self.relation_embeddings(relations)
        t = self.entity_embeddings(tails)
        return -torch.norm(h + r - t, p=1, dim=1)  # 负距离作为得分

from neo4j import GraphDatabase
from openai import OpenAI

class KGRAG:
    def __init__(self, neo4j_uri, openai_key):
        self.driver = GraphDatabase.driver(neo4j_uri, auth=("neo4j", "password"))
        self.llm = OpenAI(api_key=openai_key)
    
    def retrieve_subgraph(self, entities):
        """检索与实体相关的子图"""
        with self.driver.session() as session:
            result = session.run("""
                MATCH (e)-[r]-(related)
                WHERE e.name IN $entities
                RETURN e.name AS source, type(r) AS relation,
                       related.name AS target, r.description AS desc
                LIMIT 30
            """, entities=entities)
            return [record.data() for record in result]
    
    def answer(self, question):
        # Step 1: 从问题中提取实体（可用 LLM 或 NER 模型）
        entities = self.extract_entities(question)
        
        # Step 2: 检索知识图谱子图
        subgraph = self.retrieve_subgraph(entities)
        
        # Step 3: 将子图转为文本上下文
        context = self.subgraph_to_text(subgraph)
        
        # Step 4: LLM 生成回答
        prompt = f"""基于以下知识回答用户问题。
知识：{context}
问题：{question}
如果不确定，请直接说明知识不足。"""
        
        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content

    def extract_entities(self, text):
        # 简化版：关键词匹配（生产环境使用 NER 模型）
        known_entities = ["爱因斯坦", "相对论", "德国", "量子力学"]
        return [e for e in known_entities if e in text]
    
    def subgraph_to_text(self, subgraph):
        facts = []
        for edge in subgraph:
            facts.append(f"{edge['source']} 的 {edge['relation']} 是 {edge['target']}。")
        return "\n".join(facts)