📜 第16课：规则推理

from collections import defaultdict from typing import List, Tuple, Set, Callable class Rule: """推理规则""" def __init__(self, name, premises, conclusion, confidence=1.0): """ premises: [(s_var, p, o_var), ...] 变量用?开头 conclusion: (s_var, p, o_var) """ self.name = name self.premises = premises self.conclusion = conclusion self.confidence = confidence def get_variables(self): """提取规则中的变量""" variables = set() for s, p, o in self.premises + [self.conclusion]: if s.startswith("?"): variables.add(s) if o.startswith("?"): variables.add(o) return variables class ForwardReasoner: ">>>前向推理引擎""" def __init__(self): self.facts = set() # {(s, p, o)} self.rules = [] # [Rule] self.derived = {} # {(s,p,o): [规则名]} self.spo_index = defaultdict(set) def add_fact(self, s, p, o, source="asserted"): triple = (s, p, o) if triple not in self.facts: self.facts.add(triple) self.spo_index[(s, p)].add(o) self.spo_index[(None, p)].add((s, o)) self.spo_index[(s, None)].add((p, o)) self.derived.setdefault(triple, []).append(source) return True return False def add_rule(self, rule): self.rules.append(rule) def _match_premise(self, premise, bindings_list): """匹配单个前提条件""" s_var, p, o_var = premise new_bindings = [] for bindings in bindings_list: s = bindings.get(s_var, s_var) if s_var.startswith("?") else s_var o = bindings.get(o_var, o_var) if o_var.startswith("?") else o_var for fact in self.facts: fs, fp, fo = fact if fp != p: continue new_binding = dict(bindings) ok = True if s_var.startswith("?"): if s_var in new_binding: if new_binding[s_var] != fs: ok = False else: new_binding[s_var] = fs elif s != fs: ok = False if o_var.startswith("?"): if o_var in new_binding: if new_binding[o_var] != fo: ok = False else: new_binding[o_var] = fo elif o != fo: ok = False if ok: new_bindings.append(new_binding) return new_bindings def _apply_rule(self, rule): """应用单条规则，返回新推导的事实""" bindings_list = [{}] for premise in rule.premises: bindings_list = self._match_premise(premise, bindings_list) if not bindings_list: return [] new_facts = [] s_var, p, o_var = rule.conclusion for bindings in bindings_list: s = bindings.get(s_var, s_var) if s_var.startswith("?") else s_var o = bindings.get(o_var, o_var) if o_var.startswith("?") else o_var new_facts.append((s, p, o, rule.name)) return new_facts def reason(self, max_iterations=100): """执行前向推理（不动点迭代）""" iteration = 0 total_new = 0 while iteration < max_iterations: new_facts = [] for rule in self.rules: results = self._apply_rule(rule) new_facts.extend(results) added = 0 for s, p, o, rule_name in new_facts: if self.add_fact(s, p, o, rule_name): added += 1 total_new += added iteration += 1 if added == 0: break return {"iterations": iteration, "new_facts": total_new, "total_facts": len(self.facts)} def explain(self, s, p, o): """解释某事实的推理过程""" triple = (s, p, o) if triple not in self.derived: return f"{triple}: 原始断言，无推理过程" sources = self.derived[triple] if sources == ["asserted"]: return f"{triple}: 原始断言" return f"{triple}: 由规则 {sources} 推导" # ========== 构建推理系统 ========== reasoner = ForwardReasoner() # 添加基础事实 reasoner.add_fact("鲁迅", "出生地", "绍兴") reasoner.add_fact("绍兴", "属于", "浙江省") reasoner.add_fact("浙江省", "属于", "中国") reasoner.add_fact("老舍", "出生地", "北京") reasoner.add_fact("北京", "属于", "中国") reasoner.add_fact("鲁迅", "创作", "呐喊") reasoner.add_fact("鲁迅", "创作", "彷徨") reasoner.add_fact("呐喊", "类型", "短篇小说集") reasoner.add_fact("彷徨", "类型", "短篇小说集") # 添加推理规则 reasoner.add_rule(Rule( "国籍推理", [("?person", "出生地", "?city"), ("?city", "属于", "?province"), ("?province", "属于", "?country")], ("?person", "国籍", "?country") )) reasoner.add_rule(Rule( "直接国籍推理", [("?person", "出生地", "?city"), ("?city", "属于", "?country")], ("?person", "国籍", "?country") )) reasoner.add_rule(Rule( ">>>作者类型推理", [("?person", "创作", "?work1"), ("?person", "创作", "?work2"), ("?work1", "类型", "?type"), ("?work2", "类型", "?type")], ("?person", "擅长", "?type") )) # 执行推理 print("=== 前向推理 ===") result = reasoner.reason() print(f" 迭代次数: {result['iterations']}") print(f" 新增事实: {result['new_facts']}") print(f" 总事实数: {result['total_facts']}") print(" === 所有事实 ===") for fact in sorted(reasoner.facts): print(f" {fact}") print(" === 推理解释 ===") print(reasoner.explain("鲁迅", "国籍", "中国")) print(reasoner.explain("老舍", "国籍", "中国")) print(reasoner.explain("鲁迅", "擅长", "短篇小说集"))

class BackwardReasoner: ">>>后向推理引擎""" def __init__(self, forward_reasoner): self.fr = forward_reasoner def prove(self, goal, depth=0, max_depth=10): """尝试证明目标 (s, p, o)，返回证据树""" if depth > max_depth: return {"goal": goal, "status": "MAX_DEPTH"} s, p, o = goal # 直接检查是否已知事实 if goal in self.fr.facts: return {"goal": goal, "status": "ASSERTED"} # 尝试通过规则推导 for rule in self.fr.rules: # 检查规则结论是否匹配目标 cs, cp, co = rule.conclusion if cp != p: continue bindings = {} if cs.startswith("?"): bindings[cs] = s elif cs != s: continue if co.startswith("?"): bindings[co] = o elif co != o: continue # 递归证明所有前提 sub_goals = [] all_proved = True for ps, pp, po in rule.premises: actual_s = bindings.get(ps, ps) if ps.startswith("?") else ps actual_o = bindings.get(po, po) if po.startswith("?") else po sub_goal = (actual_s, pp, actual_o) proof = self.prove(sub_goal, depth + 1, max_depth) if proof["status"] in ("ASSERTED", "PROVED"): sub_goals.append(proof) else: all_proved = False break if all_proved: return {"goal": goal, "status": "PROVED", "rule": rule.name, "proofs": sub_goals} return {"goal": goal, "status": "FAILED"} # 后向推理测试 br = BackwardReasoner(reasoner) print("=== 后向推理 ===") print("证明: 鲁迅国籍=中国?") result = br.prove(("鲁迅", "国籍", "中国")) print(f" 状态: {result['status']}") if "rule" in result: print(f" 使用规则: {result['rule']}") print(" 证明: 鲁迅擅长=短篇小说集?") result2 = br.prove(("鲁迅", "擅长", "短篇小说集")) print(f" 状态: {result2['status']}")