526互联

综合设计——多源异构数据采集与融合应用综合实践

发布时间 2023-12-14 23:22:31作者: 酱酱酱酱江

项目代码

这个项目属于哪个课程 2023数据采集与融合技术
组名、项目简介 组名：Double 20000、项目需求：设计出一个交互友好的多源异构数据的采集与融合的小应用 、项目目标：通过在网页中上传文本、图片、视频或音频分析其中的情感 、项目开展技术路线：前端3件套、Python、fastapi
团队成员学号 042101414、052101230、102102104、102102105、102102108、102102111、102102157、102102158
这个项目目标 通过在网页中上传文本、图片、视频或音频分析其中的情感
其他参考文献 [1]李慧,庞经纬.基于文图音融合的多模态网民情感识别研究[J/OL].数据分析与知识发现:1-17[2023-12-13].http://kns.cnki.net/kcms/detail/10.1478.g2.20231011.1557.012.html.

项目整体介绍：

项目名称：多模态情感分析系统

项目背景：在当前的数字化时代，情感分析在各种应用中变得越来越重要，如客户服务、市场分析和社交媒体监控。多模态情感分析能够提供比单一模态更丰富、更准确的情感识别和分析。

项目目标：开发一个多模态情感分析系统，能够处理和分析文本、图片、音频和视频数据，从而提供综合的情感分析结果。

技术路线：

  • 前端开发：

    • 使用HTML、CSS和JavaScript进行界面设计，实现用户与系统的交互。

    • 集成文件上传功能，支持文本、图片、音频和视频文件。

  • 后端开发：

    • 使用Python进行后端逻辑的编写。

    • 利用FastAPI框架处理前端请求和数据传输。

  • 数据处理与分析：

    • 文本分析：最开始自己训练模型但是后来因为文心一言的准确率更高，因此采用文心一言的接口进行文本情感分析。

      （音频、视频、图片找不到接口，因此自己训练模型）

    • 音频分析

      • 使用RAVDESS数据集进行训练。
      • 对上传的音频文件进行特征提取和情感识别。

    • 图片分析：

      • 使用VGG模型进行图像处理。
      • 利用CK+和FER数据集进行情感分类。

    • 视频分析（找不到可以训练视频的模型，最后只能通过提取音频进行分析）

      • 提取视频中的音频部分。
      • 对提取的音频进行分析，使用同音频分析的方法。

  • 结果输出与展示：将分析结果通过前端界面展示。

最终效果：

通过在本地上传文件进行分析并且得到结果

个人分工介绍

1.查找数据集

  • 采用simplifyweibo_4_moods数据集，数据集中包含36万多条，带情感标注新浪微博，包含4种情感，其中喜悦约20万条，愤怒、厌恶、低落各约5万条。
  • simplifyweibo_4_moods数据集是从新浪微博中收集而来的，因此它反映了真实人类的情感表达。这使得情感分类模型更能够适应实际应用场景，具有更高的可解释性和可靠性。
  • 由于是针对中文微博进行情感分类，因此可以避免英文情感数据集中的语言差异问题。这就需要使用中文自然语言处理技术来处理和分析中文文本，在中文文本处理方面，该数据集提供了很好的资源。

2.选择模型

  • 选择BERT分类器模型，它是一种预训练的语言表示模型，它利用了Transformer架构和大规模文本语料库来学习通用的语言表征。BERT 模型的主要特点是双向性，即在预训练阶段能够同时利用上下文信息，从而更好地理解单词的含义。
  • BertClassifier类就是针对文本分类任务而设计的，它利用预训练的中文 BERT 模型作为基础，通过添加额外的线性层和激活函数，将 BERT 的输出转化为最终的分类输出。这样的设计使得我们可以利用 BERT 强大的语言表示能力，快速构建并微调适用于各种文本分类任务的模型。

3.训练模型

代码解释：

  • 首先使用BertClassifier类，该类继承自nn.Module，其中包含了一个预训练的中文BERT模型、一些线性层和激活函数。
  • 训练模型：定义了一个 train 函数，用于训练模型。在训练循环中，使用 DataLoader 加载数据，计算损失和精度，并更新模型参数。
  • 评估模型：定义了一个 evaluate 函数，用于评估模型在测试集上的表现。
  • 模型保存：在训练完成后，将训练好的模型参数保存到名为"BERT-weibo.pt"的文件中。

