第 5 章 · 01 Rete 前向链推理 ★

第 5 章 · 01 Rete 前向链推理 ★

本节摘要：全书高潮之一，装配管线第⑦段。reasoning 模块（11 个文件、3391 行）是 Semantica 里唯一一处"零 LLM"的智能来源：不调用任何大模型，仅凭 Rete 前向链、规则匹配与不动点迭代，在图谱上推出新事实。本节先回答"LLM 时代为什么还要确定性推理"——LLM 会幻觉，而 Rete 推理 100% 可复现、可审计，这是监管场景的刚需；再精读 Rete 网络的四类节点（AlphaNode/BetaNode/TerminalNode）、建网与事实传播的完整链路，以及门面类 Reasoner 的前向链不动点循环，最后把确定性推理与 LLM 推理摆上同一张对比桌。

内容来源：semantica/reasoning/（rete_engine.py 387 行、reasoner.py 472 行、graph_reasoner.py 162 行、__init__.py）、README.md（reasoning 模块参考）

⚠️ 注意：ReteEngine 的 AlphaNode 条件匹配器在当前版本是有意从简的——_matches() 直接 return True，README 明确标注这是已知限制（Current limitation），要求在生产合规闸门上先验证 match_patterns() 的输出。经典条件求值在路线图上；因此本章同时精读门面类 Reasoner 的完整正则匹配实现，它在语义上承担了真正的"逐条件绑定"工作。

学习目标

1. 说清确定性推理与 LLM 生成的本质区别，以及监管场景为什么必须要前者。
2. 掌握 Rete 算法的心智模型：规则+事实→匹配→激活→执行，alpha 网络/beta 网络/终结点各司其职。
3. 读懂 ReteEngine 的建网（_add_rule_to_network）与事实传播（_propagate_fact）源码。
4. 掌握门面类 Reasoner 的前向链不动点循环、?var 变量绑定与新事实推导。
5. 能对比确定性推理与 LLM 推理，知道各自的适用位置。

一、为什么 LLM 时代还需要确定性推理

一个容易被问倒的问题：既然有 LLM，为什么还要 20 世纪 70 年代发明的 Rete 算法？答案藏在 README 的一句话里：

The reasoning engines, KG construction, and provenance layer are fully deterministic; no LLM is required to use them.

"fully deterministic"（完全确定性）三个词值千金，它意味着三条性质，恰好都是 LLM 给不了的：

1. 可复现。同样的规则、同样的事实，今天跑和半年后跑，结论一个字都不差。LLM 同一 prompt 两次采样可能给出不同答案，温度设为 0 也只是概率上的众数，不是逻辑上的必然。
2. 可审计。每一条推出来的新事实都携带"用了哪条规则、基于哪些前提"的完整记录（InferenceResult.premises）。监管者问"AI 拒贷为什么"，答案必须能逐条回放，而不是"模型认为"。
3. 无幻觉。前向链只会推出规则前提满足时的结论，推不出任何"编造"——推理空间被规则集严格约束。LLM 则可能流畅地输出事实性错误。

所以在 Semantica 的分工里：LLM 负责理解自然语言、抽取实体关系（第 3 章）；确定性推理负责在结构化事实上做一切"要签字画押"的结论推导（本章）。一个典型的监管链路是：LLM 抽取 → 图谱 → Rete/Datalog 推理 → 决策记录（第 6 章）→ PROV-O 溯源（第 7 章），推理这一段全程零概率。

二、Rete 算法：三十秒经典版

Rete（拉丁语"网"）是前向链（forward chaining）规则引擎的经典匹配算法。前向链的推演方向是"从事实到结论"：

规则库（IF 条件1 AND 条件2 THEN 结论）
工作记忆（已知事实集合）
    │ ①匹配 match：找出所有条件被事实满足的规则
    │ ②冲突消解：选出要执行的激活
    │ ③执行 act：把结论写入工作记忆
    └ 循环，直到推不出新事实（不动点）
​

朴素做法每轮都拿所有规则去扫所有事实，复杂度 O(规则×事实×条件)。Rete 的核心洞察是把规则编译成一张网络并缓存中间匹配：新事实到来时只沿网络传播一次，历史匹配结果存在节点里不重算。网络分三段：

• alpha 网络：单条件过滤，每个条件一个节点，负责"这条事实自己像不像我的模式"；
• beta 网络：连接（join）节点，把左分支的部分匹配与新事实拼接，保证跨条件的变量一致；
• 终结点（terminal）：一条规则的全部条件都满足时在此激活。

Semantica 的 rete_engine.py 把这三段一一落成了类。

三、四类节点：Match 与 ReteNode 家族

先看激活记录与节点基类（rete_engine.py 第 46—62 行）：

@dataclass
class Match:
    """Pattern match."""
    rule: Rule
    facts: List[Fact]
    bindings: Dict[str, Any] = field(default_factory=dict)
    confidence: float = 1.0

class ReteNode:
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.children: List["ReteNode"] = []
​

