LSTM文本分类数据预处理实战：从清洗到序列化的完整指南

1. 先搞清楚LSTM项目的数据预处理到底要解决什么问题

如果你正在处理文本分类任务，特别是像公司行业分类这种需要理解上下文语义的场景，LSTM确实是个不错的选择。但很多人一上来就急着调模型参数，结果发现效果还不如简单的TF-IDF。问题往往出在数据预处理阶段没做到位。

LSTM对输入数据的格式非常敏感。它需要的是数值化的序列数据，而不是原始文本。预处理的核心目标就是把乱七八糟的文本数据转换成模型能理解的数字序列，同时还要保留文本的语义信息。

具体到公司行业分类这个场景，你手头可能有公司名称、业务描述、经营范围这些文本数据。预处理要做的就是把它们转换成词向量序列，并且构建一个完整的词表映射系统。这步做不好，后面模型训练再折腾也是白费功夫。

我一般会先把预处理拆成几个关键步骤：文本清洗、分词、构建词表、序列化、数据集划分。每个步骤都有容易踩坑的地方，比如中文分词时的专有名词处理，或者词表构建时的高频词筛选。

2. 文本清洗和分词的实际操作细节

拿到原始数据后，不要急着分词。先做一轮文本清洗，把明显无效的内容过滤掉。

2.1 文本清洗的关键步骤

清洗阶段主要处理以下几类问题：

• 特殊字符：保留有意义的标点，去掉乱码和无关符号
• 空格和换行：统一处理成标准空格
• 大小写：根据任务需求决定是否统一小写
• 数字处理：保留原样、替换成统一标记或删除

对于公司数据，我一般会保留数字和部分特殊符号，因为像"ABC科技股份有限公司"、"腾讯控股00700"这样的文本中，数字可能包含重要信息。

import re def clean_text(text): # 保留中文、英文、数字、常见标点 text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9，。！？；：、（）《》]', ' ', text) # 合并多个空格 text = re.sub(r'\s+', ' ', text) return text.strip()

2.2 中文分词的实用方案

分词质量直接影响后续效果。对于行业分类这种专业领域，建议使用领域词典增强的分词工具。

我比较推荐jieba分词，配合自定义词典效果更好：

import jieba # 加载行业专有名词词典 jieba.load_userdict('industry_terms.txt') def segment_text(text): # 精确模式，适合文本分析 words = jieba.lcut(text) # 过滤单字和停用词 words = [word for word in words if len(word) > 1 and word not in stopwords] return words

停用词表可以使用哈工大停用词表作为基础，再根据业务需求补充。比如在公司描述中，"有限公司"、"股份有限公司"这类词虽然高频但信息量低，可以考虑加入停用词表。

3. 构建词表的关键决策点

词表构建是预处理中最需要经验判断的环节。词表大小直接影响模型性能和训练速度。

3.1 确定词表大小的经验法则

我一般按这个思路确定词表大小：

• 小数据集（<10万条）：词表大小控制在5000-10000
• 中等数据集（10万-100万条）：10000-50000
• 大数据集（>100万条）：50000-200000

实际操作时，先统计所有词语的频率分布：

from collections import Counter def build_vocab(texts, max_vocab_size=20000): word_freq = Counter() for text in texts: words = segment_text(text) word_freq.update(words) # 按频率排序，取前max_vocab_size-2个词（留位置给特殊标记） vocab = ['<PAD>', '<UNK>'] # 填充标记和未知词标记 vocab.extend([word for word, freq in word_freq.most_common(max_vocab_size-2)]) word2idx = {word: idx for idx, word in enumerate(vocab)} return vocab, word2idx

3.2 处理低频词和未知词

词表中不可能包含所有词，总会遇到未知词。常见的处理策略：

• 统一映射到<UNK>标记
• 使用字符级或子词单元（BPE）
• 对于公司数据，可以考虑保留所有出现次数超过2次的词

在实践中，我发现对于行业分类任务，保留出现3次以上的词通常能在覆盖率和噪声控制之间取得较好平衡。

