智能客服FAQ系统效率提升实战：从意图识别到响应优化-编程实验室

背景痛点：传统 FAQ 的“慢”与“错”

去年双十一，公司客服峰值 QPS 飙到 1.2 w，老系统直接“罢工”：

关键词+正则的意图判断，命中率 68%，剩下 32% 全转人工；
每次查询都要扫一遍 8 w 条 FAQ，平均响应 680 ms，P99 1.4 s；
没有状态缓存，同一用户 3 s 内重复提问，后台重复算 3 次；
规则热更新靠发版，凌晨 2 点紧急改一条正则，全集群重启 15 min。

一句话：高并发场景下，传统 FAQ 既慢又错，维护还费劲。

技术对比：规则 vs 统计 vs 预训练

维度	规则引擎	TF-IDF+LightGBM	BERT 微调
准确率	0.68	0.81	0.93
吞吐量（单核）	4500 QPS	2200 QPS	1100 QPS
维护成本	人力堆规则	周级重训	天级微调
扩展性	差	中	好（支持多语言）

结论：BERT 单次推理慢，但结合缓存与异步批跑，综合吞吐反而翻倍，后面实战会量化。

核心实现：一条问答请求的“旅程”

1. 意图识别模块（Transformers 微调）

# intent_model.py from typing import List, Dict import torch, json from transformers import BertTokenizer, BertForSequenceClassification from pydantic import BaseModel, Field class PredictRequest(BaseModel): text: str = Field(..., min_length=2, max_length=128) class IntentModel: def __init__(self, model_dir: str, device: str = "cuda"): self.tokenizer = BertTokenizer.from_pretrained(model_dir) self.model = BertForSequenceClassification.from_pretrained(model_dir) self.model.to(device).eval() self.device = device @torch.no_grad() def predict(self, req: PredictRequest) -> Dict[str, float]: """返回 {label:prob} 字典，top5""" inputs = self.tokenizer( req.text, return_tensors="pt", truncation=True, max_length=64 ).to(self.device) logits = self.model(**inputs).logits probs = torch.nn.functional.softmax(logits, dim=-1) top5 = torch.topk(probs[0], 5) return {self.model.config.id2token[i.item()]: v.item() for v, i in zip(top5.values, top5.indices)}

训练脚本用Trainer默认参数，3 轮早停，学习率 2e-5，负采样比例 1:3，最终 F1 0.93。

2. 问答缓存层（毫秒级返回）

# cache.py import redis, json, hashlib from typing import Optional class FAQCache: def __init__(self, redis_url: str, ttl: int = 3600): self.r = redis.from_url(redis_url, decode_responses=True) self.ttl = ttl def _key(self, text: str) -> str: return "faq:v1:" + hashlib.md5(text.encode()).hexdigest() def get(self, text: str) -> Optional[dict]: data = self.r.get(self._key(text)) return json.loads(data) if data else None def set(self, text: str, payload: dict): self.r.setex(self._key(text), self.ttl, json.dumps(payload))

缓存命中率 72%，P99 延迟从 680 ms 降到 18 ms。

3. 异步流水线（FastAPI + Celery）

# main.py from fastapi import FastAPI, BackgroundTasks from intent_model import IntentModel, PredictRequest from cache import FAQCache import asyncio, os app = FastAPI() model = IntentModel(os.getenv("MODEL_DIR")) cache = FAQCache(os.getenv("REDIS_URL")) @app.post("/ask") async def ask(req: PredictRequest, bt: BackgroundTasks): hit = cache.get(req.text) if hit: return hit # 异步落库+模型推理 loop = asyncio.get_event_loop() pred = await loop.run_in_executor(None, model.predict, req) bt.add_task(cache.set, req.text, pred) return pred

Worker 并发 8 进程 + 4 线程，单节点实测 3200 QPS，GPU 利用率 82%。

避坑指南：生产踩过的 3 个深坑

OOV 词导致 UNK 占比高
- 在 tokenizer 里加never_split=["退款","退运费"]业务词；
- 用领域词表继续预训练 5 k 步，MLM 损失降到 1.9，下游 F1 +2.3%。
对话状态幂等性
- 缓存 key 仅依赖用户问题文本，不依赖 session，避免“刷新一次答案变一次”；
- 对需要多轮槽位的场景，把 slot 拼接后再算 key，保证同状态同答案。
模型热更新
- 采用双目录发布：/models/{timestamp}/，服务启动时软链到 current；
- 通过 etcd 广播版本号，各节点加载完向注册中心写“ready”，滚动 0 损切换。