一条消息从微信用户发出到机器人回复中间经过一条完整的处理链路。把这条链路从头到尾走一遍能看清每个环节在做什么。一、消息到达——回调接收用户发消息后接口平台通过 Webhook 把消息推送到程序的回调地址。这一环节的约束5秒内必须返回响应否则平台认为推送失败会重试。所以回调入口只做最轻的动作验证请求合法性、解析 JSON、消息入队、立刻返回成功码。业务逻辑全部异步处理。二、消息入队——削峰和缓冲消息放进队列后由 worker 异步消费。队列的价值是削峰——消息高峰期比如群里突然活跃队列缓冲worker 按自己的节奏处理不会被瞬时流量压垮。队列还保证消息不丢——worker 处理失败时消息回到队列重试。三、消息处理——分发到业务worker 从队列取出消息后走标准处理管线幂等检查msgId 去重→ 身份解析私聊/群聊、是谁、什么标签→ 类型分发文本/图片/系统消息走不同处理器→ 业务逻辑规则匹配、AI 回复、转人工→ 生成回复内容。每一步都可能短路重复消息直接跳过、非目标群消息忽略、无法处理的消息走默认动作。不是每条消息都走完全部管线。四、回复发送——频控和落库回复内容生成后不是立刻发出先过频控检查该用户/该群的发送间隔和日上限、再过内容审核敏感词、占位符完整性通过后调发送接口。发送成功后保存 msgId供撤回和引用记录发送日志。五、状态回流——监控和闭环发送结果回流到监控系统成功率、耗时、错误类型。失败消息按错误码分类处理限频重试、掉线告警。完整链路的日志通过 msgId 串联出问题时能从一条消息追溯到全链路。完整链路对照环节核心动作约束回调接收验证入队返回5秒内响应消息入队缓冲不丢异步消息处理去重→解析→分发→逻辑可能短路回复发送频控→审核→发送落库msgId状态回流监控错误处理闭环完整管线实现# 环节1回调入口 app.post(/webhook) def webhook(): if not verify_signature(request): # 验证合法性 return {code: 403}, 403 msg_q.put(request.json) # 入队 return {code: 1000} # 立刻返回 # 环节23worker处理 def msg_worker(): while True: d msg_q.get() try: process_one(d) except Exception as e: db.save(msg_error, {msgId: d.get(msgId), error: str(e)}) def process_one(d): # 幂等 if db.exists(messages, msgIdd[msgId]): return db.save(messages, d) # 身份解析 ident resolve_identity(d) # 非目标群消息短路 if ident[scene] group and ident[groupId] not in WATCH_GROUPS: return # 类型分发 handler HANDLERS.get(d[messageType], handle_unknown) reply handler(d, ident) if reply: send_reply_safely(d, reply) # 环节4安全发送 def send_reply_safely(d, reply): target d.get(groupId) or d[fromUser] if not rate_limiter.allow(target): # 频控 db.save(pending_sends, {target: target, content: reply}) return if not content_check(reply): # 内容审核 alert(f回复内容未通过审核: {reply[:30]}) return r api.call(sendText, {toUser: target, content: reply}) db.save(replies, {origMsgId: d[msgId], replyMsgId: r[data][msgId], time: now()}) # 环节5定时检查失败消息 def retry_failed(): for item in db.query(pending_sends): if rate_limiter.allow(item[target]): r api.call(sendText, {toUser: item[target], content: item[content]}) if r.is_success: db.delete(pending_sends, item[id])落地建议五个环节按顺序搭建先跑通回调→入队→处理→发送的最小闭环回流监控第二步再加。5秒响应约束和 msgId 幂等是两个不能省的基础设计。队列是稳定性的关键——没有队列高峰期消息处理和回调响应会互相拖垮。回调格式和接口参数参考 Eyun 开发文档。 从API调用到微信机器人运行：个人微信二次开发中一套完整的请求处理流程
一条消息从微信用户发出到机器人回复中间经过一条完整的处理链路。把这条链路从头到尾走一遍能看清每个环节在做什么。一、消息到达——回调接收用户发消息后接口平台通过 Webhook 把消息推送到程序的回调地址。这一环节的约束5秒内必须返回响应否则平台认为推送失败会重试。所以回调入口只做最轻的动作验证请求合法性、解析 JSON、消息入队、立刻返回成功码。业务逻辑全部异步处理。二、消息入队——削峰和缓冲消息放进队列后由 worker 异步消费。队列的价值是削峰——消息高峰期比如群里突然活跃队列缓冲worker 按自己的节奏处理不会被瞬时流量压垮。队列还保证消息不丢——worker 处理失败时消息回到队列重试。三、消息处理——分发到业务worker 从队列取出消息后走标准处理管线幂等检查msgId 去重→ 身份解析私聊/群聊、是谁、什么标签→ 类型分发文本/图片/系统消息走不同处理器→ 业务逻辑规则匹配、AI 回复、转人工→ 生成回复内容。每一步都可能短路重复消息直接跳过、非目标群消息忽略、无法处理的消息走默认动作。不是每条消息都走完全部管线。四、回复发送——频控和落库回复内容生成后不是立刻发出先过频控检查该用户/该群的发送间隔和日上限、再过内容审核敏感词、占位符完整性通过后调发送接口。发送成功后保存 msgId供撤回和引用记录发送日志。五、状态回流——监控和闭环发送结果回流到监控系统成功率、耗时、错误类型。失败消息按错误码分类处理限频重试、掉线告警。完整链路的日志通过 msgId 串联出问题时能从一条消息追溯到全链路。完整链路对照环节核心动作约束回调接收验证入队返回5秒内响应消息入队缓冲不丢异步消息处理去重→解析→分发→逻辑可能短路回复发送频控→审核→发送落库msgId状态回流监控错误处理闭环完整管线实现# 环节1回调入口 app.post(/webhook) def webhook(): if not verify_signature(request): # 验证合法性 return {code: 403}, 403 msg_q.put(request.json) # 入队 return {code: 1000} # 立刻返回 # 环节23worker处理 def msg_worker(): while True: d msg_q.get() try: process_one(d) except Exception as e: db.save(msg_error, {msgId: d.get(msgId), error: str(e)}) def process_one(d): # 幂等 if db.exists(messages, msgIdd[msgId]): return db.save(messages, d) # 身份解析 ident resolve_identity(d) # 非目标群消息短路 if ident[scene] group and ident[groupId] not in WATCH_GROUPS: return # 类型分发 handler HANDLERS.get(d[messageType], handle_unknown) reply handler(d, ident) if reply: send_reply_safely(d, reply) # 环节4安全发送 def send_reply_safely(d, reply): target d.get(groupId) or d[fromUser] if not rate_limiter.allow(target): # 频控 db.save(pending_sends, {target: target, content: reply}) return if not content_check(reply): # 内容审核 alert(f回复内容未通过审核: {reply[:30]}) return r api.call(sendText, {toUser: target, content: reply}) db.save(replies, {origMsgId: d[msgId], replyMsgId: r[data][msgId], time: now()}) # 环节5定时检查失败消息 def retry_failed(): for item in db.query(pending_sends): if rate_limiter.allow(item[target]): r api.call(sendText, {toUser: item[target], content: item[content]}) if r.is_success: db.delete(pending_sends, item[id])落地建议五个环节按顺序搭建先跑通回调→入队→处理→发送的最小闭环回流监控第二步再加。5秒响应约束和 msgId 幂等是两个不能省的基础设计。队列是稳定性的关键——没有队列高峰期消息处理和回调响应会互相拖垮。回调格式和接口参数参考 Eyun 开发文档。