代码如下

import torch
import numpy as np
from transformers import BertTokenizer
import pandas as pd
from torch import nn
from transformers import BertModel
from torch.optim import Adam
from tqdm import tqdm


df = pd.read_csv('simplifyweibo_4_moods.csv')
np.random.seed(112)
df_train, df_val, df_test = np.split(df.sample(frac=1, random_state=42),
                                     [int(.8*len(df)), int(.9*len(df))])  # 拆分为训练集、验证集和测试集，比例为 80:10:10。

tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')

class Dataset(torch.utils.data.Dataset):
    def __init__(self, df):
        self.labels = np.array(df['label'])
        self.texts = [tokenizer(text,
                                padding='max_length',
                                max_length = 16,
                                truncation=True,
                                return_tensors="pt")
                      for text in df['review']]

    def classes(self):
        return self.labels

    def __len__(self):
        return len(self.labels)

    def get_batch_labels(self, idx):
        # Fetch a batch of labels
        return np.array(self.labels[idx])

    def get_batch_texts(self, idx):
        # Fetch a batch of inputs
        return self.texts[idx]

    def __getitem__(self, idx):
        batch_texts = self.get_batch_texts(idx)
        batch_y = self.get_batch_labels(idx)
        return batch_texts, batch_y

# 构建模型
class BertClassifier(nn.Module):
    def __init__(self, dropout=0.5):
        super(BertClassifier, self).__init__()
        self.bert = BertModel.from_pretrained('bert-base-chinese',num_labels=15)
        self.dropout = nn.Dropout(dropout)
        self.linear = nn.Linear(768, 4)
        self.relu = nn.ReLU()

    def forward(self, input_id, mask):
        _, pooled_output = self.bert(input_ids= input_id, attention_mask=mask,return_dict=False)
        dropout_output = self.dropout(pooled_output)
        linear_output = self.linear(dropout_output)
        final_layer = self.relu(linear_output)
        return final_layer


# 训练模型
def train(model, train_data, val_data, learning_rate, epochs, batch_size):
    # 通过Dataset类获取训练和验证集
    train, val = Dataset(train_data), Dataset(val_data)
    # DataLoader根据batch_size获取数据，训练时选择打乱样本
    train_dataloader = torch.utils.data.DataLoader(train, batch_size, shuffle=True)
    val_dataloader = torch.utils.data.DataLoader(val, batch_size)
    # 判断是否使用GPU
    use_cuda = torch.cuda.is_available()
    device = torch.device("cuda" if use_cuda else "cpu")
    # 定义损失函数和优化器
    criterion = nn.CrossEntropyLoss()
    optimizer = Adam(model.parameters(), lr=learning_rate)

    if use_cuda:
        model = model.cuda()
        criterion = criterion.cuda()
    # 开始进入训练循环
    for epoch_num in range(epochs):
        # 定义两个变量，用于存储训练集的准确率和损失
        total_acc_train = 0
        total_loss_train = 0
        # 进度条函数tqdm
        for train_input, train_label in tqdm(train_dataloader):
            train_label = train_label.to(device)
            mask = train_input['attention_mask'].to(device)
            input_id = train_input['input_ids'].squeeze(1).to(device)
            # 通过模型得到输出
            output = model(input_id, mask)
            # 计算损失
            batch_loss = criterion(output, train_label.long())
            total_loss_train += batch_loss.item()
            # 计算精度
            acc = (output.argmax(dim=1) == train_label).sum().item()
            total_acc_train += acc
            # 模型更新
            model.zero_grad()
            batch_loss.backward()
            optimizer.step()
        # ------ 验证模型 -----------
        # 定义两个变量，用于存储验证集的准确率和损失
        total_acc_val = 0
        total_loss_val = 0
        # 不需要计算梯度
        with torch.no_grad():
            # 循环获取数据集，并用训练好的模型进行验证
            for val_input, val_label in val_dataloader:
                # 如果有GPU，则使用GPU，接下来的操作同训练
                val_label = val_label.to(device)
                mask = val_input['attention_mask'].to(device)
                input_id = val_input['input_ids'].squeeze(1).to(device)
                output = model(input_id, mask)
                batch_loss = criterion(output, val_label.long())
                total_loss_val += batch_loss.item()
                acc = (output.argmax(dim=1) == val_label).sum().item()
                total_acc_val += acc
        print(
            f'''Epochs: {epoch_num + 1} 
              | Train Loss: {total_loss_train / len(train_data): .3f} 
              | Train Accuracy: {total_acc_train / len(train_data): .3f} 
              | Val Loss: {total_loss_val / len(val_data): .3f} 
              | Val Accuracy: {total_acc_val / len(val_data): .3f}''')

