102_灾难性遗忘：你的模型正在“边学边忘”？破解微调稳定性困局

LLM灾难性遗忘的主要维度
├── 知识遗忘
│   ├── 事实知识丢失
│   ├── 概念关系混淆
│   └── 推理能力退化
├── 能力遗忘
│   ├── 指令理解能力下降
│   ├── 上下文处理能力减弱
│   └── 复杂任务执行能力降低
└── 行为变化
    ├── 输出风格偏移
    ├── 安全边界模糊
    └── 幻觉生成增加

class ForgettingDynamicsModel:
    """
    LLM微调过程中的遗忘动力学模拟模型
    """
    def __init__(self, model_size, initial_knowledge, forgetting_rate=0.1):
        self.model_size = model_size  # 模型大小（参数数量，以B为单位）
        self.initial_knowledge = initial_knowledge  # 初始知识量（0-1之间的分数）
        self.forgetting_rate = forgetting_rate  # 基础遗忘率
        self.knowledge_history = [initial_knowledge]  # 知识历史记录
        self.param_sensitivity = self._calculate_param_sensitivity()
        
    def _calculate_param_sensitivity(self):
        """
        计算参数敏感性，基于模型大小和架构特征
        较大的模型通常参数敏感性较低，遗忘率也较低
        """
        # 简化模型：假设参数敏感性与模型大小的平方根成反比
        base_sensitivity = 0.8
        size_factor = min(1.0, 1.0 / (self.model_size ** 0.5))
        return base_sensitivity * size_factor
    
    def update_knowledge(self, training_intensity, data_similarity, epochs):
        """
        更新知识水平，模拟微调过程中的知识变化
        
        参数：
        - training_intensity: 训练强度（0-1之间的分数）
        - data_similarity: 微调数据与预训练数据的相似度（0-1之间的分数）
        - epochs: 训练轮数
        
        返回：
        - 当前知识水平
        - 遗忘程度
        """
        # 获取当前知识水平
        current_knowledge = self.knowledge_history[-1]
        
        # 计算本次训练的实际遗忘率
        # 遗忘率与训练强度正相关，与数据相似度负相关
        effective_forgetting_rate = (self.forgetting_rate * 
                                   training_intensity * 
                                   (1 - data_similarity) *
                                   self.param_sensitivity)
        
        # 计算遗忘量
        knowledge_loss = current_knowledge * effective_forgetting_rate * epochs
        
        # 计算新任务知识获取量（简化模型）
        new_knowledge_gain = min(1.0 - current_knowledge, 
                                training_intensity * data_similarity * 0.3 * epochs)
        
        # 更新知识水平
        new_knowledge = max(0.0, min(1.0, current_knowledge - knowledge_loss + new_knowledge_gain))
        
        # 记录历史
        self.knowledge_history.append(new_knowledge)
        
        # 计算遗忘程度（相对于初始状态）
        forgetting_degree = max(0.0, self.initial_knowledge - new_knowledge)
        
        return {
            "current_knowledge": new_knowledge,
            "forgetting_degree": forgetting_degree,
            "knowledge_loss": knowledge_loss,
            "new_knowledge_gain": new_knowledge_gain,
            "effective_forgetting_rate": effective_forgetting_rate
        }
    
    def get_knowledge_trajectory(self):
        """
        获取知识水平随时间变化的轨迹
        """
        return self.knowledge_history

# 使用示例
def simulate_forgetting_scenarios():
    # 模拟不同大小模型的遗忘情况
    model_sizes = [7, 70, 175]  # 7B, 70B, 175B参数模型
    forgetting_models = {}
    
    for size in model_sizes:
        # 初始化模型，假设初始知识水平为0.85（预训练完成后的通用能力）
        models = {
            "high_similarity": ForgettingDynamicsModel(size, 0.85),
            "medium_similarity": ForgettingDynamicsModel(size, 0.85),
            "low_similarity": ForgettingDynamicsModel(size, 0.85)
        }
        forgetting_models[size] = models
    
    # 模拟10轮微调过程
    for epoch in range(1, 11):
        print(f"\n第{epoch}轮微调：")
        
        # 高相似度数据微调（如学术论文领域微调）
        print("\n高相似度数据微调结果：")
        for size in model_sizes:
            result = forgetting_models[size]["high_similarity"].update_knowledge(
                training_intensity=0.7,  # 较高训练强度
                data_similarity=0.8,     # 高数据相似度
                epochs=1
            )
            print(f"{size}B模型：遗忘程度={result['forgetting_degree']:.4f}, 当前知识={result['current_knowledge']:.4f}")
        
        # 中等相似度数据微调（如通用文档处理）
        print("\n中等相似度数据微调结果：")
        for size in model_sizes:
            result = forgetting_models[size]["medium_similarity"].update_knowledge(
                training_intensity=0.6,  # 中等训练强度
                data_similarity=0.5,     # 中等数据相似度
                epochs=1
            )
            print(f"{size}B模型：遗忘程度={result['forgetting_degree']:.4f}, 当前知识={result['current_knowledge']:.4f}")
        
        # 低相似度数据微调（如代码生成或专业领域）
        print("\n低相似度数据微调结果：")
        for size in model_sizes:
            result = forgetting_models[size]["low_similarity"].update_knowledge(
                training_intensity=0.8,  # 高训练强度
                data_similarity=0.3,     # 低数据相似度
                epochs=1
            )
            print(f"{size}B模型：遗忘程度={result['forgetting_degree']:.4f}, 当前知识={result['current_knowledge']:.4f}")

# 运行模拟
simulate_forgetting_scenarios()

import json
from datetime import datetime

class ForgettingEvaluationFramework:
    """
    LLM微调过程中的灾难性遗忘评估框架
    """
    def __init__(self, model_evaluator, reference_tasks, domain_tasks):
        self.model_evaluator = model_evaluator  # 模型评估器，用于执行各种基准测试
        self.reference_tasks = reference_tasks  # 参考任务集合（用于评估基础能力）
        self.domain_tasks = domain_tasks  # 领域特定任务集合（用于评估目标能力）
        self.evaluation_history = []  # 评估历史记录
    
    def evaluate_pre_finetuning(self, base_model):
        """
        微调前评估基准模型
        """
        print("执行微调前模型评估...")
        
        # 评估参考任务（基础能力）
        reference_results = {}
        for task_name, task_config in self.reference_tasks.items():
            print(f"  评估任务: {task_name}")
            result = self.model_evaluator.evaluate_task(
                model=base_model,
                task_name=task_name,
                task_config=task_config
            )
            reference_results[task_name] = result
        
        # 评估领域任务（初始能力）
        domain_results = {}
        for task_name, task_config in self.domain_tasks.items():
            print(f"  评估任务: {task_name}")
            result = self.model_evaluator.evaluate_task(
                model=base_model,
                task_name=task_name,
                task_config=task_config
            )
            domain_results[task_name] = result
        
        # 计算基础指标
        baseline_metrics = self._calculate_aggregated_metrics(reference_results, domain_results)
        
        # 记录基线评估
        baseline_evaluation = {
            "type": "baseline",
            "timestamp": datetime.now().isoformat(),
            "reference_results": reference_results,
            "domain_results": domain_results,
            "aggregated_metrics": baseline_metrics
        }
        
        self.evaluation_history.append(baseline_evaluation)
        return baseline_evaluation
    
    def evaluate_post_finetuning(self, finetuned_model, checkpoint_name="default"):
        """
        微调后评估模型
        """
        print("执行微调后模型评估...")
        
        # 获取基线评估
        baseline_evaluation = self._get_baseline_evaluation()
        if not baseline_evaluation:
            raise ValueError("请先执行微调前评估")
        
        # 评估参考任务（检查遗忘）
        reference_results = {}
        for task_name, task_config in self.reference_tasks.items():
            print(f"  评估任务: {task_name}")
            result = self.model_evaluator.evaluate_task(
                model=finetuned_model,
                task_name=task_name,
                task_config=task_config
            )
            reference_results[task_name] = result
        
        # 评估领域任务（检查改进）
        domain_results = {}
        for task_name, task_config in self.domain_tasks.items():
            print(f"  评估任务: {task_name}")
            result = self.model_evaluator.evaluate_task(
                model=finetuned_model,
                task_name=task_name,
                task_config=task_config
            )
            domain_results[task_name] = result
        
        # 计算聚合指标
        current_metrics = self._calculate_aggregated_metrics(reference_results, domain_results)
        
        # 计算遗忘指标
        forgetting_metrics = self._calculate_forgetting_metrics(
            baseline_evaluation["reference_results"], 
            reference_results
        )
        
        # 计算改进指标
        improvement_metrics = self._calculate_improvement_metrics(
            baseline_evaluation["domain_results"],
            domain_results
        )
        
        # 计算综合评估分数
        overall_scores = self._calculate_overall_scores(current_metrics, forgetting_metrics, improvement_metrics)
        
        # 记录当前评估
        current_evaluation = {
            "type": "checkpoint",
            "name": checkpoint_name,
            "timestamp": datetime.now().isoformat(),
            "reference_results": reference_results,
            "domain_results": domain_results,
            "aggregated_metrics": current_metrics,
            "forgetting_metrics": forgetting_metrics,
            "improvement_metrics": improvement_metrics,
            "overall_scores": overall_scores
        }
        
        self.evaluation_history.append(current_evaluation)
        return current_evaluation
    
    def _get_baseline_evaluation(self):
        """
        获取基线评估结果
        """
        for eval_result in self.evaluation_history:
            if eval_result["type"] == "baseline":
                return eval_result
        return None
    
    def _calculate_aggregated_metrics(self, reference_results, domain_results):
        """
        计算聚合评估指标
        """
        # 计算参考任务的平均分数
        reference_scores = []
        for task_name, result in reference_results.items():
            if "score" in result:
                reference_scores.append(result["score"])
        avg_reference_score = sum(reference_scores) / len(reference_scores) if reference_scores else 0
        
        # 计算领域任务的平均分数
        domain_scores = []
        for task_name, result in domain_results.items():
            if "score" in result:
                domain_scores.append(result["score"])
        avg_domain_score = sum(domain_scores) / len(domain_scores) if domain_scores else 0
        
        # 计算综合分数
        overall_score = (avg_reference_score + avg_domain_score) / 2
        
        return {
            "avg_reference_score": avg_reference_score,
            "avg_domain_score": avg_domain_score,
            "overall_score": overall_score,
            "reference_task_count": len(reference_results),
            "domain_task_count": len(domain_results)
        }
    
    def _calculate_forgetting_metrics(self, baseline_results, current_results):
        """
        计算遗忘指标
        """
        forgetting_scores = {}
        task_forgetting_rates = {}
        
        # 计算每个任务的遗忘率
        total_baseline_score = 0
        total_current_score = 0
        
        for task_name, baseline_result in baseline_results.items():
            if task_name in current_results and "score" in baseline_result and "score" in current_results[task_name]:
                baseline_score = baseline_result["score"]
                current_score = current_results[task_name]["score"]
                
                # 计算遗忘率（分数下降的比例）
                if baseline_score > 0:
                    forgetting_rate = max(0, (baseline_score - current_score) / baseline_score)
                else:
                    forgetting_rate = 0
                
                task_forgetting_rates[task_name] = forgetting_rate
                total_baseline_score += baseline_score
                total_current_score += current_score
        
        # 计算平均遗忘率
        avg_forgetting_rate = sum(task_forgetting_rates.values()) / len(task_forgetting_rates) if task_forgetting_rates else 0
        
        # 计算整体遗忘程度
        overall_forgetting = max(0, (total_baseline_score - total_current_score) / total_baseline_score) if total_baseline_score > 0 else 0
        
        # 识别高遗忘任务（遗忘率超过30%）
        high_forgetting_tasks = [task_name for task_name, rate in task_forgetting_rates.items() if rate > 0.3]
        
        return {
            "avg_forgetting_rate": avg_forgetting_rate,
            "overall_forgetting": overall_forgetting,
            "task_forgetting_rates": task_forgetting_rates,
            "high_forgetting_tasks": high_forgetting_tasks,
            "high_forgetting_count": len(high_forgetting_tasks)
        }
    
    def _calculate_improvement_metrics(self, baseline_results, current_results):
        """
        计算改进指标
        """
        improvement_scores = {}
        task_improvement_rates = {}
        
        # 计算每个任务的改进率
        total_baseline_score = 0
        total_current_score = 0
        
        for task_name, baseline_result in baseline_results.items():
            if task_name in current_results and "score" in baseline_result and "score" in current_results[task_name]:
                baseline_score = baseline_result["score"]
                current_score = current_results[task_name]["score"]
                
                # 计算改进率
                if baseline_score > 0:
                    improvement_rate = (current_score - baseline_score) / baseline_score
                else:
                    # 如果基线分数为0，使用绝对改进
                    improvement_rate = current_score
                
                task_improvement_rates[task_name] = improvement_rate
                total_baseline_score += baseline_score
                total_current_score += current_score
        
        # 计算平均改进率
        avg_improvement_rate = sum(task_improvement_rates.values()) / len(task_improvement_rates) if task_improvement_rates else 0
        
        # 计算整体改进程度
        overall_improvement = (total_current_score - total_baseline_score) / total_baseline_score if total_baseline_score > 0 else total_current_score
        
        # 识别显著改进任务（改进率超过20%）
        significant_improvement_tasks = [task_name for task_name, rate in task_improvement_rates.items() if rate > 0.2]
        
        return {
            "avg_improvement_rate": avg_improvement_rate,
            "overall_improvement": overall_improvement,
            "task_improvement_rates": task_improvement_rates,
            "significant_improvement_tasks": significant_improvement_tasks,
            "significant_improvement_count": len(significant_improvement_tasks)
        }
    
    def _calculate_overall_scores(self, current_metrics, forgetting_metrics, improvement_metrics):
        """
        计算综合评估分数
        """
        # 计算平衡分数（考虑改进和遗忘的平衡）
        # 权重可以根据项目需求调整
        forgetting_weight = 0.6  # 遗忘的权重
        improvement_weight = 0.4  # 改进的权重
        
        # 将遗忘率转换为0-1的分数（遗忘越少分数越高）
        forgetting_score = 1 - forgetting_metrics["overall_forgetting"]
        
        # 确保改进分数在0-1之间
        improvement_score = min(1.0, 0.5 + improvement_metrics["overall_improvement"] * 0.5)
        
        # 计算平衡分数
        balance_score = (forgetting_weight * forgetting_score + 
                        improvement_weight * improvement_score)
        
        # 计算稳定性-有效性综合分数
        stability_efficiency_score = current_metrics["overall_score"] * balance_score
        
        # 评估是否达到目标
        is_successful = (
            forgetting_metrics["overall_forgetting"] < 0.2 and  # 遗忘率低于20%
            improvement_metrics["overall_improvement"] > 0.1 and  # 改进率高于10%
            current_metrics["avg_reference_score"] > 0.7  # 基础能力分数高于0.7
        )
        
        return {
            "balance_score": balance_score,
            "stability_efficiency_score": stability_efficiency_score,
            "is_successful": is_successful,
            "forgetting_score": forgetting_score,
            "improvement_score": improvement_score
        }
    
    def generate_evaluation_report(self, output_file=None):
        """
        生成综合评估报告
        """
        if not self.evaluation_history:
            raise ValueError("没有评估历史记录")
        
        # 获取基线评估
        baseline_evaluation = self._get_baseline_evaluation()
        if not baseline_evaluation:
            raise ValueError("找不到基线评估记录")
        
        # 收集所有检查点评估
        checkpoint_evaluations = [eval for eval in self.evaluation_history if eval["type"] == "checkpoint"]
        
        # 生成报告
        report = {
            "report_title": "LLM灾难性遗忘评估报告",
            "generation_time": datetime.now().isoformat(),
            "baseline_evaluation": baseline_evaluation["aggregated_metrics"],
            "checkpoints": [],
            "summary": {}
        }
        
        # 添加检查点评估结果
        for checkpoint in checkpoint_evaluations:
            report["checkpoints"].append({
                "name": checkpoint["name"],
                "timestamp": checkpoint["timestamp"],
                "metrics": checkpoint["aggregated_metrics"],
                "forgetting": checkpoint["forgetting_metrics"],
                "improvement": checkpoint["improvement_metrics"],
                "scores": checkpoint["overall_scores"]
            })
        
        # 计算总结指标
        if checkpoint_evaluations:
            # 获取最佳检查点（稳定性-有效性综合分数最高）
            best_checkpoint = max(checkpoint_evaluations, key=lambda x: x["overall_scores"]["stability_efficiency_score"])
            
            # 计算平均遗忘率
            avg_forgetting = sum(c["forgetting_metrics"]["overall_forgetting"] for c in checkpoint_evaluations) / len(checkpoint_evaluations)
            
            # 计算平均改进率
            avg_improvement = sum(c["improvement_metrics"]["overall_improvement"] for c in checkpoint_evaluations) / len(checkpoint_evaluations)
            
            report["summary"] = {
                "best_checkpoint": best_checkpoint["name"],
                "best_scores": best_checkpoint["overall_scores"],
                "avg_forgetting_rate": avg_forgetting,
                "avg_improvement_rate": avg_improvement,
                "checkpoint_count": len(checkpoint_evaluations),
                "success_rate": sum(1 for c in checkpoint_evaluations if c["overall_scores"]["is_successful"]) / len(checkpoint_evaluations)
            }
        
        # 保存报告
        if output_file:
            with open(output_file, "w", encoding="utf-8") as f:
                json.dump(report, f, indent=2, ensure_ascii=False)
            print(f"评估报告已保存至：{output_file}")
        
        return report

