python-NLP：2词性标注与命名实体识别

词性标注

  词性是词汇基本的语法属性，通常也称为词类。词性标注是在给定句子中判定每个词的语法范畴，确定其词性并加以标注的过程。例如，表示人、地点、事物以及其他抽象概念的名称即为名词，表示动作或状态变化的词为动词，描述或修饰名词属性、状态的词为形容词。如给定一个句子:“这儿是个非常漂亮的公园”，对其的标注结果应如下:“这儿/代词 是/动词 个/量词 非常/副词 漂亮/形容词 的/结构助词 公园 /名词”。
  在中文中，一个词的词性很多时候都不是固定的，一般表现为同音同形的词在不同场景下，其表示的语法属性截然不同，这就为词性标注带来很大的困难;但是另外一方面，从整体上看，大多数词语，尤其是实词，一般只有一到两个词性，且其中一个词性的使用频次远远大于另一个，即使每次都将高频词性作为词性选择进行标注，也能实现80%以上的准确率。如此，若我们对常用词的词性能够进行很好地识别，那么就能够覆盖绝大多数场景，满足基本的准确度要求。
  词性标注最简单的方法是从语料库中统计每个词所对应的高频词性，将其作为默认词性，但这样显然还有提升空间。目前较为主流的方法是如同分词一样，将句子的词性标注作为一个序列标注问题来解决，那么分词中常用的手段，如隐含马尔可夫模型、条件随机场模型等皆可在词性标注任务中使用。本节将继续介绍如何使用Jieba分词来完成词性标注任务。

import jieba.posseg as psg
sent="中文分词是文本处理不可或缺的一步！"
seg_list=psg.cut(sent)
print(" ".join(['{0}/{1}'.format(w,t) for w,t in seg_list]))

结果

中文/nz 分词/n 是/v 文本处理/n 不可或缺/l 的/uj 一步/m ！/x

format用法

print(“我来自{}，我今年{}岁了”.format(“东北”,15))
我来自东北，我今年15岁了
print(“我叫{1}，来自{2}，今年{0}岁”.format(15,“小王”,“天津”))
我叫小王，来自天津，今年15岁
print(“我来自{0}，我今年{1}岁了,我来自{0}”.format(“东北”,‘15’))## 我来自东北，我今年15岁了,我来自东北
print(“我来自{place}，我今年{age}岁了”.format(place=“东北”,age=15))
我来自东北，我今年15岁了

命名实体识别

  与自动分词、词性标注一样，命名实体识别也是自然语言处理的一个基础任务，是信息抽取、信息检索、机器翻译、问答系统等多种自然语言处理技术必不可少的组成部分。其目的是识别语料中人名、地名、组织机构名等命名实体。由于这些命名实体数量不断增加，通常不可能在词典中穷尽列出，且其构成方法具有各自的规律性，因此，通常把对这些词的识别在词汇形态处理(如汉语切分)任务中独立处理，称为命名实体识别(Named Entities Recognition，NER)。NER研究的命名实体一般分为3大类(实体类、时间类和数字类)和7小类(人名、地名、组织机构名、时间、日期、货币和百分比)。由于数量、时间、日期、货币等实体识别通常可以采用模式匹配的方式获得较好的识别效果，相比之下人名、地名、机构名较复杂，因此近年来的研究主要以这几种实体为主。
  命名实体识别当前并不是一个大热的研究课题，因为学术界部分认为这是一个已经解决了的问题，但是也有学者认为这个问题还没有得到很好地解决，原因主要有:命名实体识别只是在有限的文本类型(主要是新闻语料)和实体类别(主要是人名、地名)中取得了效果;与其他信息检索领域相比，实体命名评测语料较小，容易产生过拟合;命名实体识别更侧重高召回率，但在信息检索领域，高准确率更重要;通用的识别多种类型的命名实体的系统性很差。
  同时，中文的命名实体识别与英文的相比，挑战更大，目前未解决的难题更多。命名实体识别效果的评判主要看实体的边界是否划分正确以及实体的类型是否标注正确。在英文中，命名实体一般具有较为明显的形式标志(如英文实体中的每个词的首字母要大写)，因此其实体边界识别相对容易很多，主要重点是在对实体类型的确定。而在汉语中，相较于实体类别标注子任务，实体边界的识别更加困难。

时间命名实体（规则方法）

import re
from datetime import datetime,timedelta #timedelta时间差
from dateutil.parser import parse
import jieba.posseg as psg

