做社区、电商、IM 私信、商品问答这类业务的开发者，大概率都在后台见过这么一句话："哈喽，我们是源头工厂，有需要就找我呀"。它不算脏话，没有敏感词，也没有攻击性，甚至礼貌得有点过分，可它就是会把评论区、私信列表、工单池刷成一堵墙。更麻烦的是，它今天叫"源头工厂"，明天叫"一手货源"，后天变成"厂家直发，私信报价"，再过两天开始夹 emoji 间隔、全角符号、拼音首字母和拆字写法。

很多团队的第一版方案都是拉一个关键词黑名单：把"工厂""批发""拿货价""加微信"塞进去，命中就拦。上线第一天效果不错，第三天开始误杀正常用户的提问（比如"你们这家店是工厂直营吗"），第七天开始漏放（对方把字拆开、中间插符号），第三十天词库变成几百条没人敢删的祖传配置。这不是词库不够大的问题，是方案结构的问题。

本文的判断很明确：这类"招商引流话术"的识别，靠单点关键词匹配已经没有出路，需要把它当成一个分层的文本风控系统来做——文本归一化兜住变体、规则层负责高精度召回、模型层负责泛化打分、行为层负责把内容信号放大成处置动作，最后用分级处置代替"一刀切拦截"。这套结构的价值不在于模型多先进，而在于它能被你团队持续维护、能解释、能回滚。

读完这篇，你可以拿到一套可以跑通的最小实现：从环境准备、目录结构、归一化脚本、规则打分器、TF-IDF 加逻辑回归的训练脚本，到一个 FastAPI 推理服务，再到离线评估和线上灰度的验证方法。中间会讲清楚每个环节为什么这么设计、做错了会出现什么现象，以及在生产环境里真正容易翻车的地方在哪里。全文代码都完整可复制，你可以直接拿去改。

1. 一句"源头工厂"，为什么会把内容识别系统逼到墙角

先把这句话拆开看。它其实是三段式结构：问候语（"哈喽"）+ 身份宣称（"我们是源头工厂"）+ 行动号召（"有需要就找我呀"）。这种结构在营销话术里极其稳定，变的是词，不变的是骨架。你今天封了"源头工厂"，明天对方换成"我们是义乌档口""自家工厂直发""一手货源招代理"，骨架一模一样，你的黑名单一条都命中不了。

第二个难点是它天然长在灰区。同样一句"我们是源头工厂"，发在商家的店铺公告里，是完全正常的自我介绍；发在一个从没买过东西的新账号私信里，就是引流；发在一个正常用户的咨询里（"请问你们是源头工厂还是经销商"），甚至是平台希望鼓励的信息透明度。所以问题不在于"这句话是不是垃圾"，而在于"这句话在什么上下文里、由谁、以什么频率、发给谁"。脱离上下文的纯文本分类，天花板很低。

第三个难点是对抗成本极低。符号插入、全角半角混用、emoji 分隔、重复字符、繁简混写、拼音首字母、数字谐音，任何一种做法都只要改一次文案，而你的词库要跟着改一次。这种不对称意味着：如果你把所有希望寄托在人工维护词库上，这场仗从结构上就是输的。真正能压住成本的，是"归一化把变体拉回同一个平面"这一步，它让后面的规则和模型只面对一种形态。

所以这一章的小结论是：你面对的不是"某个词"，而是"一类意图 + 一类行为模式"。识别系统的设计目标应该是——用最低的维护成本，稳定识别出"身份宣称 + 引导联系 + 批量发送"这三个信号的组合，而不是去追无穷无尽的词表。想清楚这一点，后面的技术选型才有意义。

2. 话术识别到底在识别什么：概念、指标与技术选型

从工程视角看，这本质上是一个文本二分类任务：给定一条文本（还可以带上发送者、接收者、时间、频率等上下文），判断它是不是"营销引流类内容"。但直接上一个大模型做二分类并不是最优起点，因为你要考虑三件事：推理成本、可解释性、以及误杀的代价。误杀一条正常用户的咨询，可能直接损失一单生意，这个代价比漏放几条广告高得多。

