128_自我监督变体：SimCLR for Text - 推导对比学习的文本应用，代码实现无标注预训练的独特目标

传统自监督 → 掩码/因果建模
        ↓
对比学习 → SimCLR框架
        ↓
文本应用 → 语义表征学习
        ↓
预训练创新 → 高效无标注学习

import torch
import torch.nn.functional as F

def nt_xent_loss(z1, z2, temperature=0.1):
    """计算NT-Xent损失
    
    Args:
        z1: 第一个视图的表示 [batch_size, feature_dim]
        z2: 第二个视图的表示 [batch_size, feature_dim]
        temperature: 温度参数
    
    Returns:
        平均损失值
    """
    # L2归一化
    z1 = F.normalize(z1, dim=1)
    z2 = F.normalize(z2, dim=1)
    
    # 合并两个视图
    z = torch.cat([z1, z2], dim=0)
    batch_size = z1.size(0)
    
    # 计算相似度矩阵
    similarity_matrix = torch.matmul(z, z.T) / temperature
    
    # 创建掩码，区分正样本对
    mask = torch.eye(2 * batch_size, dtype=torch.bool).to(z.device)
    # 正样本对的位置：z1[i]与z2[i]，z2[i]与z1[i]
    pos_mask = torch.zeros_like(mask)
    pos_mask[:batch_size, batch_size:] = torch.eye(batch_size)
    pos_mask[batch_size:, :batch_size] = torch.eye(batch_size)
    
    # 计算损失
    # 排除对角线元素（自身）
    logits = similarity_matrix[~mask].view(2 * batch_size, -1)
    # 正样本的相似度
    pos_logits = similarity_matrix[pos_mask].view(2 * batch_size, 1)
    
    # 构造标签（正样本总是第一个）
    labels = torch.zeros(2 * batch_size, dtype=torch.long).to(z.device)
    
    # 计算交叉熵损失
    loss = F.cross_entropy(torch.cat([pos_logits, logits], dim=1), labels)
    
    return loss

import random
import numpy as np
from nltk.corpus import wordnet
import nlpaug.augmenter.word as naw
import nlpaug.augmenter.sentence as nas

def synonym_replacement(sentence, n=1):
    """同义词替换
    
    Args:
        sentence: 输入句子
        n: 替换的词的数量
    
    Returns:
        增强后的句子
    """
    words = sentence.split()
    new_words = words.copy()
    random_word_list = list(set([word for word in words if word.isalnum()]))
    random.shuffle(random_word_list)
    
    num_replaced = 0
    for random_word in random_word_list:
        synonyms = []
        for syn in wordnet.synsets(random_word):
            for lemma in syn.lemmas():
                synonym = lemma.name().replace('_', ' ')
                if synonym != random_word and synonym in words:
                    synonyms.append(synonym)
        
        if len(synonyms) > 0:
            synonym = random.choice(synonyms)
            new_words = [synonym if word == random_word else word for word in new_words]
            num_replaced += 1
            
        if num_replaced >= n:
            break
    
    return ' '.join(new_words)

def random_deletion(sentence, p=0.1):
    """随机删除词
    
    Args:
        sentence: 输入句子
        p: 删除概率
    
    Returns:
        增强后的句子
    """
    words = sentence.split()
    if len(words) <= 1:
        return sentence
    
    new_words = []
    for word in words:
        r = random.uniform(0, 1)
        if r > p:
            new_words.append(word)
    
    # 确保至少保留一个词
    if len(new_words) == 0:
        return random.choice(words)
    
    return ' '.join(new_words)

def random_swap(sentence, n=1):
    """随机交换词的位置
    
    Args:
        sentence: 输入句子
        n: 交换次数
    
    Returns:
        增强后的句子
    """
    words = sentence.split()
    if len(words) <= 1:
        return sentence
    
    new_words = words.copy()
    for _ in range(n):
        idx1, idx2 = random.sample(range(len(new_words)), 2)
        new_words[idx1], new_words[idx2] = new_words[idx2], new_words[idx1]
    
    return ' '.join(new_words)

def random_insertion(sentence, n=1):
    """随机插入词
    
    Args:
        sentence: 输入句子
        n: 插入的词的数量
    
    Returns:
        增强后的句子
    """
    words = sentence.split()
    new_words = words.copy()
    
    for _ in range(n):
        add_word(new_words)
    
    return ' '.join(new_words)

def add_word(words):
    """在句子中随机位置添加一个同义词
    
    Args:
        words: 词列表
    """
    synonyms = []
    counter = 0
    
    # 尝试找到可以插入同义词的词
    while len(synonyms) < 1 and counter < 10:
        random_word = random.choice(words) if words else ''
        for syn in wordnet.synsets(random_word):
            for lemma in syn.lemmas():
                synonym = lemma.name().replace('_', ' ')
                if synonym != random_word and synonym not in words:
                    synonyms.append(synonym)
        counter += 1
    
    # 如果找到同义词，插入到随机位置
    if len(synonyms) > 0:
        random_synonym = random.choice(synonyms)
        random_idx = random.randint(0, len(words))
        words.insert(random_idx, random_synonym)

def back_translation(sentence, source_lang='en', pivot_lang='fr'):
    """回译增强
    
    Args:
        sentence: 输入句子
        source_lang: 源语言
        pivot_lang: 中间语言
    
    Returns:
        增强后的句子
    """
    try:
        # 这里使用模拟的回译，实际应用中应使用翻译API
        # 例如，可以使用Google Translate API、DeepL API等
        # 为了演示，我们简单地返回原句的一个变形
        words = sentence.split()
        if len(words) > 3:
            # 随机交换两个词的位置模拟回译的变化
            idx1, idx2 = random.sample(range(len(words)), 2)
            words[idx1], words[idx2] = words[idx2], words[idx1]
            return ' '.join(words)
        return sentence
    except Exception as e:
        print(f"回译失败: {e}")
        return sentence

def augment_text(sentence, augmentation_type='combined', severity=1):
    """综合文本增强
    
    Args:
        sentence: 输入句子
        augmentation_type: 增强类型 ('synonym', 'deletion', 'swap', 'insertion', 'back_trans', 'combined')
        severity: 增强强度 (1-5)
    
    Returns:
        增强后的句子
    """
    if augmentation_type == 'synonym':
        n = max(1, int(len(sentence.split()) * (severity / 10)))
        return synonym_replacement(sentence, n=n)
    
    elif augmentation_type == 'deletion':
        p = 0.05 * severity
        return random_deletion(sentence, p=p)
    
    elif augmentation_type == 'swap':
        n = max(1, int(len(sentence.split()) * (severity / 20)))
        return random_swap(sentence, n=n)
    
    elif augmentation_type == 'insertion':
        n = max(1, int(len(sentence.split()) * (severity / 15)))
        return random_insertion(sentence, n=n)
    
    elif augmentation_type == 'back_trans':
        # 回译本身就是较强的增强
        return back_translation(sentence)
    
    elif augmentation_type == 'combined':
        # 综合多种增强方法
        new_sentence = sentence
        
        # 同义词替换
        n_syn = max(1, int(len(sentence.split()) * (severity / 15)))
        new_sentence = synonym_replacement(new_sentence, n=n_syn)
        
        # 随机删除
        p_del = 0.03 * severity
        new_sentence = random_deletion(new_sentence, p=p_del)
        
        # 随机交换
        n_swap = max(1, int(len(new_sentence.split()) * (severity / 25)))
        new_sentence = random_swap(new_sentence, n=n_swap)
        
        # 随机插入
        if random.random() > 0.5:  # 50%的概率进行插入
            n_ins = max(1, int(len(new_sentence.split()) * (severity / 20)))
            new_sentence = random_insertion(new_sentence, n=n_ins)
        
        return new_sentence
    
    else:
        raise ValueError(f"不支持的增强类型: {augmentation_type}")

