Python案例如何用CRF命名实体识别

wen python案例 2026-07-16 207

本文目录导读：

环境准备
基础CRF NER实现
特征提取函数
准备训练数据
训练CRF模型
模型评估
使用模型进行预测
完整的中文NER示例
模型保存与加载
模型优化建议

我将为您展示如何使用CRF（条件随机场）进行命名实体识别（NER）的完整案例。

环境准备

首先安装必要的库：

pip install sklearn-crfsuite scikit-learn nltk

基础CRF NER实现

import sklearn_crfsuite
from sklearn_crfsuite import metrics
from sklearn.model_selection import train_test_split
import nltk
from nltk.corpus import conll2002
import numpy as np
# 下载数据
nltk.download('conll2002')
nltk.download('averaged_perceptron_tagger')
# 加载数据
train_sents = list(conll2002.iob_sents('esp.train'))
test_sents = list(conll2002.iob_sents('esp.testb'))
print(f"训练样本数: {len(train_sents)}")
print(f"测试样本数: {len(test_sents)}")
print(f"第一个句子样本:\n{train_sents[0][:10]}")

特征提取函数

def word2features(sent, i):
    """提取单个词的上下文特征"""
    word = sent[i][0]
    postag = sent[i][1]
    features = {
        'bias': 1.0,  # 偏置特征
        'word.lower()': word.lower(),
        'word[-3:]': word[-3:],
        'word[-2:]': word[-2:],
        'word.isupper()': word.isupper(),
        'word.istitle()': word.istitle(),
        'word.isdigit()': word.isdigit(),
        'postag': postag,
        'postag[:2]': postag[:2],  # 词性标签的前两个字符
    }
    # 前一个词的特征
    if i > 0:
        word1 = sent[i-1][0]
        postag1 = sent[i-1][1]
        features.update({
            '-1:word.lower()': word1.lower(),
            '-1:postag': postag1,
            '-1:postag[:2]': postag1[:2],
        })
    else:
        features['BOS'] = True  # 句子开始标记
    # 后一个词的特征
    if i < len(sent)-1:
        word1 = sent[i+1][0]
        postag1 = sent[i+1][1]
        features.update({
            '+1:word.lower()': word1.lower(),
            '+1:postag': postag1,
            '+1:postag[:2]': postag1[:2],
        })
    else:
        features['EOS'] = True  # 句子结束标记
    return features
def sent2features(sent):
    """将句子转换为特征序列"""
    return [word2features(sent, i) for i in range(len(sent))]
def sent2labels(sent):
    """提取句子的标签序列"""
    return [label for token, pos, label in sent]
def sent2tokens(sent):
    """提取句子的词序列"""
    return [token for token, pos, label in sent]

准备训练数据

# 准备特征和标签
X_train = [sent2features(s) for s in train_sents]
y_train = [sent2labels(s) for s in train_sents]
X_test = [sent2features(s) for s in test_sents]
y_test = [sent2labels(s) for s in test_sents]
print("特征示例:")
print(f"第一个句子的第一个词特征: {X_train[0][0]}")
print(f"第一个句子的标签: {y_train[0][:10]}")

训练CRF模型

# 创建CRF模型
crf = sklearn_crfsuite.CRF(
    algorithm='lbfgs',          # 优化算法
    c1=0.1,                     # L1正则化系数
    c2=0.1,                     # L2正则化系数
    max_iterations=100,         # 最大迭代次数
    all_possible_transitions=True  # 考虑所有可能的转移
)
# 训练模型
print("开始训练CRF模型...")
crf.fit(X_train, y_train)
print("训练完成！")

模型评估

