简介

持续学习的核心挑战在于平衡对旧任务的稳定性对新任务的可塑性。传统机器学习理论关注单任务泛化,而持续学习需要分析跨任务的泛化性质。本文档基于PAC-Bayes框架,系统建立持续学习的泛化理论,刻画任务间迁移、稳定-可塑性权衡的理论边界。123


1. PAC-Bayes框架回顾

1.1 标准PAC-Bayes界

PAC-Bayes框架由McAllester于1999年提出,为贝叶斯学习提供泛化保证。

定理1(标准PAC-Bayes界):对于任意先验分布 、任意后验分布 、任意假设空间 ,以概率至少 有:

其中:

  • :真实风险
  • :经验风险
  • :KL散度
  • :样本数量

1.2 任务条件泛化

在持续学习中,我们需要分析任务条件泛化

定义1(任务条件PAC-Bayes界):以概率至少

其中 是任务 的先验。


2. 多任务泛化界

2.1 联合PAC-Bayes界

定理2(联合多任务PAC-Bayes界)1:对于 个任务的联合学习,以概率至少

其中:

  • 是平均KL散度
  • 是最小样本数

2.2 顺序学习泛化界

定理3(顺序学习泛化界):考虑 个任务的顺序学习,设 是学完所有任务后的参数。则:

其中 是只训练任务 后的最优参数, 是从 的遗忘量。

2.3 遗忘感知的泛化界

定义2(遗忘感知KL散度):定义任务 的遗忘感知KL散度:

其中 是任务 学完后参数的后验。

定理4(遗忘感知PAC-Bayes界):以概率至少


3. 任务间迁移的数学刻画

3.1 正向迁移的泛化分析

正向迁移(Forward Transfer):利用旧任务的知识帮助学习新任务。

定义3(正向迁移增益)

其中 是学完任务 后的后验, 是初始先验。

定理5(正向迁移界):如果任务 共享结构,则:

3.2 反向迁移的泛化分析

反向迁移(Backward Transfer):学习新任务后改善旧任务的性能。

定义4(反向迁移增益)

定理6(反向迁移界):反向迁移的量被以下界控制:

3.3 迁移系数的估计

定义5(任务迁移系数)

其中 是任务 的真实梯度。

定理7(迁移系数与泛化)

其中 是参数空间的半径。

import torch
import torch.nn as nn
import numpy as np
 
class TransferAnalysis:
    """任务间迁移分析"""
    
    def __init__(self, model):
        self.model = model
        self.task_gradients = {}  # 存储每个任务的梯度
        self.task_losses = {}     # 存储每个任务的损失
    
    def compute_task_gradient(self, task_id, task_loader, device='cuda'):
        """
        计算任务task_id的真实梯度(使用整个任务数据)
        
        Returns:
            gradient: 任务梯度向量
        """
        self.model.zero_grad()
        
        # 计算任务损失
        total_loss = 0
        for x, y in task_loader:
            x, y = x.to(device), y.to(device)
            output = self.model(x)
            loss = nn.functional.cross_entropy(output, y)
            loss.backward()
            total_loss += loss.item()
        
        # 收集梯度
        gradient = torch.cat([
            p.grad.flatten() 
            for p in self.model.parameters() 
            if p.grad is not None
        ])
        
        self.task_gradients[task_id] = gradient
        self.task_losses[task_id] = total_loss / len(task_loader)
        
        return gradient
    
    def compute_transfer_coefficient(self, task_i, task_j):
        """
        计算从任务i到任务j的迁移系数
        
        ρ > 0: 正向迁移
        ρ < 0: 负向迁移
        ρ ≈ 0: 无迁移
        """
        if task_i not in self.task_gradients or task_j not in self.task_gradients:
            raise ValueError("Tasks not found in gradient cache")
        
        g_i = self.task_gradients[task_i]
        g_j = self.task_gradients[task_j]
        
        # 计算余弦相似度
        cos_sim = torch.nn.functional.cosine_similarity(
            g_i.unsqueeze(0), 
            g_j.unsqueeze(0)
        ).item()
        
        return cos_sim
    
    def estimate_forward_transfer(self, source_task, target_task, 
                                  target_loader, device='cuda'):
        """
        估计从源任务到目标任务的正向迁移
        
        比较:预训练模型 vs 随机初始化模型在目标任务上的性能
        """
        # 加载源任务训练后的权重
        pretrained_loss = self.task_losses[target_task]
        
        # 加载随机初始化权重并计算损失
        self.model.reset_parameters()
        random_loss = self._compute_loss(self.model, target_loader, device)
        
        # 正向迁移 = 随机初始化损失 - 预训练损失
        forward_transfer = random_loss - pretrained_loss
        
        return forward_transfer
    
    def _compute_loss(self, model, loader, device):
        """计算模型在数据集上的损失"""
        model.eval()
        total_loss = 0
        with torch.no_grad():
            for x, y in loader:
                x, y = x.to(device), y.to(device)
                output = model(x)
                loss = nn.functional.cross_entropy(output, y)
                total_loss += loss.item()
        model.train()
        return total_loss / len(loader)
    
    def analyze_transfer_matrix(self, tasks, loaders, device='cuda'):
        """
        分析所有任务对之间的迁移系数
        
        Returns:
            transfer_matrix: T x T 迁移系数矩阵
        """
        n_tasks = len(tasks)
        transfer_matrix = np.zeros((n_tasks, n_tasks))
        
        # 计算所有任务的梯度
        for i, task_id in enumerate(tasks):
            self.compute_task_gradient(task_id, loaders[i], device)
        
        # 计算迁移系数矩阵
        for i in range(n_tasks):
            for j in range(n_tasks):
                if i != j:
                    transfer_matrix[i, j] = self.compute_transfer_coefficient(i, j)
        
        return transfer_matrix

