三层自动打标流水线 — 规则引擎+ML分类器+LLM抽取的混合置信度打标体系
Skill-Auto-Tagging-Pipeline-Rule-ML-LLM · 24-标签工程
1. 解决的问题
数据团队面临"500个SKU打标需要1周人工"——规则+ML+LLM三层流水线将覆盖率从30%提升至97%，人工审核量从100%降至3%，节省标注成本4万元/年
2. 核心算法逻辑
三层打标策略（按优先级从高到低）：
3. 业务应用场景
场景A：SKU全量自动打标流水线 - 业务问题：500个 SKU，10 种标签维度，传统人工打标需要 1 周/次更新，远不能满足实时性要求 - 三层方案： - 规则层（瞬时）：`IF DOS<7 → stockout_risk=critical`；`IF 退货率>10% → return_risk=high` - ML层（毫秒）：GBM 训练 ABC 分类器（特征：销售额/周转率/库龄） - LLM层（分钟）：从产品描述提取合规关键词 → 合规标签 - 覆盖分布：规则层 45%，ML层 40%，LLM层 12%，人工审核 3% - 业务价值：打标时效从 1 周→ 实时（规则+ML），LLM每日
**场景B：供应商评论语义标签（LLM层）** - **业务问题**：采购团队对供应商有大量非结构化备注（"这家工厂交期不稳定""质量检查很严格"），无法被规则或ML处理 - **LLM提取**： - 输入：`"宁波精工交期非常稳定，但价格略高，CE认证资料齐全"` - 输出：`{delivery_reliability: "high", price_competitiveness: "low", compliance_certs: ["CE"]}` - **业务价值**：采购知识从"只存在个人脑中"→ 结构化供应商标签，可被后续决策系统使用
4. 输入数据要求
请查看原始代码模板获取输入规格。
5. 输出结果
请查看原始代码模板获取输出规格。
6. 业务价值 / ROI
- ROI预估：三层流水线将500个SKU的10维标签打标时效从"1周人工→实时(规则+ML)+每日批量(LLM)"，标签覆盖率从30%→97%；人工审核量从100%降至3%，节省人力成本约4万元/年
- 实施难度：⭐⭐⭐☆☆（规则层容易，ML层需要训练数据，LLM层需要API成本控制）
- 优先级评分：⭐⭐⭐⭐⭐（打标是整个标签工程体系的数据入口，没有高质量打标，传播和Action触发都无从谈起）
- 评估依据：三层混合策略比纯LLM打标便宜90%（LLM每千tokens约¥0.1，500 SKU × 10 tags × LLM全量 vs 仅12%走LLM层）
7. 代码模板
代码块数量：2 · 路径：未检测到
"""
三层自动打标流水线
功能：规则引擎 / ML分类器 / LLM抽取 / 置信度聚合 / 人工审核队列
输入：实体数据 + Tag Schema定义
输出：标签结果 + 置信度 + 来源追踪 + 审核队列
"""
import numpy as np
import pandas as pd
import re
import json
from dataclasses import dataclass, field
from typing import Any, Optional
import warnings
warnings.filterwarnings('ignore')
@dataclass
class TagResult:
tag_id: str
value: Any
confidence: float
source: str # rule/ml/llm/manual
rule_matched: Optional[str] = None
needs_review: bool = False
class RuleEngine:
"""Layer 1: 规则引擎（确定性标签）"""
def __init__(self):
self.rules = []
def add_rule(self, tag_id: str, condition: callable, value_fn: callable,
description: str = ""):
self.rules.append({
"tag_id": tag_id, "condition": condition,
"value_fn": value_fn, "description": description
})
def evaluate(self, entity: dict) -> list:
results = []
for rule in self.rules:
try:
if rule["condition"](entity):
value = rule["value_fn"](entity)
results.append(TagResult(
tag_id=rule["tag_id"],
value=value,
confidence=1.0,
source="rule",
rule_matched=rule["description"],
))
except Exception:
continue
return results
class MLTagger:
"""Layer 2: ML分类器（统计推断标签）"""
8. 论文来源
- 2209.04485
- 2307.09288
- 2402.15758