# 使用示例
# 需要实现一个ModelEvaluator类来执行实际评估
# evaluator = ForgettingEvaluationFramework(
#     model_evaluator=MyModelEvaluator(),
#     reference_tasks=reference_tasks_config,
#     domain_tasks=domain_tasks_config
# )
# baseline = evaluator.evaluate_pre_finetuning(base_model)
# checkpoint1 = evaluator.evaluate_post_finetuning(finetuned_model, "epoch_10")
# report = evaluator.generate_evaluation_report("forgetting_evaluation_report.json")

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, Trainer, TrainingArguments

class L2RegularizedTrainer(Trainer):
    """
    支持L2正则化的自定义Trainer类
    """
    def __init__(self, model, args, l2_lambda=1e-4, **kwargs):
        super().__init__(model, args, **kwargs)
        self.l2_lambda = l2_lambda  # L2正则化系数
        
        # 保存预训练模型的参数作为参考（在微调过程中保持相对稳定）
        self.pretrained_params = {}
        for name, param in model.named_parameters():
            if param.requires_grad:
                self.pretrained_params[name] = param.data.clone().detach()
    
    def compute_loss(self, model, inputs, return_outputs=False):
        """
        计算包含L2正则化项的损失
        """
        # 计算标准损失
        loss, outputs = super().compute_loss(model, inputs, return_outputs=True)
        
        # 添加L2正则化损失
        l2_loss = 0.0
        for name, param in model.named_parameters():
            if param.requires_grad and name in self.pretrained_params:
                # 计算当前参数与预训练参数之间的平方差
                param_diff = param - self.pretrained_params[name]
                l2_loss += torch.sum(param_diff ** 2)
        
        # 应用正则化系数
        l2_loss = self.l2_lambda * l2_loss
        
        # 总损失 = 原始损失 + L2正则化损失
        total_loss = loss + l2_loss
        
        return (total_loss, outputs) if return_outputs else total_loss

# 使用示例
def train_with_l2_regularization():
    # 加载预训练模型
    model_name = "meta-llama/Llama-2-7b-hf"  # 示例模型
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForCausalLM.from_pretrained(model_name)
    
    # 准备训练数据（这里仅为示例，实际需要准备真实数据）
    train_dataset = ...  # 加载训练数据集
    eval_dataset = ...   # 加载评估数据集
    
    # 定义训练参数
    training_args = TrainingArguments(
        output_dir="./results_l2_reg",
        num_train_epochs=3,
        per_device_train_batch_size=4,
        per_device_eval_batch_size=4,
        warmup_steps=500,
        weight_decay=0.01,  # 注意：这里的weight_decay是PyTorch的权重衰减，与我们的L2正则化不同
        logging_dir="./logs",
        logging_steps=100,
        evaluation_strategy="steps",
        eval_steps=500,
        save_strategy="steps",
        save_steps=1000,
        learning_rate=2e-5,
        fp16=True,
    )
    
    # 创建自定义Trainer实例，设置L2正则化系数
    l2_lambda = 5e-5  # L2正则化系数，根据经验调整
    trainer = L2RegularizedTrainer(
        model=model,
        args=training_args,
        train_dataset=train_dataset,
        eval_dataset=eval_dataset,
        tokenizer=tokenizer,
        l2_lambda=l2_lambda
    )
    
    # 开始训练
    print(f"开始训练，使用L2正则化（系数：{l2_lambda}）...")
    trainer.train()
    
    # 保存模型
    trainer.save_model("./finetuned_model_l2_reg")
    print("训练完成，模型已保存")

# 运行示例
# train_with_l2_regularization()

class EWCTrainer(Trainer):
    """
    支持弹性权重整合(EWC)的自定义Trainer类
    """
    def __init__(self, model, args, ewc_lambda=5e4, fisher_n_samples=1000, **kwargs):
        super().__init__(model, args, **kwargs)
        self.ewc_lambda = ewc_lambda  # EWC正则化系数
        self.fisher_n_samples = fisher_n_samples  # 用于估计Fisher矩阵的样本数量
        
        # 保存预训练参数作为参考
        self.pretrained_params = {}
        for name, param in model.named_parameters():
            if param.requires_grad:
                self.pretrained_params[name] = param.data.clone().detach()
        
        # 初始化Fisher矩阵
        self.fisher_diag = None
    
    def compute_fisher(self, dataset, device="cuda"):
        """
        计算Fisher信息矩阵的对角线近似
        """
        print("计算Fisher信息矩阵...")
        
        # 初始化Fisher矩阵（对角线）
        fisher_diag = {}
        for name, param in self.model.named_parameters():
            if param.requires_grad:
                fisher_diag[name] = torch.zeros_like(param, device=device)
        
        # 将模型设置为评估模式
        self.model.eval()
        
        # 选择样本用于计算
        sample_indices = torch.randperm(len(dataset))[:self.fisher_n_samples]
        dataloader = torch.utils.data.DataLoader(
            torch.utils.data.Subset(dataset, sample_indices),
            batch_size=self.args.per_device_train_batch_size,
            shuffle=False,
            collate_fn=self.data_collator
        )
        
        # 计算每个样本的梯度并累加
        for batch in dataloader:
            # 将批次移动到设备
            batch = {k: v.to(device) for k, v in batch.items()}
            
            # 前向传播
            outputs = self.model(**batch)
            logits = outputs.logits
            
            # 计算分类概率
            probs = torch.softmax(logits, dim=-1)
            
            # 对于每个位置，使用预测概率作为标签分布（MCE准则）
            batch_size, seq_len, vocab_size = logits.size()
            for i in range(batch_size):
                # 跳过padding部分
                non_pad_indices = batch["attention_mask"][i].nonzero().squeeze()
                for idx in non_pad_indices:
                    if idx < seq_len - 1:  # 避免最后一个位置
                        # 获取当前位置的预测分布
                        prob = probs[i, idx]
                        
                        # 计算损失（使用当前分布作为目标）
                        loss = -torch.sum(prob * torch.log(prob + 1e-8))
                        
                        # 反向传播
                        self.model.zero_grad()
                        loss.backward(retain_graph=True)
                        
                        # 累加梯度平方到Fisher矩阵
                        for name, param in self.model.named_parameters():
                            if param.requires_grad and param.grad is not None:
                                fisher_diag[name] += param.grad.data ** 2
        
        # 归一化Fisher矩阵
        for name in fisher_diag:
            fisher_diag[name] /= self.fisher_n_samples
        
        self.fisher_diag = fisher_diag
        print("Fisher信息矩阵计算完成")
        return fisher_diag
    
    def compute_loss(self, model, inputs, return_outputs=False):
        """
        计算包含EWC正则化项的损失
        """
        # 确保已经计算了Fisher矩阵
        if self.fisher_diag is None:
            raise ValueError("请先调用compute_fisher()方法")
        
        # 计算标准损失
        loss, outputs = super().compute_loss(model, inputs, return_outputs=True)
        
        # 添加EWC正则化损失
        ewc_loss = 0.0
        for name, param in model.named_parameters():
            if param.requires_grad and name in self.fisher_diag and name in self.pretrained_params:
                # 计算当前参数与预训练参数之间的平方差，乘以Fisher信息
                param_diff = param - self.pretrained_params[name]
                ewc_loss += 0.5 * torch.sum(self.fisher_diag[name] * (param_diff ** 2))
        
        # 应用正则化系数
        ewc_loss = self.ewc_lambda * ewc_loss
        
        # 总损失 = 原始损失 + EWC正则化损失
        total_loss = loss + ewc_loss
        
        return (total_loss, outputs) if return_outputs else total_loss

# 使用示例
def train_with_ewc():
    # 加载预训练模型
    model_name = "meta-llama/Llama-2-7b-hf"  # 示例模型
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForCausalLM.from_pretrained(model_name)
    
    # 准备训练数据
    train_dataset = ...  # 加载训练数据集
    eval_dataset = ...   # 加载评估数据集
    
    # 定义训练参数
    training_args = TrainingArguments(
        output_dir="./results_ewc",
        num_train_epochs=3,
        per_device_train_batch_size=4,
        per_device_eval_batch_size=4,
        warmup_steps=500,
        weight_decay=0.01,
        logging_dir="./logs",
        logging_steps=100,
        evaluation_strategy="steps",
        eval_steps=500,
        save_strategy="steps",
        save_steps=1000,
        learning_rate=2e-5,
        fp16=True,
    )
    
    # 创建EWC Trainer实例
    ewc_lambda = 5e4  # EWC正则化系数
    fisher_n_samples = 5000  # Fisher矩阵估计样本数
    trainer = EWCTrainer(
        model=model,
        args=training_args,
        train_dataset=train_dataset,
        eval_dataset=eval_dataset,
        tokenizer=tokenizer,
        ewc_lambda=ewc_lambda,
        fisher_n_samples=fisher_n_samples
    )
    
    # 计算Fisher矩阵
    reference_dataset = ...  # 预训练任务的参考数据集
    trainer.compute_fisher(reference_dataset)
    
    # 开始训练
    print(f"开始训练，使用EWC正则化（系数：{ewc_lambda}）...")
    trainer.train()
    
    # 保存模型
    trainer.save_model("./finetuned_model_ewc")
    print("训练完成，模型已保存")

