大模型应用：情感分析模型微调深度分析：从基础预测到性能优化.6

import torch
import torch.nn as nn
from transformers import AutoTokenizer, AutoModelForSequenceClassification, Trainer, TrainingArguments
from datasets import Dataset
import pandas as pd
from sklearn.model_selection import train_test_split
import numpy as np
from modelscope import snapshot_download


class SentimentAnalyzer:
    def __init__(self):
        """
        初始化情感分析器
        
        参数:
            model_name: 预训练模型名称，这里使用中文BERT模型
        """
        cache_dir = "D:\\modelscope\\hub"
        model_name = "google-bert/bert-base-chinese" # 使用中文BERT基础模型
        local_model_path = snapshot_download(model_name, cache_dir=cache_dir)
        self.tokenizer = AutoTokenizer.from_pretrained(local_model_path)
        self.model = AutoModelForSequenceClassification.from_pretrained(
            local_model_path, num_labels=3    # 情感分类：0-负面, 1-中性, 2-正面
        )
        
    def prepare_data(self, texts, labels):
        """准备训练数据"""
        # 对中文文本进行分词和编码
        encodings = self.tokenizer(
            texts, 
            truncation=True,    # 超过最大长度时截断文本
            padding=True,       # 填充文本到统一长度
            max_length=128,     # 最大文本长度（token数）
            return_tensors='pt'
        )
        
        # 创建数据集
        dataset = Dataset.from_dict({
            'input_ids': encodings['input_ids'],
            'attention_mask': encodings['attention_mask'],
            'labels': labels
        })
        return dataset
    
    def fine_tune(self, train_texts, train_labels, val_texts=None, val_labels=None):
        """微调模型"""
        train_dataset = self.prepare_data(train_texts, train_labels)
        
        if val_texts is not None:
            val_dataset = self.prepare_data(val_texts, val_labels)
        else:
            val_dataset = None
        
        # 设置训练参数
        training_args = TrainingArguments(
            output_dir='./results',
            num_train_epochs=3,                   # 训练3轮
            per_device_train_batch_size=16,       # 每个设备批大小16
            per_device_eval_batch_size=16,        # 评估批大小16
            warmup_steps=500,                     # 学习率预热步数
            weight_decay=0.01,                    # 权重衰减（正则化）
            logging_dir='./logs',
            evaluation_strategy="epoch" if val_dataset else "no",
            save_strategy="epoch",
            load_best_model_at_end=True if val_dataset else False,
        )
        
        # 创建训练器
        trainer = Trainer(
            model=self.model,
            args=training_args,
            train_dataset=train_dataset,
            eval_dataset=val_dataset,
        )
        
        # 开始训练
        trainer.train()
        return trainer

# 创建中文示例数据
sample_texts = [
    "这部电影真是太棒了！剧情紧凑，演员表演出色。",
    "这个产品质量很差，完全不值得购买。",
    "还行吧，没有什么特别之处。",
    "质量很好，送货也很快，推荐购买！",
    "客服态度恶劣，购物体验极差。"
]

sample_labels = [2, 0, 1, 2, 0]  # 0:负面, 1:中性, 2:正面

# 初始化并微调模型
analyzer = SentimentAnalyzer()
trainer = analyzer.fine_tune(sample_texts, sample_labels)
print("-------------------------模型微调结束------------------------")
print(trainer)

import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.metrics import classification_report, confusion_matrix
import pandas as pd
import numpy as np
import torch

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
plt.rcParams['axes.unicode_minus'] = False  # 用来正常显示负号

class PredictionAnalyzer:
    def __init__(self, model, tokenizer):
        """
        预测分析器初始化
        
        参数:
            model: 训练好的模型
            tokenizer: 对应的分词器
        """
        self.model = model
        self.tokenizer = tokenizer
        self.model.eval()  # 设置为评估模式
    
    def predict_batch(self, texts):
        """批量预测文本情感"""
        # 对中文文本进行编码
        encodings = self.tokenizer(
            texts, 
            truncation=True, 
            padding=True, 
            max_length=128,
            return_tensors='pt'
        )
        
        # 进行预测
        with torch.no_grad():
            outputs = self.model(**encodings)
            predictions = torch.nn.functional.softmax(outputs.logits, dim=-1)
        
        return predictions.numpy()
    
    def get_detailed_predictions(self, texts, true_labels=None):
        """
        获取详细的预测结果
        
        返回包含预测标签、置信度、概率分布等信息的DataFrame
        """
        predictions = self.predict_batch(texts)
        predicted_labels = np.argmax(predictions, axis=1)
        
        results = []
        for i, text in enumerate(texts):
            result = {
                'text': text,
                'predicted_label': predicted_labels[i],
                'confidence': np.max(predictions[i]),  # 最大概率值作为置信度
                'probabilities': predictions[i]  # 各类别的概率分布
            }
            
            if true_labels is not None:
                result['true_label'] = true_labels[i]
                result['correct'] = (predicted_labels[i] == true_labels[i])
            
            results.append(result)
        
        return pd.DataFrame(results)

