深度学习笔记（7）文本标注与NER

文章目录

深度学习笔记（7）文本标注与NER
一、文本标注
- 1.1文本标注工具doccano
- 1.2 标注处理，bio标注
二、训练模型
- 1.引入库
- 2. 定义数据集
- 3.建模
- 4，模型训练
- 5.评估
- 6.训练
三.测试DEMO

一、文本标注

1.1文本标注工具doccano

https://github.com/doccano/doccano

pip install doccano

# Initialize database.
doccano init
# Create a super user.
doccano createuser --username admin --password pass
# Start a web server.
doccano webserver --port 8000

另外开启一个终端输入（注意，是在python3.9以上的虚拟环境中）

doccano task

然后浏览器输入http://127.0.0.1:8000/zh
在这里插入图片描述
点创建，然后界面如下

因为我们要做的是NER，所以使用序列标注。然后选数据集，进行导入的操作，

因为我测试的数据集是一行一个文本，所以选第二个

导入后如下
在这里插入图片描述
然后创建标签，试验创建了年，考试和地点三个标签

因为这里可能很多人来标记，所以在指南那里设定个标记规则

标记完成如上图

标记完成后导出数据集到自己项目的路径就可以了。
这个格式和我们处理的格式不一样，这里做一个格式的转换

1.2 标注处理，bio标注

什么是bio标注呢，举个例子

B (Begin): 表示实体的开始。
I (Inside): 表示实体的中间部分。
O (Outside): 表示单词不属于任何实体。
李雷昨天去了北京的天安门广场。
李雷：B-PER（人名的开始）
昨天：O（不涉及实体）
去了：O（不涉及实体）
北京：B-LOC（地理位置的开始）
的：I-LOC（地理位置的中间）
天安门：I-LOC（地理位置的中间）
广场：E-LOC（地理位置的结束）

那么刚才的例子，exam place year 还有o 是7分类为什么是7分类
bio分类使用下面的脚本

import json


# 定义 get_token 函数
def get_token(input):
    # english = 'abcdefghijklmnopqrstuvwxyz0123456789'
    english = 'abcdefghijklmnopqrstuvwxyz'
    output = []
    buffer = ''
    for s in input:
        if s in english or s in english.upper():
            buffer += s
        else:
            if buffer: output.append(buffer)
            buffer = ''
            output.append(s)
    if buffer: output.append(buffer)
    return output


# 定义 json2bio 函数
def json2bio(jsonl_path, bio_path):
    with open(jsonl_path, 'r', encoding='utf-8') as f:
        lines = f.readlines()
        for i, line in enumerate(lines):
            annotations = json.loads(line)
            text = annotations['text']
            labels = annotations['label']

            if not isinstance(labels, list):
                print(f"Warning: 'label' key is not a list in annotations: {annotations}")
                continue

            all_words = get_token(text)
            all_label = ['O'] * len(all_words)
            for label in labels:
                b_location = label[0]
                e_location = label[1]
                label_type = label[2]
                all_label[b_location] = 'B-' + label_type
                if b_location != e_location:
                    for word in range(b_location + 1, e_location):
                        all_label[word] = 'I-' + label_type

            with open(bio_path, 'a', encoding='utf-8') as f:
                for word, label in zip(all_words, all_label):
                    f.write(word + ' ' + label + '\n')
                f.write('\n')


# 调用函数时确保传递正确的文件路径
json2bio('admin.jsonl', 'output.txt')

输出的txt文件如图
在这里插入图片描述

二、训练模型

1.引入库

from pathlib import Path
import re
import numpy as np
data_dir = 'H:\\develop\\NLP\\ner\\data'

这里用\，因为\n是转义符，可能会有问题

