台湾统一企业集团旗下零售平台 uniopen 面对的并不是普通文本分类问题。商品标题、促销文案、用户内容与平台政策交织在一起,同一句话是否允许发布,往往取决于商品类别、业务场景和风险等级。uniopen 的实践重点,是通过 Amazon SageMaker AI 中的监督微调与提示词优化,让 Amazon Nova 2 Lite 更贴合零售审核政策,再用面向业务的评测和发布门禁控制生产风险。
定制的目标不是“更聪明”,而是“更符合政策”
通用模型能够识别大量语言现象,但企业审核系统还有三类更具体的要求:
- 规则边界明确:什么内容直接拦截,什么内容转人工,什么内容可以自动放行。
- 输出稳定可解析:下游服务需要固定标签、命中的政策编号和简短理由,而不是开放式长答案。
- 错误成本不对称:漏掉高风险内容与误拦正常商品,对业务造成的损失并不相同。
因此,模型效果不能只看整体准确率。假设审核标签为 ALLOW、REVIEW 和 BLOCK,平台通常更关心:
- 应拦截内容的召回率;
- 自动放行结果的精确率;
- 正常商品被错误拦截的比例;
- 不同商品类目、语言和促销场景中的最差表现;
- 推理延迟、人工复核率与单次审核成本。
这些标签和指标只是便于说明的实践示例,并不代表 uniopen 公开了相同的内部分类或阈值。
监督微调与提示词优化解决不同问题
监督微调适合把长期稳定、需要模型内化的审核模式写入模型。例如,训练样本可以覆盖政策标签、边界案例、容易混淆的表达,以及人工审核员最终确认的结论。SageMaker AI 则为训练任务、模型产物和部署流程提供统一承载环境。
提示词优化处理的是另一层问题:明确当前政策版本、规定输出结构、补充商品上下文,并告诉模型在不确定时转入人工复核。它比重新训练更容易更新,适合应对短期活动规则和经常调整的运营要求。
生产系统可以采用如下分工:
审核请求
-> 输入清洗与商品上下文拼装
-> 当前政策提示词
-> 经监督微调的 Nova 2 Lite
-> JSON 结构校验
-> 高风险规则兜底
-> ALLOW / REVIEW / BLOCK
一个可改造的提示词骨架如下。字段和规则编号均为示例,接入时应替换为企业自己的政策术语:
你是零售平台内容审核器。
决策优先级:
1. 命中明确禁止规则时输出 BLOCK。
2. 证据不足、上下文缺失或多条规则冲突时输出 REVIEW。
3. 只有确认未命中限制规则时才输出 ALLOW。
只返回 JSON:
{"decision":"ALLOW|REVIEW|BLOCK","policy_id":"规则编号或 NONE","reason":"不超过 40 字"}
商品类目:{{category}}
商品标题:{{title}}
商品描述:{{description}}
当前有效政策:{{policy_text}}
这里最重要的约束不是“只返回 JSON”,而是把不确定性显式映射到 REVIEW。如果强迫模型在放行和拦截之间二选一,边界案例很容易变成不可控的线上错误。
用业务切片建立发布门禁
离线评测集应尽量来自真实业务分布,同时保留专门的挑战集。除了随机抽样,还可以按以下维度切片:
- 食品、保健品、酒类等商品类目;
- 繁体中文、英文及中英混合文本;
- 商品标题、详情页、评价和促销内容;
- 新规则、历史误判样本和刻意规避规则的表达;
- 高流量商家与长尾商家。
下面是一段可直接运行的简化发布门禁脚本。它读取模型预测结果,检查拦截召回率、自动放行精确率和误拦率;阈值仅用于演示,不能直接视为真实生产标准。
先保存为 eval_gate.py:
import json
import sys
from collections import defaultdict
VALID = {"ALLOW", "REVIEW", "BLOCK"}
def load_rows(path):
with open(path, encoding="utf-8") as f:
rows = [json.loads(line) for line in f if line.strip()]
for row in rows:
if row["expected"] not in VALID or row["predicted"] not in VALID:
raise ValueError(f"Unknown label: {row}")
return rows
def metrics(rows):
expected_block = sum(r["expected"] == "BLOCK" for r in rows)
predicted_allow = sum(r["predicted"] == "ALLOW" for r in rows)
expected_allow = sum(r["expected"] == "ALLOW" for r in rows)
caught_block = sum(
r["expected"] == "BLOCK" and r["predicted"] == "BLOCK"
for r in rows
)
safe_allow = sum(
r["expected"] == "ALLOW" and r["predicted"] == "ALLOW"
for r in rows
)
false_block = sum(
r["expected"] == "ALLOW" and r["predicted"] == "BLOCK"
for r in rows
)
return {
"block_recall": caught_block / expected_block if expected_block else 1.0,
"allow_precision": safe_allow / predicted_allow if predicted_allow else 1.0,
"false_block_rate": false_block / expected_allow if expected_allow else 0.0,
}
rows = load_rows(sys.argv[1])
result = metrics(rows)
print("global", json.dumps(result, ensure_ascii=False))
failed = (
result["block_recall"] < 0.95
or result["allow_precision"] < 0.98
or result["false_block_rate"] > 0.01
)
by_segment = defaultdict(list)
for row in rows:
by_segment[row["segment"]].append(row)
for segment, items in sorted(by_segment.items()):
value = metrics(items)
print(segment, json.dumps(value, ensure_ascii=False))
has_block = any(r["expected"] == "BLOCK" for r in items)
if has_block and value["block_recall"] < 0.90:
failed = True
sys.exit(1 if failed else 0)
准备一个最小测试文件:
cat > predictions.jsonl <<'EOF'
{"segment":"food","expected":"ALLOW","predicted":"ALLOW"}
{"segment":"food","expected":"BLOCK","predicted":"BLOCK"}
{"segment":"health","expected":"ALLOW","predicted":"ALLOW"}
{"segment":"health","expected":"BLOCK","predicted":"BLOCK"}
{"segment":"promotion","expected":"REVIEW","predicted":"REVIEW"}
EOF
python eval_gate.py predictions.jsonl
脚本以非零状态退出时,CI/CD 流水线就可以阻止模型发布。实际使用时还应设置最小样本量,计算置信区间,并分别维护回归集、挑战集和近期线上样本,避免小数据切片产生虚假的高分。
从离线高分走到安全上线
模型通过门禁并不意味着可以立即接管全部流量。更稳妥的发布路径是:先以影子模式运行,与现有审核结果比较;再开放少量低风险流量;持续观察人工推翻率、误拦投诉、各类目分布漂移和 P95 延迟;最后才逐步扩大自动决策范围。
还要给模型之外的系统保留控制权:JSON 解析失败时转人工,高风险关键词或受监管商品继续经过确定性规则,所有请求记录模型版本、提示词版本和政策版本。出现事故时,团队才能判断问题来自数据、微调、提示词还是政策配置。
uniopen 案例带来的核心启示,是生产级模型定制并非一次训练任务。监督微调负责学习稳定政策模式,提示词承载可快速变化的上下文,业务评测衡量真实错误成本,而发布门禁决定一个新版本是否有资格接触用户。四者组合起来,通用模型才可能成为可治理、可回滚的零售审核组件。