# 运行示例
# train_with_ewc()

import torch
import random
import os
from transformers import AutoModelForCausalLM, AutoTokenizer, Trainer, TrainingArguments
from torch.utils.data import Dataset, ConcatDataset, Subset

class ReplayDataset(Dataset):
    """
    支持数据重放的数据集类
    """
    def __init__(self, primary_dataset, replay_dataset, replay_ratio=0.2):
        """
        初始化数据重放数据集
        
        参数：
        - primary_dataset: 主要训练数据集（新任务）
        - replay_dataset: 重放数据集（预训练或先前任务）
        - replay_ratio: 重放数据在批次中的比例
        """
        self.primary_dataset = primary_dataset
        self.replay_dataset = replay_dataset
        self.replay_ratio = replay_ratio
    
    def __len__(self):
        # 返回主要数据集的长度，因为我们会在__getitem__中动态混合数据
        return len(self.primary_dataset)
    
    def __getitem__(self, idx):
        # 获取主要数据项
        primary_item = self.primary_dataset[idx]
        
        # 确定是否使用重放数据（根据重放比例）
        if random.random() < self.replay_ratio and self.replay_dataset:
            # 随机选择一个重放数据项
            replay_idx = random.randint(0, len(self.replay_dataset) - 1)
            replay_item = self.replay_dataset[replay_idx]
            
            # 标记为数据类型
            replay_item["data_type"] = "replay"
            return replay_item
        else:
            # 标记为主要数据类型
            primary_item["data_type"] = "primary"
            return primary_item

class PrioritizedReplayTrainer(Trainer):
    """
    支持优先级数据重放的自定义Trainer类
    """
    def __init__(self, model, args, primary_dataset, replay_dataset,
                 replay_ratio_schedule=None, importance_estimator=None, **kwargs):
        """
        初始化优先级重放Trainer
        
        参数：
        - model: 模型实例
        - args: 训练参数
        - primary_dataset: 主要训练数据集
        - replay_dataset: 重放数据集
        - replay_ratio_schedule: 重放比例的调度策略（可选）
        - importance_estimator: 数据重要性估计器（可选）
        """
        # 创建重放数据集
        self.initial_replay_ratio = 0.2  # 默认初始重放比例
        self.current_replay_ratio = self.initial_replay_ratio
        self.replay_ratio_schedule = replay_ratio_schedule or self._get_default_schedule()
        self.importance_estimator = importance_estimator
        
        # 根据重要性对重放数据进行排序（如果提供了估计器）
        self.prioritized_replay_data = self._prioritize_replay_data(replay_dataset)
        
        # 创建动态重放数据集
        self.replay_dataset = ReplayDataset(
            primary_dataset=primary_dataset,
            replay_dataset=self.prioritized_replay_data,
            replay_ratio=self.current_replay_ratio
        )
        
        super().__init__(model, args, train_dataset=self.replay_dataset, **kwargs)
    
    def _get_default_schedule(self):
        """
        获取默认的重放比例调度策略
        在训练初期使用较高的重放比例，随着训练进行逐渐降低
        """
        def default_schedule(global_step, total_steps):
            # 线性衰减的重放比例
            progress = min(1.0, global_step / total_steps)
            # 从初始比例线性衰减到初始比例的20%
            return self.initial_replay_ratio * (1 - 0.8 * progress)
        
        return default_schedule
    
    def _prioritize_replay_data(self, replay_dataset):
        """
        根据重要性对重放数据进行优先级排序
        """
        if not self.importance_estimator or not replay_dataset:
            return replay_dataset
        
        print("正在评估重放数据的重要性...")
        
        # 评估每个数据项的重要性
        importance_scores = []
        for idx in range(len(replay_dataset)):
            item = replay_dataset[idx]
            importance = self.importance_estimator.estimate_importance(item)
            importance_scores.append((idx, importance))
        
        # 按重要性降序排序
        importance_scores.sort(key=lambda x: x[1], reverse=True)
        
        # 创建优先级排序后的数据集
        prioritized_indices = [idx for idx, _ in importance_scores]
        prioritized_dataset = Subset(replay_dataset, prioritized_indices)
        
        print("重放数据优先级排序完成")
        return prioritized_dataset
    
    def training_step(self, model, inputs):
        """
        执行单个训练步骤，并动态调整重放比例
        """
        # 获取当前全局步数
        current_step = self.state.global_step
        total_steps = self.args.max_steps if self.args.max_steps > 0 else \
                     (len(self.train_dataset) * self.args.num_train_epochs) // \
                     (self.args.per_device_train_batch_size * self.args.gradient_accumulation_steps)
        
        # 动态调整重放比例
        new_replay_ratio = self.replay_ratio_schedule(current_step, total_steps)
        if abs(new_replay_ratio - self.current_replay_ratio) > 0.01:
            self.current_replay_ratio = new_replay_ratio
            self.replay_dataset.replay_ratio = new_replay_ratio
            if current_step % 100 == 0:  # 定期打印以减少日志
                print(f"步骤 {current_step}: 重放比例调整为 {new_replay_ratio:.4f}")
        
        # 执行标准训练步骤
        return super().training_step(model, inputs)

class SimpleImportanceEstimator:
    """
    简单的数据重要性估计器
    """
    def __init__(self, model, tokenizer, importance_criteria=None):
        self.model = model
        self.tokenizer = tokenizer
        self.importance_criteria = importance_criteria or self._get_default_criteria()
    
    def _get_default_criteria(self):
        """
        获取默认的重要性评估标准
        """
        def default_criteria(text):
            # 简单的启发式规则：
            # 1. 较长的文本通常包含更多信息
            # 2. 包含特定关键词的文本可能更重要
            importance = 0.0
            
            # 长度因素（归一化到0-1）
            length = len(text.split())
            length_factor = min(1.0, length / 200)  # 假设200词为长文本
            importance += length_factor * 0.4
            
            # 关键词因素
            important_keywords = ["关键", "重要", "基础", "原理", "定义", "核心", "原则"]
            keyword_count = sum(1 for keyword in important_keywords if keyword in text)
            keyword_factor = min(1.0, keyword_count / 5)  # 最多5个关键词
            importance += keyword_factor * 0.6
            
            return importance
        
        return default_criteria
    
    def estimate_importance(self, data_item):
        """
        估计单个数据项的重要性
        """
        # 提取文本内容（假设数据项中有text字段）
        text = data_item.get("text", "")
        
        # 如果没有文本，返回0重要性
        if not text:
            return 0.0
        
        # 应用重要性评估标准
        importance = self.importance_criteria(text)
        
        return importance

# 使用示例
def train_with_prioritized_replay():
    # 加载预训练模型
    model_name = "meta-llama/Llama-2-7b-hf"  # 示例模型
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForCausalLM.from_pretrained(model_name)
    
    # 准备数据集
    primary_dataset = ...  # 主要训练数据集（新任务）
    replay_dataset = ...   # 重放数据集（预训练或先前任务）
    eval_dataset = ...     # 评估数据集
    
    # 创建重要性估计器
    importance_estimator = SimpleImportanceEstimator(model, tokenizer)
    
    # 定义训练参数
    training_args = TrainingArguments(
        output_dir="./results_replay",
        num_train_epochs=3,
        per_device_train_batch_size=4,
        per_device_eval_batch_size=4,
        warmup_steps=500,
        weight_decay=0.01,
        logging_dir="./logs",
        logging_steps=100,
        evaluation_strategy="steps",
        eval_steps=500,
        save_strategy="steps",
        save_steps=1000,
        learning_rate=2e-5,
        fp16=True,
    )
    
    # 自定义重放比例调度策略
    def custom_replay_schedule(global_step, total_steps):
        # 余弦退火式重放比例
        import math
        progress = min(1.0, global_step / total_steps)
        # 从0.3降到0.1，使用余弦退火
        return 0.1 + 0.2 * (1 + math.cos(math.pi * progress)) / 2
    
    # 创建优先级重放Trainer
    trainer = PrioritizedReplayTrainer(
        model=model,
        args=training_args,
        primary_dataset=primary_dataset,
        replay_dataset=replay_dataset,
        eval_dataset=eval_dataset,
        tokenizer=tokenizer,
        importance_estimator=importance_estimator,
        replay_ratio_schedule=custom_replay_schedule
    )
    
    # 开始训练
    print("开始训练，使用优先级数据重放...")
    trainer.train()
    
    # 保存模型
    trainer.save_model("./finetuned_model_replay")
    print("训练完成，模型已保存")

# 运行示例
# train_with_prioritized_replay()

def freeze_model_layers(model, freeze_layers_ratio=0.7):
    """
    冻结模型的底层参数
    
    参数：
    - model: 预训练模型
    - freeze_layers_ratio: 要冻结的层数比例（0-1之间）
    
    返回：
    - 冻结后的模型
    - 冻结的层数
    - 可训练的参数量
    """
    # 计算要冻结的层数
    num_layers = len(model.model.layers)  # 对于Llama等模型
    num_layers_to_freeze = int(num_layers * freeze_layers_ratio)
    
    # 冻结底层
    for i in range(num_layers_to_freeze):
        for param in model.model.layers[i].parameters():
            param.requires_grad = False
    
    # 可选择冻结嵌入层（通常包含丰富的语义知识）
    for param in model.model.embed_tokens.parameters():
        param.requires_grad = False
    
    # 计算可训练的参数量
    trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
    total_params = sum(p.numel() for p in model.parameters())
    trainable_ratio = trainable_params / total_params
    
    print(f"冻结了 {num_layers_to_freeze}/{num_layers} 层 (比例: {freeze_layers_ratio:.2f})")
    print(f"可训练参数量: {trainable_params:,} / {total_params:,} ({trainable_ratio:.2%})")
    
    return model, num_layers_to_freeze, trainable_params

# 使用示例
# model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-2-7b-hf")
# model, frozen_layers, trainable_params = freeze_model_layers(model, freeze_layers_ratio=0.8)

def freeze_model_components(model, freeze_attention=True, freeze_mlp=False, freeze_embeddings=True):
    """
    根据组件类型选择性地冻结模型参数
    
    参数：
    - model: 预训练模型
    - freeze_attention: 是否冻结注意力层
    - freeze_mlp: 是否冻结MLP层
    - freeze_embeddings: 是否冻结嵌入层
    
    返回：
    - 冻结后的模型
    - 可训练的参数量
    """
    # 冻结嵌入层
    if freeze_embeddings:
        for param in model.model.embed_tokens.parameters():
            param.requires_grad = False
        if hasattr(model.model, 'embed_positions'):
            for param in model.model.embed_positions.parameters():
                param.requires_grad = False
    
    # 冻结每一层的组件
    for layer in model.model.layers:
        # 冻结注意力层
        if freeze_attention:
            for param in layer.self_attn.parameters():
                param.requires_grad = False
        
        # 冻结MLP层
        if freeze_mlp:
            for param in layer.mlp.parameters():
                param.requires_grad = False
    
    # 计算可训练的参数量
    trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
    total_params = sum(p.numel() for p in model.parameters())
    trainable_ratio = trainable_params / total_params
    
    print(f"组件冻结配置:")
    print(f"  - 冻结注意力层: {freeze_attention}")
    print(f"  - 冻结MLP层: {freeze_mlp}")
    print(f"  - 冻结嵌入层: {freeze_embeddings}")
    print(f"可训练参数量: {trainable_params:,} / {total_params:,} ({trainable_ratio:.2%})")
    
    return model, trainable_params

# 使用示例
# model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-2-7b-hf")
# model, trainable_params = freeze_model_components(
#     model, 
#     freeze_attention=True,  # 冻结注意力层（通常包含更多通用知识）
#     freeze_mlp=False,       # 允许MLP层更新（更灵活地适应新任务）
#     freeze_embeddings=True  # 冻结嵌入层
# )

from peft import LoraConfig, get_peft_model, TaskType

def apply_lora_adapter(model, lora_r=8, lora_alpha=16, target_modules=None, lora_dropout=0.05):
    """
    为模型应用LoRA适配器
    
    参数：
    - model: 预训练模型
    - lora_r: LoRA的秩，控制适配器容量
    - lora_alpha: LoRA的缩放因子
    - target_modules: 应用LoRA的目标模块
    - lora_dropout: LoRA的dropout率
    
    返回：
    - 应用LoRA后的模型
    - 可训练的参数量
    """
    # 设置默认目标模块（适用于Llama等模型）
    if target_modules is None:
        target_modules = ["q_proj", "v_proj"]  # 通常在注意力层的查询和值投影上应用
    
    # 配置LoRA
    lora_config = LoraConfig(
        task_type=TaskType.CAUSAL_LM,
        inference_mode=False,
        r=lora_r,
        lora_alpha=lora_alpha,
        target_modules=target_modules,
        lora_dropout=lora_dropout
    )
    
    # 创建应用LoRA的模型
    peft_model = get_peft_model(model, lora_config)
    
    # 计算可训练的参数量
    trainable_params = sum(p.numel() for p in peft_model.parameters() if p.requires_grad)
    total_params = sum(p.numel() for p in peft_model.parameters())
    trainable_ratio = trainable_params / total_params
    
    print(f"LoRA适配器配置:")
    print(f"  - 秩(r): {lora_r}")
    print(f"  - 缩放因子(alpha): {lora_alpha}")
    print(f"  - 目标模块: {target_modules}")
    print(f"  - Dropout: {lora_dropout}")
    print(f"可训练参数量: {trainable_params:,} / {total_params:,} ({trainable_ratio:.2%})")
    
    return peft_model, trainable_params