UTIL_CN_NUM = {
    '零': 0, '一': 1, '二': 2, '两': 2, '三': 3, '四': 4, '五': 5, '六': 6, '七': 7, '八': 8, '九': 9,
    '0': 0, '1': 1, '2': 2, '3': 3, '4': 4, '5': 5, '6': 6, '7': 7, '8': 8, '9': 9,
}
UTIL_CN_UTIL = {'十': 10, '百': 100, '千': 1000, '万': 10000}



def check_time_valid(word):
    """
    对拼接字符串近一步处理，以进行有效性判断
    :param word: time_res中的每一项(每一项切割出来的时间)
    :return: 清洗后的句子
    """
    # match()匹配成功返回对象，否则返回None，
    # match是全匹配，即从头到尾，而$是匹配最后，从match源码来看，如果str是存在非数字的情况会直接返回None
    # 这里的意思就是清洗掉长度小于等于6的纯数字(小于等于6的意思是指非准确日期，比如2020)
    m = re.match('\d+$', word)
    if m:
        # 当正则表达式匹配成功时，判断句子的长度是否小于等于6，如果小于等于6，则返回None
        if len(word) <= 6:
            return None
    # 将"号"和"日"替换为"日",个人理解，这里是号和日后面莫名其妙跟了一串数字的情况
    word_1 = re.sub('[号|日]\d+$', '日', word)
    if word_1 != word:
        # 如果清洗出来的句子与原句子不同，则递归调用
        return check_time_valid(word_1)
    else:
        # 如果清洗出来的句子与原句子相同，则返回任意一个句子
        return word_1



def cn2dig(src):
    """
    除了年份之外的其余时间的解析
    :param src: 除了年份的其余时间(从列表的头到倒数第二个字，即假设有"月"这个字，则清洗掉"月")
    :return rsl: 返回相应的除了年份的其余时间的阿拉伯数字
    """
    if src == "":
        # 如果src为空，那么直接返回None，又进行一次清洗
        return None
    m = re.match("\d+", src)
    if m:
        # 如果m是数字则直接返回该数字
        return int(m.group(0))
    rsl = 0
    unit = 1
    for item in src[:: -1]:
        # 从后向前遍历src
        if item in UTIL_CN_UTIL.keys():
            # 如果item在UTIL_CN_UTIL中，则unit为这个字转换过来的阿拉伯数字
            # 即假设src为"三十"，那么第一个item为"十"，对应的unit为10
            unit = UTIL_CN_UTIL[item]
        elif item in UTIL_CN_NUM.keys():
            # 如果item不在UTIL_CN_UTIL而在UTIL_CN_NUM中，则转换为相应的阿拉伯数字并且与unit相乘
            # 就假设刚刚那个"三十"，第二个字为"三"，对应的num为3，rsl就为30
            num = UTIL_CN_NUM[item]
            rsl += num * unit
        else:
            # 如果都不在，那么就不是数字，就直接返回None
            return None

    if rsl < unit:
        # 如果出现"十五"这种情况，那么是先执行上面的elif，即rsl = 5，再执行if，即unit = 10，
        # 这时候rsl < unit，那么执行相加操作
        rsl += unit

    return rsl



def year2dig(year):
    """
    解析年份这个维度，主要是将中文或者阿拉伯数字统一转换为阿拉伯数字的年份
    :param year: 传入的年份(从列表的头到倒数第二个字，即假设有"年"这个字，则清洗掉"年")
    :return: 所表达的年份的阿拉伯数字或者None
    """
    res = ''
    for item in year:
        # 循环遍历这个年份的每一个字符
        if item in UTIL_CN_NUM.keys():
            # 如果这个字在UTIL_CN_NUM中，则转换为相应的阿拉伯数字
            res = res + str(UTIL_CN_NUM[item])
        else:
            # 否则直接相加
            # 这里已经是经历了多方面清洗后的结果了，基本到这里不会在item中出现异常的字符
            res = res + item

    m = re.match("\d+", res)
    if m:
        # 当m开头为数字时，执行下面操作，否则返回None
        if len(m.group(0)) == 2:
            # 这里是假设输入的话为"我要住到21年..."之类的，那么year就只有2个字符，即这里m == 21，
            # 那么就通过当前年份除100的整数部分再乘100最后加上这个数字获得最终年份
            # 即int(2020 / 100) * 100 + int("21")
            return int(datetime.today().year / 100) * 100 + int(m.group(0))
        else:
            # 否则直接返回该年份
            return int(m.group(0))
    else:
        return None



