AI red teamer （人工智能红队）系列29 – 人工智能信息安全应用 – 从零构建垃圾邮件过滤模型

垃圾邮件分类概述

垃圾邮件，即未经允许的批量信息（例如赌场广告），自数字通信发展初期以来就一直是个老大难问题。它扰乱收件箱，带来安全风险，还可能被用于网络钓鱼攻击等恶意目的。有效的垃圾邮件检测对于维护电子邮件系统和其他信息平台的完整性和可用性至关重要。

在这个教程中，我们将学习如何使用贝叶斯定理和机器学习技术构建一个有效的垃圾邮件检测模型。我们会从理论基础开始，逐步深入到实际的代码实现。

贝叶斯定理在垃圾邮件检测中的应用

贝叶斯定理基础

贝叶斯定理是概率论中的一个基本概念，它描述了基于对可能与事件相关的条件的先验知识而得出的事件概率。它的数学表达式为：

P(A|B) = (P(B|A) * P(A)) / P(B)

其中：

• P(A|B) 是事件A发生的概率，前提是B为真（后验概率）
• P(B|A) 是指当A为真时，事件B发生的概率（似然度）
• P(A) 是事件A的先验概率
• P(B) 是事件B的先验概率（边际概率）

将贝叶斯定理应用于垃圾邮件检测

在垃圾邮件检测中，我们可以这样理解各个概率：

1. 假设：我们希望根据电子邮件的特征确定其为垃圾邮件的概率
   • P(Spam|Features)：根据邮件特征判断其为垃圾邮件的概率
2. 似然度：这是在邮件是垃圾邮件的情况下观察到特征的概率
   • P(Features|Spam)：垃圾邮件中出现特征的概率
3. 先验概率：任何电子邮件都是垃圾邮件的概率，与其特征无关
   • P(Spam)：电子邮件成为垃圾邮件的先验概率
4. 边际概率：考虑到垃圾邮件和非垃圾邮件，观察到特征的总概率
   • P(Features)：邮件中出现特征的概率

利用贝叶斯定理，我们可以将其表示为：

P(Spam|Features) = (P(Features|Spam) * P(Spam)) / P(Features)

朴素贝叶斯的简化假设

朴素贝叶斯（Naive Bayes）的"朴素"假设是，在给定类别标签的情况下，电子邮件中某一特定特征的存在与任何其他特征的存在无关。这个假设大大简化了计算：

对于垃圾邮件：

P(Features|Spam) = P(feature1|Spam) * P(feature2|Spam) * ... * P(featureN|Spam)

对于非垃圾邮件：

P(Features|Not Spam) = P(feature1|Not Spam) * P(feature2|Not Spam) * ... * P(featureN|Not Spam)

利用这些概率，我们可以计算出根据邮件特征判断其为垃圾邮件或非垃圾邮件的后验概率，然后选择后验概率较高的类别作为预测类别。

计算示例

让我们通过一个具体例子来理解这个过程。假设我们有一封具有F1和F2特征的电子邮件，我们想确定这封邮件是否是垃圾邮件。

给定条件：

• P(Spam) = 0.3：任何电子邮件是垃圾邮件的先验概率
• P(Not Spam) = 0.7：任何电子邮件不是垃圾邮件的先验概率
• P(F1|Spam) = 0.4：假设邮件为垃圾邮件，特征F1的概率
• P(F2|Spam) = 0.5：假设邮件为垃圾邮件，特征F2的概率
• P(F1|Not Spam) = 0.2：假设邮件不是垃圾邮件，特征F1的概率
• P(F2|Not Spam) = 0.3：假设邮件不是垃圾邮件，特征F2的概率

使用朴素贝叶斯假设：

P(F1, F2|Spam) = P(F1|Spam) * P(F2|Spam) = 0.4 * 0.5 = 0.2
P(F1, F2|Not Spam) = P(F1|Not Spam) * P(F2|Not Spam) = 0.2 * 0.3 = 0.06

为了求出P(F1, F2)，我们使用全概率定律：

P(F1, F2) = P(F1, F2|Spam) * P(Spam) + P(F1, F2|Not Spam) * P(Not Spam)
          = (0.2 * 0.3) + (0.06 * 0.7)
          = 0.06 + 0.042
          = 0.102

因此：