# 使用示例
# model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-2-7b-hf")
# model, trainable_params = apply_lora_adapter(
#     model,
#     lora_r=8,
#     lora_alpha=16,
#     target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],  # 扩展到所有注意力投影
#     lora_dropout=0.05
# )

class ParallelAdapterManager:
    """
    管理多个领域或任务适配器的并行适配器管理器
    """
    def __init__(self, base_model, adapter_configs):
        """
        初始化并行适配器管理器
        
        参数：
        - base_model: 基础预训练模型
        - adapter_configs: 适配器配置字典，格式为{"adapter_name": lora_config}
        """
        self.base_model = base_model
        self.adapter_configs = adapter_configs
        self.adapters = {}
        self.active_adapter = None
        
        # 创建各个适配器
        for adapter_name, config in adapter_configs.items():
            print(f"创建适配器: {adapter_name}")
            # 为每个适配器创建独立的模型副本
            adapter_model = get_peft_model(base_model, config)
            self.adapters[adapter_name] = adapter_model
    
    def activate_adapter(self, adapter_name):
        """
        激活指定的适配器
        """
        if adapter_name not in self.adapters:
            raise ValueError(f"未知的适配器: {adapter_name}")
        
        self.active_adapter = adapter_name
        print(f"已激活适配器: {adapter_name}")
        return self.adapters[adapter_name]
    
    def train_adapter(self, adapter_name, train_dataset, training_args, **kwargs):
        """
        训练指定的适配器
        """
        # 激活目标适配器
        model = self.activate_adapter(adapter_name)
        
        # 创建Trainer
        trainer = Trainer(
            model=model,
            args=training_args,
            train_dataset=train_dataset,
            **kwargs
        )
        
        # 开始训练
        print(f"开始训练适配器: {adapter_name}")
        trainer.train()
        
        # 更新适配器
        self.adapters[adapter_name] = model
        
        return trainer
    
    def save_adapter(self, adapter_name, save_dir):
        """
        保存指定的适配器
        
        参数：
        - adapter_name: 适配器名称
        - save_dir: 保存目录
        """
        if adapter_name not in self.adapters:
            raise ValueError(f"未知的适配器: {adapter_name}")
        
        # 创建适配器专用保存目录
        adapter_save_dir = os.path.join(save_dir, adapter_name)
        os.makedirs(adapter_save_dir, exist_ok=True)
        
        # 保存适配器
        model = self.adapters[adapter_name]
        model.save_pretrained(adapter_save_dir)
        print(f"适配器 {adapter_name} 已保存到 {adapter_save_dir}")
    
    def load_adapter(self, adapter_name, load_dir):
        """
        加载指定的适配器
        
        参数：
        - adapter_name: 适配器名称
        - load_dir: 加载目录
        """
        # 加载适配器目录
        adapter_load_dir = os.path.join(load_dir, adapter_name)
        if not os.path.exists(adapter_load_dir):
            raise FileNotFoundError(f"适配器 {adapter_name} 目录不存在: {adapter_load_dir}")
        
        # 重新创建适配器模型并加载权重
        config = self.adapter_configs.get(adapter_name)
        if not config:
            raise ValueError(f"适配器 {adapter_name} 的配置不存在")
        
        # 创建新的适配器模型
        model = get_peft_model(self.base_model, config)
        # 加载保存的权重
        model = model.from_pretrained(adapter_load_dir)
        
        # 更新适配器字典
        self.adapters[adapter_name] = model
        print(f"适配器 {adapter_name} 已从 {adapter_load_dir} 加载")
        
        return model

class LLMContinualLearningFramework:
    """
    LLM持续学习框架，整合多种技术来缓解灾难性遗忘
    """
    def __init__(self, base_model, tokenizer, config=None):
        """
        初始化持续学习框架
        
        参数：
        - base_model: 基础预训练模型
        - tokenizer: 分词器
        - config: 配置字典，包含各种技术的参数
        """
        self.base_model = base_model
        self.tokenizer = tokenizer
        self.config = config or self._get_default_config()
        
        # 初始化组件
        self.replay_buffer = None  # 重放缓冲区
        self.regularizer = None    # 正则化器
        self.parameter_isolator = None  # 参数隔离器
        self.evaluation_framework = None  # 评估框架
        
        # 历史知识记录
        self.task_history = []
        self.knowledge_metrics = {}
        
        # 初始化各个组件
        self._initialize_components()
    
    def _get_default_config(self):
        """
        获取默认配置
        """
        return {
            "replay": {
                "enabled": True,
                "buffer_size": 10000,
                "replay_ratio": 0.2,
                "prioritization": True
            },
            "regularization": {
                "enabled": True,
                "method": "ewc",  # ewc, l2
                "lambda": 1e4,
                "fisher_sample_size": 1000
            },
            "parameter_isolation": {
                "enabled": True,
                "strategy": "freeze",  # freeze, adapter
                "freeze_ratio": 0.8,
                "adapter_config": {
                    "type": "lora",
                    "r": 8,
                    "alpha": 16
                }
            },
            "evaluation": {
                "enabled": True,
                "metrics": ["knowledge_retention", "task_performance", "forgetting_rate"],
                "eval_interval": 100
            }
        }
    
    def _initialize_components(self):
        """
        初始化框架组件
        """
        # 初始化重放缓冲区
        if self.config["replay"]["enabled"]:
            from datasets import Dataset
            self.replay_buffer = ReplayDataset(
                primary_dataset=None,  # 稍后设置
                replay_dataset=Dataset.from_dict({}),
                replay_ratio=self.config["replay"]["replay_ratio"]
            )
        
        # 初始化正则化器
        if self.config["regularization"]["enabled"]:
            method = self.config["regularization"]["method"]
            if method == "ewc":
                self.regularizer = EWCTrainer(
                    model=self.base_model,
                    lambda_=self.config["regularization"]["lambda"],
                    fisher_sample_size=self.config["regularization"]["fisher_sample_size"]
                )
        
        # 初始化评估框架
        if self.config["evaluation"]["enabled"]:
            self.evaluation_framework = ForgettingEvaluationFramework(
                model=self.base_model,
                tokenizer=self.tokenizer
            )
        
        print("持续学习框架组件初始化完成")
    
    def register_task(self, task_id, dataset, evaluation_dataset=None):
        """
        注册新任务
        
        参数：
        - task_id: 任务ID
        - dataset: 训练数据集
        - evaluation_dataset: 评估数据集
        """
        print(f"注册新任务: {task_id}")
        
        # 更新重放缓冲区
        if self.replay_buffer and task_id not in self.task_history:
            # 将当前数据添加到重放缓冲区
            self._update_replay_buffer(dataset)
        
        # 记录任务历史
        task_info = {
            "task_id": task_id,
            "dataset_size": len(dataset),
            "evaluation_dataset": evaluation_dataset
        }
        self.task_history.append(task_info)
        
        # 如果是第一个任务，计算Fisher信息矩阵（EWC需要）
        if len(self.task_history) == 1 and self.regularizer:
            self.regularizer.compute_fisher(dataset, self.tokenizer)
        
        return task_info
    
    def _update_replay_buffer(self, dataset):
        """
        更新重放缓冲区
        """
        from datasets import Dataset, concatenate_datasets
        import random
        
        # 确保不超过缓冲区大小限制
        buffer_size = self.config["replay"]["buffer_size"]
        sample_size = min(len(dataset), buffer_size // (len(self.task_history) + 1))
        
        # 从数据集中采样
        if sample_size > 0:
            # 获取当前重放数据
            current_replay_data = self.replay_buffer.replay_dataset
            
            # 从新数据集中采样
            indices = random.sample(range(len(dataset)), sample_size)
            new_samples = Dataset.from_dict(dataset[indices])
            
            # 合并并更新重放缓冲区
            if len(current_replay_data) > 0:
                updated_replay_data = concatenate_datasets([current_replay_data, new_samples])
                
                # 如果超过缓冲区大小，随机删除旧样本
                if len(updated_replay_data) > buffer_size:
                    keep_indices = random.sample(range(len(updated_replay_data)), buffer_size)
                    updated_replay_data = Dataset.from_dict(updated_replay_data[keep_indices])
            else:
                updated_replay_data = new_samples
            
            # 更新重放缓冲区
            self.replay_buffer.replay_dataset = updated_replay_data
            print(f"重放缓冲区已更新，当前大小: {len(self.replay_buffer.replay_dataset)}")
    
    def train_task(self, task_id, dataset, training_args, **kwargs):
        """
        训练特定任务
        
        参数：
        - task_id: 任务ID
        - dataset: 训练数据集
        - training_args: 训练参数
        
        返回：
        - 训练器实例
        """
        print(f"开始训练任务: {task_id}")
        
        # 准备训练数据集
        train_dataset = dataset
        if self.replay_buffer:
            # 设置主要数据集并创建混合数据集
            self.replay_buffer.primary_dataset = dataset
            train_dataset = self.replay_buffer
        
        # 应用参数隔离策略
        if self.config["parameter_isolation"]["enabled"]:
            strategy = self.config["parameter_isolation"]["strategy"]
            if strategy == "freeze":
                # 应用分层冻结
                self.base_model = freeze_model_layers(
                    self.base_model, 
                    freeze_layers_ratio=self.config["parameter_isolation"]["freeze_ratio"]
                )[0]
            elif strategy == "adapter":
                # 应用适配器
                adapter_config = self.config["parameter_isolation"]["adapter_config"]
                if adapter_config["type"] == "lora":
                    self.base_model = apply_lora_adapter(
                        self.base_model,
                        lora_r=adapter_config["r"],
                        lora_alpha=adapter_config["alpha"]
                    )[0]
        
        # 创建训练器
        if self.regularizer and hasattr(self.regularizer, 'compute_loss'):
            # 使用自定义训练器进行正则化训练
            trainer = self.regularizer.create_trainer(
                model=self.base_model,
                args=training_args,
                train_dataset=train_dataset,
                **kwargs
            )
        else:
            # 使用标准Trainer
            from transformers import Trainer
            trainer = Trainer(
                model=self.base_model,
                args=training_args,
                train_dataset=train_dataset,
                **kwargs
            )
        
        # 开始训练
        trainer.train()
        
        # 评估遗忘情况
        if self.evaluation_framework:
            metrics = self._evaluate_forgetting()
            self.knowledge_metrics[task_id] = metrics
            print(f"任务 {task_id} 训练完成，遗忘评估结果: {metrics}")
        
        return trainer
    
    def _evaluate_forgetting(self):
        """
        评估模型的遗忘情况
        """
        metrics = {}
        
        # 对每个历史任务进行评估
        for task_info in self.task_history:
            task_id = task_info["task_id"]
            eval_dataset = task_info.get("evaluation_dataset")
            
            if eval_dataset:
                # 执行评估
                results = self.evaluation_framework.evaluate(
                    model=self.base_model,
                    eval_dataset=eval_dataset,
                    task_name=task_id
                )
                metrics[task_id] = results
        
        return metrics
    
    def generate_stability_report(self, save_path=None):
        """
        生成稳定性报告
        
        参数：
        - save_path: 保存路径（可选）
        
        返回：
        - 报告内容
        """
        import json
        from datetime import datetime
        
        report = {
            "timestamp": datetime.now().isoformat(),
            "config": self.config,
            "tasks": len(self.task_history),
            "task_history": self.task_history,
            "metrics": self.knowledge_metrics
        }
        
        # 计算整体遗忘率
        all_forgetting_rates = []
        for task_id, metrics in self.knowledge_metrics.items():
            if "forgetting_rate" in metrics:
                all_forgetting_rates.append(metrics["forgetting_rate"])
        
        if all_forgetting_rates:
            report["average_forgetting_rate"] = sum(all_forgetting_rates) / len(all_forgetting_rates)
        
        # 保存报告
        if save_path:
            with open(save_path, "w", encoding="utf-8") as f:
                json.dump(report, f, ensure_ascii=False, indent=2)
            print(f"稳定性报告已保存到: {save_path}")
        
        return report

def run_continual_learning_example():
    """
    持续学习框架使用示例
    """
    # 1. 加载模型和分词器
    from transformers import AutoModelForCausalLM, AutoTokenizer
    
    model_name = "meta-llama/Llama-2-7b-hf"  # 示例模型
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForCausalLM.from_pretrained(model_name)
    
    # 2. 配置持续学习框架
    config = {
        "replay": {
            "enabled": True,
            "buffer_size": 5000,
            "replay_ratio": 0.3,
            "prioritization": True
        },
        "regularization": {
            "enabled": True,
            "method": "ewc",
            "lambda": 5e3,
            "fisher_sample_size": 500
        },
        "parameter_isolation": {
            "enabled": True,
            "strategy": "freeze",
            "freeze_ratio": 0.8
        },
        "evaluation": {
            "enabled": True,
            "metrics": ["knowledge_retention", "task_performance", "forgetting_rate"],
            "eval_interval": 50
        }
    }
    
    # 3. 初始化持续学习框架
    cl_framework = LLMContinualLearningFramework(
        base_model=model,
        tokenizer=tokenizer,
        config=config
    )
    
    # 4. 准备多个任务的数据集
    # 注意：这里是示例，实际应用中需要替换为真实数据集
    def create_synthetic_dataset(size, task_type):
        """创建合成数据集用于演示"""
        from datasets import Dataset
        import random
        
        data = []
        prompts = {
            "task1": ["解释量子计算的基本原理", "什么是纠缠现象？", "量子比特与经典比特的区别是什么？"],
            "task2": ["如何优化Python代码性能？", "解释装饰器的工作原理", "Python中的生成器有什么优势？"],
            "task3": ["什么是机器学习中的过拟合？", "解释梯度下降算法", "神经网络的基本组成部分是什么？"]
        }
        
        for _ in range(size):
            prompt = random.choice(prompts[task_type])
            data.append({"text": prompt})
        
        return Dataset.from_list(data)
    
    # 创建三个不同领域的任务数据集
    task1_train = create_synthetic_dataset(1000, "task1")  # 量子计算领域
    task1_eval = create_synthetic_dataset(200, "task1")
    
    task2_train = create_synthetic_dataset(1000, "task2")  # Python编程领域
    task2_eval = create_synthetic_dataset(200, "task2")
    
    task3_train = create_synthetic_dataset(1000, "task3")  # 机器学习领域
    task3_eval = create_synthetic_dataset(200, "task3")
    
    # 5. 注册并训练第一个任务
    cl_framework.register_task("task1", task1_train, task1_eval)
    
    training_args1 = TrainingArguments(
        output_dir="./results_task1",
        num_train_epochs=2,
        per_device_train_batch_size=4,
        warmup_steps=200,
        weight_decay=0.01,
        logging_dir="./logs_task1",
        logging_steps=50,
        evaluation_strategy="steps",
        eval_steps=100,
        save_strategy="steps",
        save_steps=500,
        learning_rate=2e-5
    )
    
    trainer1 = cl_framework.train_task(
        "task1", 
        task1_train, 
        training_args1,
        eval_dataset=task1_eval,
        tokenizer=tokenizer
    )
    
    # 6. 注册并训练第二个任务
    cl_framework.register_task("task2", task2_train, task2_eval)
    
    training_args2 = TrainingArguments(
        output_dir="./results_task2",
        num_train_epochs=2,
        per_device_train_batch_size=4,
        warmup_steps=200,
        weight_decay=0.01,
        logging_dir="./logs_task2",
        logging_steps=50,
        evaluation_strategy="steps",
        eval_steps=100,
        save_strategy="steps",
        save_steps=500,
        learning_rate=2e-5
    )
    
    trainer2 = cl_framework.train_task(
        "task2", 
        task2_train, 
        training_args2,
        eval_dataset=task2_eval,
        tokenizer=tokenizer
    )
    
    # 7. 注册并训练第三个任务
    cl_framework.register_task("task3", task3_train, task3_eval)
    
    training_args3 = TrainingArguments(
        output_dir="./results_task3",
        num_train_epochs=2,
        per_device_train_batch_size=4,
        warmup_steps=200,
        weight_decay=0.01,
        logging_dir="./logs_task3",
        logging_steps=50,
        evaluation_strategy="steps",
        eval_steps=100,
        save_strategy="steps",
        save_steps=500,
        learning_rate=2e-5
    )
    
    trainer3 = cl_framework.train_task(
        "task3", 
        task3_train, 
        training_args3,
        eval_dataset=task3_eval,
        tokenizer=tokenizer
    )
    
    # 8. 生成稳定性报告
    report = cl_framework.generate_stability_report("./continual_learning_stability_report.json")
    print(f"\n平均遗忘率: {report.get('average_forgetting_rate', 'N/A')}")
    
    return report

# 使用示例
# report = run_continual_learning_example()

class WeightedEnsembleModel:
    """
    加权平均模型集成器
    """
    def __init__(self, models, weights=None):
        """
        初始化加权集成模型
        
        参数：
        - models: 模型列表，每个元素为(model, tokenizer)元组
        - weights: 权重列表，默认为均匀权重
        """
        self.models = models
        if weights is None:
            # 默认使用均匀权重
            self.weights = [1.0 / len(models)] * len(models)
        else:
            # 确保权重归一化
            total_weight = sum(weights)
            self.weights = [w / total_weight for w in weights]
    
    def generate(self, prompt, max_length=100, temperature=0.7, **kwargs):
        """
        生成文本，集成多个模型的输出
        
        参数：
        - prompt: 输入提示
        - max_length: 最大生成长度
        - temperature: 生成温度
        
        返回：
        - 集成后的生成文本
        """
        # 获取每个模型的输出分布
        logits_list = []
        for (model, tokenizer), weight in zip(self.models, self.weights):
            # 编码输入
            inputs = tokenizer(prompt, return_tensors="pt")
            input_ids = inputs["input_ids"]
            
            # 生成序列
            with torch.no_grad():
                outputs = model.generate(
                    input_ids,
                    max_length=max_length,
                    temperature=temperature,
                    output_scores=True,
                    return_dict_in_generate=True,
                    **kwargs
                )
            
            # 获取logits并加权
            scores = outputs.scores
            weighted_scores = [score * weight for score in scores]
            logits_list.append(weighted_scores)
        
        # 集成logits
        generated_tokens = []
        for token_pos in range(len(logits_list[0])):
            # 累加所有模型在当前位置的logits
            aggregated_logits = sum(logits[token_pos] for logits in logits_list)
            
            # 根据温度调整
            if temperature > 0:
                aggregated_logits = aggregated_logits / temperature
            
            # 转换为概率分布并采样
            probs = F.softmax(aggregated_logits, dim=-1)
            next_token = torch.multinomial(probs, num_samples=1)
            generated_tokens.append(next_token)
        
        # 解码生成的序列
        generated_sequence = torch.cat(generated_tokens, dim=1)
        generated_text = self.models[0][1].decode(
            generated_sequence[0], 
            skip_special_tokens=True
        )
        
        return generated_text
    
    def set_weights(self, weights):
        """
        设置模型权重
        
        参数：
        - weights: 新的权重列表
        """
        total_weight = sum(weights)
        self.weights = [w / total_weight for w in weights]

class TaskAwareRouterEnsemble:
    """
    任务感知路由集成器
    """
    def __init__(self, models, task_embeddings, router_model=None):
        """
        初始化任务感知路由集成器
        
        参数：
        - models: 模型列表，每个元素为(model, tokenizer, task_name)元组
        - task_embeddings: 任务嵌入字典，键为任务名称，值为任务嵌入向量
        - router_model: 路由模型，用于预测任务类型
        """
        self.models = models
        self.task_embeddings = task_embeddings
        self.router_model = router_model or self._create_simple_router()
        self.task_names = [task_name for _, _, task_name in models]
    
    def _create_simple_router(self):
        """
        创建简单的任务路由模型
        基于输入提示与任务嵌入的相似度
        """
        def simple_router(prompt, task_embeddings, tokenizer):
            # 将提示编码为嵌入
            inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512)
            
            # 使用第一个模型的编码器获取嵌入
            # 注意：实际应用中可能需要更复杂的编码方式
            with torch.no_grad():
                outputs = self.models[0][0].model.embed_tokens(inputs["input_ids"])
                prompt_embedding = outputs.mean(dim=1).squeeze()
            
            # 计算与每个任务嵌入的相似度
            similarities = {}
            for task_name, task_emb in task_embeddings.items():
                sim = F.cosine_similarity(prompt_embedding.unsqueeze(0), task_emb.unsqueeze(0)).item()
                similarities[task_name] = sim
            
            return similarities
        
        return simple_router
    
    def predict_task(self, prompt):
        """
        预测输入提示的任务类型
        
        参数：
        - prompt: 输入提示
        
        返回：
        - 任务相似度字典
        """
        tokenizer = self.models[0][1]  # 使用第一个模型的分词器
        return self.router_model(prompt, self.task_embeddings, tokenizer)
    
    def generate(self, prompt, max_length=100, temperature=0.7, top_k=3, **kwargs):
        """
        根据任务类型动态选择模型生成文本
        
        参数：
        - prompt: 输入提示
        - max_length: 最大生成长度
        - temperature: 生成温度
        - top_k: 考虑的最高k个模型
        
        返回：
        - 生成文本
        """
        # 预测任务类型
        task_similarities = self.predict_task(prompt)
        
        # 排序任务相似度
        sorted_tasks = sorted(task_similarities.items(), key=lambda x: x[1], reverse=True)
        
        # 选择前k个最相关的模型
        selected_models = []
        selected_weights = []
        
        for task_name, similarity in sorted_tasks[:top_k]:
            # 找到对应的模型
            for model, tokenizer, t_name in self.models:
                if t_name == task_name:
                    selected_models.append((model, tokenizer))
                    selected_weights.append(similarity)
                    break
        
        # 归一化权重
        total_weight = sum(selected_weights)
        normalized_weights = [w / total_weight for w in selected_weights]
        
        # 使用加权集成生成文本
        if len(selected_models) == 1:
            # 单个模型直接生成
            model, tokenizer = selected_models[0]
            inputs = tokenizer(prompt, return_tensors="pt")
            with torch.no_grad():
                outputs = model.generate(
                    inputs["input_ids"],
                    max_length=max_length,
                    temperature=temperature,
                    **kwargs
                )
            return tokenizer.decode(outputs[0], skip_special_tokens=True)
        else:
            # 多个模型加权集成
            # 使用之前定义的加权集成器
            ensemble = WeightedEnsembleModel(selected_models, normalized_weights)
            return ensemble.generate(prompt, max_length, temperature, **kwargs)

