确定性推理

"""自然演绎：前向链（反复应用假言推理），并记录每条结论的证明链"""

class ProofEngine:
    def __init__(self):
        self.facts = {}                 # 命题 -> 证明依据
        self.rules = []                 # (前提集合, 结论, 规则名)

    def add_fact(self, p):
        self.facts.setdefault(p, ("公理", []))

    def add_rule(self, premises, conclusion, name):
        self.rules.append((set(premises), conclusion, name))

    def forward(self, target):
        """返回能否推出 target；推理过程中保留证明树。"""
        changed = True
        while changed:
            changed = False
            for premises, concl, name in self.rules:
                if concl in self.facts:
                    continue
                if premises.issubset(self.facts):       # 前件全部已知为真
                    self.facts[concl] = (name, sorted(premises))
                    changed = True
                    print(f"应用 {name}：由 {sorted(premises)} 推出 「{concl}」")
        return target in self.facts

    def explain(self, p, depth=0):
        """递归打印证明树。"""
        if p not in self.facts:
            return
        reason, deps = self.facts[p]
        print("  " * depth + f"└─ {p}   [{reason}]")
        for d in deps:
            self.explain(d, depth + 1)


if __name__ == "__main__":
    eng = ProofEngine()

    # 事实
    for f in ["苏格拉底是人", "下雨"]:
        eng.add_fact(f)

    # 蕴含规则（前提 -> 结论）
    eng.add_rule(["苏格拉底是人"], "苏格拉底会死", "假言推理:凡人皆有死")
    eng.add_rule(["下雨"], "天空有云", "规则:下雨则天阴")
    eng.add_rule(["天空有云"], "可能需要带伞", "规则:阴天备伞")
    eng.add_rule(["苏格拉底会死"], "苏格拉底不是神", "规则:神不死")

    print("=== 前向推理过程 ===")
    ok = eng.forward("苏格拉底不是神")
    print("\n能否证明'苏格拉底不是神'：", ok)

    print("\n=== 证明树 ===")
    eng.explain("苏格拉底不是神")

"""一阶谓词项的表示与最一般合一算法（MGU）

项的表示约定：
  变量：以 '?' 开头的字符串，如 '?x'
  常量：普通字符串，如 'a'、'john'
  函数项：('f', ['?x', 'a'])  元组 (函数名, 项列表)
"""

def is_var(t):
    return isinstance(t, str) and t.startswith("?")

def walk(t, subst):
    """沿置换链解析变量的当前绑定。"""
    while is_var(t) and t in subst:
        t = subst[t]
    return t

def occur_check(var, term, subst):
    """var 是否出现在 term 中（防止循环置换）。"""
    term = walk(term, subst)
    if term == var:
        return True
    if isinstance(term, tuple):
        return any(occur_check(var, arg, subst) for arg in term[1])
    return False

def extend(var, term, subst):
    """把 var := term 加入置换（含 occur 检查）。"""
    if occur_check(var, term, subst):
        return None                       # 无法合一，如 ?x 与 f(?x)
    new_subst = dict(subst)
    new_subst[var] = term
    return new_subst

def unify(t1, t2, subst=None):
    """返回 t1 与 t2 的 mgu（字典）；不可合一返回 None。"""
    if subst is None:
        subst = {}
    t1, t2 = walk(t1, subst), walk(t2, subst)

    if t1 == t2:
        return subst
    if is_var(t1):
        return extend(t1, t2, subst)
    if is_var(t2):
        return extend(t2, t1, subst)
    if isinstance(t1, tuple) and isinstance(t2, tuple):
        if t1[0] != t2[0] or len(t1[1]) != len(t2[1]):
            return None                   # 函数符/参数个数不同
        for a, b in zip(t1[1], t2[1]):
            subst = unify(a, b, subst)
            if subst is None:
                return None
        return subst
    return None                           # 两个不同常量，不可合一

def apply_subst(term, subst):
    """把置换作用于一个项/原子。"""
    term = walk(term, subst)
    if is_var(term) or isinstance(term, str):
        return term
    if isinstance(term, tuple):
        return (term[0], [apply_subst(a, subst) for a in term[1]])
    return term


