文本情感分析在Serverless架构下的应用

Posted 2023-01-29

文本情感分析是指对包含人们观点、喜好、情感等的主观性文本进行检测。该领域的发展和快速起步得益于社交媒体。越来越多的用户从单纯地获取互联网信息向创造互联网信息转变，例如产品评论、论坛讨论、博客等由用户发布的主观性文本。自2000年初以来，情感分析已经成长为自然语言处理中最活跃的研究领域之一。事实上，它已经从计算机科学蔓延到管理科学和社会科学，如市场营销、金融、政治学、通信、医疗科学，甚至是历史。由于其重要的商业价值引发整个社会的关注。

时至今日，文本情感分析算法已经相当成熟，其本质是对文本的分类。常用的算法模型有离散贝叶斯、RNN、LSTM、BERT等。目前非常流行的BERT作为分类网络，介绍文本情感分析案例。

情感分析模型的训练

1.数据准备

本案例使用的数据集是IMDB电影评论（下载地址为​​https://ai.stanford.edu/~amaas/data/sentiment/​​）。这是一个用于二元情感分类的数据集，它提供了25000个电影评论用于模型训练，25000个电影评论用于测试，以及其他未标记的数据。下面是其中的一个电影评论片段：

Story of a man who has unnatural feelings for a pig. Starts out with a opening scene that is a terrific example of absurd comedy. A formal orchestra audience is turned into an insane, violent mob by the crazy chantings of it’s singers. Unfortunately it stays absurd the WHOLE time with no general narrative eventually making it just too off putting. Even those from the era should be turned off. The cryptic dialogue would make Shakespeare seem easy to a third grader. On a technical level it’s better than you might think with some good cinematography by future great Vilmos Zsig mond. Future stars Sally Kirkland and Frederic Forrest can be seen briefly.

下载完数据集后，发现训练集和测试集已经分好。我们只需简单操作即可分离出训练需要的数据集：

import os 
import re 
data_root = "./data/aclImdb" 
# 简单的文本处理 
def text_preprocessing(text):    
# 去掉以"@"开头的词    
    text = re.sub(r(@.*?)[\\s],  , text)    
# 用&替换&    
    text = re.sub(r&, &, text)    
# 去掉文本尾部的空格    
    text = re.sub(r\\s+,  , text).strip()    
    return text 

def data_read(mode="train"):    
    data_list = []    
    data_root_mode = os.path.join(data_root, mode)    
    for _label in ["neg", "pos"]:        
        data_label_root = os.path.join(data_root_mode, _label)        
        for _file in os.listdir(data_label_root):            
            with open(os.path.join(data_label_root, _file), "r", encoding="utf-8") as f:                
                data_list.append((text_preprocessing(f.read()), int(_label == "pos")))    
                return data_list train_x, train_y = [list(x) for x in zip(*data_read("train"))] test_x, test_y = [list(x) for x in zip(*data_read("test"))]

其中，train_x和test_x格式类似下图的形式。

train_y和test_y格式类似下图的形式。

训练集和测试集相关数据1 

训练集和测试集相关数据2 

这样的数据集是不能直接用于训练的。模型是计算机语言，文本是人类语言，需要将文本转换为计算机编码。这里以Bert的编码转换工具BertTokenizer为例进行介绍。BertTokenizer可以执行一些基础的大小写、标点符号分割、小写转换、中文字符分割、去除重音符号等操作，最后返回的是关于词的数组。

将训练集的第一个样本进行编码：

from transformers import BertTokenizer 
# 加载 BERT tokenizer 
tokenizer = BertTokenizer.from_pretrained(bert-base-uncased, do_lower_case=True) 
MAX_LEN = 16 
encoded_sent = tokenizer.encode_plus(text=text_preprocessing("Story of a man who has unnatural feelings for a pig."),add_special_tokens=True, max_length=MAX_LEN,padding=max_length,
    return_attention_mask=True, truncation=True ) 
print(encoded_sent) 
print([tokenizer.ids_to_tokens[x] for x in encoded_sent[input_ids]])

MAX_LEN简单设置为16，后续训练时需设置为更大的值。

上述代码执行后，可以得到编码后的样本：

input_ids: [101, 2466, 1997, 1037, 2158, 2040, 2038, 21242, 5346, 2005, 1037, 10369, 1012, 102, 0, 0], token_type_ids: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], attention_mask: [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0]