def read_data(file_path):    #结果：WindowsPath('H:/develop/NLP/ner/data/train.txt')
    # 将传入的字符串路径转换为 Path 对象
    file_path = Path(file_path)

    # 使用 Path 对象的 read_text 方法读取文件内容，并指定编码为 UTF-8。
    raw_text = file_path.read_text(encoding='UTF-8').strip()#结果
    # 原始文本中可能包含多个文档，每个文档之间由 \n\t?\n 分隔
    raw_docs = re.split(r'\n\t?\n', raw_text)
    # 原始文档列表，每个文档包含多个句子
    token_docs = []
    tag_docs = []
    # 遍历每个文档
    for doc in raw_docs:
        tokens = []  # 存储当前文档中的所有单词
        tags = []  # 存储当前文档中的所有标签
        # 遍历文档中的每个句子
        for line in doc.split('\n'):
            # 每行包含一个单词和一个标签，通过空格分割
            token, tag = line.split(' ')
            tokens.append(token)
            tags.append(tag)
        # 将处理后的单词和标签添加到相应的列表中
        token_docs.append(tokens)
        tag_docs.append(tags)

    # 返回处理后的单词列表和标签列表
    return token_docs, tag_docs

train_texts, train_tags = read_data(data_dir + '\\train.txt')
# 读取数据目录中的 train.txt 文件，并获取其中的文本内容和标签
test_texts, test_tags = read_data(data_dir + '\\val.txt')
# 读取数据目录中的 val.txt 文件，并获取其中的文本内容和标签
val_texts, val_tags = read_data(data_dir + '\\test.txt')
# 读取数据目录中的 test.txt 文件，并获取其中的文本内容和标签

# 获取训练数据中所有标签的集合
unique_tags = set(tag for doc in train_tags for tag in doc)
# 创建一个字典，将标签映射到唯一的 ID
tag2id = {tag: id for id, tag in enumerate(unique_tags)}
# 创建一个字典，将 ID 映射回对应的标签
id2tag = {id: tag for tag, id in tag2id.items()}
# 获取标签列表
label_list = list(unique_tags)

# 导入 transformers 库，并创建一个 BertTokenizerFast 实例
from transformers import AutoTokenizer, BertTokenizerFast
tokenizer = BertTokenizerFast.from_pretrained('bert-base-chinese')
# 使用 tokenizer 对训练文本进行编码，参数包括：
# is_split_into_words=True: 表示文本已经分词好了，因为中文本来就是分好词的了，所以不用重新分词，只需要转成对应id就可以了
# return_offsets_mapping=True: 表示返回文本的偏移映射
# padding=True: 表示对文本进行填充，以适应最大长度
# truncation=True: 表示对文本进行截断，以适应最大长度
# max_length=512: 表示文本的最大长度为 512
train_encodings = tokenizer(train_texts, is_split_into_words=True, return_offsets_mapping=True, padding=True, truncation=True, max_length=512)
# 对验证文本进行相同的编码操作
val_encodings = tokenizer(val_texts, is_split_into_words=True, return_offsets_mapping=True, padding=True, truncation=True, max_length=512)

offsets_mapping 是一个非常重要的属性。offsets_mapping 是一个字典，它包含了原始文本中每个 token 在原始文本中的位置信息。

def encode_tags(tags, encodings):
    # 标签列表，每个标签列表对应一个文档
    labels = [[tag2id[tag] for tag in doc] for doc in tags]
    # 打印标签以供调试
    #print(labels)
    # 编码的标签列表，每个文档对应一个编码后的标签列表
    encoded_labels = []
    # 遍历标签列表和编码的 offset_mapping
    for doc_labels, doc_offset in zip(labels, encodings.offset_mapping):
        # 创建一个全由-100组成的矩阵，其中-100表示未知或不可预测的标签
        doc_enc_labels = np.ones(len(doc_offset),dtype=int) * -100
        # 创建一个数组，包含 offset_mapping 中的每个元素
        arr_offset = np.array(doc_offset)
        # 设置标签，其第一个 offset 位置为 0 且第二个 offset 位置不为 0
        # 这意味着这些标签位于原始文本的开头，但不在 BERT 模型的输入范围内
        if len(doc_labels) >= 510:#防止异常
            doc_labels = doc_labels[:510]
        doc_enc_labels[(arr_offset[:,0] == 0) & (arr_offset[:,1] != 0)] = doc_labels
        # 将编码后的标签列表添加到 encoded_labels 列表中
        encoded_labels.append(doc_enc_labels.tolist())

    # 返回编码后的标签列表
    return encoded_labels

