背景痛点:传统客服为什么总把用户逼疯
过去两年,维护公司官网客服系统时,我踩过最深的坑就是「规则引擎」。
- 关键词匹配:用户一句「我付不了款」被拆成「付」「款」两个词,结果机器人回复「请问您是要付款还是要退款?」——答非所问。
- 多轮对话失忆:用户先问「优惠券怎么用」,隔两句又问「可以叠加吗」,系统直接重启话题,用户原地爆炸。
- 响应延迟:为了兜底,规则调用链里塞了 7 个正则+3 个外部接口,平均 RT 1.8 s,高峰期 3 s 起步,转化率肉眼可见往下掉。
简单 NLP 模型(意图分类+槽位填充)稍好一点,但中文口语化表达一多就翻车,维护同义词表比写业务代码还累。
痛点总结:响应慢、理解差、维护重。于是我们把目光投向大模型,最终选了腾讯混元。
技术选型:混元 vs 其他大模型
我们内部拉了 3 个维度打分(5 分制,100 并发压测,同 8G 显存 P40 实例):
| 维度 | 混元 | 某开源 6B | 某厂 10B |
|---|---|---|---|
| 首 token 延迟 | 4 | 3 | 2 |
| 中文口语理解 | 5 | 3 | 4 |
| 多轮一致性 | 5 | 3 | 3 |
| 工具调用 | 4 | 2 | 4 |
| 综合得分 | 4.5 | 2.8 | 3.3 |
混元在中文语料上确实更「接地气」,比如「我的券被吞了」能直接映射到「优惠券未到账」意图;另外官方提供「流式+非流式」双接口,方便我们在吞吐和延迟之间来回横跳。
最终拍板:就用混元,但得自己包一层,别让业务代码裸调 API。
核心实现:Python 封装与对话状态机
1. 轻量级 SDK 封装
安装依赖:
pip install aiohttp tenacity tiktoken核心代码(hunyuan_client.py):
import aiohttp, json, time, os from tenacity import retry, stop_after_attempt, wait_random from typing import List, Dict class HunYuanClient: def __init__(self, app_id, secret_id, secret_key): self.app_id = app_id self.secret_id = secret_id self.secret_key = secret_key self.host = "hunyuan.tencentcloudapi.com" @retry(stop=stop_after_attempt(3), wait=wait_random(min=1, max=3)) async def chat(self, messages: List[Dict[str, str]], stream=False) -> str: """非流式对话,返回完整回复""" body = { "AppId": self.app_id, "SecretId": self.secret_id, "Timestamp": int(time.time()), "Messages": messages, "Stream": stream } # 省略签名算法,官方文档有示例 headers = self._build_sign_header(body) async with aiohttp.ClientSession() as session: async with session.post(f"https://{self.host}/v1/chat", json=body, headers=headers) as resp: if resp.status != 200: raise RuntimeError(f"HY error:{resp.status}") data = await resp.json() return data["Response"]["Reply"] def _build_sign_header(self, body): # 按腾讯云签名 v3 实现 ... return {"Authorization": "...", "Content-Type": "application/json"}亮点:
- 用
tenacity做指数退避重试,网络抖动时自动补偿。 - 所有异常统一抛
RuntimeError,方便外层捕获发告警。
2. 对话状态机
状态机只干三件事:
- 记录「当前意图」
- 缓存「已提供槽位」
- 决定「下一步动作」
代码(dialogue_state.py):
from dataclasses import dataclass, field from typing import Optional @dataclass class DialogueState: uid: str intent: str = "" slots: Dict[str, str] = field(default_factory=dict) history: List[Dict] = field(default_factory=list) def add_user_msg(self, text: str): self.history.append({"role": "user", "content": text}) def add_bot_msg(self, text: str): self.history.append({"role": "assistant", "content": text}) def to_messages(self, max=10) -> List[Dict]: # 只取最近�后 max 輪,防止 token 爆炸 return self.history[-max:]在路由层(chat_router.py)里,每次用户发消息:
state = await redis.get(f"chat:{uid}") or DialogueState(uid=uid) reply = await hy_client.chat(state.to_messages() + [{"role": "user", "content": text}]) state.add_user_msg(text) state.add_bot_msg(reply) await redis.set(f"chat:{uid}", state, ex=600) # 10 min 过期这样即使用户刷新页面,上下文也能续上。
性能优化:异步 + 连接池
1. 异步 IO
上面 SDK 已经全是async/await,但压测时发现 QPS 卡在 120 上不去,原因是每次chat()都新建ClientSession。
改成长连接池版本:
connector = aiohttp.TCPConnector(limit=200, limit_per_host=50) session = aiohttp.ClientSession(connector=connector)全局复用同一个session,QPS 直接飙到 420,平均 RT 从 850 ms 降到 480 ms。
2. 压测数据对比
| 场景 | 同步版 | 异步池化版 |
|---|---|---|
| 峰值 QPS | 120 | 420 |
| 平均 RT | 850 ms | 480 ms |
| CPU 占用 | 42% | 68% |
| 内存 | 1.1 GB | 1.3 GB |
结论:I/O 密集型场景,异步+连接池是刚需,CPU 还有余量就能继续加机器水平扩容。
安全实践:敏感词与令牌轮换
1. 敏感信息过滤
大模型偶尔会「口无遮拦」,我们采用「双层过滤」:
- 输入层:正则+DFA 树,100+ 敏感词模式,命中直接返回「亲亲,这个问题小客服无法回答哦」。
- 输出层:调用腾讯云内容安全 API,对 reply 再扫一遍,置信度 >0.8 自动替换为「*」。
代码片段:
async def safe_reply(reply: str) -> str: if sensitive_dfa.hit(reply): return "亲亲,这个问题小客服无法回答哦" check = await tms_client.text_moderation(reply) if check.suggest != "Pass": return "*" return reply2. 访问令牌轮换
混元使用临时签名,有效时间 5 分钟。
- 网关层每 4 分钟主动刷新一次 SecretKey 缓存,防止时钟漂移。
- 业务容器只读缓存,不落地磁盘,降低泄露风险。
- 采用子账号+最小权限,只给「hunyuan:Chat」一个 action,误用范围可控。
避坑指南:冷启动与突发流量
1. 冷启动预热
大模型容器首次调用往往要 2~3 s 拉权重,我们写了个「热身脚本」:
服务启动后先并发打 10 条假请求,把 GPU 显存占满,再注册到注册中心。
上线后 P99 首包降到 600 ms,用户无感知。
2. 突发流量峰值
去年双 11,凌晨 0 点 10 分流量直接 5 倍,混元返回 429。
应对策略:
- 前端按钮 1 秒内禁止重复点击,削掉 40% 重复请求。
- 网关层做令牌桶,单用户 3 次/秒,富余令牌放进「共享池」给新用户。
- 下游再失败就降级到「精简规则引擎」,至少能回答「发货时间」「退款政策」等高频问题。
最终保证核心接口错误率 <0.3%,用户侧基本无感。
完整落地时间线
- 一周:SDK 封装 + 状态机
- 三天:异步改造 + 压测
- 两天:安全过滤 + 轮换脚本
- 一周:灰度 5% → 30% → 100%
- 持续:每日 review bad case,微调 prompt
上线一个月,机器人独立解决率从 58% 提到 82%,人工会话量下降 40%,客服同学终于能准点下班。
还没完:成本与延迟怎么平衡?
混元按 token 计费,我们把平均会话长度从 600 token 压到 380 token,每月账单仍上涨 35%。
如果继续砍长度,多轮一致性又会掉血。
开放问题:
在「用户体验」「响应延迟」「钱包厚度」三者之间,你更愿意牺牲哪一个?或者,有什么压缩 prompt、缓存向量、模型蒸馏的奇技淫巧,能在不降智商的前提下再砍一半成本?欢迎留言一起拆招。