這篇文章寫得很冗余，但是我相信你如果真的看完，并且按照我的代碼和邏輯進行分析，對您以后的資料預處理和命名物體識別都有幫助，只有真正對這些復雜的文本進行NLP處理后，您才能適應更多的真實環境，堅持！畢竟我寫的時候也看了20多小時的視頻，又寫了20多個小時，別抱怨，加油~
上一篇文章處理后的資料格式如下圖所示，將一個個句子處理成了包含六元組的CSV檔案，這篇文章將介紹BiLSTM-CRF模型搭建及訓練、預測，最終實作醫學命名物體識別，
整個專案工程如下圖所示：
本專欄主要結合作者之前的博客、AI經驗和相關視頻及論文介紹，后面隨著深入會講解更多的Python人工智能案例及應用，基礎性文章，希望對您有所幫助，如果文章中存在錯誤或不足之處，還請海涵~作者作為人工智能的菜鳥，希望大家能與我在這一筆一劃的博客中成長起來，寫了這么多年博客，嘗試第一個付費專欄，但更多博客尤其基礎性文章，還是會繼續免費分享，但該專欄也會用心撰寫，望對得起讀者，共勉！
讀者也可以從github下載原始碼，結合原始碼來運行最終實驗，祝好~
- Keras下載地址：https://github.com/eastmountyxz/AI-for-Keras
- TensorFlow下載地址：https://github.com/eastmountyxz/AI-for-TensorFlow
文章目錄
- 一.生成映射字典
- 二.資料增強
- 三.資料準備
- 四.模型構建
- 1.BiLSTM模型構建
- 2.CRF模型融合
- 3.初始化函式完善
- 4.模型訓練
- 五.模型預測
- 1.輸出訓練誤差
- 2.預測資料
- 六.完整代碼
- 1.model.py
- 2.train.py
- 3.data_utils.py
- 4.prepare_data.py
- 5.data_process.py
- 七.總結
前文：
[Python人工智能] 一.TensorFlow2.0環境搭建及神經網路入門
[Python人工智能] 二.TensorFlow基礎及一元直線預測案例
[Python人工智能] 三.TensorFlow基礎之Session、變數、傳入值和激勵函式
[Python人工智能] 四.TensorFlow創建回歸神經網路及Optimizer優化器
[Python人工智能] 五.Tensorboard可視化基本用法及繪制整個神經網路
[Python人工智能] 六.TensorFlow實作分類學習及MNIST手寫體識別案例
[Python人工智能] 七.什么是過擬合及dropout解決神經網路中的過擬合問題
[Python人工智能] 八.卷積神經網路CNN原理詳解及TensorFlow撰寫CNN
[Python人工智能] 九.gensim詞向量Word2Vec安裝及《慶余年》中文短文本相似度計算
[Python人工智能] 十.Tensorflow+Opencv實作CNN自定義影像分類案例及與機器學習KNN影像分類演算法對比
[Python人工智能] 十一.Tensorflow如何保存神經網路引數
[Python人工智能] 十二.回圈神經網路RNN和LSTM原理詳解及TensorFlow撰寫RNN分類案例
[Python人工智能] 十三.如何評價神經網路、loss曲線圖繪制、影像分類案例的F值計算
[Python人工智能] 十四.回圈神經網路LSTM RNN回歸案例之sin曲線預測
[Python人工智能] 十五.無監督學習Autoencoder原理及聚類可視化案例詳解
[Python人工智能] 十六.Keras環境搭建、入門基礎及回歸神經網路案例
[Python人工智能] 十七.Keras搭建分類神經網路及MNIST數字影像案例分析
[Python人工智能] 十八.Keras搭建卷積神經網路及CNN原理詳解
[Python人工智能] 十九.Keras搭建回圈神經網路分類案例及RNN原理詳解
[Python人工智能] 二十.基于Keras+RNN的文本分類vs基于傳統機器學習的文本分類
[Python人工智能] 二十一.Word2Vec+CNN中文文本分類詳解及與機器學習（RF\DTC\SVM\KNN\NB\LR）分類對比
[Python人工智能] 二十二.基于大連理工情感詞典的情感分析和情緒計算
[Python人工智能] 二十三.基于機器學習和TFIDF的情感分類（含詳細的NLP資料清洗）
[Python人工智能] 二十四.易學智能GPU搭建Keras環境實作LSTM惡意URL請求分類
[Python人工智能] 二十六.基于BiLSTM-CRF的醫學命名物體識別研究（上）資料預處理
[Python人工智能] 二十七.基于BiLSTM-CRF的醫學命名物體識別研究（下）模型構建
《人工智能狂潮》讀后感——什么是人工智能？(一)
一.生成映射字典
接下來需要將每個漢字、邊界、拼音、偏旁部首等映射成向量，所以，我們首先需要來構造字典，統計多少個不同的字、邊界、拼音、偏旁部首等，然后再構建模型將不同的漢字、拼音等映射成不同的向量，
在prepare_data.py中自定義函式get_dict()生成映射字典，
為了訓練時保證每個批次輸入樣本長度一致，這里補充了PAD標記變數，用于填充，同時，每個批次資料在進行填充時是以本批次中最長的句子作為標準，因此需要將句子按長度排序，每個批次資料的長度接近從而提升運算速度，
思考
在機器學習和深度學習中，測驗集很可能出現新的特征，這些特征在訓練集中從未出現過，比如該資料集的某個漢字、拼音或偏旁部首，在測驗集中很可能第一次出現，那么，這種情況怎么解決呢？這種未登錄詞可以設定為低頻Unknown，從而解決該問題，
此時的完整代碼如下所示：
- prepare_data.py
#encoding:utf-8
import os
import pandas as pd
from collections import Counter
from data_process import split_text
from tqdm import tqdm #進度條 pip install tqdm
#詞性標注
import jieba.posseg as psg
#獲取字的偏旁和拼音
from cnradical import Radical, RunOption
#洗掉目錄
import shutil
#隨機劃分訓練集和測驗集
from random import shuffle
#遍歷檔案包
from glob import glob
train_dir = "train_data"
#----------------------------功能:文本預處理---------------------------------
def process_text(idx, split_method=None, split_name='train'):
"""
功能: 讀取文本并切割,接著打上標記及提取詞邊界、詞性、偏旁部首、拼音等特征
param idx: 檔案的名字 不含擴展名
param split_method: 切割文本方法
param split_name: 存盤資料集 默認訓練集, 還有測驗集
return
"""
#定義字典 保存所有字的標記、邊界、詞性、偏旁部首、拼音等特征
data = {}
#--------------------------------------------------------------------
# 獲取句子
#--------------------------------------------------------------------
if split_method is None:
#未給文本分割函式 -> 讀取檔案
with open(f'data/{train_dir}/{idx}.txt', encoding='utf8') as f: #f表示檔案路徑
texts = f.readlines()
else:
#給出文本分割函式 -> 按函式分割
with open(f'data/{train_dir}/{idx}.txt', encoding='utf8') as f:
outfile = f'data/train_data_pro/{idx}_pro.txt'
print(outfile)
texts = f.read()
texts = split_method(texts, outfile)
#提取句子
data['word'] = texts
print(texts)
#--------------------------------------------------------------------
# 獲取標簽(物體類別、起始位置)
#--------------------------------------------------------------------
#初始時將所有漢字標記為O
tag_list = ['O' for s in texts for x in s] #雙層回圈遍歷每句話中的漢字
#讀取ANN檔案獲取每個物體的型別、起始位置和結束位置
tag = pd.read_csv(f'data/{train_dir}/{idx}.ann', header=None, sep='\t') #Pandas讀取 分隔符為tab鍵
#0 T1 Disease 1845 1850 1型糖尿病
for i in range(tag.shape[0]): #tag.shape[0]為行數
tag_item = tag.iloc[i][1].split(' ') #每一行的第二列 空格分割
#print(tag_item)
#存在某些物體包括兩段位置區間 僅獲取起始位置和結束位置
cls, start, end = tag_item[0], int(tag_item[1]), int(tag_item[-1])
#print(cls,start,end)
#對tag_list進行修改
tag_list[start] = 'B-' + cls
for j in range(start+1, end):
tag_list[j] = 'I-' + cls
#斷言 兩個長度不一致報錯
assert len([x for s in texts for x in s])==len(tag_list)
#print(len([x for s in texts for x in s]))
#print(len(tag_list))
#--------------------------------------------------------------------
# 分割后句子匹配標簽
#--------------------------------------------------------------------
tags = []
start = 0
end = 0
#遍歷文本
for s in texts:
length = len(s)
end += length
tags.append(tag_list[start:end])
start += length
print(len(tags))
#標簽資料存盤至字典中
data['label'] = tags
#--------------------------------------------------------------------
# 提取詞性和詞邊界
#--------------------------------------------------------------------
#初始標記為M
word_bounds = ['M' for item in tag_list] #邊界 M表示中間
word_flags = [] #詞性
#分詞
for text in texts:
#帶詞性的結巴分詞
for word, flag in psg.cut(text):
if len(word)==1: #1個長度詞
start = len(word_flags)
word_bounds[start] = 'S' #單個字
word_flags.append(flag)
else:
start = len(word_flags)
word_bounds[start] = 'B' #開始邊界
word_flags += [flag]*len(word) #保證詞性和字一一對應
end = len(word_flags) - 1
word_bounds[end] = 'E' #結束邊界
#存盤
bounds = []
flags = []
start = 0
end = 0
for s in texts:
length = len(s)
end += length
bounds.append(word_bounds[start:end])
flags.append(word_flags[start:end])
start += length
data['bound'] = bounds
data['flag'] = flags
#--------------------------------------------------------------------
# 獲取拼音和偏旁特征
#--------------------------------------------------------------------
radical = Radical(RunOption.Radical) #提取偏旁部首
pinyin = Radical(RunOption.Pinyin) #提取拼音
#提取拼音和偏旁 None用特殊符號替代UNK
radical_out = [[radical.trans_ch(x) if radical.trans_ch(x) is not None else 'UNK' for x in s] for s in texts]
pinyin_out = [[pinyin.trans_ch(x) if pinyin.trans_ch(x) is not None else 'UNK' for x in s] for s in texts]
#賦值
data['radical'] = radical_out
data['pinyin'] = pinyin_out
#--------------------------------------------------------------------
# 存盤資料
#--------------------------------------------------------------------
#獲取樣本數量
num_samples = len(texts) #行數
num_col = len(data.keys()) #列數 字典自定義類別數 6
print(num_samples)
print(num_col)
dataset = []
for i in range(num_samples):
records = list(zip(*[list(v[i]) for v in data.values()])) #壓縮
dataset += records+[['sep']*num_col] #每處理一句話sep分割
#records = list(zip(*[list(v[0]) for v in data.values()]))
#for r in records:
# print(r)
#最后一行sep洗掉
dataset = dataset[:-1]
#轉換成dataframe 增加表頭
dataset = pd.DataFrame(dataset,columns=data.keys())
#保存檔案 測驗集 訓練集
save_path = f'data/prepare/{split_name}/{idx}.csv'
dataset.to_csv(save_path,index=False,encoding='utf-8')
#--------------------------------------------------------------------
# 處理換行符 w表示一個字
#--------------------------------------------------------------------
def clean_word(w):
if w=='\n':
return 'LB'
if w in [' ','\t','\u2003']: #中文空格\u2003
return 'SPACE'
if w.isdigit(): #將所有數字轉換為一種符號 數字訓練會造成干擾
return 'NUM'
return w
#對dataframe應用函式
dataset['word'] = dataset['word'].apply(clean_word)
#存盤資料
dataset.to_csv(save_path,index=False,encoding='utf-8')
#return texts, tags, bounds, flags
#return texts[0], tags[0], bounds[0], flags[0], radical_out[0], pinyin_out[0]
#----------------------------功能:預處理所有文本---------------------------------
def multi_process(split_method=None,train_ratio=0.8):
"""
功能: 對所有文本盡心預處理操作
param split_method: 切割文本方法
param train_ratio: 訓練集和測驗集劃分比例
return
"""
#洗掉目錄
if os.path.exists('data/prepare/'):
shutil.rmtree('data/prepare/')
#創建目錄
if not os.path.exists('data/prepare/train/'):
os.makedirs('data/prepare/train/')
os.makedirs('data/prepare/test/')
#獲取所有檔案名
idxs = set([file.split('.')[0] for file in os.listdir('data/'+train_dir)])
idxs = list(idxs)
#隨機劃分訓練集和測驗集
shuffle(idxs) #打亂順序
index = int(len(idxs)*train_ratio) #獲取訓練集的截止下標
#獲取訓練集和測驗集檔案名集合
train_ids = idxs[:index]
test_ids = idxs[index:]
#--------------------------------------------------------------------
# 引入多行程
#--------------------------------------------------------------------
#執行緒池方式呼叫
import multiprocessing as mp
num_cpus = mp.cpu_count() #獲取機器CPU的個數
pool = mp.Pool(num_cpus)
results = []
#訓練集處理
for idx in train_ids:
result = pool.apply_async(process_text, args=(idx,split_method,'train'))
results.append(result)
#測驗集處理
for idx in test_ids:
result = pool.apply_async(process_text, args=(idx,split_method,'test'))
results.append(result)
#關閉行程池
pool.close()
pool.join()
[r.get for r in results]
#----------------------------功能:生成映射字典---------------------------------
#統計函式：串列、頻率計算閾值
def mapping(data,threshold=10,is_word=False,sep='sep'):
#統計串列data中各種型別的個數
count = Counter(data)
#洗掉之前自定義的sep換行符
if sep is not None:
count.pop(sep)
#判斷是漢字 未登錄詞處理 出現頻率較少 設定為Unknown
if is_word:
#設定下列兩個詞頻次 排序靠前
count['PAD'] = 100000001 #填充字符 保證長度一致
count['UNK'] = 100000000 #未知標記
#降序排列
data = sorted(count.items(),key=lambda x:x[1], reverse=True)
#去除頻率小于threshold的元素
data = [x[0] for x in data if x[1]>=threshold]
#轉換成字典
id2item = data
item2id = {id2item[i]:i for i in range(len(id2item))}
else:
count['PAD'] = 100000001
data = sorted(count.items(),key=lambda x:x[1], reverse=True)
data = [x[0] for x in data]
id2item = data
item2id = {id2item[i]:i for i in range(len(id2item))}
return id2item, item2id
#生成映射字典
def get_dict():
#獲取所有內容
all_w = [] #漢字
all_bound = [] #邊界
all_flag = [] #詞性
all_label = [] #類別
all_radical = [] #偏旁
all_pinyin = [] #拼音
#讀取檔案
for file in glob('data/prepare/train/*.csv') + glob('data/prepare/test/*.csv'):
df = pd.read_csv(file,sep=',')
all_w += df['word'].tolist()
all_bound += df['bound'].tolist()
all_flag += df['flag'].tolist()
all_label += df['label'].tolist()
all_radical += df['radical'].tolist()
all_pinyin += df['pinyin'].tolist()
#保存回傳結果 字典
map_dict = {}
#呼叫統計函式
map_dict['word'] = mapping(all_w,threshold=20,is_word=True)
map_dict['bound'] = mapping(all_bound)
map_dict['flag'] = mapping(all_flag)
map_dict['label'] = mapping(all_label)
map_dict['radical'] = mapping(all_radical)
map_dict['pinyin'] = mapping(all_pinyin)
#字典保存內容
return map_dict
#-------------------------------功能:主函式--------------------------------------
if __name__ == '__main__':
#print(process_text('0',split_method=split_text,split_name='train'))
#多執行緒處理文本
#multi_process(split_text)
#生成映射字典
print(get_dict())
輸出結果如下圖所示：
至此，成功輸出結果，包括字、邊界、標記、類別、偏旁、拼音六類資料及對應的下標，比如邊界共包括PAD、S、B、E、M五種，物體型別包括31種，
如果需要對生成的資料進行存盤和呼叫，則使用如下核心代碼：
輸出結果為：
- ([‘PAD’, ‘S’, ‘B’, ‘E’, ‘M’], {‘PAD’: 0, ‘S’: 1, ‘B’: 2, ‘E’: 3, ‘M’: 4})
二.資料增強
接下來我們需要將這些下標轉換成對應的數值，再映射成向量，模型根據向量進行訓練，
第一步，創建檔案data_utils.py，
- data_utils.py
我們將檔案中的三個句子合并成一個句子，從而實作資料增強，同時，拼接檔案前獲取漢字、邊界、詞性、類別、偏旁、拼音對應的下標，再進行后續句子拼接操作，注意，這里的三個句子拼接在一定程度能讓整個文本保持一個均勻的長度，從而分批訓練的詞向量長度一致，增強資料并提升運算性能，
第二步，撰寫相關代碼，
#encoding:utf-8
import pandas as pd
import pickle
import numpy as np
from tqdm import tqdm
import os
#功能:獲取值對應的下標 引數為串列和字符
def item2id(data,w2i):
#x在字典中直接獲取 不在字典中回傳UNK
return [w2i[x] if x in w2i else w2i['UNK'] for x in data]
#----------------------------功能:拼接檔案---------------------------------
def get_data_with_windows(name='train'):
#讀取prepare_data.py生成的dict.pkl檔案 存盤字典{類別:下標}
with open(f'data/dict.pkl', 'rb') as f:
map_dict = pickle.load(f) #加載字典
#存盤所有資料
results = []
root = os.path.join('data/prepare/'+name)
files = list(os.listdir(root))
print(files)
#['10.csv', '11.csv', '12.csv',.....]
#獲取所有檔案 進度條
for file in tqdm(files):
all_data = []
path = os.path.join(root, file)
samples = pd.read_csv(path,sep=',')
max_num = len(samples)
#獲取sep換行分隔符下標 -1 20 40 60
sep_index = [-1]+samples[samples['word']=='sep'].index.tolist()+[max_num]
#print(sep_index)
#[-1, 83, 92, 117, 134, 158, 173, 200,......]
#----------------------------------------------------------------------
# 獲取句子并將句子全部都轉換成id
#----------------------------------------------------------------------
for i in range(len(sep_index)-1):
start = sep_index[i] + 1 #0 (-1+1)
end = sep_index[i+1] #20
data = []
#每個特征進行處理
for feature in samples.columns: #訪問每列
#通過函式item2id獲取下標 map_dict兩個值(串列和字典) 獲取第二個值
data.append(item2id(list(samples[feature])[start:end],map_dict[feature][1]))
#將每句話的串列合成
all_data.append(data)
#----------------------------------------------------------------------
# 資料增強
#----------------------------------------------------------------------
#前后兩個句子拼接 每個句子六個元素(漢字、邊界、詞性、類別、偏旁、拼音)
two = []
for i in range(len(all_data)-1):
first = all_data[i]
second = all_data[i+1]
two.append([first[k]+second[k] for k in range(len(first))]) #六個元素
three = []
for i in range(len(all_data)-2):
first = all_data[i]
second = all_data[i+1]
third = all_data[i+2]
three.append([first[k]+second[k]+third[k] for k in range(len(first))])
#回傳所有結果
results.extend(all_data+two+three)
return results
#-------------------------------功能:主函式--------------------------------------
if __name__ == '__main__':
print(get_data_with_windows('train'))
此時的輸出如下圖所示，可以看到tqdm列印的進度條，
0%| | 0/290 [00:00<?, ?it/s]
1%| | 2/290 [00:02<06:36, 1.38s/it]
3%|▎ | 9/290 [00:11<06:51, 1.46s/it]
13%|█▎ | 38/290 [01:08<07:01, 1.67s/it]
27%|██▋ | 79/290 [03:08<11:06, 3.16s/it]
45%|████▌ | 131/290 [06:39<11:56, 4.51s/it]
61%|██████ | 177/290 [11:41<15:11, 8.07s/it]
三.資料準備
繼續完善代碼，將結果輸出至檔案，并定義類分批管理，
- 1.先執行get_data_with_windows(‘train’)函式拼接檔案
- 2.再執行train_data = BatchManager(10, ‘train’)函式分批處理
- 3.用函式get_data_with_windows(‘test’)處理測驗集資料
該部分最終完整代碼如下：
- data_utils.py
#encoding:utf-8
import pandas as pd
import pickle
import numpy as np
from tqdm import tqdm
import os
import math
#功能:獲取值對應的下標 引數為串列和字符
def item2id(data,w2i):
#x在字典中直接獲取 不在字典中回傳UNK
return [w2i[x] if x in w2i else w2i['UNK'] for x in data]
#----------------------------功能:拼接檔案---------------------------------
def get_data_with_windows(name='train'):
#讀取prepare_data.py生成的dict.pkl檔案 存盤字典{類別:下標}
with open(f'data/dict.pkl', 'rb') as f:
map_dict = pickle.load(f) #加載字典
#存盤所有資料
results = []
root = os.path.join('data/prepare/'+name)
files = list(os.listdir(root))
print(files)
#['10.csv', '11.csv', '12.csv',.....]
#獲取所有檔案 進度條
for file in tqdm(files):
all_data = []
path = os.path.join(root, file)
samples = pd.read_csv(path,sep=',')
max_num = len(samples)
#獲取sep換行分隔符下標 -1 20 40 60
sep_index = [-1]+samples[samples['word']=='sep'].index.tolist()+[max_num]
#print(sep_index)
#[-1, 83, 92, 117, 134, 158, 173, 200,......]
#----------------------------------------------------------------------
# 獲取句子并將句子全部都轉換成id
#----------------------------------------------------------------------
for i in range(len(sep_index)-1):
start = sep_index[i] + 1 #0 (-1+1)
end = sep_index[i+1] #20
data = []
#每個特征進行處理
for feature in samples.columns: #訪問每列
#通過函式item2id獲取下標 map_dict兩個值(串列和字典) 獲取第二個值
data.append(item2id(list(samples[feature])[start:end],map_dict[feature][1]))
#將每句話的串列合成
all_data.append(data)
#----------------------------------------------------------------------
# 資料增強
#----------------------------------------------------------------------
#前后兩個句子拼接 每個句子六個元素(漢字、邊界、詞性、類別、偏旁、拼音)
two = []
for i in range(len(all_data)-1):
first = all_data[i]
second = all_data[i+1]
two.append([first[k]+second[k] for k in range(len(first))]) #六個元素
three = []
for i in range(len(all_data)-2):
first = all_data[i]
second = all_data[i+1]
third = all_data[i+2]
three.append([first[k]+second[k]+third[k] for k in range(len(first))])
#回傳所有結果
results.extend(all_data+two+three)
#return results
#資料存盤至本地 每次呼叫時間成本過大
with open(f'data/'+name+'.pkl', 'wb') as f:
pickle.dump(results, f)
#----------------------------功能:批處理---------------------------------
class BatchManager(object):
def __init__(self, batch_size, name='train'):
#呼叫函式拼接檔案
#data = get_data_with_windows(name)
#讀取檔案
with open(f'data/'+name+'.pkl', 'rb') as f:
data = pickle.load(f)
print(len(data)) #265455句話
print(len(data[0])) #6種類別
print(len(data[0][0])) #第一句包含字的數量 83
print("原始資料:", data[0])
#資料批處理
self.batch_data = self.sort_and_pad(data, batch_size)
self.len_data = len(self.batch_data)
def sort_and_pad(self, data, batch_size):
#計算總批次數量 26546
num_batch = int(math.ceil(len(data) / batch_size))
#按照句子長度排序
sorted_data = sorted(data, key=lambda x: len(x[0]))
batch_data = list()
#獲取一個批次的資料
for i in range(num_batch):
batch_data.append(self.pad_data(sorted_data[i*int(batch_size) : (i+1)*int(batch_size)]))
print("分批輸出:", batch_data[1000])
return batch_data
@staticmethod
def pad_data(data_):
#定義變數
chars = []
bounds = []
flags = []
radicals = []
pinyins = []
targets = []
#print("每個批次句子個數:", len(data_)) #10
#print("每個句子包含元素個數:", len(data_[0])) #6
#print("輸出data:", data_)
max_length = max([len(sentence[0]) for sentence in data_]) #值為1
#print(max_length)
#每個批次共有十組資料 每組資料均為六個元素
for line in data_:
char, bound, flag, target, radical, pinyin = line
padding = [0] * (max_length - len(char)) #計算補充字符數量
#注意char和chars不要寫錯 否則造成遞回回圈賦值錯誤
chars.append(char + padding)
bounds.append(bound + padding)
flags.append(flag + padding)
targets.append(target + padding)
radicals.append(radical + padding)
pinyins.append(pinyin + padding)
return [chars, bounds, flags, radicals, pinyins, targets]
#每次使用一個批次資料
def iter_batch(self, shuffle=False):
if shuffle:
random.shuffle(self.batch_data)
for idx in range(self.len_data):
yield self.batch_data[idx]
#-------------------------------功能:主函式--------------------------------------
if __name__ == '__main__':
#1.拼接檔案(第一次執行 后續可注釋)
#get_data_with_windows('train')
#2.分批處理
train_data = BatchManager(10, 'train')
#3.接著處理下測驗集資料
#get_data_with_windows('test')
原始資料及處理后的資料如下圖所示：
某些Python工具能看到中間輸出結果，可以看到我們的data_utils.py腳本成功將句子分批次補齊，每個批次處理為對應的10個句子 x 6個資料型別，
注：該部分老師丟失了視頻，是作者結合原始碼進行還原，哈哈！淚奔~
四.模型構建
此時我們專案的結構圖如下所示，包括：
- data：資料檔案夾，prepare為預處理資料，由很多包含六元組的CSV檔案組成
- train.pkl：訓練集句子六元組下標
- test.pkl：測驗集句子六元組下標
- data_process.py：獲取物體類別及個數、BIO資料標注、長短句分割
- prepare_data.py：獲取資料標簽、提取六元組（字、邊界、詞性、類別、偏旁、拼音）
- data_utils.py：獲取六元組對應的下標并進行對齊處理，后續轉換詞向量訓練
接著讓我們開始創建BiLSTM模型，
1.BiLSTM模型構建
第一步，創建模型構建腳本，
- model.py
核心代碼如下，大家可以先熟悉Model類中基本的函式、變陣列成，
#encoding:utf-8
"""
Created on Thu Jan 7 12:56:40 2021
@author: xiuzhang
"""
import tensorflow as tf
import numpy as np
#---------------------------功能：預測計算函式-----------------------------
def network(char,bound,flag,radical,pinyin,shapes,
initializer=tf.truncated_normal_initializer):
"""
功能：接收一個批次樣本的特征資料，計算網路的輸出值
:param char: int, id of chars a tensor of shape 2-D [None,None]
:param bound: int, a tensor of shape 2-D [None,None]
:param flag: int, a tensor of shape 2-D [None,None]
:param radical: int, a tensor of shape 2-D [None,None]
:param pinyin: int, a tensor of shape 2-D [None,None]
:param shapes: 詞向量形狀字典
:param initializer: 初始化函式
:return
"""
#--------------------------------------------------
#特征嵌入:將所有特征的id轉換成一個固定長度的向量
embedding = []
#五類特征轉換成詞向量再拼接
with tf.variable_scope('char_embedding'):
#獲取漢字資訊
char_lookup = tf.get_variable(
name = 'char_embedding', #名稱
shape = ['char'], #[num,dim] 行數(個數)*列數(向量維度)
initializer = initializer
)
#詞向量映射
embedding.append(tf.nn.embedding_lookup(char_lookup,char))
#-----------------------------功能：定義模型類---------------------------
class Model(object):
#初始化
def __init__(self, dict_):
#通過dict.pkl計算各個特征數量
self.num_char = len(dict_['word'][0])
self.num_bound = len(dict_['bound'][0])
self.num_flag = len(dict_['flag'][0])
self.num_radical = len(dict_['radical'][0])
self.num_pinyin = len(dict_['pinyin'][0])
self.num_entity = len(dict_['label'][0])
#字符映射成向量的維度
self.char_dim = 100
self.bound_dim = 20
self.flag_dim = 50
self.radical_dim = 50
self.pinyin_dim = 50
#shape表示為[num,dim] 行數(個數)*列數(向量維度)
#定義網路 接收批次樣本
def get_logits(self,char,bound,flag,radical,pinyin):
"""
功能：接收一個批次樣本的特征資料，計算網路的輸出值
:param char: int, id of chars a tensor of shape 2-D [None,None]
:param bound: int, a tensor of shape 2-D [None,None]
:param flag: int, a tensor of shape 2-D [None,None]
:param radical: int, a tensor of shape 2-D [None,None]
:param pinyin: int, a tensor of shape 2-D [None,None]
:return
"""
#定義字典傳參
shapes = {}
shapes['char'] = [self.num_char,self.char_dim]
shapes['bound'] = [self.num_bound,self.bound_dim]
shapes['flag'] = [self.num_flag,self.flag_dim]
shapes['radical'] = [self.num_radical,self.radical_dim]
shapes['pinyin'] = [self.num_pinyin,self.pinyin_dim]
return network(char,bound,flag,radical,pinyin,dict_input)
第二步，我們嘗試撰寫一個test.py腳本理解詞嵌入相關知識，
- test.py
# -*- coding: utf-8 -*-
"""
Created on Thu Jan 7 12:56:40 2021
@author: xiuzhang
"""
import tensorflow as tf
import numpy as np
matrix = np.array([
[1,1,1,1,1,1],
[2,2,2,2,2,2],
[3,3,3,3,3,3],
[4,4,4,4,4,4]
])
x = np.array([
[0,2,1,1,2],
[3,2,0,2,2]
])
#詞向量轉換
result = tf.nn.embedding_lookup(matrix,x)
with tf.Session() as sess:
print(sess.run(result))
其輸出結果如下圖所示，它通過embedding_lookup函式將x矩陣按matrix進行詞向量映射，比如[0,2,1,1,2]在matrix分別對應第一行、第三行、第二行、第二行和第四行，相當于每一個id對應一個向量，最終得到如下結果，
同樣下面這個函式將char漢字進行詞向量映射，
- embedding.append(tf.nn.embedding_lookup(char_lookup,char))
第三步，繼續完善model.py代碼，
我們嘗試對引數進行修改，多個引數傳遞并呼叫同一規則函式時，可以將引數插入至字典中，從而優化代碼，比如：
- 優化前
def network(char,bound,flag,radical,pinyin,shapes,initializer=…) - 優化后
def network(inputs,shapes,initializer=…)
接著定義雙向LSTM神經網路，為了提高運算效率，我們需要計算輸入Inputs句子的實際長度，而填充資料PAD（下標0）不計算，
完整代碼如下，它將詞向量輸入后處理，最侄訓傳三維矩陣，每個詞做一個多分類（31種物體類別），核心函式相當于一個編碼器，
- get_logits(self,char,bound,flag,radical,pinyin)
- network(inputs,shapes,num_entity,lstm_dim=100, initializer)
- [batch_size,max_length,num_entity]
#encoding:utf-8
"""
Created on Thu Jan 7 12:56:40 2021
@author: xiuzhang
"""
import tensorflow as tf
import numpy as np
from tensorflow.contrib import rnn
#---------------------------功能：預測計算函式-----------------------------
def network(inputs,shapes,num_entity,lstm_dim=100,
initializer=tf.truncated_normal_initializer):
"""
功能：接收一個批次樣本的特征資料，計算網路的輸出值
:param char: int, id of chars a tensor of shape 2-D [None,None] 批次數量*每個批次句子長度
:param bound: int, a tensor of shape 2-D [None,None]
:param flag: int, a tensor of shape 2-D [None,None]
:param radical: int, a tensor of shape 2-D [None,None]
:param pinyin: int, a tensor of shape 2-D [None,None]
:param shapes: 詞向量形狀字典
:param lstm_dim: 神經元的個數
:param num_entity: 物體標簽數量 31種型別
:param initializer: 初始化函式
:return
"""
#--------------------------------------------------
#特征嵌入:將所有特征的id轉換成一個固定長度的向量
#--------------------------------------------------
embedding = []
keys = list(shapes.keys())
#回圈將五類特征轉換成詞向量 后續拼接
for key in keys:
with tf.variable_scope(key+'_embedding'):
#獲取漢字資訊
lookup = tf.get_variable(
name = key + '_embedding', #名稱
shape = [key], #[num,dim] 行數(個數)*列數(向量維度)
initializer = initializer
)
#詞向量映射 漢字結果[None,None,100]
embedding.append(tf.nn.embedding_lookup(lookup,inputs[key]))
#拼接詞向量 shape[None,None,char_dim+bound_dim+flag_dim+radical_dim+pinyin_dim]
embed = tf.concat(embedding,axis=-1) #最后一個維度上拼接 -1
#lengths: 計算輸入inputs每句話的實際長度(填充內容不計算)
#填充值PAD下標為0 因此總長度減去PAD數量即為實際長度 從而提升運算效率
sign = tf.sign(tf.abs(inputs[keys[0]])) #字符長度
lengths = tf.reduce_sum(sign, reduction_indices=1)
#獲取填充序列長度 char的第二個維度
num_time = tf.shape(inputs[keys[0]])[1]
#--------------------------------------------------
#回圈神經網路編碼: 雙層雙向網路
#--------------------------------------------------
#第一層
with tf.variable_scope('BiLSTM_layer1'):
lstm_cell = {}
#第一層前向 后向
for name in ['forward','backward']:
with tf.varibale_scope(name): #設定名稱
lstm_cell[name] = rnn.BasicLSTMCell(
lstm_dim, #神經元的個數
initializer = initializer
)
#運行LSTM
outputs1,finial_states1 = tf.nn.bidirectional_dynamic_run(
lstm_cell['forward'],
lstm_cell['backward'],
embed,
dtype = tf.float32,
sequence_length = lengths #序列實際長度(該引數可省略)
)
#拼接前向LSTM和后向LSTM輸出
outputs1 = tf.concat(outputs1,axis=-1) #b,L,2*lstm_dim
#第二層
with tf.variable_scope('BiLSTM_layer2'):
lstm_cell = {}
#第一層前向 后向
for name in ['forward','backward']:
with tf.varibale_scope(name): #設定名稱
lstm_cell[name] = rnn.BasicLSTMCell(
lstm_dim, #神經元的個數
initializer = initializer
)
#運行LSTM
outputs,finial_states = tf.nn.bidirectional_dynamic_run(
lstm_cell['forward'],
lstm_cell['backward'],
embed, #是否利用第一層網路
dtype = tf.float32,
sequence_length = lengths #序列實際長度(該引數可省略)
)
#最終結果 [batch_size,maxlength,2*lstm_dim] 即200
result = tf.concat(outputs,axis=-1)
#--------------------------------------------------
#輸出映射
#--------------------------------------------------
#轉換成二維矩陣 [batch_size*maxlength,2*lstm_dim]
result = tf.reshape(result, [-1,2*lstm_dim])
#第一層映射 矩陣乘法
with tf.variable_scope('project_layer1'):
#權重
w = tf.get_variable(
name = 'w',
shape = [2*lstm_dim,lstm_dim], #轉100維
initializer = initializer
)
#bias
b = tf.get_variable(
name = 'w',
shape = [lstm_dim],
initializer = tf.zeros_initializer()
)
#運算 激活函式relu
result = tf.nn.relu(matmul(result,w)+b)
#第二層映射 矩陣乘法
with tf.variable_scope('project_layer2'):
#權重
w = tf.get_variable(
name = 'w',
shape = [lstm_dim,num_entity], #31種物體類別
initializer = initializer
)
#bias
b = tf.get_variable(
name = 'w',
shape = [num_entity],
initializer = tf.zeros_initializer()
)
#運算 激活函式relu 最后一層不激活
result = matmul(result,w)+b
#形狀轉換成三維
result = tf.reshape(result, [-1,num_time,num_entity])
#[batch_size,max_length,num_entity]
return result
#-----------------------------功能：定義模型類---------------------------
class Model(object):
#初始化
def __init__(self, dict_):
#通過dict.pkl計算各個特征數量
self.num_char = len(dict_['word'][0])
self.num_bound = len(dict_['bound'][0])
self.num_flag = len(dict_['flag'][0])
self.num_radical = len(dict_['radical'][0])
self.num_pinyin = len(dict_['pinyin'][0])
self.num_entity = len(dict_['label'][0])
#字符映射成向量的維度
self.char_dim = 100
self.bound_dim = 20
self.flag_dim = 50
self.radical_dim = 50
self.pinyin_dim = 50
#shape表示為[num,dim] 行數(個數)*列數(向量維度)
#設定LSTM的維度 神經元的個數
self.lstm_dim = 100
#定義網路 接收批次樣本
def get_logits(self,char,bound,flag,radical,pinyin):
"""
功能：接收一個批次樣本的特征資料，計算網路的輸出值
:param char: int, id of chars a tensor of shape 2-D [None,None]
:param bound: int, a tensor of shape 2-D [None,None]
:param flag: int, a tensor of shape 2-D [None,None]
:param radical: int, a tensor of shape 2-D [None,None]
:param pinyin: int, a tensor of shape 2-D [None,None]
:return: 回傳3-d tensor [batch_size,max_length,num_entity]
"""
#定義字典傳參
shapes = {}
shapes['char'] = [self.num_char,self.char_dim]
shapes['bound'] = [self.num_bound,self.bound_dim]
shapes['flag'] = [self.num_flag,self.flag_dim]
shapes['radical'] = [self.num_radical,self.radical_dim]
shapes['pinyin'] = [self.num_pinyin,self.pinyin_dim]
#輸入引數定義字典
inputs = {}
inputs['char'] = char
inputs['bound'] = bound
inputs['flag'] = flag
inputs['radical'] = radical
inputs['pinyin'] = pinyin
#return network(char,bound,flag,radical,pinyin,shapes)
return network(inputs,shapes,lstm_dim=self.lstm_dim,num_entity=self.num_entity)
下面我們補充一張該圖的演算法流程圖，基本流程：
- 首先將漢字、邊界、詞性、偏旁和拼音轉換成詞向量
- 詞嵌入拼接成270維輸入
- 經過兩個雙向LSTM，轉換成200維輸出結果，做31種物體類別的分類處理
模型之間的引數計算如下圖所示（源自白老師），LSTM有4個門控，31是輸出物體標簽的數量，100表示LSTM的神經元數，
注意，我們可以查看BILSTM原始碼幫助學習，比如其回傳值包括輸出(前向&后向)和狀態，
2.CRF模型融合
最終得到31個值（物體類別數）后，我們接下來需要做Softmax嗎？
我們不做Softmax，我們不是要每個時刻概率最大，而是需要序列概率最大，因此接下來通過條件隨機場計算損失，此時，我們每個時刻有31種選擇，假設存在一個10長度的序列，它有31的10次方個組合，而真實的序列只有一種，我們的目標是讓真實序列的概率在整個序列所有概率中最大，因此采用CRF模型，
下面開始撰寫代碼：
- model.py
重點：下面總結希望大家認真閱讀
傳統CRF++是通過統計學方法計算每個時刻隱狀態的分值，而現在我們是通過模型network來完成的，因此該模型稱為BiLSTM-CRF模型，同時，呼叫crf_log_likelihood()函式計算條件隨機場的對數似然，如下圖所示，初始時刻狀態為31個概率為0（log-1000）和Start概率為1（log0），
- BiLSTM：負責提取特征（結合背景關系），每個時刻輸出31個值
- CRF：負責計算隱狀態分值
- 該模型與隱馬爾可夫模型本質區別是計算分數方法不同，一種是基于統計學方法P(y|x)，一種是基于神經網路實作（BiLSTM），
- 最后的結果就是真實概率值在所有概率值中最大，因此條件隨機場是序列歸一化，對整個序列的分值做歸一化處理，
此時model.py的完整代碼如下：
#encoding:utf-8
"""
Created on Thu Jan 7 12:56:40 2021
@author: xiuzhang
"""
import tensorflow as tf
import numpy as np
from tensorflow.contrib import rnn
#計算條件隨機場的對數似然
from tensorflow.contrib.crf import crf_log_likelihood
#---------------------------功能：預測計算函式-----------------------------
def network(inputs,shapes,num_entity,lstm_dim=100,
initializer=tf.truncated_normal_initializer):
"""
功能：接收一個批次樣本的特征資料，計算網路的輸出值
:param char: int, id of chars a tensor of shape 2-D [None,None] 批次數量*每個批次句子長度
:param bound: int, a tensor of shape 2-D [None,None]
:param flag: int, a tensor of shape 2-D [None,None]
:param radical: int, a tensor of shape 2-D [None,None]
:param pinyin: int, a tensor of shape 2-D [None,None]
:param shapes: 詞向量形狀字典
:param lstm_dim: 神經元的個數
:param num_entity: 物體標簽數量 31種型別
:param initializer: 初始化函式
:return
"""
#--------------------------------------------------
#特征嵌入:將所有特征的id轉換成一個固定長度的向量
#--------------------------------------------------
embedding = []
keys = list(shapes.keys())
#回圈將五類特征轉換成詞向量 后續拼接
for key in keys:
with tf.variable_scope(key+'_embedding'):
#獲取漢字資訊
lookup = tf.get_variable(
name = key + '_embedding', #名稱
shape = [key], #[num,dim] 行數(個數)*列數(向量維度)
initializer = initializer
)
#詞向量映射 漢字結果[None,None,100]
embedding.append(tf.nn.embedding_lookup(lookup,inputs[key]))
#拼接詞向量 shape[None,None,char_dim+bound_dim+flag_dim+radical_dim+pinyin_dim]
embed = tf.concat(embedding,axis=-1) #最后一個維度上拼接 -1
#lengths: 計算輸入inputs每句話的實際長度(填充內容不計算)
#填充值PAD下標為0 因此總長度減去PAD數量即為實際長度 從而提升運算效率
sign = tf.sign(tf.abs(inputs[keys[0]])) #字符長度
lengths = tf.reduce_sum(sign, reduction_indices=1)
#獲取填充序列長度 char的第二個維度
num_time = tf.shape(inputs[keys[0]])[1]
#--------------------------------------------------
#回圈神經網路編碼: 雙層雙向網路
#--------------------------------------------------
#第一層
with tf.variable_scope('BiLSTM_layer1'):
lstm_cell = {}
#第一層前向 后向
for name in ['forward','backward']:
with tf.varibale_scope(name): #設定名稱
lstm_cell[name] = rnn.BasicLSTMCell(
lstm_dim, #神經元的個數
initializer = initializer
)
#運行LSTM
outputs1,finial_states1 = tf.nn.bidirectional_dynamic_run(
lstm_cell['forward'],
lstm_cell['backward'],
embed,
dtype = tf.float32,
sequence_length = lengths #序列實際長度(該引數可省略)
)
#拼接前向LSTM和后向LSTM輸出
outputs1 = tf.concat(outputs1,axis=-1) #b,L,2*lstm_dim
#第二層
with tf.variable_scope('BiLSTM_layer2'):
lstm_cell = {}
#第一層前向 后向
for name in ['forward','backward']:
with tf.varibale_scope(name): #設定名稱
lstm_cell[name] = rnn.BasicLSTMCell(
lstm_dim, #神經元的個數
initializer = initializer
)
#運行LSTM
outputs,finial_states = tf.nn.bidirectional_dynamic_run(
lstm_cell['forward'],
lstm_cell['backward'],
embed, #是否利用第一層網路
dtype = tf.float32,
sequence_length = lengths #序列實際長度(該引數可省略)
)
#最終結果 [batch_size,maxlength,2*lstm_dim] 即200
result = tf.concat(outputs,axis=-1)
#--------------------------------------------------
#輸出映射
#--------------------------------------------------
#轉換成二維矩陣 [batch_size*maxlength,2*lstm_dim]
result = tf.reshape(result, [-1,2*lstm_dim])
#第一層映射 矩陣乘法
with tf.variable_scope('project_layer1'):
#權重
w = tf.get_variable(
name = 'w',
shape = [2*lstm_dim,lstm_dim], #轉100維
initializer = initializer
)
#bias
b = tf.get_variable(
name = 'w',
shape = [lstm_dim],
initializer = tf.zeros_initializer()
)
#運算 激活函式relu
result = tf.nn.relu(matmul(result,w)+b)
#第二層映射 矩陣乘法
with tf.variable_scope('project_layer2'):
#權重
w = tf.get_variable(
name = 'w',
shape = [lstm_dim,num_entity], #31種物體類別
initializer = initializer
)
#bias
b = tf.get_variable(
name = 'w',
shape = [num_entity],
initializer = tf.zeros_initializer()
)
#運算 激活函式relu 最后一層不激活
result = matmul(result,w)+b
#形狀轉換成三維
result = tf.reshape(result, [-1,num_time,num_entity])
#[batch_size,max_length,num_entity]
return result,lengths
#-----------------------------功能：定義模型類---------------------------
class Model(object):
#初始化
def __init__(self, dict_):
#通過dict.pkl計算各個特征數量
self.num_char = len(dict_['word'][0])
self.num_bound = len(dict_['bound'][0])
self.num_flag = len(dict_['flag'][0])
self.num_radical = len(dict_['radical'][0])
self.num_pinyin = len(dict_['pinyin'][0])
self.num_entity = len(dict_['label'][0])
#字符映射成向量的維度
self.char_dim = 100
self.bound_dim = 20
self.flag_dim = 50
self.radical_dim = 50
self.pinyin_dim = 50
#shape表示為[num,dim] 行數(個數)*列數(向量維度)
#設定LSTM的維度 神經元的個數
self.lstm_dim = 100
#定義網路 接收批次樣本
def get_logits(self,char,bound,flag,radical,pinyin):
"""
功能：接收一個批次樣本的特征資料，計算網路的輸出值
:param char: int, id of chars a tensor of shape 2-D [None,None]
:param bound: int, a tensor of shape 2-D [None,None]
:param flag: int, a tensor of shape 2-D [None,None]
:param radical: int, a tensor of shape 2-D [None,None]
:param pinyin: int, a tensor of shape 2-D [None,None]
:return: 回傳3-d tensor [batch_size,max_length,num_entity]
"""
#定義字典傳參
shapes = {}
shapes['char'] = [self.num_char,self.char_dim]
shapes['bound'] = [self.num_bound,self.bound_dim]
shapes['flag'] = [self.num_flag,self.flag_dim]
shapes['radical'] = [self.num_radical,self.radical_dim]
shapes['pinyin'] = [self.num_pinyin,self.pinyin_dim]
#輸入引數定義字典
inputs = {}
inputs['char'] = char
inputs['bound'] = bound
inputs['flag'] = flag
inputs['radical'] = radical
inputs['pinyin'] = pinyin
#return network(char,bound,flag,radical,pinyin,shapes)
return network(inputs,shapes,lstm_dim=self.lstm_dim,num_entity=self.num_entity)
#--------------------------功能：定義loss CRF模型-------------------------
#引數: 模型輸出值 真實標簽序列 長度(不計算填充)
def loss(self,result,targets,lengths):
#獲取長度
b = len(lengths) #真實長度
num_steps = tf.shape(result)[1] #含填充
#轉移矩陣
with tf.variable_scope('crf_loss'):
#取log相當于概率接近0
small = -1000.0
#初始時刻狀態 兩個矩陣在最后一個維度合并
start_logits = tf.concat(
#前31個-1000概率為0 最后一個start為0取log為1
[small*tf.ones(shape=[b,1,self.num_entity]),tf.zeros(shape=[b,1,1])],
axis = -1
)
#X值拼接 每個時刻加一個狀態
pad_logits = tf.cast(small*tf.ones([b,num_steps,1]),tf.float32)
logits = tf.concat([result, pad_logits], axis=-1)
logits = tf.concat([start_logits,logits], axis=1) #第二個位置拼接
#Y值拼接
targets = tf.concat(
[tf.cast(self.num_entity*tf.ones([b,1]),tf.int32),targets],
axis = -1
)
#計算
self.trans = tf.get_variable(
name = 'trans',
#初始概率start加1 最終32個
shape = [self.num_entity+1,self.num_entity+1],
initializer = tf.truncated_normal_initializer()
)
#損失 計算條件隨機場的對數似然 每個樣本計算幾個值
log_likehood, self.trans = crf_log_likelihood(
inputs = logits, #輸入
tag_indices = targets, #目標
transition_params = self.trans,
sequence_lengths = lengths #真實樣本長度
)
#回傳所有樣本平均值 數加個負號損失最小化
return tf.reduce_mean(-log_likehood)
3.初始化函式完善
繼續修改Model類，在初始化init函式中增加如下功能：
- 定義接收資料的placeholder
- 呼叫get_logits計算模型輸出結果及句子真實長度
- 呼叫loss計算損失值
- 定義優化器，采用梯度截斷技術處理，如果導數值過大會導致步子邁得過大，造成梯度爆炸，因此限制在某個范圍內（如[-5,5]）
- 保存模型引數
該模型最終將270維的向量（字、邊界、詞性、偏旁、拼音）映射成31維向量，核心代碼如下：
class Model(object):
#---------------------------------------------------------
#初始化
def __init__(self, dict_, lr=0.0001):
#通過dict.pkl計算各個特征數量
self.num_char = len(dict_['word'][0])
self.num_bound = len(dict_['bound'][0])
self.num_flag = len(dict_['flag'][0])
self.num_radical = len(dict_['radical'][0])
self.num_pinyin = len(dict_['pinyin'][0])
self.num_entity = len(dict_['label'][0])
#字符映射成向量的維度
self.char_dim = 100
self.bound_dim = 20
self.flag_dim = 50
self.radical_dim = 50
self.pinyin_dim = 50
#shape表示為[num,dim] 行數(個數)*列數(向量維度)
#設定LSTM的維度 神經元的個數
self.lstm_dim = 100
#學習率
self.lr = lr
#---------------------------------------------------------
#定義接收資料的placeholder [None,None] 批次 句子長度
self.char_inputs = tf.placeholder(dtype=tf.int32,shape=[None,None],name='char_inputs')
self.bound_inputs = tf.placeholder(dtype=tf.int32,shape=[None,None],name='bound_inputs')
self.flag_inputs = tf.placeholder(dtype=tf.int32,shape=[None,None],name='flag_inputs')
self.radical_inputs = tf.placeholder(dtype=tf.int32,shape=[None,None],name='radical_inputs')
self.pinyin_inputs = tf.placeholder(dtype=tf.int32,shape=[None,None],name='pinyin_inputs')
self.targets = tf.placeholder(dtype=tf.int32,shape=[None,None],name='targets') #目標真實值
self.global_step = tf.Variable(0,trainable=False) #不能訓練 用于計數
#---------------------------------------------------------
#傳遞給網路 計算模型輸出值
#引數：輸入的字、邊界、詞性、偏旁、拼音下標 -> network轉換詞向量并計算
#回傳：網路輸出值、每句話的真實長度
self.logits,self.lengths = self.get_logits(
self.char_inputs,
self.bound_inputs,
self.flag_inputs,
self.radical_inputs,
self.pinyin_inputs
)
#---------------------------------------------------------
#計算損失
#引數：模型輸出值、真實標簽序列、長度(不計算填充)
#回傳：損失值
self.cost = self.loss(
self.logits,
self.targets,
self.lengths
)
#---------------------------------------------------------
#優化器優化 采用梯度截斷技術
with tf.variable_scope('optimizer'):
opt = tf.train.AdamOptimizer(self.lr) #學習率
#計算所有損失函式的導數值
grad_vars = opt.compute_gradients(self.cost)
#梯度截斷-導數值過大會導致步子邁得過大 梯度爆炸(因此限制在某個范圍內)
#grad_vars記錄每組引數導數和本身
clip_grad_vars = [[tf.clip_by_value(g,-5,5),v] for g,v in grad_vars]
#使用截斷后的梯度更新引數 該方法每應用一次global_step引數自動加1
self.train_op = opt.apply_gradients(clip_grad_vars,self.global_step)
#模型保存 保留最近5次模型
self.saver = tf.train.Saver(tf.global_variables(),max_to_keep=5)
4.模型訓練
新建 train.py 檔案，并撰寫訓練代碼，
- 第一步，首先引入BatchManager類，我們可以用之前data_utils.py腳本定義的BatchManager直接呼叫處理好的訓練集和測驗集，
- 第二步，自定義函式讀取字典dict.pkl內容，該檔案存盤了物體六元組，
- 第三步，引入model類搭建模型，
核心代碼如下圖所示，我們先嘗試運行下代碼：
在除錯程式時，我們可以增加斷點單步除錯，也可以print打樁輸出，比如：
(1) network模型分析
重點是觀察network函式（model.py）的引數變化情況，神經網路的輸出結果如下，核心功能包括：
- 呼叫tf.nn.embedding_lookup函式完成詞向量映射
- 呼叫rnn.BasicLSTMCell構建LSTM網路
- 呼叫tf.nn.bidirectional_dynamic_rnn組合BiLSTM，兩層BiLSTM
- 兩層全連接層將維度轉換成31，相當于做31分類（對應物體類別）
– result = tf.nn.relu(tf.matmul(result,w)+b)
– result = tf.matmul(result,w)+b
計算六元組個數
字: 1663
邊界: 5
詞性: 56
偏旁: 227
拼音: 989
類別: 31
""""初始化操作"""
model init: 1663 5 56 227 989 31
shapes: {'char': [1663, 100], 'bound': [5, 20], 'flag': [56, 50],
'radical': [227, 50], 'pinyin': [989, 50]}
Network Shape: ['char', 'bound', 'flag', 'radical', 'pinyin']
"""詞向量映射 每個字映射100維向量 [None,None,100]"""
Network Input: {'char': <tf.Tensor 'char_inputs:0' shape=(?, ?) dtype=int32>,...
Network Embedding: [
<tf.Tensor 'char_embedding' shape=(?, ?, 100) dtype=float32>,
<tf.Tensor 'bound_embedding' shape=(?, ?, 20) dtype=float32>,
<tf.Tensor 'flag_embedding' shape=(?, ?, 50) dtype=float32>,
<tf.Tensor 'radical_embedding' shape=(?, ?, 50) dtype=float32>,
<tf.Tensor 'pinyin_embedding' shape=(?, ?, 50) dtype=float32>
]
"""合并270維度"""
Network Embed: Tensor("concat:0", shape=(?, ?, 270), dtype=float32)
""""神經網路 2個LSTM組織(各100個神經元)"""
Network BiLSTM-1: Tensor("concat_1:0", shape=(?, ?, 200), dtype=float32)
Network BiLSTM-2: Tensor("concat_2:0", shape=(?, ?, 200), dtype=float32)
Dense-1: Tensor("project_layer1/Relu:0", shape=(?, 100), dtype=float32)
Dense-2: Tensor("project_layer2/add:0", shape=(?, 31), dtype=float32)
"""二維轉三維輸出最終結果"""
Result: Tensor("Reshape_1:0", shape=(?, ?, 31), dtype=float32)
(2) loss計算
核心功能包括：
- 獲取真實長度、輸入資料集 [批次大小, 序列長度, 31個物體類別]、真實標簽
- 計算損失
-用crf_log_likelihood計算條件隨機場的對數似然
Loss lengths: Tensor("strided_slice_1:0", shape=(), dtype=int32)
Loss Inputs: Tensor("Reshape_1:0", shape=(?, ?, 31), dtype=float32)
Loss Targets: Tensor("targets:0", shape=(?, ?), dtype=int32)
Loss Logits: Tensor("crf_loss/concat_2:0", shape=(?, ?, 32), dtype=float32)
Loss Targets: Tensor("crf_loss/concat_3:0", shape=(?, ?), dtype=int32)
Loss loglikehood: Tensor("crf_loss/sub:0", dtype=float32)
Loss Trans: <tf.Variable 'crf_loss/trans:0' shape=(32, 32) dtype=float32_ref>
Cost: Tensor("crf_loss/Mean:0", shape=(), dtype=float32)
Optimizer: name: "optimizer/Adam"
op: "AssignAdd"
input: "Variable"
input: "optimizer/Adam/value"
attr {
key: "T"
value {
type: DT_INT32
}
}
attr {
key: "_class"
value {
list {
s: "loc:@Variable"
}
}
}
attr {
key: "use_locking"
value {
b: false
}
}
最后構造優化器，采用梯度截斷技術及保存模型，
注意，可能報錯“AttributeError: module ‘tensorflow._api.v1.nn’ has no attribute ‘bidirectional_dynamic_run’”，注意版本問題，百度修改成對應的函式即可，作者是tensorflow1.15，
五.模型預測
1.輸出訓練誤差
上面將模型建立好之后，我們嘗試呼叫模型進行誤差訓練，train.py代碼如下，這里的喂資料操作可以封裝到類中實作，
# -*- coding: utf-8 -*-
"""
Created on Thu Jan 7 18:57:23 2021
@author: xiuzhang
"""
import tensorflow as tf
from data_utils import BatchManager
import pickle
from model import Model
#-----------------------------功能：讀取字典---------------------------
dict_file = 'data/dict.pkl'
def get_dict(path):
with open(path, 'rb') as f:
data = pickle.load(f)
return data
#-----------------------------功能：訓練函式---------------------------
batch_size = 20
def train():
#呼叫已定義的方法獲取處理好的資料集
train_manager = BatchManager(batch_size, name='train')
print('train:', type(train_manager)) #<class 'data_utils.BatchManager'>
#讀取字典
mapping_dict = get_dict(dict_file)
print('train:', len(mapping_dict)) #6
print('計算六元組個數')
print('字:', len(mapping_dict['word'][0])) #1663
print('邊界:', len(mapping_dict['bound'][0])) #5
print('詞性:', len(mapping_dict['flag'][0])) #56
print('偏旁:', len(mapping_dict['radical'][0])) #227
print('拼音:', len(mapping_dict['pinyin'][0])) #989
print('類別:', len(mapping_dict['label'][0]),'\n') #31
#-------------------------搭建模型---------------------------
#實體化模型 執行init初始化方法model核心函式：
# 1.get_logits：傳遞給網路 計算模型輸出值
# 2.loss：計算損失值
#-----------------------------------------------------------
model = Model(mapping_dict)
print("---------------模型構建成功---------------------\n")
#初始化訓練
init = tf.global_variables_initializer()
with tf.Session() as sess:
sess.run(init)
for i in range(10):
#呼叫iter_batch函式 迭代程序可以讓梯度下降在不斷嘗試找到最優解
for batch in train_manager.iter_batch(shuffle=True): #亂序
#print(len(batch)) #6個型別
#print(len(batch[0]),len(batch[1]),len(batch[2])) #20個
#每次獲取一個批次的資料 feed_dict喂資料 placeholder用于接收神經網路資料
_,loss = sess.run([model.train_op,model.cost],feed_dict={
model.char_inputs : batch[0],
model.bound_inputs : batch[2],
model.flag_inputs : batch[3],
model.radical_inputs : batch[4],
model.pinyin_inputs : batch[5],
model.targets : batch[1] #注意順序
})
print('loss:{}'.format(loss))
#---------------------------功能:主函式---------------------------------
if __name__ == '__main__':
train()
輸出結果如下圖示，可以看到loss從大到小，
loss:545.8291625976562
loss:901.7841796875
loss:442.2290954589844
loss:876.3251953125
loss:332.58746337890625
loss:674.8977661132812
loss:409.48663330078125
loss:220.19033813476562
.....
loss:31.463674545288086
loss:45.567161560058594
loss:98.6595458984375
loss:72.75428009033203
loss:52.30353927612305
問題：
這里需要注意一個問題，如下所示，該問題通常是詞向量映射錯誤導致，但這個問題困擾了我兩天，除錯了很長時間代碼，終于解決，淚奔~
- InvalidArgumentError: indices[0,2] = 7 is not in [0, 5)
- embedding.append(tf.nn.embedding_lookup(lookup,inputs[key]))
原因：
我們最終生成的CSV檔案格式是word、label、bound、flag、radical、pinyin順序，但是后面寫入dict.pkl檔案及feed_dict喂入資料訓練的順序不一致，這導致最終映射的詞向量不一致，造成了“InvalidArgumentError: indices[0,2] = 7 is not in [0, 5)”，
解決方法：
由于之前預處理CSV檔案按照char, target, bound, flag, radical, pinyin這個順序，所以生成的dict.pkl也需要按照這個順序讀寫，而feed_dict時讀取dict.pkl順序也需要按照這個順序，標簽是第2列，因此，修改方法：
- 所有順序需要一致，重新按char, target, bound, flag, radical, pinyin生成dict.pkl檔案；
– data_utils.py: char, target, bound, flag, radical, pinyin = line - feed_dict順序調整
– model.targets:batch[1] - 建議包含target（label）的操作，如讀取、賦值、寫入均按照統一的順序執行，除非是字典按照關鍵詞呼叫（如shapes[‘char’]），
2.預測資料
- 在Model類中定義run_step函式分批處理資料
- 在Model類中定義decode函式解碼，通過模型輸出和轉義矩陣預測
- 在Model類中定義predict函式預測
- 在train.py中分配輸出
輸出結果如下圖所示：
六.完整代碼
代碼下載地址：
- https://github.com/eastmountyxz/AI-for-Keras
1.model.py
#encoding:utf-8
"""
Created on Thu Jan 7 12:56:40 2021
@author: xiuzhang
"""
import tensorflow as tf
import numpy as np
from tensorflow.contrib import rnn
#計算條件隨機場的對數似然
from tensorflow.contrib.crf import crf_log_likelihood, viterbi_decode
#---------------------------功能：預測計算函式-----------------------------
def network(inputs,shapes,num_entity,lstm_dim=100,
initializer=tf.truncated_normal_initializer):
"""
功能：接收一個批次樣本的特征資料，計算網路的輸出值
:param char: int, id of chars a tensor of shape 2-D [None,None] 批次數量*每個批次句子長度
:param bound: int, a tensor of shape 2-D [None,None]
:param flag: int, a tensor of shape 2-D [None,None]
:param radical: int, a tensor of shape 2-D [None,None]
:param pinyin: int, a tensor of shape 2-D [None,None]
:param shapes: 詞向量形狀字典
:param lstm_dim: 神經元的個數
:param num_entity: 物體標簽數量 31種型別
:param initializer: 初始化函式
:return
"""
#--------------------------------------------------
#特征嵌入:將所有特征的id轉換成一個固定長度的向量
#--------------------------------------------------
embedding = []
keys = list(shapes.keys())
print("Network Input:", inputs)
#{'char':<tf.Tensor 'char_inputs_10:0' shape=(?, ?) dtype=int32>,
print("Network Shape:", keys)
#['char', 'bound', 'flag', 'radical', 'pinyin']
#回圈將五類特征轉換成詞向量 后續拼接
for key in keys: #char
with tf.variable_scope(key+'_embedding'):
#獲取漢字資訊
lookup = tf.get_variable(
name = key + '_embedding', #名稱
shape = shapes[key], #[num,dim] 行數(字個數)*列數(向量維度) 1663*100
initializer = initializer
)
#詞向量映射 漢字結果[None,None,100] 每個字映射100維向量 inputs對應每個字
embedding.append(tf.nn.embedding_lookup(lookup, inputs[key]))
print("Network Embedding:", embedding)
#[<tf.Tensor 'char_embedding_14:0' shape=(?, ?, 100) dtype=float32>,
#拼接詞向量 shape[None,None,char_dim+bound_dim+flag_dim+radical_dim+pinyin_dim]
embed = tf.concat(embedding,axis=-1) #最后一個維度上拼接 -1
print("Network Embed:", embed, '\n')
#Tensor("concat:0", shape=(?, ?, 270), dtype=float32)
#lengths: 計算輸入inputs每句話的實際長度(填充內容不計算)
#填充值PAD下標為0 因此總長度減去PAD數量即為實際長度 從而提升運算效率
sign = tf.sign(tf.abs(inputs[keys[0]])) #char 字符長度
lengths = tf.reduce_sum(sign, reduction_indices=1) #第二個維度
#獲取填充序列長度 char的第二個維度
num_time = tf.shape(inputs[keys[0]])[1]
print(sign, lengths, num_time)
#Tensor("Sign:0", shape=(?, ?), dtype=int32)
#Tensor("Sum:0", shape=(?,), dtype=int32)
#Tensor("strided_slice:0", shape=(), dtype=int32)
#--------------------------------------------------
#回圈神經網路編碼: 雙層雙向網路
#--------------------------------------------------
#第一層
with tf.variable_scope('BiLSTM_layer1'):
lstm_cell = {}
#第一層前向 后向
for name in ['forward','backward']:
with tf.variable_scope(name): #設定名稱
lstm_cell[name] = rnn.BasicLSTMCell(
lstm_dim #神經元的個數
)
#BiLSTM 2個LSTM組成(各100個神經元)
outputs1,finial_states1 = tf.nn.bidirectional_dynamic_rnn(
lstm_cell['forward'],
lstm_cell['backward'],
embed,
dtype = tf.float32,
sequence_length = lengths #序列實際長度(該引數可省略)
)
#拼接前向LSTM和后向LSTM輸出
outputs1 = tf.concat(outputs1,axis=-1) #b,L,2*lstm_dim
print('Network BiLSTM-1:', outputs1)
#Tensor("concat_1:0", shape=(?, ?, 200), dtype=float32)
#第二層
with tf.variable_scope('BiLSTM_layer2'):
lstm_cell = {}
#第一層前向 后向
for name in ['forward','backward']:
with tf.variable_scope(name): #設定名稱
lstm_cell[name] = rnn.BasicLSTMCell(
lstm_dim #神經元的個數
)
#BiLSTM
outputs,finial_states = tf.nn.bidirectional_dynamic_rnn(
lstm_cell['forward'],
lstm_cell['backward'],
outputs1, #是否利用第一層網路
dtype = tf.float32,
sequence_length = lengths #序列實際長度(該引數可省略)
)
#最終結果 [batch_size,maxlength,2*lstm_dim] 即200
result = tf.concat(outputs,axis=-1)
print('Network BiLSTM-2:', result)
#Tensor("concat_2:0", shape=(?, ?, 200), dtype=float32)
#--------------------------------------------------
#輸出全連接映射
#--------------------------------------------------
#轉換成二維矩陣再進行乘法操作 [batch_size*maxlength,2*lstm_dim]
result = tf.reshape(result, [-1,2*lstm_dim])
#第一層映射 矩陣乘法 200映射到100
with tf.variable_scope('project_layer1'):
#權重
w = tf.get_variable(
name = 'w',
shape = [2*lstm_dim,lstm_dim], #轉100維
initializer = initializer
)
#bias
b = tf.get_variable(
name = 'b',
shape = [lstm_dim],
initializer = tf.zeros_initializer()
)
#運算 激活函式relu
result = tf.nn.relu(tf.matmul(result,w)+b)
print("Dense-1:",result)
#Tensor("project_layer1/Relu:0", shape=(?, 100), dtype=float32)
#第二層映射 矩陣乘法 100映射到31
with tf.variable_scope('project_layer2'):
#權重
w = tf.get_variable(
name = 'w',
shape = [lstm_dim,num_entity], #31種物體類別
initializer = initializer
)
#bias
b = tf.get_variable(
name = 'b',
shape = [num_entity],
initializer = tf.zeros_initializer()
)
#運算 激活函式relu 最后一層不激活
result = tf.matmul(result,w)+b
print("Dense-2:",result)
#Tensor("project_layer2/add:0", shape=(?, 31), dtype=float32)
#形狀轉換成三維
result = tf.reshape(result, [-1,num_time,num_entity])
print('Result:', result, "\n")
#Tensor("Reshape_1:0", shape=(?, ?, 31), dtype=float32)
#[batch_size,max_length,num_entity]
return result,lengths
#-----------------------------功能：定義模型類---------------------------
class Model(object):
#---------------------------------------------------------
#初始化
def __init__(self, dict_, lr=0.0001):
#通過dict.pkl計算各個特征數量
self.num_char = len(dict_['word'][0])
self.num_bound = len(dict_['bound'][0])
self.num_flag = len(dict_['flag'][0])
self.num_radical = len(dict_['radical'][0])
self.num_pinyin = len(dict_['pinyin'][0])
self.num_entity = len(dict_['label'][0])
print('model init:', self.num_char, self.num_bound, self.num_flag,
self.num_radical, self.num_pinyin, self.num_entity)
#字符映射成向量的維度
self.char_dim = 100
self.bound_dim = 20
self.flag_dim = 50
self.radical_dim = 50
self.pinyin_dim = 50
#shape表示為[num,dim] 行數(個數)*列數(向量維度)
#設定LSTM的維度 神經元的個數
self.lstm_dim = 100
#學習率
self.lr = lr
#保存初始化字典
self.map = dict_
#---------------------------------------------------------
#定義接收資料的placeholder [None,None] 批次 句子長度
self.char_inputs = tf.placeholder(dtype=tf.int32,shape=[None,None],name='char_inputs')
self.bound_inputs = tf.placeholder(dtype=tf.int32,shape=[None,None],name='bound_inputs')
self.flag_inputs = tf.placeholder(dtype=tf.int32,shape=[None,None],name='flag_inputs')
self.radical_inputs = tf.placeholder(dtype=tf.int32,shape=[None,None],name='radical_inputs')
self.pinyin_inputs = tf.placeholder(dtype=tf.int32,shape=[None,None],name='pinyin_inputs')
self.targets = tf.placeholder(dtype=tf.int32,shape=[None,None],name='targets') #目標真實值
self.global_step = tf.Variable(0,trainable=False) #不能訓練 用于計數
#---------------------------------------------------------
#傳遞給網路 計算模型輸出值
#引數：輸入的字、邊界、詞性、偏旁、拼音下標 -> network轉換詞向量并計算
#回傳：網路輸出值、每句話的真實長度
self.logits,self.lengths = self.get_logits(
self.char_inputs,
self.bound_inputs,
self.flag_inputs,
self.radical_inputs,
self.pinyin_inputs
)
#---------------------------------------------------------
#計算損失
#引數：模型輸出值、真實標簽序列、長度(不計算填充)
#回傳：損失值
self.cost = self.loss(
self.logits,
self.targets,
self.lengths
)
print("Cost:", self.cost)
#---------------------------------------------------------
#優化器優化 采用梯度截斷技術
with tf.variable_scope('optimizer'):
opt = tf.train.AdamOptimizer(self.lr) #學習率
#計算所有損失函式的導數值
grad_vars = opt.compute_gradients(self.cost)
#梯度截斷-導數值過大會導致步子邁得過大 梯度爆炸(因此限制在某個范圍內)
#grad_vars記錄每組引數導數和本身
clip_grad_vars = [[tf.clip_by_value(g,-5,5),v] for g,v in grad_vars]
#使用截斷后的梯度更新引數 該方法每應用一次global_step引數自動加1
self.train_op = opt.apply_gradients(clip_grad_vars, self.global_step)
print("Optimizer:", self.train_op)
#模型保存 保留最近5次模型
self.saver = tf.train.Saver(tf.global_variables(),max_to_keep=5)
#---------------------------------------------------------
#定義網路 接收批次樣本
def get_logits(self,char,bound,flag,radical,pinyin):
"""
功能：接收一個批次樣本的特征資料，計算網路的輸出值
:param char: int, id of chars a tensor of shape 2-D [None,None]
:param bound: int, a tensor of shape 2-D [None,None]
:param flag: int, a tensor of shape 2-D [None,None]
:param radical: int, a tensor of shape 2-D [None,None]
:param pinyin: int, a tensor of shape 2-D [None,None]
:return: 回傳3-d tensor [batch_size,max_length,num_entity]
"""
#定義字典傳參
shapes = {}
shapes['char'] = [self.num_char,self.char_dim]
shapes['bound'] = [self.num_bound,self.bound_dim]
shapes['flag'] = [self.num_flag,self.flag_dim]
shapes['radical'] = [self.num_radical,self.radical_dim]
shapes['pinyin'] = [self.num_pinyin,self.pinyin_dim]
print("shapes:", shapes, '\n')
#{'char': [1663, 100], 'bound': [5, 20], 'flag': [56, 50],
# 'radical': [227, 50], 'pinyin': [989, 50]}
#輸入引數定義字典
inputs = {}
inputs['char'] = char
inputs['bound'] = bound
inputs['flag'] = flag
inputs['radical'] = radical
inputs['pinyin'] = pinyin
#return network(char,bound,flag,radical,pinyin,shapes)
return network(inputs,shapes,lstm_dim=self.lstm_dim,num_entity=self.num_entity)
#--------------------------功能：定義loss CRF模型-------------------------
#引數: 模型輸出值 真實標簽序列 長度(不計算填充)
def loss(self,result,targets,lengths):
#獲取長度
b = tf.shape(lengths)[0] #真實長度 該值只有一維
num_steps = tf.shape(result)[1] #含填充
print("Loss lengths:", b, num_steps)
print("Loss Inputs:", result)
print("Loss Targets:", targets)
#轉移矩陣
with tf.variable_scope('crf_loss'):
#取log相當于概率接近0
small = -1000.0
#初始時刻狀態
start_logits = tf.concat(
#前31個-1000概率為0 最后一個start為0取log為1
[small*tf.ones(shape=[b,1,self.num_entity]),tf.zeros(shape=[b,1,1])],
axis = -1 #兩個矩陣在最后一個維度合并
)
#X值拼接 每個時刻加一個狀態
pad_logits = tf.cast(small*tf.ones([b,num_steps,1]),tf.float32)
logits = tf.concat([result, pad_logits], axis=-1)
logits = tf.concat([start_logits,logits], axis=1) #第二個位置拼接
print("Loss Logits:", logits)
#Y值拼接
targets = tf.concat(
[tf.cast(self.num_entity*tf.ones([b,1]),tf.int32),targets],
axis = -1
)
print("Loss Targets:", targets)
#計算
self.trans = tf.get_variable(
name = 'trans',
#初始概率start加1 最終32個
shape = [self.num_entity+1,self.num_entity+1],
initializer = tf.truncated_normal_initializer()
)
#損失 計算條件隨機場的對數似然 每個樣本計算幾個值
log_likehood, self.trans = crf_log_likelihood(
inputs = logits, #輸入
tag_indices = targets, #目標
transition_params = self.trans,
sequence_lengths = lengths #真實樣本長度
)
print("Loss loglikehood:", log_likehood)
print("Loss Trans:", self.trans)
#回傳所有樣本平均值 數加個負號損失最小化
return tf.reduce_mean(-log_likehood)
#--------------------------功能：分步運行-------------------------
#引數: 會話、分批資料、訓練預測
def run_step(self,sess,batch,is_train=True):
if is_train:
feed_dict = {
self.char_inputs : batch[0],
self.bound_inputs : batch[2],
self.flag_inputs : batch[3],
self.radical_inputs : batch[4],
self.pinyin_inputs : batch[5],
self.targets : batch[1] #注意順序
}
#訓練計算損失
_,loss = sess.run([self.train_op,self.cost], feed_dict=feed_dict)
return loss
else: #預測沒有類標
feed_dict = {
self.char_inputs : batch[0],
self.bound_inputs : batch[2],
self.flag_inputs : batch[3],
self.radical_inputs : batch[4],
self.pinyin_inputs : batch[5],
}
#測驗計算結果
logits,lengths = sess.run([self.logits, self.lengths], feed_dict=feed_dict)
return logits,lengths
#--------------------------功能：解碼獲取id-------------------------
#引數:模型輸出值、真實長度、轉移矩陣(用于解碼)
def decode(self,logits,lengths,matrix):
#保留概率最大路徑
paths = []
small = -1000.0
#每個樣本解碼 31種類別+最后一個是0
start = np.asarray([[small]*self.num_entity+[0]])
#獲取每句話的成績和樣本真實長度
for score,length in zip(logits,lengths):
score = score[:length] #只取有效字符的輸出
pad = small*np.ones([length,1])
#拼接
logits = np.concatenate([score,pad],axis=-1)
logits = np.concatenate([start,logits],axis=0)
#解碼
path,_ = viterbi_decode(logits,matrix)
paths.append(path[1:])
#paths獲取的是id 還需要轉換成對應的物體標簽
return paths
#--------------------------功能：預測分析-------------------------
#引數: 會話、批次
def predict(self,sess,batch):
results = []
#獲取轉移矩陣
matrix = self.trans.eval()
#獲取模型結果 執行測驗
logits, lengths = self.run_step(sess, batch, is_train=False)
#呼叫解碼函式獲取paths
paths = self.decode(logits, lengths, matrix)
#查看字及對應的標記
chars = batch[0]
for i in range(len(paths)): #有多少路徑就有多少句子
#獲取第i句話真實長度
length = lengths[i]
#第i句話真實的字
chars[i][:length]
#ID轉換成對應的每個字
#map['word'][1]是字典
string = [self.map['word'][1][index] for index in chars[i][:length]]
#獲取tag
tags = [self.map['label'][0][index] for index in paths[i]]
#形成完整串列
result = [k for c,t in zip(string,tags)]
results.append(result)
#獲取預測值
return results
2.train.py
# -*- coding: utf-8 -*-
"""
Created on Thu Jan 7 18:57:23 2021
@author: xiuzhang
"""
import tensorflow as tf
from data_utils import BatchManager
import pickle
from model import Model
import time
#-----------------------------功能：讀取字典---------------------------
dict_file = 'data/dict.pkl'
def get_dict(path):
with open(path, 'rb') as f:
data = pickle.load(f)
return data
#-----------------------------功能：訓練函式---------------------------
batch_size = 20
def train():
#呼叫已定義的方法獲取處理好的資料集
train_manager = BatchManager(batch_size=20, name='train')
print('train:', type(train_manager)) #<class 'data_utils.BatchManager'>
test_manager = BatchManager(batch_size=100, name='test')
#讀取字典
mapping_dict = get_dict(dict_file)
print('train:', len(mapping_dict)) #6
print('計算六元組個數')
print('字:', len(mapping_dict['word'][0])) #1663
print('邊界:', len(mapping_dict['bound'][0])) #5
print('詞性:', len(mapping_dict['flag'][0])) #56
print('偏旁:', len(mapping_dict['radical'][0])) #227
print('拼音:', len(mapping_dict['pinyin'][0])) #989
print('類別:', len(mapping_dict['label'][0]),'\n') #31
#-------------------------搭建模型---------------------------
#實體化模型 執行init初始化方法model核心函式：
# 1.get_logits：傳遞給網路 計算模型輸出值
# 2.loss：計算損失值
#-----------------------------------------------------------
model = Model(mapping_dict)
print("---------------模型構建成功---------------------\n")
#初始化訓練
init = tf.global_variables_initializer()
with tf.Session() as sess:
sess.run(init)
for i in range(10):
j = 1
#呼叫iter_batch函式 迭代程序可以讓梯度下降在不斷嘗試找到最優解
for batch in train_manager.iter_batch(shuffle=True): #亂序
#時間計算
start = time.time()
#呼叫自定義函式
loss = model.run_step(sess,batch)
end = time.time()
#每10批輸出
if j % 10==0:
#第幾輪 每批數量 多少批次 損失 消耗時間 剩余估計時間
print('epoch:{},step:{}/{},loss:{},elapse:{},estimate:{}'.format(
i+1,j,train_manager.len_data,
loss,(end-start),
(end-start)*(train_manager.len_data-j)))
j += 1
"""
#print(len(batch)) #6個型別
#print(len(batch[0]),len(batch[1]),len(batch[2])) #20個
#每次獲取一個批次的資料 feed_dict喂資料 placeholder用于接收神經網路資料
_,loss = sess.run([model.train_op,model.cost],feed_dict={
model.char_inputs : batch[0],
model.bound_inputs : batch[2],
model.flag_inputs : batch[3],
model.radical_inputs : batch[4],
model.pinyin_inputs : batch[5],
model.targets : batch[1] #注意順序
})
print('loss:{}'.format(loss))
#InvalidArgumentError: indices[0,2] = 7 is not in [0, 5)
#注意:feed_dict對應資料必須一致,最早CSV檔案label為第2列,所有檔案寫回傳值順序一致
#data_utils.py: char, target, bound, flag, radical, pinyin = line
"""
#--------------------------------------------------
#每迭代一輪進行預測
for batch in test_manager.iter_batch(shuffle=True):
print(model.predict(sess,batch))
#----------------------------功能:主函式---------------------------------
if __name__ == '__main__':
train()
3.data_utils.py
#encoding:utf-8
import pandas as pd
import pickle
import numpy as np
from tqdm import tqdm
import os
import math
import random
#功能:獲取值對應的下標 引數為串列和字符
def item2id(data,w2i):
#x在字典中直接獲取 不在字典中回傳UNK
return [w2i[x] if x in w2i else w2i['UNK'] for x in data]
#----------------------------功能:拼接檔案---------------------------------
def get_data_with_windows(name='train'):
#讀取prepare_data.py生成的dict.pkl檔案 存盤字典{類別:下標}
with open(f'data/dict.pkl', 'rb') as f:
map_dict = pickle.load(f) #加載字典
#存盤所有資料
results = []
root = os.path.join('data/prepare/'+name)
files = list(os.listdir(root))
print(files)
#['10.csv', '11.csv', '12.csv',.....]
#獲取所有檔案 進度條
for file in tqdm(files):
all_data = []
path = os.path.join(root, file)
samples = pd.read_csv(path,sep=',')
max_num = len(samples)
#獲取sep換行分隔符下標 -1 20 40 60
sep_index = [-1]+samples[samples['word']=='sep'].index.tolist()+[max_num]
#print(sep_index)
#[-1, 83, 92, 117, 134, 158, 173, 200,......]
#----------------------------------------------------------------------
# 獲取句子并將句子全部都轉換成id
#----------------------------------------------------------------------
for i in range(len(sep_index)-1):
start = sep_index[i] + 1 #0 (-1+1)
end = sep_index[i+1] #20
data = []
#每個特征進行處理
for feature in samples.columns: #訪問每列
#通過函式item2id獲取下標 map_dict兩個值(串列和字典) 獲取第二個值
data.append(item2id(list(samples[feature])[start:end],map_dict[feature][1]))
#將每句話的串列合成
all_data.append(data)
#----------------------------------------------------------------------
# 資料增強
#----------------------------------------------------------------------
#前后兩個句子拼接 每個句子六個元素(漢字、邊界、詞性、類別、偏旁、拼音)
two = []
for i in range(len(all_data)-1):
first = all_data[i]
second = all_data[i+1]
two.append([first[k]+second[k] for k in range(len(first))]) #六個元素
three = []
for i in range(len(all_data)-2):
first = all_data[i]
second = all_data[i+1]
third = all_data[i+2]
three.append([first[k]+second[k]+third[k] for k in range(len(first))])
#回傳所有結果
results.extend(all_data+two+three)
#return results
#資料存盤至本地 每次呼叫時間成本過大
with open(f'data/'+name+'.pkl', 'wb') as f:
pickle.dump(results, f)
#----------------------------功能:批處理---------------------------------
class BatchManager(object):
def __init__(self, batch_size, name='train'):
#呼叫函式拼接檔案
#data = get_data_with_windows(name)
#讀取檔案
with open(f'data/'+name+'.pkl', 'rb') as f:
data = pickle.load(f)
print(len(data)) #265455句話
print(len(data[0])) #6種類別
print(len(data[0][0])) #第一句包含字的數量 83
print("原始資料:", data[0])
#資料批處理
self.batch_data = self.sort_and_pad(data, batch_size)
self.len_data = len(self.batch_data)
def sort_and_pad(self, data, batch_size):
#計算總批次數量 26546
num_batch = int(math.ceil(len(data) / batch_size))
#按照句子長度排序
sorted_data = sorted(data, key=lambda x: len(x[0]))
batch_data = list()
#獲取一個批次的資料
for i in range(num_batch):
batch_data.append(self.pad_data(sorted_data[i*int(batch_size) : (i+1)*int(batch_size)]))
print("分批輸出:", batch_data[100])
return batch_data
@staticmethod
def pad_data(data_):
#定義變數
chars = []
bounds = []
flags = []
radicals = []
pinyins = []
targets = []
#print("每個批次句子個數:", len(data_)) #10
#print("每個句子包含元素個數:", len(data_[0])) #6
#print("輸出data:", data_)
max_length = max([len(sentence[0]) for sentence in data_]) #值為1
#print(max_length)
#每個批次共有十組資料 每組資料均為六個元素
for line in data_:
#char, bound, flag, target, radical, pinyin = line
char, target, bound, flag, radical, pinyin = line
padding = [0] * (max_length - len(char)) #計算補充字符數量
#注意char和chars不要寫錯 否則造成遞回回圈賦值錯誤
chars.append(char + padding)
targets.append(target + padding)
bounds.append(bound + padding)
flags.append(flag + padding)
radicals.append(radical + padding)
pinyins.append(pinyin + padding)
return [chars, targets, bounds, flags, radicals, pinyins]
#每次使用一個批次資料
def iter_batch(self, shuffle=False):
if shuffle: #亂序
random.shuffle(self.batch_data)
for idx in range(self.len_data):
yield self.batch_data[idx]
#-------------------------------功能:主函式--------------------------------------
if __name__ == '__main__':
#1.拼接檔案(第一次執行 后續可注釋)
#get_data_with_windows('train')
#2.分批處理
train_data = BatchManager(10, 'train')
#3.接著處理下測驗集資料
get_data_with_windows('test')
4.prepare_data.py
#encoding:utf-8
import os
import pickle
import pandas as pd
from collections import Counter
from data_process import split_text
from tqdm import tqdm #進度條 pip install tqdm
#詞性標注
import jieba.posseg as psg
#獲取字的偏旁和拼音
from cnradical import Radical, RunOption
#洗掉目錄
import shutil
#隨機劃分訓練集和測驗集
from random import shuffle
#遍歷檔案包
from glob import glob
train_dir = "train_data"
#----------------------------功能:文本預處理---------------------------------
def process_text(idx, split_method=None, split_name='train'):
"""
功能: 讀取文本并切割,接著打上標記及提取詞邊界、詞性、偏旁部首、拼音等特征
param idx: 檔案的名字 不含擴展名
param split_method: 切割文本方法
param split_name: 存盤資料集 默認訓練集, 還有測驗集
return
"""
#定義字典 保存所有字的標記、邊界、詞性、偏旁部首、拼音等特征
data = {}
#--------------------------------------------------------------------
# 獲取句子
#--------------------------------------------------------------------
if split_method is None:
#未給文本分割函式 -> 讀取檔案
with open(f'data/{train_dir}/{idx}.txt', encoding='utf8') as f: #f表示檔案路徑
texts = f.readlines()
else:
#給出文本分割函式 -> 按函式分割
with open(f'data/{train_dir}/{idx}.txt', encoding='utf8') as f:
outfile = f'data/train_data_pro/{idx}_pro.txt'
print(outfile)
texts = f.read()
texts = split_method(texts, outfile)
#提取句子
data['word'] = texts
print(texts)
#--------------------------------------------------------------------
# 獲取標簽(物體類別、起始位置)
#--------------------------------------------------------------------
#初始時將所有漢字標記為O
tag_list = ['O' for s in texts for x in s] #雙層回圈遍歷每句話中的漢字
#讀取ANN檔案獲取每個物體的型別、起始位置和結束位置
tag = pd.read_csv(f'data/{train_dir}/{idx}.ann', header=None, sep='\t') #Pandas讀取 分隔符為tab鍵
#0 T1 Disease 1845 1850 1型糖尿病
for i in range(tag.shape[0]): #tag.shape[0]為行數
tag_item = tag.iloc[i][1].split(' ') #每一行的第二列 空格分割
#print(tag_item)
#存在某些物體包括兩段位置區間 僅獲取起始位置和結束位置
cls, start, end = tag_item[0], int(tag_item[1]), int(tag_item[-1])
#print(cls,start,end)
#對tag_list進行修改
tag_list[start] = 'B-' + cls
for j in range(start+1, end):
tag_list[j] = 'I-' + cls
#斷言 兩個長度不一致報錯
assert len([x for s in texts for x in s])==len(tag_list)
#print(len([x for s in texts for x in s]))
#print(len(tag_list))
#--------------------------------------------------------------------
# 分割后句子匹配標簽
#--------------------------------------------------------------------
tags = []
start = 0
end = 0
#遍歷文本
for s in texts:
length = len(s)
end += length
tags.append(tag_list[start:end])
start += length
print(len(tags))
#標簽資料存盤至字典中
data['label'] = tags
#--------------------------------------------------------------------
# 提取詞性和詞邊界
#--------------------------------------------------------------------
#初始標記為M
word_bounds = ['M' for item in tag_list] #邊界 M表示中間
word_flags = [] #詞性
#分詞
for text in texts:
#帶詞性的結巴分詞
for word, flag in psg.cut(text):
if len(word)==1: #1個長度詞
start = len(word_flags)
word_bounds[start] = 'S' #單個字
word_flags.append(flag)
else:
start = len(word_flags)
word_bounds[start] = 'B' #開始邊界
word_flags += [flag]*len(word) #保證詞性和字一一對應
end = len(word_flags) - 1
word_bounds[end] = 'E' #結束邊界
#存盤
bounds = []
flags = []
start = 0
end = 0
for s in texts:
length = len(s)
end += length
bounds.append(word_bounds[start:end])
flags.append(word_flags[start:end])
start += length
data['bound'] = bounds
data['flag'] = flags
#--------------------------------------------------------------------
# 獲取拼音和偏旁特征
#--------------------------------------------------------------------
radical = Radical(RunOption.Radical) #提取偏旁部首
pinyin = Radical(RunOption.Pinyin) #提取拼音
#提取拼音和偏旁 None用特殊符號替代UNK
radical_out = [[radical.trans_ch(x) if radical.trans_ch(x) is not None else 'UNK' for x in s] for s in texts]
pinyin_out = [[pinyin.trans_ch(x) if pinyin.trans_ch(x) is not None else 'UNK' for x in s] for s in texts]
#賦值
data['radical'] = radical_out
data['pinyin'] = pinyin_out
#--------------------------------------------------------------------
# 存盤資料
#--------------------------------------------------------------------
#獲取樣本數量
num_samples = len(texts) #行數
num_col = len(data.keys()) #列數 字典自定義類別數 6
print(num_samples)
print(num_col)
dataset = []
for i in range(num_samples):
records = list(zip(*[list(v[i]) for v in data.values()])) #壓縮
dataset += records+[['sep']*num_col] #每處理一句話sep分割
#records = list(zip(*[list(v[0]) for v in data.values()]))
#for r in records:
# print(r)
#最后一行sep洗掉
dataset = dataset[:-1]
#轉換成dataframe 增加表頭
dataset = pd.DataFrame(dataset,columns=data.keys())
#保存檔案 測驗集 訓練集
save_path = f'data/prepare/{split_name}/{idx}.csv'
dataset.to_csv(save_path,index=False,encoding='utf-8')
#--------------------------------------------------------------------
# 處理換行符 w表示一個字
#--------------------------------------------------------------------
def clean_word(w):
if w=='\n':
return 'LB'
if w in [' ','\t','\u2003']: #中文空格\u2003
return 'SPACE'
if w.isdigit(): #將所有數字轉換為一種符號 數字訓練會造成干擾
return 'NUM'
return w
#對dataframe應用函式
dataset['word'] = dataset['word'].apply(clean_word)
#存盤資料
dataset.to_csv(save_path,index=False,encoding='utf-8')
#return texts, tags, bounds, flags
#return texts[0], tags[0], bounds[0], flags[0], radical_out[0], pinyin_out[0]
#----------------------------功能:預處理所有文本---------------------------------
def multi_process(split_method=None,train_ratio=0.8):
"""
功能: 對所有文本盡心預處理操作
param split_method: 切割文本方法
param train_ratio: 訓練集和測驗集劃分比例
return
"""
#洗掉目錄
if os.path.exists('data/prepare/'):
shutil.rmtree('data/prepare/')
#創建目錄
if not os.path.exists('data/prepare/train/'):
os.makedirs('data/prepare/train/')
os.makedirs('data/prepare/test/')
#獲取所有檔案名
idxs = set([file.split('.')[0] for file in os.listdir('data/'+train_dir)])
idxs = list(idxs)
#隨機劃分訓練集和測驗集
shuffle(idxs) #打亂順序
index = int(len(idxs)*train_ratio) #獲取訓練集的截止下標
#獲取訓練集和測驗集檔案名集合
train_ids = idxs[:index]
test_ids = idxs[index:]
#--------------------------------------------------------------------
# 引入多行程
#--------------------------------------------------------------------
#執行緒池方式呼叫
import multiprocessing as mp
num_cpus = mp.cpu_count() #獲取機器CPU的個數
pool = mp.Pool(num_cpus)
results = []
#訓練集處理
for idx in train_ids:
result = pool.apply_async(process_text, args=(idx,split_method,'train'))
results.append(result)
#測驗集處理
for idx in test_ids:
result = pool.apply_async(process_text, args=(idx,split_method,'test'))
results.append(result)
#關閉行程池
pool.close()
pool.join()
[r.get for r in results]
#----------------------------功能:生成映射字典---------------------------------
#統計函式：串列、頻率計算閾值
def mapping(data,threshold=10,is_word=False,sep='sep',is_label=False):
#統計串列data中各種型別的個數
count = Counter(data)
#洗掉之前自定義的sep換行符
if sep is not None:
count.pop(sep)
#判斷是漢字 未登錄詞處理 出現頻率較少 設定為Unknown
if is_word:
#設定下列兩個詞頻次 排序靠前
count['PAD'] = 100000001 #填充字符 保證長度一致
count['UNK'] = 100000000 #未知標記
#降序排列
data = sorted(count.items(),key=lambda x:x[1], reverse=True)
#去除頻率小于threshold的元素
data = [x[0] for x in data if x[1]>=threshold]
#轉換成字典
id2item = data
item2id = {id2item[i]:i for i in range(len(id2item))}
elif is_label:
#label標簽不加PAD
data = sorted(count.items(),key=lambda x:x[1], reverse=True)
data = [x[0] for x in data]
id2item = data
item2id = {id2item[i]:i for i in range(len(id2item))}
else:
count['PAD'] = 100000001
data = sorted(count.items(),key=lambda x:x[1], reverse=True)
data = [x[0] for x in data]
id2item = data
item2id = {id2item[i]:i for i in range(len(id2item))}
return id2item, item2id
#生成映射字典
def get_dict():
#獲取所有內容
all_w = [] #漢字
all_label = [] #類別
all_bound = [] #邊界
all_flag = [] #詞性
all_radical = [] #偏旁
all_pinyin = [] #拼音
#讀取檔案
for file in glob('data/prepare/train/*.csv') + glob('data/prepare/test/*.csv'):
df = pd.read_csv(file,sep=',')
all_w += df['word'].tolist()
all_label += df['label'].tolist()
all_bound += df['bound'].tolist()
all_flag += df['flag'].tolist()
all_radical += df['radical'].tolist()
all_pinyin += df['pinyin'].tolist()
#保存回傳結果 字典
map_dict = {}
#呼叫統計函式
map_dict['word'] = mapping(all_w,threshold=20,is_word=True)
map_dict['label'] = mapping(all_label,is_label=True)
map_dict['bound'] = mapping(all_bound)
map_dict['flag'] = mapping(all_flag)
map_dict['radical'] = mapping(all_radical)
map_dict['pinyin'] = mapping(all_pinyin)
#字典保存內容
#return map_dict
#保存字典資料至檔案
with open(f'data/dict.pkl', 'wb') as f:
pickle.dump(map_dict,f)
#-------------------------------功能:主函式--------------------------------------
if __name__ == '__main__':
#print(process_text('0',split_method=split_text,split_name='train'))
#1.多執行緒處理文本
#multi_process(split_text)
#2.生成映射字典
#print(get_dict())
get_dict()
#3.讀取get_dict函式保存的字典檔案
with open(f'data/dict.pkl', 'rb') as f:
data = pickle.load(f)
print(data['bound'])
5.data_process.py
#encoding:utf-8
import os
import re
#----------------------------功能:獲取物體類別及個數---------------------------------
def get_entities(dirPath):
entities = {} #存盤物體類別
files = os.listdir(dirPath) #遍歷路徑
#獲取所有檔案的名字并去重 0.ann => 0
filenames = set([file.split('.')[0] for file in files])
filenames = list(filenames)
#print(filenames)
#重新構造ANN檔案名并遍歷檔案
for filename in filenames:
path = os.path.join(dirPath, filename+".ann")
#print(path)
#讀檔案
with open(path, 'r', encoding='utf8') as f:
for line in f.readlines():
#TAB鍵分割獲取物體型別
name = line.split('\t')[1]
#print(name)
value = name.split(' ')[0]
#print(value)
#物體加入字典并統計個數
if value in entities:
entities[value] += 1 #在物體集合中數量加1
else:
entities[value] = 1 #創建鍵值且值為1
#回傳物體集
return entities
#----------------------------功能:命名物體BIO標注--------------------------------
def get_labelencoder(entities):
#排序
entities = sorted(entities.items(), key=lambda x: x[1], reverse=True)
print(entities)
#獲取物體類別名稱
entities = [x[0] for x in entities]
print(entities)
#標記物體
id2label = []
id2label.append('O')
#生成物體標記
for entity in entities:
id2label.append('B-'+entity)
id2label.append('I-'+entity)
#字典鍵值生成
label2id = {id2label[i]:i for i in range(len(id2label))}
return id2label, label2id
#-------------------------功能:自定義分隔符文本分割------------------------------
def split_text(text, outfile):
#分割后的下標
split_index = []
#檔案寫入
fw = open(outfile, 'w', encoding='utf8')
#--------------------------------------------------------------------
# 文本分割
#--------------------------------------------------------------------
#第一部分 按照符號分割
pattern = '，|，|,|;|；|？|\?|\.'
#獲取字符的下標位置
for m in re.finditer(pattern, text):
"""
print(m)
start = m.span()[0] #標點符號位置
print(text[start])
start = m.span()[0] - 5
end = m.span()[1] + 5
print('****', text[start:end], '****')
"""
#特殊符號下標
idx = m.span()[0]
#判斷是否斷句 contniue表示不能直接分割句子
if text[idx-1]=='\n': #當前符號前是換行符
continue
if text[idx-1].isdigit() and text[idx+1].isdigit(): #前后都是數字或數字+空格
continue
if text[idx-1].isdigit() and text[idx+1].isspace() and text[idx+2].isdigit():
continue
if text[idx-1].islower() and text[idx+1].islower(): #前后都是小寫字母
continue
if text[idx-1].isupper() and text[idx+1].isupper(): #前后都是大寫字母
continue
if text[idx-1].islower() and text[idx+1].isdigit(): #前面是小寫字母 后面是數字
continue
if text[idx-1].isupper() and text[idx+1].isdigit(): #前面是大寫字母 后面是數字
continue
if text[idx-1].isdigit() and text[idx+1].islower(): #前面是數字 后面是小寫字母
continue
if text[idx-1].isdigit() and text[idx+1].isupper(): #前面是數字 后面是大寫字母
continue
if text[idx+1] in set('.，;；,，'): #前后都是標點符號
continue
if text[idx-1].isspace() and text[idx-2].isspace() and text[idx-3].isupper():
continue #HBA1C ，兩個空格+字母
if text[idx-1].isspace() and text[idx-3].isupper():
continue
#print('****', text[idx-20:idx+20], '****')
#將分句的下標存盤至串列中 -> 標點符號后面的字符
split_index.append(idx+1)
#--------------------------------------------------------------------
#第二部分 按照自定義符號分割
#下列形式進行句子分割
pattern2 = '\([一二三四五六七八九十零]\)|[一二三四五六七八九十零]、|'
pattern2 += '注:|附錄 |表 \d|Tab \d+|\[摘要\]|\[提要\]|表\d[^，，,;；]+?\n|'
pattern2 += '圖 \d|Fig \d|\[Abdtract\]|\[Summary\]|前 言|【摘要】|【關鍵詞】|'
pattern2 += '結 果|討 論|and |or |with |by |because of |as well as '
#print(pattern2)
for m in re.finditer(pattern2, text):
idx = m.span()[0]
#print('****', text[idx-20:idx+20], '****')
#連接詞位于單詞中間不能分割 如 goodbye
if (text[idx:idx+2] in ['or','by'] or text[idx:idx+3]=='and' or text[idx:idx+4]=='with')\
and (text[idx-1].islower() or text[idx-1].isupper()):
continue
split_index.append(idx) #注意這里不加1 找到即分割
#--------------------------------------------------------------------
#第三部分 中文字符+數字分割
#判斷序列且包含漢字的分割(2.接下來...) 同時小數不進行切割
pattern3 = '\n\d\.' #數字+點
for m in re.finditer(pattern3, text):
idx = m.span()[0]
if ischinese(text[idx+3]): #第四個字符為中文漢字 含換行
#print('****', text[idx-20:idx+20], '****')
split_index.append(idx+1)
#換行+數字+括號 (1)總體治療原則:淤在選擇降糖藥物時
for m in re.finditer('\n\(\d\)', text):
idx = m.span()[0]
split_index.append(idx+1)
#--------------------------------------------------------------------
#獲取句子分割下標后進行排序操作 增加第一行和最后一行
split_index = sorted(set([0, len(text)] + split_index))
split_index = list(split_index)
#print(split_index)
#計算機最大值和最小值
lens = [split_index[i+1]-split_index[i] for i in range(len(split_index)-1)]
#print(max(lens), min(lens))
#--------------------------------------------------------------------
# 長短句處理
#--------------------------------------------------------------------
#遍歷每一個句子 (一)xxxx 分割
other_index = []
for i in range(len(split_index)-1):
begin = split_index[i]
end = split_index[i+1]
#print("-----", text[begin:end])
#print(begin, end)
if (text[begin] in '一二三四五六七八九十零') or \
(text[begin]=='(' and text[begin+1] in '一二三四五六七八九十零'):
for j in range(begin,end):
if text[j]=='\n':
other_index.append(j+1)
#補充+排序
split_index += other_index
split_index = list(sorted(set([0, len(text)] + split_index)))
#--------------------------------------------------------------------
#第一部分 長句處理：句子長度超過150進行拆分
other_index = []
for i in range(len(split_index)-1):
begin = split_index[i]
end = split_index[i+1]
other_index.append(begin)
#句子長度超過150切割 并且最短15個字符
if end-begin>150:
for j in range(begin,end):
#這一次下標位置比上一次超過15分割
if(j+1-other_index[-1])>15:
#換行分割
if text[j]=='\n':
other_index.append(j+1)
#空格+前后數字
if text[j]==' ' and text[j-1].isnumeric() and text[j+1].isnumeric():
other_index.append(j+1)
split_index += other_index
split_index = list(sorted(set([0, len(text)] + split_index)))
#--------------------------------------------------------------------
#第二部分 洗掉空格的句子
for i in range(1, len(split_index)-1):
idx = split_index[i]
#當前下標和上一個下標對比 如果等于空格繼續比較
while idx>split_index[i-1]-1 and text[idx-1].isspace():
idx -= 1
split_index[i] = idx
split_index = list(sorted(set([0, len(text)] + split_index)))
#--------------------------------------------------------------------
#第三部分 短句處理-拼接
temp_idx = []
i = 0
while i<(len(split_index)-1):
begin = split_index[i]
end = split_index[i+1]
#先統計句子中中文字符和英文字符個數
num_ch = 0
num_en = 0
if end - begin <15:
for ch in text[begin:end]:
if ischinese(ch):
num_ch += 1
elif ch.islower() or ch.isupper():
num_en += 1
if num_ch + 0.5*num_en>5: #大于5說明長度夠用
temp_idx.append(begin)
i += 1 #注意break前i加1 否則死回圈
break
#長度小于等于5和后面的句子合并
if num_ch + 0.5*num_en<=5:
temp_idx.append(begin)
i += 2
else:
temp_idx.append(begin) #大于15直接添加下標
i += 1
split_index = list(sorted(set([0, len(text)] + temp_idx)))
#查看句子長度 由于存在\n換行一個字符
lens = [split_index[i+1]-split_index[i] for i in range(len(split_index)-1)][:-1] #洗掉最后一個換行
print(max(lens), min(lens))
#for i in range(len(split_index)-1):
# print(i, '****', text[split_index[i]:split_index[i+1]])
#存盤結果
result = []
for i in range(len(split_index)-1):
result.append(text[split_index[i]:split_index[i+1]])
fw.write(text[split_index[i]:split_index[i+1]])
fw.close()
#檢查：預處理后字符是否減少
s = ''
for r in result:
s += r
assert len(s)==len(text) #斷言
return result
#---------------------------功能:判斷字符是不是漢字-------------------------------
def ischinese(char):
if '\u4e00' <=char <= '\u9fff':
return True
return False
#-------------------------------功能:主函式--------------------------------------
if __name__ == '__main__':
dirPath = "data/train_data"
outPath = 'data/train_data_pro'
#獲取物體類別及個數
entities = get_entities(dirPath)
print(entities)
print(len(entities))
#完成物體標記 串列 字典
#得到標簽和下標的映射
label, label_dic = get_labelencoder(entities)
print(label)
print(len(label))
print(label_dic, '\n\n')
#遍歷路徑
files = os.listdir(dirPath)
filenames = set([file.split('.')[0] for file in files])
filenames = list(filenames)
for filename in filenames:
path = os.path.join(dirPath, filename+".txt") #TXT檔案
outfile = os.path.join(outPath, filename+"_pro.txt")
#print(path)
with open(path, 'r', encoding='utf8') as f:
text = f.read()
#分割文本
print(path)
split_text(text, outfile)
print("\n")
七.總結
寫到這里，這篇文章就介紹結束了，希望對您有所幫助，文章雖然很冗余，但還是能學到知識，尤其是資料預處理和BiLSTM構建知識，后續隨著作者深入，會分享更簡潔的命名物體識別代碼，繼續加油~
希望您喜歡這篇文章，從看視頻到撰寫代碼，我真的寫了一周時間，再次感謝視頻的作者白老師及B站UP主，真心希望這篇文章對您有所幫助，加油~
- https://github.com/eastmountyxz/AI-for-Keras
(By:Eastmount 2021-01-05 周二寫于武漢 http://blog.csdn.net/eastmount/ )
2020年8月18新開的“娜璋AI安全之家”，主要圍繞Python大資料分析、網路空間安全、人工智能、Web滲透及攻防技術進行講解，同時分享CCF、SCI、南核北核論文的演算法實作，娜璋之家會更加系統，并重構作者的所有文章，從零講解Python和安全，寫了近十年文章，真心想把自己所學所感所做分享出來，還請各位多多指教，真誠邀請您的關注！謝謝，
參考文獻：
- https://www.bilibili.com/video/BV1Z5411477j - 誰用了我的白樺林
- 肖仰華《知識圖譜概念與技術》
- NLP在線醫生-BiLSTM+CRF命名物體識別 - 閣下兄
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/246854.html
標籤：python