科研知识图谱构建方法

# -*- coding: utf-8 -*-"""Mini knowledge graph pipeline: NER -> RE -> fusion -> storage -> query."""import refrom collections import deque # --- 1. 语料：模拟科研论文中的句子 ---CORPUS = [    "p53 regulates apoptosis and inhibits tumor growth.",    "P53 regulates tumor growth.",    "Aspirin treats inflammation.",    "Aspirin reduces fever.",    "BRCA1 interacts with p53 in DNA repair.",    "Metformin treats diabetes.",    "Apoptosis inhibits tumor growth.",] # --- 2. 词典式实体识别：把词表里的术语当作地标 ---ENTITY_DICT = {    "gene": ["p53", "brca1"],    "disease": ["apoptosis", "tumor", "inflammation", "fever", "diabetes"],    "drug": ["aspirin", "metformin"],} # --- 3. 别名表实现知识融合：P53 与 p53 指向同一实体 ---ALIASES = {"P53": "p53", "BRCA1": "brca1"} # --- 4. 关系模板：用正则表达式在句中找"路牌" ---PATTERNS = [    ("regulates", r"(\w+)\s+regulates\s+(\w+)"),    ("treats", r"(\w+)\s+treats\s+(\w+)"),    ("reduces", r"(\w+)\s+reduces\s+(\w+)"),    ("inhibits", r"(\w+)\s+inhibits\s+(\w+)"),    ("interacts_with", r"(\w+)\s+interacts with\s+(\w+)"),] def canonical(name):    """把各种写法归一化为同一实体名。"""    return ALIASES.get(name.upper(), name.lower()) def recognize_entities(sentence):    """返回句子中出现的所有已知实体。"""    found = set()    for ent_type, words in ENTITY_DICT.items():        for w in words:            if re.search(r"\b" + re.escape(w) + r"\b", sentence, re.IGNORECASE):                found.add(canonical(w))    return found def extract_triplets(sentence):    """抽取满足'头尾都是已知实体'的三元组。"""    ents = recognize_entities(sentence)    triplets = []    for rel, pat in PATTERNS:        for m in re.finditer(pat, sentence, re.IGNORECASE):            h, t = canonical(m.group(1)), canonical(m.group(2))            if h in ents and t in ents and h != t:                triplets.append((h, rel, t))    return triplets def build_graph(sentences):    """把三元组写入邻接表，完成图谱存储。"""    graph = {}    triples = []    for s in sentences:        for h, r, t in extract_triplets(s):            triples.append((h, r, t))            graph.setdefault(h, []).append((t, r))            graph.setdefault(t, []).append((h, r + "_rev"))    # 简单融合：去掉完全重复的三元组    uniq = list(dict.fromkeys(triples))    return graph, uniq def shortest_path(graph, start, goal):    """BFS 广度优先搜索：回答两个实体之间怎么走。"""    if start not in graph or goal not in graph:        return None    queue = deque([(start, [start])])    seen = {start}    while queue:        node, path = queue.popleft()        for nxt, rel in graph[node]:            if nxt == goal:                return path + [nxt]            if nxt not in seen:                seen.add(nxt)                queue.append((nxt, path + [nxt]))    return None def main():    graph, triples = build_graph(CORPUS)    print("=== triples (deduplicated) ===")    for h, r, t in triples:        print("  (%s, %s, %s)" % (h, r, t))    print("\n=== entities in graph ===")    print(len(graph), "entities:", sorted(graph.keys()))    print("\n=== fusion demo ===")    print("P53 is normalized to p53 by ALIASES, so they share one node.")    print("\n=== shortest path query ===")    print("p53 -> tumor:", shortest_path(graph, "p53", "tumor"))    print("aspirin -> inflammation:", shortest_path(graph, "aspirin", "inflammation")) if __name__ == "__main__":    main()