AI辅助SaaS工单自动分类与路由：从NLP模型到规则引擎的混合方案

AI辅助SaaS工单自动分类与路由：从NLP模型到规则引擎的混合方案

一个中等规模的SaaS平台，每天可能涌入数百甚至上千条工单。人工分类不仅效率低，而且容易因为主观判断偏差导致工单错分——一个"系统崩溃"的紧急工单被当作"功能咨询"处理，后果可能是客户流失。本文复盘一套NLP模型+规则引擎的混合工单分类路由方案，将分类准确率从人工的72%提升到94%。

一、工单分类路由的整体架构

二、工单文本的预处理与意图识别

2.1 多模态工单的文本提取

工单来源多样，先做统一的文本提取和标准化：

import re
import jieba
from typing import List, Dict
from dataclasses import dataclass

@dataclass
class TicketText:
raw_text: str
title: str
description: str
attachments_text: List[str] # 附件OCR文本
metadata: Dict

class TicketPreprocessor:
"""工单文本预处理管道"""

# SaaS工单领域的自定义词典
CUSTOM_DICT = [
'数据库连接池', '读写分离', '分库分表', 'OOM', 'FullGC',
'限流熔断', '降级策略', '死锁', '慢查询', '索引失效',
'Token过期', 'SSO单点登录', '数据迁移', '灰度发布'
]

def __init__(self):
for word in self.CUSTOM_DICT:
jieba.add_word(word)

def preprocess(self, ticket: dict) -> TicketText:
raw = self._extract_text(ticket)

# 标准化管道
cleaned = self._pipeline(raw)

return TicketText(
raw_text=raw,
title=cleaned['title'],
description=cleaned['description'],
attachments_text=cleaned['attachments'],
metadata=self._extract_metadata(ticket)
)

def _pipeline(self, raw: dict) -> dict:
"""文本清洗管道"""
return {
'title': self._clean_text(raw.get('title', '')),
'description': self._clean_text(raw.get('description', '')),
'attachments': [
self._clean_text(t) for t in raw.get('attachments', [])
]
}

def _clean_text(self, text: str) -> str:
"""文本清洗"""
text = text.strip()
# 移除HTML标签
text = re.sub(r'<[^>]+>', '', text)
# 移除多余空白
text = re.sub(r'\\s+', ' ', text)
# 移除敏感信息（手机号、邮箱）
text = re.sub(r'1[3-9]\\d{9}', '[PHONE]', text)
text = re.sub(r'[\\w.-]+@[\\w.-]+', '[EMAIL]', text)
return text

def tokenize(self, text: str) -> List[str]:
"""中文分词 + 停用词过滤"""
tokens = jieba.lcut(text)
# 过滤停用词和标点
tokens = [t for t in tokens if len(t) > 1
and t not in self.stop_words
and not t.isspace()]
return tokens

2.2 意图识别模型

import torch
import torch.nn as nn
from transformers import BertModel, BertTokenizer

class TicketIntentClassifier(nn.Module):
"""基于BERT的工单意图分类模型"""

# 工单分类体系
CATEGORIES = [
'technical_bug', # 技术缺陷
'performance_issue', # 性能问题
'function_consulting', # 功能咨询
'billing_inquiry', # 计费相关
'account_issue', # 账号问题
'data_issue', # 数据问题
'integration_help', # 集成对接
'urgent_outage', # 紧急故障
'feature_request', # 功能建议
'other' # 其他
]

# 紧急程度
PRIORITIES = ['P0_critical', 'P1_high', 'P2_normal', 'P3_low']

def __init__(self, model_name='bert-base-chinese', num_categories=10):
super().__init__()
self.bert = BertModel.from_pretrained(model_name)
self.dropout = nn.Dropout(0.3)

# 多任务学习：同时预测分类和优先级
self.category_head = nn.Linear(768, num_categories)
self.priority_head = nn.Linear(768, 4)

def forward(self, input_ids, attention_mask):
outputs = self.bert(
input_ids=input_ids,
attention_mask=attention_mask
)
pooled = outputs.pooler_output
pooled = self.dropout(pooled)

category_logits = self.category_head(pooled)
priority_logits = self.priority_head(pooled)

return category_logits, priority_logits

def predict(self, text: str, tokenizer, device='cpu'):
"""单条工单预测"""
self.eval()
encoding = tokenizer(
text,
max_length=512,
padding='max_length',
truncation=True,
return_tensors='pt'
)

