Python机器学习实战：基于Transformer的智能文本分类系统构建指南

一、Transformer架构核心原理深度解析

1.1 自注意力机制的革命性突破

自注意力机制允许模型在处理每个词时同时关注输入序列中的所有词，克服了传统RNN序列处理的局限性。这种全局依赖关系捕捉能力是Transformer成功的关键。

import torch
import torch.nn as nn
import math

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super().__init__()
        self.d_model = d_model
        self.num_heads = num_heads
        self.d_k = d_model // num_heads
        
        self.w_q = nn.Linear(d_model, d_model)
        self.w_k = nn.Linear(d_model, d_model)
        self.w_v = nn.Linear(d_model, d_model)
        self.w_o = nn.Linear(d_model, d_model)
        
    def scaled_dot_product_attention(self, q, k, v, mask=None):
        attn_scores = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(self.d_k)
        if mask is not None:
            attn_scores = attn_scores.masked_fill(mask == 0, -1e9)
        attn_probs = torch.softmax(attn_scores, dim=-1)
        output = torch.matmul(attn_probs, v)
        return output
        
    def forward(self, q, k, v, mask=None):
        batch_size, seq_len = q.size(0), q.size(1)
        
        # 线性变换并分头
        q = self.w_q(q).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        k = self.w_k(k).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        v = self.w_v(v).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        
        # 计算注意力
        attn_output = self.scaled_dot_product_attention(q, k, v, mask)
        attn_output = attn_output.transpose(1, 2).contiguous().view(
            batch_size, seq_len, self.d_model
        )
        
        return self.w_o(attn_output)

1.2 位置编码：序列顺序的智慧

由于Transformer不包含循环结构，位置编码为模型提供了序列中词语位置的信息。

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super().__init__()
        
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * 
                           (-math.log(10000.0) / d_model))
        
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        pe = pe.unsqueeze(0).transpose(0, 1)
        
        self.register_buffer('pe', pe)
        
    def forward(self, x):
        return x + self.pe[:x.size(0), :]

二、实战：构建智能文本分类系统

2.1 环境配置与数据准备

首先配置必要的库并准备新闻文本分类数据集。

# 安装必要库
# pip install transformers datasets torch sklearn pandas numpy

import pandas as pd
from datasets import Dataset
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from sklearn.model_selection import train_test_split
import numpy as np

class TextDataPreprocessor:
    def __init__(self, model_name="bert-base-uncased", max_length=128):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.max_length = max_length
        
    def load_and_preprocess_data(self, file_path):
        """加载和预处理文本数据"""
        # 模拟新闻数据集 - 实际应用中替换为真实数据
        data = {
            'text': [
                "Technology company announces breakthrough in AI research",
                "Stock market reaches all-time high amid economic recovery",
                "Sports team wins championship after dramatic final match",
                "New scientific discovery could revolutionize medicine",
                "Political leaders meet to discuss climate change policies"
            ],
            'label': [0, 1, 2, 0, 1]  # 0:科技, 1:财经, 2:体育
        }
        
        df = pd.DataFrame(data)
        return df
        
    def tokenize_function(self, examples):
        """tokenize文本数据"""
        return self.tokenizer(
            examples['text'],
            padding='max_length',
            truncation=True,
            max_length=self.max_length,
            return_tensors="pt"
        )
        
    def prepare_dataset(self, df):
        """准备训练数据集"""
        dataset = Dataset.from_pandas(df)
        tokenized_dataset = dataset.map(self.tokenize_function, batched=True)
        return tokenized_dataset

2.2 自定义Transformer分类模型

基于预训练模型构建适合文本分类的Transformer架构。

import torch.nn as nn
from transformers import BertModel

class CustomTextClassifier(nn.Module):
    def __init__(self, model_name="bert-base-uncased", num_classes=3, dropout_rate=0.3):
        super().__init__()
        self.bert = BertModel.from_pretrained(model_name)
        self.dropout = nn.Dropout(dropout_rate)
        self.classifier = nn.Linear(self.bert.config.hidden_size, num_classes)
        self.softmax = nn.Softmax(dim=1)
        
    def forward(self, input_ids, attention_mask, labels=None):
        outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
        pooled_output = outputs.pooler_output
        pooled_output = self.dropout(pooled_output)
        logits = self.classifier(pooled_output)
        probabilities = self.softmax(logits)
        
        loss = None
        if labels is not None:
            loss_fct = nn.CrossEntropyLoss()
            loss = loss_fct(logits, labels)
            
        return {'loss': loss, 'logits': logits, 'probabilities': probabilities}

2.3 高级训练策略与回调函数

