技术方案

语义理解能力

模型稳定性

误报率水平

落地难度

传统词嵌入（fastText+GloVe）

较低

较高

中等

低

微调BERT模型

较高

较低

较高

高

静态BERT嵌入+BiLSTM

较高

较高

较低

中

bash
conda create -n bert-hate-speech python=3.8
conda activate bert-hate-speech
pip install torch==1.13.0+cu117 -f https://download.pytorch.org/whl/torch_stable.html
pip install transformers==4.28.0 sentencepiece scikit-learn==1.2.2 pandas numpy matplotlib

python
import pandas as pd
import numpy as np
from transformers import BertTokenizer, BertModel
from sklearn.model_selection import train_test_split
import nltk
from nltk.corpus import stopwords
import re

# 1. 下载停用词（首次执行需下载）
nltk.download('stopwords')
stop_words = set(stopwords.words('english'))  # ETHOS数据集为英文，适配英文停用词

# 2. 加载ETHOS数据集（自行下载后放入指定路径，数据集可通过公开学术渠道获取）
df = pd.read_csv('ethos_hate_speech_binary.csv')
df.columns = ['text', 'label']  # 重命名列，label=1为仇恨言论，0为非仇恨言论

# 3. 文本清洗：去除特殊字符、数字，转为小写
def clean_text(text):
    text = re.sub(r'[^a-zA-Z\s]', '', text)  # 保留字母和空格
    text = text.lower()  # 转为小写
    text = ' '.join([word for word in text.split() if word not in stop_words])  # 去停用词
    return text

df['cleaned_text'] = df['text'].apply(clean_text)

# 4. 加载预训练BERT模型与分词器（选用bert-base-uncased，轻量易落地）
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
bert_model = BertModel.from_pretrained('bert-base-uncased').cuda()  # GPU加速

# 5. 生成静态BERT嵌入
def generate_static_bert_embedding(text):
    # BERT分词，设置最大长度为64，不足补全，过长截断
    inputs = tokenizer(text, return_tensors='pt', padding='max_length', truncation=True, max_length=64).to('cuda')
    with torch.no_grad():  # 禁用梯度计算，节省资源
        outputs = bert_model(**inputs)
    # 获取每个token的嵌入（最后一层隐藏状态），取平均值作为句子的静态嵌入
    token_embeddings = outputs.last_hidden_state  # shape: (1, 64, 768)
    static_embedding = torch.mean(token_embeddings, dim=1).squeeze().cpu().numpy()  # 转为numpy数组
    return static_embedding

# 6. 生成所有文本的静态BERT嵌入，构建特征矩阵
df['embedding'] = df['cleaned_text'].apply(generate_static_bert_embedding)
X = np.array(df['embedding'].tolist())  # 特征矩阵
y = df['label'].values  # 标签矩阵

# 7. 分割训练集与测试集（比例8:2，固定随机种子确保可复现）
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)

python
import torch
import torch.nn as nn
from torch.utils.data import TensorDataset, DataLoader
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, specificity_score
import matplotlib.pyplot as plt

# 1. 定义模型类
class BertBiLSTM(nn.Module):
    def __init__(self, embedding_dim=768, hidden_dim=128, output_dim=1, dropout=0.3):
        super(BertBiLSTM, self).__init__()
        self.lstm = nn.LSTM(embedding_dim, hidden_dim, bidirectional=True, batch_first=True, dropout=dropout)
        self.fc1 = nn.Linear(hidden_dim*2, 64)  # 双向LSTM输出维度为hidden_dim*2
        self.fc2 = nn.Linear(64, output_dim)
        self.dropout = nn.Dropout(dropout)
        self.relu = nn.ReLU()
    
    def forward(self, x):
        # x shape: (batch_size, embedding_dim)
        x = x.unsqueeze(1)  # 转为(batch_size, 1, embedding_dim)，适配LSTM输入
        lstm_out, _ = self.lstm(x)  # lstm_out shape: (batch_size, 1, hidden_dim*2)
        lstm_out = torch.mean(lstm_out, dim=1)  # 池化，shape: (batch_size, hidden_dim*2)
        out = self.dropout(self.relu(self.fc1(lstm_out)))  # 全连接层1+激活+dropout
        out = torch.sigmoid(self.fc2(out))  # 输出层，shape: (batch_size, 1)
        return out

# 2. 初始化模型、损失函数、优化器
model = BertBiLSTM().cuda()
criterion = nn.BCELoss()  # 二分类交叉熵损失函数
optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)  # 优化器，学习率1e-4

# 3. 构建数据加载器（批量训练，提升效率）
batch_size = 32
train_dataset = TensorDataset(torch.tensor(X_train, dtype=torch.float32).cuda(), torch.tensor(y_train, dtype=torch.float32).cuda())
test_dataset = TensorDataset(torch.tensor(X_test, dtype=torch.float32).cuda(), torch.tensor(y_test, dtype=torch.float32).cuda())
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)

# 4. 模型训练（固定epochs=20，早停机制防止过拟合）
epochs = 20
best_val_f1 = 0.0
train_losses = []
val_losses = []

