🔧 第18课：知识补全

import numpy as np from typing import List, Tuple, Dict class KnowledgeCompletionEngine: """知识补全引擎 - 集成多种方法""" def __init__(self): self.triples = set() self.entity_pairs = {} # {(h,r): [t_list]} self.relation_pairs = {} # {(h,t): [r_list]} self.entity_count = {} self.relation_count = {} self.rule_patterns = [] def add_triples(self, triples): for h, r, t in triples: self.triples.add((h, r, t)) self.entity_pairs.setdefault((h, r), []).append(t) self.relation_pairs.setdefault((h, t), []).append(r) self.entity_count[h] = self.entity_count.get(h, 0) + 1 self.entity_count[t] = self.entity_count.get(t, 0) + 1 self.relation_count[r] = self.relation_count.get(r, 0) + 1 def predict_tail(self, h, r, top_k=5): """预测 (h, r, ?) 的尾实体""" scores = {} # 方法1: 规则推理 for pattern_h, pattern_r, pattern_t in self.rule_patterns: if pattern_r == r and pattern_h == h: scores[pattern_t] = scores.get(pattern_t, 0) + 0.4 # 方法2: 关系共现（如果h的其他关系指向某实体，该实体可能是答案） for (eh, er), tails in self.entity_pairs.items(): if eh == h and er != r: for t in tails: scores[t] = scores.get(t, 0) + 0.2 # 方法3: 相似实体的答案 for (eh, er), tails in self.entity_pairs.items(): if er == r and eh != h: # 检查eh和h是否共享其他关系 shared = len(set(k[1] for k in self.entity_pairs if k[0] == eh) & set(k[1] for k in self.entity_pairs if k[0] == h)) if shared > 0: for t in tails: scores[t] = scores.get(t, 0) + shared * 0.1 # 方法4: 实体流行度先验 for entity in self.entity_count: scores[entity] = scores.get(entity, 0) + self.entity_count[entity] * 0.01 ranked = sorted(scores.items(), key=lambda x: -x[1]) return ranked[:top_k] def predict_relation(self, h, t, top_k=3): """预测 (h, ?, t) 的关系""" # 直接查询 if (h, t) in self.relation_pairs: return [(r, 1.0) for r in self.relation_pairs[(h, t)]] # 路径推理 scores = {} for (h2, r1), mid_list in self.entity_pairs.items(): if h2 == h: for mid in mid_list: for (mid2, r2), tail_list in self.entity_pairs.items(): if mid2 == mid and t in tail_list: path = f"{r1}→{r2}" scores[path] = scores.get(path, 0) + 0.3 return sorted(scores.items(), key=lambda x: -x[1])[:top_k] def triple_score(self, h, r, t): ">>>三元组置信度评分""" score = 0.0 if (h, r, t) in self.triples: score += 1.0 # 已知三元组 if (h, r) in self.entity_pairs: score += 0.3 if (h, t) in self.relation_pairs: score += 0.3 return min(score, 1.0) # ========== 测试 ========== engine = KnowledgeCompletionEngine() engine.add_triples([ ("鲁迅", "创作", "呐喊"), ("鲁迅", "创作", "彷徨"), ("鲁迅", "出生地", "绍兴"), ("老舍", "创作", "骆驼祥子"), ("老舍", "出生地", "北京"), ("巴金", "创作", "家"), ("巴金", "出生地", "成都"), ("徐志摩", "创作", "再别康桥"), ("绍兴", "属于", "浙江省"), ("北京", "属于", "中国"), ("浙江省", "属于", "中国"), ]) print("=== 尾实体预测: (老舍, 创作, ?) ===") for entity, score in engine.predict_tail("老舍", "创作"): print(f" {entity}: {score:.3f}") print(" === 尾实体预测: (徐志摩, 出生地, ?) ===") for entity, score in engine.predict_tail("徐志摩", "出生地"): print(f" {entity}: {score:.3f}") print(" === 关系预测: (绍兴, ?, 中国) ===" for rel, score in engine.predict_relation("绍兴", "中国"): print(f" {rel}: {score:.3f}") print(" === 三元组置信度 ===") for t in [("鲁迅","创作","呐喊"), ("鲁迅","创作","骆驼祥子"), ("老舍","出生地","成都")]: print(f" {t}: {engine.triple_score(*t):.3f}")