实现学习率调度、早停等高级训练技巧。

from transformers import Trainer, TrainingArguments
from sklearn.metrics import accuracy_score, f1_score
import os

def compute_metrics(eval_pred):
    """计算评估指标"""
    predictions, labels = eval_pred
    predictions = np.argmax(predictions, axis=1)
    return {
        'accuracy': accuracy_score(labels, predictions),
        'f1': f1_score(labels, predictions, average='weighted')
    }

class CustomTrainer:
    def __init__(self, model, train_dataset, eval_dataset, output_dir="./results"):
        self.model = model
        self.train_dataset = train_dataset
        self.eval_dataset = eval_dataset
        self.output_dir = output_dir
        
    def setup_training_args(self):
        """配置训练参数"""
        return TrainingArguments(
            output_dir=self.output_dir,
            num_train_epochs=3,
            per_device_train_batch_size=16,
            per_device_eval_batch_size=16,
            warmup_steps=500,
            weight_decay=0.01,
            logging_dir='./logs',
            logging_steps=10,
            evaluation_strategy="steps",
            eval_steps=50,
            save_steps=100,
            load_best_model_at_end=True,
            metric_for_best_model="f1",
            greater_is_better=True,
            learning_rate=2e-5,
            fp16=True  # 混合精度训练
        )
        
    def train(self):
        """执行训练"""
        training_args = self.setup_training_args()
        
        trainer = Trainer(
            model=self.model,
            args=training_args,
            train_dataset=self.train_dataset,
            eval_dataset=self.eval_dataset,
            compute_metrics=compute_metrics,
            tokenizer=preprocessor.tokenizer
        )
        
        # 开始训练
        trainer.train()
        
        # 保存最终模型
        trainer.save_model()
        
        return trainer

三、模型优化与超参数调优

3.1 超参数自动搜索

使用Optuna进行超参数优化，找到最佳模型配置。

import optuna
from optuna.integration import PyTorchLightningPruningCallback

def objective(trial):
    """定义超参数搜索目标函数"""
    # 建议的超参数范围
    learning_rate = trial.suggest_float("learning_rate", 1e-5, 5e-5, log=True)
    batch_size = trial.suggest_categorical("batch_size", [8, 16, 32])
    dropout_rate = trial.suggest_float("dropout_rate", 0.1, 0.5)
    weight_decay = trial.suggest_float("weight_decay", 0.0, 0.1)
    
    # 使用建议的超参数创建模型
    model = CustomTextClassifier(
        num_classes=3,
        dropout_rate=dropout_rate
    )
    
    # 训练和评估模型
    trainer = CustomTrainer(
        model=model,
        train_dataset=train_dataset,
        eval_dataset=eval_dataset
    )
    
    training_args = trainer.setup_training_args()
    training_args.learning_rate = learning_rate
    training_args.per_device_train_batch_size = batch_size
    training_args.weight_decay = weight_decay
    
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=train_dataset,
        eval_dataset=eval_dataset,
        compute_metrics=compute_metrics
    )
    
    # 训练并获取评估结果
    trainer.train()
    eval_result = trainer.evaluate()
    
    return eval_result['eval_f1']

def optimize_hyperparameters():
    """执行超参数优化"""
    study = optuna.create_study(direction="maximize")
    study.optimize(objective, n_trials=20)
    
    print("最佳超参数:")
    for key, value in study.best_trial.params.items():
        print(f"{key}: {value}")
    
    return study.best_params

3.2 知识蒸馏：模型压缩技术

使用知识蒸馏技术将大模型的知识迁移到小模型，提升推理速度。

class KnowledgeDistillationTrainer:
    def __init__(self, teacher_model, student_model, temperature=3.0, alpha=0.7):
        self.teacher_model = teacher_model
        self.student_model = student_model
        self.temperature = temperature
        self.alpha = alpha
        self.kl_loss = nn.KLDivLoss(reduction="batchmean")
        self.ce_loss = nn.CrossEntropyLoss()
        
    def compute_distillation_loss(self, student_logits, teacher_logits, labels):
        """计算蒸馏损失"""
        # 软目标损失
        soft_targets = nn.functional.softmax(teacher_logits / self.temperature, dim=-1)
        soft_prob = nn.functional.log_softmax(student_logits / self.temperature, dim=-1)
        distill_loss = self.kl_loss(soft_prob, soft_targets) * (self.temperature ** 2)
        
        # 硬目标损失
        student_loss = self.ce_loss(student_logits, labels)
        
        # 组合损失
        total_loss = self.alpha * distill_loss + (1 - self.alpha) * student_loss
        return total_loss
        
    def train_step(self, batch):
        """训练步骤"""
        self.teacher_model.eval()
        self.student_model.train()
        
        with torch.no_grad():
            teacher_outputs = self.teacher_model(**batch)
            
        student_outputs = self.student_model(**batch)
        
        loss = self.compute_distillation_loss(
            student_outputs['logits'],
            teacher_outputs['logits'],
            batch['labels']
        )
        
        return loss