先统一几个后面会反复出现的术语。召回指的是"该抓的有没有抓到"，对应指标是召回率（Recall）；精确率（Precision）指的是"抓到的里面有多少是对的"，防止误杀；两者此消彼长，用一个综合指标F1平衡。混淆矩阵就是把预测结果和真实标签交叉成四个格子，能一眼看出你是漏放多还是误杀多。阈值是决策的分界线，模型输出一个 0 到 1 的分数，超过某个值才判定为正类，调阈值就是调精确率和召回率的平衡点。

技术选型上，不同方案的成本和收益差别很大，下面这张表是我在实际项目里会拿来和团队对齐的判断依据：

更稳妥的判断是：先做归一化 + 规则，再上 TF-IDF 线性模型做基线，把"分级处置"和"行为特征"这两层做扎实，最后才考虑要不要上预训练模型。原因很实际——在绝大多数业务里，真正把效果拖垮的不是模型不够强，而是变体没处理、误杀没申诉通道、阈值没人调、上线没有灰度。这些问题换什么模型都解决不了。

这里还要强调一个容易被忽略的概念：处置动作不等于拦截。行业里比较成熟的做法是分级——放行、折叠（默认收起、点击展开）、限流（限制发送频次）、拦截（直接不发出）、人工复核。分级的意义在于，你把 0.5 到 0.75 这个不确定区间交给"折叠"而不是"拦截"，用户的正常咨询只是被收起了一下，而营销内容已经失去了曝光收益。这一步能显著降低误杀的负面影响。

3. 环境准备与工程目录结构

这套东西对硬件要求很低，一台普通开发机足够跑通全流程。语言用 Python，版本建议 3.9 及以上（3.10、3.11 都可以），再低的版本在类型标注和部分依赖上会有兼容问题，不建议。依赖管理用 pip 加 requirements.txt 就够，如果你团队已经在用 poetry 或 conda，按自己的规范来即可，不影响本文的代码逻辑。

核心依赖只有四个方向：文本处理和向量化用 scikit-learn，模型持久化用 joblib，服务化用 FastAPI 加 uvicorn（uvicorn 是 ASGI 服务器，负责真正把 HTTP 请求接进来）。分词这一块我特意没有引入 jieba，因为中文短文本用字符级 n-gram往往比词级更稳——营销话术里造词、拆字、生造缩写的比例太高，词表覆盖不住，而字符 n-gram 不需要任何词典，天然能吃到"源头工厂""一手货源"这类组合模式。

先看依赖文件：

# 文件路径：requirements.txt fastapi>=0.100 uvicorn[standard]>=0.23 scikit-learn>=1.2 joblib>=1.2 pydantic>=1.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}", r"1[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_factory=list) matched: list = field(default_factory=list) 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(score=round(min(score, 1.0), 4), hits=hits, matched=matched)