BertTokenizer的每个模型都自带一个词库。编码后的input_ids表示每个样本在词库中id集成的序列，input_ids对应的句子如下：

[[CLS], story, of, a, man, who, has, unnatural, feelings, for, a, pig, ., [SEP], [PAD], [PAD]]

可以看到，句子的首尾增加了“[CLS]”“[SEP]”和“[PAD]”特殊编码。[CLS]表示用于分类场景，表示整句话的语义；[SEP]表示分隔符，放在句尾也可以表示句子结束；[PAD]针对有长度要求的场景，填充文本长度（Padding），使得文本长度达到要求，对应编码是0。

在编码后的样本中还有两个字段token_type_ids和attention_mask。token_type_ids表示编码的类型；attention_mask表示是否对该字符进行了文本长度填充，即该字符是否是文本填充字符[PAD]。

之后创建一个方法preprocessing_for_bert()来对所有的数据进行编码，并返回编码后的input_ids和attention_mask：

MAX_LEN = 256 
# 创建一个方法来切分一串文本 
def preprocessing_for_bert(data):    
    input_ids = []    
    attention_masks = []    
    for sent in data:        
        encoded_sent = tokenizer.encode_plus( text=text_preprocessing(sent), add_special_tokens=True, max_length=MAX_LEN, padding=max_length, 
        return_attention_mask=True,  truncation=True)        
        input_ids.append(encoded_sent.get(input_ids))        
        attention_masks.append(encoded_sent.get(attention_mask))    
        # 转化成张量（Tensor）格式    
        input_ids = torch.tensor(input_ids)    
        attention_masks = torch.tensor(attention_masks)    
        return input_ids, attention_masks 
        # 用preprocessing_for_bert来处理训练集和验证集 
        print(Tokenizing data...) 
        train_inputs, train_masks = preprocessing_for_bert(train_x) 
        test_inputs, test_masks = preprocessing_for_bert(test_x) 
        train_labels = torch.tensor(train_y) 
        test_labels = torch.tensor(test_y)

和之前的案例一样，对文本数据集也需要创建数据加载器。这里需要注意的是，训练集和验证集都需要转换成张量（Tensor）形式，因为后面的数据加载器TensorDataset只适配张量。

import torch 
from torch.utils.data import TensorDataset, DataLoader, RandomSampler, Sequentia-lSampler 
batch_size = 8 
# 为训练集创建数据加载器 
train_data = TensorDataset(train_inputs, train_masks, train_labels) 
train_sampler = RandomSampler(train_data) 
train_dataloader = DataLoader(train_data, sampler=train_sampler, batch_size=batch_size) 
# 为验证集创建数据加载器 
test_data = TensorDataset(test_inputs, test_masks, test_labels) 
test_sampler = SequentialSampler(test_data) 
test_dataloader = DataLoader(test_data, sampler=test_sampler, batch_size=batch_size)

这里只将batch_size定义为8，是因为IMDB数据集的MAX_LEN已经定义为256。这是一个很长的句子，会导致模型训练占用过大的空间，若是有资源则最好设置为16或32。至此，对数据集的处理已经完成。下面介绍模型的定义和训练。

2.模型定义

基础的模型采用BERT算法，这里只将BERT作为Embedding形式调用，实际上只需要几行代码即可实现：

from transformers import BertForSequenceClassification 
if torch.cuda.is_available():    
    device = torch.device("cuda") 
else:    
    device = torch.device("cpu") 
bert_classifier = BertForSequenceClassification.from_pretrained(bert-base-uncased, num_labels=2) 
# 告诉这个实例化的分类器，使用GPU还是CPU bert_classifier.to(device)

整个模型可以直接用transformers库自带的BertForSequenceClassification分类模型。这个分类模型可以简化为如下形式。

1）定义基础骨架Bert模型。

2）定义Dropout层和分类层，包含1个全连接层，神经元数量为分类数。

3）定义前向传播方法，将input_ids和attention_mask传入BERT。

4）将从Bert输出中的CLS最后一个隐藏层参数传入分类层，得到最终输出。