4. 文本序列化的实际考量

把分词后的文本转换成数字序列时，需要决定序列长度和填充策略。

4.1 序列长度选择

序列长度太短会丢失信息，太长会增加计算负担。我一般这样确定：

1. 先统计所有文本的分词后长度分布
2. 选择覆盖95%文本的长度作为标准长度
3. 对于极长文本，考虑截断或分段处理

import numpy as np def analyze_text_lengths(texts): lengths = [len(segment_text(text)) for text in texts] print(f"平均长度: {np.mean(lengths):.1f}") print(f"长度标准差: {np.std(lengths):.1f}") print(f"95%分位数: {np.percentile(lengths, 95):.1f}") return lengths

对于公司描述数据，序列长度通常在50-200之间。我一般设置128或256，既能覆盖大多数情况，又不会太浪费计算资源。

4.2 填充和截断策略

def text_to_sequence(text, word2idx, max_length=128): words = segment_text(text) # 截断或填充到固定长度 if len(words) > max_length: words = words[:max_length] # 取前max_length个词 else: words = words + ['<PAD>'] * (max_length - len(words)) # 转换为索引序列 sequence = [word2idx.get(word, word2idx['<UNK>']) for word in words] return sequence

5. 训练集、验证集、测试集的正确划分方法

数据集划分影响模型评估的可靠性。常见的错误是随机划分导致数据泄露。

5.1 按时间划分的重要性

对于公司数据，如果包含时间信息，一定要按时间划分。用旧数据训练，新数据测试，更符合实际应用场景。

def split_dataset_by_time(data, time_column, split_ratio=[0.7, 0.15, 0.15]): data_sorted = data.sort_values(time_column) n_total = len(data_sorted) train_end = int(n_total * split_ratio[0]) val_end = train_end + int(n_total * split_ratio[1]) train_data = data_sorted.iloc[:train_end] val_data = data_sorted.iloc[train_end:val_end] test_data = data_sorted.iloc[val_end:] return train_data, val_data, test_data

5.2 分层抽样保证分布一致

如果按时间划分不可行，至少要进行分层抽样，确保每个数据集中的类别分布相似：

from sklearn.model_selection import train_test_split def stratified_split(data, label_column, test_size=0.2): # 分层划分，保持类别比例 train_data, test_data = train_test_split( data, test_size=test_size, stratify=data[label_column], random_state=42 ) # 再从训练集中划分验证集 train_data, val_data = train_test_split( train_data, test_size=0.2, stratify=train_data[label_column], random_state=42 ) return train_data, val_data, test_data

6. 完整预处理流程的代码实现

下面是一个完整的预处理流程示例：

import pandas as pd import numpy as np from collections import Counter from sklearn.model_selection import train_test_split class TextPreprocessor: def __init__(self, max_vocab_size=20000, max_seq_length=128): self.max_vocab_size = max_vocab_size self.max_seq_length = max_seq_length self.word2idx = None self.vocab = None def fit(self, texts): """基于训练文本构建词表""" all_words = [] for text in texts: words = self.segment_text(text) all_words.extend(words) word_freq = Counter(all_words) self.vocab = ['<PAD>', '<UNK>'] + \ [word for word, freq in word_freq.most_common(self.max_vocab_size-2)] self.word2idx = {word: idx for idx, word in enumerate(self.vocab)} def transform(self, texts): """将文本转换为序列""" sequences = [] for text in texts: seq = self.text_to_sequence(text) sequences.append(seq) return np.array(sequences) def segment_text(self, text): # 简化的分词函数，实际使用时可替换为jieba等 words = text.split() # 这里用空格分词，实际应用需要更复杂的分词 return [word for word in words if len(word) > 1] def text_to_sequence(self, text): words = self.segment_text(text) # 截断或填充 if len(words) > self.max_seq_length: words = words[:self.max_seq_length] else: words = words + ['<PAD>'] * (self.max_seq_length - len(words)) return [self.word2idx.get(word, self.word2idx['<UNK>']) for word in words] # 使用示例 def main(): # 假设data是包含文本和标签的DataFrame data = pd.read_csv('company_data.csv') # 划分数据集 train_data, temp_data = train_test_split(data, test_size=0.3, random_state=42) val_data, test_data = train_test_split(temp_data, test_size=0.5, random_state=42) # 初始化预处理器 preprocessor = TextPreprocessor(max_vocab_size=20000, max_seq_length=128) # 只用训练数据拟合词表 preprocessor.fit(train_data['text'].tolist()) # 转换所有数据集 X_train = preprocessor.transform(train_data['text'].tolist()) X_val = preprocessor.transform(val_data['text'].tolist()) X_test = preprocessor.transform(test_data['text'].tolist()) y_train = train_data['label'].values y_val = val_data['label'].values y_test = test_data['label'].values print(f"词表大小: {len(preprocessor.vocab)}") print(f"训练集形状: {X_train.shape}") print(f"验证集形状: {X_val.shape}") print(f"测试集形状: {X_test.shape}") if __name__ == "__main__": main()

