10 - 神经符号AI与持续学习¶

class NeuroSymbolicLLM:
    """
    Type 2 神经符号AI: LLM + 符号工具

    LLM负责自然语言理解和规划
    符号工具负责精确计算和知识查询

    这是当前最实用的神经符号AI范式
    """

    def __init__(self, llm, tools: dict):  # __init__构造方法，创建对象时自动调用
        """
        参数:
            llm: 语言模型（神经网络部分）
            tools: 工具集合（符号推理部分）
                keys: 工具名称
                values: 可调用的工具函数
        """
        self.llm = llm
        self.tools = tools

    def reason(self, question: str) -> str:
        """
        结合神经和符号推理回答问题

        流程:
        1. LLM理解问题，决定是否需要工具
        2. 如需要，调用对应的符号工具
        3. 将工具结果整合到最终回答中
        """
        # Step 1: LLM分析问题，生成推理计划
        plan = self.llm.generate(f"""
        问题: {question}
        可用工具: {list(self.tools.keys())}
        请分析是否需要工具，如需要请给出调用计划。
        """)

        # Step 2: 执行工具调用（符号推理）
        tool_results = {}
        for tool_name, tool_fn in self.tools.items():
            if tool_name in plan:
                # 提取参数并调用工具
                args = self._extract_args(plan, tool_name)
                tool_results[tool_name] = tool_fn(**args)  # *args接收任意位置参数，**kwargs接收任意关键字参数

        # Step 3: LLM整合结果
        answer = self.llm.generate(f"""
        问题: {question}
        工具结果: {tool_results}
        请给出最终回答。
        """)

        return answer

    def _extract_args(self, plan, tool_name):
        """从推理计划中提取工具参数（简化实现）"""
        return {}

# 使用示例
tools = {
    'calculator': lambda expr: eval(expr),          # 精确计算
    'knowledge_graph': lambda query: "查询结果...",  # 知识查询
    'theorem_prover': lambda prop: True,            # 定理证明
}

import torch
import torch.nn as nn

class DifferentiableLogic(nn.Module):  # 继承nn.Module定义神经网络层
    """
    可微分的逻辑推理 (概率软逻辑)

    将离散的逻辑运算替换为连续可微的近似:
    - AND(a, b) → min(a, b) 或 a * b
    - OR(a, b)  → max(a, b) 或 a + b - a*b
    - NOT(a)    → 1 - a

    这样逻辑规则可以参与梯度反向传播
    """

    def __init__(self, t_norm: str = 'product'):
        """
        参数:
            t_norm: T-范数类型
                - 'product': 乘积 (AND=a*b, OR=a+b-a*b)
                - 'godel': Gödel (AND=min, OR=max)
                - 'lukasiewicz': Łukasiewicz (AND=max(0,a+b-1))
        """
        super().__init__()  # super()调用父类方法
        self.t_norm = t_norm

    def fuzzy_and(self, a: torch.Tensor, b: torch.Tensor) -> torch.Tensor:
        """可微分的AND操作"""
        if self.t_norm == 'product':
            return a * b
        elif self.t_norm == 'godel':
            return torch.min(a, b)
        elif self.t_norm == 'lukasiewicz':
            return torch.clamp(a + b - 1, min=0)

    def fuzzy_or(self, a: torch.Tensor, b: torch.Tensor) -> torch.Tensor:
        """可微分的OR操作"""
        if self.t_norm == 'product':
            return a + b - a * b
        elif self.t_norm == 'godel':
            return torch.max(a, b)
        elif self.t_norm == 'lukasiewicz':
            return torch.clamp(a + b, max=1)

    def fuzzy_not(self, a: torch.Tensor) -> torch.Tensor:
        """可微分的NOT操作"""
        return 1 - a

    def fuzzy_implies(self, a: torch.Tensor, b: torch.Tensor) -> torch.Tensor:
        """可微分的蕴含 (A → B ≡ ¬A ∨ B)"""
        return self.fuzzy_or(self.fuzzy_not(a), b)

