CMeKG代码解读(以项目为导向从零开始学习知识图谱)（一）

作者从零开始学习和知识图谱有关技术和内容，而本文的核心内容是对CMeKG的python代码进行学习和解读，供大家讨论参考共同进步。

CMeKG（Chinese Medical Knowledge Graph）是利用自然语言处理与文本挖掘技术，基于大规模医学文本数据，以人机结合的方式研发的中文医学知识图谱。

项目来源：

中文医学知识图谱CMeKG2.0版发布-自然语言处理实验室北京大学计算语言学研究所、郑州大学自然语言处理实验室与鹏城实验室人工智能研究中心智慧医疗课题组联合发布中文医学知识图谱CMeKG2.0版http://cmekg.pcl.ac.cn/ 欢迎大家试用并提出宝贵意见！ CMeKG（Chinese Medical Knowledge Graph）是利用自然语言处理与文本挖掘技术，基于大规模医学文本数据，以人机结合的方式研发的中文医学知识图谱。CMeKG的构建参考了ICD、ATC、SNOMED、MeSH等权威的国际医学标准以及规模庞大、多源...http://www5.zzu.edu.cn/nlp/info/1018/1785.htm

项目成果展示：

CMeKG中文医学知识图谱http://cmekg.pcl.ac.cn/

项目源代码获取：

https://github.com/king-yyf/CMeKG_toolshttps://github.com/king-yyf/CMeKG_tools

medical_re.py

config 类：

class config:
    batch_size = 32
    max_seq_len = 256
    num_p = 23
    learning_rate = 1e-5
    EPOCH = 2

    PATH_SCHEMA = "/Users/yangyf/workplace/model/medical_re/predicate.json"
    PATH_TRAIN = '/Users/yangyf/workplace/model/medical_re/train_data.json'
    PATH_BERT = "/Users/yangyf/workplace/model/medical_re/"
    PATH_MODEL = "/Users/yangyf/workplace/model/medical_re/model_re.pkl"
    PATH_SAVE = '/content/model_re.pkl'
    tokenizer = BertTokenizer.from_pretrained("/Users/yangyf/workplace/model/medical_re/" + 'vocab.txt')

    id2predicate = {}
    predicate2id = {}

此类中定义了部分基础变量，例如一个批的数据大小是32，最大字长是256等等。同时还定义了部分文件路径，在使用是需要根据自己电脑中文件的位置进行修改。

tokenizer = BerTokenizer.from_pretrained 是 transformers 当中的一个方法，该方法的工作流程是它会先判断 **from_pretrained **函数的参数，如果是 PRETRAINED_MODEL_ARCHIVE_MAP 已有的，就会去cache里找；如果不是，就会判断它是不是一个路径，会在这个路径下找需要的文件，一个config文件和一个bin文件。

 PRETRAINED_MODEL_ARCHIVE_MAP = {
        'bert-base-uncased': "https://s3.amazonaws.com/models.huggingface.co/bert/bert-base-uncased.tar.gz",
        'bert-large-uncased': "https://s3.amazonaws.com/models.huggingface.co/bert/bert-large-uncased.tar.gz",
        'bert-base-cased': "https://s3.amazonaws.com/models.huggingface.co/bert/bert-base-cased.tar.gz",
        'bert-base-multilingual': "https://s3.amazonaws.com/models.huggingface.co/bert/bert-base-multilingual.tar.gz",
        'bert-base-chinese': "https://s3.amazonaws.com/models.huggingface.co/bert/bert-base-chinese.tar.gz",
    }
"""
    Params:
    pretrained_model_name: either:
    - a str with the name of a pre-trained model to load selected in the list of:
    . `bert-base-uncased`
    . `bert-large-uncased`
    . `bert-base-cased`
    . `bert-base-multilingual`
    . `bert-base-chinese`
    - a path or url to a pretrained model archive containing:
    . `bert_config.json` a configuration file for the model
    . `pytorch_model.bin` a PyTorch dump of a BertForPreTraining instance
    *inputs, **kwargs: additional input for the specific Bert class
    (ex: num_labels for BertForSequenceClassification)
"""