class BertForSequenceClassification(nn.Module):    
    def __init__(self, config):        
        super().__init__(config)        
        self.num_labels = config.num_labels        
        self.bert = BertModel(config)        
        self.dropout = nn.Dropout(config.hidden_dropout_prob)        
        self.classifier = nn.Linear(config.hidden_size, config.num_labels)        
        self.init_weights()    

    def forward(self, input_ids, attention_mask, ...):        
        outputs = self.bert(input_ids, attention_mask=attention_mask)        
        pooled_output = outputs[1]        
        pooled_output = self.dropout(pooled_output)        
        logits = self.classifier(pooled_output)        
        loss = None        
        loss_fct = CrossEntropyLoss()        
        loss = loss_fct(logits.view(-1, self.num_labels), labels.view(-1))        
        return loss, output

3.模型训练

和之前的案例一样，每个模型训练前都需要定义优化器、损失函数等。这里优化器选择AdamW，初始学习率为0.00005，损失函数在模型中已经定义，默认为交叉熵损失函数。

optimizer = AdamW(bert_classifier.parameters(), lr=5e-5, eps=1e-8)

下面开始模型训练和验证，具体流程如下。

1）初始化各个计数器，如整体损失、训练步数等。

2）遍历地从训练数据加载器中获取处理过的训练样本和对应的标签。

3）将训练样本输入模型，得到前向传播的输出和损失函数结果。

4）损失函数调用backward方法进行反向传播，并使用梯度裁剪防止梯度爆炸。

5）更新优化器参数。

6）打印各个计数器的值。

7）在每个Epoch中重复步骤2～6进行训练。

8）对所有Epoch执行步骤1～7。

最后一个Epoch训练结束后，进行模型验证并保存：

模型训练：

def train(epochs=4):    
    print("Start training...")    
    for epoch_i in range(epochs):        
    # 每个epoch开始前将各个计数器归零        
        total_loss, batch_loss, batch_counts = 0, 0, 0        
        bert_classifier.train()        
    # 从数据加载器中读取数据        
    for step, batch in enumerate(train_dataloader):            
        batch_counts += 1            
        b_input_ids, b_attn_mask, b_labels = tuple(t.to(device) for t in batch)            
    # 将累计梯度清零            
    bert_classifier.zero_grad()            
    # 往模型中传入得到的input_id和mask，模型进行前向传播，进而得到logits值            
    loss, logits = bert_classifier(b_input_ids, token_type_ids=None, attention_mask=b_attn_mask, labels=b_labels)            
    batch_loss += loss.item()            
    total_loss += loss.item()           
    # 执行后向传播算法，计算梯度            
    loss.backward()            
    # 修剪梯度进行归一化，防止梯度爆炸            
    torch.nn.utils.clip_grad_norm_(bert_classifier.parameters(), 1.0)            
    # 更新模型参数，更新学习率            
    optimizer.step()            
    # 每训练100个Batch打印一次损失值和时间消耗            
    if (step % 100 == 0 and step != 0) or (step == len(train_dataloader) - 1):                
        print("Epoch: , Steps: ,  Loss: ".format(epoch_i, step, loss.item()))            
    # 将计数器清零            
    batch_loss, batch_counts = 0, 0        
    # 计算整个训练数据集的平均损失        
    avg_train_loss = total_loss / len(train_dataloader)        
    # 在最后一个epoch训练结束后，用验证集来测试模型的表现        
    if epoch_i == epochs - 1 or epoch_i % 5 == 0:            
        print("Epoch: , Avg_loss: , Acc: ".format(epoch_i, avg_train_loss, evaluate()))            
        torch.save(bert_classifier, "./aclImdb_bert_cls_new_.pth".format(epoch_i))    
        print("Training complete!")

之后介绍验证方法的构造。前向传播流程和训练过程是一样的，在获取模型输出后，将其处理为预测值。由于模型的最后一层是分类层，所以获取的模型输出是当前样本属于每个类别的置信度，选择最大置信度对应的类别即可：

preds = torch.argmax(logits, dim=1).flatten()

再进行准确率计算，直接调用sklearn中的accuracy_score方法：

from sklearn.metrics import accuracy_score accuracy_score(test_y, y_pred)

最终构造完成的代码如下：

def evaluate():    
    bert_classifier.eval（)    
    # 创建空集，记录每一个Batch的准确率    
    all_logits = []    
    for batch in test_dataloader:        
    # 加载 Batch 数据到 GPU或CPU        
        b_input_ids, b_attn_mask, b_labels = tuple(t.to(device) for t in batch)        
    # 计算 logits        
        with torch.no_grad():            
            loss, logits = bert_classifier(b_input_ids, token_type_ids=None, attention_mask=b_attn_mask, labels=b_labels)        
            all_logits.append(logits)    
            all_logits = torch.cat(all_logits, dim=0)    
            # Get accuracy over the test set    
            y_pred = torch.argmax(all_logits, dim=1).flatten().cpu().numpy()    
            return accuracy_score(test_y, y_pred)