def fuse_model_parameters(model_list, weights=None, save_path=None):
    """
    融合多个模型的参数
    
    参数：
    - model_list: 模型列表
    - weights: 权重列表，默认为均匀权重
    - save_path: 保存路径（可选）
    
    返回：
    - 融合后的模型
    """
    import copy
    
    # 默认使用均匀权重
    if weights is None:
        weights = [1.0 / len(model_list)] * len(model_list)
    else:
        # 确保权重归一化
        total_weight = sum(weights)
        weights = [w / total_weight for w in weights]
    
    # 创建第一个模型的深拷贝作为基础
    fused_model = copy.deepcopy(model_list[0])
    
    # 获取所有模型的命名参数
    model_parameters = []
    for model in model_list:
        model_parameters.append(dict(model.named_parameters()))
    
    # 融合参数
    print(f"开始融合 {len(model_list)} 个模型的参数...")
    
    # 遍历所有参数名
    for param_name, param in model_parameters[0].items():
        # 检查所有模型是否都有这个参数
        all_have_param = all(param_name in mp for mp in model_parameters)
        if not all_have_param:
            print(f"警告: 参数 {param_name} 在某些模型中不存在，跳过融合")
            continue
        
        # 初始化融合参数
        fused_param = torch.zeros_like(param.data)
        
        # 加权累加所有模型的参数
        for i, mp in enumerate(model_parameters):
            fused_param += weights[i] * mp[param_name].data
        
        # 设置融合后的参数
        fused_model.state_dict()[param_name].data.copy_(fused_param)
    
    print("参数融合完成")
    
    # 保存融合后的模型
    if save_path:
        fused_model.save_pretrained(save_path)
        print(f"融合后的模型已保存到: {save_path}")
    
    return fused_model

# 使用示例
def demonstrate_model_fusion():
    """
    演示模型参数融合
    """
    from transformers import AutoModelForCausalLM, AutoTokenizer
    
    # 假设我们有三个在不同任务上微调的模型
    model_paths = [
        "./results_task1",  # 量子计算微调模型
        "./results_task2",  # Python编程微调模型
        "./results_task3"   # 机器学习微调模型
    ]
    
    # 加载模型
    models = []
    for path in model_paths:
        try:
            model = AutoModelForCausalLM.from_pretrained(path)
            models.append(model)
            print(f"已加载模型: {path}")
        except Exception as e:
            print(f"加载模型 {path} 失败: {e}")
    
    if len(models) < 2:
        print("需要至少两个模型进行融合")
        return None
    
    # 定义权重（可以根据模型在不同任务上的性能调整）
    weights = [0.3, 0.4, 0.3]  # 假设Python编程模型更重要
    
    # 融合模型参数
    fused_model = fuse_model_parameters(models, weights, save_path="./fused_model")
    
    return fused_model

# 使用示例
# fused_model = demonstrate_model_fusion()