Match 是一条"待执行激活"：哪条规则、被哪些事实满足、变量绑定是什么。confidence 默认 1.0——注意这个细节，Rete 推理的置信度来自规则定义而非概率采样。三类功能节点接着登场（第 64—117 行）：

class AlphaNode(ReteNode):
    """Alpha node for single condition matching."""
    def __init__(self, node_id: str, condition: Any):
        super().__init__(node_id)
        self.condition = condition
        self.matches: List[Fact] = []          # 缓存：Rete 的灵魂

    def add_fact(self, fact: Fact) -> bool:
        if self._matches(fact):
            self.matches.append(fact)
            return True
        return False

    def _matches(self, fact: Fact) -> bool:
        # Simple matching - can be enhanced
        return True                            # ← 已知限制，见本章开头注意

class BetaNode(ReteNode):
    def __init__(self, node_id: str, left: ReteNode, right: ReteNode):
        ...
        self.matches: List[Tuple[Fact, Fact]] = []

    def join(self, left_fact: Fact, right_fact: Fact) -> bool:
        if self._can_join(left_fact, right_fact):
            self.matches.append((left_fact, right_fact))
            return True
        return False

class TerminalNode(ReteNode):
    def __init__(self, node_id: str, rule: Rule):
        ...
        self.activations: List[Match] = []     # 激活队列

    def activate(self, match: Match) -> None:
        self.activations.append(match)
​

三个节点各有一个累积容器（matches/activations），这正是 Rete 区别于朴素匹配的地方：匹配结果不是算完就扔，而是存进网络，下次传播只处理增量。节点间用 children 列表连成有向无环图，事实就是沿着 children 一层层往下流的 token。

四、建网：_add_rule_to_network

build_network()（第 155 行）清空旧网络后对每条规则调 _add_rule_to_network（第 192—222 行）：

def _add_rule_to_network(self, rule: Rule) -> None:
    # Create alpha nodes for each condition
    alpha_nodes = []
    for condition in rule.conditions:
        node_id = f"alpha_{self.node_counter}"
        self.node_counter += 1
        alpha_node = AlphaNode(node_id, condition)
        alpha_nodes.append(alpha_node)
        self.network[node_id] = alpha_node

    # Create beta nodes for joining
    if len(alpha_nodes) > 1:
        current = alpha_nodes[0]
        for i in range(1, len(alpha_nodes)):
            node_id = f"beta_{self.node_counter}"
            beta_node = BetaNode(node_id, current, alpha_nodes[i])
            self.network[node_id] = beta_node
            current = beta_node                # beta 链式串联
        final_node = current
    else:
        final_node = alpha_nodes[0] if alpha_nodes else None

    # Create terminal node
    if final_node:
        terminal_node = TerminalNode(node_id, rule)
        final_node.children.append(terminal_node)
        self.network[node_id] = terminal_node
​

三步结构一目了然：每个条件建一个 alpha 节点；多个条件时用 beta 节点左折叠成链（第 i 个 beta 的左输入是第 i−1 个 beta，右输入是第 i 个 alpha）；链尾挂终结点。一条两条件规则在网里的形状是 alpha_0 → beta_1 ← alpha_1 → terminal_2。共享相同条件的规则天然共享 alpha 节点——这是 Rete "规则越多、相对省得越多"的由来。get_network_stats()（第 373 行）会返回 alpha_nodes/beta_nodes/terminal_nodes 四项计数，方便确认网络形状。

五、事实传播：token 沿网流动

网络建好后，事实的加入触发传播（第 224—264 行）：

def add_fact(self, fact: Fact) -> None:
    self.facts.append(fact)
    self._propagate_fact(fact)                 # 增量传播，只处理新事实

def _propagate_fact(self, fact: Fact) -> None:
    for node_id, node in self.network.items():
        if isinstance(node, AlphaNode):
            if node.add_fact(fact):            # alpha 过滤
                self._propagate_from_alpha(node, fact)

def _propagate_from_alpha(self, alpha_node, fact):
    for child in alpha_node.children:
        if isinstance(child, BetaNode):
            for left_fact in alpha_node.matches:   # 与左侧历史匹配做 join
                if child.join(left_fact, fact):
                    for grandchild in child.children:
                        if isinstance(grandchild, TerminalNode):
                            match = Match(rule=grandchild.rule,
                                          facts=[left_fact, fact], confidence=1.0)
                            grandchild.activate(match)
        elif isinstance(child, TerminalNode):      # 单条件规则直达终结点
            child.activate(Match(rule=child.rule, facts=[fact], confidence=1.0))
​

读这段要盯住两条路径：单条件规则，事实过 alpha 即达 terminal；多条件规则，alpha 命中的事实与 matches 里的历史匹配逐对 join，拼出完整匹配就激活终结点。之后 match_patterns()（第 266 行）收集所有终结点的 activations，execute_matches()（第 314 行）逐条执行——当前版本执行就是取 match.rule.conclusion 作为推断结果。README 的 AML 反洗钱例子（金额 >1 万且国家命中高风险名单 → flag_for_compliance_review）演示了完整用法：build_network 两条规则、add_fact 一笔交易、match_patterns 直接命中合规标记。