if __name__ == "__main__":
    # 原子 P(term...) 用 ('P', [项...]) 表示
    tests = [
        (("P", ["?x", "b"]), ("P", ["a", "?y"])),
        (("Q", ["?x", "?x"]), ("Q", ["a", "b"])),                 # 失败：?x 不能同时是 a 和 b
        (("R", ["?x"]), ("R", [("f", ["?y"])]),),
        (("R", ["?x"]), ("R", [("f", ["?x"])])),                  # 失败：occur check
        (("T", ["?x", ("f", ["?y"])]), ("T", ["a", ("f", ["b"])])),
    ]
    for t1, t2 in tests:
        mgu = unify(t1, t2)
        print(f"合一 {t1}  与  {t2}")
        if mgu is None:
            print("   => 不可合一\n")
        else:
            print(f"   mgu = {mgu}")
            print(f"   合一后 = {apply_subst(t1, mgu)}\n")

"""命题逻辑归结反演证明器
表达式语法：~P 非；P&Q 与；P|Q 或；P->Q 蕴含；P<->Q 等价
"""

# ---------- 1. 递归下降解析 ----------
def parse(expr):
    tokens = expr.replace("(", " ( ").replace(")", " ) ").split()
    pos = 0

    def parse_iff():
        nonlocal pos
        left = parse_imp()
        while pos < len(tokens) and tokens[pos] == "<->":
            pos += 1
            right = parse_imp()
            left = ("iff", left, right)
        return left

    def parse_imp():
        nonlocal pos
        left = parse_or()
        while pos < len(tokens) and tokens[pos] == "->":
            pos += 1
            right = parse_or()
            left = ("imp", left, right)
        return left

    def parse_or():
        nonlocal pos
        left = parse_and()
        while pos < len(tokens) and tokens[pos] == "|":
            pos += 1
            left = ("or", left, parse_and())
        return left

    def parse_and():
        nonlocal pos
        left = parse_not()
        while pos < len(tokens) and tokens[pos] == "&":
            pos += 1
            left = ("and", left, parse_not())
        return left

    def parse_not():
        nonlocal pos
        if tokens[pos] == "~":
            pos += 1
            return ("not", parse_not())
        if tokens[pos] == "(":
            pos += 1
            node = parse_iff()
            pos += 1   # ')'
            return node
        tok = tokens[pos]
        pos += 1
        return ("lit", tok)

    tree = parse_iff()
    assert pos == len(tokens), f"无法解析的记号：{tokens[pos:]}"
    return tree

# ---------- 2. 化为 CNF ----------
def eliminate(node):
    """消去 -> 与 <->。"""
    t = node[0]
    if t == "lit":
        return node
    if t == "not":
        return ("not", eliminate(node[1]))
    a, b = eliminate(node[1]), eliminate(node[2])
    if t == "imp":                       # A->B  ≡  ~A|B
        return ("or", ("not", a), b)
    if t == "iff":                       # A<->B ≡ (~A|B)&(~B|A)
        return ("and", ("or", ("not", a), b), ("or", ("not", b), a))
    return (t, a, b)

def nnf(node):
    """否定内移，得到否定范式。"""
    t = node[0]
    if t == "lit":
        return node
    if t == "not":
        sub = node[1]
        if sub[0] == "not":              # 双重否定
            return nnf(sub[1])
        if sub[0] == "and":              # ~(A&B) ≡ ~A|~B
            return ("or", nnf(("not", sub[1])), nnf(("not", sub[2])))
        if sub[0] == "or":               # ~(A|B) ≡ ~A&~B
            return ("and", nnf(("not", sub[1])), nnf(("not", sub[2])))
        return node
    return (t, nnf(node[1]), nnf(node[2])) if t in ("and", "or") else node

def distr(a, b):
    """分配律：A | (B & C) = (A | B) & (A | C)。"""
    if a[0] == "and":
        return ("and", distr(a[1], b), distr(a[2], b))
    if b[0] == "and":
        return ("and", distr(a, b[1]), distr(a, b[2]))
    return ("or", a, b)

def cnf(node):
    if node[0] == "lit" or node[0] == "not":
        return node
    a, b = cnf(node[1]), cnf(node[2])
    if node[0] == "and":
        return ("and", a, b)
    return distr(a, b)                   # or

def to_clauses(node):
    """CNF 树 -> 子句集合；子句用 frozenset 表示，
    文字：命题名为正，('~', 名) 为负。"""
    clauses = set()
    def walk(n):
        if n[0] == "and":
            walk(n[1]); walk(n[2])
        else:
            lits = set()
            def collect(o):
                if o[0] == "or":
                    collect(o[1]); collect(o[2])
                elif o[0] == "not":
                    lits.add(("~", o[1][1]))
                else:
                    lits.add(o[1])
            collect(n)
            clauses.add(frozenset(lits))
    walk(node)
    return clauses