# 示例使用
def create_text_views(text, num_views=2, augmentation_types=None):
    """为给定文本创建多个增强视图
    
    Args:
        text: 原始文本
        num_views: 要创建的视图数量
        augmentation_types: 增强类型列表，如果为None则随机选择
    
    Returns:
        增强视图列表
    """
    if augmentation_types is None:
        # 可用的增强类型
        available_types = ['synonym', 'deletion', 'swap', 'insertion', 'back_trans', 'combined']
        
        # 为每个视图随机选择增强类型
        augmentation_types = [random.choice(available_types) for _ in range(num_views)]
    
    elif len(augmentation_types) != num_views:
        raise ValueError("增强类型数量必须与视图数量匹配")
    
    views = []
    for aug_type in augmentation_types:
        # 随机选择增强强度
        severity = random.randint(1, 3)
        view = augment_text(text, augmentation_type=aug_type, severity=severity)
        views.append(view)
    
    return views

def evaluate_augmentation_strategy(texts, augmentation_types, num_samples=100):
    """评估不同增强策略的效果
    
    Args:
        texts: 文本列表
        augmentation_types: 增强类型列表
        num_samples: 采样数量
    
    Returns:
        评估结果字典
    """
    import time
    import numpy as np
    from sklearn.metrics.pairwise import cosine_similarity
    from sentence_transformers import SentenceTransformer
    
    # 加载预训练的句子嵌入模型
    model = SentenceTransformer('all-MiniLM-L6-v2')
    
    # 采样文本
    sampled_texts = random.sample(texts, min(num_samples, len(texts)))
    
    results = {}
    
    for aug_type in augmentation_types:
        start_time = time.time()
        similarities = []
        
        for text in sampled_texts:
            # 生成增强视图
            view = augment_text(text, augmentation_type=aug_type)
            
            # 计算相似度
            orig_embedding = model.encode([text])
            view_embedding = model.encode([view])
            similarity = cosine_similarity(orig_embedding, view_embedding)[0][0]
            similarities.append(similarity)
        
        avg_similarity = np.mean(similarities)
        std_similarity = np.std(similarities)
        elapsed_time = time.time() - start_time
        
        results[aug_type] = {
            'avg_similarity': avg_similarity,
            'std_similarity': std_similarity,
            'avg_time': elapsed_time / num_samples,
            'efficiency_score': avg_similarity / (elapsed_time / num_samples)  # 效率分数
        }
    
    return results

原始文本 → 增强策略1 → 编码器 → 投影头 → z₁
原始文本 → 增强策略2 → 编码器 → 投影头 → z₂
                                       ↓
                                 NT-Xent损失

import torch
import torch.nn as nn
from transformers import BertModel, BertTokenizer, RobertaModel, RobertaTokenizer

class TextEncoder(nn.Module):
    """文本编码器类
    
    支持多种预训练模型作为编码器
    """
    def __init__(self, model_name='bert-base-uncased', freeze_encoder=False):
        """初始化文本编码器
        
        Args:
            model_name: 预训练模型名称
            freeze_encoder: 是否冻结编码器参数
        """
        super().__init__()
        
        # 加载预训练模型和分词器
        if 'roberta' in model_name.lower():
            self.model = RobertaModel.from_pretrained(model_name)
            self.tokenizer = RobertaTokenizer.from_pretrained(model_name)
        else:
            self.model = BertModel.from_pretrained(model_name)
            self.tokenizer = BertTokenizer.from_pretrained(model_name)
        
        # 冻结编码器参数
        if freeze_encoder:
            for param in self.model.parameters():
                param.requires_grad = False
        
        # 获取编码器输出维度
        self.output_dim = self.model.config.hidden_size
    
    def forward(self, texts, max_length=128):
        """前向传播
        
        Args:
            texts: 文本列表
            max_length: 最大序列长度
        
        Returns:
            编码器输出表示
        """
        # 分词
        inputs = self.tokenizer(
            texts,
            padding=True,
            truncation=True,
            max_length=max_length,
            return_tensors='pt'
        )
        
        # 移动到与模型相同的设备
        for key in inputs:
            inputs[key] = inputs[key].to(self.model.device)
        
        # 获取模型输出
        outputs = self.model(**inputs)
        
        # 使用CLS标记的输出作为句子表示
        # 也可以使用其他池化策略，如平均池化
        sentence_embeddings = outputs.pooler_output
        
        return sentence_embeddings

class ProjectionHead(nn.Module):
    """投影头类
    
    用于将编码器输出映射到潜在空间
    """
    def __init__(self, input_dim, hidden_dim=256, output_dim=128, num_layers=2):
        """初始化投影头
        
        Args:
            input_dim: 输入维度（编码器输出维度）
            hidden_dim: 隐藏层维度
            output_dim: 输出维度
            num_layers: 层数
        """
        super().__init__()
        
        layers = []
        
        # 添加隐藏层
        if num_layers == 1:
            layers.append(nn.Linear(input_dim, output_dim))
        else:
            layers.append(nn.Linear(input_dim, hidden_dim))
            layers.append(nn.ReLU())
            
            # 添加中间层
            for _ in range(num_layers - 2):
                layers.append(nn.Linear(hidden_dim, hidden_dim))
                layers.append(nn.ReLU())
            
            # 添加输出层
            layers.append(nn.Linear(hidden_dim, output_dim))
        
        # 构建投影头
        self.projection = nn.Sequential(*layers)
    
    def forward(self, x):
        """前向传播
        
        Args:
            x: 编码器输出
        
        Returns:
            投影后的表示
        """
        return self.projection(x)