几个设计取舍要说明。第一，每组只记一次命中，用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( analyzer="char", # 中文短文本用字符级更稳 ngram_range=(1, 3), # 单字、双字、三字短语 min_df=1, sublinear_tf=True, )), ("clf", LogisticRegression( max_iter=1000, class_weight="balanced", # 正负样本不平衡时自动加权 C=1.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_ok=True) 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()

代码里有几点必须讲清楚。analyzer="char"是中文短文本的常用选择，如果用默认的"word"，中文没有空格会被当成一个整词，特征全废；char_wb看起来更"高级"，但它只在词边界内切 n-gram，中文没有空格时同样不适用，这个坑很多人踩过。class_weight="balanced"在你正负样本接近 1:1 时其实没什么影响，但一旦上线后正样本比例变化，它能防止模型被多数类带偏。sublinear_tf=True用对数缩放词频，能压低高频字（比如"的""我""你"）的权重，对短文本有帮助。

最后那段自检逻辑也要保留。训练完不验证就打包上线，是很多事故的源头。取正类概率时用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(level=logging.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(title="Sales Pitch Detector", version="1.0.0") try: model = joblib.load(MODEL_PATH) except FileNotFoundError: model = None logger.warning("模型文件不存在，请先执行 scripts/train.py 生成模型") class CheckRequest(BaseModel): text: str = Field(..., max_length=2000) 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_model=CheckResponse) def check(req: CheckRequest): if not req.text or not req.text.strip(): raise HTTPException(status_code=400, detail="text 不能为空") 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( action=action, final_score=final, model_score=round(model_prob, 4), rule_score=rule.score, hits=rule.hits, fingerprint=simhash(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应该只剩identity，rule_score掉到 0.35，action变成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_division=0)) print("逐条明细：") for text, score, hits in notes: print(f" final={score} hits={hits} | {text}") if __name__ == "__main__": main()

注意这里用的是REVIEW_THRESHOLD（0.5）作为二分类的分界，而不是BLOCK_THRESHOLD（0.75）。原因是评估阶段我们关心的是"有没有被识别为可疑"，处置等级是另一个维度的事。如果你用 0.75 做评估分界，召回率会显著偏低，看起来像模型不行，其实是口径搞错了。

这个脚本跑出来的accuracy、precision、recall、f1-score只能说明流程通了，因为样本只有 8 条，任何一条的变动都会让指标大幅跳动。真正做效果评估时，评估集至少要到几百条量级，并且要包含三类容易误判的边界样本：含营销关键词的正常咨询、不含关键词的营销变体、以及一句话里既有正常诉求又有引流引导的混合样本。第三类最难，建议单独统计。

7. 常见问题与排查思路

这套系统跑起来不难，难的是跑对。下面这张表是我遇到过、以及在看别人实现时最常见的问题，按"现象—原因—排查—解决"整理：

表格里有两行值得单独强调。第一行"概率都是 0.5"是最高频的问题，九成以上是analyzer参数没设对，因为中文分词缺失会让整个文本变成一个特征，模型学不到任何东西。第六行"结果整体反了"是最隐蔽的问题，它不报错，只是逻辑反了，往往要等到运营反馈"最近拦截的都是正常用户"才会被发现，所以取概率的写法一定要用index(1)这种防御性写法，而不是依赖默认顺序。

还有一类问题不在表里，但更麻烦：词库和规则长期无人维护，导致系统逐渐失效。表现是拦截量慢慢下降、误杀慢慢上升，但因为变化是渐进的，没有报警就没人发现。解决办法是给系统加监控指标，而不只是加规则——每天统计拦截量、折叠量、申诉量、申诉通过率，一旦申诉通过率超过某个阈值，就说明误杀在恶化，需要立刻回看规则权重。

8. 最佳实践与工程建议

这一段是给准备把这套东西上生产的团队看的。前面九成的代码你都可以直接抄，但下面这些经验抄不走，需要你自己在项目里落地。

第一，规则和模型都要版本化。规则文件、模型文件、阈值配置三者必须能对应到一个版本号，并且在服务启动日志里打出来。原因很现实：出问题时你需要在十分钟内回答"昨天和今天有什么不同"。如果规则是热更新的、模型是滚动覆盖的、阈值是改在环境变量里的，你就无法回溯。更稳妥的做法是把规则集也做成配置文件（YAML 或 JSON），走正常的发布流程，而不是让运营在后台直接改。

第二，必须给用户留申诉通道。任何自动处置系统都会有误杀，区别只在于你有没有让用户表达出来的路径。具体做法是：折叠或拦截时给一个"内容被误判？点击申诉"的入口，申诉样本自动回流到标注队列。这个通道不只是为了用户体验，它是你获取高质量负样本的主要来源——真实误杀样本比任何人工构造的负样本都值钱。

第三，分级处置要克制。前面已经强调过，0.5 到 0.75 这个不确定区间应该交给"折叠"而不是"拦截"。此外，同一个账号即使被判为营销，也不建议第一次就永久限制，更稳妥的策略是阶梯式：首次折叠、再次限流、三次以上才拦截。这样即使模型误判，代价也被控制在一屏之内。

第四，注意数据合规边界。做这类识别时，你只需要文本内容、账号基础属性（注册时间、历史发送量）、以及必要的频次统计。不要顺手把聊天的完整会话、收货地址、支付信息拉进特征，也不要在日志里打印完整原文。日志里打的应该是脱敏后的片段、指纹、以及命中的规则 ID。这条不只是合规要求，也是安全底线——日志泄露的代价远高于识别效果提升带来的收益。

第五，冷启动阶段别追求模型效果。上线第一天没有标注数据是常态。这个阶段的正确做法是：规则先行 + 全量抽样人工审核，用审核结果积累第一批标注，同时把处置动作全部设成"折叠"这种低风险动作。等你积累了几百条正负样本，再训练模型、再考虑提高处置等级。反过来做——先上模型、先拦截、后补数据——几乎一定会出事故。

第六，把"效果监控"当成功能来做。至少要有四个指标：每日识别量、处置分布（pass/collapse/block 各占多少）、申诉量与申诉通过率、以及抽样准确率。前三个是实时指标，最后一个靠每周抽 200 条人工复核。指标异常时的处理顺序是：先看处置分布有没有突变，再看申诉样本集中在哪类文本，最后才去看模型和规则。这个顺序能让你在十分钟内定位到问题层级。

第七，阈值要放进配置中心而不是代码里。阈值是所有参数里最需要频繁调整的一个，而且调整往往发生在出问题的当下。如果它写在代码里，每次调都要走一次发布流程，等你发完，舆情已经过去了。把它放到配置中心或环境变量，配合灰度发布和快速回滚，是这类系统的基本要求。

9. 从能跑到好用：下一步的迭代路径

把上面这套跑通之后，你已经有了一个结构正确、可解释、可回滚的识别系统。它现在的效果大概只能算"能用"，但结构是对的，这意味着后面每一步投入都能叠加，而不是推倒重来。

接下来最值得投入的不是模型，是标注数据的质量。优先补三类样本：一类是"含营销词但正常"的负样本，它们直接决定你的误杀率；一类是"零关键词但明显引流"的正样本，它们决定模型的泛化上限；还有一类是混合样本——前半句是正常咨询、后半句突然引导私聊。这三类样本凑齐之后，你会发现自己能很清楚地看到模型的上限在哪里。

数据够了之后，再考虑模型升级。判断标准很明确：如果在线抽样复核发现的漏放案例中，有超过一半是"语义上能理解但模型没学到"的类型，那就该换更强的模型了；如果漏放主要集中在变体绕过，那说明问题在归一化，不在模型。这两种情况用错药，投入会全部打水漂。

最后提醒一句关于范围的事。这类系统的目标是减少骚扰、保护正常用户的沟通体验，不是把商家都赶走。它会误判，所以必须留申诉；它会漏放，所以必须有监控；它会被对抗，所以必须能快速迭代。把这三点当成系统的一部分而不是"后续优化"，你在设计阶段就会做出不同的选择——比如给每条处置都留一个可解释的命中理由，比如把阈值放进配置中心，比如从一开始就给规则和模型打版本号。

想继续深入的话，有几个方向比较有价值：文本归一化的对抗演进（如何用少量样本发现新的绕过手法）、近似重复检测的工程实现（大规模指纹索引怎么做）、以及分级处置策略的 A/B 实验设计（怎么科学地判断"提高拦截率"到底有没有伤害转化）。这三块决定了这个系统从"能跑"到"好用"的距离。

建议把本文的目录结构和三个核心脚本先落到自己的项目里跑一遍，尤其是归一化和规则打分这两部分，它们不依赖任何模型，当天就能上线看到效果。