EPOCHS = 10  # 训练轮数
model = BertClassifier()  # 定义的模型
LR = 1e-6  # 学习率
Batch_Size = 16  # 看你的GPU，要合理取值
train(model, df_train, df_val, LR, EPOCHS, Batch_Size)
torch.save(model.state_dict(), 'BERT-weibo.pt')

# 评估模型
def evaluate(model, test_data, batch_size):
    test = Dataset(test_data)
    test_dataloader = torch.utils.data.DataLoader(test, batch_size=batch_size)
    use_cuda = torch.cuda.is_available()
    device = torch.device("cuda" if use_cuda else "cpu")
    if use_cuda:
        model = model.cuda()
    total_acc_test = 0
    with torch.no_grad():
        for test_input, test_label in test_dataloader:
            test_label = test_label.to(device)
            mask = test_input['attention_mask'].to(device)
            input_id = test_input['input_ids'].squeeze(1).to(device)
            output = model(input_id, mask)
            acc = (output.argmax(dim=1) == test_label).sum().item()
            total_acc_test += acc
    print(f'Test Accuracy: {total_acc_test / len(test_data): .3f}')
evaluate(model, df_test, Batch_Size)

训练过程：

模型测试

使用训练好的"BERT-weibo.pt"文件测试，输入一段文字返回对应的情感。

import torch
from transformers import BertTokenizer
from torch import nn
from transformers import BertModel, BertConfig

def get_label_string(label):
    labels = {'喜悦': 0,
              '愤怒': 1,
              '厌恶': 2,
              '低落': 3
              }
    for key, value in labels.items():
        if value == label:
            return key
    return None

# 构建模型
class BertClassifier(nn.Module):
    def __init__(self, config, dropout=0.5):
        super(BertClassifier, self).__init__()
        self.bert = BertModel.from_pretrained('bert-base-chinese', config=config)
        self.dropout = nn.Dropout(dropout)
        self.linear = nn.Linear(768, 4)
        self.relu = nn.ReLU()

    def forward(self, input_id, mask):
        _, pooled_output = self.bert(input_ids=input_id, attention_mask=mask, return_dict=False)
        dropout_output = self.dropout(pooled_output)
        linear_output = self.linear(dropout_output)
        final_layer = self.relu(linear_output)
        return final_layer

model_config = BertConfig.from_pretrained('bert-base-chinese')
model = BertClassifier(config=model_config)
state_dict = torch.load('BERT-weibo.pt')
del state_dict['bert.embeddings.position_ids']
model.load_state_dict(state_dict)
model.eval()

tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')
text = '我讨厌你'
text_input = tokenizer(text, padding='max_length', max_length=16, truncation=True, return_tensors="pt")
mask = text_input['attention_mask']
input_id = text_input['input_ids']
output = model(input_id, mask)
output = output.argmax(dim=1)
output = output.item()
label_string = get_label_string(output)
print(label_string)

测试结果

心得体会：

我对使用BERT模型进行文本分类有了更深入的理解。BERT预训练模型是一种非常强大的语言模型，能够学习到丰富的上下文表示。通过微调BERT模型，可以将其应用于具体的文本分类任务，提升模型的性能，使用pandas库读取和处理csv文件，并使用BertTokenizer对文本进行编码。这样可以将文本转化为模型可接受的输入形式，包括分词、填充和截断等操作。了解到了如何使用PyTorch构建自定义的神经网络模型，通过继承nn.Module类，可以定义自己的模型结构，并在forward方法中实现前向计算。这样可以灵活地组合不同的层和激活函数，满足具体任务的需求，使用DataLoader进行数据批处理的方法提高训练效率。
通过这次实践，我对深度学习文本分类任务有了更深入的理解，同时也增加了对PyTorch框架的熟悉程度。我相信这些知识和经验将对我的未来研究和工作有很大的帮助。