🔗 第2课：RDF与三元组

# 前缀声明（类似命名空间的缩写） @prefix ex: <http://example.org/> . @prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> . @prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> . @prefix xsd: <http://www.w3.org/2001/XMLSchema#> . # 基本三元组 ex:鲁迅 ex:出生地 ex:绍兴 ; ex:创作 ex:呐喊 , ex:彷徨 ; ex:原名 "周树人" ; ex:出生年份 "1881"^^xsd:gYear . # 类型声明 ex:鲁迅 rdf:type ex:作家 . ex:呐喊 rdf:type ex:文学作品 . # RDFS标签 ex:作家 rdfs:label "作家" . ex:出生地 rdfs:label "出生地" . ex:绍兴 rdfs:label "绍兴"@zh .

from urllib.parse import quote import json class RDFNode: """RDF节点：URI、空白节点或字面量""" def __init__(self, value, node_type="uri", datatype=None, lang=None): self.value = value self.node_type = node_type # "uri", "bnode", "literal" self.datatype = datatype self.lang = lang def __repr__(self): if self.node_type == "uri": return f"<{self.value}>" elif self.node_type == "bnode": return f"_:{self.value}" else: # literal suffix = "" if self.lang: suffix = f"@{self.lang}" elif self.datatype: suffix = f"^^<{self.datatype}>" return f'"{self.value}"{suffix}' def __eq__(self, other): return isinstance(other, RDFNode) and self.value == other.value and self.node_type == other.node_type def __hash__(self): return hash((self.value, self.node_type)) class RDFTriple: """RDF三元组""" def __init__(self, subject, predicate, obj): assert subject.node_type in ("uri", "bnode"), "主体必须是URI或空白节点" assert predicate.node_type == "uri", "谓词必须是URI" self.subject = subject self.predicate = predicate self.object = obj def __repr__(self): return f"{self.subject} {self.predicate} {self.object} ." def __eq__(self, other): return (isinstance(other, RDFTriple) and self.subject == other.subject and self.predicate == other.predicate and self.object == other.object) def __hash__(self): return hash((self.subject, self.predicate, self.object)) class RDFGraph: """RDF图 - 三元组集合""" def __init__(self): self.triples = set() self.prefixes = {} self.bnode_counter = 0 def add_prefix(self, short, long): self.prefixes[short] = long def uri(self, prefix, local): """用前缀创建URI节点""" return RDFNode(self.prefixes.get(prefix, prefix) + local) def literal(self, value, datatype=None, lang=None): """创建字面量节点""" return RDFNode(value, "literal", datatype, lang) def bnode(self): """创建空白节点""" self.bnode_counter += 1 return RDFNode(f"b{self.bnode_counter}", "bnode") def add(self, s, p, o): """添加三元组""" triple = RDFTriple(s, p, o) self.triples.add(triple) return triple def remove(self, s, p, o): """移除三元组""" self.triples.discard(RDFTriple(s, p, o)) def match(self, s=None, p=None, o=None): """模式匹配查询（None为通配符）""" results = [] for t in self.triples: if (s is None or t.subject == s) and \ (p is None or t.predicate == p) and \ (o is None or t.object == o): results.append(t) return results def to_turtle(self): """导出为Turtle格式""" lines = [] for prefix, uri in self.prefixes.items(): lines.append(f"@prefix {prefix}: <{uri}> .") lines.append("") for t in sorted(self.triples, key=lambda x: repr(x)): lines.append(repr(t)) return "\n".join(lines) def to_jsonld(self): """导出为JSON-LD格式""" context = {k: v for k, v in self.prefixes.items()} graph = [] for t in self.triples: entry = { "@id": t.subject.value, t.predicate.value: t.object.value } graph.append(entry) return json.dumps({"@context": context, "@graph": graph}, ensure_ascii=False, indent=2) def size(self): return len(self.triples) # ========== 构建RDF知识图谱 ========== g = RDFGraph() g.add_prefix("ex", "http://example.org/") g.add_prefix("rdf", "http://www.w3.org/1999/02/22-rdf-syntax-ns#") g.add_prefix("rdfs", "http://www.w3.org/2000/01/rdf-schema#") g.add_prefix("xsd", "http://www.w3.org/2001/XMLSchema#") # 类型声明 g.add(g.uri("ex", "鲁迅"), g.uri("rdf", "type"), g.uri("ex", "作家")) g.add(g.uri("ex", "老舍"), g.uri("rdf", "type"), g.uri("ex", "作家")) g.add(g.uri("ex", "呐喊"), g.uri("rdf", "type"), g.uri("ex", "小说")) # 关系 g.add(g.uri("ex", "鲁迅"), g.uri("ex", "创作"), g.uri("ex", "呐喊")) g.add(g.uri("ex", "鲁迅"), g.uri("ex", "出生地"), g.uri("ex", "绍兴")) # 字面量属性 g.add(g.uri("ex", "鲁迅"), g.uri("ex", "原名"), g.literal("周树人", lang="zh")) g.add(g.uri("ex", "鲁迅"), g.uri("ex", "出生年份"), g.literal("1881", datatype="http://www.w3.org/2001/XMLSchema#gYear")) g.add(g.uri("ex", "呐喊"), g.uri("ex", "出版年份"), g.literal("1923", datatype="http://www.w3.org/2001/XMLSchema#gYear")) # 空白节点示例（代表复杂结构） address = g.bnode() g.add(g.uri("ex", "绍兴"), g.uri("ex", "位于"), address) g.add(address, g.uri("ex", "省份"), g.literal("浙江省")) g.add(address, g.uri("ex", "国家"), g.literal("中国")) print("=== RDF图统计 ===") print(f"三元组数: {g.size()}") print("\n=== 模式匹配查询 ===") # 查询所有类型声明 type_triples = g.match(p=g.uri("rdf", "type")) for t in type_triples: print(f" {t}") print("\n=== 鲁迅相关查询 ===") luxun = g.uri("ex", "鲁迅") for t in g.match(s=luxun): print(f" {t}") print("\n=== Turtle序列化 ===") print(g.to_turtle()[:500])

