NLP 数据预处理：分词、向量化与特征工程

1. 技术分析

1.1 NLP 数据预处理流程

数据预处理是 NLP 管道的重要环节：

NLP 预处理流程 原始文本 → 清洗 → 分词 → 向量化 → 特征工程

1.2 预处理步骤对比

1.3 文本表示方法

文本表示层次 字符级: 字符序列 词级: 词袋模型 句子级: 句向量 文档级: 文档向量

2. 核心功能实现

2.1 文本清洗

import re import string class TextCleaner: def __init__(self): self.patterns = { 'url': r'https?://\S+|www\.\S+', 'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'html': r'<.*?>', 'special_chars': r'[^a-zA-Z0-9\s]', 'extra_spaces': r'\s+' } def clean(self, text): text = text.lower() text = re.sub(self.patterns['url'], '', text) text = re.sub(self.patterns['email'], '', text) text = re.sub(self.patterns['html'], '', text) text = re.sub(self.patterns['special_chars'], '', text) text = re.sub(self.patterns['extra_spaces'], ' ', text) return text.strip() class ChineseTextCleaner: def __init__(self): self.patterns = { 'url': r'https?://\S+|www\.\S+', 'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'html': r'<.*?>', 'punctuation': r'[，。！？、；：""''（）【】《》]', 'extra_spaces': r'\s+' } def clean(self, text): text = re.sub(self.patterns['url'], '', text) text = re.sub(self.patterns['email'], '', text) text = re.sub(self.patterns['html'], '', text) text = re.sub(self.patterns['punctuation'], '', text) text = re.sub(self.patterns['extra_spaces'], '', text) return text.strip()

2.2 分词处理

import jieba class Tokenizer: def __init__(self, language='english'): self.language = language if language == 'chinese': self.tokenizer = jieba elif language == 'english': from nltk.tokenize import word_tokenize self.tokenizer = word_tokenize def tokenize(self, text): if self.language == 'chinese': return self.tokenizer.lcut(text) else: return self.tokenizer(text) def tokenize_batch(self, texts): return [self.tokenize(text) for text in texts] class StopwordFilter: def __init__(self, language='english'): if language == 'english': from nltk.corpus import stopwords self.stopwords = set(stopwords.words('english')) elif language == 'chinese': self.stopwords = self._load_chinese_stopwords() def _load_chinese_stopwords(self): stopwords = set() common_stopwords = [ '的', '是', '在', '和', '有', '我', '他', '她', '它', '这', '那', '个', '都', '就', '也', '很', '到', '说', '要', '去', '会', '着', '没有', '看', '好', '自己', '又' ] stopwords.update(common_stopwords) return stopwords def filter(self, tokens): return [token for token in tokens if token not in self.stopwords] def filter_batch(self, tokenized_texts): return [self.filter(tokens) for tokens in tokenized_texts]

2.3 向量化

import torch import torch.nn as nn from sklearn.feature_extraction.text import TfidfVectorizer class TFIDFVectorizer: def __init__(self, max_features=5000): self.vectorizer = TfidfVectorizer(max_features=max_features) def fit(self, texts): self.vectorizer.fit(texts) def transform(self, texts): return torch.tensor(self.vectorizer.transform(texts).toarray(), dtype=torch.float32) def fit_transform(self, texts): return torch.tensor(self.vectorizer.fit_transform(texts).toarray(), dtype=torch.float32) class WordEmbeddingVectorizer: def __init__(self, embedding_dim=100): self.embedding_dim = embedding_dim self.word_to_idx = {} self.embedding = None def fit(self, tokenized_texts): vocab = set() for tokens in tokenized_texts: vocab.update(tokens) self.word_to_idx = {word: i + 1 for i, word in enumerate(vocab)} self.word_to_idx['<UNK>'] = 0 self.embedding = nn.Embedding(len(self.word_to_idx), self.embedding_dim) def transform(self, tokenized_texts, max_len=50): sequences = [] for tokens in tokenized_texts: sequence = [] for token in tokens[:max_len]: sequence.append(self.word_to_idx.get(token, 0)) sequence += [0] * (max_len - len(sequence)) sequences.append(sequence) return torch.tensor(sequences, dtype=torch.long) class BERTVectorizer: def __init__(self, model_name='bert-base-uncased'): from transformers import BertModel, BertTokenizer self.model = BertModel.from_pretrained(model_name) self.tokenizer = BertTokenizer.from_pretrained(model_name) def encode(self, texts): inputs = self.tokenizer( texts, padding=True, truncation=True, max_length=512, return_tensors='pt' ) with torch.no_grad(): outputs = self.model(**inputs) return outputs.last_hidden_state[:, 0, :]

2.4 特征工程

class TextFeatureExtractor: def __init__(self): self.features = [] def add_length_feature(self, texts): lengths = [len(text) for text in texts] self.features.append(torch.tensor(lengths, dtype=torch.float32).unsqueeze(1)) def add_word_count_feature(self, tokenized_texts): word_counts = [len(tokens) for tokens in tokenized_texts] self.features.append(torch.tensor(word_counts, dtype=torch.float32).unsqueeze(1)) def add_punctuation_feature(self, texts): punctuation_ratios = [] for text in texts: punctuation_count = sum(1 for char in text if char in string.punctuation) ratio = punctuation_count / len(text) if len(text) > 0 else 0 punctuation_ratios.append(ratio) self.features.append(torch.tensor(punctuation_ratios, dtype=torch.float32).unsqueeze(1)) def get_features(self): if not self.features: return None return torch.cat(self.features, dim=1) class FeaturePipeline: def __init__(self, steps): self.steps = steps def fit_transform(self, texts): features = texts for step in self.steps: features = step.fit_transform(features) return features def transform(self, texts): features = texts for step in self.steps: features = step.transform(features) return features

3. 性能对比

3.1 向量化方法对比

3.2 分词器对比

3.3 预处理步骤影响

4. 最佳实践

4.1 预处理管道构建

def build_preprocessing_pipeline(language='english'): steps = [ TextCleaner() if language == 'english' else ChineseTextCleaner(), Tokenizer(language=language), StopwordFilter(language=language), TFIDFVectorizer() ] return FeaturePipeline(steps) class PreprocessingFactory: @staticmethod def create(config): if config['type'] == 'tfidf': return TFIDFVectorizer(**config['params']) elif config['type'] == 'word2vec': return WordEmbeddingVectorizer(**config['params']) elif config['type'] == 'bert': return BERTVectorizer(**config['params'])

4.2 预处理流程

class NLPPreprocessor: def __init__(self, tokenizer, vectorizer, cleaner=None): self.tokenizer = tokenizer self.vectorizer = vectorizer self.cleaner = cleaner def process(self, texts): if self.cleaner: texts = [self.cleaner.clean(text) for text in texts] tokenized = self.tokenizer.tokenize_batch(texts) if hasattr(self.vectorizer, 'fit_transform'): features = self.vectorizer.fit_transform(tokenized) else: features = self.vectorizer.encode(texts) return features def transform(self, texts): if self.cleaner: texts = [self.cleaner.clean(text) for text in texts] tokenized = self.tokenizer.tokenize_batch(texts) if hasattr(self.vectorizer, 'transform'): features = self.vectorizer.transform(tokenized) else: features = self.vectorizer.encode(texts) return features

5. 总结

NLP 数据预处理是模型效果的关键：

1. 文本清洗：去除噪声，提高数据质量
2. 分词：将文本切分为基本单元
3. 向量化：将文本转为数值表示
4. 特征工程：提取额外特征

对比数据如下：

• 好的预处理可提升 5-10% 模型效果
• BERT 向量比传统向量化方法效果更好
• 中文需要专用的分词器
• 预处理管道应根据任务定制