def parse_datetime(msg):
    """

    :param msg: ['2024年07月23日下午三点']示例  '我要从26号下午4点住到11月2号'示例
    :return:
    """
    if msg is None or len(msg) == 0:
        return None

    try:
        dt = parse(msg, fuzzy=True) #时间格式转换 为2024-07-23 00:00:00 2024-11-02 00:00:00
        return dt.strftime('%Y-%m-%d %H:%M:%S')
    except Exception as e:
        m = re.match(
            r"([0-9零一二两三四五六七八九十]+年)?([0-9一二两三四五六七八九十]+月)?([0-9一二两三四五六七八九十]+[号日])?([上中下午晚早]+)?([0-9零一二两三四五六七八九十百]+[点:\.时])?([0-9零一二三四五六七八九十百]+分?)?([0-9零一二三四五六七八九十百]+秒)?",
            msg)
        if m.group(0) is not None:
            res = {
                "year": m.group(1),
                "month": m.group(2),
                "day": m.group(3),
                "hour": m.group(5) if m.group(5) is not None else '00',
                "minute": m.group(6) if m.group(6) is not None else '00',
                "second": m.group(7) if m.group(7) is not None else '00',
            }
            params = {}

            for name in res:
                if res[name] is not None and len(res[name]) != 0:
                    tmp = None
                    if name == 'year':
                        tmp = year2dig(res[name][:-1])
                    else:
                        tmp = cn2dig(res[name][:-1])
                    if tmp is not None:
                        params[name] = int(tmp)
            target_date = datetime.today().replace(**params)
            is_pm = m.group(4)
            if is_pm is not None:
                if is_pm == u'下午' or is_pm == u'晚上' or is_pm =='中午':
                    hour = target_date.time().hour
                    if hour < 12:
                        target_date = target_date.replace(hour=hour + 12)
            return target_date.strftime('%Y-%m-%d %H:%M:%S')
        else:
            return None


def time_extract(text):
    """
    思路:
        通过jieba分词将带有时间信息的词进行切分，记录连续时间信息的词。
        使用了词性标注，提取"m(数字)"和"t(时间)"词性的词。

    规则约束:
        对句子进行解析，提取其中所有能表示日期时间的词，并进行上下文拼接

    :param text: 每一个请求文本
    :return: 解析出来后最终的句子
    """

    time_res = []
    word = ''
    key_date = {'今天': 0, '明天': 1, '后天': 2}
    for k, v in psg.cut(text):
        # k: 词语, v: 词性
        if k in key_date:
            # 当k存在于key_date中时
            if word != '':
                # 如果word不为空时, 列表中添加相应的词语
                time_res.append(word)
            # 获取系统当前时间，并且获取句子中时间的跨度(0, 1, 2)，通过当前时间 + 时间跨度获得几天后的时间
            word = (datetime.today() + timedelta(days=key_date.get(k, 0))) \
                .strftime('%Y {0} %m {1} %d {2} ').format('年', '月', '日')
        elif word != '':
            # 如果k不存在于key_date时，word不为空
            if v in ['m', 't']:
                # 当词性为数字或时间时，添加至word中
                word = word + k
            else:
                # 当词性不为数字或时间时，将word放入time_res，同时清空word
                time_res.append(word)
                word = ''
        elif v in ['m', 't']:
            # 当k不存在于key_date中，且word为空时，如果词性是数字或时间时，word为该词语
            word = k
    if word != '':
        # word中可能存放的值:
        #   1. 通过词性标注后获得的时间跨度后的时间
        #   2. 非key_date中的时间或数字
        # 即只有k不存在于key_date，word不为空，词性不为数字或时间时，word才为空，进入不了这个if语句
        time_res.append(word)

    # 如果返回的结果是None，则直接清洗，否则放入集合中
    result = list(filter(lambda x: x is not None, [check_time_valid(w) for w in time_res]))
    final_res = [parse_datetime(w) for w in result]

    return [x for x in final_res if x is not None]


text1 = '我要住到明天下午三点'
print(text1, time_extract(text1), sep=':')

text2 = '预定28号的房间'
print(text2, time_extract(text2), sep=':')

text3 = '我要从26号下午4点住到11月2号'
print(text3, time_extract(text3), sep=':')

text4 = '我要预订今天到30的房间'
print(text4, time_extract(text4), sep=':')

