知识表示

"""微型一阶谓词知识库：事实 + 规则 + 反向链查询"""

class Atom:
    """原子谓词，如 Human(?x)、Parent(bob, anna)。"""
    def __init__(self, pred, *terms):
        self.pred = pred
        self.terms = list(terms)

    @staticmethod
    def parse(expr):
        """把字符串 'Human(?x)' 解析成 Atom 对象。"""
        pred, rest = expr.split("(", 1)
        terms = rest.rstrip(")").split(",")
        return Atom(pred.strip(), *[t.strip() for t in terms])

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

    def match(self, fact, subst):
        """将本模式（可含变量）与事实（不含变量）在已有置换 subst 下匹配。"""
        if self.pred != fact.pred or len(self.terms) != len(fact.terms):
            return None
        for pat_t, fact_t in zip(self.terms, fact.terms):
            pat_t = subst.get(pat_t, pat_t) if self.is_var(pat_t) else pat_t
            if self.is_var(pat_t):
                subst = dict(subst)
                subst[pat_t] = fact_t          # 变量绑定
            elif pat_t != fact_t:
                return None
        return subst

    def substitute(self, subst):
        terms = [subst.get(t, t) if self.is_var(t) else t for t in self.terms]
        return Atom(self.pred, *terms)

    def __repr__(self):
        return f"{self.pred}({', '.join(self.terms)})"


class KnowledgeBase:
    def __init__(self):
        self.facts = []          # 事实：Atom 列表
        self.rules = []          # 规则：(前提原子列表, 结论原子)

    def add_fact(self, expr):
        self.facts.append(Atom.parse(expr))

    def add_rule(self, head, *body):
        """head 为结论，body 为若干前提，例如
        add_rule('Mortal(?x)', 'Human(?x)')  表示  Human(x) → Mortal(x)"""
        self.rules.append((Atom.parse(head), [Atom.parse(b) for b in body]))

    def query(self, goal_expr):
        """查询目标，返回所有使目标成立的变量绑定（反向链）。"""
        goal = Atom.parse(goal_expr)
        answers = []

        def prove(goals, subst):
            if not goals:                          # 所有子目标都已证明
                answers.append(subst)
                return
            first, rest = goals[0], goals[1:]
            first = first.substitute(subst)
            # 路径 A：与已知事实匹配
            for fact in self.facts:
                new_subst = first.match(fact, subst)
                if new_subst is not None:
                    prove(rest, new_subst)
            # 路径 B：使用规则，把目标替换成规则的前提
            for head, body in self.rules:
                new_subst = first.match(head, subst)
                if new_subst is not None:
                    new_body = [a.substitute(new_subst) for a in body]
                    prove(new_body + [a.substitute(new_subst) for a in rest], new_subst)

        prove([goal], {})
        return answers


if __name__ == "__main__":
    kb = KnowledgeBase()

    # ---- 事实：家庭关系 ----
    for f in [
        "Human(socrates)", "Human(bob)", "Human(anna)", "Human(tom)",
        "Parent(bob, anna)", "Parent(anna, tom)",
    ]:
        kb.add_fact(f)

    # ---- 规则 ----
    kb.add_rule("Mortal(?x)", "Human(?x)")          # 人都会死
    kb.add_rule("Ancestor(?x, ?y)", "Parent(?x, ?y)")
    kb.add_rule("Ancestor(?x, ?y)", "Parent(?x, ?z)", "Ancestor(?z, ?y)")

    # ---- 查询 ----
    print("会死者：", kb.query("Mortal(?who)"))
    print("Bob 的孩子：", kb.query("Parent(bob, ?c)"))
    print("Tom 的祖先：", kb.query("Ancestor(?a, tom)"))

"""产生式系统：综合数据库 + 规则库 + 正向推理机"""

class Rule:
    def __init__(self, name, conditions, conclusion, desc=""):
        self.name = name
        self.conditions = set(conditions)   # 前件：需要同时成立的事实集合
        self.conclusion = conclusion        # 后件：推出的新事实
        self.desc = desc

    def __repr__(self):
        return f"<{self.name}: IF {sorted(self.conditions)} THEN '{self.conclusion}'>"


