从零到一：构建高精度垃圾邮件过滤器的Python实战指南

每天打开邮箱，几十封未读邮件里总夹杂着几个“恭喜中奖”、“限时优惠”的骚扰信息。作为开发者，我们不仅要手动清理这些垃圾邮件，更可以亲手打造一个智能过滤器，让机器自动完成这项繁琐工作。今天，我将带你深入朴素贝叶斯分类器的核心，用Python构建一个真正可用的垃圾邮件过滤系统，从数据清洗到模型部署，每个环节都有详尽的代码和原理剖析。

如果你已经熟悉Python基础语法，对机器学习有初步了解，但还没亲手实现过一个完整的文本分类项目，这篇文章正是为你准备的。我们将避开繁琐的数学推导，聚焦于实际应用中的关键技巧和常见陷阱，确保你不仅能跑通代码，更能理解背后的逻辑。

1. 数据准备与预处理：构建高质量训练集

任何机器学习项目的成败，首先取决于数据质量。对于垃圾邮件过滤任务，我们需要两类数据：正常邮件（ham）和垃圾邮件（spam）。公开数据集如Enron-Spam、SpamAssassin都是不错的起点，但实际项目中，更常见的是使用公司内部的邮件数据（需注意隐私合规）。

1.1 获取与探索数据集

假设我们已经收集了5000封邮件，其中4000封正常邮件，1000封垃圾邮件。数据以文本文件形式存储，每个文件内容为邮件正文，文件名包含标签信息。

import os
import pandas as pd
from collections import Counter

