手把手教你用BGE-M3构建情感分析系统

手把手教你用BGE-M3构建情感分析系统

1. 引言：从文本嵌入到情感分类

在自然语言处理（NLP）任务中，情感分析是企业洞察用户反馈、监控舆情和优化产品体验的核心技术之一。传统方法依赖于词袋模型或LSTM等序列模型，但随着预训练语言模型的发展，基于文本嵌入（Text Embedding）的语义表征方式已成为主流。

本文将围绕BGE-M3这一多功能检索嵌入模型，手把手带你构建一个完整的二分类情感分析系统。我们将涵盖：

• 模型部署与服务启动
• 基于[CLS]向量的微调实践
• 使用池化策略提升分类性能
• ONNX导出与生产环境推理测试

通过本教程，你不仅能掌握如何将检索模型用于下游分类任务，还能了解其在实际工程中的部署路径。

2. BGE-M3 模型简介与服务部署

2.1 BGE-M3 是什么？

BGE-M3 是由 FlagAI 团队推出的三模态混合检索嵌入模型，支持：

密集 + 稀疏 + 多向量三合一检索能力

尽管它最初设计用于检索场景（如RAG、文档匹配），但其强大的语义编码能力也使其成为优秀的通用文本表征模型，适用于分类、聚类、相似度计算等多种任务。

⚠️ 注意：BGE-M3 不是生成式模型，而是双编码器类嵌入模型，输出的是固定维度的语义向量。

2.2 部署本地嵌入服务

我们使用提供的镜像快速部署 BGE-M3 服务。

启动服务（推荐方式）

bash /root/bge-m3/start_server.sh

后台运行并记录日志

nohup bash /root/bge-m3/start_server.sh > /tmp/bge-m3.log 2>&1 &

验证服务是否正常运行

netstat -tuln | grep 7860

访问http://<服务器IP>:7860可查看 Gradio 提供的交互界面。

查看日志输出

tail -f /tmp/bge-m3.log

一旦服务启动成功，即可通过 API 获取文本嵌入向量，为后续微调提供基础支持。

3. 微调 BGE-M3 实现情感分类

虽然 BGE-M3 本身不带分类头，但我们可以通过在其基础上添加全连接层进行迁移学习，实现“预训练 + 微调”范式下的情感分类。

我们将采用两种典型策略：

1. 基于 [CLS] token 的特征提取
2. 基于平均池化的多向量融合

3.1 构建分类模型：基于 [CLS] 向量

在 Transformer 架构中，[CLS] token 的最终隐藏状态常被用作整个句子的聚合表示。

定义分类网络结构

import torch import torch.nn as nn from transformers import AutoModel, AutoTokenizer class TextClassifier(nn.Module): def __init__(self, model, hidden_size=1024, num_classes=2): super().__init__() self.model = model self.classifier = nn.Sequential( nn.Dropout(0.1), nn.Linear(hidden_size, hidden_size), nn.ReLU(), nn.Linear(hidden_size, num_classes) ) def forward(self, inputs): outputs = self.model(**inputs) cls_embedding = outputs.last_hidden_state[:, 0, :] # [CLS] 向量 logits = self.classifier(cls_embedding) return logits

该模型保留原始 BGE-M3 的主干，并在其顶部叠加一个四层分类头（Dropout → Linear → ReLU → Linear）。

3.2 数据准备与 DataLoader 构建

我们构造一个简单的中文情感数据集作为示例：

texts = [ "这是一个积极的句子，充满了正能量。", "这是一个消极的句子，感觉非常糟糕。", "今天天气真好，阳光明媚。", "这个电影太无聊了，浪费时间。", "我喜欢这个产品，质量非常好。", "这个服务太差劲了，非常不满意。", "大模型对程序员来说是一个很好的工具。", "大模型对初级开发者来说是一个坏消息。" ] labels = [1, 0, 1, 0, 1, 0, 1, 0] # 1: 积极, 0: 消极

