背景痛点：传统智能客服的三座大山
去年做 7×24 小时智能客服时，我们被三件事折磨得够呛：
- 知识库更新滞后：运营同学刚把新活动规则贴进 Confluence，线上已经冒出 200 多个“为什么提示券不可用？”的工单，模型却还在用上周的“旧口播”回答。
- 长尾问题雪崩：618 大促凌晨，突然涌进“定金能合并尾款吗”这种训练语料里从没出现过的问法，BERT 分类器直接掉线，Fallback 到关键词匹配，准确率从 92% 跌到 38%。
- 响应速度失控：为了覆盖长尾，我们把 FAQ 从 2 万条膨胀到 18 万条，ElasticSearch 的召回延迟从 120 ms 涨到 600 ms，再加上生成模型 beam search 的 1.2 s，用户平均等待 1.8 s，体验“肉眼可见”地崩了。
这三座大山，让我们下定决心把系统重构成 RAG（Retrieval-Augmented Generation）架构：让生成模型只负责“说人话”，知识实时性交给动态检索。
技术对比：三种方案硬指标横评
| 维度 | 规则系统 | 纯生成式（T5/ChatGLM） | RAG |
|---|---|---|---|
| 延迟 P95 | 120 ms | 1.2 s | 380 ms |
| 准确率（Top1） | 68% | 85% | 91% |
| 知识更新成本 | 高（人工写规则） | 需全量微调 | 分钟级热插拔 |
| 长尾覆盖 | 差 | 好 | 好 |
| 幻觉风险 | 无 | 高 | 中（可控） |
| 维护人力 | 3 人/周 | 1 人/月 | 0.3 人/周 |
实测数据来自我们 4 台 A100 的灰度集群，1000 QPS 压测。RAG 用 380 ms 换来 91% 准确率，ROI 最高。
核心实现：检索器+生成器 1+1>2
系统总览
- 离线层：把 FAQ、商品详情、活动文案切成 256 token 的 Chunk，用 text2vec-large-chinese 转成 1024 维向量，写入 FAISS IVF1024+HNSW 混合索引。
- 在线层：用户 Query → Query Rewrite → 检索 Top20 → Rerrank（cross-encoder）→ Top5 → Prompt 模板 → 生成答案。
- 反馈层：用户点踩/点赞 → 日志 → 每日离线 nDCG 评估 → 低分 Chunk 自动下架。
关键代码
以下示例基于 Python 3.8，依赖 faiss-cpu==1.7.4、transformers==4.38、fastapi==0.110。
1. 知识库向量化存储
# kb_indexer.py from pathlib import Path import faiss, json, torch from transformers import AutoTokenizer, AutoModel from typing import List class VectorIndexer: def __init__(self, model_name: str = "GanymedeNil/text2vec-large-chinese"): self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModel.from_pretrained(model_name).eval().cuda() self.index = faiss.IndexHNSWFlat(1024, 64, faiss.METRIC_INNER_PRODUCT) self.index.hnsw.efConstruction = 200 @torch.inference_mode() def encode(self, texts: List[str]) -> torch.Tensor: inputs = self.tokenizer(texts, padding=True, truncation=True, max_length=256, return_tensors="pt").to("cuda") return self.model(**inputs).last_hidden_state[:, 0, :].cpu() def add_chunks(self, chunks: List[str], ids: List[int]): vecs = self.encode(chunks).numpy() faiss.normalize_L2(vecs) self.index.add_with_ids(vecs, np.array(ids, dtype=np.int64)) def save(self, path: Path): faiss.write_index(self.index, str(path/"faq.index")) (path/"faq_map.json").write_text(json.dumps({i: c for i, c in enumerate(chunks)}))2. 检索-生成流水线
# rag_service.py import faiss, json, numpy; import numpy as np from transformers import AutoTokenizer, AutoModelForCausalLM from pydantic import BaseModel class Query(BaseModel): uid: str text: str class RAGService: def __init__(self, index_path: Path, llm_path: str): self.index = faiss.read_index(str(index_path/"faq.index")) self.chunk_map = json.loads((index_path/"faq_map.json").read_text()) self.tokenizer = AutoTokenizer.from_pretrained(llm_path) self.llm = AutoModelForCausalLM.from_pretrained(llm_path).half().cuda().eval() self.rerank = self._load_cross_encoder() def rewrite(self, q: str) -> str: # 简单同义改写，可换成 T5-Prefix return q.replace("你们", "贵司").replace("吗", "吗") def retrieve(self, q: str, k: int = 20) -> List[dict]: vec = self.encode([self.rewrite(q)]) D, I = self.index.search(vec, k) return [{"id": int(i), "score": float(d), "text": self.chunk_map[str(i)]} for d, i in zip(D[0], I[0])] def rerank(self, q: str, candidates: List[dict], top_k: int = 5): pairs = [(q, c["text"]) for c in candidates] scores = self.rerank.predict(pairs) for c, s in zip(candidates, scores): c["rerank_score"] = float(s) return sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)[:top_k] def generate(self, q: str, refs: List[str], max_len: int = 150) -> str: prompt = f"背景知识：\n" + "\n".join(refs) + f"\n问题：{q}\n答案：" inputs = self.tokenizer(prompt, return_tensors="pt").to("cuda") out = self.llm.generate(**inputs, max_new_tokens=max_len, do_sample=False, pad_token_id=self.tokenizer.eos_token_id) return self.tokenizer.decode(out[0][inputs.input_ids.shape[1]:], skip_special_tokens=True) def ask(self, query: Query): cands = self.retrieve(query.text) top5 = self.rerank(query.text, cands) answer = self.generate(query.text, [t["text"] for t in top5]) return {"answer": answer, "refs": top5}3. FastAPI 入口
# main.py from fastapi import FastAPI app = FastAPI() rag = RAGService(Path("./data"), "THUDM/chatglm3-6b") @app.post("/ask") def ask(q: Query): return rag.ask(q)生产考量：高并发下的三板斧
缓存策略
- 把 Query 向量做 64bit 哈希，Redis 缓存 Top5 结果 TTL=300 s，实测 30% QPS 直接命中，P99 延迟降到 180 ms。
- 对热点商品 ID 做本地 LRU 向量缓存，减少 FAISS 查询 20%。
知识库增量更新
- 采用“双索引”滚动：线上读旧索引，离线写新索引，写完原子替换文件名，重启零中断。
- 每日凌晨拉取 CMS 变更，diff 后只重算新增/修改 Chunk，平均 3 min 完成。
异常熔断
- 检索返回空或最高分 < 0.65 时，触发熔断，直接返回“转人工”文案，避免幻觉。
- 生成模型输出包含“无法确定/我不知道”且 logits 平均熵 > 5.0 时，同样熔断。
避坑指南：踩出来的 5 个血泪教训
- 向量维度不是越高越好
- 把 1024 维升到 1536 维，召回率只涨 0.8%，延迟却 +30%，最后回退 1024。
- 避免幻觉的 prompt 技巧
- 在 prompt 末尾加“若背景知识无法回答问题，请直接回复‘请联系人工客服’”，幻觉率从 12% 降到 3%。
- nDCG 监控
- 日志里埋点“是否解决”，每天跑一次 nDCG@5，低于 0.85 自动报警，方便定位脏数据。
- 分片大小 256 token 是甜点值
- 过大易引入噪声，过小断句丢失语义，实测 256 时 F1 最高。
- 记得给 Cross-Encoder 降 batch
- 20 条候选一次喂给显卡，latency 从 90 ms 降到 25 ms，吞吐量翻倍。
结尾：多轮对话的上下文保持，你打算怎么做？
目前我们的 RAG 还是“单轮问答”模式，下一轮如果用户追问“那第二件半价呢？”——如何把上一轮检索到的 Chunk 以及用户意图无缝带进来，既不让上下文爆炸，又能继续精准检索？是把历史 Query+Answer 直接拼进 prompt，还是再做一次语义摘要后重新检索？欢迎留言聊聊你的做法。