# ---------- 3. 归结 ----------
def resolve(c1, c2):
    """两个子句的所有可能归结式。"""
    results = []
    for lit in c1:
        neg = ("~", lit[1]) if isinstance(lit, tuple) else ("~", lit)
        comp = lit[1] if isinstance(lit, tuple) else None
        # c1 中的正文字 lit，找 c2 中 ('~', lit)；反之亦然
        if isinstance(lit, tuple):       # 负文字
            if comp in c2:
                merged = (c1 - {lit}) | (c2 - {comp})
                results.append(frozenset(merged))
        else:                            # 正文字
            if neg in c2:
                merged = (c1 - {lit}) | (c2 - {neg})
                results.append(frozenset(merged))
    return results

def resolution_prove(kb_exprs, goal_expr):
    """归结反演：kb_exprs 前提列表，goal_expr 待证结论。"""
    clauses = set()
    for e in kb_exprs:
        clauses |= to_clauses(cnf(nnf(eliminate(parse(e)))))
    # 目标取否定
    clauses |= to_clauses(cnf(nnf(eliminate(parse("~(" + goal_expr + ")")))))

    print("子句集：")
    for c in sorted(clauses, key=str):
        print("   ", show_clause(c))

    new = set()
    while True:
        pairs = [(c1, c2) for c1 in clauses for c2 in clauses if c1 != c2]
        for c1, c2 in pairs:
            for r in resolve(c1, c2):
                if len(r) == 0:
                    print(f"归结 {show_clause(c1)} 与 {show_clause(c2)} => □ 空子句")
                    return True            # 矛盾！目标成立
                if not tautology(r):
                    new.add(r)
        if new <= clauses:
            return False                   # 不再产生新子句，目标不可证
        clauses |= new
        new = set()

def tautology(c):
    """重言式子句（同时含 P 与 ~P）可直接删除。"""
    for lit in c:
        if not isinstance(lit, tuple):
            if ("~", lit) in c:
                return True
    return False

def show_clause(c):
    if not c:
        return "□"
    return " ∨ ".join(l if not isinstance(l, tuple) else "¬" + l[1] for l in sorted(c, key=str))


if __name__ == "__main__":
    # 例题：若下雨则地湿 (R->W)；下雨了 (R)；证明 地是湿的 (W)
    print("【例1】R->W, R  ⊢ W ?")
    print("结论：", resolution_prove(["R->W", "R"], "W"), "(True 表示得证)\n")

    # 经典：(P->Q) & (Q->R) & P  ⊢ R
    print("【例2】P->Q, Q->R, P  ⊢ R ?")
    print("结论：", resolution_prove(["P->Q", "Q->R", "P"], "R"))

"""一阶谓词逻辑归结反演（子句已手工化为标准形式）
文字表示： ('+', ('谓词', [项...]))  正文字
          ('-', ('谓词', [项...]))  负文字
子句：frozenset(文字)；变量以 ?开头；函数项 ('f',[...])
"""
from unify import unify, apply_subst      # 复用 3.3 节的合一算法

def negate(lit):
    sign, atom = lit
    return ("-" if sign == "+" else "+", atom)

def resolve_fol(c1, c2):
    """返回两个谓词子句的所有归结式（已实例化）。"""
    results = []
    for l1 in c1:
        for l2 in c2:
            if l1[0] == l2[0]:
                continue                   # 同号不能互补
            mgu = unify(l1[1], l2[1])
            if mgu is None:
                continue
            rest1 = [apply_subst_lit(x, mgu) for x in c1 - {l1}]
            rest2 = [apply_subst_lit(x, mgu) for x in c2 - {l2}]
            resolvent = frozenset(rest1 + rest2)
            if not is_tautology(resolvent):
                results.append(resolvent)
    return results

def apply_subst_lit(lit, subst):
    sign, atom = lit
    return (sign, apply_subst(atom, subst))

def is_tautology(clause):
    for lit in clause:
        if negate(lit) in clause:
            return True
    return False

def rename(clause, k):
    """子句变量标准化改名：?x -> ?x_k，避免不同子句变量重名。"""
    def rn(term):
        if isinstance(term, str) and term.startswith("?"):
            return term + "_" + str(k)
        if isinstance(term, tuple):
            return (term[0], [rn(a) for a in term[1]])
        return term
    return frozenset((s, rn(atom)) for s, atom in clause)