CRF 命名实体识别方法

CRF原理

  CRF 本质是一个无向图，其中绿色点表示输入，红色点表示输出。点与点之间的边可以分成两类，一类是 x 与 y 之间的连线，表示其相关性；另一类是相邻时刻的 y之间的相关性。也就是说，在预测某时刻 y时，同时要考虑相邻的标签解决。当 CRF 模型收敛时，就会学到类似 P-B 和 T-I 作为相邻标签的概率非常低。
  对于 CRF，我们给出准确的数学语言描述：设 X 与 Y 是随机变量，P(Y|X) 是给定 X 时 Y 的条件概率分布，若随机变量 Y 构成的是一个马尔科夫随机场，则称条件概率分布 P(Y|X) 是条件随机场。

  本文模型训练数据集 。数据分为10个标签类别，分别为: 地址（address），书名（book），公司（company），游戏（game），政府（goverment），电影（movie），姓名（name），组织机构（organization），职位（position），景点（scene）。
CLUE Fine-Grain NER中文数据集

本文NER任务使用BIO三位标注法，即:

B-begin：代表实体开头
I-inside：代表实体内部
O-outside：代表不属于任何实体
其后面接实体类型，如 ‘B-name’,‘I-company’。

`

#coding:utf-8
import json
import sklearn_crfsuite #CRF
from sklearn import metrics
from itertools import chain


# 将数据处理成CRF库输入格式
def data_process(path):
    # 读取每一条json数据放入列表中
    # 由于该json文件含多个数据，不能直接json.loads读取，需使用for循环逐条读取
    json_data = []
    with open(path, 'r', encoding='utf-8') as fp:
        for line in fp:
            json_data.append(json.loads(line))

    # json_data中每一条数据的格式为
    '''
    {'text': '浙商银行企业信贷部叶老桂博士则从另一个角度对五道门槛进行了解读。叶老桂认为，对目前国内商业银行而言，',
     'label': {'name': {'叶老桂': [[9, 11]]}, 'company': {'浙商银行': [[0, 3]]}}}
     '''

    # 将json文件处理成如下格式
    '''
    [['浙', '商', '银', '行', '企', '业', '信', '贷', '部', '叶', '老', '桂', '博', '士', '则', '从', '另', '一', 
    '个', '角', '度', '对', '五', '道', '门', '槛', '进', '行', '了', '解', '读', '。', '叶', '老', '桂', '认', 
    '为', '，', '对', '目', '前', '国', '内', '商', '业', '银', '行', '而', '言', '，'], 
    ['B-company', 'I-company', 'I-company', 'I-company', 'O', 'O', 'O', 'O', 'O', 'B-name', 'I-name', 
    'I-name', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 
    'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O']]
    '''
    data = []
    # 遍历json_data中每组数据
    for i in range(len(json_data)):
        # 将标签全初始化为'O'
        label = ['O'] * len(json_data[i]['text'])
        # 遍历'label'中几组实体，如样例中'name'和'company'
        for n in json_data[i]['label']:
            # 遍历实体中几组文本，如样例中'name'下的'叶老桂'（有多组文本的情况，样例中只有一组）
            for key in json_data[i]['label'][n]:
                # 遍历文本中几组下标，如样例中[[9, 11]]（有时某个文本在该段中出现两次，则会有两组下标）
                for n_list in range(len(json_data[i]['label'][n][key])):
                    # 记录实体开始下标和结尾下标
                    start = json_data[i]['label'][n][key][n_list][0]
                    end = json_data[i]['label'][n][key][n_list][1]
                    # 将开始下标标签设为'B-' + n，如'B-' + 'name'即'B-name'
                    # 其余下标标签设为'I-' + n
                    label[start] = 'B-' + n
                    label[start + 1: end + 1] = ['I-' + n] * (end - start)

        # 对字符串进行字符级分割
        # 英文文本如'bag'分割成'b'，'a'，'g'三位字符，数字文本如'125'分割成'1'，'2'，'5'三位字符
        texts = []
        for t in json_data[i]['text']:
            texts.append(t)

        # 将文本和标签编成一个列表添加到返回数据中
        data.append([texts, label])
    return data


# 判断字符是否是英文
def is_english(c):
    if ord(c.lower()) >= 97 and ord(c.lower()) <= 122:
        return True
    else:
        return False

# 将文本转换为特征字典
# sklearn-crfsuite输入数据支持多种格式，这里选择字典格式
# 单个CRF与BiLSTM+CRF不同，BiLSTM会自动生成输入序列中每个字符的发射概率，而单个CRF的发射概率则是通过学习将特征映射成发射概率
# sklearn-crfsuite的数据输入格式采用字典格式，类似于做特征工程，CRF将这些特征映射成发射概率
'''
    序列中的每一个字符处理成如下格式：
    {'bias': 1.0,
     'word': '商',
     'word.isdigit()': False,
     'word.is_english()': False,
     '-1:word': '浙',
     '-1:word.isdigit()': False,
     '-1:word.is_english()': False,
     '+1:word': '银',
     '+1:word.isdigit()': False,
     '+1:word.is_english()': False}
'''
def word2features(sent, i):
    # 本代码采用大小为3的滑动窗口构造特征，特征有当前字符、字符是否为数字或英文等，当然可以增大窗口或增加其他特征
    # 特征长度可以不同
    word = sent[i][0]
    features = {
        'bias': 1.0,
        'word': word,
        'word.isdigit()': word.isdigit(),
        'word.is_english()': is_english(word),
    }

    if i > 0:
        word = sent[i - 1][0]
        features.update({
            '-1:word': word,
            '-1:word.isdigit()': word.isdigit(),
            '-1:word.is_english()': is_english(word),
        })
    else:
        # 若该字符为序列开头，则增加特征 BOS(begin of sentence)
        features['BOS'] = True
    # 该字的后一个字
    if i < len(sent) - 1:
        word = sent[i + 1][0]
        features.update({
            '+1:word': word,
            '+1:word.isdigit()': word.isdigit(),
            '+1:word.is_english()': is_english(word),
        })
    else:
        # 若该字符为序列结尾，则增加特征 EOS(end of sentence)
        features['EOS'] = True
    return features


def sent2features(sent):
    return [word2features(sent, i) for i in range(len(sent))]


def sent2labels(sent):
    return [label for label in sent]


train = data_process('./data/cluener_public/train.json')
valid = data_process('./data/cluener_public/dev.json')
print('训练集长度:', len(train))
print('验证集长度:', len(valid))
X_train = [sent2features(s[0]) for s in train]
y_train = [sent2labels(s[1]) for s in train]
X_dev = [sent2features(s[0]) for s in valid]
y_dev = [sent2labels(s[1]) for s in valid]
print(X_train[0][1])

# algorithm：lbfgs法求解该最优化问题，c1：L1正则系数，c2：L2正则系数，max_iterations：迭代次数，verbose：是否显示训练信息
crf_model = sklearn_crfsuite.CRF(algorithm='lbfgs', c1=0.1, c2=0.1, max_iterations=50,
                                 all_possible_transitions=True, verbose=True)
# 若sklearn版本大于等于0.24会报错：AttributeError: 'CRF' object has no attribute 'keep_tempfiles'
# 可降低版本 pip install -U 'scikit-learn<0.24'
# 或使用异常处理，不会影响训练效果
try:
    crf_model.fit(X_train, y_train)
except:
    pass

labels = list(crf_model.classes_)
# 由于大部分标签都是'O'，故不去关注'O'标签的预测
labels.remove("O")
y_pred = crf_model.predict(X_dev)

y_dev = list(chain.from_iterable(y_dev))
y_pred = list(chain.from_iterable(y_pred))

# 计算F1分数，average可选'micro'，'macro'，'weighted'，处理多类别F1分数的不同计算方法
# 此metrics为sklearn_crfsuite.metrics，但必须引入from sklearn_crfsuite import metrics
# 也可使用sklearn.metrics.f1_score(y_dev, y_pred, average='weighted', labels=labels)),但要求y_dev和y_pred是一维列表
print('weighted F1 score:', metrics.f1_score(y_dev, y_pred,
                      average='weighted', labels=labels))

# 排好标签顺序输入，否则默认按标签出现顺序进行排列
sorted_labels = sorted(labels, key=lambda name: (name[1:], name[0]))
# 打印详细分数报告，包括precision(精确率)，recall(召回率)，f1-score(f1分数)，support(个数)，digits=3代表保留3位小数
print(metrics.classification_report(
    y_dev, y_pred, labels=sorted_labels, digits=3
))

# 查看转移概率和发射概率
# print('CRF转移概率：', crf_model.transition_features_)
# print('CRF发射概率：', crf_model.state_features_)