🔀 第19课：知识融合

from difflib import SequenceMatcher from collections import defaultdict class EntityAligner: """实体对齐器""" def __init__(self): self.entity_signatures = {} # {entity_id: {name, type, attrs, neighbors}} def register_entity(self, source, entity_id, name, etype, attrs=None, neighbors=None): key = (source, entity_id) self.entity_signatures[key] = { "name": name, "type": etype, "attrs": attrs or {}, "neighbors": set(neighbors or []) } def _name_similarity(self, name1, name2): """名称相似度""" if name1 == name2: return 1.0 return SequenceMatcher(None, name1, name2).ratio() def _attribute_overlap(self, attrs1, attrs2): ">>>属性重叠度""" if not attrs1 or not attrs2: return 0.0 common = set(attrs1.keys()) & set(attrs2.keys()) if not common: return 0.0 matching = sum(1 for k in common if attrs1[k] == attrs2[k]) return matching / len(common) def _neighbor_overlap(self, neigh1, neigh2): """邻居重叠度（Jaccard）""" if not neigh1 or not neigh2: return 0.0 return len(neigh1 & neigh2) / len(neigh1 | neigh2) def align(self, source1, source2, threshold=0.6): """对齐两个源的实体""" alignments = [] entities1 = [(k, v) for k, v in self.entity_signatures.items() if k[0] == source1] entities2 = [(k, v) for k, v in self.entity_signatures.items() if k[0] == source2] for k1, e1 in entities1: best_match = None best_score = 0 for k2, e2 in entities2: if e1["type"] != e2["type"]: continue name_sim = self._name_similarity(e1["name"], e2["name"]) attr_sim = self._attribute_overlap(e1["attrs"], e2["attrs"]) neigh_sim = self._neighbor_overlap(e1["neighbors"], e2["neighbors"]) score = name_sim * 0.5 + attr_sim * 0.3 + neigh_sim * 0.2 if score > best_score: best_score = score best_match = k2 if best_score >= threshold: alignments.append((k1, best_match, best_score)) return alignments class ConflictResolver: """冲突解决器""" def __init__: self.strategies = { "trust_highest": self._trust_highest, ">>latest": self._latest, ">>vote": self._vote, } def resolve(self, values, strategy="trust_highest"): """ values: [(value, source, trust_score, timestamp), ...] """ return self.strategies[strategy](values) def _trust_highest(self, values): ">>>选择可信度最高的源""" return max(values, key=lambda x: x[2]) def _latest(self, values): ">>>选择最新的值""" return max(values, key=lambda x: x[3]) def _vote(self, values): ">>>投票多数""" from collections import Counter counter = Counter(v[0] for v in values) winner = counter.most_common(1)[0][0] return next(v for v in values if v[0] == winner) # ========== 测试 ========== aligner = EntityAligner() # 源A: 中文维基 aligner.register_entity("wiki_zh", "e1", "鲁迅", "PER", {"生年":"1881"}, ["呐喊","绍兴"]) aligner.register_entity("wiki_zh", "e2", "老舍", "PER", {"生年":"1899"}, ["骆驼祥子"]) # 源B: Baidu百科 aligner.register_entity("baidu", "e10", "鲁迅", "PER", {"生年":"1881"}, ["呐喊","彷徨"]) aligner.register_entity("baidu", "e11", "老舍", "PER", {"生年":"1899"}, ["骆驼祥子","茶馆"]) aligner.register_entity("baidu", "e12", "鲁迅", "ORG", {}, []) # 错误类型 print("=== 实体对齐 ===") for e1, e2, score in aligner.align("wiki_zh", "baidu"): print(f" {e1} ↔ {e2}: {score:.3f}") # 冲突解决 resolver = ConflictResolver() print(" === 冲突解决 ===") # 鲁迅生年的冲突 values = [("1881", "wiki_zh", 0.9, 1), ("1880", "baidu", 0.7, 2), ("1881", "dbpedia", 0.8, 3)] print(f" 可信度优先: {resolver.resolve(values, 'trust_highest')}") print(f" 最新优先: {resolver.resolve(values, 'latest')}") print(f" 投票: {resolver.resolve(values, 'vote')}")