4. 稳定-可塑性权衡

4.1 形式化定义

稳定性(Stability):模型保留旧知识的能力。

可塑性(Plasticity):模型学习新知识的能力。

4.2 稳定-可塑性权衡定理

定理8(基本权衡)2:对于任意持续学习算法 ,在任意任务序列上,有:

其中:

  • 是联合多任务学习的最优损失
  • 是顺序学习的最终损失

含义:稳定性和可塑性之和存在下界,无法同时最大化两者。

4.3 最优权衡曲线

定义6(Pareto最优权衡):在稳定-可塑性空间中,满足”无法在不降低稳定性的情况下提高可塑性”的点构成Pareto前沿。

定理9(Pareto前沿界):Pareto前沿由以下参数化曲线给出:

其中 是可塑性的正则化项。

4.4 信息论分析

定理10(稳定-可塑性的信息瓶颈界):设 是表示 与任务 标签 的互信息。则:


5. 遗忘的PAC-Bayes分析

5.1 任务级遗忘界

定义7( -阶段遗忘)

定理11(遗忘的PAC-Bayes界)3:以概率至少

5.2 累积遗忘界

定义8(累积遗忘)

定理12(累积遗忘界)

5.3 遗忘与容量的关系

定理13(遗忘-容量权衡):设 是网络的参数容量(定义为其能存储的独立模式数量)。则:

其中 是与任务复杂度相关的常数。

含义:当任务数量增加时,遗忘量至少线性增长,除非增加参数容量。

class ForgettingBoundEstimator:
    """遗忘界估计器"""
    
    def __init__(self, model):
        self.model = model
        self.task_params = {}  # 存储每个任务结束后的参数
        self.task_losses = {}  # 存储每个任务的经验损失
    
    def estimate_forgetting_bound(self, task_t, m_t, delta=0.05):
        """
        估计任务t的遗忘上界
        
        Args:
            task_t: 任务索引
            m_t: 任务t的样本数
            delta: 置信度参数
        
        Returns:
            bound: 遗忘的PAC-Bayes上界
        """
        # 计算与任务t最优参数的KL散度
        if task_t not in self.task_params:
            return float('inf')
        
        kl_div = self._compute_kl_divergence(task_t)
        
        # PAC-Bayes界
        bound = np.sqrt((kl_div + np.log(2 * task_t / delta)) / (2 * m_t))
        
        return bound
    
    def _compute_kl_divergence(self, task_t):
        """
        计算当前参数与任务t最优参数的KL散度
        
        近似为高斯分布之间的KL散度
        """
        current_params = self._get_flattened_params()
        task_params = self._get_flattened_params(self.task_params[task_t])
        
        # 假设参数服从各向同性高斯分布
        sigma_sq = 1.0  # 简化假设
        
        kl = torch.sum(
            (current_params - task_params) ** 2
        ) / (2 * sigma_sq)
        
        return kl.item()
    
    def _get_flattened_params(self, state_dict=None):
        """获取展平后的参数向量"""
        if state_dict is None:
            state_dict = self.model.state_dict()
        
        return torch.cat([p.flatten() for p in state_dict.values()])
    
    def estimate_cumulative_forgetting(self, m_min, T, delta=0.05):
        """
        估计累积遗忘界
        
        Args:
            m_min: 最小任务样本数
            T: 任务总数
            delta: 置信度参数
        
        Returns:
            bound: 累积遗忘上界
        """
        total_bound = 0
        
        for t in range(1, T + 1):
            # 每个任务的遗忘界
            bound_t = self.estimate_forgetting_bound(t, m_min, delta / T)
            total_bound += bound_t
        
        return total_bound

