在公共采购文本中,检测指控性语言(accusatory language)是一类典型的 NLP 管道任务。投诉信、供应商质疑函、审计意见和合同履约记录里,经常出现“评标参数存在倾向性”“材料涉嫌虚假”“资质条件疑似为特定品牌量身定制”等表达。这些句子不一定像新闻标题那样直接给出结论,而是藏在否定、假设、引用和长尾名词中。如果只用关键词匹配,很容易漏掉委婉说法,也会把“供应商否认串通行为”这类辩解句误判成指控。如果直接训练监督模型,又需要大量人工标注,而公共采购语料通常领域性强、样本偏少、标注成本高。
这篇文章围绕一个可复现的方案展开:构建级联的无监督-监督 NLP 管道,第一次先用规则、业务词表、零样本分类等无监督或弱监督信号生成弱标签,第二次再用这些弱标签训练监督分类器,最后输出指控风险等级。文章会给出完整代码骨架、弱标签生成策略、监督训练流程、评估方法、常见错误排查和生产化建议。适合正在做 NLP 文本分类、审计风险预筛、投诉自动分拣或合规系统的工程师阅读。
文中不会使用真实公开资料,示例文本只用于说明代码结构和推理过程。真实项目落地时,需要根据业务语料重新设计规则、验证模型并做脱敏处理。
1. 理解问题:为什么检测指控语言需要级联管道
1.1 指控性语言在采购场景中的表现
指控性语言并不是普通负面情感,它通常包含三个要素:
- 被质疑对象:供应商、评标委员会、招标代理等。
- 被质疑行为:评分不公、参数排他、材料作假、串通投标等。
- 负面评价倾向:明显、涉嫌、违规、不合理等。
一个典型句子可能是:“该招标文件对供货时间的设置不合理,疑似为现有供应商争取缓冲期。”这句话没有出现“投诉”“举报”等强信号词,但语义上已经构成对招标公平性的质疑。这种文本如果只靠词典命中,会落在召回盲区。
从 NLP 任务角度,它属于主观性文本检测,和情感分析、意图识别有关,但比普通情感分析更复杂。因为指控性语言经常是客观语气包裹主观质疑,例如“投标人未按招标文件要求提供业绩原件”这句话本身是在陈述事实,但放在投诉上下文里可能是在暗示资质造假。所以,模型需要同时理解词汇、句式和业务背景。
1.2 无监督和监督模型各自的边界
纯规则或关键词方案的问题很直观:
- 召回不完整,无法覆盖同义表达。
- 否定句处理困难。
- 无法感知上下文,无法判断投诉方和被投诉方。
纯监督模型的问题同样明显:
- 需要大量标注数据。
- 通用语料训练出来的模型在采购领域效果不稳定。
- 人工标注需要熟悉采购法规和文本背景,成本高。
级联管道的基本思路是:先用低成本的规则和弱监督信号生成一批可信度较高的标签,再用监督模型学习这些标签背后的语义模式,最后通过人工抽样评估和反馈修正管道。这样可以减少从零标注的负担,同时保留监督模型对上下文建模的能力。
1.3 级联管道的三段式结构
整个管道可以抽象为下面这条链路:
输入文本 -> 第一级:无监督弱标签生成 - 指控词触发 - 业务上下文过滤 - 否定词窗口过滤 - 零样本分类补充 -> 第二级:监督分类器 - TF-IDF / 嵌入表示 - 逻辑回归 / 预训练语言模型 -> 后处理 - 概率阈值 - 高风险 / 待复核 / 低风险决策 - 人工反馈回流第一级处理“哪些样本大概率是正例、哪些大概率是负例”,第二级处理“将弱标签泛化到更多表达”。两者级联后,既保留了规则的可解释性,又获得了模型对语义变化的适应能力。
2. 准备环境与语料结构
2.1 依赖安装与版本确认
下面的示例使用 Python 开发,核心依赖包括 pandas、scikit-learn,以及可选的 transformers。拟一个虚拟环境:
python -m venv .venv source .venv/bin/activate pip install --upgrade pip pip install pandas scikit-learn # 如果使用零样本分类或 BERT,还需要安装 transformers 和 torch pip install transformers torch如果原始环境没有明确版本,落地前要先确认以下依赖是否兼容:
| 依赖 | 作用 | 建议 |
|---|---|---|
| Python | 基础运行环境 | 3.9 及以上 |
| pandas | 数据处理 | 1.5 及以上 |
| scikit-learn | 特征和分类模型 | 1.2 及以上 |
| transformers | 零样本分类和预训练模型 | 按官方文档确认 |
| torch | 深度学习后端 | 与 transformers 版本匹配 |
这里再强调一句:示例代码里使用的模型名称仅表示思路,实际选择需要根据网络、显存、许可证和语种确定。
2.2 语料目录设计
真实项目中,建议把原始语料、弱标签输出、人工评估集和日志分开存放:
procurement_accord/ ├── data/ │ ├── raw_docs.json │ ├── weak_labels.json │ └── manual_eval.json ├── rules/ │ └── accusation_terms.py ├── pipeline/ │ ├── weak_label.py │ ├── classifier_model.py │ └── service.py ├── models/ │ └── saved_model.bin └── logs/ └── pipeline.log目录设计虽然简单,但对后续排查很有帮助。尤其是弱标签文件需要保留“哪个规则命中”“命中位置”“是否经过否定过滤”,否则模型有问题时很难回溯。
2.3 最小演示语料
先准备几条模拟文本,用于跑通管道。字段包含 id、来源和正文,如下:
[ { "id": "doc_001", "source": "complaint", "text": "评标委员会在打分时对某供应商的技术参数存在明显倾向性,评分规则不一致。" }, { "id": "doc_002", "source": "contract", "text": "项目验收已完成,付款申请按合同约定提交。" }, { "id": "doc_003", "source": "complaint", "text": "投标人提供的业绩材料涉嫌造假,且未按招标文件要求提供原件。" }, { "id": "doc_004", "source": "notice", "text": "中标通知书已发出,双方正在进行合同细节沟通。" }, { "id": "doc_005", "source": "complaint", "text": "招标文件中的资质要求被指为特定品牌量身定制,排斥潜在投标人。" } ]实际项目里,文本可能来自 PDF、Word、OCR 结果,需要先做格式抽取、去重和脱敏。演示数据不承担真实业务效果,只用来验证管道的代码逻辑。
3. 第一阶段:用规则和零样本模型生成弱标签
3.1 规则库为什么需要两层过滤
规则不能简单写成“包含指控词就算正例”。比如“供应商否认串通行为”包含“串通”,但它在语义上是辩解,不是指控。所以规则需要包含两层过滤:
- 指控行为词:负责触发候选,例如“质疑”“涉嫌”“串通”“倾向性”“排他性”。
- 业务上下文词:负责把候选限定在采购场景内,例如“招标”“投标”“供应商”“评标”“资质”。
业务上下文的作用是减少误报。如果一篇文本只有“质疑”两个字,没有采购相关词,那它可能是在讨论其他事件,不应直接进入采购指控分类。
下表是规则库中常见词表:
| 类型 | 示例词 | 作用 |
|---|---|---|
| 指控行为词 | 质疑、投诉、涉嫌、串通、围标、倾向性、排他性、量身定制、虚假、伪造 | 触发候选正例 |
| 业务上下文词 | 招标、投标、供应商、中标、评标、采购、合同、资质、技术参数、报价 | 限定业务范围 |
| 否定词 | 不、未、没有、否认、不存在、非、无 | 过滤否定或辩解句 |
3.2 实现规则匹配函数
下面用一个简单函数说明弱标签生成逻辑。为了可读,这里只做窗口级否定检查,不做完整依存句法分析:
import re ACCUSATION_PATTERNS = [ "质疑", "投诉", "涉嫌", "围标", "串标", "串通", "倾向性", "排他性", "量身定制", "虚假", "伪造", "不合理", "违规", "暗箱操作", ] BUSINESS_TERMS = [ "招标", "投标", "供应商", "中标", "评标", "采购", "合同", "资质", "技术参数", "报价", ] NEGATIONS = ["不", "未", "没有", "否认", "不存在", "非", "无"] def has_negation_in_window(text: str, start: int, end: int, window: int = 30) -> bool: left = max(0, start - window) right = min(len(text), end + window) segment = text[left:right] return any(neg in segment for neg in NEGATIONS) def weak_label(text: str) -> dict: matched_signals = [] for pattern in ACCUSATION_PATTERNS: for match in re.finditer(pattern, text): if has_negation_in_window(text, match.start(), match.end()): continue left = max(0, match.start() - 30) right = min(len(text), match.end() + 30) context = text[left:right] if any(re.search(term, context) for term in BUSINESS_TERMS): matched_signals.append( { "rule": pattern, "start": match.start(), "end": match.end(), } ) has_business = any(re.search(term, text) for term in BUSINESS_TERMS) if matched_signals: return {"label": 1, "relation": "rule_hit", "signals": matched_signals} if has_business: return {"label": 0, "relation": "business_only", "signals": []} return {"label": -1, "relation": "no_signal", "signals": []}这段代码有三个关键点:
- 在命中指控词后,取左右各 30 个字符作为上下文,检查是否存在否定词。
- 即使指控词命中,如果上下文缺少业务词,也不会标记为正例。
- 如果文本包含业务词但没有命中任何指控规则,会标记为
0,用于构造负样本。
需要说明,窗口级否定检查并不完美。比如“供应商并未否认串通行为”这种双重否定句式会被误判成负例,因为上下文里出现了“否认”。生产环境建议用更精细的句法分析或短文本分类模型处理否定作用域。
3.3 用零样本分类补充召回盲区
规则能覆盖显式信号,但无法覆盖“该条款对潜在投标人不公平”这种没有强信号词的表达。零样本分类可以在没有业务标注的情况下,判断文本是否属于指控类别。
示例代码如下:
from transformers import pipeline zero_shot = pipeline( "zero-shot-classification", model="typeform/distilbert-base-uncased-mnli", device=-1, # CPU ) CANDIDATE_LABELS = [ "accusatory language about procurement violation", "neutral procurement statement", ] def zero_shot_label(text: str, threshold: float = 0.5) -> int: result = zero_shot(text, CANDIDATE_LABELS) top_label = result["labels"][0] top_score = result["scores"][0] if top_label.startswith("accusatory") and top_score >= threshold: return 1 return -1这里要澄清一个概念:零样本分类模型本身是在大规模语料上训练过的,严格来说不是“无监督”,但在当前采购业务领域,它没有使用业务人工标签,所以可以放在弱监督阶段使用。中文场景请选择中文 NLI 模型,或者先做语义映射,不能直接套用英文模型名称。
零样本分类的缺点是速度慢、结果不稳定。建议只在规则未命中且文本包含业务词时调用,避免每条输入都走大模型。
3.4 弱标签合并策略和人工复核
弱标签合并可以按下面的优先级:
| 信号 | 合并结果 | 说明 |
|---|---|---|
| 规则命中且通过否定过滤 | 1 | 高置信正例 |
| 业务词命中但规则未命中,零样本判定为指控 | 1 | 扩大召回 |
| 业务词命中但规则未命中,零样本判定为非指控 | 0 | 构造负样本 |
| 既无业务词也无规则信号 | -1 | 不参与训练 |
| 规则正例和零样本负例冲突 | 以规则正例为准 | 规则更贴近业务词表,可在人工复核中修正 |
合并函数可以写成:
def merge_weak_labels(rule_result: dict, zs_label: int) -> int: if rule_result["label"] == 1 or zs_label == 1: return 1 if rule_result["label"] == 0: return 0 return -1生成弱标签后,不要直接拿去训练就结束。建议每类随机抽样 20 到 50 条,由熟悉采购业务的人员复核。这个动作能在早期发现规则遗漏和误报,也能为后续模型评估提供人工基线。
4. 第二阶段:训练监督分类器
4.1 特征表示与模型选择
弱标签让监督模型有了训练集,但模型需要把文本转成向量。常见选择有三种:
| 方案 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| TF-IDF + 逻辑回归 | 训练快、可解释性好 | 难以处理复杂语义 | 快速基线、规则上线验证 |
| Sentence Transformer 嵌入 | 语义能力强,训练成本低 | 需要下载预训练模型 | 中等规模数据 |
| 微调 BERT 类模型 | 效果最好 | 需要 GPU、训练时间长 | 数据量较大且长期迭代 |
作为最小示例,这里使用 TF-IDF 加逻辑回归。它的训练速度快,CPU 就能跑通,也方便输出特征权重,帮助理解模型学会了哪些词。
4.2 训练完整代码
先加载第一阶段生成的弱标签数据,然后划分训练集和验证集:
import pandas as pd from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.linear_model import LogisticRegression from sklearn.model_selection import train_test_split from sklearn.metrics import classification_report from sklearn.pipeline import make_pipeline # 假设 docs 是原始语料,这里对每条文本调用弱标签函数 records = [] for doc in docs: result = weak_label(doc["text"]) if result["label"] in (0, 1): records.append( { "id": doc["id"], "text": doc["text"], "label": result["label"], "relation": result["relation"], } ) df = pd.DataFrame(records) X_train, X_test, y_train, y_test = train_test_split( df["text"], df["label"], test_size=0.3, random_state=42, stratify=df["label"], ) model = make_pipeline( TfidfVectorizer(ngram_range=(1, 2), min_df=1), LogisticRegression(max_iter=1000, class_weight="balanced"), ) model.fit(X_train, y_train) print(classification_report(y_test, model.predict(X_test), target_names=["negative", "positive"]))关于参数,这里要解释几处:
ngram_range=(1, 2)表示同时使用单个词和相邻双词作为特征,能捕捉“明显倾向性”“涉嫌造假”这类短语。class_weight="balanced"让模型自动调整正负样本权重,避免因负样本过多导致正例被忽略。min_df=1表示词至少在一条样本中出现才保留。正式项目要根据语料量调整,数据多时可以改成 2 或 3,过滤低频噪声。
4.3 为什么还需要人工评估集
弱标签是从规则和零样本生成的,本身带有噪声。直接看弱标签验证集上的 F1 并不能代表真实效果。最好再准备一个小型人工评估集,由人工稳定标注,不参与规则生成。
manual_eval = [ {"text": "该供应商的报价明显低于成本价,可能影响后续履约。", "label": 1}, {"text": "项目进度正常,未发现异常情况。", "label": 0}, {"text": "技术参数评分没有给出具体明细,存在解释空间。", "label": 1}, ] manual_texts = [item["text"] for item in manual_eval] manual_labels = [item["label"] for item in manual_eval] print(classification_report(manual_labels, model.predict(manual_texts), target_names=["negative", "positive"]))上面只有 3 条数据,只能演示调用方式。真实项目的人工评估集建议至少 100 到 300 条,且来源要与训练集不同,否则评估结果会有偏差。
5. 把弱标签和监督模型整合成一条可调用管道
5.1 管道类设计
为了让管道可复用,可以封装成一个类,统一处理缓存、弱标签、模型预测和决策阈值:
import hashlib import json class AccusationPipeline: def __init__(self, classifier, zero_shot=None, threshold_review=0.5, threshold_high=0.8): self.classifier = classifier self.zero_shot = zero_shot self.threshold_review = threshold_review self.threshold_high = threshold_high self.cache = {} def _cache_key(self, text: str) -> str: normalized = " ".join(text.split()) return hashlib.sha1(normalized.encode("utf-8")).hexdigest() def detect(self, doc: dict) -> dict: text = doc["text"] cache_key = self._cache_key(text) if cache_key in self.cache: return self.cache[cache_key] rule_result = weak_label(text) zs_label = -1 if self.zero_shot is not None and rule_result["label"] != 1: zs_label = zero_shot_label(text) merged_label = merge_weak_labels(rule_result, zs_label) if merged_label in (0, 1): proba = self.classifier.predict_proba([text])[0] prediction = int(proba[1] >= self.threshold_review) else: proba = [0.0, 0.0] prediction = -1 positive_prob = round(float(proba[1]), 4) result = { "id": doc["id"], "weak_label": merged_label, "weak_relation": rule_result["relation"], "prediction": prediction, "probability": positive_prob, "decision": self._decide(positive_prob), } self.cache[cache_key] = result return result def _decide(self, positive_prob: float) -> str: if positive_prob >= self.threshold_high: return "high_risk" if positive_prob >= self.threshold_review: return "review" return "low_risk"这里要注意:缓存键做了一次空白归一化。否则同一句话因为换行或多余空格产生不同哈希值,缓存命中率会下降。更严格的做法还可以做全角转半角、小写化、去除标点,但要根据业务决定。
5.2 批量调用与结果输出
将类实例化后,逐条调用:
pipeline = AccusationPipeline( classifier=model, zero_shot=zero_shot, threshold_review=0.5, threshold_high=0.8, ) for doc in docs: result = pipeline.detect(doc) print(json.dumps(result, ensure_ascii=False))输出示例:
{"id": "doc_005", "weak_label": 1, "weak_relation": "rule_hit", "prediction": 1, "probability": 0.93, "decision": "high_risk"}这里的probability是模型对正例的预测概率,具体数值会随训练数据和模型变化。你可以把低置信度样本设置为review,由人工处理。这样既减少了漏报,又控制了误报成本。
5.3 生产环境中的调度与缓存
在实验环境,管道运行一次即可。生产环境通常要把管道拆成在线服务和离线批处理两部分:在线服务处理投诉工单或新增文本,离线批处理定期重跑历史数据。
如果使用 Redis 作为管道缓存,批量获取文本指纹时可以使用 Redis 的 pipeline 通信机制减少网络往返。但这里要把它和 NLP 管道区分开,Redis pipeline 只是批量读写缓存时的客户端通信方式,不改变 NLP 管道本身的级联逻辑。
6. 运行验证与常见问题排查
6.1 完整运行流程和检查点
建议按下面顺序验证:
- 加载语料并统计各字段缺失情况。
- 运行弱标签函数,打印
label分布。 - 查看规则命中的具体词和位置,判断规则是否过严或过宽。
- 训练模型,保存分类报告。
- 在人工评估集上验证,对比弱标签准确率。
- 使用管道对新文档预测,检查高置信度样本是否符合直觉。
可以用一句话代码检查弱标签分布:
print(df["label"].value_counts())如果label=1的数量明显过少,说明规则触发条件太严格,或业务词表和指控词表都不完整。如果label=0的样本太多,需要检查业务词表是否把大量无关文本也圈进来了。
6.2 常见错误与解决表
| 问题现象 | 常见原因 | 检查方式 | 解决建议 |
|---|---|---|---|
| 弱标签几乎全是 0 | 指控词和业务词共现条件过严 | 打印规则命中次数 | 扩充指控词和业务词表,或放宽窗口长度 |
| 模型把大部分文本判为 1 | 负样本被规则建设性地排除 | 查看训练集标签分布 | 增加业务词命中但规则未命中的负样本 |
| “否认串通”被判为正例 | 否定过滤未覆盖该上下文 | 打印上下文片段 | 检查否认、没有、否定窗口;必要时用依存句法 |
| 零样本分类结果不稳定 | 候选标签措辞不清 | 打印置信度分布 | 调整标签描述,例如“指控采购违规”和“客观陈述” |
| 模型预测和弱标签明显冲突 | 弱标签噪声大或特征选择不合适 | 抽样看冲突样本 | 提高弱标签合并阈值,或换成语义嵌入特征 |
6.3 从现象定位是第几级的问题
排查时先判断错误来自哪一级:
- 如果规则命中率低,问题在第一级规则库。
- 如果规则命中率高但模型预测结果差,问题在第二级训练或特征。
- 如果管道输出的置信度分布扭曲,需要看训练集中的标签是否造假过多。
- 如果线上新文本效果差,优先检查文本清洗是否一致,例如是否做了去 HTML、去空白、截断等操作。
这种逐级定位方式,比直接改模型参数更有效。
7. 生产化最佳实践与扩展方向
7.1 学习环境到生产环境的差异
实验阶段可以用小数据集和 Jupyter Notebook,生产环境必须考虑稳定性和可观测性。
| 维度 | 学习 / 实验 | 生产 |
|---|---|---|
| 数据 | 小型示例数据 | 大数据量、脱敏、版本管理 |
| 模型 | CPU 单机运行 | GPU 服务或离线批处理 |
| 规则 | 硬编码在 Python 文件 | 配置化、支持热更新 |
| 监控 | 手动打印输出 | 指标、日志、报警 |
| 人工反馈 | 临时抽样 | 标注平台回流 |
| 回滚 | 不需要 | 保留上一版本模型和规则 |
文本脱敏尤其重要。公共采购语料可能包含企业名称、人员姓名、项目编号和联系方式,在写入训练集前需要做实体脱敏或权限管控。
7.2 上线前检查清单
在把管道发布到生产前,可以按下面的清单自检:
- 数据清洗链路是否统一,训练和推理是否走同一套处理函数。
- 弱标签产生的规则版本是否纳入版本管理。
- 是否随机抽样了足够数量的人工评估集并完成复核。
- 是否保存了模型训练数据和评估报告。
- 是否定义了
high_risk、review、low_risk三档处理策略。 - 是否对每条预测记录保留
id、模型版本、规则版本、概率、决策结果。 - 是否设置了置信度分布、规则命中率、接口耗时等监控指标。
- 是否有回滚机制,模型或规则更新后能否快速切回旧版本。
上线前不需要追求模型完美,但一定要保证错误可追溯。
7.3 扩展方向:从二元检测到细粒度风险识别
二元检测只是第一步。实际业务更关心的是“指控类型”和“被指控对象”。扩展方向包括:
- 多标签分类:把单一标签拆成“围标串标”“参数倾向性”“资质造假”“评标不公”等类别。
- 命名实体识别:识别文本中的供应商名称、项目编号、评标专家等实体,输出“谁指控谁存在什么问题”。
- 主动学习:把模型低置信度的样本推给人工标注,标注结果回到训练集,比随机抽样效率高。
- 持续学习:定期用新标注数据重训模型,同时保留旧模型做 A/B 对比。
整体来看,这个任务的技术关键不在模型有多复杂,而在于弱标签质量、规则版本管理和人工反馈闭环。建议第一步先做出规则命中和弱