背景痛点：为什么“能跑”≠“能扛”
第一次把智能客服搬到线上时，我信心满满：BERT 微调 92% 准确率，Flask 接口 50 ms 返回，Demo 漂亮得能直接发朋友圈。结果灰度 30 min 后，群里开始刷屏：
- “用户说‘转人工’怎么被识别成‘查账单’？”
- “并发 200 时 CPU 飙到 95%，对话直接串台！”
- “上游 CRM 接口超时 5 s，整个线程池被拖死，内存蹭蹭涨。”
总结下来，新手最容易踩的坑集中在三条：
- 意图识别只看单句，没考虑上下文，导致多轮对话“前言不搭后语”。
- 会话状态放 Python dict，进程一多就互相覆盖，用户 A 看到用户 B 的订单。
- 第三方 API 没有熔断/重试，一次抖动全站“雪崩”。
痛定思痛，我把这 3 年趟过的坑写成一份“从能跑到能扛”的实战笔记，供同样想落地智能客服的你抄作业。
架构对比：规则、模型还是“我全都要”？
| 方案 | 平均延迟 | 准确率 | 维护成本 | 适用场景 |
|---|---|---|---|---|
| 纯规则引擎 | 10 ms | 70 % | 低（正则堆） | 固定问答、政策类 |
| 纯机器学习 | 250 ms | 90 % | 高（标注+重训） | 开放域闲聊 |
| BERT+规则混合 | 60 ms | 93 % | 中（双轨迭代） | 电商、金融、运营商 |
结论：线上 90% 的 query 都是“查订单”“改密码”这类高频意图，用规则 10 ms 搞定；剩下 10% 长尾丢给 BERT，既省 GPU 又保体验。混合架构是“能扛”的第一步。
核心实现：60 行代码跑通“BERT+规则”双轨
项目结构极简，方便二次开发：
text chatbot/ ├─ intent_rule.py # 规则兜底 ├─ intent_bert.py # BERT 微调 ├─ dialog_state.py # 状态机 ├─ session.py # Redis 会话 └─ main.py # Flask 入口 1. 规则引擎：O(1) 查表，10 ms 内返回 python # intent_rule.py import re from typing import Optional RULES = [ (r"转人工|人工客服", "transfer_agent"), (r"订单.*号", "query_order"), (r"改密码|重置密码", "reset_pwd"), ] def rule_predict(text: str) -> Optional[str]: text = text.lower() for pattern, intent in RULES: if re.search(pattern, text): return intent return None 2. BERT 微调：三分类，训练 3 epoch，0.93 F1 python # intent_bert.py from transformers import BertTokenizer, BertForSequenceClassification import torch MODEL_PATH = "bert-base-chinese" NUM_LABELS = 3 # 订单、密码、其他 tokenizer = BertTokenizer.from_pretrained(MODEL_PATH) model = BertForSequenceClassification.from_pretrained("./finetuned") def bert_predict(text: str) -> str: inputs = tokenizer(text, return_tensors="pt", truncation=True, padding=True, max_length=32) with torch.no_grad(): logits = model(**inputs).logits label_id = logits.argmax(-1).item() return ["query_order", "reset_pwd", "other"][label_id] 3. 对话状态机：槽位+轮次 python # dialog_state.py from enum import Enum, auto class State(Enum): INIT = auto() AWAIT_ORDER_NO = auto() CONFIRM_PWD = auto() class DialogState: def __init__(self): self.state = State.INIT self.slots = {} def update(self, intent: str, text: str): if intent == "query_order": self.state = State.AWAIT_ORDER_NO elif intent == "reset_pwd": self.state = State.CONFIRM_PWD # 更多状态迁移略 4. Redis 会话：TTL 30 min，自动续期 python # session.py import redis, json, os r = redis.Redis(host=os.getenv("REDIS_HOST", "127.0.0.1"), decode_responses=True) def get_session(uid: str): key = f"chat:{uid}" data = r.get(key) return json.loads(data) if data else {"state": "INIT", "slots": {}} def set_session(uid: str, data: dict, ttl=1800): key = f"chat:{uid}" r.setex(key, ttl, json.dumps(data)) 5. 主入口：先规则后模型，双轨融合 python # main.py from flask import Flask, request from intent_rule import rule_predict from intent_bert import bert_predict from dialog_state import DialogState from session import get_session, set_session app = Flask(__name__) @app.post("/chat") def chat(): uid = request.json["uid"] text = request.json["text"] sess = get_session(uid) intent = rule_predict(text) or bert_predict(text) ds = DialogState() ds.__dict__ = sess ds.update(intent, text) set_session(uid, ds.__dict__) return {"intent": intent, "state": ds.state.name} Flask 默认单进程，先压测 100 并发验证逻辑正确，再上 gunicorn -k gevent -w 4。 ## 性能优化：把 250 ms 压到 60 ms 的实战数据 1. 线程池对比 本地 8 核 16 G，JMeter 200 并发持续 60 s： | 线程池大小 | 平均 QPS | 99% RT | CPU 峰值 | |------------|----------|--------|----------| | 1 | 42 | 600 ms | 30 % | | 4 | 158 | 120 ms | 65 % | | 8 | 210 | 90 ms | 95 % | | 16 | 205 | 95 ms | 95 % | 结论：4 倍 CPU 核数后收益递减，线上 4 核容器直接设 --workers=4。 2. 冷启动优化 BERT 第一次推理要 3 s 初始化 CUDA 上下文，用户请求直接超时。解决思路： - 预加载：docker CMD 里先跑一条 dummy predict，再 gunicorn。 - 热备份：双容器滚动发布，K8s readinessProbe 检测 /warmup 接口返回 200 后再切流量。 - 模型常驻内存：torch.jit.trace 导出 .pt，推理速度再提 20%。  ## 避坑指南：内存泄漏与敏感词误判 1. 对话超时内存泄漏 现象：上线 3 天后内存上涨 30%，重启回落。排查步骤： - 使用 tracemalloc 快照对比，发现 DialogState 实例未释放。 - 原因：TTL 过期后 Redis 键被删除，但 Flask 进程本地缓存仍保留强引用。 - 解决：把 DialogState 改为 __slots__，并在每次 get_session 后显式 del 旧对象；同时把本地缓存改为 weakref.WeakValueDictionary。 2. 敏感词误判 “我昨天充了 200 块话费”被误判为“充”涉黄。优化策略： - 双层过滤：先白名单（业务关键词）再黑名单，白命中的直接跳过。 - 多模式匹配：flashtext 0.1 ms 完成 1 万条敏感词扫描，比正则快 50 倍。 - 日志回溯：每天拉取误判 Top100，人工加白，7 天后误判率从 2% 降到 0.3%。 ## 代码规范：让后人少骂两句 - 统一 black 格式化，行宽 88。 - 函数复杂度不超过 10，嵌套 if 一律拆 卫语句。 - 关键路径加时间复杂度注释，如： python def rule_predict(text: str) -> Optional[str]: # O(1) 规则条数固定，可视为常数时间 ... - 单元测试覆盖率 85% 以上，CI 用 GitHub Actions，每次 PR 自动跑 200 并发压力测试，防止“手滑”。 ## 延伸思考：跨渠道会话同步怎么玩？ 网页、微信、小程序三端同时咨询时，用户希望“我在网页输入订单号，微信端直接看到结果”。思路： 1. 渠道只负责收发消息，统一 uid=unionid+手机号。 2. 会话存储用 Redis Hash：key=chat:{uid}，field=channel:{wx|web|mini}，value=最新消息 ID。 3. 消息流转通过 MQ（如 Kafka），各渠道订阅自身 topic，保证弱网环境下也能最终一致。 4. 前端长轮询或 WebSocket 拉取，消息顺序以 server 时间戳为准，避免客户端时钟不一致。 留给读者的小作业：把上面的“消息 MQ”换成 Redis Stream，实现一个 500 行以内的最小跨渠道同步 Demo，欢迎 PR 交流。 --- 写完这篇，我把 3 年踩过的坑又回忆了一遍。智能客服这条链路上，意图识别只是冰山一角，真正的战场在并发、容灾、监控和不断变化的业务规则。希望这份笔记能让你少熬一次通宵，多留一点头发。如果还有疑问，评论区见，一起把“能扛”进行到底。 [](https://t.csdnimg.cn/l0Z1) ---