def fol_prove(clauses):
    """对子句集（含目标之否定）做水平饱和归结，出现 □ 即不可满足。"""
    clauses = set(clauses)
    level = 0
    while True:
        new = set()
        items = list(clauses)
        for i in range(len(items)):
            for j in range(i + 1, len(items)):
                c1 = rename(items[i], level)
                c2 = rename(items[j], level)
                for r in resolve_fol(c1, c2):
                    if len(r) == 0:
                        print(f"第 {level} 层归结得到空子句 □")
                        return True
                    if r not in clauses:
                        new.add(r)
        if new <= clauses:
            return False
        clauses |= new
        level += 1
        print(f"第 {level} 层：新增 {len(new)} 个子句，总计 {len(clauses)}")

def L(sign, pred, *terms):
    """快速构造文字。"""
    return (sign, (pred, list(terms)))

if __name__ == "__main__":
    # ===== 例1：苏格拉底三段论 =====
    # ∀x Human(x)->Mortal(x) 化为子句： ¬Human(?x) ∨ Mortal(?x)
    # Human(socrates)；目标否定：¬Mortal(socrates)
    kb = [
        frozenset([L("-", "Human", "?x"), L("+", "Mortal", "?x")]),
        frozenset([L("+", "Human", "socrates")]),
        frozenset([L("-", "Mortal", "socrates")]),       # 目标之否定
    ]
    print("【苏格拉底会死？】", fol_prove(kb), "\n")

    # ===== 例2：家庭关系 =====
    # ∀x,y Parent(x,y)-> Ancestor(x,y)
    # ∀x,y,z Parent(x,z) ∧ Ancestor(z,y) -> Ancestor(x,y)
    # 事实：Parent(bob,anna), Parent(anna,tom)
    # 目标：Ancestor(bob,tom)，其否定加入
    kb2 = [
        frozenset([L("-", "Parent", "?x", "?y"), L("+", "Ancestor", "?x", "?y")]),
        frozenset([L("-", "Parent", "?x", "?z"),
                   L("-", "Ancestor", "?z", "?y"),
                   L("+", "Ancestor", "?x", "?y")]),
        frozenset([L("+", "Parent", "bob", "anna")]),
        frozenset([L("+", "Parent", "anna", "tom")]),
        frozenset([L("-", "Ancestor", "bob", "tom")]),
    ]
    print("【Bob 是 Tom 的祖先？】", fol_prove(kb2))

"""归结反演求取答案：ANSWER 谓词记录变量绑定"""
from fol_resolution import L, rename, is_tautology
from unify import unify, apply_subst

def resolve_with_trace(c1, c2):
    results = []
    for l1 in c1:
        for l2 in c2:
            if l1[0] == l2[0] and l1[1][0] != "ANSWER":
                continue
            if l1[0] != l2[0]:
                mgu = unify(l1[1], l2[1])
                if mgu is None:
                    continue
                rest1 = [(s, apply_subst(a, mgu)) for s, a in c1 - {l1}]
                rest2 = [(s, apply_subst(a, mgu)) for s, a in c2 - {l2}]
                r = frozenset(rest1 + rest2)
                if not is_tautology(r):
                    results.append(r)
    return results

def fol_query(clauses, answer_lit, max_level=8):
    """answer_lit 形如 ANSWER(?y) 的正文字。"""
    clauses = set(clauses)
    for level in range(max_level):
        new = set()
        items = list(clauses)
        for i in range(len(items)):
            for j in range(i + 1, len(items)):
                for r in resolve_with_trace(rename(items[i], level),
                                            rename(items[j], level)):
                    only_answer = all(lit[1][0] == "ANSWER" for lit in r)
                    if only_answer and r:
                        return r          # 只剩 ANSWER 文字，读出答案
                    new.add(r)
        if new <= clauses:
            return None
        clauses |= new
    return None

if __name__ == "__main__":
    # 求：Bob 的孙辈是谁？（即 Parent(bob,?z) 且 Parent(?z,?y)）
    kb = [
        frozenset([L("-", "Parent", "?x", "?y"), L("+", "Ancestor", "?x", "?y")]),
        frozenset([L("-", "Parent", "?x", "?z"),
                   L("-", "Ancestor", "?z", "?y"),
                   L("+", "Ancestor", "?x", "?y")]),
        frozenset([L("+", "Parent", "bob", "anna")]),
        frozenset([L("+", "Parent", "anna", "tom")]),
        frozenset([L("+", "Parent", "anna", "lily")]),
        # 目标重言式：¬Ancestor(bob,?y) ∨ ANSWER(?y)
        frozenset([L("-", "Ancestor", "bob", "?y"),
                   ("+", ("ANSWER", ["?y"]))]),
    ]
    ans = fol_query(kb, None)
    if ans:
        for lit in ans:
            print("答案：Bob 的后代包括 =", lit[1][1])
    else:
        print("无解")