with torch.no_grad():
cat_logits, pri_logits = self(
encoding['input_ids'].to(device),
encoding['attention_mask'].to(device)
)

cat_probs = torch.softmax(cat_logits, dim=1).squeeze()
pri_probs = torch.softmax(pri_logits, dim=1).squeeze()

cat_idx = cat_probs.argmax().item()
pri_idx = pri_probs.argmax().item()

return {
'category': self.CATEGORIES[cat_idx],
'priority': self.PRIORITIES[pri_idx],
'category_confidence': cat_probs[cat_idx].item(),
'priority_confidence': pri_probs[pri_idx].item(),
'category_distribution': {
self.CATEGORIES[i]: cat_probs[i].item()
for i in range(len(self.CATEGORIES))
}
}

三、规则引擎兜底与混合决策

3.1 规则引擎设计

NLP模型虽然覆盖面广，但对某些强规则场景不如规则引擎精确。两者互补：

@Component
public class TicketRuleEngine {

private final List<ClassificationRule> rules;

/**
* 规则定义（支持热更新，存储在配置中心）
*/
public TicketRuleEngine() {
this.rules = List.of(
// 规则1：关键词匹配 → 紧急故障
ClassificationRule.builder()
.name("system-outage-detection")
.priority(1)
.condition(ticket ->
containsAny(ticket.getDescription(),
"系统崩溃", "服务不可用", "全部宕机", "全线报错",
"用户无法登录", "数据丢失", "生产事故"))
.result(new ClassificationResult("urgent_outage", "P0_critical"))
.build(),

// 规则2：错误码匹配 → 技术缺陷
ClassificationRule.builder()
.name("error-code-matching")
.priority(2)
.condition(ticket ->
matchesPattern(ticket.getDescription(),
"HTTP 5\\\\d{2}|Error Code: \\\\d+|OOM killer|OutOfMemory"))
.result(new ClassificationResult("technical_bug", "P1_high"))
.build(),

// 规则3：计费关键词 → 计费工单
ClassificationRule.builder()
.name("billing-keyword")
.priority(3)
.condition(ticket ->
containsAny(ticket.getDescription(),
"账单", "扣费", "余额", "发票", "充值", "退款",
"价格", "续费", "免费额度"))
.result(new ClassificationResult("billing_inquiry", "P2_normal"))
.build(),

// 规则4：客户级别加权
ClassificationRule.builder()
.name("vip-priority-escalation")
.priority(0) // 最高优先级
.condition(ticket ->
ticket.getTenantTier() == TenantTier.ENTERPRISE
&& (ticket.getDescription().contains("紧急")
|| ticket.getDescription().contains("影响业务")))
.action(ticket -> {
// VIP客户自动升级优先级
ticket.setPriority("P1_high");
ticket.setSlaHours(2); // 2小时SLA
})
.build()
);
}

/**
* 规则引擎执行（按优先级排序，首次匹配即生效）
*/
public Optional<ClassificationResult> evaluate(Ticket ticket) {
return rules.stream()
.sorted(Comparator.comparingInt(ClassificationRule::getPriority))
.filter(rule -> rule.getCondition().test(ticket))
.findFirst()
.map(rule -> {
if (rule.getAction() != null) {
rule.getAction().accept(ticket);
}
return rule.getResult();
});
}
}

3.2 混合决策器