这段代码的主要目的是将原始的标签列表转换为适合 BERT 模型输入的格式。它遍历每个文档的标签列表，并将每个标签映射到对应的 ID。然后，它创建一个全由 -100 组成的矩阵，其中 -100 表示未知或不可预测的标签。最后，它根据 offset_mapping 设置那些在原始文本开头但不在 BERT 模型输入范围内的标签。

2. 定义数据集

class NerDataset(torch.utils.data.Dataset):
    # 定义一个自定义数据集类，用于命名实体识别（NER）任务
    def __init__(self, encodings, labels):
        # 初始化方法，接收两个参数：encodings 和 labels
        # encodings 是从 tokenizer 生成的字典，包含输入序列的各种特征
        # labels 是原始的标签列表，每个标签列表对应一个文档
        self.encodings = encodings
        self.labels = labels

    def __getitem__(self, idx):
        # 定义一个特殊方法，用于从数据集中获取单个样本
        # 它接受一个索引 idx，并返回一个字典，包含编码后的特征和标签
        item = {key: torch.tensor(val[idx]) for key, val in self.encodings.items()}
        item['labels'] = torch.tensor(self.labels[idx])
        return item

    def __len__(self):
        # 定义一个特殊方法，用于返回数据集中样本的数量
        # 在这个类中，样本的数量与 labels 的长度相同
        return len(self.labels)

3.建模

train_encodings.pop("offset_mapping") # 训练不需要这个
# 从 train_encodings 中移除 "offset_mapping" 键，因为训练过程中不需要这个信息
val_encodings.pop("offset_mapping")
# 从 val_encodings 中移除 "offset_mapping" 键，因为训练过程中不需要这个信息

train_dataset = NerDataset(train_encodings, train_labels)
# 创建一个 NerDataset 实例，用于训练数据
# train_encodings 是包含训练数据特征的字典
# train_labels 是包含训练数据标签的列表

val_dataset = NerDataset(val_encodings, val_labels)
# 创建一个 NerDataset 实例，用于验证数据
# val_encodings 是包含验证数据特征的字典
# val_labels 是包含验证数据标签的列表

offset_mapping 是其中一个字典，它包含了原始文本中每个 token 在原始文本中的位置信息。这对于将模型预测的标签映射回原始文本中的实体位置至关重要。然而，在训练过程中，通常不需要这个信息，因为它会增加模型的输入大小，从而可能减慢训练速度。

因此，您在训练数据集（train_encodings 和 val_encodings）中移除了 offset_mapping 键，以减少模型的输入大小。这样做不会影响模型的训练，因为模型在训练过程中只关注输入特征（如 input_ids、attention_mask）和标签（labels），而不是 offset_mapping。

在验证和测试阶段，offset_mapping 是非常有用的，因为它可以帮助您将模型的预测结果映射回原始文本中的位置，以便评估模型的性能。因此，您没有在验证和测试数据集（val_dataset 和 test_dataset）中移除 offset_mapping。

4，模型训练

from transformers import AutoModelForTokenClassification, TrainingArguments, Trainer
# 从 transformers 库中导入 AutoModelForTokenClassification、TrainingArguments 和 Trainer 类

model = AutoModelForTokenClassification.from_pretrained('ckiplab/albert-base-chinese-ner',
                                                        num_labels=7,
                                                        ignore_mismatched_sizes=True,
                                                        id2label=id2tag,
                                                        label2id=tag2id
                                                        )
# 初始化一个 AutoModelForTokenClassification 实例，它是一个预训练的模型，用于 NER 任务
# 'ckiplab/albert-base-chinese-ner' 是模型的名称，它是基于 ALBERT 模型预训练的中文 NER 模型
# num_labels=7 表示模型将有 7 个输出标签，这与您之前提到的 7 分类相匹配
# ignore_mismatched_sizes=True 表示当模型输入的大小与训练数据不匹配时，忽略错误
# id2label 和 label2id 是字典，分别将标签 ID 映射回标签名称，以及将标签名称映射回标签 ID
# 这些字典是您之前在处理数据时创建的，用于将模型的输出标签 ID 映射回原始标签

