RAG大模型智能客服：从架构设计到生产环境部署的实战指南-平芜编程栈

背景痛点：传统客服的“老毛病”

做ToB客服的同学都懂，最怕的不是用户问题多，而是“知识库又过期了”。

规则引擎：写一条规则要三天，用户换种问法就“404”；
纯生成式LLM：满嘴跑火车，把2021年的旧文档当成“最新政策”；
多轮对话：用户追问一句“那手续费呢？”模型就失忆，开始胡说。

一句话：知识时效性、上下文一致性、答案可控性，三座大山压得人喘不过气。

技术对比：RAG vs Fine-tune，谁更香？

维度	RAG	Fine-tune
训练成本	0 GPU，只训embedding	全量/LoRA都要卡
知识更新	5 分钟灌新向量	重训+回炉至少半天
响应延迟	+10~30 ms检索	纯生成一样
答案可解释	有出处可溯源	黑盒，得靠运气
多领域切换	换索引即可	得重新微调

结论：客服场景“周更日新”的节奏，RAG几乎是唯一解。

核心实现：Flask+FAISS+LangChain，30分钟搭一套MVP

1. 系统架构图

API层：Flask+gunicorn，同步转异步
检索层：FAISS IndexIVFPQ，百万向量<30 ms
生成层：LangChain RetrievalQA，自带n-shot prompt模板
缓存层：Redis把高频query的embedding按LRU缓存1 h

2. 知识库embedding预处理

代码片段（PEP8，带类型注解）：

# kb_build.py from pathlib import Path from typing import List, Tuple import faiss, json, redis, torch from sentence_transformers import SentenceTransformer encoder = SentenceTransformer("multi-qa-MiniLM-L6-cos-v1") def chunk_overlap(text: str, chunk_size: int = 256, overlap: int = 50) -> List[str]: """滑动窗口chunk，避免截断语义""" tokens = text.split() step = chunk_size - overlap return [" ".join(tokens[i:i+chunk_size]) for i in range(0, len(tokens), step)] def build_index(jsonl_file: Path, index_path: Path) -> None: docs, vecs = [], [] for line in open(jsonl_file, encoding="utf8"): doc = json.loads(line) chunks = chunk_overlap(doc["content"]) docs.extend(chunks) vecs.extend(encoder.encode(chunks, batch_size=64, show_progress_bar=True)) d = vecs[0].shape[0] index = faiss.index_factory(d, "IVF1024,PQ64") index.train(np.array(vecs).astype("float32")) index.add(np.array(vecs).astype("float32")) faiss.write_index(index, str(index_path)) # 把docs id->text落盘，供后续溯源 json.dump(docs, open(index_path.with_suffix(".docs"), "w", encoding="utf8"))

要点：

chunk_size按512 token算，overlap≈20%防止断句
IVF1024,PQ64压缩=内存降8倍，召回@10>0.92

3. 检索+重排序

# retriever.py import faiss, numpy as np from sentence_transformers import CrossEncoder cross_encoder = CrossEncoder("cross-qa/ms-marco-MiniLM-L-6-v2") def search_index(query: str, index: faiss.Index, k: int = 25) -> Tuple[List[str], List[float]]: q_vec = encoder.encode([query]) scores, idxs = index.search(q_vec.astype("float32"), k) return idxs[0].tolist(), scores[0].tolist() def rerank(query: str, candidates: List[str]) -> List[str]: pairs = [(query, c) for c in candidates] ranks = cross_encoder.predict(pairs) return [c for _, c in sorted(zip(ranks, candidates), key=lambda x: x[0], reverse=True)]

MMR多样性排序也顺手加进来，避免top-k全是同义句：

def mmr Diversity(query_vec, cand_vecs, cand_txts, k=5, lamb=0.5): """Maximal Marginal Relevance""" selected, idx = [], list(range(len(cand_txts))) while len(selected) < k and idx: if not selected: best = np.argmax(cand_vecs @ query_vec.T) else: sim_to_q = cand_vecs @ query_vec.T sim_to_sel = cand_vecs @ cand_vecs[selected[-1]].T score = lamb * sim_to_q - (1-lamb) * sim_to_sel best = idx[np.argmax(score[idx])] selected.append(best) idx.remove(best) return [cand_txts[i] for i in selected]

4. 生成流水线与后处理

# service.py from flask import Flask, request, jsonify from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate template = """ 使用以下上下文回答用户问题。如果不知道，请说“暂无资料”，不要编造。 上下文： {context} 问题：{question} 答案： """ PROMPT = PromptTemplate(template=template, input_variables=["context","question"]) llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.1) qa = RetrievalQA(llm=llm, retriever=..., prompt=PROMPT) def post_filter(answer: str) -> str: """黑名单+正则，过滤口语化风险词""" deny = {"手续费全免", "绝对免费", "零成本"} for d in deny: answer = answer.replace(d, d[:-2]+"以当时政策为准") return answer @app.route("/ask", methods=["POST"]) def ask(): data = request.json q = data["query"] # 1. 缓存命中 if rds.exists(q): vec = pickle.loads(rds.get(q)) else: vec = encoder.encode([q]) rds.setex(q, 3600, pickle.dumps(vec)) # 2. 检索+重排 idxs, _ = search_index(vec, index, k=25) cands = [docs[i] for i in idxs] top5 = rerank(q, cands)[:5] # 3. 生成 ans = qa.run({"context": "\n".join(top5), "question": q}) return jsonify({"answer": post_filter(ans), "sources": top5})