情感分类数据集的选择与实战应用指南

# 使用 TRAE IDE 快速加载 IMDB 数据集
import tensorflow as tf
from tensorflow.keras.datasets import imdb
 
# TRAE IDE AI 助手会自动提示参数配置
(train_data, train_labels), (test_data, test_labels) = imdb.load_data(
    num_words=10000,  # AI 推荐：限制词汇量提升训练速度
    skip_top=50,      # AI 建议：跳过高频停用词
    maxlen=None,      # 保持原始长度以保留上下文
    seed=42,          # 确保结果可复现
    start_char=1,     # 起始字符索引
    oov_char=2,       # 超出词汇表的字符
    index_from=3      # 实际词汇从索引3开始
)
 
print(f"训练集大小: {len(train_data)}")
print(f"测试集大小: {len(test_data)}")
print(f"平均评论长度: {sum(len(x) for x in train_data) / len(train_data):.1f}")

# TRAE IDE 提供的数据集评估模板
def evaluate_dataset_business_fit(dataset, business_requirements):
    """
    TRAE IDE AI 助手生成的评估框架
    """
    scores = {}
    
    # 1. 领域相关性评分 (0-1)
    domain_keywords = business_requirements['domain_keywords']
    dataset_text = ' '.join(dataset.sample(1000)['text'].tolist())
    scores['domain_relevance'] = calculate_keyword_overlap(domain_keywords, dataset_text)
    
    # 2. 情感分布平衡度 (0-1)
    sentiment_dist = dataset['label'].value_counts(normalize=True)
    scores['balance_score'] = 1 - abs(sentiment_dist.std())  # 标准差越小越好
    
    # 3. 数据质量评分 (0-1)
    scores['quality_score'] = evaluate_text_quality(dataset)
    
    # 4. 规模充足性评分 (0-1)
    min_samples = business_requirements['min_samples']
    scores['scale_adequacy'] = min(1.0, len(dataset) / min_samples)
    
    return scores

import re
import spacy
from bs4 import BeautifulSoup
import emoji
 
