本文目录导读：
我来为你介绍如何使用Python实现一个文件内容的模糊粗糙条件随机场（Fuzzy Rough CRF），这是一个结合了模糊集合、粗糙集和条件随机场的模型。
基础实现
import numpy as np
from sklearn_crfsuite import CRF
from sklearn_crfsuite import metrics
from fuzzywuzzy import fuzz
import re
from collections import defaultdict
class FuzzyRoughCRF:
"""模糊粗糙条件随机场"""
def __init__(self, fuzzy_threshold=80, rough_epsilon=0.1):
self.fuzzy_threshold = fuzzy_threshold
self.rough_epsilon = rough_epsilon
self.crf = CRF(
algorithm='lbfgs',
c1=0.1,
c2=0.1,
max_iterations=100,
all_possible_transitions=True
)
def extract_fuzzy_features(self, text, i):
"""提取模糊特征"""
features = {}
word = text[i]
# 基础特征
features['bias'] = 1.0
features['word.lower()'] = word.lower()
features['word.isdigit()'] = word.isdigit()
features['word.isalpha()'] = word.isalpha()
# 模糊匹配特征（与常见模式匹配）
patterns = {
'数字模式': r'\d+',
'英文单词': r'[a-zA-Z]+',
'中文': r'[\u4e00-\u9fff]+',
'特殊符号': r'[!@#$%^&*(),.?":{}|<>]'
}
for name, pattern in patterns.items():
if re.match(pattern, word):
fuzzy_score = fuzz.ratio(word, re.search(pattern, word).group())
features[f'fuzzy_{name}'] = fuzzy_score / 100.0
# 上下文特征
if i > 0:
features['prev_word'] = text[i-1]
features['fuzzy_prev_similarity'] = fuzz.ratio(word, text[i-1]) / 100.0
else:
features['BOS'] = True
if i < len(text) - 1:
features['next_word'] = text[i+1]
features['fuzzy_next_similarity'] = fuzz.ratio(word, text[i+1]) / 100.0
else:
features['EOS'] = True
return features
def apply_rough_set(self, features):
"""应用粗糙集进行特征约简"""
# 计算下近似和上近似
lower_approx = {}
upper_approx = {}
for key, value in features.items():
if value >= (1 - self.rough_epsilon):
lower_approx[key] = value
elif value >= self.rough_epsilon:
upper_approx[key] = value
# 边界区域
boundary = {k: v for k, v in upper_approx.items()
if k not in lower_approx}
# 特征选择：使用下近似和上近似的平均值
if lower_approx or upper_approx:
selected_features = {}
# 合并下近似特征
for k, v in lower_approx.items():
selected_features[k] = v
# 边界特征加权
for k, v in boundary.items():
selected_features[f'rough_{k}'] = v * 0.5
return selected_features
return features
def word2features(self, text, i):
"""生成CRF特征"""
fuzzy_features = self.extract_fuzzy_features(text, i)
rough_features = self.apply_rough_set(fuzzy_features)
# 添加词性特征（简单实现）
pos_features = {
'noun': len(re.findall(r'[\u4e00-\u9fff]+[的]?', text[i])),
'verb': len(re.findall(r'[是在了有]', text[i])),
'adj': len(re.findall(r'[好大快高]', text[i])),
}
# 合并所有特征
features = {}
features.update(rough_features)
features.update(pos_features)
return features
def extract_sentences(self, filepath):
"""从文件提取句子"""
sentences = []
with open(filepath, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line:
# 分词
tokens = list(line)
sentences.append(tokens)
return sentences
def train(self, train_file, label_file):
"""训练模型"""
# 读取训练数据
sentences = self.extract_sentences(train_file)
# 读取标签
with open(label_file, 'r', encoding='utf-8') as f:
labels = [line.strip().split() for line in f if line.strip()]
# 生成特征
X_train = []
for sentence in sentences:
features = [self.word2features(sentence, i)
for i in range(len(sentence))]
X_train.append(features)
y_train = labels
# 训练CRF模型
self.crf.fit(X_train, y_train)
print("模型训练完成")
def predict(self, test_file):
"""预测"""
sentences = self.extract_sentences(test_file)
# 生成特征
X_test = []
for sentence in sentences:
features = [self.word2features(sentence, i)
for i in range(len(sentence))]
X_test.append(features)
# 预测
y_pred = self.crf.predict(X_test)
return sentences, y_pred
def evaluate(self, test_file, label_file):
"""评估模型"""
sentences, y_pred = self.predict(test_file)
with open(label_file, 'r', encoding='utf-8') as f:
y_true = [line.strip().split() for line in f if line.strip()]
# 计算评估指标
flat_y_true = [label for seq in y_true for label in seq]
flat_y_pred = [label for seq in y_pred for label in seq]
accuracy = metrics.flat_accuracy_score(y_true, y_pred)
f1_score = metrics.flat_f1_score(y_true, y_pred,
average='weighted')
return {
'accuracy': accuracy,
'f1_score': f1_score,
'classification_report': metrics.flat_classification_report(
y_true, y_pred
)
}
class FuzzyRoughProcessor:
"""模糊粗糙处理器的完整实现"""
def __init__(self):
self.model = FuzzyRoughCRF(fuzzy_threshold=80, rough_epsilon=0.1)
def process_file(self, input_file, output_file):
"""处理文件"""
# 预处理
sentences = self._preprocess(input_file)
# 应用模糊粗糙CRF
results = self._apply_fuzzy_rough_crf(sentences)
# 后处理
output = self._postprocess(results)
# 保存结果
with open(output_file, 'w', encoding='utf-8') as f:
f.write(output)
print(f"处理完成，结果已保存到 {output_file}")
def _preprocess(self, filepath):
"""文件预处理"""
sentences = []
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
# 基本清理
content = re.sub(r'\s+', ' ', content)
content = content.strip()
# 分句
parts = re.split(r'[。！？\n]', content)
for part in parts:
if part.strip():
# 分词（这里简单按字符分割）
tokens = list(part.strip())
sentences.append(tokens)
return sentences
def _apply_fuzzy_rough_crf(self, sentences):
"""应用模糊粗糙CRF"""
results = []
features_list = []
for sentence in sentences:
features = [self.model.word2features(sentence, i)
for i in range(len(sentence))]
features_list.append(features)
# 模拟预测（这里简化处理）
for sentence, features in zip(sentences, features_list):
result = []
for word, feat in zip(sentence, features):
# 根据特征进行标签预测
if feat.get('word.isdigit()', False):
label = 'NUM'
elif feat.get('fuzzy_中文', 0) > 0.7:
label = 'CHN'
elif feat.get('fuzzy_英文单词', 0) > 0.7:
label = 'ENG'
else:
label = 'OTHER'
result.append((word, label))
results.append(result)
return results
def _postprocess(self, results):
"""后处理结果"""
output = []
for sentence in results:
line_parts = []
for word, label in sentence:
line_parts.append(f"{word}/{label}")
output.append(' '.join(line_parts))
return '\n'.join(output)
# 使用示例
if __name__ == "__main__":
# 1. 基本使用
processor = FuzzyRoughProcessor()
# 创建示例文件
with open('input.txt', 'w', encoding='utf-8') as f:
f.write("这是一段测试文本。\n包含数字123和英文ABC。")
# 处理文件
processor.process_file('input.txt', 'output.txt')
# 2. 训练模型
model = FuzzyRoughCRF()
# 准备训练数据
train_sentences = [
['我', '喜欢', '编程'],
['#39;, '天气', '很好']
]
train_labels = [
['PRON', 'VERB', 'NOUN'],
['ADV', 'NOUN', 'ADJ']
]
# 保存训练数据
with open('train.txt', 'w', encoding='utf-8') as f:
for sent in train_sentences:
f.write(''.join(sent) + '\n')
with open('train_labels.txt', 'w', encoding='utf-8') as f:
for labels in train_labels:
f.write(' '.join(labels) + '\n')
# 训练模型
model.train('train.txt', 'train_labels.txt')
# 预测
sentences, predictions = model.predict('input.txt')
print("预测结果:")
for sent, pred in zip(sentences, predictions):
for word, label in zip(sent, pred):
print(f"{word}: {label}")
高级功能扩展
class AdvancedFuzzyRoughCRF:
"""高级模糊粗糙CRF实现"""
def __init__(self):
self.fuzzy_sets = self._init_fuzzy_sets()
self.rough_relations = self._init_rough_relations()
def _init_fuzzy_sets(self):
"""初始化模糊集"""
fuzzy_sets = {
'高频词': ['的', '了', '在', '是', '有'],
'数字': ['零', '一', '二', '三', '四', '五'],
'标点': ['，', '。', '！', '？', '；', '：']
}
# 添加模糊隶属度函数
membership = {}
for category, words in fuzzy_sets.items():
membership[category] = {}
for word in words:
membership[category][word] = 0.8 # 基础隶属度
return membership
def _init_rough_relations(self):
"""初始化粗糙关系"""
return {
'等价关系': lambda x, y: abs(x - y) < 0.1,
'相似关系': lambda x, y: abs(x - y) < 0.3,
'包含关系': lambda x, y: x in y or y in x
}
def fuzzy_membership(self, word, category):
"""计算模糊隶属度"""
if category in self.fuzzy_sets:
if word in self.fuzzy_sets[category]:
# 使用Levenshtein距离进行模糊匹配
for w, base_membership in self.fuzzy_sets[category].items():
if fuzz.ratio(word, w) > 90:
return base_membership * 0.9
return 0
return 0
def rough_approximation(self, data, relation_key='等价关系'):
"""计算粗糙集的上下近似"""
if relation_key not in self.rough_relations:
return set(), set()
relation = self.rough_relations[relation_key]
lower = set()
upper = set()
for x in data:
# 下近似：确定属于的
if all(relation(x, y) for y in data if y != x):
lower.add(x)
# 上近似：可能属于的
if any(relation(x, y) for y in data if y != x):
upper.add(x)
return lower, upper
def fuzzy_rough_features(self, text, position):
"""提取模糊粗糙特征"""
features = {}
word = text[position]
# 模糊特征
for category in self.fuzzy_sets:
membership = self.fuzzy_membership(word, category)
if membership > 0:
features[f'fuzzy_{category}'] = membership
# 粗糙特征（使用上下文）
context_window = text[max(0, position-2):min(len(text), position+3)]
# 计算上下文等价类
for i, w in enumerate(context_window):
if w != word:
similarity = fuzz.ratio(word, w) / 100.0
if similarity > 0.7:
features[f'rough_context_{i}'] = similarity
# 组合特征
if 'fuzzy_高频词' in features and 'rough_context_0' in features:
features['fuzzy_rough_combo'] = features['fuzzy_高频词'] * features['rough_context_0']
return features
# 使用高级功能
def advanced_example():
"""高级功能使用示例"""
crf = AdvancedFuzzyRoughCRF()
text = list("这是一个高级的模糊粗糙CRF实现")
for i in range(len(text)):
features = crf.fuzzy_rough_features(text, i)
if features:
print(f"位置 {i} ({text[i]}): {features}")
实际应用案例
def text_processing_pipeline():
"""文本处理流水线"""
import os
# 创建处理器
processor = TextProcessingPipeline()
# 处理文件
input_file = "input.txt"
output_file = "output.json"
if os.path.exists(input_file):
results = processor.process(input_file)
processor.save_results(results, output_file)
return results
class TextProcessingPipeline:
"""文本处理流水线"""
def __init__(self):
self.fuzzy_rough_crf = FuzzyRoughCRF()
self.processor = FuzzyRoughProcessor()
def process(self, filepath):
"""完整的处理流程"""
# 1. 读取文件
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
# 2. 预处理
cleaned = self._clean_text(content)
# 3. 应用模糊粗糙CRF
results = self.processor._apply_fuzzy_rough_crf(
self.processor._preprocess(filepath)
)
# 4. 提取信息
info = self._extract_information(results)
return {
'original': content,
'cleaned': cleaned,
'annotated': results,
'information': info
}
def _clean_text(self, text):
"""文本清洗"""
text = re.sub(r'\s+', ' ', text)
text = text.strip()
return text
def _extract_information(self, results):
"""提取信息"""
info = {
'num_count': 0,
'chinese_words': [],
'english_words': [],
'entities': []
}
for sentence in results:
for word, label in sentence:
if label == 'NUM':
info['num_count'] += 1
elif label == 'CHN':
info['chinese_words'].append(word)
elif label == 'ENG':
info['english_words'].append(word)
return info
def save_results(self, results, output_file):
"""保存结果"""
import json
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
这个实现提供了：
- 基础模糊粗糙CRF模型：结合模糊集和粗糙集的特征提取
- 文件处理功能：完整的文件读写和处理流程
- 训练和预测：支持模型训练和序列标注
- 高级特征工程：模糊隶属度计算和粗糙集近似
- 实际应用案例：文本处理流水线示例
你可以根据具体需求调整参数和特征提取方法,以适应不同的文件内容和处理任务。