六、从知识图谱推导新事实：Reasoner 门面

ReteEngine 之上，reasoner.py 提供了推荐入口 Reasoner。它先用两个 dataclass 固定数据形态（第 25—55 行）：Rule（conditions/conclusion/priority/confidence）与 InferenceResult（conclusion/rule_used/premises）——后者就是审计回放的最小单元。规则可以用字符串直接写，由 _parse_rule_definition（第 351 行）的正则 IF\s+(.+?)\s+THEN\s+(.+) 解析：

reasoner.add_fact("Person(John)")
reasoner.add_rule("IF Person(?x) THEN Human(?x)")
results = reasoner.forward_chain()
​

知识图谱不用手工转写——add_fact（第 133—147 行）直接吃 KG 字典：

if "type" in fact and ("name" in fact or "id" in fact):
    name = fact.get("name", fact.get("id"))
    etype = fact.get("type", "Entity")
    self.facts.add(f"{etype}({name})")           # 实体 → 谓词
elif "source_id" in fact or "source_name" in fact:
    ...
    self.facts.add(f"{rtype}({source}, {target})")  # 关系 → 二元谓词
​

第 3 章建好的图谱，节点变 Person(John)、边变 WorksFor(John, Acme)，推理引擎无缝接驳。真正的推导核心是 forward_chain（第 204—269 行），一个带安全阀的不动点循环：

max_iterations = self.config.get("max_iterations", 50)
while new_facts_added and iteration < max_iterations:
    pre_pass_facts = frozenset(self.facts)     # 本轮开始前的事实快照
    pass_results: Dict[str, InferenceResult] = {}
    for rule in self.rules:
        for conclusion, matched_facts in self._match_rule(rule):
            if conclusion in pass_results:     # 同轮多次推导同一结论
                existing = pass_results[conclusion]
                for fact in matched_facts:     #   合并前提，不丢证据（#733）
                    if fact not in existing.premises:
                        existing.premises.append(fact)
                continue
            if conclusion in pre_pass_facts:   # 旧事实不算新推导
                continue
            self.facts.add(conclusion)         # 立即入集，本轮后续规则可链式引用
            ...
            new_facts_added = True
​

三个工程细节值得划线：其一，pre_pass_facts 快照区分"本来就已知"与"本轮新推"；其二，新结论立即加入 self.facts 而非轮末批量入集，于是"IF A THEN B"点火后"IF B THEN C"能在同一轮跟进，减少外层迭代；其三，同一结论的多条推导路径会把前提合并进同一条 InferenceResult（issue #733），审计时证据不丢。配套的还有 add_rule 的幂等去重（第 105—126 行，相同条件+结论的规则不再重复添加，issue #732——Jupyter 里反复跑同一 cell 不会悄悄翻倍规则），以及按 priority 排序的执行优先级。backward_chain/_prove_goal（第 271—349 行）则提供反向链：给定目标事实，从规则结论反推需要的条件，递归证明——第 5.3 节演绎推理会回到它。

变量绑定藏在 _match_pattern（第 417—455 行）：模式里的 ?x 先按 re.split(r"(\?\w+)", pattern) 切出变量段，未绑定变量生成命名捕获组 (?P<x>.+?)，同一变量出现两次则生成反向引用 (?P=x)，已绑定变量直接 escape 成字面量——一个约 30 行的函数，实现了合一代数里"变量一致性"的语义。

💡 装配要点：本阶是全书"零 LLM 智能"的起点。心智模型：规则编译成 alpha（单条件过滤）→beta（连接）→terminal（激活）三段网络，事实作为 token 沿网传播，匹配结果缓存在节点里做增量更新；Reasoner 门面用不动点循环把这条链跑到"推不出新事实"为止，每条新事实都留下 InferenceResult（规则+前提+置信度）作为审计凭据。接进管线的方式：kg.build() 的产物经 add_fact 的字典转换喂给引擎，结论再流入第 6 章的 record_decision。

本节要点回顾

• 确定性推理的三条刚需性质：可复现、可审计、无幻觉；README 明言 reasoning/provenance 层 fully deterministic，不需要 LLM。
• Rete 四类对象：Match（激活记录）、AlphaNode（单条件过滤+matches 缓存）、BetaNode（左右连接）、TerminalNode（激活队列）；_matches/_can_join 当前从简，生产前需验证输出。
• 建网三步：逐条件建 alpha、beta 左折叠成链、链尾挂 terminal；共享条件天然共享节点。
• 传播两条路径：单条件直达 terminal，多条件与历史匹配 join 后激活；执行阶段取 rule.conclusion。
• Reasoner 门面：KG 字典直接转谓词事实、IF...THEN... 字符串规则、forward_chain 不动点循环（max_iterations=50 安全阀）、#733 前提合并、#732 规则去重、?var 正则合一（含反向引用）。

下一节：02 Datalog 与 SPARQL：图查询双引擎——半朴素不动点的递归 Datalog、面向 RDF 的 SPARQL，以及自然语言到查询的桥接。