知识图谱构建实战：从文本到图谱的完整流水线

import spacy

nlp = spacy.load("zh_core_web_trf")  # transformer 中文模型
doc = nlp("2024年，华为在深圳发布了昇腾AI芯片，任正非是华为创始人。")

for ent in doc.ents:
    print(f"{ent.text}\t{ent.label_}\t{ent.start_char}-{ent.end_char}")
# 输出：
# 2024年    DATE
# 华为     ORG
# 深圳     GPE
# 昇腾AI芯片 PRODUCT
# 任正非   PERSON
# 华为     ORG

# 候选生成：字符串 → 候选实体（拼音/别名/精确匹配）
def generate_candidates(entity_text, entity_index):
    candidates = []
    # 1. 精确匹配
    if entity_text in entity_index:
        candidates += entity_index[entity_text]
    # 2. 拼音/别名校对（apoc.text.clean 同思路）
    cleaned = clean_name(entity_text)
    for alias, eids in entity_index.items():
        if cleaned == clean_name(alias):
            candidates += eids
    return candidates

# 候选消歧：用上下文向量相似度打分
def disambiguate(candidates, context_vec):
    scores = []
    for eid, desc_vec in candidates:
        sim = cosine_similarity(desc_vec, context_vec)
        scores.append((eid, sim))
    return max(scores, key=lambda x: x[1])

# 模糊匹配：Jaro-Winkler 适合中文组织名
import jellyfish
print(jellyfish.jaro_winkler_similarity("华为", "华为公司"))   # 0.80
print(jellyfish.jaro_winkler_similarity("华为", "中兴"))       # 0.33

import re

RULES = [
    (r"(.+?)出生于(.+?)[，。；]", "BORN_IN"),
    (r"(.+?)毕业于(.+?)[，。；]", "GRADUATED_FROM"),
    (r"(.+?)是(.+?)的创始人[，。；]", "FOUNDED_BY"),
    (r"(.+?)成立于(.+?)[，。；]", "FOUNDED_ON"),
]

def rule_extract(text):
    triples = []
    for pattern, rel in RULES:
        for m in re.finditer(pattern, text):
            subj, obj = m.group(1).strip(), m.group(2).strip()
            triples.append((subj, rel, obj))
    return triples

print(rule_extract("任正非是华为的创始人，华为成立于1987年。"))
# [('任正非', 'FOUNDED_BY', '华为'), ('华为', 'FOUNDED_ON', '1987年')]

import spacy

nlp = spacy.load("zh_core_web_trf")
doc = nlp("华为发布了昇腾AI芯片")

# 找 nsubj（主语）和 dobj（直接宾语）之间的动词
for token in doc:
    if token.dep_ == "nsubj":
        subj = token.text
        verb = token.head.text
        objs = [t.text for t in token.head.children if t.dep_ in ("dobj", "dative")]
        for obj in objs:
            print((subj, verb, obj))
# ('华为', '发布', '昇腾AI芯片')

import json
from openai import OpenAI

client = OpenAI()

def llm_extract(sentence):
    prompt = f"""从句子中抽取所有事实三元组，只输出 JSON 数组。
每个三元组格式：{{"head": 主语, "relation": 关系, "tail": 宾语}}
允许的关系：BORN_IN, WORKS_AT, FOUNDED_BY, DEVELOPED, ACQUIRED, PART_OF
禁止臆造，句子中没有的事实不要输出。
句子：{sentence}"""
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"}
    )
    data = json.loads(resp.choices[0].message.content)
    return data.get("triples", [])

print(llm_extract("华为的创始人任正非在1987年创立了公司。"))

from transformers import AutoTokenizer, AutoModelForSequenceClassification

tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese")
model = AutoModelForSequenceClassification.from_pretrained(
    "bert-base-chinese", num_labels=12  # 12 种关系
)

def classify_relation(head, tail, sentence):
    text = f"[CLS]{sentence}[SEP]{head}[SEP]{tail}[SEP]"
    inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=128)
    logits = model(**inputs).logits
    rel_id = logits.argmax().item()
    return ID_TO_RELATION[rel_id]

# ontology.yaml：知识图谱的概念层
EntityTypes:
  - Organization: {props: [name, foundedYear, headquarters]}
  - Person: {props: [name, birthYear, nationality]}
  - Product: {props: [name, category]}
  - Technology: {props: [name, domain]}

Relations:
  - FOUNDED_BY:   {from: Organization, to: Person}
  - WORKS_AT:     {from: Person, to: Organization}
  - DEVELOPED:    {from: Organization, to: Product}
  - ACQUIRED:     {from: Organization, to: Organization}

Constraints:
  - "每家企业至少有一个名称"
  - "FOUNDED_BY 的尾实体必须是 Person"

@prefix ex: <http://example.org/> .
ex:Huawei ex:founder ex:RenZhengfei .
ex:Huawei ex:name "华为技术有限公司" .

import re

def preprocess(raw_text):
    # 去 HTML、去噪声
    text = re.sub(r"<[^>]+>", "", raw_text)
    text = re.sub(r"\s+", " ", text)
    # 段落切分（抽取的最小单元，保证共指上下文）
    paragraphs = [p.strip() for p in re.split(r"\n{2,}", text) if len(p.strip()) > 50]
    return paragraphs

# 增量策略：按变更批次入库，保留 provenance
def incremental_load(batch, batch_id):
    tx.run("""
        UNWIND $triples AS t
        MERGE (h:Entity {id: t.head_id})
        ...
        WITH h, tl, t, $batch_id AS bid
        CALL apoc.merge.relationship(h, t.relation, 
            {batchId: bid, confidence: t.confidence}, {}, tl)
        YIELD rel
        SET rel.validUntil = $validUntil   // 支持软删除/过期
    """, triples=batch, batch_id=batch_id, validUntil=None)

# 案例数据：3 条新闻
news = [
    "华为在深圳发布了昇腾AI芯片，用于大模型训练。",
    "任正非创立了华为，总部位于深圳。",
    "阿里云推出了通义千问大模型，与昇腾芯片深度适配。",
]