简介面向需要在微信公众号中接入智能对话能力的开发场景这份资源提供了一套基于Flask与ChatGPT构建的聊天机器人后端源码并包含docker一键部署方案适合有一定Python基础、想快速上线公众号机器人的开发者参考。资源包共15个文件压缩包仅52KB主要文件类型包括Python脚本app.py、chatgpt.py、log.py、Markdown部署文档、Dockerfile与docker-compose.yml容器编排配置、Pipfile依赖锁定以及LICENSE等代码结构清晰便于直接替换数据后运行。目前已有142人学习学习热度逐步积累。项目内含部署文档能引导从环境准备、依赖安装到服务启动的完整流程配套Flask系统部署文档和示例数据可有效降低小白上手难度适合用于课程设计、个人公众号机器人、Docker实践等轻量级项目改造。1. 这个项目到底解决什么问题一条能照做的微信公众号聊天机器人链路基于 Flask ChatGPT Docker 的微信公众号聊天机器人拆开看就是一条非常标准的技术链路用户在微信里发一条消息微信服务器把消息推到你部署的公网地址上Flask 收到后整理成对话上下文调一次 ChatGPT 接口拿到回复再按微信要求的 XML 格式把回复还回去整个过程用 Docker 打包成镜像换一台服务器也能一键拉起。这类项目最容易被低估的是「微信消息收发」这半边——很多人把精力全花在调 ChatGPT 上结果卡在签名校验、XML 拼包和 5 秒超时这几个坎上机器人怎么都上不了线。这篇文章会把这条路从头走一遍适合手里有公众号或者想用测试号快速验证 AI 客服场景的开发者照着做就能跑通踩过的坑也都标出来了。2. 动手前的三件事申请测试号、装好本地环境、把 Flask 骨架跑起来2.1 公众号用测试号还是正式号先想清楚你要干什么做微信公众号聊天机器人第一步不是写代码是决定用哪种号。很多人一上来就注册服务号结果卡在营业执照和 300 元认证费上项目直接搁浅。其实微信公众平台有一个「测试号」入口在 mp.weixin.qq.com 的调试沙箱里点击登录后能直接拿到一套独立的 appID 和 appsecret不需要企业资质也不需要认证自带接收用户消息、被动回复、自定义菜单这些核心接口权限足够把聊天机器人完整跑通。我早期做个人项目基本都是先用测试号验证逻辑跑通了再考虑要不要迁到正式号。正式服务号的优势在高级接口客服消息接口可以主动给用户推消息、模板消息通知类推送、网页授权以及微信支付相关能力。这些是测试号给不了的。如果你的目标只是「微信里有个 AI 助手」测试号完全够如果要做成对外运营的产品比如给用户发订单通知、做客服工单流转那就得走正式服务号。两者的关键差异我整理成一张表账号类型费用与门槛聊天机器人相关权限适合场景测试号免费微信扫码登录即可接收消息、被动回复、自定义菜单、二维码关注个人开发、功能验证、学习 Demo个人订阅号免费身份证注册接收消息有限制被动回复可用个人内容号回复能力较弱正式服务号需营业执照认证费 300 元/年客服消息、模板消息、网页授权、被动回复面向用户的生产环境产品测试号的另一个好处是有一个专属的「测试号二维码」手机微信扫码关注后就能像真实用户一样给机器人发消息联调体验和正式号几乎没有差别。所以我下面的所有步骤都基于测试号来做你用正式号也完全一样只是后台入口不同。2.2 本地环境Python、Flask 与 Docker Desktop 的准备代码层面这个项目依赖 Python 3.10 以上版本、Flask、OpenAI 的 Python SDK、Gunicorn 和 Redis 客户端。Python 的安装没什么好说的Windows 去官网下载安装包时记得勾选「Add Python to PATH」macOS 用 Homebrew 装也行。装完之后在项目目录里建虚拟环境这一步不要省不然依赖冲突会把你折腾到怀疑人生mkdir wechat-chatbot cd wechat-chatbot python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install flask openai requests gunicorn redis依赖装完后先别急着写业务逻辑用一个最简的 Flask 应用把链路起点占住。新建app.py写一个/wechat路由先只处理 GET 请求并返回一个占位字符串后面章节再补签名校验from flask import Flask app Flask(__name__) app.route(/wechat, methods[GET, POST]) def wechat(): return ok跑flask run --port 5000浏览器访问http://localhost:5000/wechat能看到 ok说明 Flask 环境没问题。开发阶段先不用 Docker 跑 Flask本地直接起服务调试速度快改代码即时生效Docker 留给后面的生产部署章节那时候再处理多进程、时区和依赖缓存的问题。如果你是 Windows 用户Docker Desktop 的安装是另一个容易翻车的地方。官方安装包装完第一次启动经常报virtualization support not detected或者Docker Desktop failed to start because...这两个报错九成是同一个原因Windows 的硬件虚拟化没开或者 WSL2 内核没装。解决路径是在 BIOS 里打开 Intel VT-x / AMD-V然后装 WSL2 内核更新包最后在 PowerShell 里执行wsl --set-default-version 2。处理完重启 Docker Desktop基本就能看到那个鲸鱼图标正常转起来了。2.3 拿到测试号后台的三行关键配置测试号后台在微信公众平台调试入口登录后页面里有两个核心区块。一个是「测试号信息」写着 appID 和 appsecret这两个值后面调 ChatGPT 和获取 access_token 都要用另一个是「接口配置信息」需要填 URL 和 Token这就是微信服务器回调你的服务的地址。URL 的格式是http://你的公网地址/wechatToken 是你自己随便定的一串字符比如mytoken123但必须和代码里写的保持一致。这里的关键问题是本地开发时你的服务在localhost:5000微信服务器不可能访问到。常见做法是用一个内网穿透工具把本机的 5000 端口暴露成一个公网 HTTPS 地址我一般用 ngrok 这类工具启动ngrok http 5000它会给你一个形如https://xxxx.ngrok.io的地址把这个地址拼上/wechat填进后台 URL 栏即可。填完点提交微信服务器会立刻向这个 URL 发一次 GET 请求做验证。如果你只写了上面那个返回ok的骨架验证会失败因为微信要求你按签名规则计算并返回echostr参数——这就是下一章要做的事。另外提醒一句测试号后台的 URL 和 Token 每次修改后都要重新提交验证本地调试时内网穿透地址变了记得同步去后台改否则微信还会往旧地址推消息。3. 把微信消息接进 FlaskURL 验证、XML 解析与被动回复3.1 URL 验证微信服务器的第一次握手微信服务器的回调验证逻辑是这样的在你提交接口配置时微信会向你的 URL 发送一个 GET 请求带上signature、timestamp、nonce、echostr四个参数。你需要把后台配置的 Token、收到的 timestamp、nonce 三个值拼在一起先按字典序排序再拼接成字符串做一次 SHA1 加密得到的结果和signature比对。一致就把echostr原样返回给微信验证通过之后微信才会把用户消息 POST 到这个地址。import hashlib from flask import Flask, request TOKEN mytoken123 # 必须与测试号后台填写的 Token 完全一致 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) tmp sorted([TOKEN, timestamp, nonce]) tmp_str .join(tmp) if hashlib.sha1(tmp_str.encode(utf-8)).hexdigest() signature: return echostr return verify failed, 403 # POST 分支下一节实现这里有几个容易忽略的细节。sorted是对三个字符串做字典序排序不是原来的顺序写错顺序签名永远对不上。tmp_str.encode(utf-8)不能省SHA1 的对象是字节串字符串直接传会报类型错误。signature和计算结果的比较用字符串等值判断即可不要用去比较可能为None的值所以前面要用request.args.get的默认值兜底。这个 GET 分支不用写日志验证完成后微信不会再主动 GET但保留一个快速返回的逻辑能让后台配置提交时秒过。3.2 消息接收与被动回复XML 的解析与拼包用户给公众号发消息后微信服务器会把消息以 XML 形式 POST 到你的/wechat接口。一段典型的文本消息 XML 长这样根节点是xml里面是ToUserName开发者微信号、FromUserName用户的 openid、CreateTime时间戳、MsgType消息类型、Content文本内容。解析这段 XML 不需要引入重量级库Python 标准库的xml.etree.ElementTree就够用。回复消息同样要拼一段 XML 回去被动回复的格式有几个硬性要求ToUserName和FromUserName要互换位置用户发给公众号你回复时两边的值对调CreateTime用当前时间戳MsgType是textContent里放回复正文。所有文本字段都要用 CDATA 包裹不然特殊字符会把 XML 结构撑破。import xml.etree.ElementTree as ET from datetime import datetime from flask import request, make_response def parse_xml(xml_str): root ET.fromstring(xml_str) return {child.tag: child.text or for child in root} def build_text_xml(to_user, from_user, content): return fxml ToUserName![CDATA[{to_user}]]/ToUserName FromUserName![CDATA[{from_user}]]/FromUserName CreateTime{int(datetime.now().timestamp())}/CreateTime MsgType![CDATA[text]]/MsgType Content![CDATA[{content}]]/Content /xml app.route(/wechat, methods[GET, POST]) def wechat(): # GET 分支同前一节 if request.method POST: payload parse_xml(request.data.decode(utf-8)) openid payload.get(FromUserName) appid payload.get(ToUserName) msg_type payload.get(MsgType) if msg_type text: user_input payload.get(Content, ).strip() # reply 这里先用占位文本下一章接入 ChatGPT reply f你说了{user_input} else: reply 我目前只能处理文字消息 xml_resp build_text_xml(openid, appid, reply) return make_response(xml_resp, 200, {Content-Type: application/xml})parse_xml里用字典推导式把 XML 子节点转成 dict取字段时用.get并给默认值因为图片、语音这类消息可能没有Content字段。build_text_xml里的 f-string 拼 XML 在内容短时没问题但如果content里恰好包含]]序列这段 XML 就会提前结束。简单处理是在拼装前把]]替换成全角版本或者干脆在写Content时避免输出这种特殊串后面接 ChatGPT 后基本不会出现但心里要有个数。这里还得提一个微信的隐藏机制被动回复必须在 5 秒内完成响应。如果你 5 秒内没返回微信会认为超时然后重试三次重试的消息里MsgId相同。所以接口里最好按MsgId做一次幂等把已经处理过的MsgId缓存起来避免同一个用户消息被重复处理、回复两次。这个点在接入 GPT 后尤其重要因为 GPT 的响应时间经常超过 5 秒后面避坑章节会专门展开。3.3 明文、兼容还是安全模式三种消息加解密方案的取舍测试号后台的「消息加解密方式」有三个选项明文模式、兼容模式、安全模式。明文模式下微信 POST 上来的 XML 是上面那种直接可读的格式回复也不用加密开发阶段最省事。兼容模式是明文和密文同时推给你安全模式则只能收到密文所有消息都要做 AES 解密回复也要用 AES 加密回去。我的建议是开发阶段用明文模式把业务逻辑跑通上线前最后再切安全模式。安全模式的完整加解密逻辑涉及 AES-128-CBC、PKCS7 填充、消息体里拼接 appid 再取前 16 字节做 key 这些细节自己手写容易在边界情况上踩坑。更稳的做法是用wechatpy库它把加解密封装好了from wechatpy.crypto import WeChatCrypto from wechatpy.exceptions import InvalidSignatureException crypto WeChatCrypto(token, encoding_aes_key, appid) # 解密收到的密文 encrypted_xml payload.get(Encrypt) try: decrypted crypto.decrypt_message(encrypted_xml, msg_signature, timestamp, nonce) except InvalidSignatureException: return signature error, 403切安全模式时后台会让你填一个 EncodingAESKey43 位字符生成后要同时存到环境变量里。注意兼容模式下Encrypt字段和明文字段同时存在代码里要先判断有没有Encrypt有就走解密分支没有就走明文解析这个分支判断写漏了切模式后接口会大面积报错。4. 接入 ChatGPT 与使用数据资料openai SDK、会话记忆与 FAQ 兜底4.1 openai SDK 的第一次对话与三个必调参数微信消息解析没问题后把回复内容替换成 ChatGPT 的生成结果。现在的 openai Python 包已经到 1.x 版本接口风格和早期的 0.x 完全不同早期写法openai.ChatCompletion.create已经废弃要用新的OpenAI客户端类import os from openai import OpenAI client OpenAI(api_keyos.environ[OPENAI_API_KEY]) def ask_chatgpt(user_id, user_input): resp client.chat.completions.create( modelgpt-3.5-turbo, messages[ {role: system, content: 你是公众号客服回答要简洁、口语化控制在 200 字以内。}, {role: user, content: user_input}, ], temperature0.7, max_tokens512, timeout10, ) return resp.choices[0].message.contenttemperature是第一个要调的参数它控制输出的随机性取值范围 0 到 2。做客服机器人我一般设在 0.5 到 0.7 之间太低回复会显得机械太高容易跑偏。max_tokens限制回复长度512 对应大概 300-400 个汉字对微信聊天场景够用设太大会拖慢响应时间影响 5 秒超时。timeout是必须加的没有它网络异常时请求会一直挂着直接占满 worker 线程。在继续之前有一个前提需要先确认运行环境要能访问api.openai.com这个域名这是所有调用的基础具体怎么让你的环境具备这个网络条件按你自己的合规方式处理这里不展开。4.2 会话记忆怎么存从内存 dict 到 RedisChatGPT 的接口本身是无状态的同样一问「你叫什么」不带上下文它每次都会重新自我介绍。聊天机器人要像真人对话必须把每个用户的历史消息攒起来下次请求时一起发给模型。最简单的做法是进程内用 dict 保存key 是用户的 openidfrom collections import defaultdict import time SESSIONS defaultdict(list) SESSION_TTL 7200 # 2 小时无对话则清空 def get_messages(openid, user_input): now time.time() msg_list SESSIONS[openid] if now - msg_list.get(updated_at, 0) SESSION_TTL: SESSIONS[openid] {messages: [], updated_at: now} msg_list SESSIONS[openid] msg_list[messages].append({role: user, content: user_input}) msg_list[messages] msg_list[messages][-20:] # 控制长度防止 token 超限 return msg_list[messages]但内存方案有个致命缺陷一旦用 Gunicorn 起多个 worker 进程每个进程的 dict 是独立的同一用户两次请求可能命中不同 worker上下文就断了。所以一旦进入 Docker 部署阶段我就会换成 Redis。Redis 的字符串结构加过期时间正好匹配这个场景用 openid 做 key存整个 messages 列表的 JSON 序列化结果import redis, json r redis.Redis.from_url(os.environ.get(REDIS_URL, redis://localhost:6379/0)) def get_messages(openid, user_input): key fsession:{openid} raw r.get(key) messages json.loads(raw) if raw else [] messages.append({role: user, content: user_input}) r.setex(key, SESSION_TTL, json.dumps(messages[-20:], ensure_asciiFalse)) return messages用setex设过期时间Redis 会自动清理超过 2 小时不活跃的会话省得自己写定时任务。ensure_asciiFalse一定要加不然中文会被转成\uXXXX存储没问题但可读性差排查问题时会很痛苦。Redis 的部署方式我放在第 5 章跟 Docker Compose 一起写。4.3 数据资料怎么用启动加载 FAQ 兜底让机器人不冷场标题里提到的「数据资料」在这个项目里通常指的就是 FAQ 语料、历史对话记录、种子问题集这类数据文件。我的习惯是把它们整理成两个文件data/faq.json存放常见问题与标准答案data/chat.log记录每次对话的原文和回复后者既是数据资产也是排查线上问题的线索。FAQ 文件的结构很简单就是一个列表每项包含关键词和回答import json, difflib FAQ_FILE data/faq.json def load_faq(): with open(FAQ_FILE, encodingutf-8) as f: return json.load(f) def match_faq(user_input, faqs): best_score, best_answer 0, for item in faqs: score difflib.SequenceMatcher(None, user_input, item[question]).ratio() if score best_score: best_score, best_answer score, item[answer] return best_answer if best_score 0.6 else 这里用difflib.SequenceMatcher做模糊匹配命中阈值设 0.6。匹配到就直接返回 FAQ 里的标准答案不走 ChatGPT省 API 调用费没匹配到才调 ChatGPT。这个兜底机制的价值在超时场景特别明显当 ChatGPT 响应超过 5 秒时微信已经判定超时了这时候再回什么都没有意义。所以我的处理顺序是先查 FAQ命中就秒回未命中再调 GPT并给 GPT 的调用套一个 3 秒的超时超时就回「我还在学习这个问题换个说法问问」——保证任何情况下都在 5 秒内给了响应。这其实也是「数据资料」在这类源码包里最常见的打开方式不是训练模型而是做一个快速命中的本地知识库用来兜底和控成本。等 FAQ 积累到几百条机器人的表现会比纯调 GPT 更像一个合格客服因为 FAQ 里的答案是人工校验过的不会出现 GPT 一本正经胡说的情况。5. 用 Docker 部署上线与绕不开的避坑点从 Dockerfile 到 HTTPS 回调5.1 一份能上生产的 Dockerfile依赖缓存、时区与 gunicorn本地调试用 Flask 自带的开发服务器没问题但生产环境必须换 Gunicorn。Flask 自带的服务器是单进程单线程的性能差不说遇到慢请求会直接阻塞整个服务。Gunicorn 是多 worker 的配合--timeout可以控制每个请求的最大时长。Dockerfile 里还有个细节先复制requirements.txt并安装依赖再复制项目代码。这样依赖层会被 Docker 缓存后续代码改动了重新构建镜像时只要requirements.txt没变依赖安装这步就不会重跑构建速度快很多。FROM python:3.11-slim ENV PYTHONUNBUFFERED1 \ TZAsia/Shanghai \ PIP_INDEX_URLhttps://pypi.tuna.tsinghua.edu.cn/simple RUN apt-get update apt-get install -y --no-install-recommends tzdata \ ln -snf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \ echo Asia/Shanghai /etc/timezone WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD [gunicorn, -w, 2, -b, 0.0.0.0:8000, --timeout, 30, app:app]TZAsia/Shanghai配合 tzdata 包解决容器日志时区问题不然日志里的时间永远比北京时间慢 8 小时排查线上问题时会让你怀疑人生。PIP_INDEX_URL指到国内镜像源解决构建时 pip 下载超时的问题这一步在服务器网络环境下几乎是必须的。Gunicorn 的-w 2表示开两个 worker 进程--timeout 30是单个请求最大允许 30 秒。worker 数不要盲目加大多 worker 会放大下一节要说的会话内存问题2 到 4 个对个人公众号足够。5.2 docker-compose 编排app、Redis 与 data 挂载单容器跑起来简单但项目里还依赖 Redis 做会话存储用 Docker Compose 一次性把两个服务编排起来是更省事的方案。写在docker-compose.ymlservices: app: build: . ports: - 8000:8000 environment: - OPENAI_API_KEY${OPENAI_API_KEY} - REDIS_URLredis://redis:6379/0 volumes: - ./data:/app/data depends_on: - redis restart: always redis: image: redis:7-alpine restart: alwaysOPENAI_API_KEY从.env文件读取docker compose会自动加载同目录下的.env文件密钥不要写进代码或镜像这是底线。./data:/app/data把宿主机上的data目录挂载进容器FAQ 文件和聊天日志就持久化在宿主机上容器重建、镜像升级后数据不丢。depends_on保证 Redis 先启动restart: always让服务在意外退出后自动拉起省去了写 systemd 服务的麻烦。启动命令就是一行docker compose up -d --build5.3 HTTPS 回调地址让微信服务器能找到你的服务微信公众平台接口配置里的 URL测试号阶段填 HTTP 地址也能通过验证但生产环境强烈建议上 HTTPS。原因有两层一是微信官方对正式服务号的回调地址要求越来越严格明文 HTTP 的稳定性差容易被运营商劫持二是你的服务端与微信服务器之间传输的是用户消息和 openidHTTPS 能防止中间人截获。用 Caddy 做流量转发是我现在最常用的方案因为 Caddy 会自动申请和续期 Lets Encrypt 证书不需要像 Nginx 那样手动维护证书文件caddy reverse-proxy --from yourdomain.com --to localhost:8000这一条命令就把 443 端口的 HTTPS 请求全部转给宿主机的 8000 端口。前提是你有一个域名并且域名解析 A 记录指向服务器的公网 IP。国内云服务器还有一个绕不开的环节是备案域名要完成 ICP 备案后电信运营商才会放行 80 和 443 端口这个周期通常要一两周建议项目启动时就先把域名和备案流程办了别等服务写好了才临时抱佛脚。5.4 部署阶段必踩的四个坑时区、worker、镜像源、超时重试第一个坑是容器日志时间差 8 小时。现象是看docker compose logs时明明下午三点的消息日志显示早上七点。原因是 Python 基础镜像默认 UTC 时区。解决就是在 Dockerfile 里装 tzdata 并设置TZAsia/Shanghai改完重建镜像即可。这个坑不影响功能但影响排查效率上线第一天就会碰上。第二个坑是 Gunicorn 多 worker 导致会话上下文串不起来。现象是用户问「我叫小明」机器人说「很高兴认识你」再问「我叫什么」机器人说「我不知道」——原因是第一个请求落在 worker A第二个落在 worker B内存里的 dict 是进程独立的B 里根本没有 A 存的上下文。解决是用 Redis 替代内存存会话这也是我 4.2 节坚持换成 Redis 的原因。临时应急可以-w 1强制单进程但只适合测试上线不合适。第三个坑是构建镜像时 pip 下载超时。现象是docker build卡在pip install很久然后报连接超时。原因是默认源在境外服务器网络访问不稳定。解决是 Dockerfile 里设置PIP_INDEX_URL走国内镜像源或者干脆用清华源。这也解释了为什么 5.1 节把镜像源直接写进 Dockerfile 里而不是每次构建手动指定。第四个坑是超时重试导致用户收到重复回复。现象是用户发一条消息收到两条相同内容的回复。原因是微信要求 5 秒内响应但 ChatGPT 的调用经常在 5 秒以上微信判定超时后自动重试同一条消息你的代码又处理了一遍。解决分三块入口按MsgId去重已经处理过的MsgId直接返回空串GPT 调用加timeout并控制在 3 秒内先走 FAQ 兜底保证秒回。三件事都做到这个坑基本就填平了。6. 上线前的最后检查与一个排障技巧机器人在本地和用内网穿透联调都跑通后往服务器部署前我习惯按下面这套顺序做最后的检查。先把构建好的镜像启动然后确认容器健康访问http://服务器IP:8000/wechat?echostr1timestamp1nonce1能看到页面提示 verify failed说明服务在跑接着看docker compose logs -f app的日志确认没有异常报错最后用手机微信给测试号发一条消息观察日志里是否出现「收到消息」和 ChatGPT 的响应耗时。日志是这里最直接的证据我在代码里加了两行打印一行收到消息时输出 openid 和正文一行回复前输出耗时import logging logging.basicConfig(levellogging.INFO) logging.info(recv [%s] %s, openid, user_input) logging.info(reply in %.2fs, elapsed)一个对我很管用的排障技巧是不要只记录 openid把会话的上下文条数也打出来。比如日志里session len6你能立刻判断是上下文太长导致 token 超限还是 Redis 连接失败导致取不到历史。否则遇到「用户说了一句话机器人答非所问」你根本分不清是模型问题还是记忆问题。我第一次把机器人推到线上时整晚都在跟超时重试较劲后来才意识到问题不在 GPT 慢而是我不该在 5 秒限制里等一个注定超时的响应把 FAQ 兜底放在 ChatGPT 前面之后问题才彻底消停。现在每次上线前我都固定跑一遍上面的检查确认日志、FAQ 兜底、MsgId 去重三件事都到位才收工。希望帮到你。本文还有配套的精品资源点击获取