class ProductionSystem:
    def __init__(self):
        self.facts = set()          # 综合数据库
        self.rules = []             # 规则库

    def add_fact(self, fact):
        self.facts.add(fact)

    def add_rule(self, rule):
        self.rules.append(rule)

    def forward_chaining(self, trace=True):
        """正向推理：从已知事实出发，反复触发规则，直到不再产生新事实。"""
        fired = set()               # 已触发过的规则，避免重复
        log = []
        changed = True

        while changed:
            changed = False
            conflict_set = []       # 冲突集：当前所有可触发的规则

            for rule in self.rules:
                if rule.name in fired:
                    continue
                if rule.conditions.issubset(self.facts):   # 前件全部满足
                    conflict_set.append(rule)

            if conflict_set:
                # 冲突消解策略：规则按名称排序（也可按优先级、专一性等）
                rule = sorted(conflict_set, key=lambda r: r.name)[0]
                self.facts.add(rule.conclusion)
                fired.add(rule.name)
                changed = True
                msg = f"触发 {rule.name}：{rule.desc} => 新事实「{rule.conclusion}」"
                log.append(msg)
                if trace:
                    print(msg)

        return log


if __name__ == "__main__":
    # 经典动物识别规则库（第八章还会扩展为完整专家系统）
    ps = ProductionSystem()
    for f in ["有毛发", "吃肉", "黄褐色", "有黑色条纹"]:
        ps.add_fact(f)

    ps.add_rule(Rule("R1", ["有毛发"], "是哺乳动物", "有毛发→哺乳动物"))
    ps.add_rule(Rule("R2", ["有奶"], "是哺乳动物", "有奶→哺乳动物"))
    ps.add_rule(Rule("R3", ["是哺乳动物", "吃肉"], "是食肉动物", "哺乳动物且吃肉→食肉动物"))
    ps.add_rule(Rule("R4", ["是哺乳动物", "有蹄"], "是有蹄类动物", "哺乳动物且有蹄→有蹄类"))
    ps.add_rule(Rule("R5", ["是食肉动物", "黄褐色", "有黑色条纹"], "是老虎",
                     "食肉+黄褐色+黑条纹→老虎"))
    ps.add_rule(Rule("R6", ["是食肉动物", "黄褐色", "有暗斑点"], "是金钱豹",
                     "食肉+黄褐色+暗斑点→金钱豹"))

    print("初始事实：", sorted(ps.facts))
    print("-" * 50)
    ps.forward_chaining()
    print("-" * 50)
    print("最终结论：", "老虎" if "是老虎" in ps.facts else "未能识别")

"""框架表示法：槽/侧面、默认值、过程性附件、继承"""

class Frame:
    def __init__(self, name):
        self.name = name
        self.slots = {}      # 槽名 -> {侧面名: 值}

    def add_slot(self, slot, value=None, default=None,
                 if_needed=None, if_added=None):
        self.slots[slot] = {
            "value": value,
            "default": default,
            "if_needed": if_needed,    # 缺值时调用的函数(frame, slot)
            "if_added": if_added,      # 写入值时触发的函数(frame, slot, val)
        }

    def get(self, slot, seen=None):
        """取值顺序：本框架值 → 默认值 → if_needed 过程 → 沿 AKO 继承"""
        seen = seen or set()
        if self.name in seen:
            return None
        seen.add(self.name)

        s = self.slots.get(slot)
        if s is not None:
            if s["value"] is not None:
                return s["value"]
            if s["default"] is not None:
                return s["default"]
            if s["if_needed"] is not None:
                return s["if_needed"](self, slot)

        # 沿 ISA / AKO 到父框架中继承
        for parent_slot in ("AKO", "ISA"):
            if parent_slot in self.slots and self.slots[parent_slot]["value"]:
                parent = FrameSystem.get_frame(self.slots[parent_slot]["value"])
                val = parent.get(slot, seen)
                if val is not None:
                    return val
        return None

    def setv(self, slot, value):
        s = self.slots.setdefault(slot, {"value": None, "default": None,
                                         "if_needed": None, "if_added": None})
        s["value"] = value
        if s["if_added"]:
            s["if_added"](self, slot, value)

    def show(self):
        print(f"【框架：{self.name}】")
        for slot, facets in self.slots.items():
            if slot in ("AKO", "ISA"):
                continue
            print(f"   {slot:　<6} = {self.get(slot)}")


class FrameSystem:
    _frames = {}

    @classmethod
    def get_frame(cls, name):
        return cls._frames.get(name)

    @classmethod
    def create(cls, name):
        f = Frame(name)
        cls._frames[name] = f
        return f