现在可以开始训练了。给train方法传入epochs参数，这里定义epochs=1，即迭代4轮训练：

train(epochs=1)

由于日志很多，这里只选择最后几个步骤的日志。可以看到，1个Epoch之后，准确率就已经达到89.8%。

Epoch: 0, Steps: 2800, Loss: 0.12946712970733643

Epoch: 0, Steps: 2900, Loss: 0.17792001366615295

Epoch: 0, Steps: 3000, Loss: 0.003091663122177124

Epoch: 0, Steps: 3100, Loss: 0.004699620418250561

Epoch: 0, Steps: 3124, Loss: 0.0766051858663559

Epoch: 0, Avg_loss: 0.09662641194868833, Acc: 0.89816

在TensorBoard中可以看到，模型训练时Loss的变化。它的波动非常大，但整体来讲是在逐渐变小，越来越收敛，符合预期效果，如下所示。

模型训练时Loss变化曲线 

整个情感分析的网络结构如下所示。

情感分析的网络结构 

部署到Serverless架构

1.本地推理代码开发

导入推理需要的模块，并定义类别：

import re

import torch

from transformers import BertTokenizer

class_list = ["neg", "pos"]

MAX_LEN = 256

定义推理模块：

class SegModelPredictor(object):    
    def __init__(self, model_math=None):        
        self.tokenizer = BertTokenizer.from_pretrained(bert-base-uncased, do_lower_ case=True)        
        self.model = torch.load(model_math)        
        self.model.eval（)        
        self.device = torch.device(cpu)        
        self.model.to(self.device)    

    def predict(self, content):        
        encoded_sent = self.tokenizer.encode_plus(text=text_preprocessing(content), add_special_tokens=True, max_length=MAX_LEN, padding=max_length, return_attention_mask=True,truncation=True)        
        input_ids = torch.tensor([encoded_sent.get(input_ids)])        
        attention_mask = torch.tensor([encoded_sent.get(attention_mask)])        
        logits = self.model(input_ids, attention_mask=attention_mask)        
        logits = torch.argmax(logits[0], dim=1).flatten().cpu().numpy()        
        result = class_list[int(logits)]        
        return result

在测试集中找两个数据进行测试，其中的路径代表了所标注样本的分类（neg和pos）：

if __name__ == "__main__":    

　　model = SegModelPredictor("./aclImdb_bert_cls_new_0.pth")    

　　print(model.predict(open("./aclImdb/test/neg/0_2.txt", "r").read()))    

　　print(model.predict(open("./aclImdb/test/pos/0_10.txt", "r").read()))    

　　print(model.predict("I am too sad to go to work"))

可以看到，模型预测效果还是不错的：

neg pos neg

2.本地推理服务开发

将上述推理代码简化、整合成推理服务：

import re 
from flask import Flask, request 
import torch 
from transformers import BertTokenizer 
app = Flask(__name__) 
class_list = ["neg", "pos"] 
MAX_LEN = 256 
def text_preprocessing(text):    
# 删除 @name    
    text = re.sub(r(@.*?)[\\s],  , text)    
    text = re.sub(r&, &, text)    
    # 删除空格    
    text = re.sub(r\\s+,  , text).strip()    
    return text 
    if torch.cuda.is_available():    
        device = torch.device("cuda") 
    else:    
        device = torch.device("cpu") 

class SegModelPredictor(object):    
    def __init__(self, model_math=None):        
        self.tokenizer = BertTokenizer.from_pretrained(bert-base-uncased, do_lower_case= True)        
        self.model = torch.load(model_math, map_location="cpu")        
        self.model.eval（)        
        self.device = device        
        self.model.to(self.device)    

    def predict(self, content):        
        encoded_sent = self.tokenizer.encode_plus(text=text_preprocessing(content), add_special_tokens=True, max_length=MAX_LEN, padding=max_length, return_attention_mask=True, truncation=True )        
        input_ids = torch.tensor([encoded_sent.get(input_ids)])        
        attention_mask = torch.tensor([encoded_sent.get(attention_mask)])        
        logits = self.model(input_ids, attention_mask=attention_mask)        
        logits = torch.argmax(logits[0], dim=1).flatten().cpu().numpy()        
        result = class_list[int(logits)]        
        return result 

