背景痛点：传统客服的三座大山
过去两年，我先后接手过两套“祖传”客服系统，它们像三座大山一样压在运维和运营身上：
- 并发请求处理靠“排队+人工扩容”，高峰期 CPU 飙到 90%，用户平均等待 8 秒以上，投诉电话比咨询还多。
- 多轮对话维护用 Redis 存 JSON，字段一多就“串台”，用户刚报完手机号，机器人转头又问“请问您的手机号是？”。
- 意图识别靠关键词+正则，新增一个意图要写 50 条规则，准确率从 70% 掉到 55%，维护人员自嘲是“正则缝纫机”。
这三座山直接导致人力成本占整体预算 60%，老板一句“降本增效”，我们就得在 3 个月内把系统翻新。于是有了这次“AI 辅助开发”的完整实践：用 Python 把 NLP、对话管理、服务治理全链路撸一遍，最终把平均响应压到 200 ms，运营成本降了 30%。下面把趟过的坑、攒下的代码、跑出的数据一次性摊开。
技术选型：规则、ML 还是大模型？
先给结论：别迷信单点 SOTA，场景优先、成本优先。
规则引擎（ES、Drools）
适合冷启动+兜底，开发半小时，运行十年，但扩展性≈0，新增意图要人肉堆规则。传统机器学习（FastText、TextCNN）
训练快、资源省，CPU 能跑 1k QPS；缺点是上下文健忘，多轮一多就“前言不搭后语”。深度学习小模型（BERT-base、ERNIE）
准确率 85%→92%，GPU 延迟 80 ms，CPU 延迟 400 ms；显存占用 1.3 GB，适合并发 < 200 的轻量场景。大模型（GPT-3.5、ChatGLM）
零样本泛化无敌，但延迟 1.5 s、单价 0.002$/千询，高并发=高账单；我们用它做“标注员”而非“主服务”。
最终组合策略：
规则兜底 → FastText 做一级路由 → BERT 精排 → GPT-3.5 生成寒暄/复杂回复，把 80% 简单咨询拦截在前两层，剩下 20% 走高精度通道，成本可控。
核心实现：对话状态机 + 异步 API
1. 对话状态管理（State Machine）
用 python-statemachine 把多轮对话拆成 4 个互斥状态：Greeting → Collect → Confirm → Answer，状态迁移由意图+槽位共同决定。
# state_machine.py from statemachine import StateMachine, State from typing import Dict, Optional class DialogSM(StateMachine): greeting = State("Greeting", initial=True) collect = State("Collect") confirm = State("Confirm") answer = State("Answer") # 事件：意图驱动 inquire = greeting.to(collect) | collect.to(confirm) affirm = confirm.to(answer) deny = confirm.to(collect) restart = answer.to(greeting) def __init__(self, uid: str): super().__init__() self.uid: str = uid self.slots: Dict[str, str] = {} def on_enter_collect(self, intent: str, entities: Dict[str, str]): """槽位收集：合并实体""" self.slots.update(entities) def on_enter_confirm(self): """确认前做实体掩码，见避坑章节""" mask_sensitive_slots(self.slots)状态机实例按uid维度存 Redis，TTL 15 min，自动过期=自动清内存。
2. Flask 异步 API
为了把 I/O 等待（NLP 推理、知识库查询）从 200 ms 压到 50 ms 以内，用asyncio + aiohttp包一层，Flask 只负责解析/校验，耗时 < 5 ms。
# app.py import asyncio, time, json from flask import Flask, request from state_machine import DialogSM from typing import Tuple app = Flask(__name__) async def nlp_predict(text: str) -> Tuple[str, dict]: """调用内部 BERT 服务，返回 (intent, entities)""" async with aiohttp.ClientSession() as session: async with session.post( "http://bert-service:8501/v1/models/classify:predict", json={"instances": [text]}, headers-token=gen_jwt() ) as resp: data = await resp.json() return data["intent"], data["entities"] @app.route("/chat", methods=["POST"]) async def chat(): uid = request.json["uid"] query = request.json["query"] sm = await redis_get_sm(uid) or DialogSM(uid) intent, entities = await nlp_predict(query) sm.send(intent, entities=entities) # 触发状态迁移 answer = await generate_answer(sm) await redis_set_sm(uid, sm) # 回写状态 return {"answer": answer, "state": sm.current_state.id}关键点：
- 所有
async def统一走asyncio.run()，避免 Flask 阻塞； - 对外保持 REST，内部走 gRPC/aiohttp，双协议隔离前后端。
3. NLP 服务鉴权 & 限流
内部推理服务用JWT + Redis 令牌桶做限流，桶容量按 GPU 显存折算：
T4 16 GB ≈ 64 并发，桶令牌 64，每秒回充 64，突发可借 30%，超量直接 429。
# limiter.py import redis, time from typing import Optional r = redis.Redis(host="redis", decode_responses=True) def acquire(token_key: str, capacity: int, refill: int) -> Optional[float]: now = time.time() ttl = r.ttl(token_key) or 0 tokens = min(capacity, int(r.get(token_key) or 0) + refill) if tokens < 1: return None # 触发限流 r.decr(token_key, 1) r.expire(token_key, ttl) return now性能考量：QPS、内存、GPU 实测
AB 测试数据（单卡 T4，batch=8）
| 模型 | 设备 | 平均延迟 | QPS | 显存/内存 |
|---|---|---|---|---|
| FastText | CPU | 12 ms | 1200 | 200 MB |
| BERT-base | CPU | 380 ms | 42 | 1.3 GB |
| BERT-base | T4 | 65 ms | 310 | 1.3 GB |
| GPT-3.5-api | – | 1.4 s | 5 | – |
线上把 80% 流量切给 FastText，15% 给 BERT，5% 给 GPT-3.5，综合 QPS 拉到 900+，成本对比纯 GPT 方案下降 70%。
上下文内存优化
- 状态机只存diff 字段，如手机号只存掩码后四位；
- 历史对话按滑动窗口（最近 5 轮）序列化，超窗自动丢弃；
- 对长文本用Sentence-Transformer 抽 384 维向量，占 1.5 KB，比原文节省 90%。
避坑指南：敏感数据、幂等、冷启动
1. 实体掩码
用户常输入“身份证 3625********1234”，要在状态机入库前完成掩码，防止日志落盘泄露。
def mask_sensitive_slots(slots: Dict[str, str]): for k, v in slots.items(): if k in {"idcard", "phone"}: slots[k] = re.sub(r"(\d{4})\d+(\d{4})", r"\1****\2", v)2. 对话超时重试的幂等
前端重试可能 3 次 POST 同一句话，用 uid+msgId 做唯一键，Redis SETNX 防重放：
msg_key = f"dup:{uid}:{msgId}" if not r.set(msg_key, 1, nx=True, ex=60): return {"answer": "", "state": sm.current_state.id, "msg": "duplicate"}3. 模型冷启动预热
Triton Server 拉起后第一次推理要编译 CUDA kernel，延迟飙到 2 s。写个readyz接口，启动脚本里顺序预热：
for text in "你好" "查订单" "再见"; do curl -X POST http://localhost:8501/v1/models/classify:predict \ -d "{\"instances\":[\"$text\"]}" done直到 readyz 返回 200 才注册到 Consul，避免流量灌进来时踩坑。
代码规范小结
- 全项目强制
python>=3.10，类型注解用from __future__ import annotations； - 所有 I/O 函数加
tenacity重试，日志里带exc_info=True； - 性能关键路径（状态机迁移、掩码、限流）打statsd 埋点，方便后期 Prometheus 拉指标。
互动：精度 vs. 延迟，你怎么选？
把 BERT 换成更大的模型，准确率能再涨 3%，但延迟翻倍；砍到 FastText，延迟 10 ms，准确率却掉 8%。线上业务你会怎么平衡？
欢迎留言聊聊你的切分策略。
附赠一份开源对话数据集（CC BY-SA）：
https://github.com/example/CustomerChat-Chinese-110k
拿去压测，记得分享结果。
踩完这些坑，最大的感受是：AI 客服不是“模型即系统”，而是“状态+策略+成本”的三体问题。先把状态机、缓存、幂等、限流这些“工程骨架”搭好，再谈模型精度，否则再准的模型也扛不住凌晨 2 点的流量洪峰。希望这份从架构到上线的避坑笔记，能帮你少熬几个通宵。