IterableDataset 类：

class IterableDataset(torch.utils.data.IterableDataset):
    def __init__(self, data, random):
        super(IterableDataset).__init__()
        self.data = data
        self.random = random
        self.tokenizer = config.tokenizer

    def __len__(self):
        return len(self.data)

    def search(self, sequence, pattern):
        n = len(pattern)
        for i in range(len(sequence)):
            if sequence[i:i + n] == pattern:
                return i
        return -1

    def process_data(self):
        idxs = list(range(len(self.data)))
        if self.random:
            np.random.shuffle(idxs)
        batch_size = config.batch_size
        max_seq_len = config.max_seq_len
        num_p = config.num_p
        batch_token_ids = np.zeros((batch_size, max_seq_len), dtype=np.int)         
        batch_mask_ids = np.zeros((batch_size, max_seq_len), dtype=np.int)          
        batch_segment_ids = np.zeros((batch_size, max_seq_len), dtype=np.int)
        batch_subject_ids = np.zeros((batch_size, 2), dtype=np.int)
        batch_subject_labels = np.zeros((batch_size, max_seq_len, 2), dtype=np.int)
        batch_object_labels = np.zeros((batch_size, max_seq_len, num_p, 2), dtype=np.int)
        batch_i = 0
        for i in idxs:
            text = self.data[i]['text']
            batch_token_ids[batch_i, :] = self.tokenizer.encode(text, max_length=max_seq_len, pad_to_max_length=True,add_special_tokens=True)
            batch_mask_ids[batch_i, :len(text) + 2] = 1
            spo_list = self.data[i]['spo_list']
            idx = np.random.randint(0, len(spo_list), size=1)[0]
            s_rand = self.tokenizer.encode(spo_list[idx][0])[1:-1]
            s_rand_idx = self.search(list(batch_token_ids[batch_i, :]), s_rand)
            batch_subject_ids[batch_i, :] = [s_rand_idx, s_rand_idx + len(s_rand) - 1]
            for i in range(len(spo_list)):
                spo = spo_list[i]
                s = self.tokenizer.encode(spo[0])[1:-1]
                p = config.prediction2id[spo[1]]
                o = self.tokenizer.encode(spo[2])[1:-1]
                s_idx = self.search(list(batch_token_ids[batch_i]), s)
                o_idx = self.search(list(batch_token_ids[batch_i]), o)
                if s_idx != -1 and o_idx != -1:#如果主体和客体都存在
                    batch_subject_labels[batch_i, s_idx, 0] = 1
                    batch_subject_labels[batch_i, s_idx + len(s) - 1, 1] = 1
                    if s_idx == s_rand_idx:
                        batch_object_labels[batch_i, o_idx, p, 0] = 1
                        batch_object_labels[batch_i, o_idx + len(o) - 1, p, 1] = 1
            batch_i += 1
            if batch_i == batch_size or i == idxs[-1]:
                yield batch_token_ids, batch_mask_ids, batch_segment_ids, batch_subject_labels, batch_subject_ids, batch_object_labels
                batch_token_ids[:, :] = 0
                batch_mask_ids[:, :] = 0
                batch_subject_ids[:, :] = 0
                batch_subject_labels[:, :, :] = 0
                batch_object_labels[:, :, :, :] = 0
                batch_i = 0

search()：

用来进行模式匹配，该函数中有两个输入参数，一个返回参数，在一给定的 **sequence **序列中找到所希望的 **pattern **，如果能找到则返回 pattern 开始的位置，找不到则返回-1。

process_data():

对数据进行处理和分类，便于以后使用。如果类中 **random **参数为真，则会使用 shuffle 方法将idxs中的数据进行打乱打散。之后创建所需的标志和掩码数组。以前两条创建语句为例：生成一个batch_size,max_seq_len大小的零数组，数据的存储方式是numpy的整型形式，数组名称是token；生成和token同样大小的零数组，用于掩码，mask。

在 **for **循环中首先从字典中取出所需处理的文本，然后使用 tokenizer.encode() 中进行解析，返回的是切分之后的编码情况。此处对 tokenizer.encode() 进行详细阐述。