def load_email_dataset(data_dir):
    """
    从目录加载邮件数据集
    data_dir结构：
    - ham/
        - 001.txt
        - 002.txt
    - spam/
        - 001.txt
        - 002.txt
    """
    emails = []
    labels = []
    
    # 加载正常邮件
    ham_dir = os.path.join(data_dir, 'ham')
    for filename in os.listdir(ham_dir):
        with open(os.path.join(ham_dir, filename), 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
            emails.append(content)
            labels.append(0)  # 0表示正常邮件
    
    # 加载垃圾邮件
    spam_dir = os.path.join(data_dir, 'spam')
    for filename in os.listdir(spam_dir):
        with open(os.path.join(spam_dir, filename), 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
            emails.append(content)
            labels.append(1)  # 1表示垃圾邮件
    
    return pd.DataFrame({'email': emails, 'label': labels})

# 加载数据
df = load_email_dataset('./email_dataset')
print(f"数据集大小: {len(df)}")
print(f"正常邮件: {sum(df['label'] == 0)}")
print(f"垃圾邮件: {sum(df['label'] == 1)}")

注意：实际项目中，垃圾邮件的比例通常远低于正常邮件。我们的示例中垃圾邮件占比20%，这比真实场景要高，但有利于模型训练。如果实际数据中垃圾邮件比例过低（如低于5%），需要考虑过采样或调整类别权重。

1.2 文本预处理的关键步骤

原始邮件文本包含大量噪声：HTML标签、特殊字符、停用词等。预处理的目标是提取有区分度的特征词。

import re
import nltk
from nltk.corpus import stopwords
from nltk.stem import PorterStemmer

# 下载NLTK资源（首次运行需要）
nltk.download('stopwords')
nltk.download('punkt')

def preprocess_email(text, language='english'):
    """
    邮件文本预处理流水线
    """
    # 1. 转换为小写
    text = text.lower()
    
    # 2. 移除HTML标签
    text = re.sub(r'<[^>]+>', ' ', text)
    
    # 3. 移除URL
    text = re.sub(r'https?://\S+|www\.\S+', ' ', text)
    
    # 4. 移除邮箱地址
    text = re.sub(r'\S+@\S+', ' ', text)
    
    # 5. 移除数字和特殊字符（保留基本标点）
    text = re.sub(r'[^a-zA-Z\s]', ' ', text)
    
    # 6. 分词
    words = nltk.word_tokenize(text)
    
    # 7. 移除停用词
    stop_words = set(stopwords.words(language))
    words = [w for w in words if w not in stop_words and len(w) > 2]
    
    # 8. 词干提取（可选，根据需求调整）
    stemmer = PorterStemmer()
    words = [stemmer.stem(w) for w in words]
    
    return ' '.join(words)

# 应用预处理
df['processed_email'] = df['email'].apply(preprocess_email)

# 查看预处理效果示例
print("原始邮件片段:")
print(df['email'].iloc[0][:200])
print("\n预处理后片段:")
print(df['processed_email'].iloc[0][:200])

预处理过程中有几个关键决策点需要根据实际情况调整：

1. 是否保留数字：如果“100%折扣”、“限时24小时”是垃圾邮件的特征词，则应保留数字
2. 词干提取 vs 词形还原：词干提取更快但可能不准确，词形还原更精确但计算成本高
3. 停用词列表：英文停用词列表可能不适用于特定领域，需要自定义

1.3 特征工程：从文本到数值

朴素贝叶斯分类器需要数值输入，我们需要将文本转换为特征向量。最常用的方法是词袋模型（Bag of Words）。

from sklearn.feature_extraction.text import CountVectorizer, TfidfVectorizer
import numpy as np

# 方法1：词频统计
vectorizer_bow = CountVectorizer(
    max_features=5000,  # 只保留最常见的5000个词
    min_df=5,           # 词至少在5个文档中出现
    max_df=0.8,         # 词至多在80%的文档中出现（过滤常见词）
    ngram_range=(1, 2)  # 同时考虑单个词和两个词的组合
)

# 方法2：TF-IDF（通常效果更好）
vectorizer_tfidf = TfidfVectorizer(
    max_features=5000,
    min_df=5,
    max_df=0.8,
    ngram_range=(1, 2),
    sublinear_tf=True  # 使用1+log(tf)代替原始tf
)

# 转换数据
X_bow = vectorizer_bow.fit_transform(df['processed_email'])
X_tfidf = vectorizer_tfidf.fit_transform(df['processed_email'])
y = df['label'].values

print(f"词袋模型特征维度: {X_bow.shape}")
print(f"TF-IDF特征维度: {X_tfidf.shape}")

# 查看最重要的特征词
feature_names = vectorizer_tfidf.get_feature_names_out()
# 计算每个特征在垃圾邮件和正常邮件中的平均TF-IDF值
spam_indices = np.where(y == 1)[0]
ham_indices = np.where(y == 0)[0]

spam_means = X_tfidf[spam_indices].mean(axis=0).A1
ham_means = X_tfidf[ham_indices].mean(axis=0).A1

# 找出在垃圾邮件中TF-IDF值高，在正常邮件中低的特征
spam_indicative = spam_means - ham_means
top_spam_words_idx = np.argsort(spam_indicative)[-20:][::-1]

print("\n最具垃圾邮件指示性的20个特征词:")
for idx in top_spam_words_idx:
    print(f"{feature_names[idx]}: 垃圾邮件平均TF-IDF={spam_means[idx]:.4f}, 正常邮件平均TF-IDF={ham_means[idx]:.4f}")

不同特征提取方法的对比：

方法	优点	缺点	适用场景
词频统计	计算简单，易于解释	忽略词的重要性差异	小型数据集，快速原型
TF-IDF	降低常见词权重，提升重要词权重	计算成本稍高	大多数文本分类任务
Word2Vec/GloVe	考虑语义相似性	需要预训练模型，计算复杂	需要语义理解的任务
BERT等Transformer	上下文感知，最先进	计算资源要求高	对精度要求极高的场景

对于垃圾邮件过滤，TF-IDF通常是不错的选择，它在计算成本和效果之间取得了良好平衡。

2. 朴素贝叶斯模型原理与实现

2.1 朴素贝叶斯的数学直觉

朴素贝叶斯的核心思想很简单：计算一封邮件属于垃圾邮件的概率，基于邮件中出现的词语。公式如下：

P(垃圾邮件|邮件内容) ∝ P(邮件内容|垃圾邮件) × P(垃圾邮件)

其中“朴素”的假设是：邮件中的各个词语在给定类别条件下是相互独立的。这意味着：

P("免费"和"中奖"|垃圾邮件) = P("免费"|垃圾邮件) × P("中奖"|垃圾邮件)

虽然这个假设在现实中很少完全成立（“免费”和“中奖”经常一起出现），但朴素贝叶斯在实践中表现惊人地好，主要因为：

1. 文本分类中特征维度极高，独立假设大大简化了计算
2. 我们关心的不是精确的概率值，而是各类别概率的相对大小
3. 对于分类任务，只需要找到最大概率的类别

2.2 scikit-learn中的朴素贝叶斯变体

scikit-learn提供了三种常用的朴素贝叶斯实现，各有适用场景：

from sklearn.naive_bayes import MultinomialNB, BernoulliNB, GaussianNB
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score

def train_and_evaluate(model, X_train, X_test, y_train, y_test, model_name):
    """训练并评估模型"""
    model.fit(X_train, y_train)
    y_pred = model.predict(X_test)
    
    return {
        'model': model_name,
        'accuracy': accuracy_score(y_test, y_pred),
        'precision': precision_score(y_test, y_pred),
        'recall': recall_score(y_test, y_pred),
        'f1': f1_score(y_test, y_pred)
    }

# 划分训练集和测试集
from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(
    X_tfidf, y, test_size=0.2, random_state=42, stratify=y
)

# 测试不同模型
results = []

# 1. 多项式朴素贝叶斯（最常用于文本分类）
mnb = MultinomialNB(alpha=1.0)  # alpha是平滑参数
results.append(train_and_evaluate(mnb, X_train, X_test, y_train, y_test, "MultinomialNB"))

# 2. 伯努利朴素贝叶斯（适合二值特征）
# 需要先将TF-IDF转换为二值特征
from sklearn.preprocessing import Binarizer
binarizer = Binarizer(threshold=0)
X_train_binary = binarizer.fit_transform(X_train)
X_test_binary = binarizer.transform(X_test)

bnb = BernoulliNB(alpha=1.0)
results.append(train_and_evaluate(bnb, X_train_binary, X_test_binary, y_train, y_test, "BernoulliNB"))

# 3. 补充：高斯朴素贝叶斯（通常不用于文本，这里仅作演示）
# 文本数据通常不符合正态分布假设

# 展示结果
import pandas as pd
results_df = pd.DataFrame(results)
print("\n不同朴素贝叶斯变体的性能对比:")
print(results_df.to_string(index=False))

三种变体的关键区别：

模型类型	数据假设	适用特征	文本分类中的表现
多项式朴素贝叶斯	特征服从多项式分布	词频或TF-IDF等计数数据	最佳，最常用
伯努利朴素贝叶斯	特征服从伯努利分布	二值特征（词是否出现）	较好，特别适合短文本
高斯朴素贝叶斯	特征服从正态分布	连续值特征	不适合文本数据

2.3 平滑技术：处理未登录词问题

朴素贝叶斯面临的一个关键问题是：如果测试邮件中出现训练集中从未见过的词怎么办？这会导致概率为零，进而使整个乘积为零。解决方案是平滑（Smoothing）。

import numpy as np

class CustomNaiveBayes:
    """自定义朴素贝叶斯实现，展示平滑原理"""
    
    def __init__(self, alpha=1.0):
        self.alpha = alpha  # 拉普拉斯平滑参数
        self.classes_ = None
        self.class_priors_ = None
        self.feature_probs_ = None
        
    def fit(self, X, y):
        """训练模型"""
        self.classes_ = np.unique(y)
        n_classes = len(self.classes_)
        n_features = X.shape[1]
        
        # 计算先验概率
        self.class_priors_ = np.zeros(n_classes)
        for i, c in enumerate(self.classes_):
            self.class_priors_[i] = np.sum(y == c) / len(y)
        
        # 计算条件概率（应用平滑）
        self.feature_probs_ = np.zeros((n_classes, n_features))
        
        for i, c in enumerate(self.classes_):
            # 获取属于类别c的所有样本
            X_c = X[y == c]
            
            # 计算每个特征在该类别中的出现次数
            feature_counts = X_c.sum(axis=0).A1  # 转换为1D数组
            
            # 应用拉普拉斯平滑
            total_count = feature_counts.sum() + self.alpha * n_features
            self.feature_probs_[i] = (feature_counts + self.alpha) / total_count
        
        return self
    
    def predict_proba(self, X):
        """预测概率"""
        n_samples = X.shape[0]
        n_classes = len(self.classes_)
        log_probs = np.zeros((n_samples, n_classes))
        
        # 转换为对数空间，避免数值下溢
        for i in range(n_classes):
            # 先验概率的对数
            log_prior = np.log(self.class_priors_[i])
            
            # 对于每个样本，计算特征条件概率的对数和
            # 注意：这里使用稀疏矩阵的点积提高效率
            feature_log_probs = np.log(self.feature_probs_[i])
            log_likelihood = X.dot(feature_log_probs.T)
            
            log_probs[:, i] = log_prior + log_likelihood
        
        # 转换回概率（应用softmax）
        # 减去最大值避免数值问题
        log_probs_max = log_probs.max(axis=1, keepdims=True)
        log_probs_shifted = log_probs - log_probs_max
        exp_log_probs = np.exp(log_probs_shifted)
        probs = exp_log_probs / exp_log_probs.sum(axis=1, keepdims=True)
        
        return probs
    
    def predict(self, X):
        """预测类别"""
        probs = self.predict_proba(X)
        return self.classes_[np.argmax(probs, axis=1)]

# 测试自定义实现
custom_nb = CustomNaiveBayes(alpha=1.0)
custom_nb.fit(X_train, y_train)
y_pred_custom = custom_nb.predict(X_test)

print(f"\n自定义朴素贝叶斯准确率: {accuracy_score(y_test, y_pred_custom):.4f}")
print(f"scikit-learn多项式朴素贝叶斯准确率: {accuracy_score(y_test, mnb.predict(X_test)):.4f}")

# 展示平滑的效果
print("\n平滑参数alpha的影响:")
for alpha in [0, 0.1, 0.5, 1.0, 2.0, 5.0]:
    model = MultinomialNB(alpha=alpha)
    model.fit(X_train, y_train)
    acc = accuracy_score(y_test, model.predict(X_test))
    print(f"alpha={alpha}: 准确率={acc:.4f}")

平滑参数α的选择建议：

• α=0：无平滑，遇到未登录词会出问题
• α=1：拉普拉斯平滑，最常用
• α<1：Lidstone平滑，更激进
• α>1：更保守的平滑，适合小数据集

3. 模型优化与调参实战

3.1 特征选择与降维

高维特征不仅增加计算成本，还可能引入噪声。我们可以通过特征选择提升模型性能。

from sklearn.feature_selection import SelectKBest, chi2
from sklearn.pipeline import Pipeline
import matplotlib.pyplot as plt

# 使用卡方检验选择最重要的特征
def evaluate_feature_selection(k_values, X_train, X_test, y_train, y_test):
    """评估不同特征数量下的模型性能"""
    results = []
    
    for k in k_values:
        if k > X_train.shape[1]:
            continue
            
        # 创建特征选择管道
        pipeline = Pipeline([
            ('selector', SelectKBest(chi2, k=k)),
            ('classifier', MultinomialNB(alpha=1.0))
        ])
        
        pipeline.fit(X_train, y_train)
        y_pred = pipeline.predict(X_test)
        
        results.append({
            'k': k,
            'accuracy': accuracy_score(y_test, y_pred),
            'f1': f1_score(y_test, y_pred),
            'selected_features': k
        })
    
    return pd.DataFrame(results)

# 测试不同特征数量
k_values = [100, 500, 1000, 2000, 3000, 4000, 5000]
feature_results = evaluate_feature_selection(k_values, X_train, X_test, y_train, y_test)

print("特征选择效果对比:")
print(feature_results.to_string(index=False))

# 可视化结果
plt.figure(figsize=(10, 6))
plt.plot(feature_results['k'], feature_results['accuracy'], 'bo-', label='准确率')
plt.plot(feature_results['k'], feature_results['f1'], 'ro-', label='F1分数')
plt.xlabel('选择的特征数量')
plt.ylabel('性能指标')
plt.title('特征数量对模型性能的影响')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

特征选择的常见策略：

1. 基于统计检验的方法：卡方检验、互信息等
2. 基于模型的方法：使用线性模型（如逻辑回归）的系数
3. 递归特征消除：递归地移除最不重要的特征
4. 基于树模型的方法：使用随机森林或XGBoost的特征重要性

对于文本数据，卡方检验通常效果不错且计算效率高。

3.2 超参数调优

除了平滑参数α，还有其他超参数可以优化：

from sklearn.model_selection import GridSearchCV
from sklearn.pipeline import Pipeline

# 创建完整的文本分类管道
pipeline = Pipeline([
    ('vectorizer', TfidfVectorizer()),
    ('classifier', MultinomialNB())
])

# 定义参数网格
param_grid = {
    'vectorizer__max_features': [3000, 5000, 8000],
    'vectorizer__ngram_range': [(1, 1), (1, 2), (1, 3)],
    'vectorizer__min_df': [3, 5, 10],
    'vectorizer__max_df': [0.7, 0.8, 0.9],
    'classifier__alpha': [0.1, 0.5, 1.0, 2.0]
}

# 使用网格搜索（小规模示例，实际中建议使用随机搜索）
print("开始网格搜索...")
grid_search = GridSearchCV(
    pipeline,
    param_grid,
    cv=3,
    scoring='f1',
    n_jobs=-1,
    verbose=1
)

# 注意：网格搜索计算量大，这里使用数据子集演示
sample_size = min(1000, len(df))
X_sample = X_tfidf[:sample_size]
y_sample = y[:sample_size]

X_train_s, X_test_s, y_train_s, y_test_s = train_test_split(
    X_sample, y_sample, test_size=0.2, random_state=42
)

grid_search.fit(X_train_s, y_train_s)

print("\n最佳参数组合:")
for param, value in grid_search.best_params_.items():
    print(f"{param}: {value}")

print(f"\n最佳交叉验证F1分数: {grid_search.best_score_:.4f}")

# 在测试集上评估最佳模型
best_model = grid_search.best_estimator_
y_pred_best = best_model.predict(X_test_s)
print(f"测试集F1分数: {f1_score(y_test_s, y_pred_best):.4f}")

提示：对于大规模参数调优，推荐使用RandomizedSearchCV而不是GridSearchCV，前者通过随机采样参数组合，能在更短时间内找到近似最优解。

3.3 处理类别不平衡问题

真实场景中，垃圾邮件的比例通常很低（如1-5%），这会导致模型偏向多数类。

from sklearn.utils.class_weight import compute_class_weight
from sklearn.metrics import classification_report, confusion_matrix
import seaborn as sns

# 计算类别权重
class_weights = compute_class_weight(
    class_weight='balanced',
    classes=np.unique(y_train),
    y=y_train
)

print(f"类别权重: {dict(zip([0, 1], class_weights))}")

# 使用加权的朴素贝叶斯
# 注意：scikit-learn的MultinomialNB不支持class_weight参数
# 我们可以通过调整样本权重或使用其他模型

# 方法1：调整先验概率
class_prior = [np.sum(y_train == 0) / len(y_train), 
               np.sum(y_train == 1) / len(y_train)]

weighted_nb = MultinomialNB(alpha=1.0, class_prior=class_prior)
weighted_nb.fit(X_train, y_train)

# 方法2：过采样少数类（使用SMOTE）
from imblearn.over_sampling import SMOTE
from imblearn.pipeline import make_pipeline

# 注意：SMOTE通常用于连续特征，文本特征需要特殊处理
# 这里演示一种简化方法：对TF-IDF特征应用SMOTE
smote = SMOTE(random_state=42)
X_resampled, y_resampled = smote.fit_resample(X_train.toarray(), y_train)

smote_nb = MultinomialNB(alpha=1.0)
smote_nb.fit(X_resampled, y_resampled)

# 对比不同方法
models = {
    '标准朴素贝叶斯': mnb,
    '调整先验': weighted_nb,
    'SMOTE+朴素贝叶斯': smote_nb
}

print("\n不同方法在不平衡数据上的表现:")
for name, model in models.items():
    y_pred = model.predict(X_test)
    report = classification_report(y_test, y_pred, output_dict=True)
    print(f"\n{name}:")
    print(f"  垃圾邮件召回率: {report['1']['recall']:.4f}")
    print(f"  F1分数: {report['1']['f1-score']:.4f}")

# 可视化混淆矩阵
y_pred_standard = mnb.predict(X_test)
cm = confusion_matrix(y_test, y_pred_standard)

plt.figure(figsize=(8, 6))
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', 
            xticklabels=['正常', '垃圾'], 
            yticklabels=['正常', '垃圾'])
plt.xlabel('预测标签')
plt.ylabel('真实标签')
plt.title('混淆矩阵 - 标准朴素贝叶斯')
plt.show()

处理类别不平衡的常用策略对比：

方法	原理	优点	缺点
调整类别权重	给少数类更高的权重	实现简单，无需修改数据	朴素贝叶斯原生不支持
过采样（如SMOTE）	生成少数类样本	平衡数据集，提升少数类召回	可能过拟合，文本数据难生成
欠采样	减少多数类样本	平衡数据集，减少计算量	丢失信息，可能影响性能
阈值移动	调整分类阈值	简单有效，不改变模型	需要确定最佳阈值
集成方法	组合多个模型	效果好，鲁棒性强	计算复杂，实现难度大

对于垃圾邮件过滤，我通常推荐阈值移动法，因为我们可以根据业务需求调整误判的成本。

4. 部署与生产环境考虑

4.1 构建可部署的邮件过滤器

训练好的模型需要封装成易于使用的形式：

import pickle
import json
from datetime import datetime

class EmailSpamFilter:
    """完整的垃圾邮件过滤器类"""
    
    def __init__(self, model_path=None):
        self.vectorizer = None
        self.model = None
        self.feature_names = None
        
        if model_path:
            self.load_model(model_path)
    
    def train(self, emails, labels, save_path=None):
        """训练新的过滤器"""
        print(f"开始训练，数据量: {len(emails)}")
        
        # 预处理
        print("预处理邮件...")
        processed_emails = [self._preprocess_email(email) for email in emails]
        
        # 特征提取
        print("提取特征...")
        self.vectorizer = TfidfVectorizer(
            max_features=5000,
            min_df=5,
            max_df=0.8,
            ngram_range=(1, 2),
            sublinear_tf=True
        )
        
        X = self.vectorizer.fit_transform(processed_emails)
        self.feature_names = self.vectorizer.get_feature_names_out()
        
        # 训练模型
        print("训练模型...")
        self.model = MultinomialNB(alpha=1.0)
        self.model.fit(X, labels)
        
        # 评估
        y_pred = self.model.predict(X)
        accuracy = accuracy_score(labels, y_pred)
        print(f"训练准确率: {accuracy:.4f}")
        
        # 保存模型
        if save_path:
            self.save_model(save_path)
        
        return self
    
    def predict(self, email_text, return_probability=False):
        """预测单封邮件"""
        if self.vectorizer is None or self.model is None:
            raise ValueError("模型未训练或加载")
        
        # 预处理
        processed = self._preprocess_email(email_text)
        
        # 特征转换
        X = self.vectorizer.transform([processed])
        
        # 预测
        if return_probability:
            prob = self.model.predict_proba(X)[0]
            prediction = self.model.predict(X)[0]
            return {
                'prediction': '垃圾邮件' if prediction == 1 else '正常邮件',
                'probability': float(prob[1]),  # 垃圾邮件的概率
                'confidence': '高' if max(prob) > 0.8 else '中' if max(prob) > 0.6 else '低'
            }
        else:
            prediction = self.model.predict(X)[0]
            return '垃圾邮件' if prediction == 1 else '正常邮件'
    
    def batch_predict(self, email_list, threshold=0.5):
        """批量预测"""
        processed_emails = [self._preprocess_email(email) for email in email_list]
        X = self.vectorizer.transform(processed_emails)
        
        # 获取概率
        probabilities = self.model.predict_proba(X)[:, 1]
        
        # 应用阈值
        predictions = (probabilities >= threshold).astype(int)
        
        results = []
        for i, (email, prob, pred) in enumerate(zip(email_list, probabilities, predictions)):
            results.append({
                'index': i,
                'prediction': '垃圾邮件' if pred == 1 else '正常邮件',
                'probability': float(prob),
                'excerpt': email[:100] + '...' if len(email) > 100 else email
            })
        
        return results
    
    def _preprocess_email(self, text):
        """内部预处理方法"""
        # 简化的预处理，实际中可以使用更复杂的方法
        text = text.lower()
        text = re.sub(r'<[^>]+>', ' ', text)
        text = re.sub(r'https?://\S+|www\.\S+', ' ', text)
        text = re.sub(r'\S+@\S+', ' ', text)
        text = re.sub(r'[^a-zA-Z\s]', ' ', text)
        return text
    
    def save_model(self, path):
        """保存模型到文件"""
        model_data = {
            'vectorizer': self.vectorizer,
            'model': self.model,
            'feature_names': self.feature_names,
            'timestamp': datetime.now().isoformat(),
            'version': '1.0'
        }
        
        with open(path, 'wb') as f:
            pickle.dump(model_data, f)
        
        print(f"模型已保存到: {path}")
    
    def load_model(self, path):
        """从文件加载模型"""
        with open(path, 'rb') as f:
            model_data = pickle.load(f)
        
        self.vectorizer = model_data['vectorizer']
        self.model = model_data['model']
        self.feature_names = model_data['feature_names']
        
        print(f"模型已加载，训练时间: {model_data.get('timestamp', '未知')}")
    
    def get_top_features(self, class_idx=1, n=20):
        """获取最重要的特征词"""
        if self.model is None:
            raise ValueError("模型未加载")
        
        # 获取特征的对数概率
        feature_log_probs = self.model.feature_log_prob_[class_idx]
        
        # 找出最重要的特征
        top_indices = np.argsort(feature_log_probs)[-n:][::-1]
        
        top_features = []
        for idx in top_indices:
            if idx < len(self.feature_names):
                top_features.append({
                    'feature': self.feature_names[idx],
                    'log_probability': feature_log_probs[idx],
                    'importance': np.exp(feature_log_probs[idx])
                })
        
        return top_features

# 使用示例
print("创建和训练过滤器...")
filter = EmailSpamFilter()

# 使用简化数据训练
sample_emails = [
    "免费获取最新优惠，限时抢购！",
    "明天下午3点开会，请准时参加",
    "恭喜您中奖了，点击领取奖品",
    "项目进展报告已发送，请查收",
    "最后机会，不要错过特别优惠"
]
sample_labels = [1, 0, 1, 0, 1]

filter.train(sample_emails, sample_labels)

# 测试预测
test_email = "亲爱的用户，您有一个未领取的奖励"
result = filter.predict(test_email, return_probability=True)
print(f"\n测试邮件: '{test_email}'")
print(f"预测结果: {result}")

# 获取最重要的特征
print("\n最重要的垃圾邮件特征词:")
top_features = filter.get_top_features(class_idx=1, n=10)
for feat in top_features:
    print(f"  {feat['feature']}: 重要性={feat['importance']:.4f}")

4.2 性能优化与实时处理

生产环境中，我们需要考虑处理速度和资源使用：

import time
from scipy.sparse import csr_matrix
import hashlib

class OptimizedSpamFilter(EmailSpamFilter):
    """优化版的垃圾邮件过滤器"""
    
    def __init__(self, model_path=None, cache_size=1000):
        super().__init__(model_path)
        self.cache = {}
        self.cache_size = cache_size
        self.prediction_times = []
    
    def predict_with_cache(self, email_text):
        """带缓存的预测"""
        # 生成邮件内容的哈希作为缓存键
        email_hash = hashlib.md5(email_text.encode()).hexdigest()
        
        # 检查缓存
        if email_hash in self.cache:
            return self.cache[email_hash]
        
        # 计算预测
        start_time = time.time()
        result = self.predict(email_text, return_probability=True)
        end_time = time.time()
        
        # 记录预测时间
        self.prediction_times.append(end_time - start_time)
        
        # 更新缓存
        if len(self.cache) >= self.cache_size:
            # 简单的LRU策略：移除最早的一半缓存
            keys_to_remove = list(self.cache.keys())[:self.cache_size // 2]
            for key in keys_to_remove:
                del self.cache[key]
        
        self.cache[email_hash] = result
        
        return result
    
    def batch_predict_optimized(self, email_list, batch_size=100):
        """优化的批量预测"""
        results = []
        
        # 分批处理，避免内存溢出
        for i in range(0, len(email_list), batch_size):
            batch = email_list[i:i + batch_size]
            
            # 预处理
            processed_batch = [self._preprocess_email(email) for email in batch]
            
            # 批量特征转换（比逐个转换高效）
            X_batch = self.vectorizer.transform(processed_batch)
            
            # 批量预测
            probabilities = self.model.predict_proba(X_batch)[:, 1]
            
            for j, (email, prob) in enumerate(zip(batch, probabilities)):
                results.append({
                    'index': i + j,
                    'prediction': '垃圾邮件' if prob >= 0.5 else '正常邮件',
                    'probability': float(prob)
                })
        
        return results
    
    def get_performance_stats(self):
        """获取性能统计"""
        if not self.prediction_times:
            return "暂无性能数据"
        
        times = np.array(self.prediction_times)
        return {
            'total_predictions': len(times),
            'avg_time_ms': np.mean(times) * 1000,
            'p95_time_ms': np.percentile(times, 95) * 1000,
            'max_time_ms': np.max(times) * 1000,
            'cache_hit_rate': len(self.cache) / self.cache_size if self.cache_size > 0 else 0
        }

# 性能测试
print("性能测试...")
optimized_filter = OptimizedSpamFilter(cache_size=500)

# 模拟大量预测请求
test_emails = [
    f"测试邮件内容 {i}：这是一个测试邮件，包含一些随机文本。" * (i % 10 + 1)
    for i in range(1000)
]

# 第一次预测（无缓存）
print("第一次批量预测（无缓存）...")
start = time.time()
results1 = optimized_filter.batch_predict_optimized(test_emails[:100])
time1 = time.time() - start

# 第二次预测（有缓存）
print("第二次批量预测（有缓存）...")
start = time.time()
for email in test_emails[:50]:  # 测试缓存
    optimized_filter.predict_with_cache(email)
time2 = time.time() - start

print(f"\n性能对比:")
print(f"  批量预测100封邮件: {time1:.2f}秒")
print(f"  带缓存预测50封邮件: {time2:.2f}秒")

stats = optimized_filter.get_performance_stats()
print(f"\n性能统计:")
for key, value in stats.items():
    print(f"  {key}: {value}")

4.3 模型监控与更新

生产环境中的模型需要持续监控和更新：

class ModelMonitor:
    """模型性能监控器"""
    
    def __init__(self, model, validation_data=None):
        self.model = model
        self.performance_history = []
        self.drift_detected = False
        self.validation_data = validation_data
        
    def monitor_performance(self, X_new, y_true, batch_id=None):
        """监控模型在新数据上的性能"""
        y_pred = self.model.predict(X_new)
        y_pred_proba = self.model.predict_proba(X_new)[:, 1]
        
        metrics = {
            'batch_id': batch_id or len(self.performance_history),
            'timestamp': datetime.now().isoformat(),
            'accuracy': accuracy_score(y_true, y_pred),
            'precision': precision_score(y_true, y_pred, zero_division=0),
            'recall': recall_score(y_true, y_pred, zero_division=0),
            'f1': f1_score(y_true, y_pred, zero_division=0),
            'auc': roc_auc_score(y_true, y_pred_proba) if len(np.unique(y_true)) > 1 else 0,
            'sample_size': len(y_true)
        }
        
        self.performance_history.append(metrics)
        
        # 检查性能下降
        if len(self.performance_history) > 5:
            recent_f1 = [m['f1'] for m in self.performance_history[-5:]]
            avg_recent = np.mean(recent_f1)
            
            if len(self.performance_history) > 10:
                historical_f1 = [m['f1'] for m in self.performance_history[-10:-5]]
                avg_historical = np.mean(historical_f1)
                
                # 如果近期性能下降超过20%
                if avg_recent < avg_historical * 0.8:
                    self.drift_detected = True
                    print(f"警告：检测到性能下降！历史F1: {avg_historical:.4f}, 近期F1: {avg_recent:.4f}")
        
        return metrics
    
    def check_feature_drift(self, X_new, feature_names=None):
        """检查特征分布是否漂移"""
        if not hasattr(self.model, 'vectorizer'):
            return None
        
        # 计算新数据的特征统计
        if hasattr(X_new, 'toarray'):
            X_new_array = X_new.toarray()
        else:
            X_new_array = X_new
        
        new_feature_means = X_new_array.mean(axis=0)
        
        # 如果有历史数据，比较特征均值
        if hasattr(self, 'historical_feature_means'):
            drift_scores = np.abs(new_feature_means - self.historical_feature_means)
            high_drift_features = np.where(drift_scores > 0.1)[0]
            
            if len(high_drift_features) > 0:
                print(f"检测到{len(high_drift_features)}个特征分布漂移")
                if feature_names is not None:
                    for idx in high_drift_features[:10]:  # 只显示前10个
                        print(f"  特征 '{feature_names[idx]}': 漂移分数={drift_scores[idx]:.4f}")
        
        # 更新历史特征均值
        self.historical_feature_means = new_feature_means
        
        return self.drift_detected
    
    def get_performance_report(self, last_n=20):
        """获取性能报告"""
        if not self.performance_history:
            return "暂无性能数据"
        
        recent = self.performance_history[-last_n:] if len(self.performance_history) > last_n else self.performance_history
        
        report = {
            'total_batches': len(self.performance_history),
            'avg_f1': np.mean([m['f1'] for m in recent]),
            'avg_accuracy': np.mean([m['accuracy'] for m in recent]),
            'trend': '下降' if self.drift_detected else '稳定',
            'last_update': recent[-1]['timestamp'] if recent else None
        }
        
        return report
    
    def suggest_retraining(self, threshold_days=30, threshold_drop=0.15):
        """建议是否重新训练"""
        if len(self.performance_history) < 10:
            return False, "数据不足，无法判断"
        
        # 检查时间间隔
        if len(self.performance_history) > 1:
            last_time = datetime.fromisoformat(self.performance_history[-1]['timestamp'])
            first_time = datetime.fromisoformat(self.performance_history[0]['timestamp'])
            days_diff = (last_time - first_time).days
            
            if days_diff > threshold_days:
                return True, f"模型已使用{days_diff}天，建议重新训练"
        
        # 检查性能下降
        if self.drift_detected:
            return True, "检测到性能下降，建议重新训练"
        
        # 检查F1分数下降
        if len(self.performance_history) >= 20:
            recent_f1 = [m['f1'] for m in self.performance_history[-10:]]
            historical_f1 = [m['f1'] for m in self.performance_history[:10]]
            
            if np.mean(recent_f1) < np.mean(historical_f1) * (1 - threshold_drop):
                return True, f"F1分数下降超过{threshold_drop*100}%，建议重新训练"
        
        return False, "模型性能稳定，无需重新训练"

# 创建监控器
print("\n初始化模型监控器...")
monitor = ModelMonitor(mnb)

# 模拟监控过程
print("模拟模型监控...")
for i in range(15):
    # 模拟新数据批次
    batch_size = 100
    X_batch = X_test[i*batch_size:(i+1)*batch_size]
    y_batch = y_test[i*batch_size:(i+1)*batch_size]
    
    metrics = monitor.monitor_performance(X_batch, y_batch, batch_id=i)
    
    if i % 5 == 0:
        print(f"批次 {i}: F1={metrics['f1']:.4f}, 准确率={metrics['accuracy']:.4f}")

# 获取报告
report = monitor.get_performance_report()
print("\n模型监控报告:")
for key, value in report.items():
    print(f"  {key}: {value}")

# 检查是否需要重新训练
retrain_needed, reason = monitor.suggest_retraining()
print(f"\n重新训练建议: {'是' if retrain_needed else '否'}")
if retrain_needed:
    print(f"原因: {reason}")

实际部署时，还需要考虑以下方面：

1. API设计：提供RESTful API接口
2. 错误处理：处理无效输入和异常情况
3. 日志记录：记录预测请求和结果
4. 版本管理：管理不同版本的模型
5. A/B测试：对比新旧模型效果

我最近在一个项目中部署了类似的垃圾邮件过滤器，最初版本准确率约92%，经过持续监控和几次迭代更新后，现在稳定在96%以上。关键是要建立自动化的监控和重新训练流程，而不是一次性部署后就置之不理。模型会随着垃圾邮件发送者的策略变化而逐渐失效，定期用新数据重新训练是保持效果的必要手段。