print(model)

ignore_mismatched_sizes=True 这段比较重要，因为我是7分类，和之前的原始模型不一样，所以要忽略不一样的，只取一样的操作。

5.评估

评估方法一般用这个模型内置的就可以了，这里要把-100的去掉，就是那些填充的

from datasets import load_metric
metric = load_metric("seqeval")

import numpy as np

def compute_metrics(p):
    predictions, labels = p
    predictions = np.argmax(predictions, axis=2)

    # 不要管-100那些，剔除掉
    true_predictions = [
        [label_list[p] for (p, l) in zip(prediction, label) if l != -100]
        for prediction, label in zip(predictions, labels)
    ]
    true_labels = [
        [label_list[l] for (p, l) in zip(prediction, label) if l != -100]
        for prediction, label in zip(predictions, labels)
    ]

    results = metric.compute(predictions=true_predictions, references=true_labels)
    return {
        "precision": results["overall_precision"],
        "recall": results["overall_recall"],
        "f1": results["overall_f1"],
        "accuracy": results["overall_accuracy"],
    }

6.训练

checkpoint = 'bert-base-chinese'
# 指定用于训练的预训练模型检查点，这里是 BERT 的基础中文模型
num_train_epochs = 1000
# 设置训练过程中完整地通过训练数据集的次数
per_device_train_batch_size=8
# 每个 GPU 上的训练批次大小
per_device_eval_batch_size=8
# 每个 GPU 上的评估批次大小

training_args = TrainingArguments(
    output_dir='./output',          # 输出目录，训练和验证的结果将保存在这里
    num_train_epochs=num_train_epochs,              # 训练epoch数量
    per_device_train_batch_size=per_device_train_batch_size,  # 每个GPU的BATCH
    per_device_eval_batch_size=per_device_eval_batch_size,
    warmup_steps=500,                # warmup次数，用于调整学习率
    weight_decay=0.01,               # 限制权重的大小
    logging_dir='./logs',
    # 设置日志文件的目录，训练过程中的日志将被保存在这里
    logging_steps=10,
    # 设置日志的间隔，即每10个训练步骤打印一次日志
    save_strategy='steps',
    # 指定何时保存模型检查点，'steps' 表示按指定的步数保存
    save_steps=1000,
    # 设置保存模型检查点的间隔，即每1000个训练步骤保存一次
    save_total_limit=1,
    # 设置最多保存的模型检查点数量，这里是1，表示只保留最新的模型
    evaluation_strategy='steps',
    # 指定何时评估模型，'steps' 表示按指定的步数评估
    eval_steps=1000,
    # 设置评估的间隔，即每1000个训练步骤评估一次
)

trainer = Trainer(
    model=model,
    # 指定要训练的模型
    args=training_args,
    # 指定训练参数
    train_dataset=train_dataset,
    # 指定训练数据集
    eval_dataset=val_dataset,
    # 指定评估数据集
    compute_metrics=compute_metrics
    # 指定用于计算评估指标的函数
)

trainer.train()

# 调用 Trainer 对象的 train 方法来开始模型的训练过程
# 这将使用之前定义的训练数据集和训练参数来训练模型

trainer.evaluate()
# 调用 Trainer 对象的 evaluate 方法来评估模型的性能
# 这将使用之前定义的评估数据集来评估模型的性能

model.save_pretrained("./checkpoint/model/%s-%sepoch" % (checkpoint, num_train_epochs))
# 调用模型的 save_pretrained 方法来保存模型
# 参数是一个字符串，表示保存模型的路径和名称
# 在这个例子中，模型将被保存到 './checkpoint/model/' 目录下
# 文件名格式为 '%s-%sepoch'，其中 '%s' 被 'bert-base-chinese' 替换
# '%sepoch' 被 '1000epoch' 替换，因为 num_train_epochs 的值为 1000