class MixtureOfExperts:
    """
    专家混合系统
    """
    def __init__(self, expert_models, gating_network=None, hidden_size=768):
        """
        初始化专家混合系统
        
        参数：
        - expert_models: 专家模型列表，每个元素为(model, tokenizer, domain)元组
        - gating_network: 门控网络，用于动态分配专家权重
        - hidden_size: 门控网络的隐藏层大小
        """
        self.expert_models = expert_models
        self.num_experts = len(expert_models)
        self.tokenizers = [tokenizer for _, tokenizer, _ in expert_models]
        self.domains = [domain for _, _, domain in expert_models]
        
        # 如果没有提供门控网络，创建一个简单的线性门控网络
        if gating_network is None:
            self.gating_network = self._create_simple_gating_network(hidden_size)
        else:
            self.gating_network = gating_network
    
    def _create_simple_gating_network(self, hidden_size):
        """
        创建简单的线性门控网络
        """
        from torch import nn
        
        # 获取第一个模型的嵌入维度
        sample_model, _, _ = self.expert_models[0]
        embed_dim = sample_model.config.hidden_size
        
        # 简单的两层前馈网络
        class SimpleGatingNetwork(nn.Module):
            def __init__(self, input_dim, hidden_dim, output_dim):
                super().__init__()
                self.fc1 = nn.Linear(input_dim, hidden_dim)
                self.relu = nn.ReLU()
                self.fc2 = nn.Linear(hidden_dim, output_dim)
                self.softmax = nn.Softmax(dim=-1)
            
            def forward(self, x):
                # x是输入嵌入
                x = self.fc1(x)
                x = self.relu(x)
                x = self.fc2(x)
                return self.softmax(x)  # 返回专家权重
        
        return SimpleGatingNetwork(embed_dim, hidden_size, self.num_experts)
    
    def get_input_embedding(self, prompt, tokenizer_idx=0):
        """
        获取输入提示的嵌入表示
        
        参数：
        - prompt: 输入提示
        - tokenizer_idx: 使用的分词器索引
        
        返回：
        - 输入嵌入
        """
        tokenizer = self.tokenizers[tokenizer_idx]
        model, _, _ = self.expert_models[tokenizer_idx]
        
        # 编码输入
        inputs = tokenizer(prompt, return_tensors="pt")
        
        # 获取输入嵌入
        with torch.no_grad():
            # 使用模型的嵌入层
            embeddings = model.model.embed_tokens(inputs["input_ids"])
            # 取平均作为整个输入的嵌入表示
            avg_embedding = embeddings.mean(dim=1)
        
        return avg_embedding
    
    def get_expert_weights(self, prompt):
        """
        获取专家权重
        
        参数：
        - prompt: 输入提示
        
        返回：
        - 专家权重张量
        """
        # 获取输入嵌入
        input_embedding = self.get_input_embedding(prompt)
        
        # 使用门控网络计算专家权重
        with torch.no_grad():
            weights = self.gating_network(input_embedding)
        
        return weights
    
    def generate(self, prompt, max_length=100, temperature=0.7, top_k_experts=2, **kwargs):
        """
        生成文本，使用专家混合系统
        
        参数：
        - prompt: 输入提示
        - max_length: 最大生成长度
        - temperature: 生成温度
        - top_k_experts: 使用权重最高的k个专家
        
        返回：
        - 生成文本
        """
        # 获取专家权重
        weights = self.get_expert_weights(prompt)
        
        # 选择前k个权重最高的专家
        top_k_weights, top_k_indices = torch.topk(weights, k=min(top_k_experts, self.num_experts), dim=-1)
        
        # 归一化选中专家的权重
        normalized_weights = F.softmax(top_k_weights, dim=-1).tolist()[0]
        
        # 准备选中的专家模型
        selected_experts = [(self.expert_models[idx][0], self.expert_models[idx][1]) for idx in top_k_indices.tolist()[0]]
        
        # 使用加权集成生成文本
        ensemble = WeightedEnsembleModel(selected_experts, normalized_weights)
        return ensemble.generate(prompt, max_length, temperature, **kwargs)
    
    def train_gating_network(self, training_data, epochs=5, learning_rate=1e-4, batch_size=8):
        """
        训练门控网络
        
        参数：
        - training_data: 训练数据，格式为[(prompt, domain_label)]
        - epochs: 训练轮数
        - learning_rate: 学习率
        - batch_size: 批次大小
        """
        import torch.optim as optim
        from torch.utils.data import DataLoader, TensorDataset
        from sklearn.preprocessing import LabelEncoder
        
        # 准备数据
        embeddings = []
        labels = []
        
        # 创建标签编码器
        label_encoder = LabelEncoder()
        label_encoder.fit(self.domains)
        
        # 处理训练数据
        for prompt, domain in training_data:
            # 获取输入嵌入
            embedding = self.get_input_embedding(prompt)
            embeddings.append(embedding)
            
            # 编码标签
            label_idx = label_encoder.transform([domain])[0]
            labels.append(label_idx)
        
        # 创建数据集和数据加载器
        embeddings_tensor = torch.cat(embeddings)
        labels_tensor = torch.tensor(labels)
        dataset = TensorDataset(embeddings_tensor, labels_tensor)
        dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)
        
        # 定义损失函数和优化器
        criterion = torch.nn.CrossEntropyLoss()
        optimizer = optim.Adam(self.gating_network.parameters(), lr=learning_rate)
        
        # 训练门控网络
        self.gating_network.train()
        for epoch in range(epochs):
            running_loss = 0.0
            for i, (inputs, labels) in enumerate(dataloader):
                # 梯度清零
                optimizer.zero_grad()
                
                # 前向传播
                outputs = self.gating_network(inputs)
                loss = criterion(outputs, labels)
                
                # 反向传播和优化
                loss.backward()
                optimizer.step()
                
                # 统计损失
                running_loss += loss.item()
                
                if (i + 1) % 10 == 0:
                    print(f'Epoch [{epoch+1}/{epochs}], Step [{i+1}/{len(dataloader)}], Loss: {running_loss/10:.4f}')
                    running_loss = 0.0
        
        print('门控网络训练完成')
        return self.gating_network

class ComprehensiveForgettingEvaluator:
    """
    综合灾难性遗忘评估器
    """
    def __init__(self, baseline_model, baseline_tokenizer):
        """
        初始化评估器
        
        参数：
        - baseline_model: 基础模型（微调前）
        - baseline_tokenizer: 基础分词器
        """
        self.baseline_model = baseline_model
        self.baseline_tokenizer = baseline_tokenizer
        self.evaluation_results = {}
    
    def evaluate_model(self, model, model_name, evaluation_datasets, metrics=None):
        """
        评估模型性能
        
        参数：
        - model: 要评估的模型
        - model_name: 模型名称（用于结果存储）
        - evaluation_datasets: 评估数据集字典，格式为 {"dataset_name": dataset}
        - metrics: 要计算的指标列表，默认为 ["perplexity", "accuracy", "f1", "bleu"]
        
        返回：
        - 评估结果字典
        """
        if metrics is None:
            metrics = ["perplexity", "accuracy", "f1", "bleu"]
        
        results = {}
        
        # 对每个数据集进行评估
        for dataset_name, dataset in evaluation_datasets.items():
            print(f"评估模型 {model_name} 在数据集 {dataset_name} 上的性能...")
            dataset_results = {}
            
            # 计算困惑度
            if "perplexity" in metrics:
                perplexity = self._calculate_perplexity(model, dataset)
                dataset_results["perplexity"] = perplexity
                print(f"  困惑度: {perplexity:.4f}")
            
            # 计算准确率（如果适用）
            if "accuracy" in metrics and hasattr(dataset, "labels"):
                accuracy = self._calculate_accuracy(model, dataset)
                dataset_results["accuracy"] = accuracy
                print(f"  准确率: {accuracy:.4f}")
            
            # 计算F1分数（如果适用）
            if "f1" in metrics and hasattr(dataset, "labels"):
                f1 = self._calculate_f1(model, dataset)
                dataset_results["f1"] = f1
                print(f"  F1分数: {f1:.4f}")
            
            # 计算BLEU分数（如果适用）
            if "bleu" in metrics and hasattr(dataset, "target_texts"):
                bleu = self._calculate_bleu(model, dataset)
                dataset_results["bleu"] = bleu
                print(f"  BLEU分数: {bleu:.4f}")
            
            # 计算知识保留指标
            if "knowledge_retention" in metrics:
                knowledge_ret = self._calculate_knowledge_retention(model, dataset)
                dataset_results["knowledge_retention"] = knowledge_ret
                print(f"  知识保留率: {knowledge_ret:.4f}")
            
            results[dataset_name] = dataset_results
        
        # 存储评估结果
        self.evaluation_results[model_name] = results
        return results
    
    def _calculate_perplexity(self, model, dataset, max_samples=1000):
        """
        计算困惑度
        """
        import torch
        
        model.eval()
        total_loss = 0
        total_tokens = 0
        
        # 限制样本数量以提高效率
        samples = dataset[:max_samples] if len(dataset) > max_samples else dataset
        
        with torch.no_grad():
            for sample in samples:
                if hasattr(sample, "text"):
                    text = sample.text
                elif "text" in sample:
                    text = sample["text"]
                else:
                    continue
                
                # 编码文本
                inputs = self.baseline_tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
                
                # 计算损失
                outputs = model(**inputs, labels=inputs["input_ids"])
                loss = outputs.loss
                
                # 累加损失和token数量
                total_loss += loss.item() * inputs["input_ids"].size(1)
                total_tokens += inputs["input_ids"].size(1)
        
        # 计算平均困惑度
        if total_tokens > 0:
            avg_loss = total_loss / total_tokens
            perplexity = torch.exp(torch.tensor(avg_loss)).item()
            return perplexity
        else:
            return float('inf')
    
    def _calculate_accuracy(self, model, dataset, max_samples=500):
        """
        计算准确率
        """
        import torch
        
        model.eval()
        correct = 0
        total = 0
        
        # 限制样本数量
        samples = dataset[:max_samples] if len(dataset) > max_samples else dataset
        
        with torch.no_grad():
            for sample in samples:
                if hasattr(sample, "text") and hasattr(sample, "labels"):
                    text = sample.text
                    label = sample.labels
                elif "text" in sample and "labels" in sample:
                    text = sample["text"]
                    label = sample["labels"]
                else:
                    continue
                
                # 编码文本
                inputs = self.baseline_tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
                
                # 生成预测
                outputs = model(**inputs)
                predictions = torch.argmax(outputs.logits, dim=-1)
                
                # 简单的分类逻辑（根据需要调整）
                # 这里假设输出的第一个token代表分类结果
                # 实际应用中需要根据具体任务调整
                if predictions[0, 0].item() == label:
                    correct += 1
                total += 1
        
        return correct / total if total > 0 else 0
    
    def _calculate_f1(self, model, dataset, max_samples=500):
        """
        计算F1分数
        """
        from sklearn.metrics import f1_score
        import torch
        
        model.eval()
        y_true = []
        y_pred = []
        
        # 限制样本数量
        samples = dataset[:max_samples] if len(dataset) > max_samples else dataset
        
        with torch.no_grad():
            for sample in samples:
                if hasattr(sample, "text") and hasattr(sample, "labels"):
                    text = sample.text
                    label = sample.labels
                elif "text" in sample and "labels" in sample:
                    text = sample["text"]
                    label = sample["labels"]
                else:
                    continue
                
                # 编码文本
                inputs = self.baseline_tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
                
                # 生成预测
                outputs = model(**inputs)
                predictions = torch.argmax(outputs.logits, dim=-1)
                
                # 存储真实标签和预测结果
                y_true.append(label)
                y_pred.append(predictions[0, 0].item())
        
        # 计算F1分数
        if len(y_true) > 0:
            return f1_score(y_true, y_pred, average="macro")
        else:
            return 0
    
    def _calculate_bleu(self, model, dataset, max_samples=100):
        """
        计算BLEU分数
        """
        from nltk.translate.bleu_score import sentence_bleu
        
        model.eval()
        total_bleu = 0
        count = 0
        
        # 限制样本数量
        samples = dataset[:max_samples] if len(dataset) > max_samples else dataset
        
        with torch.no_grad():
            for sample in samples:
                if hasattr(sample, "text") and hasattr(sample, "target_texts"):
                    text = sample.text
                    target_texts = sample.target_texts
                elif "text" in sample and "target_texts" in sample:
                    text = sample["text"]
                    target_texts = sample["target_texts"]
                else:
                    continue
                
                # 生成文本
                inputs = self.baseline_tokenizer(text, return_tensors="pt")
                outputs = model.generate(inputs["input_ids"], max_length=100)
                generated_text = self.baseline_tokenizer.decode(outputs[0], skip_special_tokens=True)
                
                # 计算BLEU分数
                # 注意：实际应用中可能需要更复杂的预处理
                reference = [text.split() for text in target_texts]
                candidate = generated_text.split()
                
                bleu_score = sentence_bleu(reference, candidate)
                total_bleu += bleu_score
                count += 1
        
        return total_bleu / count if count > 0 else 0
    
    def _calculate_knowledge_retention(self, model, dataset, max_samples=200):
        """
        计算知识保留率
        比较微调前后模型在相同任务上的表现
        """
        # 计算微调后模型的困惑度
        finetuned_perplexity = self._calculate_perplexity(model, dataset, max_samples)
        
        # 计算基础模型的困惑度
        baseline_perplexity = self._calculate_perplexity(self.baseline_model, dataset, max_samples)
        
        # 计算知识保留率
        # 这里使用困惑度的相对变化来衡量知识保留
        # 如果微调后困惑度更低，说明保留了知识且可能有提升
        if baseline_perplexity > 0:
            perplexity_ratio = finetuned_perplexity / baseline_perplexity
            # 将困惑度比率映射到0-1的知识保留率
            # 当困惑度相同时，保留率为1
            # 当困惑度增加时，保留率降低
            knowledge_retention = max(0, 2 - perplexity_ratio) if perplexity_ratio <= 2 else 0
            return knowledge_retention
        else:
            return 0
    
    def calculate_forgetting_rate(self, pretrained_results, finetuned_results):
        """
        计算遗忘率
        
        参数：
        - pretrained_results: 预训练模型的评估结果
        - finetuned_results: 微调后模型的评估结果
        
        返回：
        - 遗忘率字典
        """
        forgetting_rates = {}
        
        # 遍历所有数据集
        for dataset_name in pretrained_results:
            if dataset_name in finetuned_results:
                dataset_forgetting = {}
                
                # 遍历所有指标
                for metric_name in pretrained_results[dataset_name]:
                    if metric_name in finetuned_results[dataset_name]:
                        pretrained_score = pretrained_results[dataset_name][metric_name]
                        finetuned_score = finetuned_results[dataset_name][metric_name]
                        
                        # 对于困惑度，分数越低越好
                        # 对于其他指标，分数越高越好
                        if metric_name == "perplexity":
                            # 困惑度增加表示性能下降
                            if pretrained_score > 0:
                                relative_change = (finetuned_score - pretrained_score) / pretrained_score
                                # 转换为0-1的遗忘率，困惑度增加越多，遗忘率越高
                                dataset_forgetting[metric_name] = max(0, relative_change)
                            else:
                                dataset_forgetting[metric_name] = 0
                        else:
                            # 对于其他指标，性能下降表示遗忘
                            if pretrained_score > 0:
                                relative_change = (pretrained_score - finetuned_score) / pretrained_score
                                # 转换为0-1的遗忘率
                                dataset_forgetting[metric_name] = max(0, relative_change)
                            else:
                                dataset_forgetting[metric_name] = 0
                
                # 计算平均遗忘率
                if dataset_forgetting:
                    avg_forgetting = sum(dataset_forgetting.values()) / len(dataset_forgetting)
                    dataset_forgetting["average"] = avg_forgetting
                
                forgetting_rates[dataset_name] = dataset_forgetting
        
        # 计算总体平均遗忘率
        all_forgetting_rates = []
        for dataset in forgetting_rates:
            if "average" in forgetting_rates[dataset]:
                all_forgetting_rates.append(forgetting_rates[dataset]["average"])
        
        if all_forgetting_rates:
            overall_average = sum(all_forgetting_rates) / len(all_forgetting_rates)
            forgetting_rates["overall_average"] = overall_average
        
        return forgetting_rates
    
    def generate_evaluation_report(self, output_file=None):
        """
        生成评估报告
        
        参数：
        - output_file: 输出文件路径（可选）
        
        返回：
        - 评估报告内容
        """
        import json
        
        # 生成报告内容
        report = {
            "models_evaluated": list(self.evaluation_results.keys()),
            "evaluation_details": self.evaluation_results
        }
        
        # 如果有预训练和微调后的结果，计算遗忘率
        if "pretrained" in self.evaluation_results and "finetuned" in self.evaluation_results:
            forgetting_rates = self.calculate_forgetting_rate(
                self.evaluation_results["pretrained"],
                self.evaluation_results["finetuned"]
            )
            report["forgetting_rates"] = forgetting_rates
        
        # 转换为JSON格式
        report_json = json.dumps(report, indent=2, ensure_ascii=False)
        
        # 如果指定了输出文件，保存报告
        if output_file:
            with open(output_file, "w", encoding="utf-8") as f:
                f.write(report_json)
            print(f"评估报告已保存到: {output_file}")
        
        return report_json
    
    def visualize_results(self, metrics=None, output_dir="./visualizations"):
        """
        可视化评估结果
        
        参数：
        - metrics: 要可视化的指标列表
        - output_dir: 输出目录
        """
        import matplotlib.pyplot as plt
        import os
        
        # 创建输出目录
        os.makedirs(output_dir, exist_ok=True)
        
        # 获取所有模型名称
        model_names = list(self.evaluation_results.keys())
        
        # 如果没有指定指标，使用所有可用指标
        if metrics is None:
            metrics = set()
            for model in self.evaluation_results:
                for dataset in self.evaluation_results[model]:
                    metrics.update(self.evaluation_results[model][dataset].keys())
            metrics = list(metrics)
        
        # 为每个指标创建可视化
        for metric in metrics:
            # 收集数据
            data = {}
            datasets = set()
            
            for model in model_names:
                data[model] = {}
                for dataset in self.evaluation_results[model]:
                    if metric in self.evaluation_results[model][dataset]:
                        data[model][dataset] = self.evaluation_results[model][dataset][metric]
                        datasets.add(dataset)
            
            # 为每个数据集创建单独的图表
            for dataset in datasets:
                # 准备数据
                x = model_names
                y = [data[model].get(dataset, 0) for model in model_names]
                
                # 创建图表
                plt.figure(figsize=(10, 6))
                plt.bar(x, y)
                plt.xlabel("模型")
                plt.ylabel(metric)
                plt.title(f"{metric} 在 {dataset} 上的表现")
                plt.xticks(rotation=45)
                plt.tight_layout()
                
                # 保存图表
                output_file = os.path.join(output_dir, f"{metric}_{dataset}.png")
                plt.savefig(output_file)
                plt.close()
                print(f"图表已保存到: {output_file}")