@Service
public class HybridTicketRouter {

private final TicketRuleEngine ruleEngine;
private final NLPModelClient nlpClient;
private final TicketRepository ticketRepo;

private static final double NLP_CONFIDENCE_THRESHOLD = 0.85;

/**
* 混合决策流程：
* 1. 规则引擎优先（强规则直接决策）
* 2. NLP模型兜底（覆盖长尾场景）
* 3. 低置信度转人工审核
*/
public RoutingResult route(Ticket ticket) {
// Step 1: 规则引擎
Optional<ClassificationResult> ruleResult = ruleEngine.evaluate(ticket);
if (ruleResult.isPresent()) {
ClassificationResult result = ruleResult.get();
return autoRoute(ticket, result, "rule_engine");
}

// Step 2: NLP模型
NLPPrediction prediction = nlpClient.predict(
ticket.getTitle() + " " + ticket.getDescription());

if (prediction.getCategoryConfidence() >= NLP_CONFIDENCE_THRESHOLD) {
ClassificationResult result = ClassificationResult.builder()
.category(prediction.getCategory())
.priority(prediction.getPriority())
.confidence(prediction.getCategoryConfidence())
.build();
return autoRoute(ticket, result, "nlp_model");
}

// Step 3: 低置信度 → 人工审核池
ticket.setStatus(TicketStatus.PENDING_REVIEW);
ticket.setRoutingNote(String.format(
"NLP置信度: %.2f, Top-3分类: %s",
prediction.getCategoryConfidence(),
prediction.getTopCategories(3)
));
ticketRepo.save(ticket);

return RoutingResult.builder()
.ticketId(ticket.getId())
.routed(false)
.assignedGroup("manual_review_pool")
.reason("nlp_low_confidence:" + prediction.getCategoryConfidence())
.build();
}

/**
* SLA驱动的优先级排序
*/
private String resolveGroup(ClassificationResult result, Ticket ticket) {
// 基础路由表
Map<String, String> categoryToGroup = Map.of(
"technical_bug", "tech_support_l2",
"performance_issue", "performance_team",
"function_consulting", "product_support",
"billing_inquiry", "billing_team",
"urgent_outage", "sre_oncall"
);

String group = categoryToGroup.getOrDefault(
result.getCategory(), "general_support");

// P0/P1紧急工单 → 7×24值班组
if (List.of("P0_critical", "P1_high").contains(result.getPriority())) {
group = "sre_oncall_724";
}

// 企业版客户 → 专属支持通道
if (ticket.getTenantTier() == TenantTier.ENTERPRISE) {
group = group + "_vip";
}

return group;
}
}

四、在线学习与持续优化

4.1 人工标注反馈闭环

class OnlineLearningPipeline:
"""模型在线学习管道"""

def __init__(self, model, label_threshold=100):
self.model = model
self.label_threshold = label_threshold
self.feedback_buffer = []

def collect_feedback(self, ticket_id: str,
predicted: str, actual: str,
reviewer: str):
"""收集人工审核反馈"""
self.feedback_buffer.append({
'ticket_id': ticket_id,
'predicted': predicted,
'actual': actual,
'reviewer': reviewer,
'timestamp': datetime.now(),
'is_correct': predicted == actual
})

def should_retrain(self) -> bool:
"""当错误反馈积累到阈值时触发重训练"""
errors = [f for f in self.feedback_buffer
if not f['is_correct']]
return len(errors) >= self.label_threshold

def retrain(self):
"""增量微调模型"""
if not self.should_retrain():
return

# 构建微调数据集（仅使用被纠正的样本）
retrain_samples = [
f for f in self.feedback_buffer
if not f['is_correct']
]

# 加载原工单文本 + 正确标签
train_data = self._load_training_data(retrain_samples)

# 执行微调（低学习率，避免灾难性遗忘）
self.model.train()
optimizer = torch.optim.AdamW(
self.model.parameters(), lr=2e-5)

for epoch in range(3):
for batch in train_data:
loss = self._train_step(batch, optimizer)

# 评估微调效果
accuracy = self._evaluate()
print(f"Online learning completed. "
f"New accuracy: {accuracy:.3f}")

# 清空缓冲区
self.feedback_buffer.clear()

4.2 分类质量监控

@Component
public class ClassificationMonitor {

private final MeterRegistry meterRegistry;

/**
* 实时监控分类质量指标
*/
@Scheduled(fixedRate = 300_000) // 每5分钟
public void reportMetrics() {
LocalDateTime lastHour = LocalDateTime.now().minusHours(1);

// 自动分类率 = NLP直出 / 总工单
double autoRate = ticketRepo.countAutoClassified(lastHour)
/ (double) ticketRepo.countTotal(lastHour);
meterRegistry.gauge("ticket.auto_classification.rate", autoRate);

// 人工纠正率 = 审核后被修改的分类数 / 审核总数
double correctionRate = ticketRepo.countCorrected(lastHour)
/ (double) ticketRepo.countReviewed(lastHour);
meterRegistry.gauge("ticket.correction.rate", correctionRate);

// 各类别准确率
for (String category : TicketIntentClassifier.CATEGORIES) {
double accuracy = calcCategoryAccuracy(category, lastHour);
meterRegistry.gauge(
"ticket.accuracy." + category,
Tags.of("category", category),
accuracy);
}

// 告警：自动分类率突降
if (autoRate < 0.60) {
alertService.send("NLP分类模型可能退化，自动分类率降至 "
+ String.format("%.1f%%", autoRate * 100));
}
}
}

五、总结

混合方案上线后的关键成果：

三条核心经验：