P(Spam|F1, F2) = (0.2 * 0.3) / 0.102 = 0.06 / 0.102 ≈ 0.588

P(Not Spam|F1, F2) = (0.06 * 0.7) / 0.102 = 0.042 / 0.102 ≈ 0.412

由于P(Spam|F1, F2) > P(Not Spam|F1, F2)，该邮件被归类为垃圾邮件。

数据集介绍与获取

我们将使用SMS垃圾邮件收集数据集来探索贝叶斯垃圾邮件分类。该数据集是为开发和评估基于文本的垃圾邮件过滤器而量身定制的资源，由巴西坎皮纳斯大学和西班牙Optenet的研究人员共同完成。

该数据集包含5574条文本信息，这些信息被标注为ham（合法）或spam（垃圾邮件），是构建和测试垃圾邮件检测模型的绝佳资源。

下载数据集

import requests
import zipfile
import io
import os
import pandas as pd

# 下载数据集
url = "https://archive.ics.uci.edu/static/public/228/sms+spam+collection.zip"
response = requests.get(url)

if response.status_code == 200:
    print("下载成功")

    # 解压数据集
    with zipfile.ZipFile(io.BytesIO(response.content)) as z:
        z.extractall("sms_spam_collection")
        print("解压成功")

    # 列出解压后的文件
    extracted_files = os.listdir("sms_spam_collection")
    print("解压的文件:", extracted_files)
else:
    print("下载失败")

加载数据集

# 加载数据集
df = pd.read_csv(
    "sms_spam_collection/SMSSpamCollection",
    sep="\t",  # 制表符分隔
    header=None,  # 没有标题行
    names=["label", "message"]  # 手动指定列名
)

# 查看数据集基本信息
print("=== 数据集概览 ===")
print(df.head())
print("\n=== 数据集描述 ===")
print(df.describe())
print("\n=== 数据集信息 ===")
print(df.info())

# 检查缺失值和重复值
print("\n=== 缺失值检查 ===")
print(df.isnull().sum())
print("\n=== 重复值检查 ===")
print("重复条目数量:", df.duplicated().sum())

# 删除重复值
df = df.drop_duplicates()
print("删除重复值后的数据集大小:", df.shape)

数据预处理

数据预处理是机器学习管道中的关键步骤。对于文本数据，我们需要将原始文本转换为机器学习算法能够理解的格式。让我们逐步了解每个预处理步骤的目的和实现。

准备必要的工具

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

# 下载必要的NLTK数据文件
nltk.download("punkt")
nltk.download("punkt_tab")
nltk.download("stopwords")

print("=== 预处理前的数据示例 ===") 
print(df.head(5))

文本小写化

将文本转换为小写可以确保分类器对单词一视同仁，无论其原始大小写如何。这样做可以减少特征维度并提高一致性。

# 将所有消息文本转换为小写
df["message"] = df["message"].str.lower()
print("\n=== 小写化后 ===")
print(df["message"].head(5))

删除标点符号和数字

我们需要仔细平衡清理数据和保留重要信息。某些符号如$和!在垃圾邮件检测中可能包含重要上下文。

# 删除非必要的标点和数字，但保留有用的符号如$和!
df["message"] = df["message"].apply(lambda x: re.sub(r"[^a-z\s$!]", "", x))
print("\n=== 删除标点和数字后（保留$和!）===")
print(df["message"].head(5))

文本分词

分词将消息文本划分为单个单词或标记，这是进一步分析的基础。

# 将每条消息分割为单个标记
df["message"] = df["message"].apply(word_tokenize)
print("\n=== 分词后 ===")
print(df["message"].head(5))

删除停用词

停用词是常见的词语，如"and"、"the"或"is"，通常不会添加有意义的上下文。删除这些词可以减少噪音并将模型的重点放在更有意义的词语上。

# 定义英语停用词集合并从标记中删除它们
stop_words = set(stopwords.words("english"))
df["message"] = df["message"].apply(lambda x: [word for word in x if word not in stop_words])
print("\n=== 删除停用词后 ===")
print(df["message"].head(5))

词干提取

词干提取通过将单词简化为其基本形式来规范化单词。这有助于合并同一词根的不同形式，减少词汇量。

# 对每个标记进行词干提取
stemmer = PorterStemmer()
df["message"] = df["message"].apply(lambda x: [stemmer.stem(word) for word in x])
print("\n=== 词干提取后 ===")
print(df["message"].head(5))