class SimpleSPARQL: """简易SPARQL引擎 - 支持基本SELECT查询""" def __init__(self, graph): self.graph = graph def _match_pattern(self, pattern, bindings=None): """匹配单个三元组模式，返回变量绑定列表""" if bindings is None: bindings = [{}] results = [] s_pat, p_pat, o_pat = pattern for binding in bindings: # 解析模式中的变量或常量 s = binding.get(s_pat, s_pat) if isinstance(s_pat, str) and s_pat.startswith("?") else s_pat p = binding.get(p_pat, p_pat) if isinstance(p_pat, str) and p_pat.startswith("?") else p_pat o = binding.get(o_pat, o_pat) if isinstance(o_pat, str) and o_pat.startswith("?") else o_pat # 在图中查找匹配 for triple in self.graph.match( s=None if isinstance(s, str) and s.startswith("?") else s, p=None if isinstance(p, str) and p.startswith("?") else p, o=None if isinstance(o, str) and o.startswith("?") else o ): new_binding = dict(binding) ok = True if isinstance(s, str) and s.startswith("?"): if s in new_binding and new_binding[s] != triple.subject: ok = False else: new_binding[s] = triple.subject if isinstance(p, str) and p.startswith("?"): if p in new_binding and new_binding[p] != triple.predicate: ok = False else: new_binding[p] = triple.predicate if isinstance(o, str) and o.startswith("?"): if o in new_binding and new_binding[o] != triple.object: ok = False else: new_binding[o] = triple.object if ok: results.append(new_binding) return results def select(self, variables, patterns): """SPARQL SELECT查询 variables: 需要返回的变量列表如 ["?s", "?o"] patterns: 三元组模式列表如 [("?s", "ex:创作", "?o")] """ bindings = None for pattern in patterns: bindings = self._match_pattern(pattern, bindings) if not bindings: return [] results = [] for b in bindings: row = {v: b.get(v, "未绑定") for v in variables} results.append(row) return results # 创建SPARQL引擎 sparql = SimpleSPARQL(g) # 查询1：所有作家 print("=== 查询：所有作家 ===") rdf_type = g.uri("rdf", "type") ex_writer = g.uri("ex", "作家") results = sparql.select(["?s"], [("?s", rdf_type, ex_writer)]) for r in results: print(f" 作家: {r['?s']}") # 查询2：鲁迅创作的所有作品 print("\n=== 查询：鲁迅的作品 ===") ex_create = g.uri("ex", "创作") luxun = g.uri("ex", "鲁迅") results = sparql.select(["?work"], [(luxun, ex_create, "?work")]) for r in results: print(f" 作品: {r['?work']}") # 查询3：两跳查询 - 哪些作品与绍兴有关 print("\n=== 查询：与绍兴相关的作品（两跳） ===") ex_birthplace = g.uri("ex", "出生地") shaoxing = g.uri("ex", "绍兴") results = sparql.select(["?writer", "?work"], [ ("?writer", ex_birthplace, shaoxing), ("?writer", ex_create, "?work") ]) for r in results: print(f" {r['?writer']} 创作 {r['?work']}")

