PyTorch深度学习框架60天进阶学习计划 - 第22天：命名实体识别实战
PyTorch深度学习框架60天进阶学习计划 - 第22天：命名实体识别实战
使用BiLSTM-CRF实现医疗文本实体抽取
在医疗领域，命名实体识别(NER)是信息抽取的重要基础任务，可以从非结构化的医疗文本中识别出药物、疾病、症状、治疗方法等关键实体信息。今天，我们将学习如何使用BiLSTM-CRF模型实现医疗文本的实体抽取，深入理解CRF层转移矩阵的学习机制，以及分析维特比解码算法的动态规划实现。
目录
- 命名实体识别基础与医疗NER特点
- BiLSTM-CRF模型架构解析
- CRF层与转移矩阵学习
- 维特比解码算法详解
- 实战：使用PyTorch实现医疗NER模型
- 模型训练与评估
- 实验结果分析与优化方向
1. 命名实体识别基础与医疗NER特点
命名实体识别(Named Entity Recognition, NER)是从文本中识别和抽取特定类型实体的任务。在医疗领域，NER具有以下特点：
- 专业术语多：医疗文本包含大量专业术语，如疾病名称、药物名称等
- 缩写和简写常见：如"T2DM"代表"2型糖尿病"
- 实体边界模糊：医疗实体可能包含多个词，边界判定困难
- 上下文依赖强：同一词汇在不同上下文中可能表示不同实体类型
常见的医疗命名实体类型包括：
| 实体类型 | 描述 | 示例 |
|---|---|---|
| 疾病(Disease) | 各类疾病名称 | 2型糖尿病、高血压、肺炎 |
| 症状(Symptom) | 病症描述 | 头痛、发热、恶心 |
| 药物(Drug) | 药品名称 | 阿司匹林、胰岛素、青霉素 |
| 治疗(Treatment) | 治疗方式 | 手术、放化疗、物理治疗 |
| 检查(Test) | 检查项目 | 血常规、CT、核磁共振 |
| 解剖部位(Body) | 人体部位 | 肝脏、肺部、心脏 |
2. BiLSTM-CRF模型架构解析
BiLSTM-CRF是目前NER任务中最常用的模型之一，它结合了双向LSTM捕获上下文信息的能力和CRF捕获标签间依赖关系的优势。
模型架构如下：
-
嵌入层(Embedding Layer)：将输入文本转换为密集向量表示
- 词嵌入(Word Embedding)
- 字符级嵌入(Character Embedding)(可选)
-
BiLSTM层：捕获序列上下文信息
- 前向LSTM：从左到右处理序列
- 后向LSTM：从右到左处理序列
- 合并两个方向的隐藏状态
-
CRF层：建模标签之间的依赖关系，确保预测的标签序列符合约束条件
BiLSTM-CRF相比单纯的BiLSTM模型的优势在于，CRF层能够考虑整个标签序列的合法性，避免出现非法的标签转移（如"I-Disease"出现在"O"之后而不是"B-Disease"之后）。
3. CRF层与转移矩阵学习
条件随机场(Conditional Random Field, CRF)是一种判别式模型，用于标注和分割序列数据。在NER任务中，CRF可以学习标签之间的转移概率，确保预测结果符合标签规则。
转移矩阵学习
CRF层的核心是转移矩阵(Transition Matrix)，记为A，其中A[i,j]表示从标签i转移到标签j的分数(score)。
假设我们有标签集合Y，对于给定的输入序列x和可能的标签序列y，CRF层计算分数：
s c o r e ( x , y ) = ∑ i = 0 n A y i − 1 , y i + ∑ i = 1 n P i , y i score(x,y) = \sum_{i=0}^{n} A_{y_{i-1},y_i} + \sum_{i=1}^{n} P_{i,y_i} score(x,y)=i=0∑nAyi−1,yi+i=1∑nPi,yi
其中：
- n是序列长度
- A_{y_{i-1},y_i}是从标签y_{i-1}转移到标签y_i的分数
- P_{i,y_i}是位置i的词被赋予标签y_i的分数(由BiLSTM输出)
转移矩阵是模型训练过程中学习的参数。训练目标是最大化正确标签序列的概率：
P ( y ∣ x ) = e s c o r e ( x , y ) ∑ y ′ ∈ Y n e s c o r e ( x , y ′ ) P(y|x) = \frac{e^{score(x,y)}}{\sum_{y' \in Y^n} e^{score(x,y')}} P(y∣x)=∑y′∈Ynescore(x,y′)escore(x,y)
训练时，我们最小化负对数似然：
L = − log P ( y ∣ x ) = − s c o r e ( x , y ) + log ∑ y ′ ∈ Y n e s c o r e ( x , y ′ ) L = -\log P(y|x) = -score(x,y) + \log \sum_{y' \in Y^n} e^{score(x,y')} L=−logP(y∣x)=−score(x,y)+logy′∈Yn∑escore(x,y′)
4. 维特比解码算法详解
在推理阶段，给定一个输入序列，我们需要找到分数最高的标签序列。然而，可能的标签序列数量随序列长度指数增长，无法枚举所有可能性。
维特比(Viterbi)算法是一种动态规划算法，可以高效地找到最优标签序列：
- 初始化：设置起始状态的分数
- 递推：对于序列中的每个位置，计算到达每个可能标签的最高分数路径
- 终止：找到最后一个位置的最高分数及对应标签
- 回溯：从终止状态回溯，构建完整的最优标签序列
算法的核心是维护一个动态规划表dp，其中dp[i,j]表示前i个位置，以标签j结尾的最高分数。递推公式为：
d p [ i , j ] = max k ∈ Y ( d p [ i − 1 , k ] + A k , j + P i , j ) dp[i,j] = \max_{k \in Y} (dp[i-1,k] + A_{k,j} + P_{i,j}) dp[i,j]=k∈Ymax(dp[i−1,k]+Ak,j+Pi,j)
其中Y是所有可能的标签集合。
回溯过程则从最后一个位置的最优标签开始，根据记录的来源标签，逐步向前构建完整的标签序列。
5. 实战：使用PyTorch实现医疗NER模型
下面我们将使用PyTorch实现一个完整的BiLSTM-CRF模型用于医疗NER任务。
数据预处理
首先，定义数据处理代码：
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import numpy as np
from sklearn.metrics import classification_report
import matplotlib.pyplot as plt
import pandas as pd
import re
# 1. 数据预处理
class MedicalNERDataset(Dataset):
def __init__(self, data_path, vocab=None, tag_to_idx=None, max_len=100):
"""
初始化医疗NER数据集
Args:
data_path: 数据文件路径
vocab: 词表，用于将词转换为ID
tag_to_idx: 标签到索引的映射
max_len: 序列最大长度
"""
self.data = self.load_data(data_path)
self.max_len = max_len
# 构建词表和标签集合
if vocab is None:
self.vocab = self.build_vocab(self.data)
self.word_to_idx = {word: idx for idx, word in enumerate(self.vocab)}
else:
self.vocab = vocab
self.word_to_idx = {word: idx for idx, word in enumerate(self.vocab)}
if tag_to_idx is None:
tags = set()
for _, tags_list in self.data:
for tag in tags_list:
tags.add(tag)
self.tag_to_idx = {tag: idx for idx, tag in enumerate(sorted(list(tags)))}
else:
self.tag_to_idx = tag_to_idx
self.idx_to_tag = {idx: tag for tag, idx in self.tag_to_idx.items()}
def load_data(self, data_path):
"""加载BIO格式的NER数据"""
data = []
sentence, tags = [], []
with open(data_path, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if not line: # 空行表示句子结束
if sentence:
data.append((sentence, tags))
sentence, tags = [], []
else:
parts = line.split()
if len(parts) >= 2: # 确保至少有词和标签
word, tag = parts[0], parts[-1]
sentence.append(word)
tags.append(tag)
# 处理最后一个句子
if sentence:
data.append((sentence, tags))
return data
def build_vocab(self, data):
"""构建词表"""
vocab = set()
for sentence, _ in data:
for word in sentence:
vocab.add(word)
# 添加特殊标记
vocab = sorted(list(vocab))
vocab = ['<PAD>', '<UNK>'] + vocab
return vocab
def __len__(self):
return len(self.data)
def __getitem__(self, idx):
"""获取转换后的数据项"""
words, tags = self.data[idx]
# 截断或填充到固定长度
if len(words) > self.max_len:
words = words[:self.max_len]
tags = tags[:self.max_len]
# 将词和标签转换为ID
word_ids = [self.word_to_idx.get(word, 1) for word in words] # 1 是 <UNK> 的索引
tag_ids = [self.tag_to_idx[tag] for tag in tags]
# 填充
padding_length = self.max_len - len(words)
word_ids = word_ids + [0] * padding_length # 0 是 <PAD> 的索引
tag_ids = tag_ids + [0] * padding_length # 假设PAD对应的标签索引为0
# 创建mask
mask = [1] * len(words) + [0] * padding_length
return {
'word_ids': torch.LongTensor(word_ids),
'tag_ids': torch.LongTensor(tag_ids),
'mask': torch.ByteTensor(mask),
'seq_len': len(words)
}
# 2. 构建BiLSTM-CRF模型
class BiLSTM_CRF(nn.Module):
def __init__(self, vocab_size, tag_to_idx, embedding_dim=100, hidden_dim=200, dropout=0.5):
"""
初始化BiLSTM-CRF模型
Args:
vocab_size: 词表大小
tag_to_idx: 标签到索引的映射
embedding_dim: 词嵌入维度
hidden_dim: LSTM隐藏层维度
dropout: Dropout概率
"""
super(BiLSTM_CRF, self).__init__()
self.vocab_size = vocab_size
self.tag_to_idx = tag_to_idx
self.embedding_dim = embedding_dim
self.hidden_dim = hidden_dim
self.tagset_size = len(tag_to_idx)
# 词嵌入层
self.word_embeds = nn.Embedding(vocab_size, embedding_dim, padding_idx=0)
self.dropout = nn.Dropout(dropout)
# BiLSTM层
self.lstm = nn.LSTM(embedding_dim, hidden_dim // 2,
num_layers=1, bidirectional=True, batch_first=True)
# 全连接层，映射到标签空间
self.hidden2tag = nn.Linear(hidden_dim, self.tagset_size)
# CRF层的转移矩阵：形状为 [tagset_size, tagset_size]
# A[i,j]表示从标签i转移到标签j的得分
self.transitions = nn.Parameter(torch.randn(self.tagset_size, self.tagset_size))
# 约束: 句子开始不能转移到非起始标签，句子结束不能从非结束标签转移
# 可以在训练中学习，也可以手动设置(这里我们初始化为较大的负值)
self.transitions.data[:, 0] = -10000.0 # 任何标签都不能转移到PAD
self.transitions.data[0, :] = -10000.0 # PAD不能转移到任何标签
def _get_lstm_features(self, sentence, mask):
"""获取BiLSTM的输出特征"""
embeds = self.word_embeds(sentence) # [batch_size, seq_len, embedding_dim]
embeds = self.dropout(embeds)
# 应用mask来处理变长序列
packed_input = torch.nn.utils.rnn.pack_padded_sequence(
embeds, mask.sum(1).int().cpu(), batch_first=True, enforce_sorted=False
)
packed_output, _ = self.lstm(packed_input)
lstm_out, _ = torch.nn.utils.rnn.pad_packed_sequence(packed_output, batch_first=True)
lstm_feats = self.hidden2tag(lstm_out) # [batch_size, seq_len, tagset_size]
return lstm_feats
def _forward_alg(self, feats, mask):
"""前向算法计算所有可能的标签序列的总分数(Z(x))"""
batch_size = feats.size(0)
seq_len = feats.size(1)
tag_size = feats.size(2)
mask = mask.float()
# 在位置0，PAD的发射概率应该为0，其他标签为-10000
# 从而保证任何路径都是以START_TAG开始
init_alphas = torch.full((batch_size, self.tagset_size), -10000.0).to(feats.device)
init_alphas[:, 0] = 0.
# forward_var[i]表示标签i的前向变量
forward_var = init_alphas
# 遍历句子中的每个词
for t in range(seq_len):
# 获取当前时间步的活跃状态(未被掩码)
active_mask = mask[:, t].view(-1, 1).repeat(1, tag_size)
# 计算从任意标签到当前标签的所有可能路径分数
# emit_score: [batch_size, tag_size]
emit_score = feats[:, t]
# 计算转移分数
# forward_var的形状是[batch_size, tag_size]
# self.transitions的形状是[tag_size, tag_size]
# forward_var.unsqueeze(2): [batch_size, tag_size, 1]
# self.transitions.unsqueeze(0): [1, tag_size, tag_size]
# next_tag_var: [batch_size, tag_size, tag_size]
next_tag_var = forward_var.unsqueeze(2) + self.transitions.unsqueeze(0)
# 对第二个维度求log-sum-exp，得到每个标签的前向变量
# 形状为[batch_size, tag_size]
forward_var_t = torch.logsumexp(next_tag_var, dim=1)
# 将发射分数加入
forward_var_t = forward_var_t + emit_score
# 对于被掩码的部分，保持原值
forward_var = torch.where(active_mask.bool(), forward_var_t, forward_var)
# 对最后一个有效时间步求log-sum-exp，得到整个序列的分数
# 这里简化处理，忽略了转移到STOP_TAG的分数
forward_var = torch.logsumexp(forward_var, dim=1)
return forward_var
def _score_sentence(self, feats, tags, mask):
"""计算给定标签序列的分数"""
batch_size = feats.size(0)
seq_len = feats.size(1)
mask = mask.float()
# 累积分数初始化为0
score = torch.zeros(batch_size).to(feats.device)
# 获取第一个标签的发射分数
score = score + feats[:, 0, tags[:, 0]]
# 遍历句子中的每个词(从第二个开始)
for t in range(1, seq_len):
# 累加转移分数和发射分数
active_mask = mask[:, t]
# 只对有效的位置累加分数
# 转移分数: self.transitions[tags[:, t-1], tags[:, t]]
# 发射分数: feats[:, t, tags[:, t]]
transition_score = self.transitions[tags[:, t-1], tags[:, t]] * active_mask
emission_score = feats[:, t, tags[:, t]] * active_mask
score = score + transition_score + emission_score
return score
def _viterbi_decode(self, feats, mask):
"""使用维特比算法解码最优标签序列"""
batch_size = feats.size(0)
seq_len = feats.size(1)
tag_size = feats.size(2)
# 保存回溯路径
backpointers = torch.zeros((batch_size, seq_len, tag_size), dtype=torch.long).to(feats.device)
# 初始化维特比变量
viterbi_var = torch.full((batch_size, tag_size), -10000.0).to(feats.device)
viterbi_var[:, 0] = 0 # 从PAD开始
# 遍历句子
for t in range(seq_len):
# 创建一个临时变量保存当前时间步的维特比变量
forward_var_t = viterbi_var.unsqueeze(2).repeat(1, 1, tag_size)
transition_t = self.transitions.unsqueeze(0).repeat(batch_size, 1, 1)
# 计算当前时间步的所有可能转移
forward_var_t = forward_var_t + transition_t
# 找出对每个目标标签的最高分数和路径
best_tag_id = torch.argmax(forward_var_t, dim=1)
best_path_score = torch.max(forward_var_t, dim=1)[0]
# 更新维特比变量
viterbi_var = best_path_score + feats[:, t]
# 记录回溯路径
backpointers[:, t, :] = best_tag_id
# 获取最高分数和对应的结束标签
best_score, best_tag_id = torch.max(viterbi_var, dim=1)
# 使用回溯指针解码最优路径
best_paths = torch.zeros((batch_size, seq_len), dtype=torch.long).to(feats.device)
# 对每个样本进行回溯
for b in range(batch_size):
best_tag = best_tag_id[b]
best_paths[b, -1] = best_tag
# 从后向前回溯
for t in range(seq_len - 1, 0, -1):
best_tag = backpointers[b, t, best_tag]
best_paths[b, t-1] = best_tag
return best_paths
def neg_log_likelihood(self, sentence, tags, mask):
"""计算负对数似然损失"""
feats = self._get_lstm_features(sentence, mask)
forward_score = self._forward_alg(feats, mask)
gold_score = self._score_sentence(feats, tags, mask)
return torch.mean(forward_score - gold_score)
def forward(self, sentence, mask):
"""前向传播，用于预测"""
# 获取BiLSTM特征
lstm_feats = self._get_lstm_features(sentence, mask)
# 使用维特比算法解码
tag_seq = self._viterbi_decode(lstm_feats, mask)
return tag_seq
# 3. 辅助函数
def train_epoch(model, data_loader, optimizer, device):
"""训练一个epoch"""
model.train()
total_loss = 0
for batch in data_loader:
# 将数据移到设备
word_ids = batch['word_ids'].to(device)
tag_ids = batch['tag_ids'].to(device)
mask = batch['mask'].to(device)
# 前向计算
model.zero_grad()
loss = model.neg_log_likelihood(word_ids, tag_ids, mask)
# 反向传播
loss.backward()
optimizer.step()
total_loss += loss.item()
return total_loss / len(data_loader)
def evaluate(model, data_loader, tag_to_idx, device):
"""评估模型"""
model.eval()
idx_to_tag = {idx: tag for tag, idx in tag_to_idx.items()}
true_tags_list = []
pred_tags_list = []
with torch.no_grad():
for batch in data_loader:
# 将数据移到设备
word_ids = batch['word_ids'].to(device)
tag_ids = batch['tag_ids'].to(device)
mask = batch['mask'].to(device)
seq_lens = batch['seq_len']
# 预测
pred_ids = model(word_ids, mask)
# 根据实际序列长度处理标签
for i, seq_len in enumerate(seq_lens):
true_tags = [idx_to_tag[tag.item()] for tag in tag_ids[i][:seq_len]]
pred_tags = [idx_to_tag[tag.item()] for tag in pred_ids[i][:seq_len]]
true_tags_list.extend(true_tags)
pred_tags_list.extend(pred_tags)
# 使用sklearn计算指标
report = classification_report(true_tags_list, pred_tags_list)
return report, true_tags_list, pred_tags_list
def predict_entities(sentence, model, word_to_idx, idx_to_tag, device, max_len=100):
"""预测文本中的实体"""
# 文本处理
words = list(sentence)
if len(words) > max_len:
words = words[:max_len]
# 转换为ID
word_ids = [word_to_idx.get(word, 1) for word in words] # 1是<UNK>的索引
padding_length = max_len - len(words)
word_ids = word_ids + [0] * padding_length # 0是<PAD>的索引
mask = [1] * len(words) + [0] * padding_length
# 转换为tensor
word_ids = torch.LongTensor(word_ids).unsqueeze(0).to(device)
mask = torch.ByteTensor(mask).unsqueeze(0).to(device)
# 预测
model.eval()
with torch.no_grad():
pred_ids = model(word_ids, mask)
# 转换为标签
pred_tags = [idx_to_tag[tag.item()] for tag in pred_ids[0][:len(words)]]
# 提取实体
entities = []
entity = {"type": "", "text": "", "start": 0}
for i, (word, tag) in enumerate(zip(words, pred_tags)):
if tag.startswith('B-'):
if entity["text"]:
entities.append(entity.copy())
entity = {"type": tag[2:], "text": word, "start": i}
elif tag.startswith('I-') and entity["text"] and tag[2:] == entity["type"]:
entity["text"] += word
elif tag == 'O':
if entity["text"]:
entities.append(entity.copy())
entity = {"type": "", "text": "", "start": 0}
# 添加最后一个实体
if entity["text"]:
entities.append(entity)
return entities
# 4. 示例医疗文本数据生成（用于演示）
def create_sample_data(file_path):
"""创建示例医疗NER训练数据"""
sentences = [
"患者 诊断 为 2型 糖尿病 ， 并 出现 蛋白尿 症状 。",
"医生 建议 服用 二甲 双 胍 控制 血糖 。",
"患者 肝功能 检查 结果 显示 转氨酶 升高 。",
"该 患者 有 高血压 病史 ， 目前 正在 服用 硝苯地平 。",
"病人 出现 头痛 发热 等 流感 症状 。",
"核磁共振 检查 显示 左肺 有 小 结节 。",
"患者 近期 出现 心悸 胸闷 ， 考虑 冠心病 可能 。",
"医生 给 病人 开具 了 阿司匹林 处方 。",
"化验 结果 显示 血压 偏高 ， 建议 调整 用药 。",
"患者 患有 帕金森病 ， 表现 为 肢体 震颤 。"
]
# BIO标注
bio_tags = [
["O", "O", "O", "B-Disease", "I-Disease", "O", "O", "O", "B-Symptom", "O", "O"],
["O", "O", "O", "B-Drug", "I-Drug", "I-Drug", "O", "B-Test", "O"],
["O", "B-Test", "O", "O", "O", "B-Test", "B-Symptom", "O"],
["O", "O", "O", "B-Disease", "O", "O", "O", "O", "O", "B-Drug", "O"],
["O", "O", "O", "B-Symptom", "B-Symptom", "O", "B-Disease", "O", "O"],
["B-Test", "O", "O", "B-Body", "O", "O", "B-Disease", "O"],
["O", "O", "O", "B-Symptom", "B-Symptom", "O", "O", "B-Disease", "O", "O"],
["O", "O", "O", "O", "O", "B-Drug", "O", "O"],
["B-Test", "O", "O", "B-Symptom", "B-Symptom", "O", "O", "O", "O", "O"],
["O", "O", "B-Disease", "O", "O", "O", "B-Symptom", "I-Symptom", "O"]
]
# 写入文件
with open(file_path, 'w', encoding='utf-8') as f:
for sentence, tags in zip(sentences, bio_tags):
words = sentence.split()
for word, tag in zip(words, tags):
f.write(f"{word} {tag}\n")
f.write("\n") # 句子之间用空行分隔
# 5. 主程序：训练和评估模型
def main():
# 创建示例数据（实际应用中，应该使用真实医疗语料）
train_file = "medical_ner_train.txt"
test_file = "medical_ner_test.txt"
create_sample_data(train_file)
create_sample_data(test_file) # 实际应用中，测试集应该与训练集不同
# 超参数设置
batch_size = 32
embedding_dim = 100
hidden_dim = 200
epochs = 10
learning_rate = 0.001
# 创建数据集
train_dataset = MedicalNERDataset(train_file)
word_to_idx = train_dataset.word_to_idx
tag_to_idx = train_dataset.tag_to_idx
test_dataset = MedicalNERDataset(test_file,
vocab=train_dataset.vocab,
tag_to_idx=train_dataset.tag_to_idx)
# 创建数据加载器
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=batch_size)
# 设置设备
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# 初始化模型
model = BiLSTM_CRF(len(train_dataset.vocab), tag_to_idx,
embedding_dim=embedding_dim,
hidden_dim=hidden_dim)
model.to(device)
# 优化器
optimizer = optim.Adam(model.parameters(), lr=learning_rate)
# 损失和准确率记录
losses = []
# 训练循环
print("开始训练...")
for epoch in range(epochs):
# 训练一个epoch
train_loss = train_epoch(model, train_loader, optimizer, device)
losses.append(train_loss)
print(f"Epoch {epoch+1}/{epochs}, Loss: {train_loss:.4f}")
# 每两个epoch评估一次
if (epoch + 1) % 2 == 0:
report, _, _ = evaluate(model, test_loader, tag_to_idx, device)
print(f"Evaluation Report:\n{report}")
# 保存模型
torch.save({
'model_state_dict': model.state_dict(),
'word_to_idx': word_to_idx,
'tag_to_idx': tag_to_idx,
'idx_to_tag': {idx: tag for tag, idx in tag_to_idx.items()}
}, "medical_ner_model.pth")
# 绘制损失曲线
plt.figure(figsize=(10, 6))
plt.plot(range(1, epochs + 1), losses)
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.title('Training Loss Curve')
plt.grid(True)
plt.savefig('training_loss.png')
plt.close()
# 模型测试
print("\n模型测试:")
# 加载模型（实际应用中，应该从保存的模型文件加载）
idx_to_tag = {idx: tag for tag, idx in tag_to_idx.items()}
# 测试文本
test_sentences = [
"患者出现持续性头痛和发热，被诊断为流感。",
"医生建议每日口服阿司匹林和布洛芬来缓解症状。",
"肝功能检查显示转氨酶轻度升高，考虑脂肪肝。"
]
for sentence in test_sentences:
entities = predict_entities(sentence, model, word_to_idx, idx_to_tag, device)
print(f"\n原文: {sentence}")
print("识别实体:")
for entity in entities:
print(f" - 类型: {entity['type']}, 实体: {entity['text']}")
# 可视化CRF转移矩阵
visualize_transitions(model, tag_to_idx)
return model, word_to_idx, tag_to_idx
def visualize_transitions(model, tag_to_idx):
"""可视化CRF转移矩阵"""
# 获取转移矩阵
transitions = model.transitions.detach().cpu().numpy()
# 获取标签列表
tags = sorted(tag_to_idx.keys(), key=lambda x: tag_to_idx[x])
# 创建DataFrame
df_transitions = pd.DataFrame(transitions, index=tags, columns=tags)
# 绘制热力图
plt.figure(figsize=(12, 10))
plt.imshow(transitions, cmap='Blues')
plt.colorbar()
plt.xticks(range(len(tags)), tags, rotation=45)
plt.yticks(range(len(tags)), tags)
plt.xlabel('转移到')
plt.ylabel('转移自')
plt.title('CRF转移矩阵热力图')
# 标注数值
for i in range(len(tags)):
for j in range(len(tags)):
text = plt.text(j, i, f'{transitions[i, j]:.1f}',
ha="center", va="center", color="white" if transitions[i, j] < 0 else "black")
plt.tight_layout()
plt.savefig('transition_matrix.png')
plt.close()
# 在实际环境中运行
if __name__ == "__main__":
main()
6. 模型训练与评估
现在我们已经实现了完整的BiLSTM-CRF模型用于医疗NER任务。下面让我们来解析模型训练和评估的关键环节：
训练过程
训练BiLSTM-CRF模型的核心是计算和优化负对数似然损失：
-
前向传播：
- 通过BiLSTM获取每个位置的特征表示
- 计算当前标签序列的分数（路径分数）
- 计算所有可能标签序列的总分数（归一化因子）
-
损失计算：
- 损失 = -（当前标签序列分数 - 所有可能标签序列的对数和）
- 直观理解：最大化正确标签序列的概率
-
反向传播：
- 计算梯度并更新模型参数，包括：
- BiLSTM的权重
- 词嵌入层参数
- CRF层的转移矩阵
- 计算梯度并更新模型参数，包括：
CRF层转移矩阵的学习
CRF层的转移矩阵是模型的关键组成部分，它捕获了标签之间的依赖关系。在医疗NER任务中，这一点尤为重要，例如：
- “I-Disease”（疾病内部）标签通常只会出现在"B-Disease"（疾病开始）之后
- “B-Symptom”（症状开始）不太可能紧跟在"I-Drug"（药物内部）之后
转移矩阵的学习过程：
- 初始化：随机初始化转移矩阵参数，并设置一些约束（如不允许转移到PAD标签）
- 训练：在训练过程中，通过最大化正确标签序列的概率，模型逐渐学习到合理的转移概率
- 优化：转移矩阵参数通过梯度下降方法进行更新
下面是一个直观理解转移矩阵的示例：
| 从\到 | O | B-Disease | I-Disease | B-Symptom | I-Symptom |
|---|---|---|---|---|---|
| O | 高 | 高 | 低 | 高 | 低 |
| B-Disease | 中 | 低 | 高 | 低 | 低 |
| I-Disease | 高 | 中 | 高 | 低 | 低 |
| B-Symptom | 中 | 低 | 低 | 低 | 高 |
| I-Symptom | 高 | 低 | 低 | 中 | 高 |
这个矩阵表明，例如从"B-Disease"（疾病开始）到"I-Disease"（疾病内部）的转移概率很高，而从"B-Disease"到"I-Symptom"的转移概率很低。
维特比解码算法
在推理阶段，我们使用维特比算法找到最可能的标签序列。这是一个动态规划算法，可以通过以下步骤直观理解：
- 初始化：设置起始状态的分数
- 递推：对于序列中的每个位置t，每个可能的标签y：
- 计算从前一个位置的所有可能标签转移到当前标签y的分数
- 选择最大分数路径并记录来源标签
- 终止：找到最后一个位置的最高分数及对应标签
- 回溯：从终止状态回溯，构建完整的最优标签序列
维特比算法的工作流程如下图所示：
O B-Disease I-Disease B-Symptom I-Symptom
位置1 s1,1 s1,2 s1,3 s1,4 s1,5
/|\ /|\ /|\ /|\ /|\
/ | \ / | \ / | \ / | \ / | \
位置2 s2,1 s2,2 s2,3 s2,4 s2,5 ...
/|\ ...
/ | \
位置3 s3,1 ...
其中s_t,i表示位置t的标签i的最高分数路径。箭头表示可能的转移，我们只保留得分最高的一条。
7. 实验结果分析与优化方向
结果分析
当我们运行上述代码，训练医疗NER模型后，可以分析以下几个方面的结果：
-
训练损失曲线：
- 理想情况下，损失应该随着训练进行而下降，最终趋于稳定
- 如果损失不降反升，可能是学习率过大或训练不稳定
-
实体识别性能：
- 实体级别的评估指标：精确率(Precision)、召回率(Recall)、F1值
- 不同实体类型的识别效果（医疗NER中，药物名称通常比症状更容易识别）
-
标签转移矩阵分析：
- 学习到的转移矩阵可视化，观察标签之间学习到的依赖关系
- 验证是否符合医疗文本的语言学规律
下面是一个假设的实验结果示例表：
| 实体类型 | 精确率 | 召回率 | F1值 |
|---|---|---|---|
| Disease | 0.85 | 0.82 | 0.83 |
| Symptom | 0.78 | 0.74 | 0.76 |
| Drug | 0.92 | 0.89 | 0.91 |
| Test | 0.81 | 0.79 | 0.80 |
| Body | 0.84 | 0.77 | 0.80 |
| 总体 | 0.84 | 0.80 | 0.82 |
从结果可以看出，药物(Drug)实体的识别效果最好，这可能是因为药物名称通常有固定格式和明确边界。而症状(Symptom)实体的识别效果较差，这可能是因为症状描述通常更加多样化和主观。
模型优化方向
针对医疗NER任务，可以从以下几个方向优化BiLSTM-CRF模型：
-
增强特征表示：
- 添加字符级嵌入，捕获形态学信息
- 整合医疗词典特征，利用先验知识
- 使用预训练的医疗领域词向量或语言模型（如医疗BERT）
-
模型架构优化：
- 增加BiLSTM层数，提高模型容量
- 尝试使用Transformer替代BiLSTM，获取更好的上下文表示
- 加入注意力机制，关注关键词汇
-
数据增强：
- 扩充医疗实体词典
- 使用同义词替换生成更多训练样本
- 利用规则模板生成合成训练数据
-
后处理规则：
- 加入医疗实体字典匹配
- 设计特定规则处理边界模糊的实体
- 实体合并和消歧（相同实体多次出现）
-
领域适应：
- 针对不同医疗子领域（如心脏病学、肿瘤学）进行模型微调
- 结合医疗本体，建立实体间关系知识
代码运行流程图
下面是我们实现的医疗NER系统的完整运行流程图：
┌────────────────┐ ┌────────────────┐ ┌────────────────┐
│ │ │ │ │ │
│ 数据预处理 │──────▶ 模型构建 │──────▶ 模型训练 │
│ │ │ │ │ │
└────────────────┘ └────────────────┘ └────────────────┘
│ │
│ │
▼ ▼
┌────────────────┐ ┌────────────────┐ ┌────────────────┐
│ │ │ │ │ │
│ 模型评估 │◀─────┤ 维特比解码 │◀─────┤ 实体抽取 │
│ │ │ │ │ │
└────────────────┘ └────────────────┘ └────────────────┘
详细流程说明：
-
数据预处理：
- 加载BIO标注的医疗NER数据
- 构建词表和标签集
- 数据转换和批处理
-
模型构建：
- 词嵌入层：将词转换为密集向量
- BiLSTM层：捕获上下文信息
- CRF层：建模标签依赖关系
-
模型训练：
- 计算负对数似然损失
- 优化模型参数
- 学习CRF转移矩阵
-
维特比解码：
- 前向计算得到发射概率
- 使用维特比算法寻找最优标签序列
- 回溯解码完整路径
-
实体抽取：
- 根据BIO标签序列提取实体
- 确定实体类型和边界
- 实体整合
-
模型评估：
- 计算实体级别评估指标
- 分析不同类型实体的识别效果
- 可视化转移矩阵
CRF层转移矩阵可视化示例
CRF层的转移矩阵可视化可以帮助我们理解标签之间的依赖关系。下图是一个假设的医疗NER任务中CRF转移矩阵的热力图示例：
O B-Disease I-Disease B-Symptom I-Symptom B-Drug I-Drug ...
O [高] [高] [低] [高] [低] [高] [低]
B-Disease[中] [低] [高] [低] [低] [低] [低]
I-Disease[高] [中] [高] [低] [低] [低] [低]
B-Symptom[中] [低] [低] [低] [高] [低] [低]
I-Symptom[高] [低] [低] [中] [高] [低] [低]
B-Drug [中] [低] [低] [低] [低] [低] [高]
I-Drug [高] [低] [低] [低] [低] [中] [高]
...
颜色越深表示转移概率越高，越浅表示转移概率越低。通过这个热力图，我们可以直观地看到：
- 从"B-"标签（实体开始）到相应"I-"标签（实体内部）的转移概率较高
- 从"O"（非实体）到各个"B-"标签（新实体开始）的转移概率较高
- 从"I-"标签（实体内部）到不同类型"B-"标签的转移概率较低
总结
通过今天的学习，我们掌握了：
-
医疗命名实体识别任务的特点与挑战：
- 医疗术语专业性强、缩写多、边界模糊
- 实体类型丰富，包括疾病、症状、药物、治疗、检查等
-
BiLSTM-CRF模型的工作原理：
- 词嵌入层：将词转换为向量表示
- BiLSTM层：捕获序列上下文信息
- CRF层：建模标签间依赖关系
-
CRF层转移矩阵学习机制：
- 转移矩阵表示标签间转移概率
- 通过最大化正确标签序列概率学习
- 约束非法转移的发生
-
维特比解码算法：
- 动态规划求解最优标签序列
- 前向计算最高分数路径
- 回溯重建完整标签序列
-
PyTorch实现完整BiLSTM-CRF模型：
- 灵活的数据处理流程
- 高效的模型训练与评估
- 实体抽取与结果可视化
医疗NER是医疗信息抽取的基础任务，掌握这一技术可以帮助我们从大量非结构化医疗文本中自动提取有价值的信息，为医疗知识图谱构建、临床决策支持等下游任务提供支持。
清华大学全五版的《DeepSeek教程》完整的文档需要的朋友，关注我私信：deepseek 即可获得。
怎么样今天的内容还满意吗？再次感谢朋友们的观看，关注GZH：凡人的AI工具箱，回复666，送您价值199的AI大礼包。最后，祝您早日实现财务自由，还请给个赞，谢谢！