重新组合标记

最后，我们将标记重新连接成字符串，以便与后续的特征提取方法兼容。

# 将标记重新连接为单个字符串
df["message"] = df["message"].apply(lambda x: " ".join(x))
print("\n=== 最终预处理结果 ===")
print(df["message"].head(5))

特征提取

特征提取将预处理的文本转换为机器学习算法可以处理的数字向量。我们使用词袋模型（Bag of Words）方法，它将文本表示为词汇表中每个词的出现次数。

理解词袋模型

词袋模型的核心思想是：

1. 构建一个包含数据集中所有唯一词汇的词典
2. 将每个文档表示为词汇表中每个词的计数向量
3. 向量的每个维度对应词汇表中的一个词

为了捕获一些局部序列信息，我们还包含了bigrams（连续的词对）。

实现特征提取

from sklearn.feature_extraction.text import CountVectorizer

# 初始化CountVectorizer，包含unigrams和bigrams
vectorizer = CountVectorizer(
    min_df=1,           # 词必须至少出现在1个文档中
    max_df=0.9,         # 删除出现在90%以上文档中的词
    ngram_range=(1, 2)  # 包含unigrams和bigrams
)

# 拟合并转换消息列
X = vectorizer.fit_transform(df["message"])

# 转换标签为数字格式（1代表垃圾邮件，0代表正常邮件）
y = df["label"].apply(lambda x: 1 if x == "spam" else 0)

print(f"特征矩阵形状: {X.shape}")
print(f"词汇表大小: {len(vectorizer.vocabulary_)}")
print(f"垃圾邮件数量: {sum(y)}")
print(f"正常邮件数量: {len(y) - sum(y)}")

特征提取示例说明

让我们通过一个简单的例子来理解CountVectorizer的工作原理。假设我们有以下文档：

1. "free prize waiting"
2. "spam message offers free prize"
3. "important news today"

使用ngram_range=(1,2)，我们会得到：

• Unigrams: "free", "prize", "waiting", "spam", "message", "offers", "important", "news", "today"
• Bigrams: "free prize", "prize waiting", "spam message", "message offers", "offers free", "important news"

每个文档然后被表示为这些特征的计数向量。

模型训练与评估

现在我们将训练一个多项式朴素贝叶斯分类器来检测垃圾邮件。我们使用管道（Pipeline）来简化工作流程，并通过网格搜索来找到最佳参数。

构建训练管道

from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
from sklearn.metrics import classification_report, confusion_matrix
import numpy as np

# 构建包含向量化和分类的管道
pipeline = Pipeline([
    ("vectorizer", CountVectorizer(min_df=1, max_df=0.9, ngram_range=(1, 2))),
    ("classifier", MultinomialNB())
])

# 定义超参数网格
param_grid = {
    "classifier__alpha": [0.01, 0.1, 0.15, 0.2, 0.25, 0.5, 0.75, 1.0]
}

# 使用5折交叉验证和F1分数进行网格搜索
grid_search = GridSearchCV(
    pipeline,
    param_grid,
    cv=5,
    scoring="f1",
    n_jobs=-1  # 使用所有可用的CPU核心
)

# 在完整数据集上进行网格搜索
print("开始训练模型...")
grid_search.fit(df["message"], y)

# 获取最佳模型
best_model = grid_search.best_estimator_
print(f"最佳参数: {grid_search.best_params_}")
print(f"最佳F1分数: {grid_search.best_score_:.4f}")

模型评估

# 分割数据集用于最终评估
X_train, X_test, y_train, y_test = train_test_split(
    df["message"], y, test_size=0.2, random_state=42, stratify=y
)

# 使用最佳模型进行预测
best_model.fit(X_train, y_train)
y_pred = best_model.predict(X_test)

# 打印分类报告
print("\n=== 分类报告 ===")
print(classification_report(y_test, y_pred, target_names=['正常邮件', '垃圾邮件']))

# 打印混淆矩阵
print("\n=== 混淆矩阵 ===")
cm = confusion_matrix(y_test, y_pred)
print(cm)

# 计算准确率
accuracy = np.sum(y_pred == y_test) / len(y_test)
print(f"\n模型准确率: {accuracy:.4f}")