运行结束后pytorch_model.bin就是模型

三.测试DEMO

import torch
import numpy as np


def get_token(input):
    # english = 'abcdefghijklmnopqrstuvwxyz0123456789'
    english = 'abcdefghijklmnopqrstuvwxyz'
    output = []
    buffer = ''
    for s in input:
        if s in english or s in english.upper():
            buffer += s
        else:
            if buffer: output.append(buffer)
            buffer = ''
            output.append(s)
    if buffer: output.append(buffer)
    return output


from transformers import AutoModelForTokenClassification, TrainingArguments, Trainer

model = AutoModelForTokenClassification.from_pretrained('./output/checkpoint-2000')
print(model)
print(model.config.id2label)

from transformers import BertTokenizerFast

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

上面是一个分词器

if __name__ == '__main__':
    # 定义输入文本
    input_str = '2022研究生考试时间是什么时候？'
    # 使用 get_token 函数将输入文本转换为字符列表
    input_char = get_token(input_str)
    # 使用 tokenizer 对输入文本进行编码
    input_tensor = tokenizer(input_char, is_split_into_words=True, padding=True, truncation=True,
                             return_offsets_mapping=True, max_length=512, return_tensors="pt")
    # 获取编码后的 tokens
    input_tokens = input_tensor.tokens()
    # 获取编码中的 offset_mapping
    offsets = input_tensor["offset_mapping"]
    # 创建一个忽略 mask，用于忽略不在原始文本中的 token
    ignore_mask = offsets[0, :, 1] == 0
    # 移除编码中的 offset_mapping 信息，因为模型训练时不需要这个信息
    input_tensor.pop("offset_mapping")
    # 获取模型的输出
    outputs = model(**input_tensor)
    # 获取预测概率和预测标签
    probabilities = torch.nn.functional.softmax(outputs.logits, dim=-1)[0].tolist()
    predictions = outputs.logits.argmax(dim=-1)[0].tolist()
    # 打印预测标签
    print(predictions)
    # 初始化结果列表
    results = []

    # 遍历预测标签
    tokens = input_tensor.tokens()
    idx = 0
    while idx < len(predictions):
        # 如果当前 token是特殊字符，跳过
        if ignore_mask[idx]:
            idx += 1
            continue
        # 获取当前预测标签
        pred = predictions[idx]
        # 获取当前预测标签的名称
        label = model.config.id2label[pred]#比如B-year
        # 如果标签不是 O，则处理实体
        if label != "O":
            # 打印的时候移除 B- 或 I- 前缀
            label = label[2:]
            # 获取实体在原始文本中的开始和结束位置
            start = idx
            end = start + 1
            # 获取所有 token I-label
            all_scores = []
            all_scores.append(probabilities[start][predictions[start]])
            while (
                    end < len(predictions)
                    and model.config.id2label[predictions[end]] == f"I-{label}"
            ):
                all_scores.append(probabilities[end][predictions[end]])
                end += 1
                idx += 1
            # 计算所有 token I-label 的平均概率 比如天安门三个字天80%安70%门90%那概率就是（80+70+90）/3
            score = np.mean(all_scores).item()
            # 获取实体在原始文本中的单词
            word = input_tokens[start:end]
            # 添加结果到结果列表
            results.append(
                {
                    "entity_group": label,
                    "score": score,
                    "word": word,
                    "start": start,
                    "end": end,
                }
            )
        idx += 1
    # 遍历结果列表并打印
    for i in range(len(results)):
        print(results[i])

    #print(results)

{'entity_group': 'year', 'score': 0.9746097803115845, 'word': ['2', '0', '0', '9', '年'], 'start': 1, 'end': 6}
{'entity_group': 'exam', 'score': 0.9772411584854126, 'word': ['高', '考'], 'start': 6, 'end': 8}
{'entity_group': 'place', 'score': 0.9826750457286835, 'word': ['北', '京'], 'start': 9, 'end': 11}
{'entity_group': 'place', 'score': 0.41835154096285504, 'word': ['0', '0', '9'], 'start': 17, 'end': 20}