7. 预处理阶段的常见问题排查

当预处理结果不理想时，按这个顺序排查：

7.1 数据质量检查

先看原始数据：

• 是否有大量空值或无效文本
• 文本长度分布是否合理
• 类别分布是否均衡

def check_data_quality(data, text_column, label_column): # 检查空值 print(f"空值数量: {data[text_column].isnull().sum()}") # 检查文本长度分布 text_lengths = data[text_column].str.len() print(f"文本长度统计:\n{text_lengths.describe()}") # 检查类别分布 label_dist = data[label_column].value_counts() print(f"类别分布:\n{label_dist}")

7.2 词表质量检查

检查构建的词表是否合理：

• 高频词是否都是有意义的词汇
• 是否包含了重要的专业术语
• 停用词过滤是否适当

def check_vocab_quality(vocab, top_n=50): print("前50个高频词:") for i, word in enumerate(vocab[:min(top_n, len(vocab))]): print(f"{i}: {word}")

7.3 序列化结果验证

随机抽样检查几个样本的序列化结果：

• 序列长度是否正确
• 填充位置是否合理
• 未知词处理是否正常

def validate_sequences(sequences, vocab, original_texts, num_samples=5): idx2word = {idx: word for word, idx in vocab.items()} for i in range(min(num_samples, len(sequences))): seq = sequences[i] original_text = original_texts[i] print(f"样本 {i}:") print(f"原始文本: {original_text}") print(f"序列: {seq[:20]}...") # 只显示前20个token print(f"转换回文本: {' '.join([idx2word.get(idx, '<UNK>') for idx in seq[:20]])}...") print("-" * 50)

8. 预处理结果的保存和复用

预处理结果需要保存以便后续使用：

import pickle def save_preprocessor(preprocessor, filepath): with open(filepath, 'wb') as f: pickle.dump({ 'word2idx': preprocessor.word2idx, 'vocab': preprocessor.vocab, 'max_vocab_size': preprocessor.max_vocab_size, 'max_seq_length': preprocessor.max_seq_length }, f) def load_preprocessor(filepath): with open(filepath, 'rb') as f: saved_data = pickle.load(f) preprocessor = TextPreprocessor( max_vocab_size=saved_data['max_vocab_size'], max_seq_length=saved_data['max_seq_length'] ) preprocessor.word2idx = saved_data['word2idx'] preprocessor.vocab = saved_data['vocab'] return preprocessor

预处理阶段花的时间绝对值得。我见过太多项目因为预处理不到位，导致后续模型调参怎么调都效果不好。把数据清洗干净、词表构建合理、数据集划分正确，这些基础工作做扎实了，LSTM模型才能发挥出真正的威力。

实际落地时，最该盯住的不是模型结构多复杂，而是输入数据的质量。很多效果问题追根溯源都是预处理阶段埋下的坑。