class NeuralLogicNetwork(nn.Module):
    """
    神经逻辑网络: 将逻辑规则嵌入神经网络

    示例任务: 关系推理
    - 输入: 实体对(x, y)的特征
    - 规则: parent(x,y) ∧ parent(y,z) → grandparent(x,z)
    - 输出: 各种关系的概率
    """

    def __init__(self, entity_dim: int, n_relations: int):
        """
        参数:
            entity_dim: 实体嵌入维度
            n_relations: 关系类型数量
        """
        super().__init__()

        # 实体嵌入层
        self.entity_encoder = nn.Linear(entity_dim, 128)

        # 关系预测器（神经部分）
        self.relation_predictor = nn.Sequential(
            nn.Linear(256, 128),  # 两个实体拼接
            nn.ReLU(),
            nn.Linear(128, n_relations),
            nn.Sigmoid()  # 输出概率 [0, 1]
        )

        # 可微分逻辑引擎（符号部分）
        self.logic = DifferentiableLogic(t_norm='product')

    def predict_relation(self, entity_a: torch.Tensor,
                         entity_b: torch.Tensor) -> torch.Tensor:
        """预测两个实体间的关系概率"""
        a_emb = self.entity_encoder(entity_a)
        b_emb = self.entity_encoder(entity_b)
        pair = torch.cat([a_emb, b_emb], dim=-1)
        return self.relation_predictor(pair)

    def apply_transitivity_rule(
        self,
        p_parent_xy: torch.Tensor,
        p_parent_yz: torch.Tensor
    ) -> torch.Tensor:
        """
        应用传递性规则: parent(x,y) ∧ parent(y,z) → grandparent(x,z)
        使用可微分逻辑保持梯度流
        """
        return self.logic.fuzzy_and(p_parent_xy, p_parent_yz)

    def forward(self, entities_a, entities_b, entities_c=None):
        """
        前向传播: 神经预测 + 逻辑规则推导
        """
        # 神经部分: 直接预测关系
        direct_preds = self.predict_relation(entities_a, entities_b)

        if entities_c is not None:
            # 逻辑推理部分: 利用传递性增强预测
            p_ab = self.predict_relation(entities_a, entities_b)
            p_bc = self.predict_relation(entities_b, entities_c)

            # 应用规则: parent(a,b) ∧ parent(b,c) → grandparent(a,c)
            p_grandparent = self.apply_transitivity_rule(
                p_ab[:, 0],  # parent关系的概率
                p_bc[:, 0]
            )

            return direct_preds, p_grandparent

        return direct_preds

import torch
import torch.nn as nn
from typing import Dict

