天池下的瑞金醫院MMC人工智能輔助構建知識圖譜

淺談知識圖譜------天池下的瑞金醫院MMC人工智能輔助構建知識圖譜

前言

數據說明

問題

網絡模型和效果展示

代碼

實體的定義和處理

句子的切分和處理

代碼和數據集：

前言

知識圖譜是個很大的概念，可惜我沒數據，借用瑞金醫院的數據集，來談下命名識別。

數據說明

數據使用 brat 進行標注，每個 .txt 文件對應一個 .ann 標注文件。

txt文件對應一篇糖尿病下的論文，

ann文件有3列，以 \t 分隔，第一列為實體編號，第二列為實體類別，第三列為實體位置信息。實體位置信息共3列， 以空格分隔，分別代表實體的開始位置，結束位置，實體文本。

問題

這里我引用冠軍隊伍的代碼，他們當時所面臨的問題如下：

（1）他們是對一篇文章去做實體標注，文章的字數可能很長(幾千到上萬字)，不可能直接輸入到一個 RNN 中；

（2）樣本中文章可能由于格式轉換的一些原因，沒有一個很好的句子邊界，甚至一個詞匯當中存在換行符 \n 或者句號 的情況，因此用換行 符或者句號去切割句子不一定合適。

（3）如果按照固定窗口大小的滑動窗口去切句子，剛好把一個實體切分成2個部分怎么辦？

中文文本，面臨是否要分詞的選擇；

下面是他們的解決方案：

網絡模型和效果展示

網絡模型為了便于上下文的關聯采用了雙向的lstm，為了使滑動的時候不丟到相關聯的詞語采用了一層CRF，作為最后最后一層的預測。

代碼

代碼主要分為三個部分，實體的定義和處理、句子的切分和處理、模型的搭建，除此之外還有預測評估的部分

實體的定義和處理

class Entity(object): def __init__(self, ent_id, category, start_pos, end_pos, text): self.ent_id = ent_id self.category = category self.start_pos = start_pos self.end_pos = end_pos self.text = text def __gt__(self, other): return self.start_pos > other.start_pos def offset(self, offset_val): return Entity(self.ent_id, self.category, self.start_pos + offset_val, self.end_pos + offset_val, self.text) def __repr__(self): return '({}, {}, ({}, {}), {})'.format(self.ent_id, self.category, self.start_pos, self.end_pos, self.text)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

class Entities(object): def __init__(self, ents): self.ents = sorted(ents) self.ent_dict = dict(zip([ent.ent_id for ent in ents], ents)) def __getitem__(self, key): if isinstance(key, int) or isinstance(key, slice): return self.ents[key] else: return self.ent_dict.get(key, None) def offset(self, offset_val): ents = [ent.offset(offset_val) for ent in self.ents] return Entities(ents) def vectorize(self, vec_len, cate2idx): res_vec = np.zeros(vec_len, dtype=int) for ent in self.ents: res_vec[ent.start_pos: ent.end_pos] = cate2idx[ent.category] return res_vec def find_entities(self, start_pos, end_pos): res = [] for ent in self.ents: if ent.start_pos > end_pos: break sp, ep = (max(start_pos, ent.start_pos), min(end_pos, ent.end_pos)) if ep > sp: new_ent = Entity(ent.ent_id, ent.category, sp, ep, ent.text[:(ep - sp)]) res.append(new_ent) return Entities(res) def merge(self): merged_ents = [] for ent in self.ents: if len(merged_ents) == 0: merged_ents.append(ent) elif (merged_ents[-1].end_pos == ent.start_pos and merged_ents[-1].category == ent.category): merged_ent = Entity(ent_id=merged_ents[-1].ent_id, category=ent.category, start_pos=merged_ents[-1].start_pos, end_pos=ent.end_pos, text=merged_ents[-1].text + ent.text) merged_ents[-1] = merged_ent else: merged_ents.append(ent) return Entities(merged_ents)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

句子的切分和處理

data_dir = 'ruijin_round1_train2_20181022/' ent2idx = dict(zip(ENTITIES, range(1, len(ENTITIES) + 1))) idx2ent = dict([(v, k) for k, v in ent2idx.items()]) # print(idx2ent) docs = Documents(data_dir=data_dir) # ShuffleSplit（） 隨機排列交叉驗證，生成一個用戶給定數量的獨立的訓練/測試數據劃分。樣例首先被打散然后劃分為一對訓練測試集合。 # n_splits:劃分訓練集、測試集的次數，默認為10 # test_size: 測試集比例或樣本數量， # random_state:隨機種子值，默認為None，可以通過設定明確的random_state，使得偽隨機生成器的結果可以重復。 rs = ShuffleSplit(n_splits=1, test_size=20, random_state=2018) train_doc_ids, test_doc_ids = next(rs.split(docs)) train_docs, test_docs = docs[train_doc_ids], docs[test_doc_ids] num_cates = max(ent2idx.values()) + 1 sent_len = 64 vocab_size = 3000 emb_size = 100 sent_pad = 10 sent_extrator = SentenceExtractor(window_size=sent_len, pad_size=sent_pad) train_sents = sent_extrator(train_docs) test_sents = sent_extrator(test_docs) train_data = Dataset(train_sents, cate2idx=ent2idx) train_data.build_vocab_dict(vocab_size=vocab_size) test_data = Dataset(test_sents, word2idx=train_data.word2idx, cate2idx=ent2idx)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