一个简单的例子来展示 tokenizer.encode() 的作用，此部分来自天才小呵呵的博文：

Transformers包tokenizer.encode()方法源码阅读笔记_天才小呵呵的博客-CSDN博客_tokenizer.encode1 引言Hugging Face公司出的transformer包，能够超级方便的引入预训练模型，BERT、ALBERT、GPT2… tokenizer = BertTokenizer.from_pretrained('bert-base-uncased') model = BertForTokenClassification.from_pretrained('bert-bas...https://blog.csdn.net/qq_33293040/article/details/105439750

sentence = "Hello, my sun is cuting."
input_ids_method1 = torch.tensor(
    tokenizer.encode(sentence, add_special_tokens=True))  # Batch size 1
    # tensor([ 101, 7592, 1010, 2026, 2365, 2003, 3013, 2075, 1012,  102])

代码运行后的输出包括101和102表示开头和结尾的输出[cls]、[sep]。附上 tokenizer.encode() 的源码解析各参数的作用。

    def encode(
        self,
        text: str,  # 需要转化的句子
        text_pair: Optional[str] = None,   
        add_special_tokens: bool = True, 
        max_length: Optional[int] = None,  
        stride: int = 0,
        truncation_strategy: str = "longest_first",
        pad_to_max_length: bool = False,
        return_tensors: Optional[str] = None,
        **kwargs
    ):
        """
        Converts a string in a sequence of ids (integer), using the tokenizer and vocabulary.

        Same as doing ``self.convert_tokens_to_ids(self.tokenize(text))``.

        Args:
            text (:obj:`str` or :obj:`List[str]`):
                The first sequence to be encoded. This can be a string, a list of strings (tokenized string using
                the `tokenize` method) or a list of integers (tokenized string ids using the `convert_tokens_to_ids`
                method)
            text_pair (:obj:`str` or :obj:`List[str]`, `optional`, defaults to :obj:`None`):
                Optional second sequence to be encoded. This can be a string, a list of strings (tokenized
                string using the `tokenize` method) or a list of integers (tokenized string ids using the
                `convert_tokens_to_ids` method)
            add_special_tokens (:obj:`bool`, `optional`, defaults to :obj:`True`):
                If set to ``True``, the sequences will be encoded with the special tokens relative
                to their model.
            max_length (:obj:`int`, `optional`, defaults to :obj:`None`):
                If set to a number, will limit the total sequence returned so that it has a maximum length.
                If there are overflowing tokens, those will be added to the returned dictionary
            stride (:obj:`int`, `optional`, defaults to ``0``):
                If set to a number along with max_length, the overflowing tokens returned will contain some tokens
                from the main sequence returned. The value of this argument defines the number of additional tokens.
            truncation_strategy (:obj:`str`, `optional`, defaults to `longest_first`):
                String selected in the following options:

                - 'longest_first' (default) Iteratively reduce the inputs sequence until the input is under max_length
                  starting from the longest one at each token (when there is a pair of input sequences)
                - 'only_first': Only truncate the first sequence
                - 'only_second': Only truncate the second sequence
                - 'do_not_truncate': Does not truncate (raise an error if the input sequence is longer than max_length)
            pad_to_max_length (:obj:`bool`, `optional`, defaults to :obj:`False`):
                If set to True, the returned sequences will be padded according to the model's padding side and
                padding index, up to their max length. If no max length is specified, the padding is done up to the
                model's max length. The tokenizer padding sides are handled by the class attribute `padding_side`
                which can be set to the following strings:

                - 'left': pads on the left of the sequences
                - 'right': pads on the right of the sequences
                Defaults to False: no padding.
            return_tensors (:obj:`str`, `optional`, defaults to :obj:`None`):
                Can be set to 'tf' or 'pt' to return respectively TensorFlow :obj:`tf.constant`
                or PyTorch :obj:`torch.Tensor` instead of a list of python integers.
            **kwargs: passed to the `self.tokenize()` method
        """
        encoded_inputs = self.encode_plus(
            text,
            text_pair=text_pair,
            max_length=max_length,
            add_special_tokens=add_special_tokens,
            stride=stride,
            truncation_strategy=truncation_strategy,
            pad_to_max_length=pad_to_max_length,
            return_tensors=return_tensors,
            **kwargs,
        )

        return encoded_inputs["input_ids"]

**add_special_tokens: bool = True **将句子转化成对应模型的输入形式，默认开启

** max_length **设置最大长度，如果不设置的话原模型设置的最大长度是512，此时，如果句子长度超过512会报错：

** pad_to_max_length: bool = False** 是否按照最长长度补齐，默认关闭，此处可以通过**tokenizer.padding_side = 'left' **设置补齐的位置在左边插入。

**truncation_strategy: str = "longest_first" **截断机制，有四种方式来读取句子内容：- ‘longest_first’(默认)：一直迭代，读到不能再读，读满为止- ‘only_first’: 只读入第一个序列- ‘only_second’: 只读入第二个序列- ‘do_not_truncate’: 不做截取，长了就报错

**return_tensors: Optional[str] = None **返回的数据类型，默认是None,可以选择tensorflow版本 ('tf') 和pytorch版本('pt')。

然后源码在mask列表里面选择第 batch_i 行，将里面文本长度加2的值置1表明文本长度，预留出开头和结尾的输出。spo是关系抽取中的三元组，s是主体，o是客体，p是关系。找到一个随机主体，然后在文本中进行匹配，找到该主体位置并记录。然后对spo_list进行遍历，如果主体、客体都在文本中存在，则记录下主体位置和关系，同时如果该主体与随机主体相同，则记录下客体的位置。

小循环的次数为spo_list的长度，大循环的结束条件有两个一是批数达到上限二是数据达到上限。

这里附上对于 yield 方法的解析：

一个带有 yield 的函数就是一个 generator，它和普通函数不同，生成一个 generator 看起来像函数调用，但不会执行任何函数代码，直到对其调用 next()（在 for 循环中会自动调用 next()）才开始执行。虽然执行流程仍按函数的流程执行，但每执行到一个 yield 语句就会中断，并返回一个迭代值，下次执行时从 yield 的下一个语句继续执行。看起来就好像一个函数在正常执行的过程中被 yield 中断了数次，每次中断都会通过 yield 返回当前的迭代值。yield 的好处是显而易见的，把一个函数改写为一个 generator 就获得了迭代能力，比起用类的实例保存状态来计算下一个 next() 的值，不仅代码简洁，而且执行流程异常清晰。

get_stream():

def get_stream(self):
    return cycle(self.process_data())

在此类中接下来两个函数，用来对上文中的process_data()函数进行循环无尽的生成，对于cycle函数的解释如下。

itertools. cycle ( iterable )
Make an iterator returning elements from the iterable and saving a copy of each. When the iterable is exhausted, return elements from the saved copy. Repeats indefinitely. Roughly equivalent to:
def cycle(iterable):
# cycle('ABCD') --> A B C D A B C D A B C D ...
saved = []
for element in iterable:
yield element
saved.append(element)
while saved:
for element in saved:
yield element

Model4s类

class Model4s(nn.Module):
    def __init__(self, hidden_size=768):
        super(Model4s, self).__init__()
        self.bert = BertModel.from_pretrained(config.PATH_BERT)
        for param in self.bert.parameters():
            param.requires_grad = True
        self.dropout = nn.Dropout(p=0.2)
        self.linear = nn.Linear(in_features=hidden_size, out_features=2, bias=True)
        self.sigmoid = nn.Sigmoid()

    def forward(self, input_ids, input_mask, segment_ids, hidden_size=768):
        hidden_states = self.bert(input_ids,
                                  attention_mask=input_mask,
                                  token_type_ids=segment_ids)[0]  # (batch_size, sequence_length, hidden_size)
        output = self.sigmoid(self.linear(self.dropout(hidden_states))).pow(2)

        return output, hidden_states

先从名字理解该类，此类用于主体数据模型的训练。

init():

在这里定义并初始化了一些以后会使用的函数。

super().init()，就是继承父类的init方法，关于super().init()的详细解释可以查看此篇博文，个人认为介绍的十分详细。python中super().init()_java/python知识分享-CSDN博客_super().init()super（）.init__（） 1、子类构造函数调用super().init()1.1、第一个直接继承父类，可以调用name1.2、第二个继承自父类，覆盖初始化化def init，增加属性age，不能调用name属性1.3、第三个继承自父类，覆盖初始化化def init，并继承初始化属性name，可以调用2、继承顺序3、python2和3的区别1、子类构造函数调用super().i...https://blog.csdn.net/a__int/article/details/104600972

然后加载一个初始化的训练模型，路径在前面已经定义过了。之后由于是最终的训练模型所以将require_grad设置为真。

根据PyTorch的自动求导机制，如果一个tensor设置require_grad为True的情况下，才会对这个tensor以及由这个tensor计算出来的其他tensor求导，并将导数值存在tensor的grad属性中，便于优化器来更新参数。

所以一般情况下，只有对训练过程的tensor设置require_grad为True，便于求导和更新参数。而验证过程只是检查当前模型的泛化能力，只需要求出loss，所以不需要根据val set的loss对参数进行更新，因此require_grad为False。

作者：zhFang1999
链接：https://www.zhihu.com/question/436410778/answer/1645941506
来源：知乎

对于nn.Dropout的解释可以参见这篇博文：深度学习中Dropout原理解析_Microstrong-CSDN博客_dropout1. Dropout简介1.1 Dropout出现的原因在机器学习的模型中，如果模型的参数太多，而训练样本又太少，训练出来的模型很容易产生过拟合的现象。在训练神经网络的时候经常会遇到过拟合的问题，过拟合具体表现在：模型在训练数据上损失函数较小，预测准确率较高；但是在测试数据上损失函数比较大，预测准确率较低。过拟合是很多机器学习的通病。如果模型过拟合，那么得到的模型几乎不能用。为了解决过拟合问题，一...https://blog.csdn.net/program_developer/article/details/80737724

对于nn.Linear的解释可以参照这两篇博文：其中第二篇博客最清晰的解释了in_features和out_features，第一篇则较为全面的介绍了整个nn.linear的使用方式并给出公式说明。

Pytorch.nn.Linear 解析（数学角度）_Medlen-CSDN博客_nn.linearpytorch.nn.Linear 是一个类，下面是它的一些初始化参数in_features : 输入样本的张量大小out_features : 输出样本的张量大小bias : 偏置它主要是对输入数据做一个线性变换。y=xAT+by=xA^T+by=xAT+b这里A是权重矩阵，b是偏置。他们都是根据 in_features 生成测试代码：m = torch.nn.Linea...https://blog.csdn.net/weixin_38481963/article/details/105258389PyTorch的nn.Linear（）详解_风雪夜归人o的博客-CSDN博客_nn.linear PyTorch的nn.Linear（）是用于设置网络中的全连接层的，需要注意的是全连接层的输入与输出都是二维张量，一般形状为[batch_size, size]，不同于卷积层要求输入输出是四维张量。其用法与形参说明如下： in_features指的是输入的二维张量的大小，即输入的[batch_size, size]中的size。 out_features指的是输出的二维张量的大小，即...https://blog.csdn.net/qq_42079689/article/details/102873766?depth_1-utm_source=distribute.pc_relevant.none-task&utm_source=distribute.pc_relevant.none-task

该函数中的最后一句引用了nn中的Sigmoid函数，而该函数形式为：

$\text{Sigmoid}(x) = \sigma(x) = \frac{1}{1 + \exp(-x)}$

forward()：

这个函数说实话还并没有看明白，在这个python文件中并没有使用，应结合在实际py文件中的使用，在此推测一下应该是使用初始化好的模型对主体进行相关计算，计算出隐藏层数，然后使用sigmoid计算出一个数字（经过平方），以供日后使用

标签： python 知识图谱 github

本文转载自: https://blog.csdn.net/chen_nnn/article/details/122795768
版权归原作者 chen_nnn 所有，如有侵权，请联系我们删除。