背景痛点：传统客服系统“三座大山”
压得人喘不过气
去年我在一家做 SaaS 的小公司接手客服系统，老架构用 MySQL 硬扛会话日志，意图识别靠关键词if-else，高峰期一上量就集体“社死”。总结下来，三座大山必须搬掉：
- 会话持久化：HTTP 短连接 + 无状态，刷新页面就“失忆”，用户反复描述问题，体验感直线下降。
- 并发处理：同步阻塞 + 单实例，QPS 刚过 50，CPU 就飙到 90%，客服同事只能排队“人工复读”。
- NLU 精度：关键词匹配对同义词、口语化表达基本抓瞎，意图识别准确率 60% 徘徊，老板天天在群里甩截图。
不重构就要背锅，于是决定用 Python 生态撸一套“小而美”的智能客服，目标只有三句话：高并发、低延迟、能看懂人话。
技术选型：为什么放弃 Rasa、Dialogflow，自己造轮子
- Rasa：功能全，但依赖重、镜像 3 GB+，小团队服务器只有 4 核 8 G，跑起来像开飞机拉牛车。
- Dialogflow：按请求计费，月活 10 w 次就要上千块，预算被财务一票否决；而且数据出境，敏感行业直接劝退。
- 自建方案：Python + Transformer + 规则引擎，模型只有 110 M，Docker 镜像 600 MB；规则兜底，标注数据少也能跑；最重要的是——完全免费、可离线、可魔改。
于是拍板：用 BERT 做意图分类，正则 + 规则做实体抽取，Redis 管会话状态，Flask 当网关，轻量又能打。
核心实现：四步搭好对话引擎
1. 整体架构
┌---------┐ ┌---------┐ ┌---------┐ 用户 -->│ Flask API │-->│ 意图/实体 │-->│ 状态机 │--> 回复 └---------┘ └---------┘ └---------┘2. Flask 网关层：异步 + Gunicorn
# app.py from flask import Flask, request, jsonify import asyncio from gevent.pywsgi import WSGIServer app = Flask(__name__) @app.post("/chat") def chat(): user_id = request.json["user_id"] query = request.json["query"] # 异步推理 loop = asyncio.new_event_loop() ans = loop.run_until_complete(dialogue_manager.async_reply(user_id, query)) return jsonify(ans) if __name__ == "__main__": # 生产用 gevent，4 worker 压测 800 QPS 延迟 250 ms WSGIServer(("0.0.0.0", 8000), app).serve_forever()3. BERT 微调：30 分钟搞定意图分类
训练数据格式：text \t intent_label
“我想查订单” query_order “物流走到哪了” query_logistics ...预处理代码：
# preprocess.py import pandas as pd from sklearn.model_selection import train_test_split from transformers import BertTokenizer tokenizer = BertTokenizer.from_pretrained("bert-base-chinese") df = pd.read_csv("raw.csv", sep="\t", names=["text", "intent"]) train, test = train_test_split(df, test_size=0.1, random_state=42) def encode(texts): return tokenizer(texts.tolist(), padding=True, truncation=True, max_length=32, return_tensors="pt") train_enc = encode(train["text"]) test_enc = encode(test["text"]) torch.save({"input_ids": train_enc["input_ids"], "attention_mask": train_enc["attention_mask"], "labels": train["intent"].values}, "train.pt")微调脚本（关键参数已注释）：
# train_intent.py from transformers import BertForSequenceClassification, Trainer, TrainingArguments model = BertForSequenceClassification.from_pretrained( "bert-base-chinese", num_labels=len(intent2id)) args = TrainingArguments( output_dir="./intent_model", per_device_train_batch_size=64, num_train_epochs=3, learning_rate=3e-5, logging_steps=50, save_total_limit=2, load_best_model_at_end=True, metric_for_best_model="accuracy") trainer = Trainer(model=model, args=args, train_dataset=train_ds, eval_dataset=test_ds) trainer.train()训练完把intent_model推到目录，推理侧直接torch.load缓存到内存，单条 30 ms 内完成。
4. Redis 状态机：TTL + 持久化双保险
# state.py import redis, json, time r = redis.Redis(host="127.0.0.1", decode_responses=True) class DialogueState: def __init__(self, user_id, ttl=600): self.key = f"ds:{user_id}" self.ttl = ttl def get(self): data = r.get(self.key) return json.loads(data) if data else {"hist": [], "slots": {}} def update(self, **kwargs): pipe = r.pipeline() pipe.set(self.key, json.dumps(kwargs)) pipe.expire(self.key, self.ttl) pipe.execute()多轮对话场景举例：用户说“帮我订一张票”，状态机记录slots={"dest": None}；再问“从北京到杭州”，实体抽取把dest补全，即可走订票接口。
性能优化：300 ms 不是玄学
- 异步 IO：推理用
asyncio.to_thread把模型丢给线程池，防止 Flask 阻塞；Gunicorn 配geventworker，压测 800 QPS 平均 RT 250 ms。 - 模型服务化：把 BERT 放到独立
torchserve容器，Flask 通过 gRPC 调用，升级模型无需重启网关。 - 缓存机制：意图结果按文本哈希缓存 60 s，热门问题直接命中，QPS 降低 35%。
避坑指南：别让 Demo 上线就翻船
- 对话上下文丢失：页面刷新带
user_id重新生成？前端必须落库user_id到 localStorage，后端对空状态机做“兜底提示”。 - 敏感词过滤：先过正则黑名单，再过 BERT 白名单模型，双层防护；运营可在后台秒级热更新正则，无需发版。
- 日志隐私：返回前把手机号、地址用正则脱敏，再落盘，防止 GDPR/网安审计踩雷。
requirements.txt（实测无冲突）
Flask==2.3.3 transformers==4.30.2 torch==2.0.1 redis==4.5.5 gevent==23.7.0 gunicorn==21.2.0 pandas==2.0.3 scikit-learn==1.3.0延伸思考：标注数据只有 2 k 条，NER 怎么救？
- 远程监督：用外部知识库做弱标注，把订单号、手机号正则结果直接当实体标签，再人工 10% 抽检矫正。
- 数据增强：同义词替换 + 随机 Mask，生成 5 倍样本；对中文可用
opencc简繁切换再回译，实体边界不变。 - 迁移 + 主动学习：先用
bert-base-chinese跑 80% 高置信样本，再挑模型最懵的 200 条让人标，一轮就能涨 6~7 个百分点。
写在最后
整套代码从 0 到上线只花了三周，白天写业务、晚上调模型，踩了无数个“以为能复现论文”的坑。现在系统每天扛 5 万次调用，意图准确率稳在 92%，平均响应 280 ms，客服小姐姐终于有时间摸鱼……哦不，专注高价值客户。希望这份笔记能帮你少掉几根头发，早日把智能客服搬上生产线。