class DomainKnowledgeEvaluator:
    """
    领域知识评估器
    专门用于评估LLM在特定领域知识上的保留情况
    """
    def __init__(self, model, tokenizer):
        """
        初始化领域知识评估器
        
        参数：
        - model: 要评估的模型
        - tokenizer: 分词器
        """
        self.model = model
        self.tokenizer = tokenizer
        self.evaluation_sets = {}
    
    def add_evaluation_set(self, domain_name, questions, ground_truths):
        """
        添加领域评估集
        
        参数：
        - domain_name: 领域名称
        - questions: 问题列表
        - ground_truths: 参考答案列表
        """
        if len(questions) != len(ground_truths):
            raise ValueError("问题数量必须与参考答案数量相同")
        
        self.evaluation_sets[domain_name] = {
            "questions": questions,
            "ground_truths": ground_truths
        }
        
        print(f"已添加 {domain_name} 领域的评估集，包含 {len(questions)} 个问题")
    
    def load_evaluation_set_from_json(self, file_path, domain_name=None):
        """
        从JSON文件加载评估集
        
        参数：
        - file_path: JSON文件路径
        - domain_name: 领域名称（如果文件中没有指定）
        """
        import json
        
        try:
            with open(file_path, "r", encoding="utf-8") as f:
                data = json.load(f)
            
            # 如果没有指定领域名称，尝试从文件中获取
            if domain_name is None:
                if "domain_name" in data:
                    domain_name = data["domain_name"]
                else:
                    raise ValueError("文件中没有指定领域名称，且未提供domain_name参数")
            
            # 验证数据格式
            if "questions" not in data or "ground_truths" not in data:
                raise ValueError("JSON文件必须包含'questions'和'ground_truths'字段")
            
            self.add_evaluation_set(
                domain_name, 
                data["questions"], 
                data["ground_truths"]
            )
            
            return True
        except Exception as e:
            print(f"加载评估集失败: {e}")
            return False
    
    def generate_domain_knowledge_qa(self, domain_name, num_questions=50, output_file=None):
        """
        生成领域知识问答对
        注意：这需要一个强大的模型来生成高质量的问答对
        
        参数：
        - domain_name: 领域名称
        - num_questions: 生成的问题数量
        - output_file: 输出文件路径
        
        返回：
        - 生成的问答对字典
        """
        # 定义生成领域问题的提示
        question_gen_prompt = f"""
        请为{domain_name}领域生成{num_questions}个专业知识问题。
        问题应该涵盖该领域的核心概念、重要原理和实用技术。
        问题难度应该从基础到高级都有覆盖。
        每个问题应该具体、明确，且能够检验对该领域的真实理解。
        
        请以以下格式输出问题列表：
        [
            "问题1",
            "问题2",
            ...
        ]
        """
        
        # 生成问题
        print(f"正在生成{domain_name}领域的问题...")
        inputs = self.tokenizer(question_gen_prompt, return_tensors="pt")
        outputs = self.model.generate(
            inputs["input_ids"],
            max_length=2000,
            temperature=0.7,
            num_return_sequences=1
        )
        questions_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        
        # 提取问题列表（这里简化处理，实际应用中可能需要更复杂的解析）
        # 尝试从生成文本中提取JSON格式的问题列表
        import re
        import json
        
        # 尝试提取JSON格式的内容
        json_match = re.search(r'\[.*?\]', questions_text, re.DOTALL)
        if json_match:
            try:
                questions = json.loads(json_match.group())
                print(f"成功提取了{len(questions)}个问题")
            except json.JSONDecodeError:
                print("无法解析JSON格式，将使用备用方法提取问题")
                # 备用方法：按行分割
                lines = questions_text.strip().split('\n')
                questions = [line.strip('"\' .1234567890-') for line in lines if line.strip()]
                questions = questions[:num_questions]
        else:
            # 备用方法：按行分割
            lines = questions_text.strip().split('\n')
            questions = [line.strip('"\' .1234567890-') for line in lines if line.strip()]
            questions = questions[:num_questions]
        
        # 生成参考答案
        print(f"正在为问题生成参考答案...")
        ground_truths = []
        
        for i, question in enumerate(questions):
            print(f"生成问题{i+1}/{len(questions)}的参考答案...")
            
            # 定义生成答案的提示
            answer_gen_prompt = f"""
            请为以下{domain_name}领域的问题提供准确、全面的专业答案：
            {question}
            
            答案应该：
            1. 准确无误，符合该领域的最新知识
            2. 全面覆盖问题的各个方面
            3. 使用专业但清晰的语言
            4. 包含必要的细节和解释
            """
            
            # 生成答案
            inputs = self.tokenizer(answer_gen_prompt, return_tensors="pt")
            outputs = self.model.generate(
                inputs["input_ids"],
                max_length=1000,
                temperature=0.3,  # 降低温度以获得更准确的答案
                num_return_sequences=1
            )
            answer = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
            ground_truths.append(answer)
        
        # 构建评估集
        evaluation_set = {
            "domain_name": domain_name,
            "questions": questions,
            "ground_truths": ground_truths
        }
        
        # 保存到文件
        if output_file:
            with open(output_file, "w", encoding="utf-8") as f:
                json.dump(evaluation_set, f, ensure_ascii=False, indent=2)
            print(f"评估集已保存到: {output_file}")
        
        # 添加到评估器
        self.add_evaluation_set(domain_name, questions, ground_truths)
        
        return evaluation_set
    
    def evaluate_domain_knowledge(self, domain_name, evaluate_model=None, max_questions=None):
        """
        评估模型在特定领域的知识掌握情况
        
        参数：
        - domain_name: 领域名称
        - evaluate_model: 要评估的模型（默认使用初始化时的模型）
        - max_questions: 最大评估问题数量
        
        返回：
        - 评估结果字典
        """
        # 检查领域是否存在
        if domain_name not in self.evaluation_sets:
            raise ValueError(f"领域 {domain_name} 不存在于评估集中")
        
        # 使用指定模型或默认模型
        model = evaluate_model if evaluate_model else self.model
        
        # 获取评估集
        evaluation_set = self.evaluation_sets[domain_name]
        questions = evaluation_set["questions"]
        ground_truths = evaluation_set["ground_truths"]
        
        # 限制评估问题数量
        if max_questions and max_questions < len(questions):
            questions = questions[:max_questions]
            ground_truths = ground_truths[:max_questions]
        
        print(f"开始评估模型在 {domain_name} 领域的知识，共 {len(questions)} 个问题")
        
        # 存储每个问题的评估结果
        question_results = []
        total_score = 0
        
        # 对每个问题进行评估
        for i, (question, ground_truth) in enumerate(zip(questions, ground_truths)):
            print(f"评估问题 {i+1}/{len(questions)}: {question}")
            
            # 生成答案
            inputs = self.tokenizer(question, return_tensors="pt")
            outputs = model.generate(
                inputs["input_ids"],
                max_length=500,
                temperature=0.3,
                num_return_sequences=1
            )
            generated_answer = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
            
            # 评分生成的答案
            score, feedback = self._score_answer(generated_answer, ground_truth)
            
            # 存储结果
            question_results.append({
                "question": question,
                "ground_truth": ground_truth,
                "generated_answer": generated_answer,
                "score": score,
                "feedback": feedback
            })
            
            total_score += score
        
        # 计算平均分数
        average_score = total_score / len(questions) if questions else 0
        
        # 构建总体评估结果
        result = {
            "domain_name": domain_name,
            "total_questions": len(questions),
            "average_score": average_score,
            "question_results": question_results
        }
        
        print(f"领域 {domain_name} 知识评估完成，平均分数: {average_score:.2f}/1.0")
        
        return result
    
    def _score_answer(self, generated_answer, ground_truth):
        """
        对生成的答案进行评分
        
        参数：
        - generated_answer: 模型生成的答案
        - ground_truth: 参考答案
        
        返回：
        - (分数, 反馈)
        """
        # 这里使用简化的评分方法
        # 实际应用中可能需要更复杂的评分逻辑，如使用BLEU、ROUGE、BERTScore等
        # 或者使用另一个模型来评分
        
        # 检查关键知识点的覆盖情况
        # 这里简化为检查参考答案中的关键词是否在生成答案中出现
        import re
        from nltk.tokenize import word_tokenize
        from nltk.corpus import stopwords
        import string
        
        try:
            # 下载必要的NLTK资源（如果尚未下载）
            import nltk
            nltk.download('punkt', quiet=True)
            nltk.download('stopwords', quiet=True)
        except:
            pass
        
        # 文本预处理函数
        def preprocess_text(text):
            # 转为小写
            text = text.lower()
            # 移除标点符号
            text = text.translate(str.maketrans('', '', string.punctuation))
            # 分词
            tokens = word_tokenize(text)
            # 移除停用词
            stop_words = set(stopwords.words('english'))
            tokens = [token for token in tokens if token not in stop_words and len(token) > 1]
            return set(tokens)
        
        # 计算关键词覆盖率
        try:
            # 预处理参考答案和生成答案
            ground_truth_tokens = preprocess_text(ground_truth)
            generated_tokens = preprocess_text(generated_answer)
            
            # 如果参考答案的关键词集合为空，返回默认分数
            if not ground_truth_tokens:
                return 0.5, "无法从参考答案中提取有效关键词"
            
            # 计算关键词覆盖率
            common_tokens = ground_truth_tokens.intersection(generated_tokens)
            coverage = len(common_tokens) / len(ground_truth_tokens)
            
            # 基于覆盖率确定分数
            if coverage >= 0.8:
                score = 1.0
                feedback = "优秀：覆盖了几乎所有关键知识点"
            elif coverage >= 0.6:
                score = 0.8
                feedback = "良好：覆盖了大部分关键知识点"
            elif coverage >= 0.4:
                score = 0.6
                feedback = "一般：覆盖了部分关键知识点"
            elif coverage >= 0.2:
                score = 0.4
                feedback = "较差：只覆盖了少数关键知识点"
            else:
                score = 0.2
                feedback = "差：几乎没有覆盖关键知识点"
        except Exception as e:
            # 如果评分过程出错，返回默认分数
            print(f"评分过程出错: {e}")
            return 0.5, "评分过程出错，使用默认分数"
        
        return score, feedback
    
    def evaluate_all_domains(self, evaluate_model=None, max_questions_per_domain=None):
        """
        评估模型在所有领域的知识掌握情况
        
        参数：
        - evaluate_model: 要评估的模型
        - max_questions_per_domain: 每个领域最大评估问题数量
        
        返回：
        - 所有领域的评估结果字典
        """
        all_results = {}
        
        # 对每个领域进行评估
        for domain_name in self.evaluation_sets:
            try:
                result = self.evaluate_domain_knowledge(
                    domain_name,
                    evaluate_model,
                    max_questions_per_domain
                )
                all_results[domain_name] = result
            except Exception as e:
                print(f"评估领域 {domain_name} 时出错: {e}")
                all_results[domain_name] = {"error": str(e)}
        
        # 计算总体平均分数
        domain_scores = []
        for domain_name, result in all_results.items():
            if "average_score" in result:
                domain_scores.append(result["average_score"])
        
        if domain_scores:
            overall_average_score = sum(domain_scores) / len(domain_scores)
            all_results["overall_average_score"] = overall_average_score
            print(f"所有领域的总体平均分数: {overall_average_score:.2f}/1.0")
        
        return all_results
    
    def generate_knowledge_retention_report(self, pretrained_results, finetuned_results, output_file=None):
        """
        生成知识保留报告
        
        参数：
        - pretrained_results: 预训练模型的评估结果
        - finetuned_results: 微调后模型的评估结果
        - output_file: 输出文件路径
        
        返回：
        - 报告内容
        """
        import json
        
        # 计算每个领域的知识保留率
        knowledge_retention = {}
        
        # 遍历所有领域
        for domain_name in pretrained_results:
            if domain_name in finetuned_results and domain_name != "overall_average_score":
                # 获取预训练和微调后的分数
                pretrained_score = pretrained_results[domain_name].get("average_score", 0)
                finetuned_score = finetuned_results[domain_name].get("average_score", 0)
                
                # 计算知识保留率
                if pretrained_score > 0:
                    retention_rate = finetuned_score / pretrained_score
                    # 转换为百分比
                    retention_rate_percent = min(100, retention_rate * 100)
                else:
                    retention_rate_percent = 0
                
                # 计算分数变化
                score_change = finetuned_score - pretrained_score
                
                # 判断知识保留情况
                if retention_rate_percent >= 90:
                    status = "优秀：知识保留良好"
                elif retention_rate_percent >= 70:
                    status = "良好：知识大部分保留"
                elif retention_rate_percent >= 50:
                    status = "一般：知识部分保留"
                else:
                    status = "较差：知识严重遗忘"
                
                knowledge_retention[domain_name] = {
                    "pretrained_score": pretrained_score,
                    "finetuned_score": finetuned_score,
                    "score_change": score_change,
                    "retention_rate_percent": retention_rate_percent,
                    "status": status
                }
        
        # 计算总体知识保留率
        if "overall_average_score" in pretrained_results and "overall_average_score" in finetuned_results:
            overall_pretrained = pretrained_results["overall_average_score"]
            overall_finetuned = finetuned_results["overall_average_score"]
            
            if overall_pretrained > 0:
                overall_retention = min(100, (overall_finetuned / overall_pretrained) * 100)
            else:
                overall_retention = 0
            
            knowledge_retention["overall"] = {
                "pretrained_score": overall_pretrained,
                "finetuned_score": overall_finetuned,
                "score_change": overall_finetuned - overall_pretrained,
                "retention_rate_percent": overall_retention,
                "status": "总体知识保留情况"
            }
        
        # 构建报告
        report = {
            "knowledge_retention": knowledge_retention,
            "generated_at": "2024-01-01"  # 实际应用中应使用当前日期
        }
        
        # 转换为JSON格式
        report_json = json.dumps(report, indent=2, ensure_ascii=False)
        
        # 保存到文件
        if output_file:
            with open(output_file, "w", encoding="utf-8") as f:
                f.write(report_json)
            print(f"知识保留报告已保存到: {output_file}")
        
        return report_json

