上个月老板把我叫进办公室,甩过来一句"咱们做个微信机器人吧,能自动回消息那种,简单吧?"
我当时心里想,不就是个机器人嘛,调几个接口的事。结果一上手才发现,坑多得能栽跟头栽到怀疑人生。第一版我闷头写了一周,消息收发都还不稳定,老板天天问"好了没",我天天回"快了快了"。
后来实在扛不住,重新梳理了思路,把整个流程拆成五步,按步骤一步步来,三天就跑通了。今天就把这五步和我踩过的坑一起写出来,给同样在做的朋友省点时间。
先说个大原则:别一上来就写代码。我第一版就是打开编辑器就开干,写到一半发现思路不对,又推翻重来。后来学乖了,先把五步在纸上列出来,每步要干啥、依赖啥、可能踩啥坑,想清楚了再动手,效率高了一大截。
第一步:注册实例,拿到wId和Token
这是最基础的一步,但也别小看。我一开始就是没搞清楚实例和账号的关系,折腾了半天。
具体操作:在 Eyun平台 上创建一个微信实例,绑定你要做成机器人的微信号。创建完会拿到两个关键凭证:wId(实例ID)和Token(访问令牌)。这俩东西后面所有接口调用都要用,相当于你和微信之间的通行证。
踩的坑:
一开始没把Token存好,硬编码在代码里,后来要轮换的时候满世界找,差点漏改一个地方导致线上挂掉。后来统一放配置中心,用环境变量注入,再没出过这事。
wId和Token的对应关系搞混了,多个实例的时候传错了凭证,消息发到了另一个号上,客户那边一脸懵。建议加个实例配置表,别靠脑子记。
Token过期没自动刷新,跑着跑着突然鉴权失败,全线崩了。后来加了定时刷新,提前十分钟续期,并做了刷新失败的重试和告警,再没因为这个挂过。
第二步:配置回调,5秒内必须响应
微信消息进来靠的是回调(Webhook)。你的服务得有个公网能访问的接口,微信那边有消息就POST过来。
这一步的核心约束是:5秒内必须返回响应,否则微信会认为你处理失败,触发重试。重试一来,同一条消息你处理两遍,回复两条一样的内容,用户体验直接崩。
踩的坑:
最开始在回调接口里直接处理业务逻辑(查数据库、调AI),一个请求要跑三四秒,时不时超时。后来改成收到消息先落库,立刻返回success,再异步处理,才稳住。这个改造是整个项目最关键的一步,没改之前天天被超时折磨。
本地开发没有公网IP,回调配不上。一开始用内网穿透工具凑合,后来上线换了正经的域名加HTTPS,才算靠谱。内网穿透工具不稳定,断线重连一波消息就丢了,本地调试还行,千万别拿到线上用。
签名校验写错了,以为是token不对,折腾了半天才发现是签名算法的实现问题。这种低级错误最磨人,建议直接用平台提供的SDK里的校验函数,别自己手写。
重试机制没处理好,同一条消息处理了三遍,用户收到三条一样的回复。后来加了消息去重,用消息ID做幂等,同一条消息不管来几次只处理一次。
第三步:消息处理引擎:接收→分类→路由→回复
这是整个机器人的核心。消息进来之后,不是直接回,而是要走一套处理流程:接收→分类→路由→回复。
我一开始图省事,写了个大if-else,所有消息都在一个函数里处理。结果消息类型一多,代码乱成一团,加个新功能要改半天,还容易把旧功能改坏。后来重构成管道模式,每个处理环节独立,才清爽下来。
这里多说一句分类器的实现。分类器要干的事就是看消息体里有没有特定字段,判断这是文本、图片还是语音。听起来简单,但有个坑:有些消息是混合类型,比如用户转发一条带图的链接,既有文本又有图还有链接。我的处理是按优先级判断——链接优先级最高,其次是图片,最后是文本。具体优先级怎么排,得根据你的业务场景来,没有标准答案。
关于消息处理的完整流程和字段说明,可以对照 Eyun开发文档 里的消息收发章节,每个字段都标得挺清楚,省得自己猜来猜去。
消息处理引擎的核心逻辑大概长这样:
class MessageEngine: def __init__(self): self.classifier = MsgClassifier() # 分类器:判断消息类型 self.router = MsgRouter() # 路由器:分发到对应处理器 self.replier = MsgReplier() # 回复器:组装并发送回复 def handle(self, raw_msg): # 1. 接收:解析原始消息 msg = self.parse(raw_msg) if not msg: return None # 2. 分类:判断是文本/图片/语音/链接 msg_type = self.classifier.classify(msg) # 3. 路由:根据类型和内容分发到对应处理器 handler = self.router.get_handler(msg_type, msg) if not handler: return self.replier.reply(msg, "暂不支持此类消息") # 4. 处理+回复:处理器返回结果,回复器发送 try: result = handler.process(msg) return self.replier.reply(msg, result) except Exception as e: # 兜底:出错别让用户看到报错,给个友好提示 log.error(f"处理失败: {e}", exc_info=True) return self.replier.reply(msg, "稍等,我查一下")这段代码的关键就一点:每个环节独立,可替换可扩展。今天接文本处理,明天要加图片,后天要换AI模型,都不用动主流程,只改对应的处理器就行。
第四步:接入AI:让机器人"听得懂人话"
前三步做完,机器人已经能收发消息了,但回的还是写死的规则。要让它"智能",得接大模型API。
我接的是通用的大模型接口,让AI干两件事:一是理解用户意图(用户到底想问啥),二是生成回复(机器人该说啥)。
踩的坑:
一开始把所有消息都丢给AI处理,结果"你好""在吗"这种简单消息也要等AI回,延迟两三秒,用户觉得卡。后来加了层关键词预过滤,高频简单消息直接走规则秒回,复杂问题才丢给AI。
AI偶尔会胡说八道(幻觉),有次用户问价格,AI编了个不存在的套餐,客户投诉过来才知道。后来加了回复校验,涉及价格、库存这类敏感信息,AI生成后必须过一遍校验才发出去。
prompt没调好,AI回复又臭又长,用户没耐心看。后来限定回复字数,要求口语化短句,体验才好起来。
AI怎么接、prompt怎么调,每个人的业务不一样,没有标准答案。我的经验是多试多对比,把真实用户的对话记录拉出来,看AI哪里回得不好,针对性改prompt,比闷头调参数有效。
还有一个容易忽略的点:AI的响应时间不稳定。大部分时候两秒内出结果,偶尔要等五六秒甚至超时。用户等不了那么久,所以我加了个超时降级——AI三秒没回就先给个"正在为您查询,请稍等",后台继续等AI结果,出来了再补发一条。这样用户至少知道机器人在干活,不是卡死了。
第五步:上线监控:日志+告警+降级
机器人跑起来不算完,能不能稳定跑才是关键。这一步很多人忽略,等到线上出事才补,代价很大。
我做了三件事:
日志:每条消息的处理过程全记录,包括分类结果、路由去向、AI响应、最终回复。出问题能回溯。
告警:关键指标设阈值——回调超时率、AI响应时间、消息处理成功率。超阈值立刻报警,别等用户投诉。
降级:AI挂了不能整个机器人挂。AI不可用时自动降级到规则回复,虽然没那么智能,但至少能兜住不冷场。这块我在上线第一周就遇到了AI服务波动,幸亏提前做了降级,不然那天机器人直接哑火,客户体验会很差。
补充一个教训:告警别设太敏感。我一开始把阈值设得很紧,结果半夜被报警叫醒三四次,过去一看全是正常波动。后来调松了阈值,加了波动平滑(连续三次超阈值才报警),才消停。告警太频繁等于没有告警,人麻了就不管了。
关于监控指标怎么选、告警怎么配,这块说实话得结合自己的业务来,没有通用方案。我的经验是先从最关键的几个指标开始,别一上来就想监控一切,反而啥都看不清。可以参考 Eyun平台 上别人跑通的项目配置,看看人家盯哪些指标,比自己瞎摸索强。
搭建周期表
步骤 | 我实际耗时 | 难点在哪 |
|---|---|---|
注册实例 | 0.5天 | 搞清凭证管理和多实例对应 |
配置回调 | 1天 | 5秒响应约束+异步处理改造 |
消息处理引擎 | 1天 | 架构设计,别写成一坨if-else |
接入AI | 0.5天 | prompt调优+幻觉控制 |
上线监控 | 0.5天 | 指标选取和告警阈值调参 |
合计 | 约3.5天 |
这个时间是我重构之后的数据。第一版没规划好,零零碎碎搞了一周还没成型。所以动手之前先想清楚架构,磨刀不误砍柴工这话真不是白说的。
结尾:3天跑通的核心是别重复造轮子
回头看,第一版慢,不是因为我能力不行,是因为我在重复造轮子——消息收发自己封装、签名校验自己写、回调重试自己搞,这些底层的东西平台都提供现成的,我却从头写了一遍。
后来想通了,这些通用的部分直接用平台的封装,自己的精力全放在业务逻辑上:消息怎么分类、AI怎么接、回复怎么优化。这才是机器人真正有差异化的地方。
最后给个建议:做之前先把整体流程画出来,哪步用现成的、哪步要自己写,心里有数再动手。我第一版就是边写边想,写到一半发现路走错了,推翻重来,白白浪费好几天。
再补充几个实操小经验:
先用小号测试,别拿主号直接上,万一被封号哭都来不及。测试通过再上正式号。
消息回复加随机延迟,别秒回。秒回太明显像机器人,加个1-3秒的随机延迟,更像真人操作。
做好频率控制,别一口气发太多消息,容易触发风控。我一般控制在每分钟不超过20条,安全边际留足。
希望这篇能帮到正在做的朋友,有具体问题欢迎评论区聊。