自定义 Dataset 类

from torch.utils.data import Dataset class TextDataset(Dataset): def __init__(self, texts, labels, tokenizer, max_length=128): self.texts = texts self.labels = labels self.tokenizer = tokenizer self.max_length = max_length def __len__(self): return len(self.texts) def __getitem__(self, idx): text = self.texts[idx] label = self.labels[idx] encoding = self.tokenizer.encode_plus( text, add_special_tokens=True, max_length=self.max_length, padding='max_length', truncation=True, return_tensors='pt' ) return { 'input_ids': encoding['input_ids'].flatten(), 'attention_mask': encoding['attention_mask'].flatten(), 'label': torch.tensor(label, dtype=torch.long) }

创建训练/验证集

from sklearn.model_selection import train_test_split from torch.utils.data import DataLoader train_texts, val_texts, train_labels, val_labels = train_test_split( texts, labels, test_size=0.2, random_state=42 ) tokenizer = AutoTokenizer.from_pretrained("BAAI/bge-m3") model = AutoModel.from_pretrained("BAAI/bge-m3") train_dataset = TextDataset(train_texts, train_labels, tokenizer) val_dataset = TextDataset(val_texts, val_labels, tokenizer) train_dataloader = DataLoader(train_dataset, batch_size=2, shuffle=True) val_dataloader = DataLoader(val_dataset, batch_size=2)

3.3 训练流程配置

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') classifier = TextClassifier(model).to(device) optimizer = torch.optim.AdamW(classifier.parameters(), lr=2e-5) scheduler = get_linear_schedule_with_warmup( optimizer, num_warmup_steps=0, num_training_steps=len(train_dataloader) * 3 )

训练循环函数

def train_epoch(model, dataloader, optimizer, scheduler, device): model.train() total_loss = 0 for batch in tqdm(dataloader, desc="Training"): input_ids = batch['input_ids'].to(device) attention_mask = batch['attention_mask'].to(device) labels = batch['label'].to(device) optimizer.zero_grad() outputs = model({'input_ids': input_ids, 'attention_mask': attention_mask}) loss = nn.CrossEntropyLoss()(outputs, labels) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() scheduler.step() total_loss += loss.item() return total_loss / len(dataloader)

验证函数

def evaluate(model, dataloader, device): model.eval() correct_predictions = 0 total_predictions = 0 with torch.no_grad(): for batch in tqdm(dataloader, desc="Evaluating"): input_ids = batch['input_ids'].to(device) attention_mask = batch['attention_mask'].to(device) labels = batch['label'].to(device) outputs = model({'input_ids': input_ids, 'attention_mask': attention_mask}) _, predictions = torch.max(outputs, dim=1) correct_predictions += (predictions == labels).sum().item() total_predictions += labels.size(0) accuracy = correct_predictions / total_predictions return accuracy

开始训练

print("开始训练...") for epoch in range(3): train_loss = train_epoch(classifier, train_dataloader, optimizer, scheduler, device) val_accuracy = evaluate(classifier, val_dataloader, device) print(f"Epoch {epoch+1}, Loss: {train_loss:.4f}, Acc: {val_accuracy:.4f}")

训练完成后保存模型权重：

torch.save(classifier.state_dict(), 'text_classifier.pth')

4. 改进方案：基于池化的分类模型

仅使用 [CLS] token 可能会丢失部分上下文信息。我们可以改用平均池化（Mean Pooling）或最大池化（Max Pooling）对所有 token 的输出进行聚合。

4.1 定义池化分类器

