📖 第28课：知识图谱构建

class KnowledgeGraph: def __init__(self): self.entities = set() self.relations = [] self.entity_attrs = {} def add_triple(self, subject, relation, obj): self.entities.add(subject) self.entities.add(obj) self.relations.append((subject, relation, obj)) def add_entity_attr(self, entity, attr, value): if entity not in self.entity_attrs: self.entity_attrs[entity] = {} self.entity_attrs[entity][attr] = value def query_by_entity(self, entity): results = [] for s, r, o in self.relations: if s == entity: results.append((r, o, 'outgoing')) if o == entity: results.append((r, s, 'incoming')) return results def query_by_relation(self, relation): return [(s, o) for s, r, o in self.relations if r == relation] def multi_hop_query(self, start_entity, hops=2): paths = [[start_entity]] for _ in range(hops): new_paths = [] for path in paths: last = path[-1] for s, r, o in self.relations: if s == last and o not in path: new_paths.append(path + [r, o]) paths = new_paths return paths kg = KnowledgeGraph() # 构建知识图谱 triples = [ ("马云", "创建", "阿里巴巴"), ("阿里巴巴", "总部位于", "杭州"), ("阿里巴巴", "行业", "电子商务"), ("张勇", "担任", "CEO"), ("张勇", "就职于", "阿里巴巴"), ("腾讯", "总部位于", "深圳"), ("马化腾", "创建", "腾讯"), ("腾讯", "行业", "社交"), ("腾讯", "投资", "美团"), ("美团", "总部位于", "北京"), ("美团", "行业", "本地生活"), ] for s, r, o in triples: kg.add_triple(s, r, o) print("=== 知识图谱查询 ===") print(f"实体数: {len(kg.entities)}, 关系数: {len(kg.relations)}") print("\n阿里巴巴的关联:") for rel, entity, direction in kg.query_by_entity("阿里巴巴"): arrow = "→" if direction == "outgoing" else "←" print(f" {entity} {arrow} [{rel}]") print("\n总部位于关系:") for s, o in kg.query_by_relation("总部位于"): print(f" {s} → {o}") print("\n马云的2跳查询:") paths = kg.multi_hop_query("马云", hops=2) for path in paths: print(f" {' → '.join(path)}")

class KGQA: def __init__(self, kg): self.kg = kg def answer(self, question): # 简单的基于模板的KG问答 import re patterns = [ (r'(.+?)的(.+?)是什么', self._query_attr), (r'(.+?)的(.+)', self._query_relation), (r'谁(.+?)了(.+)', self._query_reverse), ] for pattern, handler in patterns: match = re.match(pattern, question) if match: return handler(match) return "无法理解该问题" def _query_attr(self, match): entity = match.group(1) attr = match.group(2) if entity in self.kg.entity_attrs and attr in self.kg.entity_attrs[entity]: return self.kg.entity_attrs[entity][attr] # 尝试关系查询 for rel, obj, direction in self.kg.query_by_entity(entity): if attr in rel: return obj if direction == 'outgoing' else f"{obj} (反向)" return f"未知: {entity}的{attr}" def _query_relation(self, match): entity = match.group(1) relation = match.group(2) for rel, obj, direction in self.kg.query_by_entity(entity): if relation in rel or rel in relation: return obj if direction == 'outgoing' else f"{obj} (反向)" return f"未知: {entity}的{relation}" def _query_reverse(self, match): relation = match.group(1) obj = match.group(2) for s, r, o in self.kg.relations: if o == obj and relation in r: return s return f"未知: 谁{relation}了{obj}" qa = KGQA(kg) questions = [ "阿里巴巴的行业是什么", "阿里巴巴的总部位于", "谁创建了腾讯", "马云的2跳查询", ] print("=== 知识图谱问答 ===") for q in questions: if '2跳' in q: paths = kg.multi_hop_query("马云", hops=2) answer = ' | '.join(['→'.join(p) for p in paths[:3]]) else: answer = qa.answer(q) print(f" 问: {q}") print(f" 答: {answer}")