for epoch in range(epochs):
    # 训练阶段
    model.train()
    train_loss = 0.0
    for batch_x, batch_y in train_loader:
        optimizer.zero_grad()  # 清空梯度
        outputs = model(batch_x).squeeze()  # 挤压维度，适配损失函数输入
        loss = criterion(outputs, batch_y)
        loss.backward()  # 反向传播
        optimizer.step()  # 更新参数
        train_loss += loss.item() * batch_x.size(0)
    train_loss = train_loss / len(train_loader.dataset)
    train_losses.append(train_loss)
    
    # 验证阶段
    model.eval()
    val_loss = 0.0
    val_preds = []
    val_labels = []
    with torch.no_grad():
        for batch_x, batch_y in test_loader:
            outputs = model(batch_x).squeeze()
            loss = criterion(outputs, batch_y)
            val_loss += loss.item() * batch_x.size(0)
            # 生成预测标签（大于0.5为1，否则为0）
            preds = (outputs >= 0.5).float().cpu().numpy()
            val_preds.extend(preds)
            val_labels.extend(batch_y.cpu().numpy())
    val_loss = val_loss / len(test_loader.dataset)
    val_losses.append(val_loss)
    
    # 计算验证集评估指标
    val_acc = accuracy_score(val_labels, val_preds)
    val_precision = precision_score(val_labels, val_preds)
    val_recall = recall_score(val_labels, val_preds)
    val_f1 = f1_score(val_labels, val_preds)
    val_specificity = specificity_score(val_labels, val_preds)
    
    # 早停机制：保存F1分数最高的模型
    if val_f1 > best_val_f1:
        best_val_f1 = val_f1
        torch.save(model.state_dict(), 'bert_bilstm_best.pth')
    
    # 打印每轮训练信息
    print(f"Epoch {epoch+1}/{epochs} | Train Loss: {train_loss:.4f} | Val Loss: {val_loss:.4f}")
    print(f"Val Acc: {val_acc:.4f} | Val Precision: {val_precision:.4f} | Val Recall: {val_recall:.4f} | Val F1: {val_f1:.4f} | Val Specificity: {val_specificity:.4f}")

# 5. 绘制训练损失与验证损失曲线（可视化模型训练效果）
plt.figure(figsize=(10, 6))
plt.plot(range(1, epochs+1), train_losses, label='Train Loss')
plt.plot(range(1, epochs+1), val_losses, label='Val Loss')
plt.xlabel('Epochs')
plt.ylabel('Loss')
plt.title('Train Loss vs Val Loss')
plt.legend()
plt.show()

模型方案

准确率（Accuracy）

精确率（Precision）

召回率（Recall）

F1分数（F1-Score）

特异性（Specificity）

BiLSTM + fastText + GloVe

77.50%

76.32%

75.12%

76.85%

79.66%

微调BERT模型

76.64%

79.15%

78.53%

78.83%

74.31%

本文方案（静态BERT嵌入+BiLSTM）

80.15%

80.32%

79.11%

79.71%

83.03%

python
from flask import Flask, request, jsonify
import torch
import re
from transformers import BertTokenizer, BertModel
import nltk
from nltk.corpus import stopwords

# 初始化Flask应用
app = Flask(__name__)

# 加载停用词、分词器、BERT模型、最优训练模型
nltk.download('stopwords')
stop_words = set(stopwords.words('english'))
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
bert_model = BertModel.from_pretrained('bert-base-uncased').cuda()
model = BertBiLSTM().cuda()
model.load_state_dict(torch.load('bert_bilstm_best.pth'))
model.eval()  # 切换为评估模式

# 复用数据预处理函数
def clean_text(text):
    text = re.sub(r'[^a-zA-Z\s]', '', text)
    text = text.lower()
    text = ' '.join([word for word in text.split() if word not in stop_words])
    return text

def generate_static_bert_embedding(text):
    inputs = tokenizer(text, return_tensors='pt', padding='max_length', truncation=True, max_length=64).to('cuda')
    with torch.no_grad():
        outputs = bert_model(**inputs)
    token_embeddings = outputs.last_hidden_state
    static_embedding = torch.mean(token_embeddings, dim=1).squeeze().cpu().numpy()
    return static_embedding

# 定义检测接口
@app.route('/hate_speech_detect', methods=['POST'])
def detect():
    data = request.get_json()
    text = data.get('text', '')
    if not text:
        return jsonify({'code': 400, 'message': '请输入待检测文本', 'result': {}})
    
    # 预处理+预测
    cleaned_text = clean_text(text)
    embedding = generate_static_bert_embedding(cleaned_text)
    embedding_tensor = torch.tensor(embedding, dtype=torch.float32).unsqueeze(0).cuda()
    with torch.no_grad():
        output = model(embedding_tensor).squeeze().cpu().numpy()
    label = 1 if output >= 0.5 else 0
    confidence = float(output) if label == 1 else float(1 - output)
    
    # 返回结果
    result = {
        'text': text,
        'cleaned_text': cleaned_text,
        'label': label,  # 1:仇恨言论，0:非仇恨言论
        'label_name': '仇恨言论' if label == 1 else '非仇恨言论',
        'confidence': round(confidence, 4)
    }
    return jsonify({'code': 200, 'message': '检测成功', 'result': result})

# 启动服务
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=False)