# 使用示例
def analyze_predictions_demo():
    """预测分析演示函数"""
    # 创建测试数据（中文）
    test_texts = [
        "这个产品真的很棒，我非常喜欢！",
        "质量太差了，完全不符合描述。",
        "一般般吧，没什么特别的感觉。",
        "物流速度快，包装完好，很满意！",
        "售后服务态度极差，再也不会购买了。"
    ]
    test_labels = [2, 0, 1, 2, 0]  # 真实标签
    
    # 在实际使用中，这里需要已经训练好的模型
    # analyzer = SentimentAnalyzer()
    # 假设我们已经有了训练好的analyzer对象
    
    print("正在进行预测分析...")
    # 模拟预测结果（实际使用时取消注释下面两行）
    # analyzer = PredictionAnalyzer(analyzer.model, analyzer.tokenizer)
    # results_df = analyzer.get_detailed_predictions(test_texts, test_labels)
    
    # 为了演示效果，我们创建了一些包含错误预测的结果
    results_df = pd.DataFrame({
        'text': test_texts,
        'true_label': test_labels,
        'predicted_label': [2, 0, 2, 2, 0],  # 模拟预测结果，第三个预测错误
        'confidence': [0.95, 0.87, 0.65, 0.92, 0.88],
        'correct': [True, True, False, True, True],
        'probabilities': [
            [0.02, 0.03, 0.95],
            [0.87, 0.10, 0.03],
            [0.25, 0.10, 0.65],
            [0.05, 0.03, 0.92],
            [0.88, 0.08, 0.04]
        ]
    })
    
    print("预测结果详情:")
    print("=" * 80)
    for _, row in results_df.iterrows():
        status = "✓" if row['correct'] else "✗"
        true_sentiment = ["负面", "中性", "正面"][row['true_label']]
        pred_sentiment = ["负面", "中性", "正面"][row['predicted_label']]
        print(f"{status} 文本: {row['text']}")
        print(f"   真实情感: {true_sentiment}, 预测情感: {pred_sentiment}, 置信度: {row['confidence']:.3f}")
        print()
    
    return results_df

# 执行预测分析
results_df = analyze_predictions_demo()

def analyze_confidence_distribution(results_df):
    """分析预测置信度的分布情况"""
    print("正在进行置信度分析...")
    
    fig, axes = plt.subplots(1, 2, figsize=(15, 5))
    
    # 正确和错误预测的置信度分布对比
    correct_conf = results_df[results_df['correct']]['confidence']
    incorrect_conf = results_df[~results_df['correct']]['confidence']
    
    # 绘制直方图
    axes[0].hist([correct_conf, incorrect_conf], 
                bins=10, alpha=0.7, 
                label=['正确预测', '错误预测'],
                color=['green', 'red'])
    axes[0].set_xlabel('置信度')
    axes[0].set_ylabel('频次')
    axes[0].set_title('正确 vs 错误预测的置信度分布')
    axes[0].legend()
    axes[0].grid(True, alpha=0.3)
    
    # 按情感类别分析置信度分布
    labels = ['负面', '中性', '正面']
    colors = ['red', 'gray', 'green']
    
    for label in range(3):
        label_data = results_df[results_df['true_label'] == label]['confidence']
        if len(label_data) > 0:
            axes[1].hist(label_data, bins=8, alpha=0.6, 
                        label=labels[label], color=colors[label])
    
    axes[1].set_xlabel('置信度')
    axes[1].set_ylabel('频次')
    axes[1].set_title('各情感类别的预测置信度分布')
    axes[1].legend()
    axes[1].grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.show()
    
    # 输出统计信息
    print("\n置信度统计信息:")
    print("=" * 40)
    print(f"总体平均置信度: {results_df['confidence'].mean():.3f}")
    if len(correct_conf) > 0:
        print(f"正确预测平均置信度: {correct_conf.mean():.3f}")
    if len(incorrect_conf) > 0:
        print(f"错误预测平均置信度: {incorrect_conf.mean():.3f}")
    
    # 分析置信度阈值的影响
    print("\n不同置信度阈值下的表现:")
    print("置信度阈值 | 覆盖率 | 准确率")
    print("-" * 35)
    for threshold in [0.5, 0.6, 0.7, 0.8, 0.9]:
        high_conf_data = results_df[results_df['confidence'] >= threshold]
        if len(high_conf_data) > 0:
            coverage = len(high_conf_data) / len(results_df)
            accuracy = high_conf_data['correct'].mean()
            print(f"    {threshold}    | {coverage:.3f}  | {accuracy:.3f}")

