背景痛点：为什么传统方案总被用户吐槽“答非所问”
过去一年，我们团队陆续接了三个行业的智能客服外包：银行、医疗 SaaS 与跨境电商。上线前大家都觉得“不就是 FAQ 检索嘛”，结果灰度测试第一天就被用户喷到怀疑人生。总结下来，高频踩坑集中在四点：
- 多轮对话上下文缺失——用户追问“那手续费呢？”，机器人却回到“请问您想办什么业务？”
- 行业术语理解偏差——“我买了 3C 认证产品”被拆成“3C”+“认证”+“产品”，结果召回一堆无关售后政策
- 非结构化语料爆炸——PDF 手册、HTML 页面、钉钉群文件，格式乱到飞起，ES 里一搜全是噪声
- 冷启动响应慢——新业务上线，知识库只有 200 条样本，ES 的 BM25 直接“半宕机”，平均响应 1.8 s
痛定思痛，我们决定扔掉“正则+关键词”老套路，用 BERT+Faiss 重构整条链路。下文把踩过的坑、调参的魔法和血泪教训一次性写全，代码全部可落地，拿去就能跑。
技术对比：正则、ES 与向量化方案怎么选
先给出一张 3×4 的对比表，方便一眼看穿优劣（ 越多越香）：
| 维度 | 正则匹配 | ES/BM25 | BERT+Faiss |
|---|---|---|---|
| 开发速度 | |||
| 语义泛化 | |||
| 多语言 | |||
| 运维成本 |
结论速览
- 正则：适合极固定、无歧义的指令，如“查询余额请回复 1”。业务一旦膨胀，规则交叉堪比蜘蛛网
- ES：BM25 对关键词召回率/recall rate 不错，但无法解决“同义不同词”问题，且对长尾 query 精度骤降
- BERT+Faiss：一次微调就能把行业语料吃进 embedding 空间，配合 IVF_PQ 压缩，单机 2000 万条 768 维向量延迟稳定在 30 ms 内，GPU 内存占用 <4 GB，真香
核心实现：从脏数据到 30 ms 检索
1. 数据清洗：让 PDF 乖乖吐出纯文本
非结构化语料最头疼的是“半吊子”PDF：有扫描件、有表格、还有横版 PPT。下面这段代码用pdfplumber按页拆，表格单独存，异常页自动写日志，保证下游只拿到干净文本。
# data_ingestion.py import pdfplumber, pathlib, logging from typing import List logging.basicConfig(level=logging.INFO) def extract_text(pdf_path: str, min_table_len: int = 3) -> List[str]: """ 提取正文与表格文本，过滤空页。 :param pdf_path: 文件路径 :param min_table_len: 表格最少行数，低于则忽略 :return: 每段文本块 """ chunks: List[str] = [] try: with pdfplumber.open(pdf_path) as pdf: for page in pdf.pages: # 正文 text = page.extract_text() if text and len(text.strip()) > 20: chunks.append(text.strip()) # 表格 for table in page.extract_tables(): if table and len(table) >= min_table_len: flat = " ".join([" ".join(row) for row in table if row]) chunks.append(flat) except Exception as e: logging.exception(f"bad pdf: {pdf_path}, e={e}") return chunks跑批时把extract_text包进ProcessPoolExecutor，8 核 CPU 每分钟能啃 600 份 20 页 PDF，OCR 阶段用easyocr兜底，这里不展开。
2. 语义编码：领域自适应 Fine-tuning
直接用bert-base-chinese做特征，发现“银承”“福费廷”这类金融黑话全部被打包进[UNK]。于是用“继续预训练+任务微调”两阶段：
- 继续预训练（MLM）
把 1.2 GB 行业语料扔给transformers的run_mlm.py，学习率 1e-4，batch 64，跑 3 个 epoch，loss 从 3.8 降到 2.1 - 任务微调（Sentence Pair）
构造 8 万组<query, answer>正例 + 16 万负例，用cosent损失拉近正样本对，训练 2 epoch，最终召回率提升 18%
关键代码片段（简化）：
# train_semantic.py from transformers import BertTokenizer, BertForSequenceClassification from dataset import PairDataset from torch.utils.data import DataLoader import torch tokenizer = BertTokenizer.from_pretrained("bert-base-chinese") model = BertForSequenceClassification.from_pretrained( "./continue_pretrain", num_labels=2) def train(model, dataloader, lr: float = 2e-5, epochs: int = 2): optimizer = torch.optim.AdamW(model.parameters(), lr=lr) model.train() for epoch in range(epochs): for batch in dataloader: optimizer.zero_grad() out = model(**batch) loss = out.loss loss.backward() optimizer.step() torch.save(model.state_dict(), "sent_bert.bin")3. 检索优化：Faiss IVF_PQ 调参指南
向量维度 768，数据量 2000 万，目标延迟 <30 ms，内存 <4 GB。核心参数如下：
nlist=4096，把数据切成 4 k 聚类桶M=48，PQ 分成 48 子空间，每子空间 16 bit，压缩比≈8nprobe=32，查询时扫 32 个桶，recall@10 保持 0.92- GPU 版
IndexIVFPQ先train()再add()，训练样本 50 万足够
# build_index.py import faiss, numpy.p as np d = 768 nb = 2_000_000 xb = np.random.random((nb, d)).astype('float32') # 伪数据演示 index = faiss.index_factory(d, "IVF4096,PQ48") index.train(xb) index.add(xb) faiss.write_index(index, "faq.index")实测单机 Tesla T4，batch=32，平均延迟 27 ms，CPU 回退模式 65 ms，满足业务要求。
生产考量：增量更新与敏感过滤
1. 实时性：增量索引更新策略
全量重建 2000 万条要 3 h，业务等不起。采用“双 Buffer + 版本号”机制：
- 内存里维护两份
IndexIVFPQ指针，一主一备 - 新数据先写 Kafka，Flume 攒 5 min 小批，向量编码后写临时
index_delta - 每 30 min 把
index_delta与主索引merge成新索引，切换指针，原子替换 - 老索引延迟删除，保证回滚
这样新增 5 万条延迟控制在 5 min 内，对客服无感知。
2. 安全性：三级敏感信息过滤
金融、医疗场景最怕泄露身份证、银行卡号。采用“正则+关键词+模型”三级：
- ** 正则：
\b\d{15,18}\b直接打码 - 关键词：维护 1.3 万敏感词 Trie 树，O(n) 扫描
- 模型：用
bert-base-chinese二分类“正常/敏感”，recall 0.96，precision 0.92
过滤链路放在“写入前”，一旦命中直接落审计日志，不入索引，保证“可溯源、不可见”。
避坑指南：三次线上故障复盘
OOM：GPU 显存飙到 32 GB
原因：index.add_with_ids()一次性喂 500 万向量
解决：分 10 万一批，加torch.cuda.empty_cache()，显存降到 4 GB向量维度不一致
原因：训练用 768，线上误用 512 维轻量模型
解决：在build_index.py里加断言assert xb.shape[1] == d，CI 阶段拦截PQ 压缩后精度雪崩
原因：M=96，每子空间 8 bit，过压缩
解决：把M降到 48，nprobe从 16 提到 32，recall@10 从 0.78 拉回 0.92
代码规范：PEP8 与类型注解示例
团队强制black+flake8，关键函数必须写 Google Style docstring。示例：
def search( query: str, top_k: int = 10, index: faiss.IndexIVFPQ = None, model: BertModel = None, tokenizer: BertTokenizer = None, ) -> List[Tuple[int, float]]: """ 语义检索入口。 :param query: 用户问题 :param top_k: 返回条数 :param index: Faiss 索引 :param model: 编码模型 :param tokenizer: 分词器 :return: [(idx, score), ...] """ inputs = tokenizer(query, return_tensors="pt", max_length=64, truncation=True) with torch.no_grad(): vec = model(**inputs).pooler_output.numpy().astype("float32") faiss.normalize_L2(vec) scores, idxs = index.search(vec, top_k) return list(zip(idxs[0], scores[0]))结论与开放讨论
经过三个月灰度，我们的 BERT+Faiss 方案在 20 万条真实会话中，把问答匹配准确率从 56% 提到 78%，平均响应时间 29 ms，机器内存 3.7 GB，基本达到“准实时+高精准”目标。但仍有几个开放问题留给大家一起思考：
- 如何平衡语义精度与响应延迟？当数据量再上 1 亿条，是否考虑两阶段检索（粗排+精排）？
- 多轮对话的上下文向量如何优雅融合？是把历史 query 平均池化，还是做层级注意力？
- 增量更新频率继续缩短到 1 min，会不会因 merge 过于频繁反而造成 CPU 抖动？
欢迎留言聊聊你们的做法，一起把智能客服做得“更像人”。