class TextSimCLR(nn.Module):
    """文本SimCLR模型
    
    结合编码器、投影头和对比损失函数
    """
    def __init__(self, encoder_model_name='bert-base-uncased', 
                 projection_hidden_dim=256, 
                 projection_output_dim=128,
                 projection_num_layers=2,
                 temperature=0.1):
        """初始化TextSimCLR模型
        
        Args:
            encoder_model_name: 编码器模型名称
            projection_hidden_dim: 投影头隐藏层维度
            projection_output_dim: 投影头输出维度
            projection_num_layers: 投影头层数
            temperature: 温度参数
        """
        super().__init__()
        
        # 初始化编码器
        self.encoder = TextEncoder(model_name=encoder_model_name)
        
        # 初始化投影头
        self.projection_head = ProjectionHead(
            input_dim=self.encoder.output_dim,
            hidden_dim=projection_hidden_dim,
            output_dim=projection_output_dim,
            num_layers=projection_num_layers
        )
        
        # 温度参数
        self.temperature = temperature
    
    def forward(self, text_views1, text_views2):
        """前向传播
        
        Args:
            text_views1: 第一组增强视图
            text_views2: 第二组增强视图
        
        Returns:
            模型输出和损失
        """
        # 通过编码器获取表示
        z1 = self.encoder(text_views1)
        z2 = self.encoder(text_views2)
        
        # 通过投影头
        z1_proj = self.projection_head(z1)
        z2_proj = self.projection_head(z2)
        
        # 计算对比损失
        loss = self._compute_contrastive_loss(z1_proj, z2_proj)
        
        return {
            'loss': loss,
            'z1': z1,
            'z2': z2,
            'z1_proj': z1_proj,
            'z2_proj': z2_proj
        }
    
    def _compute_contrastive_loss(self, z1, z2):
        """计算对比损失
        
        Args:
            z1: 第一组投影后的表示
            z2: 第二组投影后的表示
        
        Returns:
            对比损失
        """
        # 合并两个视图
        z = torch.cat([z1, z2], dim=0)
        batch_size = z1.size(0)
        
        # 计算相似度矩阵
        similarity_matrix = torch.matmul(z, z.T) / self.temperature
        
        # 创建掩码，区分正样本对
        mask = torch.eye(2 * batch_size, dtype=torch.bool).to(z.device)
        # 正样本对的位置：z1[i]与z2[i]，z2[i]与z1[i]
        pos_mask = torch.zeros_like(mask)
        pos_mask[:batch_size, batch_size:] = torch.eye(batch_size)
        pos_mask[batch_size:, :batch_size] = torch.eye(batch_size)
        
        # 计算损失
        # 排除对角线元素（自身）
        logits = similarity_matrix[~mask].view(2 * batch_size, -1)
        # 正样本的相似度
        pos_logits = similarity_matrix[pos_mask].view(2 * batch_size, 1)
        
        # 构造标签（正样本总是第一个）
        labels = torch.zeros(2 * batch_size, dtype=torch.long).to(z.device)
        
        # 计算交叉熵损失
        loss = F.cross_entropy(torch.cat([pos_logits, logits], dim=1), labels)
        
        return loss
    
    def get_sentence_embeddings(self, texts, max_length=128):
        """获取句子嵌入
        
        Args:
            texts: 文本列表
            max_length: 最大序列长度
        
        Returns:
            句子嵌入向量
        """
        with torch.no_grad():
            return self.encoder(texts, max_length=max_length)

from torch.utils.data import Dataset, DataLoader
import pandas as pd

class ContrastiveTextDataset(Dataset):
    """对比学习文本数据集
    
    为每个样本生成两个增强视图
    """
    def __init__(self, texts, augmentation_types1='combined', 
                 augmentation_types2='combined', severity_range=(1, 3)):
        """初始化数据集
        
        Args:
            texts: 文本列表
            augmentation_types1: 第一个视图的增强类型
            augmentation_types2: 第二个视图的增强类型
            severity_range: 增强强度范围
        """
        self.texts = texts
        self.augmentation_types1 = augmentation_types1
        self.augmentation_types2 = augmentation_types2
        self.severity_range = severity_range
    
    def __len__(self):
        return len(self.texts)
    
    def __getitem__(self, idx):
        text = self.texts[idx]
        
        # 为第一个视图生成增强
        severity1 = random.randint(self.severity_range[0], self.severity_range[1])
        view1 = augment_text(text, augmentation_type=self.augmentation_types1, severity=severity1)
        
        # 为第二个视图生成增强
        severity2 = random.randint(self.severity_range[0], self.severity_range[1])
        view2 = augment_text(text, augmentation_type=self.augmentation_types2, severity=severity2)
        
        return {
            'original': text,
            'view1': view1,
            'view2': view2
        }

def collate_fn(batch):
    """数据批处理函数
    
    Args:
        batch: 数据批次
    
    Returns:
        处理后的批次数据
    """
    original_texts = [item['original'] for item in batch]
    view1_texts = [item['view1'] for item in batch]
    view2_texts = [item['view2'] for item in batch]
    
    return {
        'original': original_texts,
        'view1': view1_texts,
        'view2': view2_texts
    }

def create_dataloader(texts, batch_size=32, shuffle=True, num_workers=4):
    """创建数据加载器
    
    Args:
        texts: 文本列表
        batch_size: 批次大小
        shuffle: 是否打乱数据
        num_workers: 工作进程数量
    
    Returns:
        数据加载器
    """
    dataset = ContrastiveTextDataset(texts)
    dataloader = DataLoader(
        dataset,
        batch_size=batch_size,
        shuffle=shuffle,
        collate_fn=collate_fn,
        num_workers=num_workers
    )
    return dataloader

import torch
import torch.optim as optim
from torch.cuda.amp import autocast, GradScaler
import numpy as np
import matplotlib.pyplot as plt
from tqdm import tqdm
import os
import json

def train_simclr(model, dataloader, val_dataloader=None, 
                learning_rate=1e-4, weight_decay=1e-6, 
                num_epochs=100, device='cuda', 
                save_dir='./checkpoints', log_dir='./logs',
                save_every=10, use_amp=False):
    """训练SimCLR模型
    
    Args:
        model: SimCLR模型
        dataloader: 训练数据加载器
        val_dataloader: 验证数据加载器
        learning_rate: 学习率
        weight_decay: 权重衰减
        num_epochs: 训练轮数
        device: 训练设备
        save_dir: 模型保存目录
        log_dir: 日志保存目录
        save_every: 每多少轮保存一次模型
        use_amp: 是否使用自动混合精度
    
    Returns:
        训练历史记录
    """
    # 确保保存目录存在
    os.makedirs(save_dir, exist_ok=True)
    os.makedirs(log_dir, exist_ok=True)
    
    # 将模型移动到指定设备
    model = model.to(device)
    
    # 优化器设置
    optimizer = optim.AdamW(
        model.parameters(),
        lr=learning_rate,
        weight_decay=weight_decay
    )
    
    # 学习率调度器
    scheduler = optim.lr_scheduler.CosineAnnealingLR(
        optimizer, T_max=num_epochs, eta_min=learning_rate * 0.01
    )
    
    # 混合精度训练
    scaler = GradScaler(enabled=use_amp)
    
    # 训练历史记录
    history = {
        'train_loss': [],
        'val_loss': [],
        'learning_rate': []
    }
    
    # 开始训练
    for epoch in range(num_epochs):
        model.train()
        epoch_loss = 0.0
        
        # 进度条
        pbar = tqdm(dataloader, desc=f'Epoch {epoch+1}/{num_epochs}')
        
        for batch in pbar:
            # 清零梯度
            optimizer.zero_grad()
            
            # 获取增强视图
            view1 = batch['view1']
            view2 = batch['view2']
            
            # 混合精度前向传播
            with autocast(enabled=use_amp):
                outputs = model(view1, view2)
                loss = outputs['loss']
            
            # 反向传播
            scaler.scale(loss).backward()
            scaler.step(optimizer)
            scaler.update()
            
            # 记录损失
            batch_loss = loss.item()
            epoch_loss += batch_loss * len(view1)
            
            # 更新进度条
            pbar.set_postfix({'loss': batch_loss})
        
        # 计算平均训练损失
        avg_train_loss = epoch_loss / len(dataloader.dataset)
        history['train_loss'].append(avg_train_loss)
        history['learning_rate'].append(optimizer.param_groups[0]['lr'])
        
        # 验证
        if val_dataloader is not None:
            model.eval()
            val_loss = 0.0
            
            with torch.no_grad():
                for batch in val_dataloader:
                    view1 = batch['view1']
                    view2 = batch['view2']
                    
                    with autocast(enabled=use_amp):
                        outputs = model(view1, view2)
                        loss = outputs['loss']
                    
                    val_loss += loss.item() * len(view1)
            
            avg_val_loss = val_loss / len(val_dataloader.dataset)
            history['val_loss'].append(avg_val_loss)
            
            print(f'Epoch {epoch+1}/{num_epochs}, ' \
                  f'Train Loss: {avg_train_loss:.4f}, ' \
                  f'Val Loss: {avg_val_loss:.4f}, ' \
                  f'LR: {optimizer.param_groups[0]["lr"]:.6f}')
        else:
            print(f'Epoch {epoch+1}/{num_epochs}, ' \
                  f'Train Loss: {avg_train_loss:.4f}, ' \
                  f'LR: {optimizer.param_groups[0]["lr"]:.6f}')
        
        # 更新学习率
        scheduler.step()
        
        # 保存模型
        if (epoch + 1) % save_every == 0:
            checkpoint_path = os.path.join(
                save_dir, f'simclr_epoch_{epoch+1}.pt'
            )
            torch.save({
                'epoch': epoch + 1,
                'model_state_dict': model.state_dict(),
                'optimizer_state_dict': optimizer.state_dict(),
                'scheduler_state_dict': scheduler.state_dict(),
                'loss': avg_train_loss,
            }, checkpoint_path)
            
            print(f'Model saved to {checkpoint_path}')
        
        # 保存历史记录
        with open(os.path.join(log_dir, 'training_history.json'), 'w') as f:
            json.dump(history, f, indent=4)
        
        # 绘制训练曲线
        plot_training_history(history, log_dir)
    
    return history

