自搭建智能客服系统实战：从零构建高可用对话引擎-平芜编程栈

背景痛点：为什么放着 SaaS 不用，偏要自己撸一套？

去年公司做金融风控，客服每天都要回答大量敏感问题：额度、征信、负债率……数据一旦外泄就是上热搜的节奏。我们先后试过两家头部 SaaS：

数据必须走对方云端，合同里却写着“不可抗力导致泄露不担责”。
想改个“根据用户评级动态回答”的逻辑，报价 30 万起，排期 2 个月。

老板一句话：“自己干，预算 20 万，3 个月上线。”于是就有了今天的自搭建智能客服项目。

技术选型：Rasa、Dialogflow 还是纯自研？

我们拉了一张表，把能下到的开源方案都跑了一遍，核心指标三条：吞吐量、训练数据量、二次开发灵活度。

框架	单机 QPS	所需标注句	中文支持	备注
Dialogflow ES	120	0（官方预训练）	需翻墙	数据出境，直接 pass
Rasa 2.x	350	2k+	社区版分词一般	需要 GPU 才跑得动 Bert
Rasa 3.x + Duckling	420	1k+	同上	实体抽取准，但重
Bert+自研	600+	500 条即可微调	完全可控	代码量大，需要 NLP 背景

最终拍板：用Bert+自研做意图识别，Spacy管实体，FastAPI写微服务，Redis存会话。理由一句话：吞吐量高、数据留在本地、改需求不用求人。

核心实现：三步搭出对话引擎

1. FastAPI 微服务骨架（带 JWT 鉴权）

先搭最外层 API，保证“高可用”不是口号。下面这段代码单文件能跑，依赖只有fastapi[all]和pyjwt。

# main.py import time import jwt from fastapi import FastAPI, Depends, HTTPException from pydantic import BaseModel SECRET = "replace_with_env_var" app = FastAPI(title="chatbot") class User(BaseModel): username: str password: str def create_token(data: dict, expire=3600): data.update({"exp": int(time.time()) + expire}) return jwt.encode(data, SECRET, algorithm="HS256") async def get_user(token: str): try: payload = jwt.decode(token, SECRET, algorithms=["HS256"]) return payload["username"] except jwt.PyJWTError: raise HTTPException(status_code=401, detail="Invalid token") @app.post("/login") async def login(user: User): if user.password != "demo123": # 实际走 DB+哈希 raise HTTPException(status_code=400, detail="Wrong pwd") return {"token": create_token({"username": user.username})} @app.post("/chat") async def chat(utterance: str, user=Depends(get_user)): # TODO: 调用下游 NLU & DM return {"reply": f"Hi {user}, you said: {utterance}"}

本地uvicorn main:app --reload就能调试，后期上 Kubernetes 直接打镜像，零改动。

2. 意图识别模块（Spacy 微调）

训练数据只标 500 句，覆盖“查额度、问还款、转人工”三大意图。格式用 JSONL：

{"text": "我下个月要还多少钱", "label": "ask_repay"} {"text": "额度能不能提高一点", "label": "ask_quota"} {"text": "找客服", "label": "to_human"}

训练脚本（核心 30 行，PEP8 已检查，timeit 看耗时）：

# train_intent.py import time, json, spacy from sklearn.metrics import classification_report def load_data(path): with open(path) as f: return [json.loads(l) for l in f] def train(nlp, train_path, n_iter=5): data = load_data(train_path) textcat = nlp.add_pipe("textcat", last=True) for _, label in data: textcat.add_label(label) optimizer = nlp.initialize() for i in range(n_iter): t0 = time.perf_counter() losses = {} for text, label in data: doc = nlp.make_doc(text) gold = {"cats": {label: 1.0}} nlp.update([doc], [gold], sgd=optimizer, losses=losses) print(f"epoch {i} loss={losses['textcat']:.3f} time={time.perf_counter()-t0:.2f}s") return nlp if __name__ == "__main__": nlp = spacy.blank("zh") nlp = train(nlp, "intent.jsonl") nlp.to_disk("intent_model")

验证结果：500 句训练 + 100 句测试，意图准确率 96.4%，足够产线用。

3. 对话状态机 & Redis 上下文

多轮对话最怕“前言不搭后语”。思路：把每一轮的状态写成 dict，key 是user_id，value 序列化后扔 Redis，TTL 设 15 分钟。

# dm.py import json, redis, time from spacy import load nlp = load("intent_model") r = redis.Redis(host="localhost", decode_responses=True) class DialogManager: def __init__(self, uid): self.uid = uid self.key = f"chat:{uid}" def load(self): data = r.get(self.key) return json.loads(data) if data else {"intent": None, "slots": {}} def save(self, state): r.setex(self.key, 900, json.dumps(state)) # 15 min def reply(self, text): t0 = time.perf_counter() doc = nlp(text) intent = max(doc.cats, key=doc.cats.get) state = self.load() # 简单状态机：intent 驱动 if intent == "ask_quota": answer = "您的额度为 50,000 元。" elif intent == "ask_repay": answer = "下期还款 3,200 元，截止 15 号。" else: answer = "正在为您转人工，请稍候。" state["intent"] = intent self.save(state) print(f"infer_time={time.perf_counter()-t0:.3f}s") # 性能打点 return answer

把DialogManager集成到/chat接口，就完成“能记住上句”的多轮对话。

生产考量：让 demo 像工业品一样稳

1. 压力测试：JMeter 1000 并发怎么调？

环境：4C8G 容器，单副本。
初始：QPS 280，P99 延迟 900 ms，CPU 打满。
优化：
1. uvicorn workers=4改成gunicorn -k uvicorn.workers.UvicornWorker -w 4。
2. Redis 连接池max_connections=50，避免短连接。
3. Spacy 模型常驻内存，去掉每次load。
结果：QPS 提到 680，P99 降到 320 ms，CPU 仍有余量。

2. 安全防护：日志脱敏 + SQL 注入

对话日志最容易泄露手机号、身份证。统一写一层sanitize：

import re def sanitize(text: str) -> str: text = re.sub(r"\d{11}", "", text) text = re.sub(r"\d{17}[\dX]", "🆔", text) return text

入库用 ORM 并且预编译语句，拒绝拼接；FastAPI 的依赖注入也天然防注入。上线前用sqlmap跑一轮，全部 404 才给绿灯。

避坑指南：踩过的坑，帮你先填平

多轮 session 泄漏
开发期把user_id放在 URL 参数里，测试小哥复制链接给同事，结果 A 看到 B 的聊天上下文。教训：user_id必须放进 JWT，从 Header 取，杜绝浏览器地址栏泄露。
中文分词领域词典
金融场景“分期、白条、代偿”这类词，默认分词会拆成“代/偿”。给 Spacy 加一条tokenizer_exceptions：
```
nlp.tokenizer.add_special_case("代偿", [{"ORTH": "代偿"}])
```
意图识别准确率从 92% 提到 96%，就这一行。
Redis 热 key 打垮
促销当天在线 5 万人，同一个key过期瞬间 5k 并发回源，Redis CPU 100%。解决：过期时间加随机 jitter（0–300 s），把尖峰削平。