class ElasticWeightConsolidation:
    """
    EWC (Elastic Weight Consolidation)

    核心思想:
    - 计算每个权重对旧任务的"重要性"（Fisher信息矩阵）
    - 训练新任务时，惩罚重要权重的变化
    - 类比: 像弹簧一样把重要权重拉回到旧值附近

    损失函数: L = L_new + (λ/2) Σ F_i (θ_i - θ*_i)²

    参考: Kirkpatrick et al., "Overcoming Catastrophic Forgetting" (2017)
    """

    def __init__(self, model: nn.Module, ewc_lambda: float = 400.0):
        """
        参数:
            model: 神经网络模型
            ewc_lambda: EWC正则化强度
        """
        self.model = model
        self.ewc_lambda = ewc_lambda

        # 存储旧任务的参数和Fisher矩阵
        self.saved_params: Dict[str, torch.Tensor] = {}
        self.fisher: Dict[str, torch.Tensor] = {}

    def compute_fisher(self, data_loader, n_samples: int = 200):
        """
        计算Fisher信息矩阵（对角近似）

        F_i = E[( ∂log p(y|x,θ) / ∂θ_i )²]

        直觉: 如果某个权重的梯度经常很大，
              说明这个权重对当前任务很重要
        """
        fisher = {name: torch.zeros_like(param)
                  for name, param in self.model.named_parameters()}

        self.model.eval()  # eval()开启评估模式（关闭Dropout等）
        count = 0

        for inputs, targets in data_loader:
            if count >= n_samples:
                break

            self.model.zero_grad()  # 清零梯度，防止梯度累积
            outputs = self.model(inputs)
            loss = nn.functional.cross_entropy(outputs, targets)
            loss.backward()  # 反向传播计算梯度

            # 累加梯度的平方（Fisher的对角近似）
            for name, param in self.model.named_parameters():
                if param.grad is not None:
                    fisher[name] += param.grad.data ** 2

            count += inputs.size(0)

        # 取平均
        for name in fisher:
            fisher[name] /= count

        self.fisher = fisher

        # 保存当前参数（旧任务的最优参数 θ*）
        self.saved_params = {
            name: param.data.clone()
            for name, param in self.model.named_parameters()
        }

    def ewc_loss(self) -> torch.Tensor:
        """
        计算EWC正则化损失

        L_ewc = (λ/2) Σ_i F_i (θ_i - θ*_i)²

        F_i大的权重（对旧任务重要）变化受到更大惩罚
        """
        loss = torch.tensor(0.0, device=next(self.model.parameters()).device)

        for name, param in self.model.named_parameters():
            if name in self.fisher:
                # 权重变化 * Fisher重要性
                loss += (self.fisher[name] *
                        (param - self.saved_params[name]) ** 2).sum()

        return 0.5 * self.ewc_lambda * loss

    def train_new_task(self, data_loader, optimizer, epochs=10):
        """
        训练新任务（带EWC正则化）
        """
        self.model.train()  # train()开启训练模式

        for epoch in range(epochs):
            total_loss = 0
            for inputs, targets in data_loader:
                optimizer.zero_grad()

                # 新任务损失
                outputs = self.model(inputs)
                task_loss = nn.functional.cross_entropy(outputs, targets)

                # EWC正则化损失
                reg_loss = self.ewc_loss()

                # 总损失
                loss = task_loss + reg_loss
                loss.backward()
                optimizer.step()  # 根据梯度更新模型参数

                total_loss += loss.item()  # .item()将单元素张量转为Python数值

            print(f"Epoch {epoch+1}: loss={total_loss/len(data_loader):.4f}")

class GenerativeReplay:
    """
    生成式回放: 用生成模型代替存储旧数据

    核心思想:
    - 训练一个生成模型（如VAE/GAN）来生成旧任务的"伪数据"
    - 学新任务时，混合真实新数据和生成的旧数据
    - 优势: 不需要存储旧数据（隐私友好）

    参考: Shin et al., "Continual Learning with Deep Generative Replay" (2017)
    """

    def __init__(self, task_model: nn.Module, generator: nn.Module):
        """
        参数:
            task_model: 任务模型（分类器等）
            generator: 生成模型（VAE/GAN）
        """
        self.task_model = task_model
        self.generator = generator

    def generate_replay_data(self, n_samples: int = 100):
        """用生成模型生成旧任务的伪数据"""
        self.generator.eval()
        with torch.no_grad():  # 禁用梯度计算，节省内存
            # 从噪声生成数据
            z = torch.randn(n_samples, self.generator.latent_dim)
            fake_data = self.generator.decode(z)

            # 用当前任务模型为伪数据打标签
            fake_labels = self.task_model(fake_data).argmax(dim=-1)

        return fake_data, fake_labels

    def train_step(self, real_data, real_labels, replay_ratio: float = 0.5):
        """
        混合真实数据和回放数据训练

        参数:
            real_data: 新任务的真实数据
            real_labels: 新任务的真实标签
            replay_ratio: 回放数据的比例
        """
        n_replay = int(len(real_data) * replay_ratio)

        if n_replay > 0:
            # 生成旧任务回放数据
            fake_data, fake_labels = self.generate_replay_data(n_replay)

            # 混合新旧数据
            all_data = torch.cat([real_data, fake_data])
            all_labels = torch.cat([real_labels, fake_labels])
        else:
            all_data, all_labels = real_data, real_labels

        # 正常训练
        outputs = self.task_model(all_data)
        loss = nn.functional.cross_entropy(outputs, all_labels)

        return loss

import torch
import torch.nn as nn
import numpy as np