model_path = "./model/aclImdb_bert_cls.pth" 
model = SegModelPredictor(model_path) 
@app.route(/invoke, methods=[POST]) 
def invoke():    
    return result: model.predict(request.get_data().decode("utf-8")) 

if __name__ == __main__:    
    app.run(debug=True, host=0.0.0.0, port=9000)

然后在本地启动推理服务，可以看到如下日志：

* Serving Flask app index (lazy loading)

* Environment: production    

WARNING: This is a development server. Do not use it in a production deployment.    Use a production WSGI server instead.

* Debug mode: on

* Running on all addresses.    

WARNING: This is a development server. Do not use it in a production deployment.

* Running on http://192.168.1.4:9000/ (Press CTRL+C to quit)

* Restarting with stat

* Debugger is active!

* Debugger PIN: 124-955-177

新启动一个终端，然后输入：

curl --location --request POST http://0.0.0.0:9000/invoke \\ --header Content-Type: text/plain \\ --data-raw I am too sad to go to work

可以看到输出结果为：

"result": "neg"

项目Serverless化

1.部署前准备

通过容器镜像将业务部署到Serverless架构，首先编写相关的Dockerfile文件：

FROM python:3.7-slim

# Create app directory

WORKDIR /usr/src/app

# Bundle app source COPY . .

RUN pip install torch==1.7.0 torchvision==0.8.0 transformers==3.5.0 flask numpy  -i https://pypi.tuna.tsinghua.edu.cn/simple

编写符合Serverless Devs规范的Yaml文件：

# s.yaml 
edition: 1.0.0 
name: emotional 
access: default 
services:    
  emotional        
  component: fc        
  props:            
    region: cn-shanghai            
    service:                
      name: emotional                
      description: emotional  service            
      function:                
        name: emotional -function                
        runtime: custom-container                
        caPort: 8080                
        codeUri: ./                
        timeout: 60                
        gpuMemorySize: 8192                
        instanceType: g1                
        customContainerConfig:                    
          image: registry.cn-shanghai.aliyuncs.com/custom-container/emotional:0.0.1                    
          command: ["python"]                    
          args: ["index.py"]            
          triggers:                
            - name: httpTrigger                    
            type: http                    
            config:                        
              authType: anonymous                        
              methods:                            
                - GET                            
                - POST            
                customDomains:                
                  - domainName: auto                    
                  protocol: HTTP                    
                  routeConfigs:                        
                    - path: /*

2.项目部署

首先构建镜像，此处可以通过Serverless Devs开发者工具进行构建：

s build --use-docker

构建完成之后，可以通过工具直接进行部署：

s deploy --use-local -y

部署完成后，还可以进一步执行相关预留实例操作，以最小化冷启动影响：

# 配置预留实例

$ s provision put --target 1 --qualifier LATEST

# 查询预留实例是否就绪

$ s provision get --qualifier LATEST

完成上述操作后，可以通过invoke命令进行函数的调用与测试，也可以通过返回的地址进行函数的可视化测试。

项目总结

为了降低冷启动带来的影响、部署难度，以及提升计算性能，本项目通过Serverless Devs开发者工具将业务部署到阿里云Serverless架构的GPU实例，并执行了预留操作，在一定程度上大大降低了冷启动对项目的影响。

针对人工智能应用的GPU基础设施，我们通常会面临设计周期长、运维复杂度高、集群利用率低、成本较高等问题。Serverless将这些问题从用户侧转移至云厂商侧，使得用户无须关心底层GPU基础设施的方方面面，全身心聚焦于业务本身，大大简化了业务达成路径。Serverless架构具备以下优点。

1）在成本优先的人工智能应用场景，其优点如下。

·提供弹性预留模式，从而按需为用户保留GPU工作实例，比自建GPU集群成本优势低。

·提供GPU共享虚拟化，支持以1/2、独占方式使用GPU，允许业务以更精细的方式配置GPU实例。

2）在效率优先的人工智能应用场景：摆脱运维GPU集群的繁重负担（驱动/CUDA版本管理、机器运行管理、GPU坏卡管理），使得开发者专注于代码开发、聚焦于业务目标。

本项目基于Serverless架构的情感分析，通过GPU实例与预留模式的加持，在性能上有了质的飞跃。相信随着时间的推移，人工智能项目会在Serverless架构上有更深入的应用、更为完善的最佳实践。