本文目录导读：
我将为您展示如何使用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进行命名实体识别,包括特征工程、模型训练、评估和预测的全流程，您可以根据实际需求调整特征提取函数和模型参数。