# 执行置信度分析
analyze_confidence_distribution(results_df)

def comprehensive_performance_analysis(results_df):
    """综合性能分析可视化"""
    print("正在进行综合性能分析...")
    
    fig, axes = plt.subplots(2, 2, figsize=(15, 12))
    
    # 1. 混淆矩阵
    true_labels = results_df['true_label'].values
    predicted_labels = results_df['predicted_label'].values
    
    cm = confusion_matrix(true_labels, predicted_labels)
    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', 
                xticklabels=['负面', '中性', '正面'],
                yticklabels=['负面', '中性', '正面'], ax=axes[0,0])
    axes[0,0].set_title('混淆矩阵', fontsize=14, fontweight='bold')
    axes[0,0].set_xlabel('预测标签', fontsize=12)
    axes[0,0].set_ylabel('真实标签', fontsize=12)
    
    # 2. 真实分布 vs 预测分布
    true_counts = pd.Series(true_labels).value_counts().sort_index()
    pred_counts = pd.Series(predicted_labels).value_counts().sort_index()
    
    x = np.arange(len(true_counts))
    width = 0.35
    
    axes[0,1].bar(x - width/2, true_counts, width, label='真实分布', 
                 alpha=0.7, color='blue')
    axes[0,1].bar(x[:len(pred_counts)] + width/2, pred_counts, width, label='预测分布', 
                 alpha=0.7, color='orange')
    axes[0,1].set_xlabel('情感类别', fontsize=12)
    axes[0,1].set_ylabel('样本数量', fontsize=12)
    axes[0,1].set_title('真实分布 vs 预测分布', fontsize=14, fontweight='bold')
    axes[0,1].set_xticks(x)
    axes[0,1].set_xticklabels(['负面', '中性', '正面'])
    axes[0,1].legend()
    axes[0,1].grid(True, alpha=0.3)
    
    # 3. 各类别准确率
    class_accuracy = []
    labels = ['负面', '中性', '正面']
    
    for i in range(3):
        mask = np.array(true_labels) == i
        if mask.sum() > 0:
            acc = (np.array(predicted_labels)[mask] == i).mean()
            class_accuracy.append(acc)
        else:
            class_accuracy.append(0)
    
    bars = axes[1,0].bar(labels, class_accuracy, 
                        color=['red', 'gray', 'green'], alpha=0.7)
    axes[1,0].set_ylabel('准确率', fontsize=12)
    axes[1,0].set_title('各情感类别准确率', fontsize=14, fontweight='bold')
    axes[1,0].set_ylim(0, 1)
    axes[1,0].grid(True, alpha=0.3)
    
    # 在柱状图上添加数值标签
    for i, bar in enumerate(bars):
        height = bar.get_height()
        axes[1,0].text(bar.get_x() + bar.get_width()/2., height + 0.01,
                      f'{height:.3f}', ha='center', va='bottom', fontweight='bold')
    
    # 4. 置信度箱型图
    confidence_data = []
    for i in range(3):
        mask = np.array(true_labels) == i
        confidence_data.append(results_df[mask]['confidence'].values)
    
    box_plot = axes[1,1].boxplot(confidence_data, labels=labels, patch_artist=True)
    # 设置箱型图颜色
    colors = ['lightcoral', 'lightgray', 'lightgreen']
    for patch, color in zip(box_plot['boxes'], colors):
        patch.set_facecolor(color)
    
    axes[1,1].set_ylabel('预测置信度', fontsize=12)
    axes[1,1].set_title('各情感类别预测置信度分布', fontsize=14, fontweight='bold')
    axes[1,1].grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.show()
    
    # 输出详细的分类报告
    print("\n详细分类报告:")
    print("=" * 50)
    print(classification_report(true_labels, predicted_labels, 
                              target_names=['负面', '中性', '正面']))