class ForgettingMonitor:
    """
    灾难性遗忘实时监控器
    在微调过程中实时监控模型性能变化
    """
    def __init__(self, base_model, base_tokenizer, evaluation_datasets):
        """
        初始化遗忘监控器
        
        参数：
        - base_model: 基础模型（微调前）
        - base_tokenizer: 分词器
        - evaluation_datasets: 评估数据集字典
        """
        self.base_model = base_model
        self.base_tokenizer = base_tokenizer
        self.evaluation_datasets = evaluation_datasets
        self.monitoring_logs = []
        self.baseline_results = None
    
    def establish_baseline(self):
        """
        建立基线性能
        
        返回：
        - 基线评估结果
        """
        print("正在建立基线性能...")
        evaluator = ComprehensiveForgettingEvaluator(self.base_model, self.base_tokenizer)
        self.baseline_results = evaluator.evaluate_model(
            self.base_model,
            "baseline",
            self.evaluation_datasets
        )
        print("基线性能建立完成")
        return self.baseline_results
    
    def monitor(self, current_model, step, epoch, save_checkpoint=True, checkpoint_dir="./monitor_checkpoints"):
        """
        监控当前模型性能
        
        参数：
        - current_model: 当前模型
        - step: 当前步骤
        - epoch: 当前轮次
        - save_checkpoint: 是否保存检查点
        - checkpoint_dir: 检查点目录
        
        返回：
        - 监控结果
        """
        import os
        
        # 如果还没有基线性能，先建立基线
        if self.baseline_results is None:
            self.establish_baseline()
        
        print(f"监控步骤 {step} (轮次 {epoch}) 的模型性能...")
        
        # 评估当前模型
        evaluator = ComprehensiveForgettingEvaluator(self.base_model, self.base_tokenizer)
        current_results = evaluator.evaluate_model(
            current_model,
            f"step_{step}_epoch_{epoch}",
            self.evaluation_datasets
        )
        
        # 计算遗忘率
        forgetting_rates = evaluator.calculate_forgetting_rate(
            self.baseline_results,
            current_results
        )
        
        # 创建监控记录
        monitoring_record = {
            "step": step,
            "epoch": epoch,
            "evaluation_results": current_results,
            "forgetting_rates": forgetting_rates
        }
        
        # 保存监控记录
        self.monitoring_logs.append(monitoring_record)
        
        # 保存检查点
        if save_checkpoint:
            os.makedirs(checkpoint_dir, exist_ok=True)
            
            # 保存模型检查点
            checkpoint_path = os.path.join(checkpoint_dir, f"model_step_{step}_epoch_{epoch}")
            current_model.save_pretrained(checkpoint_path)
            
            # 保存监控记录
            import json
            log_file = os.path.join(checkpoint_dir, f"monitoring_log_step_{step}_epoch_{epoch}.json")
            with open(log_file, "w", encoding="utf-8") as f:
                json.dump(monitoring_record, f, indent=2, ensure_ascii=False)
            
            print(f"检查点已保存到: {checkpoint_path}")
            print(f"监控记录已保存到: {log_file}")
        
        # 打印关键指标
        if "overall_average" in forgetting_rates:
            overall_forgetting = forgetting_rates["overall_average"]
            print(f"当前步骤的总体遗忘率: {overall_forgetting:.4f}")
            
            # 如果遗忘率过高，发出警告
            if overall_forgetting > 0.3:
                print("警告: 遗忘率过高，考虑调整训练参数或使用正则化方法")
        
        return monitoring_record
    
    def create_callback(self, eval_interval=1000, save_checkpoint=True, checkpoint_dir="./monitor_checkpoints"):
        """
        创建Hugging Face Trainer回调
        
        参数：
        - eval_interval: 评估间隔（步数）
        - save_checkpoint: 是否保存检查点
        - checkpoint_dir: 检查点目录
        
        返回：
        - TrainerCallback对象
        """
        from transformers import TrainerCallback
        
        class ForgettingMonitorCallback(TrainerCallback):
            """
            用于监控灾难性遗忘的Trainer回调
            """
            def __init__(self, monitor, eval_interval, save_checkpoint, checkpoint_dir):
                self.monitor = monitor
                self.eval_interval = eval_interval
                self.save_checkpoint = save_checkpoint
                self.checkpoint_dir = checkpoint_dir
            
            def on_step_end(self, args, state, control, **kwargs):
                # 每eval_interval步进行一次评估
                if state.global_step % self.eval_interval == 0 and state.global_step > 0:
                    # 获取当前模型
                    model = kwargs.get("model")
                    if model:
                        # 监控模型性能
                        self.monitor.monitor(
                            model,
                            state.global_step,
                            state.epoch,
                            self.save_checkpoint,
                            self.checkpoint_dir
                        )
                
                return control
        
        return ForgettingMonitorCallback(self, eval_interval, save_checkpoint, checkpoint_dir)
    
    def generate_monitoring_report(self, output_file=None):
        """
        生成监控报告
        
        参数：
        - output_file: 输出文件路径
        
        返回：
        - 报告内容
        """
        import json
        
        # 构建报告
        report = {
            "baseline_results": self.baseline_results,
            "monitoring_logs": self.monitoring_logs,
            "total_monitoring_steps": len(self.monitoring_logs)
        }
        
        # 分析遗忘趋势
        if len(self.monitoring_logs) > 0:
            forgetting_trends = []
            for log in self.monitoring_logs:
                step = log["step"]
                epoch = log["epoch"]
                overall_forgetting = log["forgetting_rates"].get("overall_average", 0)
                
                forgetting_trends.append({
                    "step": step,
                    "epoch": epoch,
                    "overall_forgetting": overall_forgetting
                })
            
            report["forgetting_trends"] = forgetting_trends
        
        # 转换为JSON格式
        report_json = json.dumps(report, indent=2, ensure_ascii=False)
        
        # 保存到文件
        if output_file:
            with open(output_file, "w", encoding="utf-8") as f:
                f.write(report_json)
            print(f"监控报告已保存到: {output_file}")
        
        return report_json
    
    def visualize_forgetting_trends(self, output_file="./forgetting_trends.png"):
        """
        可视化遗忘趋势
        
        参数：
        - output_file: 输出文件路径
        """
        import matplotlib.pyplot as plt
        
        # 检查是否有监控数据
        if len(self.monitoring_logs) < 2:
            print("监控数据不足，无法生成趋势图")
            return False
        
        # 准备数据
        steps = []
        epochs = []
        overall_forgetting = []
        dataset_forgetting = {}
        
        # 提取所有数据集名称
        all_datasets = set()
        for log in self.monitoring_logs:
            for dataset in log["forgetting_rates"]:
                if dataset != "overall_average":
                    all_datasets.add(dataset)
        
        # 初始化数据集遗忘率列表
        for dataset in all_datasets:
            dataset_forgetting[dataset] = []
        
        # 收集数据
        for log in self.monitoring_logs:
            steps.append(log["step"])
            epochs.append(log["epoch"])
            
            # 总体遗忘率
            overall_forgetting.append(log["forgetting_rates"].get("overall_average", 0))
            
            # 各数据集遗忘率
            for dataset in all_datasets:
                avg_forgetting = log["forgetting_rates"].get(dataset, {}).get("average", 0)
                dataset_forgetting[dataset].append(avg_forgetting)
        
        # 创建图表
        plt.figure(figsize=(12, 6))
        
        # 绘制总体遗忘率
        plt.plot(steps, overall_forgetting, 'o-', label="总体遗忘率", linewidth=2)
        
        # 绘制各数据集遗忘率
        for dataset in all_datasets:
            plt.plot(steps, dataset_forgetting[dataset], '.-', label=dataset)
        
        # 设置图表属性
        plt.xlabel("训练步数")
        plt.ylabel("遗忘率")
        plt.title("灾难性遗忘趋势监控")
        plt.grid(True, linestyle='--', alpha=0.7)
        plt.legend(loc='best')
        
        # 保存图表
        plt.tight_layout()
        plt.savefig(output_file)
        plt.close()
        
        print(f"遗忘趋势图已保存到: {output_file}")
        return True