实际应用示例

让我们用训练好的模型来评估一些新的短信，看看它在实际情况下的表现。

准备测试消息

# 测试消息样例
test_messages = [
    "Congratulations! You've won a $1000 Walmart gift card. Go to http://bit.ly/1234 to claim now.",
    "Hey, are we still meeting up for lunch today?",
    "Urgent! Your account has been compromised. Verify your details here: www.fakebank.com/verify",
    "Reminder: Your appointment is scheduled for tomorrow at 10am.",
    "FREE entry in a weekly competition to win an iPad. Just text WIN to 80085 now!",
    "Can you pick up some milk on your way home?",
    "WINNER! You have been selected to receive a cash prize of $500. Call now!"
]

# 使用最佳模型进行预测
predictions = best_model.predict(test_messages)
prediction_probabilities = best_model.predict_proba(test_messages)

# 显示预测结果
print("=== 预测结果 ===")
for i, msg in enumerate(test_messages):
    prediction = "垃圾邮件" if predictions[i] == 1 else "正常邮件"
    spam_prob = prediction_probabilities[i][1]  # 垃圾邮件概率
    ham_prob = prediction_probabilities[i][0]   # 正常邮件概率

    print(f"\n消息: {msg}")
    print(f"预测: {prediction}")
    print(f"垃圾邮件概率: {spam_prob:.3f}")
    print(f"正常邮件概率: {ham_prob:.3f}")
    print("-" * 80)

模型保存与加载

为了在生产环境中使用模型，我们需要保存训练好的模型。

import joblib

# 保存训练好的模型
model_filename = 'spam_detection_model.joblib'
joblib.dump(best_model, model_filename)
print(f"模型已保存到 {model_filename}")

# 加载模型示例
def load_and_predict(model_path, messages):
    """
    加载保存的模型并进行预测
    """
    loaded_model = joblib.load(model_path)
    predictions = loaded_model.predict(messages)
    probabilities = loaded_model.predict_proba(messages)
    return predictions, probabilities

# 使用示例
# predictions, probabilities = load_and_predict(model_filename, test_messages)

上传模型，获取 flag

模型性能分析

特征重要性分析

我们可以查看哪些词汇对垃圾邮件分类最有影响：

# 获取特征名称和对应的权重
feature_names = best_model.named_steps['vectorizer'].get_feature_names_out()
classifier = best_model.named_steps['classifier']

# 获取每个类别的log概率
log_probs = classifier.feature_log_prob_
spam_log_probs = log_probs[1]  # 垃圾邮件类别
ham_log_probs = log_probs[0]   # 正常邮件类别

# 计算特征的重要性分数（垃圾邮件概率 - 正常邮件概率）
feature_importance = spam_log_probs - ham_log_probs

# 获取最重要的垃圾邮件特征
top_spam_indices = np.argsort(feature_importance)[-20:]
print("=== 最重要的垃圾邮件特征 ===")
for idx in reversed(top_spam_indices):
    print(f"{feature_names[idx]}: {feature_importance[idx]:.3f}")

# 获取最重要的正常邮件特征  
top_ham_indices = np.argsort(feature_importance)[:20]
print("\n=== 最重要的正常邮件特征 ===")
for idx in top_ham_indices:
    print(f"{feature_names[idx]}: {feature_importance[idx]:.3f}")

总结与优化建议

通过这个完整的垃圾邮件检测项目，我们学习了：

1. 理论基础：理解了贝叶斯定理在文本分类中的应用
2. 数据处理：掌握了文本预处理的各个步骤及其重要性
3. 特征工程：学会了使用词袋模型将文本转换为数字特征
4. 模型训练：使用朴素贝叶斯分类器和交叉验证进行模型优化
5. 模型评估：通过多种指标评估模型性能

进一步优化的方向

1. 高级特征工程：可以考虑使用TF-IDF、Word2Vec或更现代的词嵌入技术
2. 模型复杂度：尝试其他算法如支持向量机、随机森林或深度学习模型
3. 特征选择：使用统计方法选择最有效的特征
4. 数据增强：收集更多样化的数据来提高模型的泛化能力
5. 实时处理：优化模型以支持实时垃圾邮件检测

这个项目展示了机器学习在信息安全领域的实际应用，为理解更复杂的AI安全系统打下了基础。