公司动态
个人微信二次开发:3种方案轻松实现AI机器人接入微信
上个月帮一个朋友做了个AI聊天机器人本地跑得挺溜答问也还算靠谱做个demo、演示演示都没问题。结果上线没两天客户一句话给我整不会了这玩意儿能不能直接接到微信上我们客户都在微信群里问问题懒得再开个网页。我当时的第一反应是——微信又不开放机器人API这咋整总不能让客户去申请公众号、再走认证流程吧那周期太长。硬着头皮去查资料翻了不少帖子发现还真有路子而且不止一条。折腾了大半个月把三种方案都跑了一遍这里把踩的坑和最后选的方案写出来给后来人省点时间。先说结论没有最优方案只有最合适的方案。下面三种各自有各自的命门看你项目卡在哪个约束上。我最后选的不是听起来最高级的那个而是最适合朋友那个项目体量的。方案一轮询模式定时拉消息原理最笨但最容易理解的玩法。写个定时任务每隔几秒调一次消息接口看有没有新消息进来。有就取回来交给AI处理处理完再调发送接口把回复推回去。我一开始就是用这个方案跑的因为不需要公网IP本地起个服务就能测特别适合刚上手验证想法。第一版我甚至直接在自家电脑上跑插着网线、挂着脚本就出门吃饭了回来一看消息都回上了当时还挺有成就感。优点实现极简一个循环加个sleep就搞定不需要公网IP和域名本地就能跑调试方便出问题一眼能看出来加个日志啥都清楚缺点延迟大我设的5秒轮询用户发完消息平均要等5秒才回体验拉胯群里有人还以为机器人掉线了频繁拉取容易触发限流有几次直接被接口封了一阵那阵子消息全堵着没消息也在拉纯属浪费资源服务器和带宽都在空转适用场景个人玩具项目、内部小工具、消息量很低的场景。我后来给一个内部反馈群用过一天也就几十条消息轮询完全够用也懒得折腾服务器。方案二Webhook回调模式微信主动推原理这个就反过来了。你在公网部署一个服务把URL配置到那边用户一发消息微信会主动POST到你这个地址你接到请求立刻处理、立刻回。整个过程是事件驱动的没有轮询的空转有点像点外卖和定时去厨房看的区别。接入的时候我参考了 Eyun开发文档 里关于回调验签和消息加解密的部分少走了点弯路。这块文档讲得挺细尤其是签名校验那个地方我第一版写错了字段顺序调了一晚上才发现问题文档里其实写得很明白是我自己没细看。优点实时性好用户发消息基本秒回体验上一个档次不浪费资源有消息才处理没消息服务器就闲着是官方支持的玩法稳定性有保障不用怕哪天突然封了缺点必须有公网IP和域名本地调试得用内网穿透穿透工具有时不稳调试时被坑过几次服务必须能被外网访问安全性要额外注意端口别乱开要求5秒内返回AI处理慢了直接超时用户那边显示服务异常适用场景正式上线的产品、对响应速度有要求的项目。后来正式环境我们就用这个配合 Eyun平台 的消息通道整体跑下来挺稳没出过什么大幺蛾子。方案三长连接模式WebSocket保持连接原理这种方式不是直接对接微信官方而是借助一些第三方协议或者企业微信的能力跟服务端保持一条长连接。消息来了解析、处理、回写都在同一条连接上完成不用反复建连。说实话这个我研究的时间最长因为涉及到的协议细节多而且坑也藏得深——光是断线重连的退避策略我就调了两三天。但跑通之后体验是真的好延迟低、资源占用小高并发下也不容易崩。优点实时性最好几乎零延迟发出去对方那边立刻有反应资源消耗低一条连接搞定收发不用频繁握手适合高并发场景扛得住消息洪峰缺点实现复杂连接管理、断线重连、心跳保活都要自己写代码量大依赖第三方能力稳定性看对方脸色对方一抖你跟着抖调试难度高出问题排查费劲连接状态不好看适用场景消息量大、对实时性要求极高的场景。一般小项目用不上我也就是为了对比才跑了一遍跑完就把代码封存了维护成本扛不住没必要。三种方案横向对比光说不直观我做了张表把几个关键维度摆一起看维度轮询模式Webhook回调长连接延迟高秒级低毫秒级极低资源消耗高空转多低低实现难度简单中等复杂稳定性一般高看依赖公网要求不需要需要看实现适用规模小中大大我自己最后的结论是验证想法用轮询正式上线用Webhook长连接留给对极致实时有要求的场景。大部分中小项目Webhook足够了没必要上来就搞复杂的越复杂的东西越容易出问题。一段能跑的Webhook对接代码下面这段是我最后定稿用的Webhook接收AI处理回复的完整流程Python写的去掉了业务相关的部分保留主干逻辑。新手拿来改改就能跑别被网上那些东拼西凑的教程绕晕。from flask import Flask, request import hashlib import requests app Flask(__name__) TOKEN 你的token AI_URL 替换成你自己的AI服务地址 def verify_signature(signature, timestamp, nonce): # 回调签名校验校验不过的直接拒 params sorted([TOKEN, timestamp, nonce]) sign hashlib.sha1(.join(params).encode()).hexdigest() return sign signature def call_ai(user_msg): # 调AI接口生成回复这里换成你自己的机器人 resp requests.post( AI_URL, json{message: user_msg}, timeout4 ) return resp.json().get(reply, 我这边没想明白稍等转人工) app.route(/wx/callback, methods[GET, POST]) def callback(): if request.method GET: # 首次配置URL时的验证 return request.args.get(echostr, ) # POST是真实消息推送 signature request.args.get(msg_signature, ) timestamp request.args.get(timestamp, ) nonce request.args.get(nonce, ) if not verify_signature(signature, timestamp, nonce): return fail # 这里简化了XML解析实际用XML库解析 user_msg extract_text(request.data) reply call_ai(user_msg) # 组装回复要求5秒内返回 return build_reply_xml(reply) if __name__ __main__: app.run(host0.0.0.0, port80)几个踩坑提醒都是血泪call_ai的 timeout 一定要设而且要小于5秒。我有一次AI接口卡了8秒那边直接报超时用户收不到回复来回投诉。签名校验那段字段顺序不能错sorted是关键。第一版我手写顺序写反了调到半夜才反应过来。生产环境务必加业务鉴权和频控光靠平台那层验签不够。有人拿你的回调地址刷服务器扛不住。首次配置URL那个GET验证很多人漏掉没它配置根本过不去别问我怎么知道的。XML解析别用字符串切割老老实实上XML库否则遇到转义字符就崩。选型建议再啰嗦一句选型。如果你的情况符合下面任一条直接上Webhook有公网服务器或者能搞到内网穿透用户量超过几百对响应速度有要求需要稳定长期运行不想半夜被叫起来重启只有一种情况建议用轮询就是你自己玩、客户群就几个人、不想折腾服务器。长连接除非有明确的实时性诉求否则不建议上来就用光维护成本就够喝一壶。我朋友的机器人最后就是Webhook方案上的线又把 Eyun开发文档 里讲的消息加解密规范过了一遍做加固跑了一个月没出大问题。整体来说方案不复杂难的是把细节抠明白文档读细点能省一半调试时间。希望这篇能帮到正在折腾对接的朋友少踩点我踩过的坑。有问题评论区聊能答的我都会回。