简介一份基于Flask与ChatGPT的微信公众号聊天机器人后端工程面向已有Python基础、希望快速搭建公众号智能对话服务的开发者也适合用于学习Flask接口开发、ChatGPT API接入和Docker容器化部署。压缩包内含可运行的完整源码和部署文档按说明配置Python 3.7及以上环境即可启动对新手比较友好。资源共15个文件以Python脚本、Markdown部署文档、Docker配置Dockerfile与docker-compose.yml及依赖锁定文件为主压缩后仅52KB体积轻巧、目录结构清晰便于阅读和二次开发。已有143人学习下载。项目包含应用入口、聊天核心逻辑、日志记录、系统部署说明及工作流配置等内容可帮助读者梳理公众号消息处理链路掌握ChatGPT接口对接与Docker一键部署的完整方法并支持直接替换数据或继续扩展功能。无论是个人公众号接入智能回复还是学习工程化部署都能从中获得可直接落地的思路。1. 从微信公众号到ChatGPT这个机器人项目到底在解决什么公众号后台的自动回复做到怀疑人生大概是很多运营者的共同经历。关键词匹配写得再细致用户换个说法就答非所问更不用说深夜时段根本没人盯着后台。这个项目解决的就是这件事用 Python 的 Flask 写一个轻量服务把微信公众号收到的用户消息转发给 ChatGPT再把 ChatGPT 的回复通过微信接口推回去最后用 Docker 把整个服务打包成可一键部署的镜像。你不需要从零搭一套 NLP 系统也不需要买第三方客服机器人自己就能掌控所有对话数据和话术。适合的人群比较明确手里有公众号、想让对话体验明显上一个台阶的开发者以及想研究微信开放接口与外部 API 如何对接的 Python 学习者。整条链路的难点不在 Flask而在微信回调的时序约束——微信要求 5 秒内必须响应而 ChatGPT 正常情况下就要 2 到 5 秒高峰期更久。怎么处理这个时间差是整篇文章的主线。2. 先把链路画清楚公众号到 ChatGPT 的消息旅程与选型理由2.1 微信公众号开发者模式的接入逻辑为什么要先做 Token 验证在碰任何代码之前得先理解消息是怎么从用户手机跑到你服务器上的。用户在公众号里发一条文字消息微信服务器会把它打包成一个 HTTP 请求POST 到你配置的回调 URL 上。但微信不是随便把消息发给你它要先确认「这个地址真的是你登记的服务器」于是就有了开发者模式的 Token 验证机制。你在公众号后台的「基本配置 → 服务器配置」里填三个东西URL你的回调地址、Token自己定的一个字符串、EncodingAESKey消息加密密钥。点击提交时微信会往这个 URL 发一个 GET 请求带四个参数signature、timestamp、nonce、echostr。你的服务端需要用 Token 和 timestamp、nonce 一起做一次签名计算算出来的结果和 signature 一致就把 echostr 原样返回。微信收到正确的 echostr才会把服务器配置启用成功。这里有个很多人第一次做就翻车的点URL 必须是公网可访问的地址而且端口要通。本地调试时你访问localhost:5000自己能通但微信服务器访问不到所以验证永远失败。常见做法是本地写完代码直接docker build后推到一台有公网 IP 的云主机上跑通验证再回头做本地迭代。GET /wechat?signaturexxxtimestamp123nonceabcechostrpong HTTP/1.1这个 GET 请求里signature 是微信用 Token、timestamp、nonce 三个值做 sha1 得到的。为什么微信不直接比对你返回的字符串而要你重新算一遍签名因为 echostr 可能被中间层拦截或伪造但签名必须知道你的 Token 才能算出来这相当于一个身份证明。2.2 Flask 在整条链路里的真实职责路由、XML 解析与请求时序很多人以为这个项目的核心是 AI 对话其实 Flask 做的事更像一个「翻译官」。微信服务器是发起方Flask 是被动接收方它把微信的 XML 消息解开提取出用户说的话转发给 ChatGPT再把回复按微信要求的 XML 格式包好返回去。整个过程里 Flask 只暴露一个路由、处理两种 HTTP 方法本质上就是一个接口服务。为什么不建议用 DjangoDjango 自带 ORM、Admin、中间件体系对这个项目来说太重了一个路由引入整个框架的启动开销没必要。FastAPI 的异步能力很香但如果你或你的团队不熟悉 ASGI 部署碰到 uvicorn 和 gunicorn 的配合问题反而多。Flask 是这三个里生态最成熟、资料最多、踩坑记录最好查的公众号开发这种需求十年前就有人在 Flask 上做几乎每个你能想到的报错都有人贴过解决方案。整条链路的时序关系也要在动手前理清用户发消息 → 微信服务器 POST 到 Flask → Flask 解析、调 ChatGPT → 微信等待响应。微信的等待上限是 5 秒超过就判定超时并重试。也就是说Flask 不仅要「能处理消息」还要「处理得足够快」否则 ChatGPT 还没算完微信已经把同一个请求又推了一遍。这个约束直接决定了代码结构也解释了为什么后面要引入异步和客服消息接口。另外要明确一个边界本项目适用于认证的服务号或测试号。订阅号的消息接口权限受限部分功能开不全测试号则没有这个限制适合先跑通逻辑。2.3 为什么用 Docker环境一致性、回调域名的迁移成本与后续扩展微信回调 URL 必须是一个固定的公网地址这意味着你的服务大概率要跑在一台云主机上。云主机上装 Python 环境、装依赖、调试端口和本地开发环境很难保证完全一致——A 同学本地 Python 3.9 能跑服务器上 3.11 就报某个依赖编译失败这种「换台机器就坏」的事在实操里太常见了。Docker 解决的就是这个一致性。把 Python 版本、pip 依赖、运行命令全部写进镜像任何一台装了 Docker 的机器docker run起来就是同一个环境。迁移时也不用重新配一遍服务器镜像推到新机器直接跑。后续项目要加功能比如接入一个知识库检索只需要在镜像里加依赖重新 build旧的容器停掉新的容器顶上回滚也就是再跑一次旧镜像的事。对比项裸机部署Docker 部署Python 版本不一致容易踩坑镜像内固定依赖冲突手动排查隔离在容器内迁移服务器重装环境重新拉起镜像回滚版本手动改代码切换镜像版本日志查看手动找文件docker logs或挂载卷还有一个容易被忽略的好处Docker 容器的重启策略。云主机偶尔重启裸机部署的服务可能要等手动启动而容器设了--restartalways后会自动拉起公众号后台不会因为服务器没起来而报「暂时无法提供服务」。3. 用 Flask 把微信消息接进来核心代码与参数说明3.1 最小可用的 Token 验证接口跑不通它什么都做不了先写最基础的 Flask 应用只做一件事处理微信的 GET 验证请求。这个接口跑不通后面全是空谈。import hashlib from flask import Flask, request WECHAT_TOKEN your-token-here app Flask(__name__) app.route(/wechat, methods[GET, POST]) def wechat(): if request.method GET: signature request.args.get(signature, ) timestamp request.args.get(timestamp, ) nonce request.args.get(nonce, ) echostr request.args.get(echostr, ) token WECHAT_TOKEN tmp_list [token, timestamp, nonce] tmp_list.sort() tmp_str .join(tmp_list) hash_str hashlib.sha1(tmp_str.encode(utf-8)).hexdigest() if hash_str signature: return echostr return bad signature, 403 # POST 处理逻辑在 3.2 中补充 return ok这段代码的逻辑不复杂但有几个细节值得说。微信的签名算法是把 token、timestamp、nonce 三个字符串按 ASCII 码排序拼接成一个字符串再做 sha1。sort()默认就是按 ASCII 码排不要自己写排序规则。encode(utf-8)必须有否则sha1收到的是字符串而不是字节串Python 会直接抛 TypeError。3.2 接收用户消息并调用 ChatGPT绕开 5 秒超时的两种写法用户消息是以 POST 请求到达的消息体是 XML。先用标准库解析一下import xml.etree.ElementTree as ET def parse_wechat_xml(xml_data): root ET.fromstring(xml_data) msg {child.tag: child.text for child in root} return msg解析出来的字典包含几个关键字段ToUserName是公众号原始 IDFromUserName是用户的 openidContent是文本内容MsgType是消息类型。这里必须先判断类型用户发的不一定是文字可能是图片、语音甚至是关注事件这一类没有Content直接调 ChatGPT 会拿 None 当 prompt白扣一次 API 费用。最直观的写法是同步处理收到消息、调用 ChatGPT、把回复拼成 XML 返回。但实际跑起来你就会发现ChatGPT 的响应时间在 2 到 10 秒之间波动微信只等你 5 秒。一旦超时微信会重试同一个请求重试还是超时就再重试最终用户那边显示「该公众号暂时无法提供服务」而你的 OpenAI 账单上多了三四笔调用。from openai import OpenAI client OpenAI(api_keysk-xxx) app.route(/wechat, methods[POST]) def handle_message(): msg parse_wechat_xml(request.data) if msg.get(MsgType) ! text: return reply call_gpt(msg.get(Content, )) return build_reply_xml(msg, reply) def call_gpt(prompt): resp client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.7, max_tokens500, ) return resp.choices[0].message.contenttemperature0.7是在创意和稳定之间取的中间值公众号客服场景不建议调太高否则同一句话每次回答都不一样。max_tokens500限制回复长度ChatGPT 有时会一口气写两三千字微信回复消息有长度上限截断反而体验更差。同步写法的问题刚才说了应对办法是改成异步。先把微信请求立刻应答掉不让它重试再在后台线程里慢慢调 ChatGPT完成后用客服消息接口主动推给用户。这是公众号 AI 机器人目前的主流做法。import threading app.route(/wechat, methods[POST]) def handle_message(): msg parse_wechat_xml(request.data) if msg.get(MsgType) ! text: return threading.Thread( targetasync_reply, args(msg, msg.get(Content, )), daemonTrue, ).start() return def async_reply(msg, user_content): try: answer call_gpt(user_content) push_customer_msg(msg[FromUserName], answer) except Exception as e: app.logger.error(reply failed: %s, e)先返回空字符串HTTP 状态码是 200微信收到 200 就认为消息处理成功不会重试。这一步解决了超时重试的问题也避免了重复扣费。然后threading.Thread在后台做耗时的 API 调用推消息用客服消息接口。客服消息需要 access_token这个 token 有效期为 7200 秒不能每次现取否则频繁调用获取接口会被限流。常见做法是缓存到内存里过期前提前 200 秒刷新import time import requests token_cache {value: , expires_at: 0} def get_access_token(): if token_cache[expires_at] time.time(): return token_cache[value] resp requests.get( https://api.weixin.qq.com/cgi-bin/token, params{ grant_type: client_credential, appid: APP_ID, secret: APP_SECRET, }, timeout10, ).json() token_cache[value] resp[access_token] token_cache[expires_at] time.time() resp[expires_in] - 200 return token_cache[value] def push_customer_msg(openid, content): url fhttps://api.weixin.qq.com/cgi-bin/message/custom/send?access_token{get_access_token()} requests.post( url, json{touser: openid, msgtype: text, text: {content: content}}, timeout5, )这里减掉的 200 秒是安全余量。实际运营中 token 在边界时间失效的概率不低踩过一次坑后你就会理解为什么所有公众号 SDK 都建议提前刷新。不过要注意客服消息接口有频率限制认证服务号是每分钟 20 条峰值流量下需要做消息队列做削峰。3.3 回复消息的 XML 组装格式错了微信直接丢弃无论是同步返回还是客服消息推送最终都要拼一段 XML 给微信。同步返回的格式是固定模板import time def build_reply_xml(msg, reply_content): return f xml ToUserName![CDATA[{msg[FromUserName]}]]/ToUserName FromUserName![CDATA[{msg[ToUserName]}]]/FromUserName CreateTime{int(time.time())}/CreateTime MsgType![CDATA[text]]/MsgType Content![CDATA[{reply_content}]]/Content /xml .strip()ToUserName和FromUserName必须对调因为微信的视角里回复的接收者就是消息的发送者。CreateTime用当前时间戳微信会拿它做消息去重。CDATA包裹是为了防止回复内容里的特殊字符破坏 XML 结构比如用户问代码问题ChatGPT 回了带br或引号的文本不包 CDATA 微信解析就报错消息直接丢。另外一个容易踩的坑是消息加密。如果公众号后台把消息加解密方式设成了「安全模式」上面收到的 XML 就是密文必须先解密才能解析回复时也要加密回去。调试期建议先用明文模式跑通上线前再切安全模式否则排查问题时加密和解密混在一起很难定位是业务逻辑问题还是加解密问题。4. Docker 部署避坑指南从本地跑通到公网可见的四个大坑4.1 生产环境的 Dockerfile 写法为什么不用 flask run本地开发时flask run或直接python app.py都没问题但生产环境还这么干就要出事。Flask 自带的开发服务器是单进程、单线程的线上并发一上来就卡住。这里用 gunicorn 作为 WSGI 服务器两个 worker 就能应付公众号日常流量。FROM python:3.11-slim WORKDIR /app ENV PYTHONUNBUFFERED1 \ TZAsia/Shanghai RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime \ echo $TZ /etc/timezone COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 5000 CMD [gunicorn, app:app, -b, 0.0.0.0:5000, -w, 2, --timeout, 60]PYTHONUNBUFFERED1让 Python 日志不缓冲否则容器里看不到实时输出排查问题很痛苦。TZAsia/Shanghai配合ln -snf解决时区问题详情看 4.2。--timeout 60是 gunicorn 的 worker 超时时间如果某个请求处理超过 60 秒会被强制杀掉这个值要大于 ChatGPT 最长响应时间但又不能太大否则线程堆积。4.2 坑一容器时区漂移日志时间差 8 小时现象容器跑起来了公众号也能回复但docker logs里看到的时间戳全部比北京慢 8 小时。查问题时会发现「用户 12 点发的消息日志显示 4 点」很难定位真实问题。原因python:3.11-slim这个基础镜像默认时区是 UTC容器里没有/etc/timezone配置所有系统调用返回的时间都基于 UTC。Python 的datetime.now()拿到的也是 UTC 时间。解决在 Dockerfile 里显式设置时区上面已经写了。还要确认代码里获取时间戳用的是time.time()而不是datetime.utcnow()后者不管容器时区怎么设拿到的都是 UTC。统一用time.time()或者datetime.now()不要在代码里再手动加 8 小时。4.3 坑二公众号后台「配置失败」Token 验证 403现象后台点提交微信提示「配置失败」或「Token 验证失败」浏览器直接访问 URL 能看到bad signature。原因这个坑通常有三个来源。第一URL 路径写错代码路由是/wechat后台填成了/wechat/或者反过来。第二app.route里只写了methods[POST]微信的 GET 验证请求打过来直接 405。第三返回值有问题return echostr被框架当成了 JSON 处理。解决先用 curl 在服务器本地测 GET 请求看是不是返回 echostr。确认是签名算法问题还是路径问题。注意后台填的 URL 必须和你代码路由完全一致大小写、末尾斜杠都算。如果前面有 Nginx 做代理检查代理转发时是不是把 query string 弄丢了signature 和 echostr 全在 query string 里。提示不要自己在入口处加一层自定义的「IP 白名单」或拦截中间件微信服务器回调的出口 IP 是动态的加白名单会导致验证请求被拦截后台一直报配置失败。4.4 坑三ChatGPT 响应超过 5 秒用户收到「暂时无法提供服务」现象用户发一条消息公众号立刻回复「该公众号暂时无法提供服务请稍后再试」日志里 OpenAI 调用还在正常跑但用户永远看不到答案。原因同步调 ChatGPT 超过了微信的 5 秒等待上限。微信掷出超时后重试重试三次都超时就放弃并把错误提示返回给用户。更隐蔽的是重试的每一次都真实触发了 OpenAI 的 API 调用账单上会出现多条重复扣费。解决用第 3 章说的异步方案先返回空串、再推客服消息。如果高峰时段 ChatGPT 特别慢可以在异步线程里多加一层兜底超过 8 秒还没拿到回复给用户推一条「答案还在路上请稍候查看」避免用户以为机器人死了。这个兜底文案本身也是一次客服消息发送占用的配额很小但体验提升明显。4.5 坑四容器重启后日志和上下文全没现象云主机重启容器自动拉起来了但之前的对话记录、运行日志全没了。用户问「你刚才不是说……」机器人什么都不记得。原因容器文件系统是临时层容器重启时内部写入的数据全部清空。数据库、日志文件如果写在容器内等同于没有持久化。解决用数据卷挂载把日志目录从容器里拿出来。docker run -d --name wechat-bot \ -p 5000:5000 \ -v /data/wechat-bot/logs:/app/logs \ --restartalways \ wechat-bot:1.0/data/wechat-bot/logs是宿主机目录/app/logs是容器内日志目录两边一一对应。代码里要把日志配置成写入/app/logs下的文件gunicorn 的--access-logfile和--error-logfile也指到这个目录这样宿主机上直接tail -f就能看日志。对话上下文的问题单靠数据卷解决不了因为多 worker 下内存本来就不共享需要引入 Redis这在第 5 章讲。5. 上线前的验证手段与两个进阶技巧5.1 用 curl 模拟微信回调先于真实用户完成冒烟测试每改一次代码就拿着手机给公众号发消息测试效率太低。更可靠的做法是先用脚本模拟微信的请求把整条链路在本地触发一遍。验证 GET 签名时先用脚本算出签名值import hashlib token your-token timestamp 123 nonce abc tmp .join(sorted([token, timestamp, nonce])) sig hashlib.sha1(tmp.encode(utf-8)).hexdigest() print(sig)再把结果拼进 curl 请求curl http://localhost:5000/wechat?signature上面算出的sigtimestamp123nonceabcechostrpong期望输出是pong。POST 模拟用户发消息需要构造一段真实结构的 XMLcurl -X POST http://localhost:5000/wechat \ -H Content-Type: application/xml \ -d xmlToUserName![CDATA[gh_test]]/ToUserNameFromUserName![CDATA[openid_test]]/FromUserNameCreateTime1234567890/CreateTimeMsgType![CDATA[text]]/MsgTypeContent![CDATA[你好]]/Content/xml如果走了异步方案POST 应该立刻返回 200 空串然后等一两秒再去看日志里客服消息是否推送成功。5.2 进阶用 Redis 存会话上下文让机器人有记忆ChatGPT 的接口本身是无状态的同一个用户问「你好」和接着问「我叫张三」它不会记得前面说了什么。要做成有记忆的对话得自己按 openid 存历史。import redis r redis.Redis.from_url(redis://localhost:6379/0) def build_messages(openid, user_content): history r.lrange(fsession:{openid}, 0, -1) messages [{role: system, content: 你是一个公众号助手}] for item in history: role, content item.decode().split(|, 1) messages.append({role: role, content: content}) messages.append({role: user, content: user_content}) r.rpush(fsession:{openid}, fuser|{user_content}) r.ltrim(fsession:{openid}, -10, -1) return messagesltrim保留最近 10 轮对话防止历史无限增长撑爆 token 上限。想限制机器人记住更少的内容把这个值调成 5 或 3 都可以。5.3 上线检查清单改完代码准备发布前按这个顺序自查一遍后台服务器配置已启用Token 验证通过用真实微信号发一条消息确认 5 秒内收到「空串」响应、稍后收到客服消息回复查docker logs确认没有异常堆栈确认 access_token 没有频繁刷新重启容器后会话上下文能通过 Redis 恢复。这套链路我前前后后搭过好几版最深的教训是所有外呼接口必须先确认时序再写业务逻辑。微信的 5 秒超时是一个硬约束不绕开它模型再聪明用户也收不到。设计接口时先问「如果上游 10 秒才回我的服务怎么处理」比写完再补要省心得多。希望帮到你。本文还有配套的精品资源点击获取
简介一份基于Flask与ChatGPT的微信公众号聊天机器人后端工程面向已有Python基础、希望快速搭建公众号智能对话服务的开发者也适合用于学习Flask接口开发、ChatGPT API接入和Docker容器化部署。压缩包内含可运行的完整源码和部署文档按说明配置Python 3.7及以上环境即可启动对新手比较友好。资源共15个文件以Python脚本、Markdown部署文档、Docker配置Dockerfile与docker-compose.yml及依赖锁定文件为主压缩后仅52KB体积轻巧、目录结构清晰便于阅读和二次开发。已有143人学习下载。项目包含应用入口、聊天核心逻辑、日志记录、系统部署说明及工作流配置等内容可帮助读者梳理公众号消息处理链路掌握ChatGPT接口对接与Docker一键部署的完整方法并支持直接替换数据或继续扩展功能。无论是个人公众号接入智能回复还是学习工程化部署都能从中获得可直接落地的思路。1. 从微信公众号到ChatGPT这个机器人项目到底在解决什么公众号后台的自动回复做到怀疑人生大概是很多运营者的共同经历。关键词匹配写得再细致用户换个说法就答非所问更不用说深夜时段根本没人盯着后台。这个项目解决的就是这件事用 Python 的 Flask 写一个轻量服务把微信公众号收到的用户消息转发给 ChatGPT再把 ChatGPT 的回复通过微信接口推回去最后用 Docker 把整个服务打包成可一键部署的镜像。你不需要从零搭一套 NLP 系统也不需要买第三方客服机器人自己就能掌控所有对话数据和话术。适合的人群比较明确手里有公众号、想让对话体验明显上一个台阶的开发者以及想研究微信开放接口与外部 API 如何对接的 Python 学习者。整条链路的难点不在 Flask而在微信回调的时序约束——微信要求 5 秒内必须响应而 ChatGPT 正常情况下就要 2 到 5 秒高峰期更久。怎么处理这个时间差是整篇文章的主线。2. 先把链路画清楚公众号到 ChatGPT 的消息旅程与选型理由2.1 微信公众号开发者模式的接入逻辑为什么要先做 Token 验证在碰任何代码之前得先理解消息是怎么从用户手机跑到你服务器上的。用户在公众号里发一条文字消息微信服务器会把它打包成一个 HTTP 请求POST 到你配置的回调 URL 上。但微信不是随便把消息发给你它要先确认「这个地址真的是你登记的服务器」于是就有了开发者模式的 Token 验证机制。你在公众号后台的「基本配置 → 服务器配置」里填三个东西URL你的回调地址、Token自己定的一个字符串、EncodingAESKey消息加密密钥。点击提交时微信会往这个 URL 发一个 GET 请求带四个参数signature、timestamp、nonce、echostr。你的服务端需要用 Token 和 timestamp、nonce 一起做一次签名计算算出来的结果和 signature 一致就把 echostr 原样返回。微信收到正确的 echostr才会把服务器配置启用成功。这里有个很多人第一次做就翻车的点URL 必须是公网可访问的地址而且端口要通。本地调试时你访问localhost:5000自己能通但微信服务器访问不到所以验证永远失败。常见做法是本地写完代码直接docker build后推到一台有公网 IP 的云主机上跑通验证再回头做本地迭代。GET /wechat?signaturexxxtimestamp123nonceabcechostrpong HTTP/1.1这个 GET 请求里signature 是微信用 Token、timestamp、nonce 三个值做 sha1 得到的。为什么微信不直接比对你返回的字符串而要你重新算一遍签名因为 echostr 可能被中间层拦截或伪造但签名必须知道你的 Token 才能算出来这相当于一个身份证明。2.2 Flask 在整条链路里的真实职责路由、XML 解析与请求时序很多人以为这个项目的核心是 AI 对话其实 Flask 做的事更像一个「翻译官」。微信服务器是发起方Flask 是被动接收方它把微信的 XML 消息解开提取出用户说的话转发给 ChatGPT再把回复按微信要求的 XML 格式包好返回去。整个过程里 Flask 只暴露一个路由、处理两种 HTTP 方法本质上就是一个接口服务。为什么不建议用 DjangoDjango 自带 ORM、Admin、中间件体系对这个项目来说太重了一个路由引入整个框架的启动开销没必要。FastAPI 的异步能力很香但如果你或你的团队不熟悉 ASGI 部署碰到 uvicorn 和 gunicorn 的配合问题反而多。Flask 是这三个里生态最成熟、资料最多、踩坑记录最好查的公众号开发这种需求十年前就有人在 Flask 上做几乎每个你能想到的报错都有人贴过解决方案。整条链路的时序关系也要在动手前理清用户发消息 → 微信服务器 POST 到 Flask → Flask 解析、调 ChatGPT → 微信等待响应。微信的等待上限是 5 秒超过就判定超时并重试。也就是说Flask 不仅要「能处理消息」还要「处理得足够快」否则 ChatGPT 还没算完微信已经把同一个请求又推了一遍。这个约束直接决定了代码结构也解释了为什么后面要引入异步和客服消息接口。另外要明确一个边界本项目适用于认证的服务号或测试号。订阅号的消息接口权限受限部分功能开不全测试号则没有这个限制适合先跑通逻辑。2.3 为什么用 Docker环境一致性、回调域名的迁移成本与后续扩展微信回调 URL 必须是一个固定的公网地址这意味着你的服务大概率要跑在一台云主机上。云主机上装 Python 环境、装依赖、调试端口和本地开发环境很难保证完全一致——A 同学本地 Python 3.9 能跑服务器上 3.11 就报某个依赖编译失败这种「换台机器就坏」的事在实操里太常见了。Docker 解决的就是这个一致性。把 Python 版本、pip 依赖、运行命令全部写进镜像任何一台装了 Docker 的机器docker run起来就是同一个环境。迁移时也不用重新配一遍服务器镜像推到新机器直接跑。后续项目要加功能比如接入一个知识库检索只需要在镜像里加依赖重新 build旧的容器停掉新的容器顶上回滚也就是再跑一次旧镜像的事。对比项裸机部署Docker 部署Python 版本不一致容易踩坑镜像内固定依赖冲突手动排查隔离在容器内迁移服务器重装环境重新拉起镜像回滚版本手动改代码切换镜像版本日志查看手动找文件docker logs或挂载卷还有一个容易被忽略的好处Docker 容器的重启策略。云主机偶尔重启裸机部署的服务可能要等手动启动而容器设了--restartalways后会自动拉起公众号后台不会因为服务器没起来而报「暂时无法提供服务」。3. 用 Flask 把微信消息接进来核心代码与参数说明3.1 最小可用的 Token 验证接口跑不通它什么都做不了先写最基础的 Flask 应用只做一件事处理微信的 GET 验证请求。这个接口跑不通后面全是空谈。import hashlib from flask import Flask, request WECHAT_TOKEN your-token-here app Flask(__name__) app.route(/wechat, methods[GET, POST]) def wechat(): if request.method GET: signature request.args.get(signature, ) timestamp request.args.get(timestamp, ) nonce request.args.get(nonce, ) echostr request.args.get(echostr, ) token WECHAT_TOKEN tmp_list [token, timestamp, nonce] tmp_list.sort() tmp_str .join(tmp_list) hash_str hashlib.sha1(tmp_str.encode(utf-8)).hexdigest() if hash_str signature: return echostr return bad signature, 403 # POST 处理逻辑在 3.2 中补充 return ok这段代码的逻辑不复杂但有几个细节值得说。微信的签名算法是把 token、timestamp、nonce 三个字符串按 ASCII 码排序拼接成一个字符串再做 sha1。sort()默认就是按 ASCII 码排不要自己写排序规则。encode(utf-8)必须有否则sha1收到的是字符串而不是字节串Python 会直接抛 TypeError。3.2 接收用户消息并调用 ChatGPT绕开 5 秒超时的两种写法用户消息是以 POST 请求到达的消息体是 XML。先用标准库解析一下import xml.etree.ElementTree as ET def parse_wechat_xml(xml_data): root ET.fromstring(xml_data) msg {child.tag: child.text for child in root} return msg解析出来的字典包含几个关键字段ToUserName是公众号原始 IDFromUserName是用户的 openidContent是文本内容MsgType是消息类型。这里必须先判断类型用户发的不一定是文字可能是图片、语音甚至是关注事件这一类没有Content直接调 ChatGPT 会拿 None 当 prompt白扣一次 API 费用。最直观的写法是同步处理收到消息、调用 ChatGPT、把回复拼成 XML 返回。但实际跑起来你就会发现ChatGPT 的响应时间在 2 到 10 秒之间波动微信只等你 5 秒。一旦超时微信会重试同一个请求重试还是超时就再重试最终用户那边显示「该公众号暂时无法提供服务」而你的 OpenAI 账单上多了三四笔调用。from openai import OpenAI client OpenAI(api_keysk-xxx) app.route(/wechat, methods[POST]) def handle_message(): msg parse_wechat_xml(request.data) if msg.get(MsgType) ! text: return reply call_gpt(msg.get(Content, )) return build_reply_xml(msg, reply) def call_gpt(prompt): resp client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.7, max_tokens500, ) return resp.choices[0].message.contenttemperature0.7是在创意和稳定之间取的中间值公众号客服场景不建议调太高否则同一句话每次回答都不一样。max_tokens500限制回复长度ChatGPT 有时会一口气写两三千字微信回复消息有长度上限截断反而体验更差。同步写法的问题刚才说了应对办法是改成异步。先把微信请求立刻应答掉不让它重试再在后台线程里慢慢调 ChatGPT完成后用客服消息接口主动推给用户。这是公众号 AI 机器人目前的主流做法。import threading app.route(/wechat, methods[POST]) def handle_message(): msg parse_wechat_xml(request.data) if msg.get(MsgType) ! text: return threading.Thread( targetasync_reply, args(msg, msg.get(Content, )), daemonTrue, ).start() return def async_reply(msg, user_content): try: answer call_gpt(user_content) push_customer_msg(msg[FromUserName], answer) except Exception as e: app.logger.error(reply failed: %s, e)先返回空字符串HTTP 状态码是 200微信收到 200 就认为消息处理成功不会重试。这一步解决了超时重试的问题也避免了重复扣费。然后threading.Thread在后台做耗时的 API 调用推消息用客服消息接口。客服消息需要 access_token这个 token 有效期为 7200 秒不能每次现取否则频繁调用获取接口会被限流。常见做法是缓存到内存里过期前提前 200 秒刷新import time import requests token_cache {value: , expires_at: 0} def get_access_token(): if token_cache[expires_at] time.time(): return token_cache[value] resp requests.get( https://api.weixin.qq.com/cgi-bin/token, params{ grant_type: client_credential, appid: APP_ID, secret: APP_SECRET, }, timeout10, ).json() token_cache[value] resp[access_token] token_cache[expires_at] time.time() resp[expires_in] - 200 return token_cache[value] def push_customer_msg(openid, content): url fhttps://api.weixin.qq.com/cgi-bin/message/custom/send?access_token{get_access_token()} requests.post( url, json{touser: openid, msgtype: text, text: {content: content}}, timeout5, )这里减掉的 200 秒是安全余量。实际运营中 token 在边界时间失效的概率不低踩过一次坑后你就会理解为什么所有公众号 SDK 都建议提前刷新。不过要注意客服消息接口有频率限制认证服务号是每分钟 20 条峰值流量下需要做消息队列做削峰。3.3 回复消息的 XML 组装格式错了微信直接丢弃无论是同步返回还是客服消息推送最终都要拼一段 XML 给微信。同步返回的格式是固定模板import time def build_reply_xml(msg, reply_content): return f xml ToUserName![CDATA[{msg[FromUserName]}]]/ToUserName FromUserName![CDATA[{msg[ToUserName]}]]/FromUserName CreateTime{int(time.time())}/CreateTime MsgType![CDATA[text]]/MsgType Content![CDATA[{reply_content}]]/Content /xml .strip()ToUserName和FromUserName必须对调因为微信的视角里回复的接收者就是消息的发送者。CreateTime用当前时间戳微信会拿它做消息去重。CDATA包裹是为了防止回复内容里的特殊字符破坏 XML 结构比如用户问代码问题ChatGPT 回了带br或引号的文本不包 CDATA 微信解析就报错消息直接丢。另外一个容易踩的坑是消息加密。如果公众号后台把消息加解密方式设成了「安全模式」上面收到的 XML 就是密文必须先解密才能解析回复时也要加密回去。调试期建议先用明文模式跑通上线前再切安全模式否则排查问题时加密和解密混在一起很难定位是业务逻辑问题还是加解密问题。4. Docker 部署避坑指南从本地跑通到公网可见的四个大坑4.1 生产环境的 Dockerfile 写法为什么不用 flask run本地开发时flask run或直接python app.py都没问题但生产环境还这么干就要出事。Flask 自带的开发服务器是单进程、单线程的线上并发一上来就卡住。这里用 gunicorn 作为 WSGI 服务器两个 worker 就能应付公众号日常流量。FROM python:3.11-slim WORKDIR /app ENV PYTHONUNBUFFERED1 \ TZAsia/Shanghai RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime \ echo $TZ /etc/timezone COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 5000 CMD [gunicorn, app:app, -b, 0.0.0.0:5000, -w, 2, --timeout, 60]PYTHONUNBUFFERED1让 Python 日志不缓冲否则容器里看不到实时输出排查问题很痛苦。TZAsia/Shanghai配合ln -snf解决时区问题详情看 4.2。--timeout 60是 gunicorn 的 worker 超时时间如果某个请求处理超过 60 秒会被强制杀掉这个值要大于 ChatGPT 最长响应时间但又不能太大否则线程堆积。4.2 坑一容器时区漂移日志时间差 8 小时现象容器跑起来了公众号也能回复但docker logs里看到的时间戳全部比北京慢 8 小时。查问题时会发现「用户 12 点发的消息日志显示 4 点」很难定位真实问题。原因python:3.11-slim这个基础镜像默认时区是 UTC容器里没有/etc/timezone配置所有系统调用返回的时间都基于 UTC。Python 的datetime.now()拿到的也是 UTC 时间。解决在 Dockerfile 里显式设置时区上面已经写了。还要确认代码里获取时间戳用的是time.time()而不是datetime.utcnow()后者不管容器时区怎么设拿到的都是 UTC。统一用time.time()或者datetime.now()不要在代码里再手动加 8 小时。4.3 坑二公众号后台「配置失败」Token 验证 403现象后台点提交微信提示「配置失败」或「Token 验证失败」浏览器直接访问 URL 能看到bad signature。原因这个坑通常有三个来源。第一URL 路径写错代码路由是/wechat后台填成了/wechat/或者反过来。第二app.route里只写了methods[POST]微信的 GET 验证请求打过来直接 405。第三返回值有问题return echostr被框架当成了 JSON 处理。解决先用 curl 在服务器本地测 GET 请求看是不是返回 echostr。确认是签名算法问题还是路径问题。注意后台填的 URL 必须和你代码路由完全一致大小写、末尾斜杠都算。如果前面有 Nginx 做代理检查代理转发时是不是把 query string 弄丢了signature 和 echostr 全在 query string 里。提示不要自己在入口处加一层自定义的「IP 白名单」或拦截中间件微信服务器回调的出口 IP 是动态的加白名单会导致验证请求被拦截后台一直报配置失败。4.4 坑三ChatGPT 响应超过 5 秒用户收到「暂时无法提供服务」现象用户发一条消息公众号立刻回复「该公众号暂时无法提供服务请稍后再试」日志里 OpenAI 调用还在正常跑但用户永远看不到答案。原因同步调 ChatGPT 超过了微信的 5 秒等待上限。微信掷出超时后重试重试三次都超时就放弃并把错误提示返回给用户。更隐蔽的是重试的每一次都真实触发了 OpenAI 的 API 调用账单上会出现多条重复扣费。解决用第 3 章说的异步方案先返回空串、再推客服消息。如果高峰时段 ChatGPT 特别慢可以在异步线程里多加一层兜底超过 8 秒还没拿到回复给用户推一条「答案还在路上请稍候查看」避免用户以为机器人死了。这个兜底文案本身也是一次客服消息发送占用的配额很小但体验提升明显。4.5 坑四容器重启后日志和上下文全没现象云主机重启容器自动拉起来了但之前的对话记录、运行日志全没了。用户问「你刚才不是说……」机器人什么都不记得。原因容器文件系统是临时层容器重启时内部写入的数据全部清空。数据库、日志文件如果写在容器内等同于没有持久化。解决用数据卷挂载把日志目录从容器里拿出来。docker run -d --name wechat-bot \ -p 5000:5000 \ -v /data/wechat-bot/logs:/app/logs \ --restartalways \ wechat-bot:1.0/data/wechat-bot/logs是宿主机目录/app/logs是容器内日志目录两边一一对应。代码里要把日志配置成写入/app/logs下的文件gunicorn 的--access-logfile和--error-logfile也指到这个目录这样宿主机上直接tail -f就能看日志。对话上下文的问题单靠数据卷解决不了因为多 worker 下内存本来就不共享需要引入 Redis这在第 5 章讲。5. 上线前的验证手段与两个进阶技巧5.1 用 curl 模拟微信回调先于真实用户完成冒烟测试每改一次代码就拿着手机给公众号发消息测试效率太低。更可靠的做法是先用脚本模拟微信的请求把整条链路在本地触发一遍。验证 GET 签名时先用脚本算出签名值import hashlib token your-token timestamp 123 nonce abc tmp .join(sorted([token, timestamp, nonce])) sig hashlib.sha1(tmp.encode(utf-8)).hexdigest() print(sig)再把结果拼进 curl 请求curl http://localhost:5000/wechat?signature上面算出的sigtimestamp123nonceabcechostrpong期望输出是pong。POST 模拟用户发消息需要构造一段真实结构的 XMLcurl -X POST http://localhost:5000/wechat \ -H Content-Type: application/xml \ -d xmlToUserName![CDATA[gh_test]]/ToUserNameFromUserName![CDATA[openid_test]]/FromUserNameCreateTime1234567890/CreateTimeMsgType![CDATA[text]]/MsgTypeContent![CDATA[你好]]/Content/xml如果走了异步方案POST 应该立刻返回 200 空串然后等一两秒再去看日志里客服消息是否推送成功。5.2 进阶用 Redis 存会话上下文让机器人有记忆ChatGPT 的接口本身是无状态的同一个用户问「你好」和接着问「我叫张三」它不会记得前面说了什么。要做成有记忆的对话得自己按 openid 存历史。import redis r redis.Redis.from_url(redis://localhost:6379/0) def build_messages(openid, user_content): history r.lrange(fsession:{openid}, 0, -1) messages [{role: system, content: 你是一个公众号助手}] for item in history: role, content item.decode().split(|, 1) messages.append({role: role, content: content}) messages.append({role: user, content: user_content}) r.rpush(fsession:{openid}, fuser|{user_content}) r.ltrim(fsession:{openid}, -10, -1) return messagesltrim保留最近 10 轮对话防止历史无限增长撑爆 token 上限。想限制机器人记住更少的内容把这个值调成 5 或 3 都可以。5.3 上线检查清单改完代码准备发布前按这个顺序自查一遍后台服务器配置已启用Token 验证通过用真实微信号发一条消息确认 5 秒内收到「空串」响应、稍后收到客服消息回复查docker logs确认没有异常堆栈确认 access_token 没有频繁刷新重启容器后会话上下文能通过 Redis 恢复。这套链路我前前后后搭过好几版最深的教训是所有外呼接口必须先确认时序再写业务逻辑。微信的 5 秒超时是一个硬约束不绕开它模型再聪明用户也收不到。设计接口时先问「如果上游 10 秒才回我的服务怎么处理」比写完再补要省心得多。希望帮到你。本文还有配套的精品资源点击获取