if __name__ == "__main__":
    # ---- 定义父框架：教职工 ----
    staff = FrameSystem.create("教职工")
    staff.add_slot("类别")
    staff.add_slot("工作单位", default="人工智能学院")
    staff.add_slot("健康状况", default="良好")

    # ---- 子框架：教师，AKO 继承自教职工 ----
    teacher = FrameSystem.create("教师")
    teacher.add_slot("AKO", value="教职工")
    teacher.add_slot("学历", default="博士研究生")
    teacher.add_slot("职称")
    # if_needed 过程性知识：没有填写职称时按年龄推断
    teacher.add_slot("退休年龄",
                     if_needed=lambda f, s: 65 if f.get("性别") == "男" else 60)

    # ---- 实例框架：具体某位教师 ----
    wang = FrameSystem.create("王老师")
    wang.add_slot("ISA", value="教师")
    wang.setv("姓名", "王知行")
    wang.setv("性别", "男")
    wang.setv("职称", "教授")
    wang.setv("所授课程", "人工智能导论")

    wang.show()
    print("\n—— 继承与推理效果 ——")
    print("工作单位（继承自教职工）：", wang.get("工作单位"))
    print("学历（继承自教师默认值）：", wang.get("学历"))
    print("退休年龄（if_needed 过程推算）：", wang.get("退休年龄"))

"""语义网络：有向图（三元组）+ 沿继承关系的属性推理"""

class SemanticNetwork:
    def __init__(self):
        self.triples = []          # [(主体, 关系, 客体)]
        # 可继承、指向"父概念"的关系
        self.isa_rels = {"ISA", "AKO", "subclass-of", "instance-of"}

    def add(self, subj, rel, obj):
        self.triples.append((subj, rel, obj))

    def direct(self, subj, rel):
        """直接关系查询。"""
        return [o for s, r, o in self.triples if s == subj and r == rel]

    def parents(self, node):
        """沿 ISA/AKO 一步可达的父概念。"""
        return [o for s, r, o in self.triples
                if s == node and r in self.isa_rels]

    def inheritable_props(self, node):
        """继承推理：收集节点自身及所有祖先概念的属性（have/can/属性...）。"""
        result = {}
        visited = set()
        stack = [node]
        while stack:
            cur = stack.pop()
            if cur in visited:
                continue
            visited.add(cur)
            for s, r, o in self.triples:
                if s == cur and r not in self.isa_rels and r != "part-of":
                    result.setdefault(r, []).append((o, cur))
            stack.extend(self.parents(cur))
        return result

    def all_nodes(self):
        return sorted({s for s, _, _ in self.triples} | {o for _, _, o in self.triples})

    def describe(self, node):
        print(f"「{node}」可知的属性（含继承）：")
        for rel, vals in self.inheritable_props(node).items():
            for val, source in vals:
                tag = "自身" if source == node else f"继承自「{source}」"
                print(f"   --{rel}--> {val}   （{tag}）")


if __name__ == "__main__":
    net = SemanticNetwork()
    # 概念层次
    net.add("动物", "AKO", "生物")
    net.add("鸟", "AKO", "动物")
    net.add("鱼", "AKO", "动物")
    net.add("燕子", "ISA", "鸟")
    net.add("麻雀", "ISA", "鸟")
    # 属性
    net.add("生物", "能", "新陈代谢")
    net.add("动物", "能", "运动")
    net.add("动物", "have", "心脏")
    net.add("鸟", "have", "翅膀")
    net.add("鸟", "能", "飞行")
    net.add("鱼", "have", "鳃")
    net.add("鱼", "能", "游泳")
    net.add("燕子", "颜色", "黑色")
    net.add("燕子", "build", "泥巢")

    net.describe("燕子")
    print()
    print("直接会飞的：", net.direct("鸟", "能"))
    print("燕子的父概念链：", net.parents("燕子"))

"""用面向对象方式表示'动物分类学'知识"""

class Animal:
    """动物基类：所有动物共享的知识。"""
    kingdom = "动物界"

    def __init__(self, name, age):
        self.name = name
        self.age = age

    def move(self):
        raise NotImplementedError      # 由具体子类实现（多态）

    def info(self):
        return f"{self.name}，{self.age} 岁，属于{self.kingdom}，移动方式：{self.move()}"


class Bird(Animal):
    """鸟类继承动物类，补充鸟的共同知识。"""
    def __init__(self, name, age, can_fly=True):
        super().__init__(name, age)
        self.can_fly = can_fly
        self.body_cover = "羽毛"

    def move(self):
        return "飞行" if self.can_fly else "行走"

    def lay_egg(self):
        return f"{self.name} 产了一枚卵"


class Fish(Animal):
    def move(self):
        return "游泳"

    def breathe(self):
        return f"{self.name} 用鳃呼吸"


class Penguin(Bird):
    """企鹅是鸟但不会飞——继承体系下的特化。"""
    def __init__(self, name, age):
        super().__init__(name, age, can_fly=False)