# 执行综合性能分析
comprehensive_performance_analysis(results_df)

def error_pattern_analysis(results_df):
    """错误模式趋势分析"""
    print("正在进行错误模式分析...")
    
    # 筛选错误预测
    errors_df = results_df[~results_df['correct']].copy()
    
    if len(errors_df) == 0:
        print(" 没有发现错误预测！模型表现完美。")
        return
    
    print(f"发现 {len(errors_df)} 个错误预测，占总预测的 {len(errors_df)/len(results_df)*100:.1f}%")
    
    # 创建错误类型标签
    errors_df['error_type'] = errors_df.apply(
        lambda x: f"{['负面','中性','正面'][x['true_label']]}→{['负面','中性','正面'][x['predicted_label']]}", 
        axis=1
    )
    
    fig, axes = plt.subplots(1, 3, figsize=(18, 5))
    
    # 1. 错误类型分布饼图
    error_counts = errors_df['error_type'].value_counts()
    axes[0].pie(error_counts.values, labels=error_counts.index, autopct='%1.1f%%',
                startangle=90, colors=plt.cm.Set3(np.linspace(0, 1, len(error_counts))))
    axes[0].set_title('错误类型分布', fontsize=14, fontweight='bold')
    
    # 2. 错误预测的置信度分布
    axes[1].hist(errors_df['confidence'], bins=10, alpha=0.7, color='red', edgecolor='black')
    axes[1].set_xlabel('置信度', fontsize=12)
    axes[1].set_ylabel('频次', fontsize=12)
    axes[1].set_title('错误预测的置信度分布', fontsize=14, fontweight='bold')
    axes[1].grid(True, alpha=0.3)
    
    # 添加统计信息
    avg_conf = errors_df['confidence'].mean()
    axes[1].axvline(avg_conf, color='darkred', linestyle='--', 
                   label=f'平均置信度: {avg_conf:.3f}')
    axes[1].legend()
    
    # 3. 文本长度与错误率的关系
    results_df['text_length'] = results_df['text'].apply(len)
    results_df['is_error'] = ~results_df['correct']
    
    # 按文本长度分组计算错误率
    length_bins = pd.cut(results_df['text_length'], bins=5)
    error_rate_by_length = results_df.groupby(length_bins)['is_error'].mean()
    
    # 绘制错误率趋势线
    x_positions = range(len(error_rate_by_length))
    axes[2].plot(x_positions, error_rate_by_length.values, 
                marker='o', linewidth=2, markersize=8, color='purple')
    axes[2].set_xlabel('文本长度分组', fontsize=12)
    axes[2].set_ylabel('错误率', fontsize=12)
    axes[2].set_title('文本长度与错误率关系', fontsize=14, fontweight='bold')
    axes[2].set_xticks(x_positions)
    axes[2].set_xticklabels([str(x) for x in error_rate_by_length.index], rotation=45)
    axes[2].grid(True, alpha=0.3)
    
    # 添加趋势线
    z = np.polyfit(x_positions, error_rate_by_length.values, 1)
    p = np.poly1d(z)
    axes[2].plot(x_positions, p(x_positions), "r--", alpha=0.7, label='趋势线')
    axes[2].legend()
    
    plt.tight_layout()
    plt.show()
    
    # 显示最常见的错误类型
    print("\n 最常见的错误类型:")
    print("-" * 30)
    for error_type, count in error_counts.head().items():
        percentage = count / len(errors_df) * 100
        print(f"  {error_type}: {count} 次 ({percentage:.1f}%)")
    
    # 分析高置信度错误
    high_conf_errors = errors_df[errors_df['confidence'] > 0.8]
    if len(high_conf_errors) > 0:
        print(f"\n  发现 {len(high_conf_errors)} 个高置信度错误（置信度 > 0.8）")
        print("这些错误可能揭示了模型的系统性偏见:")
        for _, error in high_conf_errors.head(3).iterrows():
            print(f"  文本: {error['text']}")
            print(f"  错误: {['负面','中性','正面'][error['true_label']]} → {['负面','中性','正面'][error['predicted_label']]}")
            print(f"  置信度: {error['confidence']:.3f}\n")

