细粒度情感分类，这个是通过python pytorch实现的一个细粒度情感分类

from config import * import torch import torch.utils.data as data from transformers import BertTokenizer import pandas as pd import random # 获得实体位置索引 def get_ent_pos(lst): items = [] for i in range(len(lst)): # B-ASP 开头对应 id=1 # I-ASP 对应 id=2 if lst[i] == 1: item = [i] while True: i += 1 # 一直到 I-ASP 结束 if i >= len(lst) or lst[i] != 2: items.append(item) break else: item.append(i) i += 1 return items # [CLS]这个手机外观时尚，美中不足的是拍照像素低。[SEP] # 0 0 0 0 0 1 2 0 0 0 0 0 0 0 0 0 1 2 2 2 0 0 0 # 输出结果: [[5, 6], [16, 17, 18, 19]] # print(get_ent_pos([0, 0, 0, 0, 0, 1, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 2, 2, 2, 0, 0, 0])) # exit() def get_ent_weight(max_len, ent_pos): cdm = [] cdw = [] for i in range(max_len): dst = min(abs(i - ent_pos[0]), abs(i - ent_pos[-1])) if dst <= SRD: cdm.append(1) cdw.append(1) else: cdm.append(0) cdw.append(1 - (dst - SRD + 1) / max_len) return cdm, cdw # print(get_ent_weight(23, [5, 6])) # exit() class Dataset(data.Dataset): def __init__(self, type="train"): super().__init__() file_path = TRAIN_FILE_PATH if type == "train" else TEST_FILE_PATH self.df = pd.read_csv(file_path) self.tokenizer = BertTokenizer.from_pretrained(BERT_MODEL_NAME) def __len__(self): # 合并句子，当前句子与下一个句子合并，总样本数量 -1 return len(self.df) - 1 def __getitem__(self, index): # 相邻两个句子拼接 text1, bio1, pola1 = self.df.loc[index] text2, bio2, pola2 = self.df.loc[index + 1] # 注意拼接符号中间的空格 " ; " " 0 " " -1 "，与原格式匹配 text = text1 + " ; " + text2 bio = bio1 + " O " + bio2 pola = pola1 + " -1 " + pola2 # print(text) # print(bio) # print(pola) # exit() # 按自己的规则分词 ***触类旁通*** # eg: "英 寸 液 晶 屏 显 示 效 果 出 色 ; forever 清 晰 度 高" # bert 会把 forever 标记为 "[UNK]"，但是情感分类的时候，会把它索引成好多值 tokens = ["[CLS]"] + text.split(" ") + ["[SEP]"] input_ids = self.tokenizer.convert_tokens_to_ids(tokens) # BIO 标签转 id bio_arr = ["O"] + bio.split(" ") + ["O"] bio_label = [BIO_MAP[l] for l in bio_arr] # 情感值转数字 pola_arr = ["-1"] + pola.split(" ") + ["-1"] # 同样功能 pola_label = [int(i) for i in pola_arr] pola_label = list(map(int, pola_arr)) # print(input_ids) # print(bio_label) # print(pola_label) # exit() return input_ids, bio_label, pola_label def collate_fn(self, batch): # 统计最大句子长度 batch.sort(key=lambda x: len(x[0]), reverse=True) max_len = len(batch[0][0]) # 变量初始化 batch_input_ids = [] batch_bio_label = [] batch_mask = [] batch_ent_cdm = [] batch_ent_cdw = [] batch_pola_label = [] batch_pairs = [] for input_ids, bio_label, pola_label in batch: # 获取实体位置，没有实体跳过 # ent_pos 形状 [[实体1索引], [实体2索引],,,] # eg：[[3, 4, 5], [13, 14, 15]] ent_pos = get_ent_pos(bio_label) if len(ent_pos) == 0: continue # 填充句子长度 pad_len = max_len - len(input_ids) batch_input_ids.append(input_ids + [BERT_PAD_ID] * pad_len) batch_mask.append([1] * len(input_ids) + [0] * pad_len) batch_bio_label.append(bio_label + [BIO_O_ID] * pad_len) # 实体和情感分类对应 pairs = [] for pos in ent_pos: # 此时 pola = 0 or 1 pola = pola_label[pos[0]] # 异常值替换，下面相同 pola = 0 if pola == -1 else pola pairs.append((pos, pola)) batch_pairs.append(pairs) # print(batch_pairs) # exit() # 随机取一个实体 sg_ent_pos = random.choice(ent_pos) cdm, cdw = get_ent_weight(max_len, sg_ent_pos) # print(sg_ent_pos) # print(cdm, cdw) # exit() # 计算加权参数 batch_ent_cdm.append(cdm) batch_ent_cdw.append(cdw) # 实体第一个字的情感极性，不出意外的话，pola 为 0 or 1 pola = pola_label[sg_ent_pos[0]] # 排除意外情况，总共有三种选择: 0 1 -1 pola = 0 if pola == -1 else pola batch_pola_label.append(pola) # print("input_ids", input_ids) # print("bio_label", bio_label) # print("pola_label", pola_label) # print("ent_pos", ent_pos) # print("pairs", pairs) # print("batch_pairs", batch_pairs) # print("ent_pos", ent_pos) # print("sg_ent_pos", sg_ent_pos) # print("cdm", cdm) # print("cdw", cdw) # print("pola", pola) # print("batch_pola_label", batch_pola_label) # exit() return ( torch.tensor(batch_input_ids), torch.tensor(batch_mask).bool(), torch.tensor(batch_bio_label), # *** 这三个对应同一个实体 *** torch.tensor(batch_ent_cdm), torch.tensor(batch_ent_cdw), torch.tensor(batch_pola_label), # *** End *** batch_pairs, ) def get_pola(model, input_ids, mask, ent_label): # 变量初始化 b_input_ids = [] b_mask = [] b_ent_cdm = [] b_ent_cdw = [] b_ent_pos = [] # 根据label解析实体位置 ent_pos = get_ent_pos(ent_label) n = len(ent_pos) if n == 0: return None, None # n个实体一起预测，同一个句子复制n份，作为一个batch # 虽然 input_ids 只是一个句子，但是这个句子可能包含很多实体 # 不同句子对应不同的实体，当前句子的情感极性就表示对应实体的情感 b_input_ids.extend([input_ids] * n) b_mask.extend([mask] * n) b_ent_pos.extend(ent_pos) for sg_ent_pos in ent_pos: cdm, cdw = get_ent_weight(len(input_ids), sg_ent_pos) b_ent_cdm.append(cdm) b_ent_cdw.append(cdw) # 列表转 tensor，注意 torch.stack() 与 torch.cat() 之间的联系 b_input_ids = torch.stack(b_input_ids, dim=0).to(DEVICE) b_mask = torch.stack(b_mask, dim=0).to(DEVICE) b_ent_cdm = torch.tensor(b_ent_cdm).to(DEVICE) b_ent_cdw = torch.tensor(b_ent_cdw).to(DEVICE) b_ent_pola = model.get_pola(b_input_ids, b_mask, b_ent_cdm, b_ent_cdw) return b_ent_pos, b_ent_pola if __name__ == "__main__": dataset = Dataset() loader = data.DataLoader(dataset, batch_size=2, collate_fn=dataset.collate_fn) print(iter(loader).__next__())