6. 任务相似性与泛化

6.1 任务相似性度量

定义9(Hessian相似性)

定义10(梯度相似性)

6.2 相似性与遗忘的关系

定理14(相似性-遗忘关系):设任务 的Hessian相似性为 。则在学习任务 后,任务 的遗忘满足:

推论

  • (任务相似):遗忘较小
  • (任务正交):遗忘取决于参数偏移
  • (任务冲突):可能发生严重遗忘

6.3 任务聚类

定理15(任务分组界):如果任务可以被划分为 个簇,使得簇内任务相似度高、簇间相似度低,则:

其中 是簇 内的遗忘, 是簇间遗忘。

class TaskSimilarityAnalyzer:
    """任务相似性分析器"""
    
    def __init__(self, model):
        self.model = model
        self.hessians = {}
        self.gradients = {}
    
    def compute_task_hessian(self, task_id, task_loader, device='cuda'):
        """
        近似计算任务Hessian矩阵
        
        使用Fisher信息矩阵作为Hessian的近似
        """
        self.model.zero_grad()
        
        # 累积梯度外积(Fisher信息矩阵)
        fisher = None
        for x, y in task_loader:
            x, y = x.to(device), y.to(device)
            output = self.model(x)
            loss = nn.functional.cross_entropy(output, y)
            loss.backward()
            
            # 收集梯度
            grad = torch.cat([
                p.grad.flatten() 
                for p in self.model.parameters() 
                if p.grad is not None
            ])
            
            if fisher is None:
                fisher = torch.outer(grad, grad)
            else:
                fisher += torch.outer(grad, grad)
            
            self.model.zero_grad()
        
        fisher /= len(task_loader)
        self.hessians[task_id] = fisher
        
        return fisher
    
    def compute_hessian_similarity(self, task_i, task_j):
        """
        计算两个任务的Hessian相似性
        """
        if task_i not in self.hessians or task_j not in self.hessians:
            raise ValueError("Hessians not computed")
        
        H_i = self.hessians[task_i]
        H_j = self.hessians[task_j]
        
        # Frobenius内积
        inner_prod = torch.sum(H_i * H_j).item()
        
        # Frobenius范数
        norm_i = torch.norm(H_i, p='fro').item()
        norm_j = torch.norm(H_j, p='fro').item()
        
        # 相似性
        similarity = inner_prod / (norm_i * norm_j + 1e-8)
        
        return similarity
    
    def cluster_tasks(self, n_clusters, tasks):
        """
        基于相似性对任务进行聚类
        
        使用谱聚类算法
        """
        n_tasks = len(tasks)
        similarity_matrix = np.zeros((n_tasks, n_tasks))
        
        # 计算任务对之间的相似性
        for i in range(n_tasks):
            for j in range(i + 1, n_tasks):
                sim = self.compute_hessian_similarity(i, j)
                similarity_matrix[i, j] = sim
                similarity_matrix[j, i] = sim
        
        # 使用简单的k-means聚类
        # 实际应用中可使用谱聚类
        from sklearn.cluster import KMeans
        
        # 将相似性转换为距离
        distance_matrix = 1 - similarity_matrix
        
        # 使用k-means
        kmeans = KMeans(n_clusters=n_clusters, random_state=42)
        labels = kmeans.fit_predict(distance_matrix)
        
        return labels, similarity_matrix

7. 实践应用

7.1 早停的泛化分析

定理16(早停的PAC-Bayes保证):设 是验证损失最小的迭代。在 处停止提供以下保证:

