背景痛点：传统客服系统到底卡在哪？
去年我在一家电商公司做后端，客服系统天天被投诉：
- 用户说“我要退货”，系统却理解成“我要兑换”，意图识别准确率不到70%，客服小姐姐人工兜底到崩溃。
- 会话(Session)状态靠MySQL硬扛，用户刷新页面就丢上下文，体验堪比“金鱼记忆”。
- 大促峰值 1 k QPS 时，老系统直接 502，老板在群里疯狂艾特“谁在线？”
痛定思痛，我决定用开源方案重构，目标只有一句话：高可用、高准确、可水平扩展。GitHub 逛了一圈，最终锁定「峰答AI」——中文友好、协议宽松、社区活跃，于是有了这篇从零到生产的踩坑笔记。
技术选型：峰答AI vs. Rasa vs. Dialogflow
| 维度 | 峰答AI(GitHub) | Rasa 开源 | Dialogflow 商用 |
|---|---|---|---|
| 中文预训练 | 内置BERT-wwm-ext，开箱即用 | 需自训，语料收集耗时 | 支持，但免费版QPS低 |
| 私有部署 | 完全离线，数据不出内网 | 同左 | 必须走谷歌云，合规风险高 |
| 二次开发 | Python，协议Apache-2.0，可商用 | 同左 | 黑盒，只能调Webhook |
| 社区资料 | 中文Issue响应快，示例多 | 英文为主，示例偏英文 | 官方文档全，但中文案例少 |
结论：
- 如果团队“英文+数据科学”能力一般，峰答AI最友好。
- 如果未来要卖私有化部署，Apache协议无后顾之忧。
- 于是拍板：以峰答AI为核心，Flask写业务层，Redis管会话，Docker一把梭。
实现细节：30 分钟跑通第一个API
1. 项目骨架
chatbot/ ├─ api/ # Flask REST层 ├─ nlp/ # 峰答AI模型封装 ├─ common/ # 工具函数 ├─ docker-compose.yml # 一键编排 └─ tests/ # Locust压测脚本2. Flask REST API（含JWT鉴权）
# api/app.py from flask import Flask, request, jsonify from flask_jwt_extended import JWTManager, jwt_required, create_access_token from nlp.fengda import FengdaAgent from common.redis_cli import RedisClient import os app = Flask(__name__) app.config["JWT_SECRET_KEY"] = os.getenv("JWT_SECRET") jwt = JWTManager(app) agent = FengdaAgent() redis = RedisClient() @app.route("/login", methods=["POST"]) def login(): """简单示例：仅校验固定秘钥""" token = create_access_token(identity=request.json.get("api_key", "")) return jsonify(access_token=token) @app.route("/chat", methods=["POST"]) @jwt_required() def chat(): user_id = request.json["user_id"] query = request.json["query"] # 幂等性：用msg_id去重 msg_id = request.json.get("msg_id") if redis.already_replied(user_id, msg_id): return jsonify({"reply": redis.get_reply(user_id, msg_id)}) # 调用峰答AI reply = agent.answer(query, context=redis.get_context(user_id)) # 回写Redis redis.save_turn(user_id, query, reply, msg_id, ttl=600) return jsonify({"reply": reply})时间复杂度：
- 意图识别 ≈ O(L) L为句长，BERT线性。
- Redis读写 ≈ O(1)，整体P99 latency 80 ms（单卡CPU）。
3. Redis键设计模式
Key TTL 含义 ------------------------------------------ ctx:{user_id} 600s 当前会话上下文（JSON） reply:{user_id}:{mid} 600s 幂等缓存 freq:{user_id} 60s 接口限流计数# common/redis_cli.py import redis import json class RedisClient: def __init__(self): self.r = redis.Redis(host='redis', port=6379, decode_responses=True) def save_turn(self, uid, q, a, mid, ttl): pipe = self.r.pipeline(transaction=True) pipe.hset(f"ctx:{uid}", mapping={"q": q, "a": a}) pipe.expire(f"ctx:{uid}", ttl) pipe.setnx(f"reply:{uid}:{mid}", a) pipe.expire(f"reply:{uid}:{mid}", ttl) pipe.execute() def get_context(self, uid): return self.r.hgetall(f"ctx:{uid}") def already_replied(self, uid, mid): return self.r.exists(f"reply:{uid}:{mid}")4. 对话状态机（含超时重试）
峰答AI返回结构：{"intent":"EXCHANGE","slots":{"item":"手机"},"confidence":0.92}
业务层再包一层状态机，防止中途插话：
# nlp/state_machine.py from transitions import Machine class DialogState: states = ["IDLE", "AWAIT_ITEM", "AWAIT_REASON", "DONE"] def __init__(self): self.machine = Machine(model=self, states=DialogState.states, initial="IDLE") def step(self, intent, slots): if self.state == "IDLE" and intent == "EXCHANGE": self.to_AWAIT_ITEM() return "请问订单编号？" if self.state == "AWAIT_ITEM" and slots.get("item"): self.to_AWAIT_REASON() return "请问退货原因？" if self.state == "AWAIT_REASON": self.to_DONE() return "已登记，稍后短信通知。" # 超时兜底 return "抱歉，能再描述一次吗？"超时重试：Redis键ttl=600s，前端每轮拉/status接口，若返回"EXPIRED"则自动重置状态机。
生产考量：压测、敏感词、GPU
1. Locust 2000 QPS 实战
# tests/locustfile.py from locust import HttpUser, task, between class ChatUser(HttpUser): wait_time = between(0.5, 2) token = "eyJ0eXAiOiJKV1..." @task def ask(self): self.client.post("/chat", json={ "user_id": "u123", "query": "怎么退货", "msg_id": "m456" }, headers={"Authorization": f"Bearer {self.token}"})启动：locust -f tests/locustfile.py --host=http://api:5000 -u 400 -r 50 --run-time 5m
结果（4 核 8 G，单卡 CPU）：
- RPS ≈ 2100
- P95 latency 120 ms
- 错误率 0.05%（主要是JWT过期）
2. 敏感词过滤：AC自动机
# common/ac.py import ahocorasick class SensitiveFilter: def __init__(self, word_list): self.ac = ahocorasick.Automaton() for w in word_list: self.ac.add_word(w, w) self.ac.make_automaton() def mask(self, text): # O(n+m) m为关键词总长 return self.ac.iter(text)在/chat接口最前端调用，命中则直接返回“亲亲，请注意文明用语哦～”。
3. Docker GPU 避坑
错误示范：docker run --gpus all ...在Compose里无效。
正确姿势：
services: fengda: runtime: nvidia environment: - NVIDIA_VISIBLE_DEVICES=0否则容器里torch.cuda.is_available()永远False，BERT退回到CPU，延迟飙到 600 ms。
避坑指南：中文分词与容器化
中文分词歧义
峰答AI底层用BERT-wwm-ext，对OOV词自带子词，但“南京市长江大桥”仍可能被切成“南京/市长/江大桥”。
解决：- 在
agent.answer()前加一层自定义词典，把公司产品名、活动名全扔进去。 - 词典格式：一行一词，加载到
jieba.load_userdict()，再喂给峰答AI，准确率从 88% → 94%。
- 在
GPU显存占用狂涨
默认batch_size=32，显存 8 G 的卡直接OOM。
调优：- 把
batch_size降到 8，开torch.onnx转模型，显存降到 3 G，吞吐只掉 5%。 - 在
docker-compose.yml里加mem_limit: 6g，防止容器把宿主机卡死。
- 把
代码规范小结
- 全项目
black + isort一把梭，CI自动检查PEP8。 - 关键算法时间复杂度已在注释标注，方便后续Review。
- 所有I/O操作（Redis、MySQL）统一用
asyncio+aioredis，避免阻塞事件循环。
上线效果 & 真实体感
两周内测，意图准确率 94%，平均响应 80 ms，大促 3 k QPS 零宕机。客服同学终于有时间喝口茶，老板也难得在群里发“辛苦了”而不是“谁在线？”——那一刻，感觉头发都长回来一点。
开放讨论：多轮对话的上下文衰减机制怎么设计？
目前我用固定 600 s TTL，但真实场景里：
- 用户聊 30 分钟前订单，上下文仍要保留；
- 用户去洗个澡回来继续聊，历史却要适当“忘记”，防止模型跑偏。
你的做法是什么？
- 按时间指数衰减？
- 按意图重要度加权？
- 还是让模型自己学一个“遗忘门”？
欢迎留言聊聊你的踩坑经验，一起把峰答AI玩得更溜。