class PlanarFlow(nn.Module):
    """
    平面流（Planar Flow）— 最简单的归一化流

    变换: f(z) = z + u · h(w^T z + b)
    其中 u, w ∈ R^d, b ∈ R, h 是激活函数

    参考: Rezende & Mohamed, "Variational Inference with Normalizing Flows" (2015)
    """

    def __init__(self, dim: int):
        super().__init__()
        self.w = nn.Parameter(torch.randn(dim))     # 方向向量
        self.u = nn.Parameter(torch.randn(dim))     # 缩放向量
        self.b = nn.Parameter(torch.zeros(1))       # 偏置

    def forward(self, z: torch.Tensor):
        """
        前向变换 + 计算对数行列式雅可比

        参数:
            z: 输入 (batch, dim)
        返回:
            z': 变换后的输出
            log_det: 对数行列式 (batch,)
        """
        # 确保可逆性（通过reparameterize u）
        u_hat = self._get_u_hat()

        # 线性部分
        linear = z @ self.w + self.b  # (batch,)

        # 非线性变换
        h = torch.tanh(linear)           # (batch,)
        h_prime = 1 - h ** 2            # tanh的导数

        # f(z) = z + u_hat * h(w^T z + b)
        z_new = z + u_hat.unsqueeze(0) * h.unsqueeze(1)  # unsqueeze增加一个维度

        # 对数行列式雅可比
        # det(I + u * (h' * w)^T) = 1 + u^T (h' * w)
        psi = h_prime.unsqueeze(1) * self.w.unsqueeze(0)  # (batch, dim)
        log_det = torch.log(torch.abs(1 + psi @ u_hat) + 1e-8)

        return z_new, log_det

    def _get_u_hat(self) -> torch.Tensor:
        """重参数化u，确保变换可逆"""
        wu = self.w @ self.u
        m_wu = -1 + torch.log1p(torch.exp(wu))  # softplus
        u_hat = self.u + (m_wu - wu) * self.w / (self.w @ self.w + 1e-8)
        return u_hat

class NormalizingFlow(nn.Module):
    """
    归一化流: K个可逆变换的组合

    z_0 ~ N(0, I) → f_1 → f_2 → ... → f_K → x

    精确计算似然:
    log p(x) = log p(z_0) - Σ log|det J_k|
    """

    def __init__(self, dim: int, n_flows: int = 8, flow_type: str = 'planar'):
        """
        参数:
            dim: 数据维度
            n_flows: 流的层数（越多表达能力越强）
            flow_type: 流的类型
        """
        super().__init__()
        self.dim = dim

        if flow_type == 'planar':
            self.flows = nn.ModuleList([PlanarFlow(dim) for _ in range(n_flows)])  # 列表推导式，简洁创建列表
        else:
            raise ValueError(f"未知流类型: {flow_type}")

    def forward(self, z: torch.Tensor):
        """前向传播: 简单分布 → 复杂分布"""
        log_det_sum = torch.zeros(z.shape[0], device=z.device)

        for flow in self.flows:
            z, log_det = flow(z)
            log_det_sum += log_det.squeeze()  # squeeze去除大小为1的维度

        return z, log_det_sum

    def log_likelihood(self, x: torch.Tensor) -> torch.Tensor:
        """
        计算数据的精确对数似然

        需要逆变换（从x到z），然后:
        log p(x) = log p(z_0) + Σ log|det J_k^{-1}|
        """
        # 注: 平面流的逆变换没有解析解，需要用其他架构
        # 这里展示正向计算（从z到x的对数似然）
        z0 = torch.randn_like(x)
        x_gen, log_det = self.forward(z0)

        # 先验分布的log概率（标准正态）
        log_pz = -0.5 * (z0 ** 2 + np.log(2 * np.pi)).sum(dim=-1)

        return log_pz + log_det

    def sample(self, n_samples: int = 100) -> torch.Tensor:
        """从学到的分布中采样"""
        z0 = torch.randn(n_samples, self.dim)
        samples, _ = self.forward(z0)
        return samples

    def train_step(self, data: torch.Tensor, optimizer):
        """
        最大似然训练
        """
        # 从先验采样
        z0 = torch.randn_like(data)
        x_gen, log_det = self.forward(z0)

        # 负对数似然损失
        log_pz = -0.5 * (z0 ** 2 + np.log(2 * np.pi)).sum(dim=-1)
        nll = -(log_pz + log_det).mean()

        optimizer.zero_grad()
        nll.backward()
        optimizer.step()

        return nll.item()