# 预测
y_pred = crf.predict(X_test)
# 计算准确率
labels = list(crf.classes_)
labels.remove('O')  # 移除O标签用于计算准确率
# 评估指标
print("\n=== 模型评估结果 ===")
print(f"F1-score (宏观平均): {metrics.flat_f1_score(y_test, y_pred, average='weighted', labels=labels):.4f}")
print(f"准确率: {metrics.flat_accuracy_score(y_test, y_pred):.4f}")
# 分类报告
sorted_labels = sorted(labels, key=lambda name: (name[1:], name[0]))
print("\n分类报告:")
print(metrics.flat_classification_report(y_test, y_pred, labels=sorted_labels, digits=3))

使用模型进行预测

def predict_entities(text, crf_model):
    """对输入文本进行命名实体识别"""
    # 分词和词性标注
    tokens = nltk.word_tokenize(text)
    pos_tags = nltk.pos_tag(tokens, lang='spa')  # 这里用西班牙语模型
    # 转换为特征格式
    sent = [(token, pos, 'O') for token, pos in pos_tags]
    features = sent2features(sent)
    # 预测
    predicted_tags = crf_model.predict([features])[0]
    # 提取实体
    entities = []
    current_entity = []
    current_label = None
    for token, tag in zip(tokens, predicted_tags):
        if tag.startswith('B-'):
            if current_entity:
                entities.append((current_label, ' '.join(current_entity)))
            current_entity = [token]
            current_label = tag[2:]
        elif tag.startswith('I-'):
            if current_label == tag[2:]:
                current_entity.append(token)
            else:
                if current_entity:
                    entities.append((current_label, ' '.join(current_entity)))
                current_entity = [token]
                current_label = tag[2:]
        else:
            if current_entity:
                entities.append((current_label, ' '.join(current_entity)))
                current_entity = []
                current_label = None
    # 添加最后一个实体
    if current_entity:
        entities.append((current_label, ' '.join(current_entity)))
    return entities
# 测试预测
test_text = "Juan visitó Madrid y Barcelona la semana pasada."
entities = predict_entities(test_text, crf)
print("\n输入文本:", test_text)
print("识别的实体:", entities)

完整的中文NER示例

# 使用自建数据集进行中文NER
import jieba
import jieba.posseg as pseg
# 创建示例训练数据
train_data = [
    ("我在北京工作", "O O B-LOC O"),
    ("李明是个医生", "B-PER O O O O"),
    ("我去了北京大学", "O O O B-ORG I-ORG"),
    ("王小明在上海读书", "B-PER I-PER O B-LOC O O O"),
]
def prepare_chinese_data(data):
    """准备中文CRF训练数据"""
    sentences = []
    labels = []
    for text, label_str in data:
        # 分词
        words = list(jieba.cut(text))
        label_seq = label_str.split()
        # 词性标注
        pos_tags = [word.flag for word in pseg.cut(text)]
        # 构建句子格式 (word, pos, label)
        sent = [(word, pos, label) for word, pos, label in zip(words, pos_tags, label_seq)]
        sentences.append(sent)
    return sentences
# 准备中文数据
chinese_sents = prepare_chinese_data(train_data)
# 特征提取（中文适配）
def chinese_word2features(sent, i):
    """中文特征提取"""
    word = sent[i][0]
    postag = sent[i][1]
    features = {
        'bias': 1.0,
        'word': word,
        'word[:1]': word[:1] if len(word) > 0 else '',
        'word[-1:]': word[-1:] if len(word) > 0 else '',
        'postag': postag,
        'word.isdigit()': word.isdigit(),
        'word.isalpha()': word.isalpha(),
    }
    # 上下文特征
    if i > 0:
        word1 = sent[i-1][0]
        features.update({
            '-1:word': word1,
            '-1:postag': sent[i-1][1],
        })
    if i < len(sent) - 1:
        word1 = sent[i+1][0]
        features.update({
            '+1:word': word1,
            '+1:postag': sent[i+1][1],
        })
    return features
