做社区、电商、IM 私信、商品问答这类业务的开发者大概率都在后台见过这么一句话哈喽我们是源头工厂有需要就找我呀。它不算脏话没有敏感词也没有攻击性甚至礼貌得有点过分可它就是会把评论区、私信列表、工单池刷成一堵墙。更麻烦的是它今天叫源头工厂明天叫一手货源后天变成厂家直发私信报价再过两天开始夹 emoji 间隔、全角符号、拼音首字母和拆字写法。很多团队的第一版方案都是拉一个关键词黑名单把工厂批发拿货价加微信塞进去命中就拦。上线第一天效果不错第三天开始误杀正常用户的提问比如你们这家店是工厂直营吗第七天开始漏放对方把字拆开、中间插符号第三十天词库变成几百条没人敢删的祖传配置。这不是词库不够大的问题是方案结构的问题。本文的判断很明确这类招商引流话术的识别靠单点关键词匹配已经没有出路需要把它当成一个分层的文本风控系统来做——文本归一化兜住变体、规则层负责高精度召回、模型层负责泛化打分、行为层负责把内容信号放大成处置动作最后用分级处置代替一刀切拦截。这套结构的价值不在于模型多先进而在于它能被你团队持续维护、能解释、能回滚。读完这篇你可以拿到一套可以跑通的最小实现从环境准备、目录结构、归一化脚本、规则打分器、TF-IDF 加逻辑回归的训练脚本到一个 FastAPI 推理服务再到离线评估和线上灰度的验证方法。中间会讲清楚每个环节为什么这么设计、做错了会出现什么现象以及在生产环境里真正容易翻车的地方在哪里。全文代码都完整可复制你可以直接拿去改。1. 一句源头工厂为什么会把内容识别系统逼到墙角先把这句话拆开看。它其实是三段式结构问候语哈喽 身份宣称我们是源头工厂 行动号召有需要就找我呀。这种结构在营销话术里极其稳定变的是词不变的是骨架。你今天封了源头工厂明天对方换成我们是义乌档口自家工厂直发一手货源招代理骨架一模一样你的黑名单一条都命中不了。第二个难点是它天然长在灰区。同样一句我们是源头工厂发在商家的店铺公告里是完全正常的自我介绍发在一个从没买过东西的新账号私信里就是引流发在一个正常用户的咨询里请问你们是源头工厂还是经销商甚至是平台希望鼓励的信息透明度。所以问题不在于这句话是不是垃圾而在于这句话在什么上下文里、由谁、以什么频率、发给谁。脱离上下文的纯文本分类天花板很低。第三个难点是对抗成本极低。符号插入、全角半角混用、emoji 分隔、重复字符、繁简混写、拼音首字母、数字谐音任何一种做法都只要改一次文案而你的词库要跟着改一次。这种不对称意味着如果你把所有希望寄托在人工维护词库上这场仗从结构上就是输的。真正能压住成本的是归一化把变体拉回同一个平面这一步它让后面的规则和模型只面对一种形态。所以这一章的小结论是你面对的不是某个词而是一类意图 一类行为模式。识别系统的设计目标应该是——用最低的维护成本稳定识别出身份宣称 引导联系 批量发送这三个信号的组合而不是去追无穷无尽的词表。想清楚这一点后面的技术选型才有意义。2. 话术识别到底在识别什么概念、指标与技术选型从工程视角看这本质上是一个文本二分类任务给定一条文本还可以带上发送者、接收者、时间、频率等上下文判断它是不是营销引流类内容。但直接上一个大模型做二分类并不是最优起点因为你要考虑三件事推理成本、可解释性、以及误杀的代价。误杀一条正常用户的咨询可能直接损失一单生意这个代价比漏放几条广告高得多。先统一几个后面会反复出现的术语。召回指的是该抓的有没有抓到对应指标是召回率Recall精确率Precision指的是抓到的里面有多少是对的防止误杀两者此消彼长用一个综合指标F1平衡。混淆矩阵就是把预测结果和真实标签交叉成四个格子能一眼看出你是漏放多还是误杀多。阈值是决策的分界线模型输出一个 0 到 1 的分数超过某个值才判定为正类调阈值就是调精确率和召回率的平衡点。技术选型上不同方案的成本和收益差别很大下面这张表是我在实际项目里会拿来和团队对齐的判断依据方案维护成本泛化能力可解释性适合的阶段纯关键词黑名单低开高走词库会失控很弱变体即失效强能直接给出命中词冷启动兜底不建议单独使用正则 归一化中等规则可复用中等能覆盖结构变体强能解释命中原因规则层主力配合模型使用TF-IDF 线性模型低训练秒级中等偏上对同分布文本有效中等可看特征权重首选基线快速上线FastText / 小模型中等需要较多标注较强较弱标注量上来后的升级选项预训练模型微调高需要 GPU 和标注强弱需要额外解释手段有明确效果瓶颈时再上大模型判别中主要是推理成本强对语义理解好中等可以让它给理由人工复核辅助、疑难样本判定更稳妥的判断是先做归一化 规则再上 TF-IDF 线性模型做基线把分级处置和行为特征这两层做扎实最后才考虑要不要上预训练模型。原因很实际——在绝大多数业务里真正把效果拖垮的不是模型不够强而是变体没处理、误杀没申诉通道、阈值没人调、上线没有灰度。这些问题换什么模型都解决不了。这里还要强调一个容易被忽略的概念处置动作不等于拦截。行业里比较成熟的做法是分级——放行、折叠默认收起、点击展开、限流限制发送频次、拦截直接不发出、人工复核。分级的意义在于你把 0.5 到 0.75 这个不确定区间交给折叠而不是拦截用户的正常咨询只是被收起了一下而营销内容已经失去了曝光收益。这一步能显著降低误杀的负面影响。3. 环境准备与工程目录结构这套东西对硬件要求很低一台普通开发机足够跑通全流程。语言用 Python版本建议 3.9 及以上3.10、3.11 都可以再低的版本在类型标注和部分依赖上会有兼容问题不建议。依赖管理用 pip 加 requirements.txt 就够如果你团队已经在用 poetry 或 conda按自己的规范来即可不影响本文的代码逻辑。核心依赖只有四个方向文本处理和向量化用 scikit-learn模型持久化用 joblib服务化用 FastAPI 加 uvicornuvicorn 是 ASGI 服务器负责真正把 HTTP 请求接进来。分词这一块我特意没有引入 jieba因为中文短文本用字符级 n-gram往往比词级更稳——营销话术里造词、拆字、生造缩写的比例太高词表覆盖不住而字符 n-gram 不需要任何词典天然能吃到源头工厂一手货源这类组合模式。先看依赖文件# 文件路径requirements.txt fastapi0.100 uvicorn[standard]0.23 scikit-learn1.2 joblib1.2 pydantic1.10安装命令很直接建议先建虚拟环境再装避免污染系统 Python# 创建并激活虚拟环境Windows 用 .venv\Scripts\activate python -m venv .venv source .venv/bin/activate # 安装依赖 pip install -r requirements.txt目录结构建议按预处理 / 规则 / 模型 / 服务四层拆开这样后面加规则、换模型、改接口时互不影响。很多团队一开始图省事把所有逻辑塞进一个文件等到规则有两百条、模型换了三版之后就没人敢动那个文件了。目录规划本身不是技术难点但它决定了这个系统半年后还能不能维护sales-pitch-detector/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 服务入口 │ ├── preprocess/ │ │ ├── __init__.py │ │ └── normalize.py # 文本归一化 │ ├── rules/ │ │ ├── __init__.py │ │ └── scorer.py # 规则打分器 │ └── features/ │ ├── __init__.py │ └── fingerprint.py # 文本指纹SimHash ├── data/ │ └── samples.py # 演示用标注样例 ├── models/ │ └── spam_clf.joblib # 训练产物脚本自动生成 ├── scripts/ │ ├── train.py # 训练脚本 │ └── evaluate.py # 离线评估脚本 └── requirements.txt还有一个前置条件必须说清楚这套系统需要你准备一批自己的标注数据。不同平台的文案风格差异极大电商私信、游戏公屏、社区评论、招聘平台留言正负样本的分布完全不同。本文代码里内置的样例数据只用于跑通流程、验证代码正确性样本量在几十条量级不能用来评估真实效果也不要把它当成结论。真正的效果评估必须在你自己的标注集上做。4. 文本归一化与特征设计把变体拉回同一个平面归一化是整个系统的地基也是投入产出比最高的一步。它的目标是把哈喽我们是源头工厂有需要就找我呀、哈喽 我们是源头工厂 有需要就找我呀、哈啰我们是源头工厂有需要就找我呀这三种写法尽量收敛成同一个紧凑串。收敛得越好后面的规则和模型就越不需要为变体买单。归一化的处理顺序是有讲究的不能随便打乱。第一步必须是 Unicode 归一化NFKC它负责把全角字母数字、全角空格、兼容字符统一成标准形态这一步在前后面的正则才能写得干净。第二步去掉零宽字符这类字符肉眼看不见却能把工厂两个字从中间劈开是很多人排查半天找不到原因的地方。第三步统一小写、去掉分隔噪声符号、压缩重复字符最后再收敛成只保留中文、字母、数字的紧凑串。具体实现如下# 文件路径app/preprocess/normalize.py import re import unicodedata # 零宽字符集合翻译时映射为 None 即删除 ZERO_WIDTH dict.fromkeys(map(ord, \u200b\u200c\u200d\ufeff), None) def normalize(text: str) - str: 把一条文本收敛成紧凑的判别形态。 处理顺序不可随意调换 1. NFKC 归一化必须在最前负责全角转半角、兼容字符统一 2. 去零宽字符防止词语被隐形字符劈开 3. 转小写统一英文和拼音首字母 4. 去掉常见分隔噪声防止“源-头-工-厂”这类拆分绕过 5. 压缩重复字符防止“有有有需要”这类填充 6. 只保留中文、字母、数字 if not text: return text unicodedata.normalize(NFKC, text).translate(ZERO_WIDTH) text text.lower() # 去掉空白与常见装饰性符号 text re.sub(r[\s\.\-_\*\~·•、。!?,;:\“”‘’()\[\]【】], , text) # 三个及以上重复字符压缩成两个保留一点强度信息又不至于干扰 text re.sub(r(.)\1{2,}, r\1\1, text) # 只留中文、字母、数字 text re.sub(r[^\u4e00-\u9fa5a-z0-9], , text) return text代码里有几个细节值得单独说。第一步的NFKC是关键它把全角变成A把变成1很多看起来毫无规律的变体其实只是全角半角混排这一步能一网打尽。第五步把三个以上重复字符压成两个而不是一个是因为便宜便宜便宜和便宜在语义强度上确实不同保留两个既能降低噪声又不会把强度信息完全抹掉。最后的字符过滤会连 emoji 一起去掉这在你只做文本判别时是合理的。4.1 文本指纹识别一句话发给一万个人归一化还有一个非常有价值的副产品文本指纹。营销引流的核心特征不是文案长什么样而是同一套文案被批量发送。如果你能算出一条文本的近似指纹就能在内容判别之外再加一条极强的信号过去 10 分钟内有 N 个不同账号发了指纹相同或高度相似的内容。这里用 SimHash 实现它的好处是对小改动不敏感——改几个标点、加两个 emoji指纹的汉明距离依然很小# 文件路径app/features/fingerprint.py import hashlib from app.preprocess.normalize import normalize def _shingles(text: str, size: int 2) - list: 把归一化文本切成字符 shingle作为指纹的最小单元。 text normalize(text) if not text: return [] if len(text) size: return [text] return [text[i:i size] for i in range(len(text) - size 1)] def simhash(text: str, bits: int 64) - int: 计算 64 位 SimHash同样的输入永远得到同样的输出。 vector [0] * bits for token in _shingles(text): digest int(hashlib.md5(token.encode(utf-8)).hexdigest(), 16) for i in range(bits): vector[i] 1 if (digest i) 1 else -1 result 0 for i in range(bits): if vector[i] 0: result | 1 i return result def hamming_distance(a: int, b: int) - int: 汉明距离两个指纹有多少位不同。 return bin(a ^ b).count(1) def is_near_duplicate(a: int, b: int, threshold: int 3) - bool: 距离小于等于阈值视为近似重复。阈值 3 是常见经验值需按业务调。 return hamming_distance(a, b) threshold这十几行代码带来的收益往往比把模型从线性换成深度模型更大。原因在于它抓到的是行为层面的事实——批量发送本身就是违规行为的直接证据而不是靠语义猜测。实际落地时你可以把指纹存进 Redis 之类的缓存用SET加过期时间记录指纹 - 最近发送账号集合再做频次统计。4.2 规则层的三类信号规则层不是简单堆正则而是要按信号类别组织。我把它们分成三组身份宣称我在卖货/我是工厂/我招代理、行动号召找我/私聊/报价/拿货以及联系方式手机号、微信号、邮箱。这三组信号单独出现时都可能有正常含义但两组以上同时出现判别力就非常强了。这正是那句源头工厂 有需要就找我的骨架命中身份宣称加行动号召规则分直接到 0.7已经站在了可疑区间里。而一句请问你们是源头工厂吗只命中身份宣称得到 0.35不足以触发任何处置——误杀就这样被结构性地避免掉了。5. 完整实现规则召回 模型打分 决策融合这一章是整个流程的核心分三层写规则层负责高精度可解释的召回模型层负责泛化和兜底服务层负责把两者融合成分级处置动作。三层各司其职任何一层出问题都能单独定位、单独回滚。5.1 规则层分组打分与命中解释规则层最重要的设计不是正则写得多巧而是分组加权。分组让每条规则的贡献可解释加权让你能通过调权重来控制误杀率而不用改规则本身。# 文件路径app/rules/scorer.py import re from dataclasses import dataclass, field from app.preprocess.normalize import normalize # 组 1身份宣称——明确表达自己在供货、批发、招商 IDENTITY_PATTERNS [ r(源头|厂家|工厂|一手|批发|档口|供货商|供应链), r(直销|直供|入驻|招商|招代理|找客户|找合作|求合作), ] # 组 2行动号召——引导对方发生联系动作 CALL_TO_ACTION_PATTERNS [ r(有需要就找我|有需要找我|可以联系我|欢迎咨询|欢迎来撩), r(私聊|私信|戳我|滴滴我|找我聊), r(报价|拿货价|出厂价|一件代发|免费寄样|量大从优), ] # 组 3联系方式——必须匹配原始文本因为归一化会删掉 . 等符号 CONTACT_PATTERNS [ r(?:vx|wx|v信|威信|薇信|加v|加微|扣扣|qq)\s*[:]?\s*[a-z0-9]{5,20}, r1[3-9]\d{9}, r[a-z0-9._%-][a-z0-9.-]\.[a-z]{2,}, ] WEIGHT_IDENTITY 0.35 WEIGHT_CALL_TO_ACTION 0.35 WEIGHT_CONTACT 0.40 dataclass class RuleResult: score: float hits: list field(default_factorylist) matched: list field(default_factorylist) def rule_score(text: str) - RuleResult: 按信号组打分返回 0~1 的分数和可解释的命中明细。 raw (text or ).lower() norm normalize(text) hits, matched [], [] score 0.0 for pattern in IDENTITY_PATTERNS: if re.search(pattern, norm): score WEIGHT_IDENTITY hits.append(identity) matched.append(pattern) break for pattern in CALL_TO_ACTION_PATTERNS: if re.search(pattern, norm): score WEIGHT_CALL_TO_ACTION hits.append(call_to_action) matched.append(pattern) break # 联系方式用原始文本匹配归一化会破坏邮箱和号码的结构 for pattern in CONTACT_PATTERNS: if re.search(pattern, raw): score WEIGHT_CONTACT hits.append(contact) matched.append(pattern) break return RuleResult(scoreround(min(score, 1.0), 4), hitshits, matchedmatched)几个设计取舍要说明。第一每组只记一次命中用break跳出避免同一组里多条正则重复累加导致分数虚高。第二联系方式必须在原始文本上匹配因为归一化把、.、空格都删了邮箱会被压成一个无法识别的长串这是新手最容易踩的坑。第三matched字段保存了命中的正则原文这在排查误杀时极其有用——你能直接告诉运营这条是因为命中了哪条规则被处理的而不是一句模型说的。可解释性在生产环境里的价值往往比一两个点的准确率更重要。5.2 模型层字符 n-gram 加线性分类器模型层我选 TF-IDF 加逻辑回归理由很实际训练秒级完成、样本需求低、推理快、权重可读、几乎没有运维负担。对一个刚起步的识别系统来说这比一上来就微调预训练模型理性得多。文本特征用字符级 1 到 3 元组能覆盖中文的常见字、双字词和三字短语组合对源头工厂一手货源一件代发这类短语有天然捕捉力。# 文件路径scripts/train.py import os import joblib from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.linear_model import LogisticRegression from sklearn.pipeline import Pipeline from app.preprocess.normalize import normalize # 演示样例仅用于验证代码可以跑通不代表真实数据分布 # 真实项目中请替换为你自己标注的数据集并要求正负样本比例接近 1:1 POSITIVE_SAMPLES [ 哈喽我们是源头工厂有需要就找我呀, 工厂直销一件代发量大从优加v详聊, 你好我们是义乌档口批发拿货价便宜, 一手货源招代理欢迎咨询价格美丽, 厂家直供支持免费寄样私信我报价, 我们是供应链源头有需要可以联系我, 自家工厂出货价格比市场低欢迎来撩, 批发档口新到一批货老板们看过来, ] NEGATIVE_SAMPLES [ 请问你们是源头工厂还是经销商, 我买的东西三天了还没发货, 这个商品的售后电话是多少, 想咨询一下你们工厂直营店在哪里, 订单显示已签收但我没有收到, 能不能帮我改一下收货地址, 这个型号和上次买的有什么区别, 发票什么时候能开出来, ] def build_pipeline() - Pipeline: return Pipeline([ (tfidf, TfidfVectorizer( analyzerchar, # 中文短文本用字符级更稳 ngram_range(1, 3), # 单字、双字、三字短语 min_df1, sublinear_tfTrue, )), (clf, LogisticRegression( max_iter1000, class_weightbalanced, # 正负样本不平衡时自动加权 C1.0, )), ]) def main() - None: texts [normalize(t) for t in POSITIVE_SAMPLES NEGATIVE_SAMPLES] labels [1] * len(POSITIVE_SAMPLES) [0] * len(NEGATIVE_SAMPLES) pipeline build_pipeline() pipeline.fit(texts, labels) os.makedirs(models, exist_okTrue) joblib.dump(pipeline, models/spam_clf.joblib) print(f训练完成样本数{len(texts)}正样本{len(POSITIVE_SAMPLES)}) print(模型已保存到 models/spam_clf.joblib) # 快速自检打印几条典型输入的预测概率 for probe in [哈喽我们是源头工厂有需要找我, 请问什么时候发货]: proba pipeline.predict_proba([normalize(probe)])[0] positive proba[list(pipeline.classes_).index(1)] print(f {probe} - 正类概率 {positive:.4f}) if __name__ __main__: main()代码里有几点必须讲清楚。analyzerchar是中文短文本的常用选择如果用默认的word中文没有空格会被当成一个整词特征全废char_wb看起来更高级但它只在词边界内切 n-gram中文没有空格时同样不适用这个坑很多人踩过。class_weightbalanced在你正负样本接近 1:1 时其实没什么影响但一旦上线后正样本比例变化它能防止模型被多数类带偏。sublinear_tfTrue用对数缩放词频能压低高频字比如的我你的权重对短文本有帮助。最后那段自检逻辑也要保留。训练完不验证就打包上线是很多事故的源头。取正类概率时用list(pipeline.classes_).index(1)而不是直接写[0][1]是为了避免类别顺序变化导致的静默错误——这种错误不会报错只会让结果莫名其妙地反了。5.3 服务层把两个分数融合成分级动作服务层要把规则分和模型分融合再结合行为特征输出一个处置动作。融合权重没有标准答案我一般从模型 0.5、规则 0.5开始在标注集上按误杀率约束去调。真正的关键是阈值分级以及行为特征的加成逻辑。# 文件路径app/main.py import logging import joblib from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from app.preprocess.normalize import normalize from app.rules.scorer import rule_score from app.features.fingerprint import simhash logger logging.getLogger(pitch-detector) logging.basicConfig(levellogging.INFO) MODEL_PATH models/spam_clf.joblib BLOCK_THRESHOLD 0.75 # 大于等于则拦截 REVIEW_THRESHOLD 0.50 # 介于两者之间则折叠/人工复核 WEIGHT_MODEL 0.5 WEIGHT_RULE 0.5 app FastAPI(titleSales Pitch Detector, version1.0.0) try: model joblib.load(MODEL_PATH) except FileNotFoundError: model None logger.warning(模型文件不存在请先执行 scripts/train.py 生成模型) class CheckRequest(BaseModel): text: str Field(..., max_length2000) sender_account_age_days: int 30 sender_daily_msg_count: int 0 class CheckResponse(BaseModel): action: str final_score: float model_score: float rule_score: float hits: list fingerprint: int def _model_probability(text: str) - float: if model is None: return 0.0 proba model.predict_proba([normalize(text)])[0] return float(proba[list(model.classes_).index(1)]) def _behavior_bonus(req: CheckRequest) - float: 行为特征加成新账号、高频发送会把可疑度往上抬。 bonus 0.0 if req.sender_account_age_days 7: bonus 0.05 if req.sender_daily_msg_count 50: bonus 0.05 return bonus app.post(/v1/check, response_modelCheckResponse) def check(req: CheckRequest): if not req.text or not req.text.strip(): raise HTTPException(status_code400, detailtext 不能为空) rule rule_score(req.text) model_prob _model_probability(req.text) combined WEIGHT_MODEL * model_prob WEIGHT_RULE * rule.score final round(min(combined _behavior_bonus(req), 1.0), 4) if final BLOCK_THRESHOLD: action block elif final REVIEW_THRESHOLD: action collapse else: action pass logger.info( check action%s final%.4f model%.4f rule%.4f hits%s, action, final, model_prob, rule.score, rule.hits, ) return CheckResponse( actionaction, final_scorefinal, model_scoreround(model_prob, 4), rule_scorerule.score, hitsrule.hits, fingerprintsimhash(req.text), )这里有两处值得展开。第一是融合方式用加权和而不是纯max加权和的好处是两路信号都弱时不会误判两路都强时分数会被推高max会让单路强信号直接决定结果误杀风险更高。第二是行为加成的幅度刻意做得很小各 0.05因为行为特征的误判代价更高——新注册用户是正常用户一天发 60 条消息也可能是个活跃买家。把它们作为加分项而不是决定项是最稳妥的做法。另外注意action的命名用的是collapse而不是review因为在真实产品里落地的往往是折叠展示这个前端动作而不是进入人工队列。虽然它背后可以对应同一个区间但用产品语言命名能让前后端对齐得更快。6. 运行验证与效果评估代码写完不验证等于没写。验证分两步先确认服务能跑起来、接口返回符合预期再回到离线数据集上确认指标和错误样本。先训练再启动服务# 步骤 1训练模型生成 models/spam_clf.joblib python -m scripts.train # 步骤 2启动服务 uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload训练脚本的预期输出大致是这样正类概率的具体数值取决于你的数据不作为效果结论训练完成样本数16正样本8 模型已保存到 models/spam_clf.joblib 哈喽我们是源头工厂有需要找我 - 正类概率 0.9xxx 请问什么时候发货 - 正类概率 0.0xxx判断成功的标准是目标句的正类概率明显高于 0.5正常咨询句明显低于 0.5。如果两句都接近 0.5说明特征没起作用第一个该看的是analyzer参数有没有写成char第二个该看的是归一化是不是把整句清空了。服务起来之后用 curl 验证接口curl -X POST http://127.0.0.1:8000/v1/check \ -H Content-Type: application/json \ -d { text: 哈喽我们是源头工厂有需要就找我呀, sender_account_age_days: 3, sender_daily_msg_count: 80 }预期返回的结构是这样的{ action: block, final_score: 0.8, model_score: 0.8, rule_score: 0.7, hits: [identity, call_to_action], fingerprint: 1234567890123456789 }final_score的具体数值会随模型和数据变化但结构必须稳定hits里应该能看到identity和call_to_action两组命中action应该是block或者至少collapse。如果你把文本改成请问你们是源头工厂还是经销商hits应该只剩identityrule_score掉到 0.35action变成pass——这就是分组加权带来的误杀保护用这一对输入做回归测试非常有效。再验证一下变体处理能力把符号插进去curl -X POST http://127.0.0.1:8000/v1/check \ -H Content-Type: application/json \ -d {text: 哈 喽我 们 是 源-头-工-厂有 需 要 就 找 我 呀}归一化之后这串会被压成紧凑形式规则层照样命中。如果这条返回pass第一个排查点就是归一化函数里的分隔符正则有没有覆盖到你插入的那类符号。离线评估这一块建议单独写脚本不要混在训练里。评估集要和训练集分开否则指标永远好看上线永远打脸# 文件路径scripts/evaluate.py import joblib from sklearn.metrics import classification_report, confusion_matrix from app.preprocess.normalize import normalize from app.rules.scorer import rule_score # 独立评估集刻意包含“含关键词但正常”的样本用来暴露误杀 EVAL_TEXTS [ 哈喽我们是源头工厂有需要就找我呀, 工厂直销一件代发加v详聊, 一手货源招代理欢迎咨询, 请问你们是源头工厂还是经销商, 订单三天了还没发货怎么办, 能不能开个发票, 你们工厂直营店在哪个城市, 这个型号有现货吗, ] EVAL_LABELS [1, 1, 1, 0, 0, 0, 0, 0] BLOCK_THRESHOLD 0.75 REVIEW_THRESHOLD 0.50 def main() - None: model joblib.load(models/spam_clf.joblib) preds, notes [], [] for text in EVAL_TEXTS: rule rule_score(text) proba model.predict_proba([normalize(text)])[0] model_prob float(proba[list(model.classes_).index(1)]) final 0.5 * model_prob 0.5 * rule.score preds.append(1 if final REVIEW_THRESHOLD else 0) notes.append((text, round(final, 4), rule.hits)) print(混淆矩阵) print(confusion_matrix(EVAL_LABELS, preds)) print() print(classification_report(EVAL_LABELS, preds, zero_division0)) print(逐条明细) for text, score, hits in notes: print(f final{score} hits{hits} | {text}) if __name__ __main__: main()注意这里用的是REVIEW_THRESHOLD0.5作为二分类的分界而不是BLOCK_THRESHOLD0.75。原因是评估阶段我们关心的是有没有被识别为可疑处置等级是另一个维度的事。如果你用 0.75 做评估分界召回率会显著偏低看起来像模型不行其实是口径搞错了。这个脚本跑出来的accuracy、precision、recall、f1-score只能说明流程通了因为样本只有 8 条任何一条的变动都会让指标大幅跳动。真正做效果评估时评估集至少要到几百条量级并且要包含三类容易误判的边界样本含营销关键词的正常咨询、不含关键词的营销变体、以及一句话里既有正常诉求又有引流引导的混合样本。第三类最难建议单独统计。7. 常见问题与排查思路这套系统跑起来不难难的是跑对。下面这张表是我遇到过、以及在看别人实现时最常见的问题按现象—原因—排查—解决整理问题现象可能原因排查方式解决方案模型对所有输入都给接近 0.5 的概率analyzer用了默认的word中文被当成一个整词打印pipeline.named_steps[tfidf].get_feature_names_out()[:20]看特征改成analyzerchar、ngram_range(1,3)插入符号的变体能绕过归一化正则没覆盖到那类符号直接调用normalize()打印结果对比补充分隔符正则把该符号加进字符集邮箱、手机号规则永远命中不了在归一化后的文本上匹配联系方式打印normalize(ab.com)看输出联系方式改在原始小写文本上匹配正常咨询被大量拦截规则权重过高或阈值过低用hits字段统计拦截样本的命中分布降WEIGHT_IDENTITY或把阈值从 0.5 提到 0.75服务启动报FileNotFoundError模型文件未生成检查models/spam_clf.joblib是否存在先执行python -m scripts.train预测结果整体反了取正类概率时写死索引打印model.classes_看顺序用list(model.classes_).index(1)取值训练时警告类别样本过少正负样本严重不平衡打印两类样本数量补样本或设class_weightbalanced同一条文本两次请求结果不同模型被重新加载或混用了不同版本检查部署实例加载的模型文件时间戳模型文件加版本号服务启动时打日志接口返回 422请求体字段类型或长度不符合CheckRequest定义看 FastAPI 返回的detail字段按 schema 传参text控制在 2000 字符内线上 CPU 飙高单条请求重复加载模型或反复编译正则看火焰图确认热点模型在模块级加载一次正则预编译表格里有两行值得单独强调。第一行概率都是 0.5是最高频的问题九成以上是analyzer参数没设对因为中文分词缺失会让整个文本变成一个特征模型学不到任何东西。第六行结果整体反了是最隐蔽的问题它不报错只是逻辑反了往往要等到运营反馈最近拦截的都是正常用户才会被发现所以取概率的写法一定要用index(1)这种防御性写法而不是依赖默认顺序。还有一类问题不在表里但更麻烦词库和规则长期无人维护导致系统逐渐失效。表现是拦截量慢慢下降、误杀慢慢上升但因为变化是渐进的没有报警就没人发现。解决办法是给系统加监控指标而不只是加规则——每天统计拦截量、折叠量、申诉量、申诉通过率一旦申诉通过率超过某个阈值就说明误杀在恶化需要立刻回看规则权重。8. 最佳实践与工程建议这一段是给准备把这套东西上生产的团队看的。前面九成的代码你都可以直接抄但下面这些经验抄不走需要你自己在项目里落地。第一规则和模型都要版本化。规则文件、模型文件、阈值配置三者必须能对应到一个版本号并且在服务启动日志里打出来。原因很现实出问题时你需要在十分钟内回答昨天和今天有什么不同。如果规则是热更新的、模型是滚动覆盖的、阈值是改在环境变量里的你就无法回溯。更稳妥的做法是把规则集也做成配置文件YAML 或 JSON走正常的发布流程而不是让运营在后台直接改。第二必须给用户留申诉通道。任何自动处置系统都会有误杀区别只在于你有没有让用户表达出来的路径。具体做法是折叠或拦截时给一个内容被误判点击申诉的入口申诉样本自动回流到标注队列。这个通道不只是为了用户体验它是你获取高质量负样本的主要来源——真实误杀样本比任何人工构造的负样本都值钱。第三分级处置要克制。前面已经强调过0.5 到 0.75 这个不确定区间应该交给折叠而不是拦截。此外同一个账号即使被判为营销也不建议第一次就永久限制更稳妥的策略是阶梯式首次折叠、再次限流、三次以上才拦截。这样即使模型误判代价也被控制在一屏之内。第四注意数据合规边界。做这类识别时你只需要文本内容、账号基础属性注册时间、历史发送量、以及必要的频次统计。不要顺手把聊天的完整会话、收货地址、支付信息拉进特征也不要在日志里打印完整原文。日志里打的应该是脱敏后的片段、指纹、以及命中的规则 ID。这条不只是合规要求也是安全底线——日志泄露的代价远高于识别效果提升带来的收益。第五冷启动阶段别追求模型效果。上线第一天没有标注数据是常态。这个阶段的正确做法是规则先行 全量抽样人工审核用审核结果积累第一批标注同时把处置动作全部设成折叠这种低风险动作。等你积累了几百条正负样本再训练模型、再考虑提高处置等级。反过来做——先上模型、先拦截、后补数据——几乎一定会出事故。第六把效果监控当成功能来做。至少要有四个指标每日识别量、处置分布pass/collapse/block 各占多少、申诉量与申诉通过率、以及抽样准确率。前三个是实时指标最后一个靠每周抽 200 条人工复核。指标异常时的处理顺序是先看处置分布有没有突变再看申诉样本集中在哪类文本最后才去看模型和规则。这个顺序能让你在十分钟内定位到问题层级。第七阈值要放进配置中心而不是代码里。阈值是所有参数里最需要频繁调整的一个而且调整往往发生在出问题的当下。如果它写在代码里每次调都要走一次发布流程等你发完舆情已经过去了。把它放到配置中心或环境变量配合灰度发布和快速回滚是这类系统的基本要求。9. 从能跑到好用下一步的迭代路径把上面这套跑通之后你已经有了一个结构正确、可解释、可回滚的识别系统。它现在的效果大概只能算能用但结构是对的这意味着后面每一步投入都能叠加而不是推倒重来。接下来最值得投入的不是模型是标注数据的质量。优先补三类样本一类是含营销词但正常的负样本它们直接决定你的误杀率一类是零关键词但明显引流的正样本它们决定模型的泛化上限还有一类是混合样本——前半句是正常咨询、后半句突然引导私聊。这三类样本凑齐之后你会发现自己能很清楚地看到模型的上限在哪里。数据够了之后再考虑模型升级。判断标准很明确如果在线抽样复核发现的漏放案例中有超过一半是语义上能理解但模型没学到的类型那就该换更强的模型了如果漏放主要集中在变体绕过那说明问题在归一化不在模型。这两种情况用错药投入会全部打水漂。最后提醒一句关于范围的事。这类系统的目标是减少骚扰、保护正常用户的沟通体验不是把商家都赶走。它会误判所以必须留申诉它会漏放所以必须有监控它会被对抗所以必须能快速迭代。把这三点当成系统的一部分而不是后续优化你在设计阶段就会做出不同的选择——比如给每条处置都留一个可解释的命中理由比如把阈值放进配置中心比如从一开始就给规则和模型打版本号。想继续深入的话有几个方向比较有价值文本归一化的对抗演进如何用少量样本发现新的绕过手法、近似重复检测的工程实现大规模指纹索引怎么做、以及分级处置策略的 A/B 实验设计怎么科学地判断提高拦截率到底有没有伤害转化。这三块决定了这个系统从能跑到好用的距离。建议把本文的目录结构和三个核心脚本先落到自己的项目里跑一遍尤其是归一化和规则打分这两部分它们不依赖任何模型当天就能上线看到效果。 
长河君 · 2026/10/26 10:35:21 · 阅读 1.2k 分层文本风控实战：识别“源头工厂”类营销引流话术
做社区、电商、IM 私信、商品问答这类业务的开发者大概率都在后台见过这么一句话哈喽我们是源头工厂有需要就找我呀。它不算脏话没有敏感词也没有攻击性甚至礼貌得有点过分可它就是会把评论区、私信列表、工单池刷成一堵墙。更麻烦的是它今天叫源头工厂明天叫一手货源后天变成厂家直发私信报价再过两天开始夹 emoji 间隔、全角符号、拼音首字母和拆字写法。很多团队的第一版方案都是拉一个关键词黑名单把工厂批发拿货价加微信塞进去命中就拦。上线第一天效果不错第三天开始误杀正常用户的提问比如你们这家店是工厂直营吗第七天开始漏放对方把字拆开、中间插符号第三十天词库变成几百条没人敢删的祖传配置。这不是词库不够大的问题是方案结构的问题。本文的判断很明确这类招商引流话术的识别靠单点关键词匹配已经没有出路需要把它当成一个分层的文本风控系统来做——文本归一化兜住变体、规则层负责高精度召回、模型层负责泛化打分、行为层负责把内容信号放大成处置动作最后用分级处置代替一刀切拦截。这套结构的价值不在于模型多先进而在于它能被你团队持续维护、能解释、能回滚。读完这篇你可以拿到一套可以跑通的最小实现从环境准备、目录结构、归一化脚本、规则打分器、TF-IDF 加逻辑回归的训练脚本到一个 FastAPI 推理服务再到离线评估和线上灰度的验证方法。中间会讲清楚每个环节为什么这么设计、做错了会出现什么现象以及在生产环境里真正容易翻车的地方在哪里。全文代码都完整可复制你可以直接拿去改。1. 一句源头工厂为什么会把内容识别系统逼到墙角先把这句话拆开看。它其实是三段式结构问候语哈喽 身份宣称我们是源头工厂 行动号召有需要就找我呀。这种结构在营销话术里极其稳定变的是词不变的是骨架。你今天封了源头工厂明天对方换成我们是义乌档口自家工厂直发一手货源招代理骨架一模一样你的黑名单一条都命中不了。第二个难点是它天然长在灰区。同样一句我们是源头工厂发在商家的店铺公告里是完全正常的自我介绍发在一个从没买过东西的新账号私信里就是引流发在一个正常用户的咨询里请问你们是源头工厂还是经销商甚至是平台希望鼓励的信息透明度。所以问题不在于这句话是不是垃圾而在于这句话在什么上下文里、由谁、以什么频率、发给谁。脱离上下文的纯文本分类天花板很低。第三个难点是对抗成本极低。符号插入、全角半角混用、emoji 分隔、重复字符、繁简混写、拼音首字母、数字谐音任何一种做法都只要改一次文案而你的词库要跟着改一次。这种不对称意味着如果你把所有希望寄托在人工维护词库上这场仗从结构上就是输的。真正能压住成本的是归一化把变体拉回同一个平面这一步它让后面的规则和模型只面对一种形态。所以这一章的小结论是你面对的不是某个词而是一类意图 一类行为模式。识别系统的设计目标应该是——用最低的维护成本稳定识别出身份宣称 引导联系 批量发送这三个信号的组合而不是去追无穷无尽的词表。想清楚这一点后面的技术选型才有意义。2. 话术识别到底在识别什么概念、指标与技术选型从工程视角看这本质上是一个文本二分类任务给定一条文本还可以带上发送者、接收者、时间、频率等上下文判断它是不是营销引流类内容。但直接上一个大模型做二分类并不是最优起点因为你要考虑三件事推理成本、可解释性、以及误杀的代价。误杀一条正常用户的咨询可能直接损失一单生意这个代价比漏放几条广告高得多。先统一几个后面会反复出现的术语。召回指的是该抓的有没有抓到对应指标是召回率Recall精确率Precision指的是抓到的里面有多少是对的防止误杀两者此消彼长用一个综合指标F1平衡。混淆矩阵就是把预测结果和真实标签交叉成四个格子能一眼看出你是漏放多还是误杀多。阈值是决策的分界线模型输出一个 0 到 1 的分数超过某个值才判定为正类调阈值就是调精确率和召回率的平衡点。技术选型上不同方案的成本和收益差别很大下面这张表是我在实际项目里会拿来和团队对齐的判断依据方案维护成本泛化能力可解释性适合的阶段纯关键词黑名单低开高走词库会失控很弱变体即失效强能直接给出命中词冷启动兜底不建议单独使用正则 归一化中等规则可复用中等能覆盖结构变体强能解释命中原因规则层主力配合模型使用TF-IDF 线性模型低训练秒级中等偏上对同分布文本有效中等可看特征权重首选基线快速上线FastText / 小模型中等需要较多标注较强较弱标注量上来后的升级选项预训练模型微调高需要 GPU 和标注强弱需要额外解释手段有明确效果瓶颈时再上大模型判别中主要是推理成本强对语义理解好中等可以让它给理由人工复核辅助、疑难样本判定更稳妥的判断是先做归一化 规则再上 TF-IDF 线性模型做基线把分级处置和行为特征这两层做扎实最后才考虑要不要上预训练模型。原因很实际——在绝大多数业务里真正把效果拖垮的不是模型不够强而是变体没处理、误杀没申诉通道、阈值没人调、上线没有灰度。这些问题换什么模型都解决不了。这里还要强调一个容易被忽略的概念处置动作不等于拦截。行业里比较成熟的做法是分级——放行、折叠默认收起、点击展开、限流限制发送频次、拦截直接不发出、人工复核。分级的意义在于你把 0.5 到 0.75 这个不确定区间交给折叠而不是拦截用户的正常咨询只是被收起了一下而营销内容已经失去了曝光收益。这一步能显著降低误杀的负面影响。3. 环境准备与工程目录结构这套东西对硬件要求很低一台普通开发机足够跑通全流程。语言用 Python版本建议 3.9 及以上3.10、3.11 都可以再低的版本在类型标注和部分依赖上会有兼容问题不建议。依赖管理用 pip 加 requirements.txt 就够如果你团队已经在用 poetry 或 conda按自己的规范来即可不影响本文的代码逻辑。核心依赖只有四个方向文本处理和向量化用 scikit-learn模型持久化用 joblib服务化用 FastAPI 加 uvicornuvicorn 是 ASGI 服务器负责真正把 HTTP 请求接进来。分词这一块我特意没有引入 jieba因为中文短文本用字符级 n-gram往往比词级更稳——营销话术里造词、拆字、生造缩写的比例太高词表覆盖不住而字符 n-gram 不需要任何词典天然能吃到源头工厂一手货源这类组合模式。先看依赖文件# 文件路径requirements.txt fastapi0.100 uvicorn[standard]0.23 scikit-learn1.2 joblib1.2 pydantic1.10安装命令很直接建议先建虚拟环境再装避免污染系统 Python# 创建并激活虚拟环境Windows 用 .venv\Scripts\activate python -m venv .venv source .venv/bin/activate # 安装依赖 pip install -r requirements.txt目录结构建议按预处理 / 规则 / 模型 / 服务四层拆开这样后面加规则、换模型、改接口时互不影响。很多团队一开始图省事把所有逻辑塞进一个文件等到规则有两百条、模型换了三版之后就没人敢动那个文件了。目录规划本身不是技术难点但它决定了这个系统半年后还能不能维护sales-pitch-detector/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 服务入口 │ ├── preprocess/ │ │ ├── __init__.py │ │ └── normalize.py # 文本归一化 │ ├── rules/ │ │ ├── __init__.py │ │ └── scorer.py # 规则打分器 │ └── features/ │ ├── __init__.py │ └── fingerprint.py # 文本指纹SimHash ├── data/ │ └── samples.py # 演示用标注样例 ├── models/ │ └── spam_clf.joblib # 训练产物脚本自动生成 ├── scripts/ │ ├── train.py # 训练脚本 │ └── evaluate.py # 离线评估脚本 └── requirements.txt还有一个前置条件必须说清楚这套系统需要你准备一批自己的标注数据。不同平台的文案风格差异极大电商私信、游戏公屏、社区评论、招聘平台留言正负样本的分布完全不同。本文代码里内置的样例数据只用于跑通流程、验证代码正确性样本量在几十条量级不能用来评估真实效果也不要把它当成结论。真正的效果评估必须在你自己的标注集上做。4. 文本归一化与特征设计把变体拉回同一个平面归一化是整个系统的地基也是投入产出比最高的一步。它的目标是把哈喽我们是源头工厂有需要就找我呀、哈喽 我们是源头工厂 有需要就找我呀、哈啰我们是源头工厂有需要就找我呀这三种写法尽量收敛成同一个紧凑串。收敛得越好后面的规则和模型就越不需要为变体买单。归一化的处理顺序是有讲究的不能随便打乱。第一步必须是 Unicode 归一化NFKC它负责把全角字母数字、全角空格、兼容字符统一成标准形态这一步在前后面的正则才能写得干净。第二步去掉零宽字符这类字符肉眼看不见却能把工厂两个字从中间劈开是很多人排查半天找不到原因的地方。第三步统一小写、去掉分隔噪声符号、压缩重复字符最后再收敛成只保留中文、字母、数字的紧凑串。具体实现如下# 文件路径app/preprocess/normalize.py import re import unicodedata # 零宽字符集合翻译时映射为 None 即删除 ZERO_WIDTH dict.fromkeys(map(ord, \u200b\u200c\u200d\ufeff), None) def normalize(text: str) - str: 把一条文本收敛成紧凑的判别形态。 处理顺序不可随意调换 1. NFKC 归一化必须在最前负责全角转半角、兼容字符统一 2. 去零宽字符防止词语被隐形字符劈开 3. 转小写统一英文和拼音首字母 4. 去掉常见分隔噪声防止“源-头-工-厂”这类拆分绕过 5. 压缩重复字符防止“有有有需要”这类填充 6. 只保留中文、字母、数字 if not text: return text unicodedata.normalize(NFKC, text).translate(ZERO_WIDTH) text text.lower() # 去掉空白与常见装饰性符号 text re.sub(r[\s\.\-_\*\~·•、。!?,;:\“”‘’()\[\]【】], , text) # 三个及以上重复字符压缩成两个保留一点强度信息又不至于干扰 text re.sub(r(.)\1{2,}, r\1\1, text) # 只留中文、字母、数字 text re.sub(r[^\u4e00-\u9fa5a-z0-9], , text) return text代码里有几个细节值得单独说。第一步的NFKC是关键它把全角变成A把变成1很多看起来毫无规律的变体其实只是全角半角混排这一步能一网打尽。第五步把三个以上重复字符压成两个而不是一个是因为便宜便宜便宜和便宜在语义强度上确实不同保留两个既能降低噪声又不会把强度信息完全抹掉。最后的字符过滤会连 emoji 一起去掉这在你只做文本判别时是合理的。4.1 文本指纹识别一句话发给一万个人归一化还有一个非常有价值的副产品文本指纹。营销引流的核心特征不是文案长什么样而是同一套文案被批量发送。如果你能算出一条文本的近似指纹就能在内容判别之外再加一条极强的信号过去 10 分钟内有 N 个不同账号发了指纹相同或高度相似的内容。这里用 SimHash 实现它的好处是对小改动不敏感——改几个标点、加两个 emoji指纹的汉明距离依然很小# 文件路径app/features/fingerprint.py import hashlib from app.preprocess.normalize import normalize def _shingles(text: str, size: int 2) - list: 把归一化文本切成字符 shingle作为指纹的最小单元。 text normalize(text) if not text: return [] if len(text) size: return [text] return [text[i:i size] for i in range(len(text) - size 1)] def simhash(text: str, bits: int 64) - int: 计算 64 位 SimHash同样的输入永远得到同样的输出。 vector [0] * bits for token in _shingles(text): digest int(hashlib.md5(token.encode(utf-8)).hexdigest(), 16) for i in range(bits): vector[i] 1 if (digest i) 1 else -1 result 0 for i in range(bits): if vector[i] 0: result | 1 i return result def hamming_distance(a: int, b: int) - int: 汉明距离两个指纹有多少位不同。 return bin(a ^ b).count(1) def is_near_duplicate(a: int, b: int, threshold: int 3) - bool: 距离小于等于阈值视为近似重复。阈值 3 是常见经验值需按业务调。 return hamming_distance(a, b) threshold这十几行代码带来的收益往往比把模型从线性换成深度模型更大。原因在于它抓到的是行为层面的事实——批量发送本身就是违规行为的直接证据而不是靠语义猜测。实际落地时你可以把指纹存进 Redis 之类的缓存用SET加过期时间记录指纹 - 最近发送账号集合再做频次统计。4.2 规则层的三类信号规则层不是简单堆正则而是要按信号类别组织。我把它们分成三组身份宣称我在卖货/我是工厂/我招代理、行动号召找我/私聊/报价/拿货以及联系方式手机号、微信号、邮箱。这三组信号单独出现时都可能有正常含义但两组以上同时出现判别力就非常强了。这正是那句源头工厂 有需要就找我的骨架命中身份宣称加行动号召规则分直接到 0.7已经站在了可疑区间里。而一句请问你们是源头工厂吗只命中身份宣称得到 0.35不足以触发任何处置——误杀就这样被结构性地避免掉了。5. 完整实现规则召回 模型打分 决策融合这一章是整个流程的核心分三层写规则层负责高精度可解释的召回模型层负责泛化和兜底服务层负责把两者融合成分级处置动作。三层各司其职任何一层出问题都能单独定位、单独回滚。5.1 规则层分组打分与命中解释规则层最重要的设计不是正则写得多巧而是分组加权。分组让每条规则的贡献可解释加权让你能通过调权重来控制误杀率而不用改规则本身。# 文件路径app/rules/scorer.py import re from dataclasses import dataclass, field from app.preprocess.normalize import normalize # 组 1身份宣称——明确表达自己在供货、批发、招商 IDENTITY_PATTERNS [ r(源头|厂家|工厂|一手|批发|档口|供货商|供应链), r(直销|直供|入驻|招商|招代理|找客户|找合作|求合作), ] # 组 2行动号召——引导对方发生联系动作 CALL_TO_ACTION_PATTERNS [ r(有需要就找我|有需要找我|可以联系我|欢迎咨询|欢迎来撩), r(私聊|私信|戳我|滴滴我|找我聊), r(报价|拿货价|出厂价|一件代发|免费寄样|量大从优), ] # 组 3联系方式——必须匹配原始文本因为归一化会删掉 . 等符号 CONTACT_PATTERNS [ r(?:vx|wx|v信|威信|薇信|加v|加微|扣扣|qq)\s*[:]?\s*[a-z0-9]{5,20}, r1[3-9]\d{9}, r[a-z0-9._%-][a-z0-9.-]\.[a-z]{2,}, ] WEIGHT_IDENTITY 0.35 WEIGHT_CALL_TO_ACTION 0.35 WEIGHT_CONTACT 0.40 dataclass class RuleResult: score: float hits: list field(default_factorylist) matched: list field(default_factorylist) def rule_score(text: str) - RuleResult: 按信号组打分返回 0~1 的分数和可解释的命中明细。 raw (text or ).lower() norm normalize(text) hits, matched [], [] score 0.0 for pattern in IDENTITY_PATTERNS: if re.search(pattern, norm): score WEIGHT_IDENTITY hits.append(identity) matched.append(pattern) break for pattern in CALL_TO_ACTION_PATTERNS: if re.search(pattern, norm): score WEIGHT_CALL_TO_ACTION hits.append(call_to_action) matched.append(pattern) break # 联系方式用原始文本匹配归一化会破坏邮箱和号码的结构 for pattern in CONTACT_PATTERNS: if re.search(pattern, raw): score WEIGHT_CONTACT hits.append(contact) matched.append(pattern) break return RuleResult(scoreround(min(score, 1.0), 4), hitshits, matchedmatched)几个设计取舍要说明。第一每组只记一次命中用break跳出避免同一组里多条正则重复累加导致分数虚高。第二联系方式必须在原始文本上匹配因为归一化把、.、空格都删了邮箱会被压成一个无法识别的长串这是新手最容易踩的坑。第三matched字段保存了命中的正则原文这在排查误杀时极其有用——你能直接告诉运营这条是因为命中了哪条规则被处理的而不是一句模型说的。可解释性在生产环境里的价值往往比一两个点的准确率更重要。5.2 模型层字符 n-gram 加线性分类器模型层我选 TF-IDF 加逻辑回归理由很实际训练秒级完成、样本需求低、推理快、权重可读、几乎没有运维负担。对一个刚起步的识别系统来说这比一上来就微调预训练模型理性得多。文本特征用字符级 1 到 3 元组能覆盖中文的常见字、双字词和三字短语组合对源头工厂一手货源一件代发这类短语有天然捕捉力。# 文件路径scripts/train.py import os import joblib from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.linear_model import LogisticRegression from sklearn.pipeline import Pipeline from app.preprocess.normalize import normalize # 演示样例仅用于验证代码可以跑通不代表真实数据分布 # 真实项目中请替换为你自己标注的数据集并要求正负样本比例接近 1:1 POSITIVE_SAMPLES [ 哈喽我们是源头工厂有需要就找我呀, 工厂直销一件代发量大从优加v详聊, 你好我们是义乌档口批发拿货价便宜, 一手货源招代理欢迎咨询价格美丽, 厂家直供支持免费寄样私信我报价, 我们是供应链源头有需要可以联系我, 自家工厂出货价格比市场低欢迎来撩, 批发档口新到一批货老板们看过来, ] NEGATIVE_SAMPLES [ 请问你们是源头工厂还是经销商, 我买的东西三天了还没发货, 这个商品的售后电话是多少, 想咨询一下你们工厂直营店在哪里, 订单显示已签收但我没有收到, 能不能帮我改一下收货地址, 这个型号和上次买的有什么区别, 发票什么时候能开出来, ] def build_pipeline() - Pipeline: return Pipeline([ (tfidf, TfidfVectorizer( analyzerchar, # 中文短文本用字符级更稳 ngram_range(1, 3), # 单字、双字、三字短语 min_df1, sublinear_tfTrue, )), (clf, LogisticRegression( max_iter1000, class_weightbalanced, # 正负样本不平衡时自动加权 C1.0, )), ]) def main() - None: texts [normalize(t) for t in POSITIVE_SAMPLES NEGATIVE_SAMPLES] labels [1] * len(POSITIVE_SAMPLES) [0] * len(NEGATIVE_SAMPLES) pipeline build_pipeline() pipeline.fit(texts, labels) os.makedirs(models, exist_okTrue) joblib.dump(pipeline, models/spam_clf.joblib) print(f训练完成样本数{len(texts)}正样本{len(POSITIVE_SAMPLES)}) print(模型已保存到 models/spam_clf.joblib) # 快速自检打印几条典型输入的预测概率 for probe in [哈喽我们是源头工厂有需要找我, 请问什么时候发货]: proba pipeline.predict_proba([normalize(probe)])[0] positive proba[list(pipeline.classes_).index(1)] print(f {probe} - 正类概率 {positive:.4f}) if __name__ __main__: main()代码里有几点必须讲清楚。analyzerchar是中文短文本的常用选择如果用默认的word中文没有空格会被当成一个整词特征全废char_wb看起来更高级但它只在词边界内切 n-gram中文没有空格时同样不适用这个坑很多人踩过。class_weightbalanced在你正负样本接近 1:1 时其实没什么影响但一旦上线后正样本比例变化它能防止模型被多数类带偏。sublinear_tfTrue用对数缩放词频能压低高频字比如的我你的权重对短文本有帮助。最后那段自检逻辑也要保留。训练完不验证就打包上线是很多事故的源头。取正类概率时用list(pipeline.classes_).index(1)而不是直接写[0][1]是为了避免类别顺序变化导致的静默错误——这种错误不会报错只会让结果莫名其妙地反了。5.3 服务层把两个分数融合成分级动作服务层要把规则分和模型分融合再结合行为特征输出一个处置动作。融合权重没有标准答案我一般从模型 0.5、规则 0.5开始在标注集上按误杀率约束去调。真正的关键是阈值分级以及行为特征的加成逻辑。# 文件路径app/main.py import logging import joblib from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from app.preprocess.normalize import normalize from app.rules.scorer import rule_score from app.features.fingerprint import simhash logger logging.getLogger(pitch-detector) logging.basicConfig(levellogging.INFO) MODEL_PATH models/spam_clf.joblib BLOCK_THRESHOLD 0.75 # 大于等于则拦截 REVIEW_THRESHOLD 0.50 # 介于两者之间则折叠/人工复核 WEIGHT_MODEL 0.5 WEIGHT_RULE 0.5 app FastAPI(titleSales Pitch Detector, version1.0.0) try: model joblib.load(MODEL_PATH) except FileNotFoundError: model None logger.warning(模型文件不存在请先执行 scripts/train.py 生成模型) class CheckRequest(BaseModel): text: str Field(..., max_length2000) sender_account_age_days: int 30 sender_daily_msg_count: int 0 class CheckResponse(BaseModel): action: str final_score: float model_score: float rule_score: float hits: list fingerprint: int def _model_probability(text: str) - float: if model is None: return 0.0 proba model.predict_proba([normalize(text)])[0] return float(proba[list(model.classes_).index(1)]) def _behavior_bonus(req: CheckRequest) - float: 行为特征加成新账号、高频发送会把可疑度往上抬。 bonus 0.0 if req.sender_account_age_days 7: bonus 0.05 if req.sender_daily_msg_count 50: bonus 0.05 return bonus app.post(/v1/check, response_modelCheckResponse) def check(req: CheckRequest): if not req.text or not req.text.strip(): raise HTTPException(status_code400, detailtext 不能为空) rule rule_score(req.text) model_prob _model_probability(req.text) combined WEIGHT_MODEL * model_prob WEIGHT_RULE * rule.score final round(min(combined _behavior_bonus(req), 1.0), 4) if final BLOCK_THRESHOLD: action block elif final REVIEW_THRESHOLD: action collapse else: action pass logger.info( check action%s final%.4f model%.4f rule%.4f hits%s, action, final, model_prob, rule.score, rule.hits, ) return CheckResponse( actionaction, final_scorefinal, model_scoreround(model_prob, 4), rule_scorerule.score, hitsrule.hits, fingerprintsimhash(req.text), )这里有两处值得展开。第一是融合方式用加权和而不是纯max加权和的好处是两路信号都弱时不会误判两路都强时分数会被推高max会让单路强信号直接决定结果误杀风险更高。第二是行为加成的幅度刻意做得很小各 0.05因为行为特征的误判代价更高——新注册用户是正常用户一天发 60 条消息也可能是个活跃买家。把它们作为加分项而不是决定项是最稳妥的做法。另外注意action的命名用的是collapse而不是review因为在真实产品里落地的往往是折叠展示这个前端动作而不是进入人工队列。虽然它背后可以对应同一个区间但用产品语言命名能让前后端对齐得更快。6. 运行验证与效果评估代码写完不验证等于没写。验证分两步先确认服务能跑起来、接口返回符合预期再回到离线数据集上确认指标和错误样本。先训练再启动服务# 步骤 1训练模型生成 models/spam_clf.joblib python -m scripts.train # 步骤 2启动服务 uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload训练脚本的预期输出大致是这样正类概率的具体数值取决于你的数据不作为效果结论训练完成样本数16正样本8 模型已保存到 models/spam_clf.joblib 哈喽我们是源头工厂有需要找我 - 正类概率 0.9xxx 请问什么时候发货 - 正类概率 0.0xxx判断成功的标准是目标句的正类概率明显高于 0.5正常咨询句明显低于 0.5。如果两句都接近 0.5说明特征没起作用第一个该看的是analyzer参数有没有写成char第二个该看的是归一化是不是把整句清空了。服务起来之后用 curl 验证接口curl -X POST http://127.0.0.1:8000/v1/check \ -H Content-Type: application/json \ -d { text: 哈喽我们是源头工厂有需要就找我呀, sender_account_age_days: 3, sender_daily_msg_count: 80 }预期返回的结构是这样的{ action: block, final_score: 0.8, model_score: 0.8, rule_score: 0.7, hits: [identity, call_to_action], fingerprint: 1234567890123456789 }final_score的具体数值会随模型和数据变化但结构必须稳定hits里应该能看到identity和call_to_action两组命中action应该是block或者至少collapse。如果你把文本改成请问你们是源头工厂还是经销商hits应该只剩identityrule_score掉到 0.35action变成pass——这就是分组加权带来的误杀保护用这一对输入做回归测试非常有效。再验证一下变体处理能力把符号插进去curl -X POST http://127.0.0.1:8000/v1/check \ -H Content-Type: application/json \ -d {text: 哈 喽我 们 是 源-头-工-厂有 需 要 就 找 我 呀}归一化之后这串会被压成紧凑形式规则层照样命中。如果这条返回pass第一个排查点就是归一化函数里的分隔符正则有没有覆盖到你插入的那类符号。离线评估这一块建议单独写脚本不要混在训练里。评估集要和训练集分开否则指标永远好看上线永远打脸# 文件路径scripts/evaluate.py import joblib from sklearn.metrics import classification_report, confusion_matrix from app.preprocess.normalize import normalize from app.rules.scorer import rule_score # 独立评估集刻意包含“含关键词但正常”的样本用来暴露误杀 EVAL_TEXTS [ 哈喽我们是源头工厂有需要就找我呀, 工厂直销一件代发加v详聊, 一手货源招代理欢迎咨询, 请问你们是源头工厂还是经销商, 订单三天了还没发货怎么办, 能不能开个发票, 你们工厂直营店在哪个城市, 这个型号有现货吗, ] EVAL_LABELS [1, 1, 1, 0, 0, 0, 0, 0] BLOCK_THRESHOLD 0.75 REVIEW_THRESHOLD 0.50 def main() - None: model joblib.load(models/spam_clf.joblib) preds, notes [], [] for text in EVAL_TEXTS: rule rule_score(text) proba model.predict_proba([normalize(text)])[0] model_prob float(proba[list(model.classes_).index(1)]) final 0.5 * model_prob 0.5 * rule.score preds.append(1 if final REVIEW_THRESHOLD else 0) notes.append((text, round(final, 4), rule.hits)) print(混淆矩阵) print(confusion_matrix(EVAL_LABELS, preds)) print() print(classification_report(EVAL_LABELS, preds, zero_division0)) print(逐条明细) for text, score, hits in notes: print(f final{score} hits{hits} | {text}) if __name__ __main__: main()注意这里用的是REVIEW_THRESHOLD0.5作为二分类的分界而不是BLOCK_THRESHOLD0.75。原因是评估阶段我们关心的是有没有被识别为可疑处置等级是另一个维度的事。如果你用 0.75 做评估分界召回率会显著偏低看起来像模型不行其实是口径搞错了。这个脚本跑出来的accuracy、precision、recall、f1-score只能说明流程通了因为样本只有 8 条任何一条的变动都会让指标大幅跳动。真正做效果评估时评估集至少要到几百条量级并且要包含三类容易误判的边界样本含营销关键词的正常咨询、不含关键词的营销变体、以及一句话里既有正常诉求又有引流引导的混合样本。第三类最难建议单独统计。7. 常见问题与排查思路这套系统跑起来不难难的是跑对。下面这张表是我遇到过、以及在看别人实现时最常见的问题按现象—原因—排查—解决整理问题现象可能原因排查方式解决方案模型对所有输入都给接近 0.5 的概率analyzer用了默认的word中文被当成一个整词打印pipeline.named_steps[tfidf].get_feature_names_out()[:20]看特征改成analyzerchar、ngram_range(1,3)插入符号的变体能绕过归一化正则没覆盖到那类符号直接调用normalize()打印结果对比补充分隔符正则把该符号加进字符集邮箱、手机号规则永远命中不了在归一化后的文本上匹配联系方式打印normalize(ab.com)看输出联系方式改在原始小写文本上匹配正常咨询被大量拦截规则权重过高或阈值过低用hits字段统计拦截样本的命中分布降WEIGHT_IDENTITY或把阈值从 0.5 提到 0.75服务启动报FileNotFoundError模型文件未生成检查models/spam_clf.joblib是否存在先执行python -m scripts.train预测结果整体反了取正类概率时写死索引打印model.classes_看顺序用list(model.classes_).index(1)取值训练时警告类别样本过少正负样本严重不平衡打印两类样本数量补样本或设class_weightbalanced同一条文本两次请求结果不同模型被重新加载或混用了不同版本检查部署实例加载的模型文件时间戳模型文件加版本号服务启动时打日志接口返回 422请求体字段类型或长度不符合CheckRequest定义看 FastAPI 返回的detail字段按 schema 传参text控制在 2000 字符内线上 CPU 飙高单条请求重复加载模型或反复编译正则看火焰图确认热点模型在模块级加载一次正则预编译表格里有两行值得单独强调。第一行概率都是 0.5是最高频的问题九成以上是analyzer参数没设对因为中文分词缺失会让整个文本变成一个特征模型学不到任何东西。第六行结果整体反了是最隐蔽的问题它不报错只是逻辑反了往往要等到运营反馈最近拦截的都是正常用户才会被发现所以取概率的写法一定要用index(1)这种防御性写法而不是依赖默认顺序。还有一类问题不在表里但更麻烦词库和规则长期无人维护导致系统逐渐失效。表现是拦截量慢慢下降、误杀慢慢上升但因为变化是渐进的没有报警就没人发现。解决办法是给系统加监控指标而不只是加规则——每天统计拦截量、折叠量、申诉量、申诉通过率一旦申诉通过率超过某个阈值就说明误杀在恶化需要立刻回看规则权重。8. 最佳实践与工程建议这一段是给准备把这套东西上生产的团队看的。前面九成的代码你都可以直接抄但下面这些经验抄不走需要你自己在项目里落地。第一规则和模型都要版本化。规则文件、模型文件、阈值配置三者必须能对应到一个版本号并且在服务启动日志里打出来。原因很现实出问题时你需要在十分钟内回答昨天和今天有什么不同。如果规则是热更新的、模型是滚动覆盖的、阈值是改在环境变量里的你就无法回溯。更稳妥的做法是把规则集也做成配置文件YAML 或 JSON走正常的发布流程而不是让运营在后台直接改。第二必须给用户留申诉通道。任何自动处置系统都会有误杀区别只在于你有没有让用户表达出来的路径。具体做法是折叠或拦截时给一个内容被误判点击申诉的入口申诉样本自动回流到标注队列。这个通道不只是为了用户体验它是你获取高质量负样本的主要来源——真实误杀样本比任何人工构造的负样本都值钱。第三分级处置要克制。前面已经强调过0.5 到 0.75 这个不确定区间应该交给折叠而不是拦截。此外同一个账号即使被判为营销也不建议第一次就永久限制更稳妥的策略是阶梯式首次折叠、再次限流、三次以上才拦截。这样即使模型误判代价也被控制在一屏之内。第四注意数据合规边界。做这类识别时你只需要文本内容、账号基础属性注册时间、历史发送量、以及必要的频次统计。不要顺手把聊天的完整会话、收货地址、支付信息拉进特征也不要在日志里打印完整原文。日志里打的应该是脱敏后的片段、指纹、以及命中的规则 ID。这条不只是合规要求也是安全底线——日志泄露的代价远高于识别效果提升带来的收益。第五冷启动阶段别追求模型效果。上线第一天没有标注数据是常态。这个阶段的正确做法是规则先行 全量抽样人工审核用审核结果积累第一批标注同时把处置动作全部设成折叠这种低风险动作。等你积累了几百条正负样本再训练模型、再考虑提高处置等级。反过来做——先上模型、先拦截、后补数据——几乎一定会出事故。第六把效果监控当成功能来做。至少要有四个指标每日识别量、处置分布pass/collapse/block 各占多少、申诉量与申诉通过率、以及抽样准确率。前三个是实时指标最后一个靠每周抽 200 条人工复核。指标异常时的处理顺序是先看处置分布有没有突变再看申诉样本集中在哪类文本最后才去看模型和规则。这个顺序能让你在十分钟内定位到问题层级。第七阈值要放进配置中心而不是代码里。阈值是所有参数里最需要频繁调整的一个而且调整往往发生在出问题的当下。如果它写在代码里每次调都要走一次发布流程等你发完舆情已经过去了。把它放到配置中心或环境变量配合灰度发布和快速回滚是这类系统的基本要求。9. 从能跑到好用下一步的迭代路径把上面这套跑通之后你已经有了一个结构正确、可解释、可回滚的识别系统。它现在的效果大概只能算能用但结构是对的这意味着后面每一步投入都能叠加而不是推倒重来。接下来最值得投入的不是模型是标注数据的质量。优先补三类样本一类是含营销词但正常的负样本它们直接决定你的误杀率一类是零关键词但明显引流的正样本它们决定模型的泛化上限还有一类是混合样本——前半句是正常咨询、后半句突然引导私聊。这三类样本凑齐之后你会发现自己能很清楚地看到模型的上限在哪里。数据够了之后再考虑模型升级。判断标准很明确如果在线抽样复核发现的漏放案例中有超过一半是语义上能理解但模型没学到的类型那就该换更强的模型了如果漏放主要集中在变体绕过那说明问题在归一化不在模型。这两种情况用错药投入会全部打水漂。最后提醒一句关于范围的事。这类系统的目标是减少骚扰、保护正常用户的沟通体验不是把商家都赶走。它会误判所以必须留申诉它会漏放所以必须有监控它会被对抗所以必须能快速迭代。把这三点当成系统的一部分而不是后续优化你在设计阶段就会做出不同的选择——比如给每条处置都留一个可解释的命中理由比如把阈值放进配置中心比如从一开始就给规则和模型打版本号。想继续深入的话有几个方向比较有价值文本归一化的对抗演进如何用少量样本发现新的绕过手法、近似重复检测的工程实现大规模指纹索引怎么做、以及分级处置策略的 A/B 实验设计怎么科学地判断提高拦截率到底有没有伤害转化。这三块决定了这个系统从能跑到好用的距离。建议把本文的目录结构和三个核心脚本先落到自己的项目里跑一遍尤其是归一化和规则打分这两部分它们不依赖任何模型当天就能上线看到效果。