7.2 正则化强度的选择

定理17(最优正则化):设 是EWC等方法的正则化强度。最优 满足:

class OptimalRegularizationEstimator:
    """最优正则化强度估计"""
    
    def __init__(self, model):
        self.model = model
        self.task_info = {}
    
    def estimate_optimal_lambda(self, tasks, task_loaders, device='cuda'):
        """
        估计EWC等方法的最优正则化强度
        """
        total_ratio = 0
        
        for t, loader in enumerate(task_loaders):
            # 计算梯度范数
            grad_norm = self._compute_grad_norm(loader, device)
            
            # 计算Fisher信息矩阵的最小特征值
            lambda_min = self._compute_min_fisher_eigenvalue(loader, device)
            
            total_ratio += (grad_norm ** 2) / lambda_min
        
        # 最优lambda
        lambda_star = total_ratio / len(tasks)
        
        return lambda_star
    
    def _compute_grad_norm(self, loader, device):
        """计算任务梯度范数"""
        self.model.zero_grad()
        
        total_loss = 0
        for x, y in loader:
            x, y = x.to(device), y.to(device)
            output = self.model(x)
            loss = nn.functional.cross_entropy(output, y)
            loss.backward()
            total_loss += loss.item()
        
        grad = torch.cat([
            p.grad.flatten() 
            for p in self.model.parameters() 
            if p.grad is not None
        ])
        
        return grad.norm().item()
    
    def _compute_min_fisher_eigenvalue(self, loader, device, n_samples=100):
        """
        计算Fisher信息矩阵的最小特征值
        
        使用随机幂迭代法近似
        """
        # 首先累积Fisher矩阵
        self.model.zero_grad()
        n_params = sum(p.numel() for p in self.model.parameters())
        
        # 使用少量样本近似
        samples = []
        for i, (x, y) in enumerate(loader):
            if i >= n_samples:
                break
            samples.append((x, y))
        
        # 计算Fisher
        F = torch.zeros(n_params, n_params, device=device)
        for x, y in samples:
            x, y = x.to(device), y.to(device)
            output = self.model(x)
            loss = nn.functional.cross_entropy(output, y)
            loss.backward()
            
            grad = torch.cat([
                p.grad.flatten() 
                for p in self.model.parameters() 
                if p.grad is not None
            ])
            
            F += torch.outer(grad, grad)
            self.model.zero_grad()
        
        F /= len(samples)
        
        # 使用幂迭代法估计最小特征值
        v = torch.randn(n_params, device=device)
        v = v / v.norm()
        
        for _ in range(50):  # 迭代次数
            v_new = F @ v
            # 归一化(排除零空间分量)
            norm = v_new.norm()
            if norm > 1e-10:
                v_new = v_new / norm
            v = v_new
        
        # Rayleigh商估计最小特征值
        lambda_min = (v @ F @ v).item()
        
        return max(lambda_min, 1e-10)  # 避免除零

8. 总结

核心定理汇总

定理内容应用
定理2联合多任务PAC-Bayes界多任务学习的泛化保证
定理3顺序学习泛化界持续学习的整体泛化分析
定理5正向迁移界迁移学习效果估计
定理8稳定-可塑性权衡最优策略的理论指导
定理11遗忘的PAC-Bayes界遗忘量上界估计
定理14相似性-遗忘关系任务相似性对遗忘的影响

实践建议

  1. 估计任务相似性:在训练前计算任务间的Hessian/梯度相似性
  2. 预测遗忘风险:基于相似性矩阵预测每个任务的遗忘量
  3. 优化正则化强度:使用定理17估计最优正则化参数
  4. 任务聚类:将相似任务分组,减少跨组干扰

理论启示

  • 遗忘不可避免:PAC-Bayes界显示遗忘量有不可减小的下界
  • 任务结构关键:任务相似性是决定遗忘量的核心因素
  • 权衡必要:稳定性和可塑性无法同时最大化

参考资料


相关阅读

Footnotes

  1. McAllester, D. (1999). PAC-Bayesian model averaging. COLT. 2

  2. Chaudhry et al. (2019). Efficient lifelong learning with A-GEM. ICLR. 2

  3. Pentina & Lampert (2015). A PAC-Bayesian bound for lifelong learning. ICML. 2