# RDFS推理演示 class RDFSReasoner: """简单的RDFS推理机""" def __init__(self, graph): self.graph = graph self.inferred = set() def apply_domain_range(self): """应用domain和range推理规则""" rdf_type = self.graph.uri("rdf", "type") rdfs_domain = self.graph.uri("rdfs", "domain") rdfs_range = self.graph.uri("rdfs", "range") # 收集所有domain/range声明 domains = {} ranges = {} for t in self.graph.match(p=rdfs_domain): domains[t.subject.value] = t.object for t in self.graph.match(p=rdfs_range): ranges[t.subject.value] = t.object # 对每个三元组，如果谓词有domain，则主体属于domain类 for t in self.graph.triples: if t.predicate.value in domains: new_triple = RDFTriple(t.subject, rdf_type, domains[t.predicate.value]) if new_triple not in self.graph.triples: self.inferred.add(new_triple) print(f" 推理(domain): {t.subject} → rdf:type → {domains[t.predicate.value]}") if t.predicate.value in ranges: new_triple = RDFTriple(t.object, rdf_type, ranges[t.predicate.value]) if new_triple not in self.graph.triples: self.inferred.add(new_triple) print(f" 推理(range): {t.object} → rdf:type → {ranges[t.predicate.value]}") def apply_subclass(self): """应用subClassOf传递推理""" rdf_type = self.graph.uri("rdf", "type") rdfs_subclass = self.graph.uri("rdfs", "subClassOf") # 收集subClassOf关系 subclass_map = {} for t in self.graph.match(p=rdfs_subclass): subclass_map[t.subject.value] = t.object.value # 如果A subClassOf B，则A的实例也是B的实例 for t in self.graph.match(p=rdf_type): cls_value = t.object.value if cls_value in subclass_map: parent = self.graph.uri("ex", subclass_map[cls_value].split("/")[-1]) new_triple = RDFTriple(t.subject, rdf_type, parent) if new_triple not in self.graph.triples and new_triple not in self.inferred: self.inferred.add(new_triple) print(f" 推理(subClassOf): {t.subject} → rdf:type → {parent}") # 添加RDFS声明 g.add(g.uri("ex", "创作"), g.uri("rdfs", "domain"), g.uri("ex", "作家")) g.add(g.uri("ex", "创作"), g.uri("rdfs", "range"), g.uri("ex", "作品")) g.add(g.uri("ex", "小说家"), g.uri("rdfs", "subClassOf"), g.uri("ex", "作家")) g.add(g.uri("ex", "呐喊"), g.uri("ex", "类型"), g.literal("短篇小说集")) print("=== RDFS推理 ===") reasoner = RDFSReasoner(g) reasoner.apply_domain_range() reasoner.apply_subclass() print(f"\n推理出 {len(reasoner.inferred)} 条新知识")

# 使用RDFlib操作RDF（如果安装了rdflib） try: from rdflib import Graph, Namespace, Literal, RDF, RDFS, XSD g2 = Graph() EX = Namespace("http://example.org/") g2.bind("ex", EX) # 添加三元组 g2.add((EX["鲁迅"], EX["创作"], EX["呐喊"])) g2.add((EX["鲁迅"], RDF.type, EX["作家"])) g2.add((EX["鲁迅"], EX["原名"], Literal("周树人", lang="zh"))) g2.add((EX["鲁迅"], EX["出生年份"], Literal(1881, datatype=XSD.gYear))) print("=== RDFlib操作 ===") print(f"三元组数: {len(g2)}") print("\nTurtle序列化:") print(g2.serialize(format="turtle")[:300]) # SPARQL查询 q = """ SELECT ?s ?o WHERE { ?s ex:创作 ?o . } """ print("\nSPARQL查询结果:") for row in g2.query(q): print(f" {row.s} → {row.o}") except ImportError: print("RDFlib未安装，可运行: pip install rdflib") print("本课的自定义实现已完整演示RDF核心概念")