def plot_training_history(history, log_dir):
    """绘制训练历史曲线
    
    Args:
        history: 训练历史记录
        log_dir: 日志保存目录
    """
    plt.figure(figsize=(12, 5))
    
    # 损失曲线
    plt.subplot(1, 2, 1)
    plt.plot(history['train_loss'], label='Train Loss')
    if 'val_loss' in history and history['val_loss']:
        plt.plot(history['val_loss'], label='Validation Loss')
    plt.title('Loss vs. Epochs')
    plt.xlabel('Epoch')
    plt.ylabel('Loss')
    plt.legend()
    plt.grid(True, alpha=0.3)
    
    # 学习率曲线
    plt.subplot(1, 2, 2)
    plt.plot(history['learning_rate'], label='Learning Rate')
    plt.title('Learning Rate vs. Epochs')
    plt.xlabel('Epoch')
    plt.ylabel('Learning Rate')
    plt.legend()
    plt.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.savefig(os.path.join(log_dir, 'training_curves.png'))
    plt.close()

def get_optimizer(model, learning_rate=1e-4, encoder_lr_scale=0.1, 
                 weight_decay=1e-6, warmup_epochs=10, total_epochs=100,
                 dataloader=None):
    """获取优化器和学习率调度器
    
    Args:
        model: SimCLR模型
        learning_rate: 基础学习率
        encoder_lr_scale: 编码器学习率缩放因子
        weight_decay: 权重衰减
        warmup_epochs: 预热轮数
        total_epochs: 总训练轮数
        dataloader: 数据加载器（用于计算总步数）
    
    Returns:
        优化器和学习率调度器
    """
    # 分层参数组
    encoder_params = list(model.encoder.parameters())
    projection_params = list(model.projection_head.parameters())
    
    # 为不同部分设置不同的学习率
    param_groups = [
        {'params': encoder_params, 'lr': learning_rate * encoder_lr_scale},
        {'params': projection_params, 'lr': learning_rate}
    ]
    
    # 创建优化器
    optimizer = optim.AdamW(param_groups, weight_decay=weight_decay)
    
    # 如果提供了数据加载器，使用总步数而不是轮数
    if dataloader is not None:
        total_steps = total_epochs * len(dataloader)
        warmup_steps = warmup_epochs * len(dataloader)
    else:
        total_steps = total_epochs * 100  # 假设每轮100步
        warmup_steps = warmup_epochs * 100
    
    # 创建学习率调度器
    # 结合预热和余弦退火
    def lr_lambda(current_step):
        # 预热阶段
        if current_step < warmup_steps:
            return float(current_step) / float(max(1, warmup_steps))
        # 余弦退火阶段
        progress = float(current_step - warmup_steps) / \
                   float(max(1, total_steps - warmup_steps))
        return max(0.0, 0.5 * (1.0 + math.cos(math.pi * progress)))
    
    scheduler = optim.lr_scheduler.LambdaLR(optimizer, lr_lambda)
    
    return optimizer, scheduler

def train_with_gradient_accumulation(model, dataloader, 
                                   accumulation_steps=4, **kwargs):
    """使用梯度累积进行训练
    
    Args:
        model: SimCLR模型
        dataloader: 数据加载器
        accumulation_steps: 梯度累积步数
        **kwargs: 其他训练参数
    
    Returns:
        训练历史记录
    """
    # 确保保存目录存在
    save_dir = kwargs.get('save_dir', './checkpoints')
    log_dir = kwargs.get('log_dir', './logs')
    os.makedirs(save_dir, exist_ok=True)
    os.makedirs(log_dir, exist_ok=True)
    
    # 将模型移动到指定设备
    device = kwargs.get('device', 'cuda')
    model = model.to(device)
    
    # 优化器设置
    learning_rate = kwargs.get('learning_rate', 1e-4)
    weight_decay = kwargs.get('weight_decay', 1e-6)
    optimizer = optim.AdamW(
        model.parameters(),
        lr=learning_rate,
        weight_decay=weight_decay
    )
    
    # 学习率调度器
    num_epochs = kwargs.get('num_epochs', 100)
    scheduler = optim.lr_scheduler.CosineAnnealingLR(
        optimizer, T_max=num_epochs, eta_min=learning_rate * 0.01
    )
    
    # 混合精度训练
    use_amp = kwargs.get('use_amp', False)
    scaler = GradScaler(enabled=use_amp)
    
    # 训练历史记录
    history = {
        'train_loss': [],
        'val_loss': [],
        'learning_rate': []
    }
    
    # 开始训练
    for epoch in range(num_epochs):
        model.train()
        epoch_loss = 0.0
        
        # 进度条
        pbar = tqdm(dataloader, desc=f'Epoch {epoch+1}/{num_epochs}')
        
        for i, batch in enumerate(pbar):
            # 获取增强视图
            view1 = batch['view1']
            view2 = batch['view2']
            
            # 混合精度前向传播
            with autocast(enabled=use_amp):
                outputs = model(view1, view2)
                loss = outputs['loss'] / accumulation_steps  # 缩放损失
            
            # 反向传播
            scaler.scale(loss).backward()
            
            # 累积梯度
            if (i + 1) % accumulation_steps == 0:
                scaler.step(optimizer)
                scaler.update()
                optimizer.zero_grad()
            
            # 记录损失
            batch_loss = loss.item() * accumulation_steps  # 恢复原始损失值
            epoch_loss += batch_loss * len(view1)
            
            # 更新进度条
            pbar.set_postfix({'loss': batch_loss})
        
        # 确保最后一个不完整的累积组也更新梯度
        if (i + 1) % accumulation_steps != 0:
            scaler.step(optimizer)
            scaler.update()
            optimizer.zero_grad()
        
        # 计算平均训练损失
        avg_train_loss = epoch_loss / len(dataloader.dataset)
        history['train_loss'].append(avg_train_loss)
        history['learning_rate'].append(optimizer.param_groups[0]['lr'])
        
        # 验证和其他步骤与之前相同
        # ...
        
        # 更新学习率
        scheduler.step()
    
    return history

def apply_mixup(x1, x2, alpha=0.2):
    """对两个批次的样本应用MixUp
    
    Args:
        x1: 第一个批次的样本
        x2: 第二个批次的样本
        alpha: MixUp参数
    
    Returns:
        混合后的样本和混合权重
    """
    lam = np.random.beta(alpha, alpha)
    mixed_x = lam * x1 + (1 - lam) * x2
    return mixed_x, lam