class Sentence(object): """ 定義被切分的句子的類： text:句子的文本 doc_id:句子所述文檔id offset:句子相對文檔的偏移距離 ents:句子包含的實體列表 """ def __init__(self, doc_id, offset, text, ents): self.text = text self.doc_id = doc_id self.offset = offset self.ents = ents def __repr__(self): """ 內部魔法函數：以text顯示類 :return: """ return self.text def __gt__(self, other): #內部魔法函數：按類的offset偏移距離對類進行排序 return self.offset > other.offset def __getitem__(self, key): """ 內部魔法函數:預測結果評估時，去除句子兩端延申的部分 :param key: :return: """ if isinstance(key, int): return self.text[key] if isinstance(key, slice): text = self.text[key] start = key.start or 0 stop = key.stop or len(self.text) if start < 0: start += len(self.text) if stop < 0: stop += len(self.text) #改變實體相對于句子的偏移距離 ents = self.ents.find_entities(start, stop).offset(-start) #改變句子相對于文檔的偏移距離 offset = self.offset + start return Sentence(self.doc_id, offset, text, ents) def _repr_html_(self): """ 內部函數：網頁顯示不同的實體以不同的顏色區分 :return: """ ents = [] for ent in self.ents: ents.append({'start': ent.start_pos, 'end': ent.end_pos, 'label': ent.category}) ex = {'text': self.text, 'ents': ents, 'title': None, 'settings': {}} return displacy.render(ex, style='ent', options={'colors': COLOR_MAP}, manual=True, minify=True) class SentenceExtractor(object): #句子切分器，窗口為windows，兩端分別延申pad_size def __init__(self, window_size=50, pad_size=10): self.window_size = window_size self.pad_size = pad_size def extract_doc(self, doc): #句子切分函數，切分的時候注意每個切分的句子相對于文檔的偏移距離，預測的時候還需要還原 num_sents = math.ceil(len(doc.text) / self.window_size) doc = doc.pad(pad_left=self.pad_size, pad_right=num_sents * self.window_size - len(doc.text) + self.pad_size) sents = [] for cur_idx in range(self.pad_size, len(doc.text) - self.pad_size, self.window_size): sent_text = doc.text[cur_idx - self.pad_size: cur_idx + self.window_size + self.pad_size] ents = [] for ent in doc.ents.find_entities(start_pos=cur_idx - self.pad_size, end_pos=cur_idx + self.window_size + self.pad_size): ents.append(ent.offset(-cur_idx + self.pad_size)) sent = Sentence(doc.doc_id, offset=cur_idx - 2 * self.pad_size, text=sent_text, ents=Entities(ents)) sents.append(sent) return sents def __call__(self, docs): #內部函數：將類當成函數形式的調用 sents = [] for doc in docs: sents += self.extract_doc(doc) return sents

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

模型的構建

def build_lstm_crf_model(num_cates, seq_len, vocab_size, model_opts=dict()): opts = { 'emb_size': 256, 'emb_trainable': True, 'emb_matrix': None, 'lstm_units': 256, 'optimizer': keras.optimizers.Adam() } opts.update(model_opts) input_seq = Input(shape=(seq_len,), dtype='int32') if opts.get('emb_matrix') is not None: embedding = Embedding(vocab_size, opts['emb_size'], weights=[opts['emb_matrix']], trainable=opts['emb_trainable']) else: embedding = Embedding(vocab_size, opts['emb_size']) x = embedding(input_seq) lstm = LSTM(opts['lstm_units'], return_sequences=True) x = Bidirectional(lstm)(x) crf = CRF(num_cates, sparse_target=True) output = crf(x) model = Model(input_seq, output) model.compile(opts['optimizer'], loss=crf.loss_function, metrics=[crf.accuracy]) return model

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

代碼和數據集：

我把代碼和數據集打包了

鏈接：https://pan.baidu.com/s/1mvjPuoGRChTpIqCYrLB6VA

提取碼：z9tz

復制這段內容后打開百度網盤手機App，操作更方便哦–來自百度網盤超級會員V3的分享

醫療 知識圖譜

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。