核心内容摘要
探寻“污黄软件”的神秘领域：不止于“污”，更有“黄”的无限可能
背景痛点传统智能客服的三座大山去年做 7×24 小时智能客服时我们被三件事折磨得够呛知识库更新滞后运营同学刚把新活动规则贴进 Confluence线上已经冒出 200 多个“为什么提示券不可用”的工单模型却还在用上周的“旧口播”回答。
长尾问题雪崩618 大促凌晨突然涌进“定金能合并尾款吗”这种训练语料里从没出现过的问法BERT 分类器直接掉线Fallback 到关键词匹配准确率从 92% 跌到 38%。
响应速度失控为了覆盖长尾我们把 FAQ 从 2 万条膨胀到 18 万条ElasticSearch 的召回延迟从 120 ms 涨到 600 ms再加上生成模型 beam search 的
2 s用户平均等待
8 s体验“肉眼可见”地崩了。
这三座大山让我们下定决心把系统重构成 RAGRetrieval-Augmented Generation架构让生成模型只负责“说人话”知识实时性交给动态检索。
技术对比三种方案硬指标横评维度规则系统纯生成式T5/ChatGLMRAG延迟 P95120 ms
2 s380 ms准确率Top168%85%91%知识更新成本高人工写规则需全量微调分钟级热插拔长尾覆盖差好好幻觉风险无高中可控维护人力3 人/周1 人/月
3 人/周实测数据来自我们 4 台 A100 的灰度集群1000 QPS 压测。
RAG 用 380 ms 换来 91% 准确率ROI 最高。
核心实现检索器生成器 112系统总览离线层把 FAQ、商品详情、活动文案切成 256 token 的 Chunk用 text2vec-large-chinese 转成 1024 维向量写入 FAISS IVF1024HNSW 混合索引。
在线层用户 Query → Query Rewrite → 检索 Top20 → Rerrankcross-encoder→ Top5 → Prompt 模板 → 生成答案。
反馈层用户点踩/点赞 → 日志 → 每日离线 nDCG 评估 → 低分 Chunk 自动下架。
关键代码以下示例基于 Python
8依赖 faiss-cpu
1.
7.
transformers
4.
fastapi
110。
知识库向量化存储# 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, paddingTrue, truncationTrue, max_length256, return_tensorspt).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, dtypenp.int
) 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)}))
检索-生成流水线# 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
- 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
: 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, keylambda x: x[rerank_score], reverseTrue)[:top_k] def generate(self, q: str, refs: List[str], max_len: int
- str: prompt f背景知识\n \n.join(refs) f\n问题{q}\n答案 inputs self.tokenizer(prompt, return_tensorspt).to(cuda) out self.llm.generate(**inputs, max_new_tokensmax_len, do_sampleFalse, pad_token_idself.tokenizer.eos_token_id) return self.tokenizer.decode(out[0][inputs.input_ids.shape[1]:], skip_special_tokensTrue) 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}
FastAPI 入口# main.py from fastapi import FastAPI app FastAPI() rag RAGService(Path(./data), THUDM/chatglm
b) app.post(/ask) def ask(q: Query): return rag.ask(q)生产考量高并发下的三板斧缓存策略把 Query 向量做 64bit 哈希Redis 缓存 Top5 结果 TTL300 s实测 30% QPS 直接命中P99 延迟降到 180 ms。
对热点商品 ID 做本地 LRU 向量缓存减少 FAISS 查询 20%。
知识库增量更新采用“双索引”滚动线上读旧索引离线写新索引写完原子替换文件名重启零中断。
每日凌晨拉取 CMS 变更diff 后只重算新增/修改 Chunk平均 3 min 完成。
异常熔断检索返回空或最高分
65 时触发熔断直接返回“转人工”文案避免幻觉。
生成模型输出包含“无法确定/我不知道”且 logits 平均熵
0 时同样熔断。
避坑指南踩出来的 5 个血泪教训向量维度不是越高越好把 1024 维升到 1536 维召回率只涨
8%延迟却 30%最后回退 1024。
避免幻觉的 prompt 技巧在 prompt 末尾加“若背景知识无法回答问题请直接回复‘请联系人工客服’”幻觉率从 12% 降到 3%。
nDCG 监控日志里埋点“是否解决”每天跑一次 nDCG5低于
85 自动报警方便定位脏数据。
分片大小 256 token 是甜点值过大易引入噪声过小断句丢失语义实测 256 时 F1 最高。
记得给 Cross-Encoder 降 batch20 条候选一次喂给显卡latency 从 90 ms 降到 25 ms吞吐量翻倍。
结尾多轮对话的上下文保持你打算怎么做目前我们的 RAG 还是“单轮问答”模式下一轮如果用户追问“那第二件半价呢”——如何把上一轮检索到的 Chunk 以及用户意图无缝带进来既不让上下文爆炸又能继续精准检索是把历史 QueryAnswer 直接拼进 prompt还是再做一次语义摘要后重新检索欢迎留言聊聊你的做法。