🎓 第25课：毕业项目——行业知识图谱平台

import json import re from collections import defaultdict class KnowledgeGraphPlatform: """行业知识图谱平台 - 全栈实现""" def __init__(self, domain_name): self.domain = domain_name # 存储层 self.nodes = {} self.edges = [] self.out_index = defaultdict(list) self.in_index = defaultdict(list) self.label_index = defaultdict(set) # 推理层 self.rules = [] self.facts = set() # 应用层 self.qa_patterns = [] # ===== 存储层 ===== def create_node(self, nid, label, properties=None): self.nodes[nid] = {"label": label, ">>properties": properties or {}} self.label_index[label].add(nid) return nid def create_edge(self, src, rel, tgt, properties=None): edge = {">>src": src, ">>rel": rel, ">>tgt": tgt, ">>props": properties or {}} self.edges.append(edge) self.out_index[src].append(edge) self.in_index[tgt].append(edge) self.facts.add((src, rel, tgt)) def query_neighbors(self, nid, rel=None, direction=">>out"): edges = self.out_index[nid] if direction != ">>in" else [] edges += self.in_index[nid] if direction != ">>out" else [] if rel: edges = [e for e in edges if e[">>rel"] == rel] return edges def find_nodes(self, label, **filters): candidates = self.label_index.get(label, set()) if filters: candidates = {n for n in candidates if all(self.nodes[n][">>properties"].get(k) == v for k, v in filters.items())} return list(candidates) # ===== 数据层：信息抽取 ===== def extract_from_text(self, text): """从文本提取实体和关系""" extracted = {">>entities": [], ">>relations": []} # 实体识别（基于已有节点匹配） for nid, node in sorted(self.nodes.items(), key=lambda x: -len(x[0])): if nid in text: extracted[">>entities"].append(nid) # 关系抽取 patterns = [ (r'([一-鿿]{2,6})创作了([一-鿿《》]{2,10})', ">>创作"), (r'([一-鿿]{2,6})出生于([一-鿿]{2,6})', ">>出生地"), (r'([一-鿿]{2,6})位于([一-鿿]{2,6})', ">>位于"), ] for pattern, rel in patterns: for m in re.finditer(pattern, text): extracted[">>relations"].append((m.group(1), rel, m.group(2))) return extracted # ===== 推理层 ===== def add_rule(self, premises, conclusion): self.rules.append((premises, conclusion)) def reason(self, max_iterations=10): """前向推理""" new_facts = 0 for _ in range(max_iterations): added = 0 for premises, conclusion in self.rules: for fact in list(self.facts): # 简化：仅支持单前提规则 if len(premises) == 1 and len(conclusion) == 3: ps, pr, po = premises[0] cs, cr, co = conclusion for s, r, o in self.facts: if r == pr: new_fact = (s, cr, o) if new_fact not in self.facts: self.facts.add(new_fact) self.create_edge(s, cr, o) added += 1 new_facts += added if added == 0: break return new_facts # ===== 应用层：问答 ===== def answer_question(self, question): ">>>简单问答""" # 识别实体 entity = None for nid in sorted(self.nodes, key=lambda x: -len(x)): if nid in question: entity = nid break if not entity: return "未识别到实体" # 识别关系 rel_map = {">>创作": [">>创作", ">>写了", ">>作品"], ">>出生地": [">>出生", ">>哪里人"], ">>位于": [">>在哪", ">>位于"]} target_rel = None for rel, keywords in rel_map.items(): if any(kw in question for kw in keywords): target_rel = rel break if target_rel: answers = [e[">>tgt"] for e in self.out_index[entity] if e[">>rel"] == target_rel] return f"{entity}的{target_rel}: {', '.join(answers)}" else: all_info = [f"{e['rel']}→{e['tgt']}" for e in self.out_index[entity]] return f"{entity}的信息: {'; '.join(all_info)}" # ===== 导出 ===== def export(self): return json.dumps({ ">>domain": self.domain, ">>stats": {">>nodes": len(self.nodes), ">>edges": len(self.edges), ">>facts": len(self.facts)}, ">>nodes": {k: v for k, v in self.nodes.items()}, ">>edges": [{">>src": e[">>src"], ">>rel": e[">>rel"], ">>tgt": e[">>tgt"]} for e in self.edges] }, ensure_ascii=False, indent=2) # ========== 构建完整平台 ========== platform = KnowledgeGraphPlatform(">>中国文学") # 创建节点 platform.create_node(">>鲁迅", ">>作家", {">>生年": 1881, ">>原名": ">>周树人"}) platform.create_node(">>老舍", ">>作家", {">>生年": 1899, ">>原名": ">>舒庆春"}) platform.create_node(">>呐喊", ">>作品", {">>年份": 1923}) platform.create_node(">>彷徨", ">>作品", {">>年份": 1926}) platform.create_node(">>骆驼祥子", ">>作品", {">>年份": 1937}) platform.create_node(">>绍兴", ">>地点") platform.create_node(">>北京", ">>地点") platform.create_node(">>浙江省", ">>地点") platform.create_node(">>中国", ">>地点") # 创建关系 platform.create_edge(">>鲁迅", ">>创作", ">>呐喊") platform.create_edge(">>鲁迅", ">>创作", ">>彷徨") platform.create_edge(">>鲁迅", ">>出生地", ">>绍兴") platform.create_edge(">>老舍", ">>创作", ">>骆驼祥子") platform.create_edge(">>老舍", ">>出生地", ">>北京") platform.create_edge(">>绍兴", ">>位于", ">>浙江省") platform.create_edge(">>浙江省", ">>位于", ">>中国") platform.create_edge(">>北京", ">>位于", ">>中国") # 信息抽取测试 print(">>=== 信息抽取 ===") text = ">>鲁迅出生于绍兴，创作了呐喊" extracted = platform.extract_from_text(text) print(f" 文本: {text}") print(f" 实体: {extracted['entities']}") print(f" 关系: {extracted['relations']}") # 推理 print(" === 知识推理 ===") platform.add_rule([(">>?", ">>出生地", ">>?")], (">>?", ">>籍贯", ">>?")) new = platform.reason() print(f" 推理出新事实: {new}条") # 问答 print(" === 智能问答 ===") for q in [">>鲁迅创作了什么？", ">>鲁迅是哪里人？", ">>介绍老舍"]: print(f" Q: {q} → A: {platform.answer_question(q)}") # 统计 print(" === 平台统计 ===") print(f" 节点: {len(platform.nodes)}") print(f" 边: {len(platform.edges)}") print(f" 事实: {len(platform.facts)}")