NLP流水线实践笔记
“NLP 项目最难的” —— 一个被脏文本折腾到凌晨的人
半年前接手了一个文本分类项目，当时的状况大致这样：数据散落在三个不同的数据库表里，预处理脚本版本不统一，模型训练全靠 Jupyter Notebook，部署是直接把 pkl 文件 rsync 到服务器。
问题的起点
半年前接手了一个文本分类项目，当时的状况大致这样：数据散落在三个不同的数据库表里，预处理脚本版本不统一，模型训练全靠 Jupyter Notebook，部署是直接把 pkl 文件 rsync 到服务器。
第一次线上事故是因为分词逻辑不一致导致的：训练环境用的 jieba 精确模式，部署环境用的全模式，输入"北京大学"，训练时切分成"北京大学"，部署时切分成"北京"“大学”，特征向量直接就对不上了。
排查了三个小时，最后在凌晨两点发现是一个实习生把部署脚本里的一行注释删了。那一刻就知道，这套方式必须要改。
先说场景
项目本身不复杂：用户上传一些文本，系统判断属于哪个类别（大概 20 个类别），然后分发到不同处理队列。数据量每天大概 3-5 万条，延迟要求控制在 200ms 以内。
看起来简单，但实际问题不少：
- 文本质量差异很大，有完整句子，也有半截话、乱码、emoji 混合
- 领域专业词汇多，通用分词器表现一般
- 类别不平衡，某些类别样本很少
- 业务端对准确率和召回率都有要求，不能只优化 F1
- 线上环境资源有限，模型不能太大
这些条件决定了我们不能直接拿预训练模型微调就完事，必须把预处理、特征工程、模型选择和部署优化都当成独立问题来处理。
文本预处理：看起来简单，实际麻烦
预处理这部分花的时间比预期多，主要是因为业务场景的特殊性。
基础清洗
一开始写了个很基础的清洗函数：
import re
import jieba
from typing import List
def basic_clean(text: str) -> str:
"""基础文本清洗"""
if not isinstance(text, str):
return ""
# 去掉多余空格
text = re.sub(r'\s+', ' ', text)
# 去掉特殊符号，保留中英文、数字、基本标点
text = re.sub(r'[^一-龥a-zA-Z0-9，。！？、；：""'',.!?;:"\'-]', ' ', text)
# 统一标点符号
text = text.replace('，', '，').replace('。', '。')
return text.strip()
这个函数用了两个月，后来发现两个问题：
- 过滤掉特殊符号时把一些有用的信息删了，比如
$50里的$、v1.2里的.、C++里的+ - 统一标点符号没有考虑中文半角混用的情况
改了一下：
def improved_clean(text: str) -> str:
"""改进后的文本清洗"""
if not isinstance(text, str):
return ""
# 转半角标点为全角
text = text.replace(',', '，').replace('.', '。')
text = text.replace('!', '！').replace('?', '？')
# 保留版本号、货币符号、代码相关的符号
text = re.sub(r'[^一-龥a-zA-Z0-9$.,+#\-，。！？、；：""'',.!?;:"\'/\\]', ' ', text)
# 去掉多余空格
text = re.sub(r'\s+', ' ', text)
return text.strip()
分词优化
通用分词器在专业词汇上的表现不好，比如"分布式一致性哈希"，jieba 切成"分布式"“一致性"“哈希”，虽然也对，但语义信息会分散。
试了两种方案：
方案1：自定义词典
# 加载专业词汇词典
jieba.load_userdict('domain_words.txt')
# domain_words.txt 内容示例
# 分布式一致性哈希 3 n
# 向量数据库 3 n
# 零信任架构 3 n
这个方案简单有效，但需要人工维护词典，新词汇出现就要更新。
方案2：基于 BERT 的预分词
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained('bert-base-chinese')
def bert_tokenize(text: str) -> List[str]:
"""使用 BERT tokenizer 进行预分词"""
tokens = tokenizer.tokenize(text)
# 去掉 [CLS] 和 [SEP]
return tokens[1:-1] if tokens else []
BERT tokenizer 的优势是能更好地处理未知词，但缺点是分词结果比较细粒度，可能会影响后续特征提取。
我们最终结合了两种方案：先加载专业词典，再用 BERT tokenizer 做一次后处理，把明显应该合并的 token 合并掉。
def combined_tokenize(text: str) -> List[str]:
"""组合分词方案"""
# 先用 jieba + 自定义词典
jieba_tokens = list(jieba.cut(text))
# 再用 BERT tokenizer 细化
bert_tokens = []
for token in jieba_tokens:
sub_tokens = tokenizer.tokenize(token)
if len(sub_tokens) == 1:
bert_tokens.append(sub_tokens[0])
else:
# 保留原始 token，除非明显是拆分错误
if all(t.startswith('##') for t in sub_tokens[1:]):
bert_tokens.extend(sub_tokens)
else:
bert_tokens.append(token)
return bert_tokens
这个方案效果不错，但有个问题：速度慢。在 Intel i7-12700K 上处理 1000 条文本需要 2.3 秒，远超 jieba 的 0.15 秒。
后来做了个妥协：训练时用组合分词，推理时只用 jieba 加自定义词典。虽然损失一点精度，但延迟从 2.3 秒降到了 0.18 秒。
归一化处理
文本归一化这部分踩过几个坑：
def normalize_text(text: str) -> str:
"""文本归一化"""
# 数字归一化：把连续数字替换为 <NUM>
text = re.sub(r'\d+', '<NUM>', text)
# URL 归一化
text = re.sub(r'http[s]?://\S+', '<URL>', text)
# 邮箱归一化
text = re.sub(r'\S+@\S+', '<EMAIL>', text)
# 日期归一化
text = re.sub(r'\d{4}[-/年]\d{1,2}[-/月]\d{1,2}[日号]?', '<DATE>', text)
return text
问题出在数字归一化上：我们业务里有些数字是有含义的，比如版本号 v1.2、价格 100元、ID user_123456。这些被统一替换成 <NUM> 后，信息就丢了。
改进版：
def smart_normalize(text: str) -> str:
"""智能归一化"""
# 先保护一些特殊模式
protected_patterns = {
'version': r'\bv\d+\.\d+(\.\d+)?\b', # v1.2.3
'price': r'\d+(元|块|美元|USD)?', # 100元
'id': r'[a-z]+_\d+', # user_123456
}
protected = {}
for name, pattern in protected_patterns.items():
matches = re.finditer(pattern, text)
for i, match in enumerate(matches):
placeholder = f'<{name}_{i}>'
protected[placeholder] = match.group()
text = text.replace(match.group(), placeholder)
# 再做通用归一化
text = re.sub(r'\d{4}[-/年]\d{1,2}[-/月]\d{1,2}[日号]?', '<DATE>', text)
text = re.sub(r'http[s]?://\S+', '<URL>', text)
text = re.sub(r'\S+@\S+', '<EMAIL>', text)
# 最后还原保护的文本
for placeholder, original in protected.items():
text = text.replace(placeholder, original)
return text
特征工程：从 TF-IDF 到 Embedding
特征工程这块经历了几次迭代。
第一版：TF-IDF + 朴素贝叶斯
一开始为了快速上线，用了最基础的方案：
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
# 构建 pipeline
pipeline = Pipeline([
('vectorizer', TfidfVectorizer(
max_features=5000,
ngram_range=(1, 2),
min_df=2,
max_df=0.95
)),
('classifier', MultinomialNB())
])
# 训练
pipeline.fit(X_train, y_train)
这个方案优点：训练快、推理快、容易解释。
缺点也很明显：准确率只有 78%，对短文本效果差，无法捕捉语义信息。
第二版：TF-IDF + LightGBM
换了更强的分类器：
import lightgbm as lgb
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.preprocessing import LabelEncoder
# 文本向量化
vectorizer = TfidfVectorizer(
max_features=8000,
ngram_range=(1, 3),
min_df=3,
max_df=0.90
)
X_train_tfidf = vectorizer.fit_transform(X_train)
# 标签编码
le = LabelEncoder()
y_train_encoded = le.fit_transform(y_train)
# 训练模型
model = lgb.LGBMClassifier(
n_estimators=300,
learning_rate=0.05,
max_depth=6,
num_leaves=31,
min_child_samples=20,
subsample=0.8,
colsample_bytree=0.8,
random_state=42
)
model.fit(X_train_tfidf, y_train_encoded)
准确率提升到了 85%，但推理延迟从 15ms 增加到了 45ms，接近业务要求的上限。
第三版：Sentence Embedding + LightGBM
为了提升准确率，尝试了基于 BERT 的句向量：
from sentence_transformers import SentenceTransformer
import numpy as np
# 加载预训练模型
sentence_model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
# 生成句向量
def get_embeddings(texts: List[str]) -> np.ndarray:
"""批量生成句向量"""
return sentence_model.encode(texts, show_progress_bar=False)
# 训练集嵌入
X_train_emb = get_embeddings(X_train)
X_test_emb = get_embeddings(X_test)
# 训练模型
model = lgb.LGBMClassifier(
n_estimators=200,
learning_rate=0.1,
max_depth=8,
num_leaves=63,
random_state=42
)
model.fit(X_train_emb, y_train_encoded)
准确率提升到了 89%，但推理延迟飙到了 180ms，直接超出业务要求。
各方案在准确率和延迟之间的取舍，比单看 F1 更能解释为什么最终选了混合特征。
Embedding 方案准确率最高却超标，混合特征在 88% 准确率和 75ms 延迟之间找到了可部署的平衡点。
最终方案：混合特征
最后定了混合特征方案：TF-IDF 负责捕捉关键词，Embedding 负责捕捉语义，拼接后用 LightGBM 分类。
from scipy.sparse import hstack
def create_hybrid_features(texts: List[str]) -> np.ndarray:
"""创建混合特征"""
# TF-IDF 特征
tfidf_features = vectorizer.transform(texts)
# Embedding 特征
embedding_features = get_embeddings(texts)
# 拼接
combined = hstack([tfidf_features, embedding_features])
return combined
准确率 88%，延迟 75ms，在业务可接受范围内。
模型训练与优化
模型训练这部分遇到的坑比想象的多。
类别不平衡处理
训练集里类别分布很不均匀：最多的类别占 15%，最少的只有 1.2%。导致模型倾向于预测多数类别。
试了几种方案：
方案1：过采样少数类
from imblearn.over_sampling import SMOTE
smote = SMOTE(random_state=42)
X_resampled, y_resampled = smote.fit_resample(X_train, y_train)
效果一般，而且 SMOTE 生成的样本质量不稳定。
方案2：调整类别权重
import numpy as np
# 计算类别权重
class_counts = np.bincount(y_train_encoded)
total_samples = len(y_train_encoded)
class_weights = {
i: total_samples / (len(class_counts) * count)
for i, count in enumerate(class_counts)
}
# 应用权重
model = lgb.LGBMClassifier(
n_estimators=200,
class_weight=class_weights,
# ... 其他参数
)
这个方案效果不错，而且实现简单。
超参数优化
用 Optuna 做超参数优化：
import optuna
def objective(trial):
"""Optuna 优化目标函数"""
params = {
'n_estimators': trial.suggest_int('n_estimators', 100, 500),
'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.2),
'max_depth': trial.suggest_int('max_depth', 4, 10),
'num_leaves': trial.suggest_int('num_leaves', 15, 127),
'min_child_samples': trial.suggest_int('min_child_samples', 10, 50),
'subsample': trial.suggest_float('subsample', 0.6, 1.0),
'colsample_bytree': trial.suggest_float('colsample_bytree', 0.6, 1.0),
'reg_alpha': trial.suggest_float('reg_alpha', 0, 1.0),
'reg_lambda': trial.suggest_float('reg_lambda', 0, 1.0),
}
model = lgb.LGBMClassifier(**params, random_state=42)
model.fit(X_train_tfidf, y_train_encoded)
# 交叉验证
cv_scores = cross_val_score(model, X_train_tfidf, y_train_encoded, cv=5)
return cv_scores.mean()
# 运行优化
study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=50)
print(f"最佳参数: {study.best_params}")
print(f"最佳分数: {study.best_value:.4f}")
跑了 50 次试验，找到了一组不错的参数，但比默认参数只提升了 1.2% 的准确率。算是验证了 LightGBM 默认参数已经足够好用的说法。
模型部署：从 pkl 到 ONNX
部署这块经历了几次重构。
第一版：直接部署 pkl 文件
一开始用的是最直接的方式：
# 部署脚本
import joblib
import pickle
# 加载模型
model = joblib.load('model.pkl')
vectorizer = joblib.load('vectorizer.pkl')
label_encoder = joblib.load('label_encoder.pkl')
# 推理函数
def predict(text: str) -> dict:
"""单条预测"""
# 预处理
cleaned = improved_clean(text)
normalized = smart_normalize(cleaned)
tokens = ' '.join(jieba.cut(normalized))
# 特征提取
features = vectorizer.transform([tokens])
# 预测
proba = model.predict_proba(features)[0]
predicted_class = model.predict(features)[0]
label = label_encoder.inverse_transform([predicted_class])[0]
return {
'label': label,
'confidence': float(max(proba)),
'probabilities': {
label_encoder.classes_[i]: float(p)
for i, p in enumerate(proba)
}
}
用 Flask 起服务：
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/predict', methods=['POST'])
def api_predict():
data = request.json
text = data.get('text', '')
if not text:
return jsonify({'error': 'text is required'}), 400
result = predict(text)
return jsonify(result)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)
这个方案简单粗暴，但有几个问题：
- 每次请求都要重新加载分词词典，初始化成本高
- 模型加载在全局，多线程可能有问题
- 无法批量处理，吞吐量低
第二版：ONNX 优化
为了降低延迟，尝试了 ONNX 转换：
import onnxruntime as ort
from skl2onnx import convert_sklearn
from skl2onnx.common.data_types import FloatTensorType
# 转换 LightGBM 模型到 ONNX
initial_type = [('float_input', FloatTensorType([None, X_train_tfidf.shape[1]]))]
onnx_model = convert_sklearn(
model,
initial_types=initial_type,
target_opset=12
)
# 保存
with open('model.onnx', 'wb') as f:
f.write(onnx_model.SerializeToString())
# ONNX 推理
ort_session = ort.InferenceSession('model.onnx')
def predict_onnx(text: str) -> dict:
"""使用 ONNX 进行推理"""
# 预处理和特征提取（同上）
features = vectorizer.transform([text]).astype(np.float32)
# ONNX 推理
ort_inputs = {ort_session.get_inputs()[0].name: features}
ort_outputs = ort_session.run(None, ort_inputs)
probabilities = ort_outputs[0][0]
predicted_class = np.argmax(probabilities)
return {
'label': label_encoder.inverse_transform([predicted_class])[0],
'confidence': float(max(probabilities))
}
ONNX 转换后推理延迟从 45ms 降到了 32ms，提升了 29%。但转化的过程踩了不少坑：
- LightGBM 的某些参数在 ONNX 转换时不支持，需要手动降级
- TF-IDF 矩阵是稀疏的，转成密集后内存占用暴增
- ONNX Runtime 版本和 skl2onnx 版本要严格对应，否则会报错
第三版：批量处理 + 缓存
最终的部署方案支持批量处理和结果缓存：
import hashlib
from typing import List
from functools import lru_cache
class TextClassifier:
"""文本分类器"""
def __init__(self):
self.model = joblib.load('model.pkl')
self.vectorizer = joblib.load('vectorizer.pkl')
self.label_encoder = joblib.load('label_encoder.pkl')
def preprocess_batch(self, texts: List[str]) -> List[str]:
"""批量预处理"""
return [
' '.join(jieba.cut(smart_normalize(improved_clean(text))))
for text in texts
]
def predict_batch(self, texts: List[str]) -> List[dict]:
"""批量预测"""
# 预处理
processed = self.preprocess_batch(texts)
# 特征提取
features = self.vectorizer.transform(processed)
# 批量预测
proba = self.model.predict_proba(features)
predictions = self.model.predict(features)
# 格式化结果
results = []
for i, pred in enumerate(predictions):
results.append({
'label': self.label_encoder.inverse_transform([pred])[0],
'confidence': float(max(proba[i]))
})
return results
@lru_cache(maxsize=1000)
def predict_cached(self, text: str) -> dict:
"""带缓存的预测"""
return self.predict_batch([text])[0]
# FastAPI 服务
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
classifier = TextClassifier()
class PredictionRequest(BaseModel):
text: str
class BatchPredictionRequest(BaseModel):
texts: List[str]
@app.post('/predict')
def predict(request: PredictionRequest):
if not request.text:
raise HTTPException(status_code=400, detail='text is required')
result = classifier.predict_cached(request.text)
return result
@app.post('/predict_batch')
def predict_batch(request: BatchPredictionRequest):
if not request.texts:
raise HTTPException(status_code=400, detail='texts is required')
results = classifier.predict_batch(request.texts)
return {'results': results}
批量处理后吞吐量提升了 4 倍，缓存让相同文本的延迟降低到 2ms 以内。
监控与反馈
上线后做了基础的监控：
import time
from collections import defaultdict
class PredictionMonitor:
"""预测监控"""
def __init__(self):
self.request_count = 0
self.latency_sum = 0
self.label_distribution = defaultdict(int)
def record(self, label: str, latency: float):
"""记录一次预测"""
self.request_count += 1
self.latency_sum += latency
self.label_distribution[label] += 1
def get_stats(self) -> dict:
"""获取统计信息"""
avg_latency = self.latency_sum / self.request_count if self.request_count > 0 else 0
return {
'total_requests': self.request_count,
'average_latency_ms': avg_latency * 1000,
'label_distribution': dict(self.label_distribution)
}
monitor = PredictionMonitor()
# 在预测函数里记录
def predict_with_monitor(text: str) -> dict:
start_time = time.time()
result = classifier.predict_cached(text)
latency = time.time() - start_time
monitor.record(result['label'], latency)
return result
监控上线后发现了几个问题：
- 某些类别的预测频率远高于训练集分布，说明数据可能漂移了
- 延迟在凌晨 2-4 点会突然升高，是因为服务器有定时任务在跑
- 周末的请求量比工作日少 40%，但准确率反而高了 5%，推测周末的文本质量更高
这些信息让我们能及时调整训练数据和部署策略。
踩过的主要坑
1. 分词不一致导致的灾难
前面提到过，训练和部署用了不同分词模式。这个坑其实可以避免：把分词逻辑封装成独立的模块，训练和部署都用同一个。
# tokenizer.py
import jieba
from typing import List
class TextTokenizer:
"""文本分词器"""
def __init__(self, user_dict_path: str = None):
if user_dict_path:
jieba.load_userdict(user_dict_path)
def tokenize(self, text: str) -> List[str]:
"""分词"""
text = improved_clean(text)
text = smart_normalize(text)
return list(jieba.cut(text, cut_all=False)) # 强制用精确模式
2. 内存泄露
服务运行一段时间后内存持续增长。排查发现有几个原因：
- LRU 缓存设置太大，导致大量长文本被缓存
- TF-IDF 向量没有及时释放
- 每次预测都创建新的临时对象
改进：
# 限制缓存大小和缓存对象的大小
@lru_cache(maxsize=500)
def predict_cached(text: str) -> dict:
if len(text) > 1000: # 长文本不缓存
return self.predict_batch([text])[0]
return self.predict_batch([text])[0]
# 及时释放资源
import gc
def predict_batch_with_cleanup(self, texts: List[str]) -> List[dict]:
"""批量预测并清理资源"""
results = self.predict_batch(texts)
gc.collect() # 手动触发垃圾回收
return results
3. 特征顺序不一致
某次部署后准确率突然下降。查下来是 TF-IDF 特征提取时，词汇表的顺序和训练时不一致。
解决：
# 训练时保存词汇表
import json
vocab_dict = {word: idx for idx, word in enumerate(vectorizer.get_feature_names_out())}
with open('vocab.json', 'w') as f:
json.dump(vocab_dict, f)
# 部署时加载词汇表
with open('vocab.json', 'r') as f:
vocab_dict = json.load(f)
# 使用固定词汇表
vectorizer = TfidfVectorizer(vocabulary=vocab_dict)
经验总结
折腾了半年，这条链路算是基本稳定了。回过头看，有几件事是对的：
先跑通，再优化。没有一开始就上最复杂的方案，而是从最基础的 TF-IDF + 朴素贝叶斯开始，逐步迭代。
监控比模型更重要。如果没有线上监控，永远不知道模型在真实环境下的表现。
性能优化要基于真实数据。很多时候优化方向是基于假设，但实际瓶颈可能完全不同。比如我们以为模型推理是瓶颈，结果发现分词才是。
简单方案往往足够。ONNX 优化了半天，但大部分延迟其实是在分词和特征提取上。
保留回退方案。每次重大改动都保留一个可回滚的版本，这是从多次事故中学会的教训。
留给未来的问题
现在这套方案还有几个地方没完全解决：
在线学习：数据会持续漂移，但定期重新训练成本较高。正在研究增量学习方案。
多语言支持：现在只支持中文，业务方有其他语言的需求。
模型解释：有时候业务方会问"为什么这么分类”，目前只能给个置信度，更细粒度的解释还做不到。
A/B 测试：多版本模型同时上线，按流量分配，这个还没做。
这些问题有些是技术问题，有些是资源问题。技术部分可以继续搞，资源部分得看业务优先级。
最后
NLP 流水线不是一堆组件的简单堆砌，而是一个需要不断调整和优化的系统。从原始文本到线上预测，每一步都有坑，每一步都需要经验。
工具会换，模型会变，但核心问题不变：如何让文本数据变成可理解的、可计算的、可维护的特征，再用这些特征做出稳定的预测。
这件事没有终点，只有不断的踩坑和填坑。
希望这篇文章对正在路上的人有点用。
主要参考：
- jieba 分词文档：https://github.com/fxsjy/jieba
- LightGBM 官方文档：https://lightgbm.readthedocs.io/
- ONNX Runtime 文档：https://onnxruntime.ai/docs/
- FastAPI 文档：https://fastapi.tiangolo.com/
版权声明: 本文首发于 指尖魔法屋-NLP流水线实践笔记（https://blog.thinkmoon.cn/post/156-nlp-pipeline-practice-preprocessing-to-deployment/） 转载或引用必须申明原指尖魔法屋来源及源地址！
评论
使用 GitHub 账号登录后即可留言，支持 Markdown。