核心内容摘要
图片如何适应容器框
背景痛点消息风暴与状态丢失的双重夹击去年给一家电商客户做企微客服机器人上线首日就踩了两个大坑早上 10 点促销推送瞬间 3 k 用户同时机器人Web 服务直接 502。
用户问完“优惠券怎么用”后紧接着追问“能叠加满减吗”机器人却忘了上一轮给的券码对话从头开始被投诉“人工智障”。
这两个问题本质上是“高并发”与“多轮状态”叠加后的典型症状企微的“接收消息”API 是HTTP 长轮询一次最多拉 20 条默认 30 s 间隔如果处理慢下一轮请求还没回来消息就堆积成山。
多轮对话需要把“槽位”(slot) 存在内存或缓存里但进程重启或水平扩容后状态灰飞烟灭。
传统 if/else 规则引擎在 200 QPS 以内还能扛超过 500 QPS 时 CPU 飙满意图分支一多维护就是灾难。
于是我们把目光转向 AI 辅助开发让模型扛语义让规则扛边界两者互补。
技术选型规则、ML、DL 的三角权衡先给出实测数据4 核 8 G Docker 内网压测单模型单实例方案平均 QPS准确率冷启动时间备注规则引擎
8
780 s分支300 后不可维护传统 MLTF-IDFLR
1
853 min需要人工标注 5 k 条轻量 BERTMiniLM蒸馏
1
9115 s模型 30 MBGPU 可选结论对并发高、意图多、需求迭代快的场景BERT 蒸馏模型在准确率与吞吐量上都更优冷启动成本用“AI 辅助标注”来抹平先用规则跑一周收集日志 → 主动学习 → 人工纠偏 10% 样本就能训出
9 模型。
核心实现三条代码搞定“收、识、答”
异步消息队列——把长轮询变成生产消费企微官方只保证“最多重试 3 次”如果 5 s 内没回 200同一条消息会再次推送极易重复。
我们采用“拉模式”定时器每 2 s 拉一次https://qyapi.weixin.qq.com/cgi-bin/message/get_msg_list把原始 JSON 直接塞进 Redis Streamgrouprobot返回 200解耦业务消费者协程异步处理即使重启也不丢事件。
核心代码Python
11PEP8import httpx, redis, asyncio, json async def pull_job(): r redis.Redis(host
127.
0.
1, decode_responsesTrue) while True: async with httpx.AsyncClient() as cli: res await cli.post( https://qyapi.weixin.qq.com/cgi-bin/message/get_msg_list, params{access_token: await get_token()}, json{limit: 20, seq: r.get(seq) or 0} ) data res.json() seq data[seq] for msg in data[msg_list]: r.xadd(wecom:stream, {payload: json.dumps(msg)}) r.set(seq, seq) await asyncio.sleep(
2)
轻量化意图分类——30 MB 模型跑在 CPU模型结构MiniLM 6 层 全连接 128 → 48 个意图量化 int8。
推理框架用 ONNXRuntime单 CPU 线程 4 ms完全摆脱 GPU 绑定。
训练完把model.onnx与label2id.json放到models/目录推理封装如下import onnxruntime as ort, json, numpy as np from transformers import BertTokenizerFast class IntentEngine: def __init__(self, model_pathmodels/model.onnx): self.sess ort.InferenceSession(model_path) self.tok BertTokenizerFast.from_pretrained(models/) with open(models/label2id.json) as f: self.id2label {int(v): k for k, v in json.load(f).items()} def predict(self, text: str, threshold
0.
: inp self.tok(text, return_tensorsnp, max_length32, truncationTrue) logits, self.sess.run(None, {k: v for k, v in inp.items()}) idx int(np.argmax(logits)) prob float(logits[idx]) return self.id2label[idx] if prob threshold else unknown
对话状态机——带超时重置的槽位填充把“状态”抽象成DialogueState用 Redis Hash 存储keyuser_idttl300 s。
状态机只关心三件事意图、已填槽、最后活跃时间。
超时后自动回到INIT。
import pydantic, time, redis r redis.Redis(decode_responsesTrue) class DialogueState(pydantic.BaseModel): intent: str slots: dict pydantic.Field(default_factorydict) ts: float 0 def load_state(uid: str) - DialogueState: data r.hgetall(fdlg:{uid}) if not data or time.time() - float(data.get(ts,
) 300: return DialogueState() return DialogueState(**data) def save_state(uid: str, state: DialogueState): state.ts time.time() r.hset(fdlg:{uid}, mappingstate.dict()) r.expire(fdlg:{uid},
业务层根据意图调用不同回复模板并更新槽位若意图为unknown则转人工。
性能优化缓存、去重、幂等三板斧缓存模型结果用户问题重复率约 28 %用 Redis 缓存text_hash, intentttl10 min直接砍掉 1/3 的 ONNX 调用。
消息去重企微每条消息有msgid消费者先SETNX msgid 1消息处理完再写ACK保证重启也不重复。
幂等性对“发客服消息”接口用client_msg_iduuiduser_id实现幂等5 分钟内重复调用企微自动丢弃。
代码片段def handle_msg(payload: dict): msg_id payload[msgid] if not r.setnx(fack:{msg_id},
: return # 已处理 uid payload[from][user_id] state load_state(uid) intent intent_eng.predict(payload[text]) # ...业务逻辑... save_state(uid, state)避坑指南token、敏感词与合规access_token 频率限制企微强制 2000 次/小时我们采用单例刷新分布式锁把 token 存在 Rediskeywecom:tokenttl7000 s刷新脚本用SET NX EX 10抢锁防止多实例并发刷新。
敏感词过滤先过本地 DFA 树10 万条
2 ms再调云厂商文本审核 API双保险。
若命中机器人直接回“亲亲换个词试试~”并打日志留痕。
合规性记录user_idquestionanswer7×24 小时加密落盘方便审计对“退款”“投诉”等高危意图强制转人工机器人不直接承诺。
延伸思考飞书/钉钉也能复用企微、飞书、钉钉的“拉模式”大同小异差异主要在签名算法与消息格式。
把核心层意图分类、状态机、缓存抽象成独立包再为每个平台写20 行适配器即可飞书把tenant_access_key换成自建 app 的token消息 ID 字段叫message_id钉钉用topapi/message/get轮询消息体嵌套在content字段。
只要保证“收、识、答”三步接口不变迁移基本一天搞定。
小结让 AI 做 90 % 的体力活整个项目下来最大感受是——别硬写规则让模型先跑。
先用规则收集数据 → 主动学习 → 蒸馏小模型 → 缓存 状态机兜底一套组合拳把并发、状态、合规三个大坑同时填上。
最终机器人上线稳定
5 k QPS平均响应 180 ms准确率 91 %夜里终于不用再盯着告警短信失眠。
如果你也在企微、飞书或钉钉上折腾智能客服不妨把这套模板拿过去改两行相信能少掉不少头发。
祝开发顺利早日让机器人替你值班。