# 训练中文CRF模型
def train_chinese_crf(sentences):
    """训练中文NER模型"""
    X = []
    y = []
    for sent in sentences:
        X.append([chinese_word2features(sent, i) for i in range(len(sent))])
        y.append([label for word, pos, label in sent])
    # 训练模型
    crf_chinese = sklearn_crfsuite.CRF(
        algorithm='lbfgs',
        c1=0.1,
        c2=0.1,
        max_iterations=50,
    )
    crf_chinese.fit(X, y)
    return crf_chinese
# 训练中文模型
print("\n训练中文NER模型...")
chinese_crf = train_chinese_crf(chinese_sents)
print("训练完成！")
# 中文预测函数
def predict_chinese_ner(text, crf_model):
    """中文命名实体识别预测"""
    # 分词和词性标注
    words = []
    pos_tags = []
    for word, flag in pseg.cut(text):
        words.append(word)
        pos_tags.append(flag)
    # 构建特征
    sent = [(word, pos, 'O') for word, pos in zip(words, pos_tags)]
    features = [chinese_word2features(sent, i) for i in range(len(sent))]
    # 预测
    predicted = crf_model.predict([features])[0]
    # 提取实体
    entities = []
    entity = []
    label = None
    for word, tag in zip(words, predicted):
        if tag.startswith('B-'):
            if entity:
                entities.append((label, ''.join(entity)))
            entity = [word]
            label = tag[2:]
        elif tag.startswith('I-'):
            if label == tag[2:]:
                entity.append(word)
            else:
                if entity:
                    entities.append((label, ''.join(entity)))
                entity = [word]
                label = tag[2:]
        else:
            if entity:
                entities.append((label, ''.join(entity)))
                entity = []
                label = None
    if entity:
        entities.append((label, ''.join(entity)))
    return entities, list(zip(words, predicted))
# 测试中文预测
test_texts = [
    "我在北京工作",
    "李明是一名优秀的医生",
    "王小明在北京大学学习",
]
print("\n中文NER预测结果:")
for text in test_texts:
    entities, tagged = predict_chinese_ner(text, chinese_crf)
    print(f"\n输入: {text}")
    print(f"实体: {entities}")
    print(f"标注: {tagged}")

模型保存与加载

import pickle
# 保存模型
def save_model(model, filename='crf_ner_model.pkl'):
    with open(filename, 'wb') as f:
        pickle.dump(model, f)
    print(f"模型已保存到 {filename}")
# 加载模型
def load_model(filename='crf_ner_model.pkl'):
    with open(filename, 'rb') as f:
        model = pickle.load(f)
    print(f"模型已从 {filename} 加载")
    return model
# 使用示例
save_model(crf, 'crf_ner_model.pkl')
loaded_model = load_model('crf_ner_model.pkl')

模型优化建议

def optimize_crf_parameters(X_train, y_train, X_test, y_test):
    """CRF参数优化示例"""
    from sklearn.metrics import make_scorer
    from sklearn.model_selection import GridSearchCV
    # 定义参数网格
    param_grid = {
        'c1': [0.01, 0.1, 1.0],
        'c2': [0.01, 0.1, 1.0],
        'max_iterations': [50, 100],
        'algorithm': ['lbfgs', 'l2sgd']
    }
    # 创建CRF
    crf = sklearn_crfsuite.CRF()
    # 网格搜索
    gs = GridSearchCV(
        crf,
        param_grid,
        cv=3,
        verbose=True,
        n_jobs=-1
    )
    # 训练
    gs.fit(X_train, y_train)
    print(f"最佳参数: {gs.best_params_}")
    print(f"最佳分数: {gs.best_score_:.4f}")
    return gs.best_estimator_
# 注意：网格搜索可能需要较长时间，仅在数据量大时使用

这个完整案例展示了如何使用CRF进行命名实体识别,包括特征工程、模型训练、评估和预测的全流程，您可以根据实际需求调整特征提取函数和模型参数。

本文地址： https://www.bmjup.com/post/22790.html