🗺️ 第1课：知识图谱概述

class KnowledgeGraph: """简单知识图谱实现 - 基于三元组存储""" def __init__(self, name="MyKG"): self.name = name self.triples = [] # 三元组列表 self.entities = set() # 实体集合 self.relations = set() # 关系集合 self.entity_attrs = {} # 实体属性 {entity: {attr: val}} self.outgoing = {} # 出边索引 {head: {(rel, tail)}} self.incoming = {} # 入边索引 {tail: {(rel, head)}} def add_triple(self, head, relation, tail): """添加三元组 (head, relation, tail)""" triple = (head, relation, tail) if triple in self.triples: return False self.triples.append(triple) self.entities.add(head) self.entities.add(tail) self.relations.add(relation) # 更新索引 self.outgoing.setdefault(head, set()).add((relation, tail)) self.incoming.setdefault(tail, set()).add((relation, head)) return True def add_entity_attr(self, entity, attr, value): """为实体添加属性""" self.entities.add(entity) self.entity_attrs.setdefault(entity, {})[attr] = value def query_by_entity(self, entity): """查询与某实体相关的所有三元组""" results = [] for h, r, t in self.triples: if h == entity or t == entity: results.append((h, r, t)) return results def query_by_relation(self, relation): """查询包含特定关系的所有三元组""" return [(h, r, t) for h, r, t in self.triples if r == relation] def get_neighbors(self, entity): """获取实体的所有邻居（一步可达）""" neighbors = set() for r, t in self.outgoing.get(entity, set()): neighbors.add(t) for r, h in self.incoming.get(entity, set()): neighbors.add(h) return neighbors def shortest_path(self, start, end): """BFS求两实体间的最短路径（无向）""" from collections import deque if start == end: return [start] visited = {start} queue = deque([(start, [start])]) while queue: current, path = queue.popleft() for neighbor in self.get_neighbors(current): if neighbor not in visited: new_path = path + [neighbor] if neighbor == end: return new_path visited.add(neighbor) queue.append((neighbor, new_path)) return None def stats(self): """返回图谱统计信息""" return { "名称": self.name, "实体数": len(self.entities), "关系数": len(self.relations), "三元组数": len(self.triples), "平均度": round(sum(len(v) for v in self.outgoing.values()) / max(len(self.entities), 1), 2) } def __repr__(self): s = self.stats() return f"KnowledgeGraph('{self.name}': {s['实体数']}实体, {s['三元组数']}三元组)" # ========== 构建中国文学知识图谱 ========== kg = KnowledgeGraph("中国文学") # 添加人物基本三元组 kg.add_triple("鲁迅", "出生地", "绍兴") kg.add_triple("鲁迅", "创作", "呐喊") kg.add_triple("鲁迅", "创作", "彷徨") kg.add_triple("鲁迅", "任职", "北京大学") kg.add_triple("鲁迅", "好友", "许寿裳") kg.add_triple("许寿裳", "出生地", "绍兴") kg.add_triple("北京大学", "位于", "北京") kg.add_triple("北京大学", "创建于", "1898年") kg.add_triple("老舍", "出生地", "北京") kg.add_triple("老舍", "创作", "骆驼祥子") kg.add_triple("老舍", "创作", "茶馆") kg.add_triple("老舍", "任职", "北京大学") kg.add_triple("绍兴", "属于", "浙江省") kg.add_triple("浙江省", "属于", "中国") kg.add_triple("北京", "属于", "中国") # 添加实体属性 kg.add_entity_attr("鲁迅", "生卒", "1881-1936") kg.add_entity_attr("鲁迅", "原名", "周树人") kg.add_entity_attr("老舍", "生卒", "1899-1966") kg.add_entity_attr("老舍", "原名", "舒庆春") kg.add_entity_attr("呐喊", "出版年", "1923") # 统计信息 print("=== 知识图谱统计 ===") print(kg) for k, v in kg.stats().items(): print(f" {k}: {v}") # 查询示例 print("\n=== 鲁迅相关三元组 ===") for h, r, t in kg.query_by_entity("鲁迅"): print(f" ({h}, {r}, {t})") print("\n=== '创作' 关系 ===") for h, r, t in kg.query_by_relation("创作"): print(f" {h} → {t}") # 路径查询 print("\n=== 路径查询 ===") path = kg.shortest_path("鲁迅", "中国") print(f"鲁迅 → 中国: {' → '.join(path) if path else '不可达'}") path2 = kg.shortest_path("老舍", "呐喊") print(f"老舍 → 呐喊: {' → '.join(path2) if path2 else '不可达'}") # 邻居查询 print("\n=== 北京大学的邻居 ===") print(kg.get_neighbors("北京大学")) # 属性查询 print("\n=== 鲁迅属性 ===") print(kg.entity_attrs.get("鲁迅", {}))

import networkx as nx import json # 将自定义KG转为NetworkX有向图 G = nx.DiGraph() for h, r, t in kg.triples: G.add_edge(h, t, relation=r) # 计算图指标 print("=== 图结构指标 ===") print(f"节点数: {G.number_of_nodes()}") print(f"边数: {G.number_of_edges()}") print(f"密度: {nx.density(G):.4f}") print(f"是否有向: {G.is_directed()}") # 度中心性 degree_centrality = nx.degree_centrality(G) print("\n=== 度中心性 Top5 ===") top5 = sorted(degree_centrality.items(), key=lambda x: -x[1])[:5] for node, cent in top5: print(f" {node}: {cent:.4f}") # 连通分量 if nx.is_weakly_connected(G): print("\n图是弱连通的（所有节点可达）") else: components = list(nx.weakly_connected_components(G)) print(f"\n弱连通分量数: {len(components)}") # 导出为JSON格式（可用于前端可视化） graph_data = { "nodes": [{"id": n, "label": n} for n in G.nodes()], "edges": [{"source": u, "target": v, "label": d["relation"]} for u, v, d in G.edges(data=True)] } print(f"\n导出JSON: {len(graph_data['nodes'])}节点, {len(graph_data['edges'])}边") print("样例边:", json.dumps(graph_data["edges"][0], ensure_ascii=False))

class KGPipeline: """知识图谱构建流水线模拟""" def __init__(self): self.stages = [] def add_stage(self, name, description, input_type, output_type): self.stages.append({ "name": name, "desc": description, "input": input_type, "output": output_type }) def run(self, raw_data): """模拟执行流水线""" print("🚀 知识图谱构建流水线启动\n") data = raw_data for i, stage in enumerate(self.stages, 1): print(f"📌 阶段{i}: {stage['name']}") print(f" 说明: {stage['desc']}") print(f" 输入: {stage['input']} → 输出: {stage['output']}") print() print("✅ 流水线完成！原始数据 → 结构化知识图谱") # 定义标准流水线 pipeline = KGPipeline() pipeline.add_stage("数据采集", "从多源获取原始数据", "多源数据", "原始语料") pipeline.add_stage("实体识别", "识别文本中的命名实体", "原始语料", "实体列表") pipeline.add_stage("关系抽取", "提取实体间语义关系", "实体+语料", "三元组") pipeline.add_stage("实体链接", ">将实体指称映射到知识库", "实体列表", "消歧实体") pipeline.add_stage("知识融合", ">合并多源知识，解决冲突", "多源三元组", ">融合三元组") pipeline.add_stage("知识存储", ">持久化存储到图数据库", ">融合三元组", ">图数据库") pipeline.add_stage(">知识推理", ">基于已有知识推导新知识", ">图数据库", ">扩充图谱") pipeline.run("多源原始数据")