# TRAE IDE 智能推荐的预处理管道
class AdvancedTextPreprocessor:
    """
    工业级文本预处理，TRAE IDE 实时语法检查确保代码质量
    """
    
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm", disable=["parser", "ner"])
        # TRAE IDE 自动提示：根据任务类型选择最适合的 spaCy 模型
        
    def clean_text(self, text):
        """多阶段文本清洗"""
        # 阶段1：HTML标签清理
        text = BeautifulSoup(text, "html.parser").get_text()
        
        # 阶段2：URL和邮箱标准化
        text = re.sub(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', ' URL ', text)
        text = re.sub(r'\\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Z|a-z]{2,}\\b', ' EMAIL ', text)
        
        # 阶段3：表情符号处理
        text = emoji.demojize(text, delimiters=(" ", " "))  # 转换为文本描述
        
        # 阶段4：拼写纠正（TRAE IDE 智能推荐）
        text = self.correct_spelling(text)
        
        return text.strip()
    
    def handle_negation(self, text):
        """否定处理：关键的情感分析预处理"""
        # TRAE IDE 代码智能补全：基于情感分析的否定处理策略
        negation_patterns = [
            r"not\\s+(\w+)", r"never\\s+(\w+)", r"no\\s+(\w+)", 
            r"n't\\s+(\w+)", r"cannot\\s+(\w+)"
        ]
        
        for pattern in negation_patterns:
            text = re.sub(pattern, r"NOT_\1", text, flags=re.IGNORECASE)
            
        return text
    
    def augment_data(self, texts, labels, augmentation_factor=0.3):
        """数据增强：提升模型泛化能力"""
        augmented_texts, augmented_labels = [], []
        
        for text, label in zip(texts, labels):
            # 同义词替换
            augmented_texts.append(self.synonym_replacement(text))
            augmented_labels.append(label)
            
            # 随机插入
            if len(text.split()) > 3:
                augmented_texts.append(self.random_insertion(text))
                augmented_labels.append(label)
                
            # 随机交换
            if len(text.split()) > 4:
                augmented_texts.append(self.random_swap(text))
                augmented_labels.append(label)
                
        return augmented_texts, augmented_labels
 
# TRAE IDE 实时错误检查确保代码健壮性
preprocessor = AdvancedTextPreprocessor()

# TRAE IDE 智能标注质量检查工具
def quality_control_annotation(dataset, sample_size=1000):
    """
    多维度标注质量评估，TRAE IDE 自动代码优化
    """
    # 随机抽样进行人工复核
    sample = dataset.sample(sample_size)
    
    # 1. 计算标注者间一致性 (Inter-annotator Agreement)
    fleiss_kappa = calculate_fleiss_kappa(sample[['annotator1', 'annotator2', 'annotator3']])
    
    # 2. 检测可疑标注模式
    suspicious_patterns = detect_annotation_bias(sample)
    
    # 3. 情感一致性检查
    sentiment_consistency = check_sentiment_consistency(sample)
    
    quality_report = {
        'fleiss_kappa': fleiss_kappa,
        'suspicious_patterns': suspicious_patterns,
        'sentiment_consistency': sentiment_consistency,
        'overall_quality': 'PASS' if fleiss_kappa > 0.75 else 'FAIL'
    }
    
    return quality_report
 
# TRAE IDE 一键生成质量报告
quality_report = quality_control_annotation(your_dataset)
print(f"标注质量: {quality_report['overall_quality']}")

# TRAE IDE 智能项目模板生成的情感分析系统
import torch
import torch.nn as nn
from transformers import BertModel, BertTokenizer
from torch.utils.data import DataLoader, Dataset
import pandas as pd
from sklearn.metrics import classification_report, confusion_matrix
import mlflow  # TRAE IDE 自动推荐：实验跟踪
 
class SentimentDataset(Dataset):
    """
    自定义数据集类，TRAE IDE 实时类型检查
    """
    def __init__(self, texts, labels, tokenizer, max_length=512):
        self.texts = texts
        self.labels = labels
        self.tokenizer = tokenizer
        self.max_length = max_length
        
    def __len__(self):
        return len(self.texts)
    
    def __getitem__(self, idx):
        text = str(self.texts[idx])
        label = self.labels[idx]
        
        # TRAE IDE 智能提示：使用最适合的编码策略
        encoding = self.tokenizer(
            text,
            truncation=True,
            padding='max_length',
            max_length=self.max_length,
            return_tensors='pt'
        )
        
        return {
            'text': text,
            'input_ids': encoding['input_ids'].flatten(),
            'attention_mask': encoding['attention_mask'].flatten(),
            'label': torch.tensor(label, dtype=torch.long)
        }
 
class SentimentClassifier(nn.Module):
    """
    基于BERT的情感分类器，TRAE IDE 架构建议优化
    """
    def __init__(self, n_classes, pretrained_model='bert-base-uncased'):
        super().__init__()
        self.bert = BertModel.from_pretrained(pretrained_model)
        self.dropout = nn.Dropout(0.3)  # TRAE IDE 推荐：防止过拟合
        self.classifier = nn.Linear(self.bert.config.hidden_size, n_classes)
        
    def forward(self, input_ids, attention_mask):
        outputs = self.bert(
            input_ids=input_ids,
            attention_mask=attention_mask
        )
        
        pooled_output = outputs.pooler_output
        output = self.dropout(pooled_output)
        return self.classifier(output)
 
def train_model(model, train_loader, val_loader, epochs=5):
    """
    训练函数，集成 MLflow 跟踪，TRAE IDE 自动补全
    """
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = model.to(device)
    
    optimizer = torch.optim.AdamW(model.parameters(), lr=2e-5)
    criterion = nn.CrossEntropyLoss()
    
    # TRAE IDE 智能推荐：学习率调度策略
    scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(
        optimizer, mode='min', patience=2, factor=0.5
    )
    
    mlflow.start_run()  # 开始实验跟踪
    
    for epoch in range(epochs):
        model.train()
        total_loss = 0
        
        for batch in train_loader:
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['label'].to(device)
            
            optimizer.zero_grad()
            outputs = model(input_ids, attention_mask)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()
            
            total_loss += loss.item()
        
        # 验证阶段
        val_loss, val_accuracy = evaluate_model(model, val_loader)
        scheduler.step(val_loss)
        
        # MLflow 记录指标
        mlflow.log_metric("train_loss", total_loss / len(train_loader), epoch)
        mlflow.log_metric("val_loss", val_loss, epoch)
        mlflow.log_metric("val_accuracy", val_accuracy, epoch)
        
        print(f"Epoch {epoch+1}/{epochs} - Val Accuracy: {val_accuracy:.4f}")
    
    mlflow.end_run()
    return model
 
def evaluate_model(model, data_loader):
    """
    模型评估，TRAE IDE 实时错误检查
    """
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model.eval()
    
    total_loss = 0
    correct = 0
    total = 0
    
    with torch.no_grad():
        for batch in data_loader:
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['label'].to(device)
            
            outputs = model(input_ids, attention_mask)
            loss = nn.CrossEntropyLoss()(outputs, labels)
            
            total_loss += loss.item()
            _, predicted = torch.max(outputs.data, 1)
            total += labels.size(0)
            correct += (predicted == labels).sum().item()
    
    accuracy = correct / total
    avg_loss = total_loss / len(data_loader)
    
    return avg_loss, accuracy
 
# 生产级部署代码
if __name__ == "__main__":
    # TRAE IDE 智能环境检测
    print(f"PyTorch 版本: {torch.__version__}")
    print(f"CUDA 可用: {torch.cuda.is_available()}")
    
    # 数据加载（使用之前预处理的数据）
    train_df = pd.read_csv('processed_train.csv')
    val_df = pd.read_csv('processed_val.csv')
    
    # 初始化tokenizer
    tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
    
    # 创建数据集
    train_dataset = SentimentDataset(
        train_df['text'].values, 
        train_df['label'].values, 
        tokenizer
    )
    
    val_dataset = SentimentDataset(
        val_df['text'].values, 
        val_df['label'].values, 
        tokenizer
    )
    
    # 创建数据加载器
    train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
    val_loader = DataLoader(val_dataset, batch_size=32)
    
    # 初始化模型
    model = SentimentClassifier(n_classes=2)
    
    # 训练模型
    trained_model = train_model(model, train_loader, val_loader)
    
    print("✅ 模型训练完成！TRAE IDE 智能部署助手已就绪")

# TRAE IDE 性能分析器推荐的优化方案
import torch.cuda.amp as amp
 
def train_with_mixed_precision(model, train_loader, epochs=5):
    """
    混合精度训练，速度提升 1.5-2x，TRAE IDE 内存优化
    """
    scaler = amp.GradScaler()
    optimizer = torch.optim.AdamW(model.parameters(), lr=2e-5)
    
    for epoch in range(epochs):
        for batch in train_loader:
            optimizer.zero_grad()
            
            # 自动混合精度
            with amp.autocast():
                outputs = model(batch['input_ids'], batch['attention_mask'])
                loss = criterion(outputs, batch['labels'])
            
            scaler.scale(loss).backward()
            scaler.step(optimizer)
            scaler.update()

# 在 TRAE IDE 中，输入需求描述
# "我需要构建一个电商评论情感分析系统，数据量大约10万条"
 
# TRAE IDE AI 助手自动生成推荐代码：
def recommend_dataset(business_type="ecommerce", data_size=100000):
    """
    基于业务类型和数据规模推荐最适合的数据集
    """
    recommendations = {
        "dataset": "Amazon Reviews - Electronics",
        "reason": "电商领域相关性最高，数据质量优秀",
        "preprocessing": ["HTML清理", "拼写纠正", "否定处理"],
        "expected_accuracy": "92-95%",
        "training_time": "2-3 hours (GPU)"
    }
    return recommendations

# TRAE IDE 智能调试助手
class TRAE_Debugger:
    """
    TRAE IDE 集成的调试工具，实时分析模型性能
    """
    
    def analyze_misclassifications(self, model, val_loader, n_examples=10):
        """
        分析误分类样本，提供改进建议
        """
        misclassified = self.get_misclassified_samples(model, val_loader)
        
        # TRAE IDE 智能分析
        analysis = {
            "common_patterns": self.find_error_patterns(misclassified),
            "data_quality_issues": self.check_data_issues(misclassified),
            "model_limitations": self.identify_model_weaknesses(misclassified),
            "recommended_actions": self.suggest_improvements(misclassified)
        }
        
        return analysis