# 在训练循环中应用MixUp
def train_with_mixup(model, dataloader, alpha=0.2, **kwargs):
    """使用MixUp进行训练
    
    Args:
        model: SimCLR模型
        dataloader: 数据加载器
        alpha: MixUp参数
        **kwargs: 其他训练参数
    
    Returns:
        训练历史记录
    """
    # ... 初始化代码 ...
    
    for epoch in range(num_epochs):
        # ...
        for batch in dataloader:
            view1 = batch['view1']
            view2 = batch['view2']
            
            # 前向传播获取表示
            with autocast(enabled=use_amp):
                z1 = model.encoder(view1)
                z2 = model.encoder(view2)
                
                # 应用MixUp到表示
                if np.random.random() < 0.5:  # 50%的概率应用MixUp
                    z1_mixed, lam = apply_mixup(z1, z2, alpha)
                    z2_mixed, _ = apply_mixup(z2, z1, alpha)
                    
                    # 通过投影头
                    z1_proj = model.projection_head(z1_mixed)
                    z2_proj = model.projection_head(z2_mixed)
                else:
                    z1_proj = model.projection_head(z1)
                    z2_proj = model.projection_head(z2)
                    lam = 1.0
                
                # 计算损失
                loss = model._compute_contrastive_loss(z1_proj, z2_proj)
            
            # ... 反向传播和优化 ...
    
    return history

