超越简单分类构建面向真实世界的多层文本分类系统摘要文本分类是自然语言处理(NLP)的基础任务但大多数教程停留在电影评论情感分析这类简单案例。本文将深入探讨如何构建一个面向真实世界复杂场景的文本分类系统处理多标签、类别不平衡、概念漂移等实际问题。我们将结合传统机器学习与深度学习方法构建一个混合架构的文本分类器。一、真实世界文本分类的挑战1.1 从理想场景到现实困境典型的文本分类教程假设类别数量固定且平衡文本长度相对均匀类别定义清晰且互斥训练数据与测试数据同分布然而现实世界的文本分类面临类别不平衡某些类别样本极少长尾分布多标签分类单文本属于多个类别概念漂移类别定义随时间变化标注不一致不同标注者对同一文本归类不同领域适应模型需要适应新的文本领域1.2 系统设计目标我们的系统设计需要满足高可扩展性能够处理新增类别而无需完全重新训练增量学习能力适应新数据和新类别多粒度分类支持从粗粒度到细粒度的分类不确定性量化提供分类置信度评估可解释性帮助理解分类决策依据二、混合架构设计2.1 系统整体架构我们提出一个三阶段混合架构import numpy as np import pandas as pd from typing import List, Dict, Tuple, Optional import hashlib import json from dataclasses import dataclass from collections import defaultdict dataclass class TextDocument: 文本文档数据结构 id: str text: str raw_labels: List[str] # 原始标签 processed_text: Optional[str] None embeddings: Optional[np.ndarray] None metadata: Optional[Dict] None class HierarchicalTextClassifier: 层次化文本分类器 def __init__(self, feature_extractor_type: str hybrid, classifier_type: str cascaded, uncertainty_threshold: float 0.3): 初始化分类器 Args: feature_extractor_type: 特征提取器类型 [tfidf, embedding, hybrid] classifier_type: 分类器类型 [flat, hierarchical, cascaded] uncertainty_threshold: 不确定性阈值高于此值触发人工审核 self.feature_extractor_type feature_extractor_type self.classifier_type classifier_type self.uncertainty_threshold uncertainty_threshold self.label_hierarchy {} # 标签层次结构 self.classifiers {} # 存储不同层次的分类器 self.label_embeddings {} # 标签的语义嵌入 # 设置随机种子以确保可重复性 self.random_seed 1773010800058 % (2**32 - 1) # 使用提供的种子 np.random.seed(self.random_seed) def build_label_hierarchy(self, labels: List[List[str]], method: str taxonomy): 构建标签层次结构 Args: labels: 多标签数据每个文档的标签列表 method: 构建方法 [taxonomy, clustering, embedding] # 计算标签共现矩阵 label_cooccurrence self._compute_cooccurrence_matrix(labels) if method taxonomy: # 基于规则或外部知识库构建分类体系 self._build_taxonomy_hierarchy(labels) elif method clustering: # 基于标签语义相似度的聚类 self._build_clustering_hierarchy(label_cooccurrence) else: # 基于预训练嵌入的层次构建 self._build_embedding_hierarchy(labels) def _compute_cooccurrence_matrix(self, labels: List[List[str]]) - np.ndarray: 计算标签共现矩阵 all_labels sorted(list(set([l for doc_labels in labels for l in doc_labels]))) label_to_idx {label: i for i, label in enumerate(all_labels)} n_labels len(all_labels) cooccurrence np.zeros((n_labels, n_labels)) for doc_labels in labels: indices [label_to_idx[l] for l in doc_labels] for i in indices: for j in indices: cooccurrence[i, j] 1 # 归一化 np.fill_diagonal(cooccurrence, 0) # 对角线置零 return cooccurrence2.2 分层特征提取模块传统文本分类通常使用单一特征表示我们提出分层特征提取class HybridFeatureExtractor: 混合特征提取器 def __init__(self, use_semantic: bool True, use_lexical: bool True, use_syntactic: bool False, use_domain_features: bool True): self.use_semantic use_semantic self.use_lexical use_lexical self.use_syntactic use_syntactic self.use_domain_features use_domain_features # 初始化不同特征提取器 if self.use_semantic: self._init_semantic_extractor() if self.use_lexical: self._init_lexical_extractor() def _init_semantic_extractor(self): 初始化语义特征提取器 try: # 使用Sentence Transformer获取高质量文本嵌入 from sentence_transformers import SentenceTransformer self.semantic_model SentenceTransformer(all-MiniLM-L6-v2) self.semantic_dim 384 except ImportError: # 备选方案使用BERT的池化输出 from transformers import AutoTokenizer, AutoModel self.tokenizer AutoTokenizer.from_pretrained(bert-base-uncased) self.bert_model AutoModel.from_pretrained(bert-base-uncased) self.semantic_dim 768 def _init_lexical_extractor(self): 初始化词汇特征提取器 from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.feature_extraction.text import CountVectorizer # 多粒度TF-IDF self.tfidf_char TfidfVectorizer( analyzerchar_wb, ngram_range(2, 5), max_features5000 ) self.tfidf_word TfidfVectorizer( ngram_range(1, 3), max_features10000, stop_wordsenglish ) def extract_features(self, texts: List[str]) - Dict[str, np.ndarray]: 提取混合特征 features {} # 1. 语义特征 if self.use_semantic: if hasattr(self, semantic_model): semantic_features self.semantic_model.encode(texts) else: semantic_features self._extract_bert_features(texts) features[semantic] semantic_features # 2. 词汇特征 if self.use_lexical: # 字符级特征 char_features self.tfidf_char.fit_transform(texts).toarray() # 词级特征 word_features self.tfidf_word.fit_transform(texts).toarray() features[lexical] np.hstack([char_features, word_features]) # 3. 文本统计特征 stats_features np.array([self._extract_text_stats(t) for t in texts]) features[statistical] stats_features # 4. 领域特定特征示例技术文档特征 if self.use_domain_features: domain_features np.array([self._extract_domain_features(t) for t in texts]) features[domain] domain_features return features def _extract_text_stats(self, text: str) - np.ndarray: 提取文本统计特征 words text.split() sentences text.split(.) features [ len(text), # 文本长度 len(words), # 词数 len(sentences), # 句子数 np.mean([len(w) for w in words]) if words else 0, # 平均词长 len(set(words)) / len(words) if words else 0, # 词汇多样性 sum(1 for c in text if c.isdigit()) / max(len(text), 1), # 数字比例 sum(1 for c in text if c.isupper()) / max(len(text), 1), # 大写比例 ] return np.array(features) def _extract_domain_features(self, text: str) - np.ndarray: 提取领域特定特征以技术文档为例 # 代码片段检测 code_patterns [def , class , import , function , return ] code_score sum(1 for pattern in code_patterns if pattern in text) # 技术术语检测 tech_terms [algorithm, database, API, framework, library] tech_score sum(1 for term in tech_terms if term.lower() in text.lower()) # URL和版本号检测 import re url_count len(re.findall(rhttps?://\S, text)) version_count len(re.findall(r\bv\d\.\d(\.\d)?\b, text)) return np.array([code_score, tech_score, url_count, version_count])三、处理类别不平衡与增量学习3.1 自适应采样策略class AdaptiveSampler: 自适应采样器处理类别不平衡 def __init__(self, sampling_strategy: str adaptive): self.sampling_strategy sampling_strategy self.class_distribution None def fit_resample(self, X, y): 根据类别分布进行重采样 from collections import Counter import numpy as np self.class_distribution Counter(y) total_samples len(y) if self.sampling_strategy adaptive: return self._adaptive_sampling(X, y) elif self.sampling_strategy focal: return self._focal_sampling(X, y) else: return X, y def _adaptive_sampling(self, X, y): 自适应采样为少数类过采样为多数类欠采样 from sklearn.utils import resample import pandas as pd # 将数据转换为DataFrame以便处理 if not isinstance(X, pd.DataFrame): X_df pd.DataFrame(X) else: X_df X y_series pd.Series(y) df pd.concat([X_df, y_series], axis1) df.columns list(X_df.columns) [target] # 计算每个类别的目标样本数 n_classes len(self.class_distribution) target_samples len(y) // n_classes resampled_data [] for class_label in self.class_distribution.keys(): class_data df[df[target] class_label] n_samples len(class_data) if n_samples target_samples: # 过采样少数类 sampled resample(class_data, replaceTrue, n_samplestarget_samples, random_state1773010800058 % 1000) elif n_samples target_samples * 1.5: # 欠采样多数类 sampled resample(class_data, replaceFalse, n_samplestarget_samples, random_state1773010800058 % 1000) else: sampled class_data resampled_data.append(sampled) # 合并重采样后的数据 balanced_df pd.concat(resampled_data, ignore_indexTrue) X_balanced balanced_df.drop(target, axis1).values y_balanced balanced_df[target].values return X_balanced, y_balanced def _focal_sampling(self, X, y): Focal采样根据分类难度调整样本权重 # 这里简化为根据类别频率调整权重 class_weights {} total sum(self.class_distribution.values()) for cls, count in self.class_distribution.items(): # 少数类获得更高权重 class_weights[cls] total / (len(self.class_distribution) * count) # 计算样本权重 sample_weights np.array([class_weights[label] for label in y]) return X, y, sample_weights3.2 增量学习实现class IncrementalClassifier: 支持增量学习的分类器 def __init__(self, base_classifierNone, drift_detectorNone, memory_size1000): self.base_classifier base_classifier or self._create_default_classifier() self.drift_detector drift_detector or self._create_drift_detector() self.memory_size memory_size self.memory_buffer [] # 存储新样本用于增量更新 self.concept_drift_detected False def _create_default_classifier(self): 创建默认的增量学习分类器 from sklearn.linear_model import SGDClassifier return SGDClassifier( losslog_loss, penaltyelasticnet, alpha0.0001, l1_ratio0.15, max_iter1000, tol1e-3, random_state1773010800058 % 1000, learning_rateadaptive, eta00.01 ) def _create_drift_detector(self): 创建概念漂移检测器 # 使用ADWIN算法检测漂移 try: from skmultiflow.drift_detection import ADWIN return ADWIN() except ImportError: # 简化版本基于准确率下降的检测 class SimpleDriftDetector: def __init__(self, threshold0.1, window_size100): self.threshold threshold self.window_size window_size self.accuracy_window [] def add_element(self, prediction, true_label): correct 1 if prediction true_label else 0 self.accuracy_window.append(correct) if len(self.accuracy_window) self.window_size: self.accuracy_window.pop(0) if len(self.accuracy_window) self.window_size: current_acc np.mean(self.accuracy_window[-50:]) previous_acc np.mean(self.accuracy_window[:50]) return (previous_acc - current_acc) self.threshold return False return SimpleDriftDetector() def partial_fit(self, X, y, classesNone): 增量学习接口 # 检测概念漂移 if len(self.memory_buffer) 0: predictions self.base_classifier.predict(X[:10]) for pred, true in zip(predictions, y[:10]): if self.drift_detector.add_element(pred, true): self.concept_drift_detected True print(概念漂移检测到可能需要重新训练) # 添加到内存缓冲区 self.memory_buffer.extend(list(zip(X, y))) if len(self.memory_buffer) self.memory_size: self.memory_buffer self.memory_buffer[-self.memory_size:] # 定期更新模型 if len(self.memory_buffer) 100: X_batch, y_batch zip(*self.memory_buffer[-100:]) self.base_class