zoo = [Bird("白鹭", 3), Fish("鲤鱼", 1), Penguin("QQ", 2)]
for a in zoo:
    print(a.info())                 # 同一个 info()，行为随对象类型不同（多态）

print(zoo[0].lay_egg())
print(zoo[1].breathe())

"""迷你知识图谱：三元组存储 + 多跳查询 + 规则推理 + 简单问答"""
from collections import defaultdict

class KnowledgeGraph:
    def __init__(self):
        self.triples = set()
        self.spo = defaultdict(list)     # (主体, 关系) -> [客体]
        self.ops = defaultdict(list)     # (客体, 关系) -> [主体]

    def add(self, s, p, o):
        if (s, p, o) not in self.triples:
            self.triples.add((s, p, o))
            self.spo[(s, p)].append(o)
            self.ops[(o, p)].append(s)

    def query(self, s, p, o):
        """三元组中任意一项为 None 表示查询变量，返回所有匹配。"""
        ans = []
        for (a, b, c) in self.triples:
            if (s is None or s == a) and (p is None or p == b) and (o is None or o == c):
                ans.append((a, b, c))
        return ans

    def hop(self, entity, *relations):
        """沿关系链做多跳查询，如 hop('红楼梦', '作者', '国籍')。"""
        current = [entity]
        for r in relations:
            nxt = []
            for e in current:
                nxt.extend(self.spo.get((e, r), []))
            current = nxt
        return current

    def add_rule(self, new_rel, rel_chain):
        """关系推理规则：若存在 rel_chain 关系链，则蕴含 new_rel 关系。
        例：new_rel='祖父', chain=('父亲','父亲')"""
        added = []
        for s in self.query(None, rel_chain[0], None):
            tails = self.hop(s[0], *rel_chain)
            for t in tails:
                before = len(self.triples)
                self.add(s[0], new_rel, t)
                if len(self.triples) > before:
                    added.append((s[0], new_rel, t))
        return added


if __name__ == "__main__":
    kg = KnowledgeGraph()
    # ---- 文学领域知识 ----
    data = [
        ("红楼梦", "作者", "曹雪芹"), ("曹雪芹", "朝代", "清朝"),
        ("红楼梦", "体裁", "古典小说"), ("西游记", "作者", "吴承恩"),
        ("吴承恩", "朝代", "明朝"), ("三国演义", "作者", "罗贯中"),
        ("罗贯中", "朝代", "元末明初"), ("水浒传", "作者", "施耐庵"),
        ("施耐庵", "朝代", "元末明初"),
        ("贾宝玉", "登场于", "红楼梦"), ("孙悟空", "登场于", "西游记"),
        # ---- 地理知识 ----
        ("中国", "首都", "北京"), ("北京", "位于", "中国"),
        ("法国", "首都", "巴黎"), ("巴黎", "位于", "法国"),
        ("巴黎", "地标", "埃菲尔铁塔"),
    ]
    for t in data:
        kg.add(*t)

    # 1) 单条件查询：四大名著的作者
    print("《红楼梦》的作者：", kg.hop("红楼梦", "作者"))

    # 2) 多跳查询：《红楼梦》作者的朝代
    print("《红楼梦》作者所处朝代：", kg.hop("红楼梦", "作者", "朝代"))

    # 3) 反向查询：谁登场于《西游记》
    who = [s for s, p, o in kg.query(None, "登场于", "西游记")]
    print("《西游记》中的角色：", who)

    # 4) 两跳到达：从《红楼梦》出发，经'作者'能到谁，再经'朝代'
    print("全部三元组数量：", len(kg.triples))

    # 5) 规则推理：定义'名著作者时代' = 作者→朝代 的压缩关系
    for s, _, book in kg.query(None, "登场于", None):
        pass
    new_facts = kg.add_rule("作者朝代", ("作者", "朝代"))
    print("规则推理新增三元组：")
    for f in sorted(new_facts):
        print("  ", f)

    # 6) 埃菲尔铁塔在哪个国家（两跳：地标→位于）
    print("埃菲尔铁塔所在国家：", kg.hop("埃菲尔铁塔", "位于"))

    # 7) 简单模板问答
    def qa(question):
        q = question.replace("？", "").replace("?", "")
        if "的作者是谁" in q:
            book = q.split("的作者是谁")[0].strip("《》")
            return "、".join(kg.hop(book, "作者"))
        if "在哪个国家" in q:
            place = q.split("在哪个国家")[0]
            return "、".join(kg.hop(place, "位于"))
        return "抱歉，这个问题超出了我的知识范围。"

    print(qa("《红楼梦》的作者是谁？"))
    print(qa("埃菲尔铁塔在哪个国家？"))