基于jieba分词的文本多分类
- 目标
- 数据准备
- 参数配置
- 数据处理
- 模型构建
- 主程序
- 测试与评估
- 测试结果
目标
本文基于给定的词表，将输入的文本基于jieba分词分割为若干个词，然后将词基于词表进行初步编码，之后经过网络层，输出在已知类别标签上的概率分布，从而实现一个简单文本的多分类。
数据准备
词表文件chars.txt
类别标签文件schema.json
{
"停机保号": 0,
"密码重置": 1,
"宽泛业务问题": 2,
"亲情号码设置与修改": 3,
"固话密码修改": 4,
"来电显示开通": 5,
"亲情号码查询": 6,
"密码修改": 7,
"无线套餐变更": 8,
"月返费查询": 9,
"移动密码修改": 10,
"固定宽带服务密码修改": 11,
"UIM反查手机号": 12,
"有限宽带障碍报修": 13,
"畅聊套餐变更": 14,
"呼叫转移设置": 15,
"短信套餐取消": 16,
"套餐余量查询": 17,
"紧急停机": 18,
"VIP密码修改": 19,
"移动密码重置": 20,
"彩信套餐变更": 21,
"积分查询": 22,
"话费查询": 23,
"短信套餐开通立即生效": 24,
"固话密码重置": 25,
"解挂失": 26,
"挂失": 27,
"无线宽带密码修改": 28
}
训练集数据train.json训练集数据
验证集数据valid.json验证集数据
参数配置
config.py
# -*- coding: utf-8 -*-
"""
配置参数信息
"""
Config = {
"model_path": "model_output",
"schema_path": "../data/schema.json",
"train_data_path": "../data/train.json",
"valid_data_path": "../data/valid.json",
"vocab_path":"../chars.txt",
"max_length": 20,
"hidden_size": 128,
"epoch": 10,
"batch_size": 32,
"optimizer": "adam",
"learning_rate": 1e-3,
}
数据处理
loader.py
# -*- coding: utf-8 -*-
import json
import re
import os
import torch
import random
import jieba
import numpy as np
from torch.utils.data import Dataset, DataLoader
"""
数据加载
"""
class DataGenerator:
def __init__(self, data_path, config):
self.config = config
self.path = data_path
self.vocab = load_vocab(config["vocab_path"])
self.config["vocab_size"] = len(self.vocab)
self.schema = load_schema(config["schema_path"])
self.config["class_num"] = len(self.schema)
self.load()
def load(self):
self.data = []
with open(self.path, encoding="utf8") as f:
for line in f:
line = json.loads(line)
#加载训练集
if isinstance(line, dict):
questions = line["questions"]
label = line["target"]
label_index = torch.LongTensor([self.schema[label]])
for question in questions:
input_id = self.encode_sentence(question)
input_id = torch.LongTensor(input_id)
self.data.append([input_id, label_index])
else:
assert isinstance(line, list)
question, label = line
input_id = self.encode_sentence(question)
input_id = torch.LongTensor(input_id)
label_index = torch.LongTensor([self.schema[label]])
self.data.append([input_id, label_index])
return
def encode_sentence(self, text):
input_id = []
if self.config["vocab_path"] == "words.txt":
for word in jieba.cut(text):
input_id.append(self.vocab.get(word, self.vocab["[UNK]"]))
else:
for char in text:
input_id.append(self.vocab.get(char, self.vocab["[UNK]"]))
input_id = self.padding(input_id)
return input_id
#补齐或截断输入的序列，使其可以在一个batch内运算
def padding(self, input_id):
input_id = input_id[:self.config["max_length"]]
input_id += [0] * (self.config["max_length"] - len(input_id))
return input_id
def __len__(self):
return len(self.data)
def __getitem__(self, index):
return self.data[index]
#加载字表或词表
def load_vocab(vocab_path):
token_dict = {}
with open(vocab_path, encoding="utf8") as f:
for index, line in enumerate(f):
token = line.strip()
token_dict[token] = index + 1 #0留给padding位置，所以从1开始
return token_dict
#加载schema
def load_schema(schema_path):
with open(schema_path, encoding="utf8") as f:
return json.loads(f.read())
#用torch自带的DataLoader类封装数据
def load_data(data_path, config, shuffle=True):
dg = DataGenerator(data_path, config)
dl = DataLoader(dg, batch_size=config["batch_size"], shuffle=shuffle)
return dl
if __name__ == "__main__":
from config import Config
dg = DataGenerator("valid_tag_news.json", Config)
print(dg[1])
主要实现一个自定义数据加载器 DataGenerator，用于加载和处理文本数据。它通过词汇表和标签映射将输入文本转化为索引序列，并进行补齐或截断。
模型构建
model.py
# -*- coding: utf-8 -*-
import torch
import torch.nn as nn
from torch.optim import Adam, SGD
"""
建立网络模型结构
"""
class TorchModel(nn.Module):
def __init__(self, config):
super(TorchModel, self).__init__()
hidden_size = config["hidden_size"]
vocab_size = config["vocab_size"] + 1
max_length = config["max_length"]
class_num = config["class_num"]
self.embedding = nn.Embedding(vocab_size, hidden_size, padding_idx=0)
self.layer = nn.Linear(hidden_size, hidden_size)
self.classify = nn.Linear(hidden_size, class_num)
self.pool = nn.AvgPool1d(max_length)
self.activation = torch.relu #relu做激活函数
self.dropout = nn.Dropout(0.1)
self.loss = nn.functional.cross_entropy #loss采用交叉熵损失
#当输入真实标签，返回loss值；无真实标签，返回预测值
def forward(self, x, target=None):
x = self.embedding(x) #input shape:(batch_size, sen_len)
x = self.layer(x) #input shape:(batch_size, sen_len, input_dim)
x = self.pool(x.transpose(1,2)).squeeze() #input shape:(batch_size, sen_len, input_dim)
predict = self.classify(x) #input shape:(batch_size, input_dim)
if target is not None:
return self.loss(predict, target.squeeze())
else:
return predict
def choose_optimizer(config, model):
optimizer = config["optimizer"]
learning_rate = config["learning_rate"]
if optimizer == "adam":
return Adam(model.parameters(), lr=learning_rate)
elif optimizer == "sgd":
return SGD(model.parameters(), lr=learning_rate)
定义了一个神经网络模型 TorchModel，继承自 nn.Module，用于文本分类任务。模型包括嵌入层、线性层、平均池化层和分类层，使用 ReLU 激活函数和 Dropout 防止过拟合。前向传播根据输入返回预测值或损失值（若提供标签）。choose_optimizer 函数根据配置选择 Adam 或 SGD 优化器，并设置学习率。模型通过交叉熵损失进行训练。
主程序
main.py
# -*- coding: utf-8 -*-
import torch
import os
import random
import os
import numpy as np
import logging
from config import Config
from model import TorchModel, choose_optimizer
from evaluate import Evaluator
from loader import load_data, load_schema
logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
"""
模型训练主程序
"""
def main(config):
#创建保存模型的目录
if not os.path.isdir(config["model_path"]):
os.mkdir(config["model_path"])
#加载训练数据
train_data = load_data(config["train_data_path"], config)
#加载模型
model = TorchModel(config)
# 标识是否使用gpu
cuda_flag = torch.cuda.is_available()
if cuda_flag:
logger.info("gpu可以使用，迁移模型至gpu")
model = model.cuda()
#加载优化器
optimizer = choose_optimizer(config, model)
#加载效果测试类
evaluator = Evaluator(config, model, logger)
#训练
for epoch in range(config["epoch"]):
epoch += 1
model.train()
logger.info("epoch %d begin" % epoch)
train_loss = []
for index, batch_data in enumerate(train_data):
optimizer.zero_grad()
if cuda_flag:
batch_data = [d.cuda() for d in batch_data]
input_id, labels = batch_data #输入变化时这里需要修改，比如多输入，多输出的情况
loss = model(input_id, labels)
train_loss.append(loss.item())
if index % int(len(train_data) / 2) == 0:
logger.info("batch loss %f" % loss)
loss.backward()
# print(loss.item())
# print(model.classify.weight.grad)
optimizer.step()
logger.info("epoch average loss: %f" % np.mean(train_loss))
evaluator.eval(epoch)
model_path = os.path.join(config["model_path"], "epoch_%d.pth" % epoch)
torch.save(model.state_dict(), model_path)
return model, train_data
def ask(model, question):
input_id = train_data.dataset.encode_sentence(question)
model.eval()
model = model.cpu()
cls = torch.argmax(model(torch.LongTensor([input_id])))
schemes = load_schema(Config["schema_path"])
ans = ""
for name, val in schemes.items():
if val == cls:
ans = name
return ans
if __name__ == "__main__":
model, train_data = main(Config)
print(ask(model, "积分是怎么积的"))
while True:
question = input("请输入问题：")
res = ask(model, question)
print("命中问题：", res)
print("-----------")
实现一个基于 PyTorch 的文本分类模型的训练和推理过程。首先，通过 main 函数创建模型训练的主流程。代码首先检查是否有 GPU 可用，并将模型迁移至 GPU（如果可用）。然后加载训练数据、模型、优化器以及效果评估类。训练过程中，模型使用交叉熵损失函数计算训练误差并进行反向传播更新参数，每个 epoch 后记录并输出平均损失。同时，训练结束后，将模型保存至指定路径。
在训练完成后，ask 函数用于推理，输入问题并通过模型进行预测。它首先将输入问题转化为模型所需的格式，然后利用训练好的模型进行分类，最后返回匹配的答案。整个程序支持通过命令行输入问题，模型根据训练结果给出对应的答案。
在主程序中，首先进行一次初始化训练，之后进入循环，可以持续输入问题并得到模型的预测答案。
测试与评估
evaluate.py
# -*- coding: utf-8 -*-
import torch
from loader import load_data
"""
模型效果测试
"""
class Evaluator:
def __init__(self, config, model, logger):
self.config = config
self.model = model
self.logger = logger
self.valid_data = load_data(config["valid_data_path"], config, shuffle=False)
self.stats_dict = {"correct":0, "wrong":0} #用于存储测试结果
def eval(self, epoch):
self.logger.info("开始测试第%d轮模型效果：" % epoch)
self.stats_dict = {"correct":0, "wrong":0} #清空前一轮的测试结果
self.model.eval()
for index, batch_data in enumerate(self.valid_data):
if torch.cuda.is_available():
batch_data = [d.cuda() for d in batch_data]
input_id, labels = batch_data #输入变化时这里需要修改，比如多输入，多输出的情况
with torch.no_grad():
pred_results = self.model(input_id) #不输入labels，使用模型当前参数进行预测
self.write_stats(labels, pred_results)
self.show_stats()
return
def write_stats(self, labels, pred_results):
assert len(labels) == len(pred_results)
for true_label, pred_label in zip(labels, pred_results):
pred_label = torch.argmax(pred_label)
if int(true_label) == int(pred_label):
self.stats_dict["correct"] += 1
else:
self.stats_dict["wrong"] += 1
return
def show_stats(self):
correct = self.stats_dict["correct"]
wrong = self.stats_dict["wrong"]
self.logger.info("预测集合条目总量：%d" % (correct +wrong))
self.logger.info("预测正确条目：%d，预测错误条目：%d" % (correct, wrong))
self.logger.info("预测准确率：%f" % (correct / (correct + wrong)))
self.logger.info("--------------------")
return
定义一个 Evaluator 类，用于评估深度学习模型在验证集上的表现。Evaluator 初始化时接受配置文件、模型和日志记录器，并加载验证数据。eval 方法用于进行模型评估，在每轮评估开始时清空统计信息，设置模型为评估模式，然后通过遍历验证数据集进行预测。预测结果通过 write_stats 方法与真实标签进行比对，统计正确和错误的预测条目。最后，show_stats 方法输出总预测条目数、正确条目数、错误条目数以及准确率。该类的作用是帮助监控模型在验证集上的性能，便于调整和优化模型。
测试结果
请输入问题：在官网上如何修改移动密码
命中问题： 移动密码修改
-----------
请输入问题：我想多加一个号码作为亲情号
命中问题： 亲情号码设置与修改
-----------
请输入问题：我已经交足了话费请立即帮我开机
命中问题： 话费查询
-----------
请输入问题：密码想换一下
命中问题： 密码修改