def evaluate_model_quality(model, dataloader, device='cuda'):
    """评估模型表示质量
    
    Args:
        model: 训练好的SimCLR模型
        dataloader: 评估数据加载器
        device: 评估设备
    
    Returns:
        评估指标字典
    """
    from sklearn.cluster import KMeans
    from sklearn.metrics import silhouette_score, davies_bouldin_score
    
    model.eval()
    embeddings = []
    original_texts = []
    
    with torch.no_grad():
        for batch in tqdm(dataloader, desc='Evaluating'):
            texts = batch['original']
            batch_embeddings = model.get_sentence_embeddings(texts).cpu().numpy()
            embeddings.extend(batch_embeddings)
            original_texts.extend(texts)
    
    # 转换为numpy数组
    embeddings = np.array(embeddings)
    
    # 计算聚类指标
    n_clusters = min(10, len(embeddings) // 10)
    if n_clusters > 1:
        kmeans = KMeans(n_clusters=n_clusters, random_state=42)
        cluster_labels = kmeans.fit_predict(embeddings)
        
        silhouette = silhouette_score(embeddings, cluster_labels)
        davies_bouldin = davies_bouldin_score(embeddings, cluster_labels)
    else:
        silhouette = 0.0
        davies_bouldin = 0.0
    
    # 计算检索指标（简化版）
    # 这里我们使用余弦相似度作为检索指标
    from sklearn.metrics.pairwise import cosine_similarity
    
    # 取前100个样本进行检索评估
    n_eval = min(100, len(embeddings))
    sim_matrix = cosine_similarity(embeddings[:n_eval])
    
    # 对角线为自身相似度，设为-1
    np.fill_diagonal(sim_matrix, -1)
    
    # 计算平均精度均值（简化版）
    mean_precision = []
    for i in range(n_eval):
        # 找到最相似的样本
        top_idx = np.argmax(sim_matrix[i])
        # 在这个简化评估中，我们假设如果两个文本相似（这里简单地以长度作为判断）
        # 这只是一个演示，实际应用中应该使用更复杂的相似度判断
        text1_len = len(original_texts[i].split())
        text2_len = len(original_texts[top_idx].split())
        is_relevant = abs(text1_len - text2_len) < max(3, 0.2 * text1_len)
        mean_precision.append(1.0 if is_relevant else 0.0)
    
    mean_precision = np.mean(mean_precision)
    
    return {
        'silhouette_score': silhouette,
        'davies_bouldin_score': davies_bouldin,
        'mean_precision': mean_precision
    }

class DownstreamTaskModel(nn.Module):
    """下游任务模型
    
    使用SimCLR编码器作为基础，添加任务特定的头部
    """
    def __init__(self, simclr_model, num_classes=2, freeze_encoder=False):
        """初始化下游任务模型
        
        Args:
            simclr_model: 预训练的SimCLR模型
            num_classes: 类别数量
            freeze_encoder: 是否冻结编码器
        """
        super().__init__()
        
        # 使用SimCLR的编码器
        self.encoder = simclr_model.encoder
        
        # 冻结编码器参数
        if freeze_encoder:
            for param in self.encoder.parameters():
                param.requires_grad = False
        
        # 添加任务特定的分类头
        self.classifier = nn.Sequential(
            nn.Linear(self.encoder.output_dim, 256),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(256, num_classes)
        )
    
    def forward(self, texts):
        """前向传播
        
        Args:
            texts: 输入文本
        
        Returns:
            分类logits
        """
        # 获取编码器表示
        embeddings = self.encoder(texts)
        
        # 分类
        logits = self.classifier(embeddings)
        
        return logits

def fine_tune_downstream(model, train_loader, val_loader, 
                        num_epochs=20, learning_rate=1e-4,
                        weight_decay=1e-6, device='cuda'):
    """微调下游任务模型
    
    Args:
        model: 下游任务模型
        train_loader: 训练数据加载器
        val_loader: 验证数据加载器
        num_epochs: 训练轮数
        learning_rate: 学习率
        weight_decay: 权重衰减
        device: 训练设备
    
    Returns:
        微调后的模型和训练历史
    """
    # 将模型移动到指定设备
    model = model.to(device)
    
    # 优化器
    optimizer = optim.AdamW(
        model.parameters(),
        lr=learning_rate,
        weight_decay=weight_decay
    )
    
    # 损失函数
    criterion = nn.CrossEntropyLoss()
    
    # 学习率调度器
    scheduler = optim.lr_scheduler.CosineAnnealingLR(
        optimizer, T_max=num_epochs
    )
    
    # 训练历史
    history = {
        'train_loss': [],
        'train_acc': [],
        'val_loss': [],
        'val_acc': []
    }
    
    # 开始微调
    for epoch in range(num_epochs):
        # 训练阶段
        model.train()
        train_loss = 0.0
        correct = 0
        total = 0
        
        for batch in tqdm(train_loader, desc=f'Epoch {epoch+1}/{num_epochs}'):
            texts = batch['text']
            labels = batch['label'].to(device)
            
            # 清零梯度
            optimizer.zero_grad()
            
            # 前向传播
            outputs = model(texts)
            loss = criterion(outputs, labels)
            
            # 反向传播和优化
            loss.backward()
            optimizer.step()
            
            # 统计
            train_loss += loss.item() * len(texts)
            _, predicted = outputs.max(1)
            total += labels.size(0)
            correct += predicted.eq(labels).sum().item()
        
        # 计算平均训练损失和准确率
        avg_train_loss = train_loss / len(train_loader.dataset)
        train_acc = correct / total
        
        # 验证阶段
        model.eval()
        val_loss = 0.0
        correct = 0
        total = 0
        
        with torch.no_grad():
            for batch in val_loader:
                texts = batch['text']
                labels = batch['label'].to(device)
                
                outputs = model(texts)
                loss = criterion(outputs, labels)
                
                val_loss += loss.item() * len(texts)
                _, predicted = outputs.max(1)
                total += labels.size(0)
                correct += predicted.eq(labels).sum().item()
        
        # 计算平均验证损失和准确率
        avg_val_loss = val_loss / len(val_loader.dataset)
        val_acc = correct / total
        
        # 更新历史记录
        history['train_loss'].append(avg_train_loss)
        history['train_acc'].append(train_acc)
        history['val_loss'].append(avg_val_loss)
        history['val_acc'].append(val_acc)
        
        # 更新学习率
        scheduler.step()
        
        print(f'Epoch {epoch+1}/{num_epochs}, ' \
              f'Train Loss: {avg_train_loss:.4f}, ' \
              f'Train Acc: {train_acc:.4f}, ' \
              f'Val Loss: {avg_val_loss:.4f}, ' \
              f'Val Acc: {val_acc:.4f}')
    
    return model, history

def visualize_results(simclr_history, downstream_history, save_dir='./results'):
    """可视化训练和评估结果
    
    Args:
        simclr_history: SimCLR训练历史
        downstream_history: 下游任务训练历史
        save_dir: 结果保存目录
    """
    import matplotlib.pyplot as plt
    import seaborn as sns
    import numpy as np
    
    os.makedirs(save_dir, exist_ok=True)
    
    # 设置绘图风格
    sns.set(style="whitegrid")
    
    # 绘制SimCLR训练损失
    plt.figure(figsize=(10, 6))
    plt.plot(simclr_history['train_loss'], label='Train Loss')
    if 'val_loss' in simclr_history and simclr_history['val_loss']:
        plt.plot(simclr_history['val_loss'], label='Validation Loss')
    plt.title('SimCLR Training Loss')
    plt.xlabel('Epoch')
    plt.ylabel('Loss')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.savefig(os.path.join(save_dir, 'simclr_loss.png'), dpi=300, bbox_inches='tight')
    plt.close()
    
    # 绘制下游任务性能
    plt.figure(figsize=(15, 5))
    
    # 损失曲线
    plt.subplot(1, 2, 1)
    plt.plot(downstream_history['train_loss'], label='Train Loss')
    plt.plot(downstream_history['val_loss'], label='Validation Loss')
    plt.title('Downstream Task Loss')
    plt.xlabel('Epoch')
    plt.ylabel('Loss')
    plt.legend()
    plt.grid(True, alpha=0.3)
    
    # 准确率曲线
    plt.subplot(1, 2, 2)
    plt.plot(downstream_history['train_acc'], label='Train Accuracy')
    plt.plot(downstream_history['val_acc'], label='Validation Accuracy')
    plt.title('Downstream Task Accuracy')
    plt.xlabel('Epoch')
    plt.ylabel('Accuracy')
    plt.legend()
    plt.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.savefig(os.path.join(save_dir, 'downstream_performance.png'), dpi=300, bbox_inches='tight')
    plt.close()
    
    # 绘制t-SNE可视化
    def plot_tsne(embeddings, labels, save_path):
        from sklearn.manifold import TSNE
        
        # 应用t-SNE降维
        tsne = TSNE(n_components=2, random_state=42, perplexity=30)
        embeddings_2d = tsne.fit_transform(embeddings)
        
        # 绘制散点图
        plt.figure(figsize=(10, 8))
        scatter = plt.scatter(
            embeddings_2d[:, 0], 
            embeddings_2d[:, 1], 
            c=labels, 
            cmap='viridis',
            alpha=0.7,
            s=50
        )
        
        # 添加颜色条
        plt.colorbar(scatter)
        plt.title('t-SNE Visualization of Text Embeddings')
        plt.xlabel('Dimension 1')
        plt.ylabel('Dimension 2')
        plt.grid(True, alpha=0.3)
        plt.savefig(save_path, dpi=300, bbox_inches='tight')
        plt.close()
    
    # 示例：加载嵌入和标签并进行t-SNE可视化
    # 这里需要根据实际情况加载数据
    # plot_tsne(embeddings, labels, os.path.join(save_dir, 'tsne_visualization.png'))

def compare_with_baselines(downstream_results, baseline_results, save_dir='./results'):
    """比较模型与基线方法的性能
    
    Args:
        downstream_results: 下游任务结果
        baseline_results: 基线方法结果
        save_dir: 结果保存目录
    """
    import matplotlib.pyplot as plt
    import numpy as np
    
    os.makedirs(save_dir, exist_ok=True)
    
    # 准备数据
    models = list(baseline_results.keys()) + ['SimCLR']
    accuracies = list(baseline_results.values()) + [downstream_results['val_acc'][-1]]
    
    # 绘制条形图
    plt.figure(figsize=(10, 6))
    bars = plt.bar(models, accuracies, color=['gray']*(len(models)-1) + ['blue'])
    
    # 在条形上方添加数值标签
    for bar in bars:
        height = bar.get_height()
        plt.text(bar.get_x() + bar.get_width()/2., height + 0.01,
                f'{height:.3f}', ha='center', va='bottom')
    
    plt.title('Performance Comparison with Baselines')
    plt.ylabel('Accuracy')
    plt.ylim(0, max(accuracies) * 1.1)
    plt.grid(True, axis='y', alpha=0.3)
    
    plt.tight_layout()
    plt.savefig(os.path.join(save_dir, 'baseline_comparison.png'), dpi=300, bbox_inches='tight')
    plt.close()

def optimize_temperature(model, dataloader, temperature_range=[0.05, 0.1, 0.2, 0.5, 1.0], device='cuda'):
    """优化温度参数
    
    Args:
        model: SimCLR模型
        dataloader: 验证数据加载器
        temperature_range: 温度参数候选值
        device: 计算设备
    
    Returns:
        最佳温度参数和对应的损失值
    """
    model.eval()
    original_temperature = model.temperature
    best_temperature = original_temperature
    best_loss = float('inf')
    
    results = {}
    
    with torch.no_grad():
        for temp in temperature_range:
            model.temperature = temp
            total_loss = 0.0
            
            for batch in tqdm(dataloader, desc=f'Evaluating temperature {temp}'):
                view1 = batch['view1']
                view2 = batch['view2']
                
                outputs = model(view1, view2)
                total_loss += outputs['loss'].item() * len(view1)
            
            avg_loss = total_loss / len(dataloader.dataset)
            results[temp] = avg_loss
            
            if avg_loss < best_loss:
                best_loss = avg_loss
                best_temperature = temp
    
    # 恢复原始温度参数
    model.temperature = original_temperature
    
    # 可视化不同温度参数的损失
    import matplotlib.pyplot as plt
    
    temps = list(results.keys())
    losses = list(results.values())
    
    plt.figure(figsize=(10, 6))
    plt.plot(temps, losses, 'o-')
    plt.axvline(x=best_temperature, color='r', linestyle='--', label=f'Best T={best_temperature}')
    plt.title('Temperature Parameter Optimization')
    plt.xlabel('Temperature')
    plt.ylabel('Loss')
    plt.xscale('log')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.savefig('temperature_optimization.png', dpi=300, bbox_inches='tight')
    plt.close()
    
    return best_temperature, best_loss, results

class TextMoCo(nn.Module):
    """文本MoCo模型
    
    使用动量编码器和队列机制
    """
    def __init__(self, encoder_model_name='bert-base-uncased',
                 projection_dim=128, queue_size=65536, 
                 momentum=0.999, temperature=0.07):
        """初始化MoCo模型
        
        Args:
            encoder_model_name: 编码器模型名称
            projection_dim: 投影维度
            queue_size: 队列大小
            momentum: 动量更新参数
            temperature: 温度参数
        """
        super().__init__()
        
        # 在线编码器
        self.encoder_q = TextEncoder(model_name=encoder_model_name)
        self.projection_q = ProjectionHead(
            input_dim=self.encoder_q.output_dim,
            output_dim=projection_dim
        )
        
        # 目标编码器（动量更新）
        self.encoder_k = TextEncoder(model_name=encoder_model_name)
        self.projection_k = ProjectionHead(
            input_dim=self.encoder_k.output_dim,
            output_dim=projection_dim
        )
        
        # 初始化目标编码器参数，使其与在线编码器相同
        for param_q, param_k in zip(
            self.encoder_q.parameters(), self.encoder_k.parameters()
        ):
            param_k.data.copy_(param_q.data)
            param_k.requires_grad = False
        
        for param_q, param_k in zip(
            self.projection_q.parameters(), self.projection_k.parameters()
        ):
            param_k.data.copy_(param_q.data)
            param_k.requires_grad = False
        
        # 队列设置
        self.queue_size = queue_size
        self.momentum = momentum
        self.temperature = temperature
        
        # 创建负样本队列
        self.register_buffer('queue', torch.randn(projection_dim, queue_size))
        self.queue = nn.functional.normalize(self.queue, dim=0)
        self.register_buffer('queue_ptr', torch.zeros(1, dtype=torch.long))
    
    @torch.no_grad()
    def _momentum_update_key_encoder(self):
        """动量更新目标编码器
        """
        for param_q, param_k in zip(
            self.encoder_q.parameters(), self.encoder_k.parameters()
        ):
            param_k.data = param_k.data * self.momentum + param_q.data * (1. - self.momentum)
        
        for param_q, param_k in zip(
            self.projection_q.parameters(), self.projection_k.parameters()
        ):
            param_k.data = param_k.data * self.momentum + param_q.data * (1. - self.momentum)
    
    @torch.no_grad()
    def _dequeue_and_enqueue(self, keys):
        """更新队列
        
        Args:
            keys: 要入队的键
        """
        batch_size = keys.shape[0]
        
        ptr = int(self.queue_ptr)
        assert self.queue_size % batch_size == 0  # 确保批次大小是队列大小的因数
        
        # 替换队列中的条目
        self.queue[:, ptr:ptr + batch_size] = keys.T
        ptr = (ptr + batch_size) % self.queue_size  # 循环更新指针
        
        self.queue_ptr[0] = ptr
    
    def forward(self, im_q, im_k):
        """前向传播
        
        Args:
            im_q: 查询视图
            im_k: 键视图
        
        # 使用在线编码器计算查询特征
        q = self.encoder_q(im_q)
        q = self.projection_q(q)
        q = nn.functional.normalize(q, dim=1)
        
        # 使用目标编码器计算键特征
        with torch.no_grad():
            self._momentum_update_key_encoder()  # 更新目标编码器
            k = self.encoder_k(im_k)
            k = self.projection_k(k)
            k = nn.functional.normalize(k, dim=1)
        
        # 计算logits
        # positive logits: Nx1
        l_pos = torch.einsum('nc,nc->n', [q, k]).unsqueeze(-1)
        # negative logits: NxK
        l_neg = torch.einsum('nc,ck->nk', [q, self.queue.clone().detach()])
        
        # 合并logits
        logits = torch.cat([l_pos, l_neg], dim=1)
        
        # 应用温度
        logits /= self.temperature
        
        # 二进制交叉熵损失
        labels = torch.zeros(logits.shape[0], dtype=torch.long).to(logits.device)
        loss = nn.CrossEntropyLoss()(logits, labels)
        
        # 更新队列
        self._dequeue_and_enqueue(k)
        
        return {
            'loss': loss,
            'q': q,
            'k': k
        }

class MultiTaskSimCLR(nn.Module):
    """多任务SimCLR模型
    
    结合对比学习和其他预训练任务
    """
    def __init__(self, encoder_model_name='bert-base-uncased',
                 projection_dim=128, temperature=0.1,
                 mlm_probability=0.15):
        """初始化多任务SimCLR模型
        
        Args:
            encoder_model_name: 编码器模型名称
            projection_dim: 投影维度
            temperature: 温度参数
            mlm_probability: 掩码语言建模的掩码概率
        """
        super().__init__()
        
        # 编码器（使用BERT作为基础）
        from transformers import BertForMaskedLM
        self.encoder = BertModel.from_pretrained(encoder_model_name)
        self.tokenizer = BertTokenizer.from_pretrained(encoder_model_name)
        
        # 投影头（用于对比学习）
        self.projection_head = ProjectionHead(
            input_dim=self.encoder.config.hidden_size,
            output_dim=projection_dim
        )
        
        # MLM头（用于掩码语言建模）
        self.mlm_head = BertForMaskedLM.from_pretrained(
            encoder_model_name
        ).cls
        
        # 温度参数
        self.temperature = temperature
        self.mlm_probability = mlm_probability
    
    def forward(self, text_views1, text_views2):
        """前向传播
        
        Args:
            text_views1: 第一组增强视图
            text_views2: 第二组增强视图
        
        Returns:
            模型输出和损失
        """
        # 分词并添加掩码（用于MLM）
        inputs1 = self._prepare_mlm_inputs(text_views1)
        inputs2 = self._prepare_mlm_inputs(text_views2)
        
        # 将输入移至模型设备
        device = next(self.encoder.parameters()).device
        for key in inputs1:
            inputs1[key] = inputs1[key].to(device)
            inputs2[key] = inputs2[key].to(device)
        
        # 编码器前向传播
        outputs1 = self.encoder(**inputs1)
        outputs2 = self.encoder(**inputs2)
        
        # 获取序列表示和池化表示
        sequence_output1 = outputs1.last_hidden_state
        sequence_output2 = outputs2.last_hidden_state
        pooled_output1 = outputs1.pooler_output
        pooled_output2 = outputs2.pooler_output
        
        # 对比学习部分
        z1_proj = self.projection_head(pooled_output1)
        z2_proj = self.projection_head(pooled_output2)
        contrastive_loss = self._compute_contrastive_loss(z1_proj, z2_proj)
        
        # 掩码语言建模部分
        mlm_loss1 = self._compute_mlm_loss(sequence_output1, inputs1)
        mlm_loss2 = self._compute_mlm_loss(sequence_output2, inputs2)
        mlm_loss = (mlm_loss1 + mlm_loss2) / 2
        
        # 总损失
        total_loss = contrastive_loss + mlm_loss
        
        return {
            'loss': total_loss,
            'contrastive_loss': contrastive_loss,
            'mlm_loss': mlm_loss,
            'z1': pooled_output1,
            'z2': pooled_output2
        }
    
    def _prepare_mlm_inputs(self, texts):
        """准备掩码语言建模的输入
        
        Args:
            texts: 文本列表
        
        Returns:
            分词后的输入，包含掩码标记
        """
        # 分词
        inputs = self.tokenizer(
            texts,
            padding=True,
            truncation=True,
            max_length=128,
            return_tensors='pt'
        )
        
        # 创建掩码输入
        input_ids = inputs['input_ids'].clone()
        labels = input_ids.clone()
        
        # 随机掩码标记
        probability_matrix = torch.full(labels.shape, self.mlm_probability)
        special_tokens_mask = [
            self.tokenizer.get_special_tokens_mask(val, already_has_special_tokens=True)
            for val in labels.tolist()
        ]
        probability_matrix.masked_fill_(torch.tensor(special_tokens_mask, dtype=torch.bool), value=0.0)
        masked_indices = torch.bernoulli(probability_matrix).bool()
        labels[~masked_indices] = -100  # 非掩码标记不计算损失
        
        # 80%的概率替换为[MASK]
        indices_replaced = torch.bernoulli(torch.full(labels.shape, 0.8)).bool() & masked_indices
        input_ids[indices_replaced] = self.tokenizer.convert_tokens_to_ids(self.tokenizer.mask_token)
        
        # 10%的概率替换为随机标记
        indices_random = torch.bernoulli(torch.full(labels.shape, 0.5)).bool() & masked_indices & ~indices_replaced
        random_words = torch.randint(len(self.tokenizer), labels.shape, dtype=torch.long)
        input_ids[indices_random] = random_words[indices_random]
        
        # 10%的概率保持不变
        
        inputs['input_ids'] = input_ids
        inputs['labels'] = labels
        
        return inputs
    
    def _compute_mlm_loss(self, sequence_output, inputs):
        """计算掩码语言建模损失
        
        Args:
            sequence_output: 序列输出
            inputs: 输入，包含labels
        
        Returns:
            MLM损失
        """
        prediction_scores = self.mlm_head(sequence_output)
        loss_fct = nn.CrossEntropyLoss()
        mlm_loss = loss_fct(
            prediction_scores.view(-1, self.tokenizer.vocab_size),
            inputs['labels'].view(-1)
        )
        return mlm_loss
    
    def _compute_contrastive_loss(self, z1, z2):
        """计算对比损失
        
        Args:
            z1: 第一组投影后的表示
            z2: 第二组投影后的表示
        
        Returns:
            对比损失
        """
        # 与标准SimCLR相同的对比损失计算
        # ...

def low_resource_experiment(monolingual_data, num_samples_list=[100, 500, 1000, 5000],
                          target_language='fr', device='cuda'):
    """低资源语言实验
    
    Args:
        monolingual_data: 单语语料库
        num_samples_list: 不同样本数量
        target_language: 目标语言
        device: 计算设备
    
    Returns:
        实验结果
    """
    results = {}
    
    for num_samples in num_samples_list:
        print(f'\n实验：使用{num_samples}个样本')
        
        # 采样数据
        sampled_data = random.sample(monolingual_data, num_samples)
        
        # 创建数据加载器
        train_loader = create_dataloader(sampled_data, batch_size=16)
        
        # 初始化模型
        model = TextSimCLR(encoder_model_name='bert-base-multilingual-cased').to(device)
        
        # 训练模型
        history = train_simclr(
            model,
            train_loader,
            num_epochs=50,
            learning_rate=1e-4,
            device=device,
            save_dir=f'./checkpoints_{num_samples}',
            save_every=10
        )
        
        # 评估模型（使用交叉语言任务）
        # 这里需要根据实际情况准备评估数据
        # eval_results = evaluate_cross_lingual_task(model, target_language)
        
        results[num_samples] = {
            'history': history,
            # 'eval_results': eval_results
        }
    
    # 可视化不同样本数量的性能
    plt.figure(figsize=(10, 6))
    
    for num_samples, res in results.items():
        plt.plot(res['history']['train_loss'], label=f'{num_samples} samples')
    
    plt.title('Training Loss with Different Sample Sizes')
    plt.xlabel('Epoch')
    plt.ylabel('Loss')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.savefig('low_resource_experiment.png', dpi=300, bbox_inches='tight')
    plt.close()
    
    return results

def zero_shot_classification(model, queries, candidates, top_k=5):
    """零样本分类
    
    Args:
        model: 训练好的SimCLR模型
        queries: 查询文本
        candidates: 候选类别描述
        top_k: 返回前k个结果
    
    Returns:
        分类结果
    """
    model.eval()
    
    with torch.no_grad():
        # 获取查询的嵌入
        query_embeddings = model.get_sentence_embeddings(queries)
        
        # 获取候选类别的嵌入
        candidate_embeddings = model.get_sentence_embeddings(candidates)
        
        # 计算余弦相似度
        query_embeddings = F.normalize(query_embeddings, dim=1)
        candidate_embeddings = F.normalize(candidate_embeddings, dim=1)
        
        similarities = torch.matmul(query_embeddings, candidate_embeddings.T)
        
        # 获取top-k结果
        top_scores, top_indices = similarities.topk(top_k, dim=1)
    
    # 整理结果
    results = []
    for i, query in enumerate(queries):
        query_results = []
        for j in range(top_k):
            query_results.append({
                'candidate': candidates[top_indices[i][j]],
                'score': top_scores[i][j].item()
            })
        results.append({
            'query': query,
            'predictions': query_results
        })
    
    return results

class CrossModalSimCLR(nn.Module):
    """跨模态SimCLR模型
    
    用于学习文本和图像的联合表示
    """
    def __init__(self, text_model_name='bert-base-uncased',
                 vision_model_name='resnet50',
                 projection_dim=128, temperature=0.07):
        """初始化跨模态SimCLR模型
        
        Args:
            text_model_name: 文本编码器模型名称
            vision_model_name: 视觉编码器模型名称
            projection_dim: 投影维度
            temperature: 温度参数
        """
        super().__init__()
        
        # 文本编码器
        self.text_encoder = TextEncoder(model_name=text_model_name)
        
        # 视觉编码器（简化版）
        # 在实际应用中，应该使用预训练的视觉模型
        self.vision_encoder = nn.Sequential(
            nn.Conv2d(3, 64, kernel_size=7, stride=2, padding=3),
            nn.ReLU(),
            nn.AdaptiveAvgPool2d((1, 1)),
            nn.Flatten()
        )
        self.vision_output_dim = 64
        
        # 文本和视觉的投影头
        self.text_projection = ProjectionHead(
            input_dim=self.text_encoder.output_dim,
            output_dim=projection_dim
        )
        self.vision_projection = ProjectionHead(
            input_dim=self.vision_output_dim,
            output_dim=projection_dim
        )
        
        # 温度参数
        self.temperature = temperature
    
    def forward(self, texts, images):
        """前向传播
        
        Args:
            texts: 文本输入
            images: 图像输入
        
        Returns:
            模型输出和损失
        """
        # 获取文本和图像表示
        text_embeddings = self.text_encoder(texts)
        image_embeddings = self.vision_encoder(images)
        
        # 投影到潜在空间
        text_features = self.text_projection(text_embeddings)
        image_features = self.vision_projection(image_embeddings)
        
        # L2归一化
        text_features = F.normalize(text_features, dim=1)
        image_features = F.normalize(image_features, dim=1)
        
        # 计算对比损失
        loss = self._compute_cross_modal_loss(text_features, image_features)
        
        return {
            'loss': loss,
            'text_features': text_features,
            'image_features': image_features
        }
    
    def _compute_cross_modal_loss(self, text_features, image_features):
        """计算跨模态对比损失
        
        Args:
            text_features: 文本特征
            image_features: 图像特征
        
        Returns:
            跨模态对比损失
        """
        batch_size = text_features.size(0)
        
        # 计算相似度矩阵
        logits = torch.matmul(text_features, image_features.t()) / self.temperature
        
        # 正样本是对角线元素（文本和对应的图像）
        labels = torch.arange(batch_size, device=logits.device)
        
        # 双向损失
        loss_i2t = F.cross_entropy(logits, labels)  # 图像到文本
        loss_t2i = F.cross_entropy(logits.t(), labels)  # 文本到图像
        
        # 总损失
        loss = (loss_i2t + loss_t2i) / 2
        
        return loss

SimCLR文本应用 → 理论推导 → 架构设计 → 代码实现 → 优化策略 → 应用场景