# 执行错误模式分析
error_pattern_analysis(results_df)

def systematic_error_analysis(results_df):
    """系统性错误分析"""
    print("正在进行系统性错误分析...")
    print("=" * 60)
    
    errors_df = results_df[~results_df['correct']]
    
    if len(errors_df) == 0:
        print("模型表现完美，没有发现系统性错误。")
        return
    
    print("系统性错误分析报告")
    print("=" * 60)
    
    # 1. 高置信度错误分析
    high_conf_errors = errors_df[errors_df['confidence'] > 0.8]
    print(f"\n1. 高置信度错误分析（置信度 > 0.8）")
    print(f"   发现 {len(high_conf_errors)} 个高置信度错误，占所有错误的 {len(high_conf_errors)/len(errors_df)*100:.1f}%")
    
    if len(high_conf_errors) > 0:
        print("   高置信度错误样本示例:")
        for i, (_, error) in enumerate(high_conf_errors.head(3).iterrows()):
            true_sentiment = ['负面','中性','正面'][error['true_label']]
            pred_sentiment = ['负面','中性','正面'][error['predicted_label']]
            print(f"     {i+1}. 文本: {error['text'][:30]}...")
            print(f"        真实: {true_sentiment}, 预测: {pred_sentiment}, 置信度: {error['confidence']:.3f}")
    
    # 2. 特定错误模式分析
    print(f"\n2. 特定错误模式分析")
    error_patterns = errors_df.groupby(['true_label', 'predicted_label']).size()
    
    print("   错误模式分布:")
    for (true, pred), count in error_patterns.items():
        true_name = ['负面','中性','正面'][true]
        pred_name = ['负面','中性','正面'][pred]
        percentage = count / len(errors_df) * 100
        print(f"     {true_name} → {pred_name}: {count} 次 ({percentage:.1f}%)")
    
    # 3. 中性情感识别挑战分析
    neutral_errors = errors_df[errors_df['true_label'] == 1]
    if len(neutral_errors) > 0:
        print(f"\n3. 中性情感识别挑战")
        print(f"   中性情感错误数: {len(neutral_errors)}")
        total_neutral = len(results_df[results_df['true_label'] == 1])
        if total_neutral > 0:
            neutral_accuracy = (total_neutral - len(neutral_errors)) / total_neutral
            print(f"   中性情感准确率: {neutral_accuracy:.3f}")
        
        # 分析中性情感被误判的模式
        neutral_error_patterns = neutral_errors['predicted_label'].value_counts()
        print("   中性情感被误判为:")
        for pred, count in neutral_error_patterns.items():
            pred_name = ['负面','中性','正面'][pred]
            print(f"     {pred_name}: {count} 次")
    
    # 4. 文本特征分析
    print(f"\n4. 文本特征与错误关系分析")
    
    # 定义情感词词典
    positive_words = ['好', '棒', '优秀', '满意', '喜欢', '推荐', '不错', '精彩']
    negative_words = ['差', '糟糕', '垃圾', '讨厌', '失望', '差评', '不好', '问题']
    
    def detect_sentiment_words(text):
        """检测文本中的情感词"""
        pos_count = sum(1 for word in positive_words if word in text)
        neg_count = sum(1 for word in negative_words if word in text)
        
        if pos_count > neg_count:
            return 'positive'
        elif neg_count > pos_count:
            return 'negative'
        else:
            return 'neutral'
    
    # 分析情感词与预测一致性
    results_df['sentiment_words'] = results_df['text'].apply(detect_sentiment_words)
    sentiment_analysis = results_df.groupby('sentiment_words').agg({
        'correct': 'mean',
        'confidence': 'mean'
    }).round(3)
    
    print("   情感词与预测准确性关系:")
    for sentiment, row in sentiment_analysis.iterrows():
        accuracy = row['correct']
        confidence = row['confidence']
        print(f"     {sentiment}: 准确率 {accuracy}, 平均置信度 {confidence}")

# 执行系统性错误分析
systematic_error_analysis(results_df)