四、模型部署与API服务

4.1 使用FastAPI构建推理服务

创建高性能的模型推理API服务。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
import torch

app = FastAPI(title="文本分类API", version="1.0")

class ClassificationRequest(BaseModel):
    text: str
    max_length: int = 128

class ClassificationResponse(BaseModel):
    label: int
    confidence: float
    class_name: str

class TextClassificationService:
    def __init__(self, model_path: str):
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model = CustomTextClassifier.from_pretrained(model_path)
        self.model.to(self.device)
        self.model.eval()
        self.tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
        self.class_names = ["科技", "财经", "体育"]
        
    async def predict(self, text: str, max_length: int = 128):
        """预测文本类别"""
        inputs = self.tokenizer(
            text,
            padding='max_length',
            truncation=True,
            max_length=max_length,
            return_tensors="pt"
        )
        
        inputs = {k: v.to(self.device) for k, v in inputs.items()}
        
        with torch.no_grad():
            outputs = self.model(**inputs)
            probabilities = outputs['probabilities']
            confidence, predicted = torch.max(probabilities, 1)
            
        return {
            'label': predicted.item(),
            'confidence': confidence.item(),
            'class_name': self.class_names[predicted.item()]
        }

# 初始化服务
service = TextClassificationService("./best_model")

@app.post("/classify", response_model=ClassificationResponse)
async def classify_text(request: ClassificationRequest):
    try:
        result = await service.predict(request.text, request.max_length)
        return ClassificationResponse(**result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    return {"status": "healthy", "model_loaded": True}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

4.2 性能监控与日志系统

实现完整的监控和日志记录系统。

import logging
import time
from prometheus_client import Counter, Histogram, generate_latest
import json

# 设置监控指标
REQUEST_COUNT = Counter('classification_requests_total', 'Total classification requests')
REQUEST_DURATION = Histogram('classification_duration_seconds', 'Request duration')
PREDICTION_CONFIDENCE = Histogram('prediction_confidence', 'Prediction confidence distribution')

class MonitoringService:
    def __init__(self):
        self.setup_logging()
        
    def setup_logging(self):
        """配置结构化日志"""
        logging.basicConfig(
            level=logging.INFO,
            format='{"timestamp": "%(asctime)s", "level": "%(levelname)s", "message": "%(message)s"}',
            handlers=[
                logging.FileHandler('classification_service.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
        
    def log_prediction(self, text, prediction, confidence, duration):
        """记录预测日志"""
        log_entry = {
            "text": text[:100] + "..." if len(text) > 100 else text,
            "prediction": prediction,
            "confidence": round(confidence, 4),
            "processing_time": round(duration, 4)
        }
        self.logger.info(json.dumps(log_entry))

# 在API端点中添加监控
@app.post("/classify", response_model=ClassificationResponse)
async def classify_text(request: ClassificationRequest):
    start_time = time.time()
    REQUEST_COUNT.inc()
    
    try:
        result = await service.predict(request.text, request.max_length)
        duration = time.time() - start_time
        
        REQUEST_DURATION.observe(duration)
        PREDICTION_CONFIDENCE.observe(result['confidence'])
        
        # 记录日志
        monitoring.log_prediction(
            request.text, 
            result['class_name'], 
            result['confidence'], 
            duration
        )
        
        return ClassificationResponse(**result)
    except Exception as e:
        duration = time.time() - start_time
        REQUEST_DURATION.observe(duration)
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/metrics")
async def metrics():
    return generate_latest()

五、系统性能评估与对比

5.1 基准测试结果

5.2 实际应用场景测试

# 测试不同领域的文本分类
test_cases = [
    "苹果公司发布新款iPhone，搭载最新A系列芯片",
    "道琼斯指数今日上涨200点，科技股表现强劲",
    "湖人队赢得NBA总冠军，詹姆斯获得MVP",
    "科学家发现新型超导材料，有望改变能源行业",
    "美联储宣布维持利率不变，市场反应积极"
]

async def test_classification():
    for text in test_cases:
        result = await service.predict(text)
        print(f"文本: {text}")
        print(f"分类: {result['class_name']} (置信度: {result['confidence']:.3f})")
        print("-" * 50)

# 运行测试
# asyncio.run(test_classification())