class PoolingClassifier(nn.Module): def __init__(self, model, hidden_size=1024, num_classes=2, pooling_type="mean"): super().__init__() self.model = model self.pooling_type = pooling_type self.classifier = nn.Sequential( nn.Dropout(0.1), nn.Linear(hidden_size, hidden_size), nn.ReLU(), nn.Linear(hidden_size, num_classes) ) def forward(self, inputs): outputs = self.model(**inputs) last_hidden_state = outputs.last_hidden_state attention_mask = inputs['attention_mask'] if self.pooling_type == "mean": mask = attention_mask.unsqueeze(-1).expand_as(last_hidden_state) masked_hidden = last_hidden_state * mask sum_hidden = torch.sum(masked_hidden, dim=1) sum_mask = torch.clamp(mask.sum(1), min=1e-9) pooled_output = sum_hidden / sum_mask elif self.pooling_type == "max": mask = attention_mask.unsqueeze(-1).bool() masked_hidden = last_hidden_state.masked_fill(~mask, -1e9) pooled_output, _ = torch.max(masked_hidden, dim=1) logits = self.classifier(pooled_output) return logits

4.2 初始化与训练

base_model = AutoModel.from_pretrained("BAAI/bge-m3") classifier = PoolingClassifier(base_model, pooling_type="mean").to(device)

其余训练流程与前述一致，只需替换模型定义即可。

✅ 实践建议：对于长文本或信息分布较散的文本，平均池化通常优于 [CLS]。

5. 生产部署：导出 ONNX 模型

为了在高性能、低延迟的生产环境中运行模型（如C++服务、边缘设备），我们将训练好的模型导出为ONNX 格式。

5.1 导出为 ONNX

def export_to_onnx(model, tokenizer, output_path='onnx_models/text_classifier.onnx'): os.makedirs(os.path.dirname(output_path), exist_ok=True) model.eval() text = "示例句子" inputs = tokenizer(text, return_tensors="pt") input_names = ['input_ids', 'attention_mask'] output_names = ['logits'] dynamic_axes = { 'input_ids': {0: 'batch_size', 1: 'sequence_length'}, 'attention_mask': {0: 'batch_size', 1: 'sequence_length'}, 'logits': {0: 'batch_size'} } torch.onnx.export( model, (inputs['input_ids'], inputs['attention_mask']), output_path, export_params=True, opset_version=14, do_constant_folding=True, input_names=input_names, output_names=output_names, dynamic_axes=dynamic_axes ) onnx.checker.check_model(output_path) print(f"ONNX模型已导出并验证通过: {output_path}") return output_path

5.2 使用 ONNX Runtime 进行推理

import onnxruntime as ort import numpy as np def onnx_inference(onnx_path, tokenizer, text): session = ort.InferenceSession(onnx_path) inputs = tokenizer(text, max_length=128, padding='max_length', truncation=True, return_tensors="np") onnx_inputs = { 'input_ids': inputs['input_ids'], 'attention_mask': inputs['attention_mask'] } outputs = session.run(None, onnx_inputs) prediction = np.argmax(outputs[0], axis=1)[0] return "积极" if prediction == 1 else "消极" # 测试 result = onnx_inference('onnx_models/text_classifier.onnx', tokenizer, "这个手机真的很棒！") print(result) # 输出：积极

6. 总结

本文系统地展示了如何利用 BGE-M3 这一强大的文本嵌入模型构建情感分析系统，涵盖了从模型部署、微调训练到生产导出的完整链路。

核心要点回顾：

1. BGE-M3 是多功能嵌入模型，虽非专为分类设计，但其高质量语义编码能力可迁移至下游任务。
2. 微调策略选择：
   • [CLS] 向量适合短文本、结构清晰的任务；
   • 平均池化更适合信息分散或长文本场景。
3. ONNX 导出是通向生产的桥梁，支持跨平台、高性能推理。
4. 工程建议：
   • 在 GPU 上训练，CPU 上导出；
   • 使用 FP16 减少模型体积；
   • 动态轴设置确保变长输入兼容性。

通过本次实践，你可以将这套方法迁移到其他文本分类任务（如意图识别、垃圾检测、多语言情感分析）中，充分发挥 BGE-M3 的多语言、高精度优势。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。