简介:这份资源是一套基于Flask框架、ChatGPT接口与Docker容器化部署的微信公众号聊天机器人完整项目,面向具备Python基础、希望快速搭建智能对话公众号的开发者与学习者。项目将后端服务、模型调用与容器编排整合在一起,适合用作课程设计、个人练手或二次开发的基础模板。压缩包共15个文件,约52KB,包含3个py源码文件、3个md部署与说明文档、2个yml容器编排配置,以及Pipfile依赖清单、Dockerfile、LICENSE等工程化文件,覆盖从代码到部署的完整链路。目前已有142人学习下载。读者可获得可直接运行的机器人后端源码、Flask与Docker两套部署文档、依赖锁定文件与容器化配置,便于理解微信公众号消息接入、ChatGPT对话逻辑与一键部署流程,并在此基础上替换数据、扩展功能或进行排错调试。
1. 从一条公众号消息到 ChatGPT 回复:这套 Flask 后端到底能跑通什么
公众号后台收到一条用户消息,微信服务器会在 5 秒内要求你返回结果,超时就直接重试三次。这个硬约束决定了任何"公众号接大模型"的项目,核心难点从来不是调 API,而是怎么在 5 秒内把异步的模型回复塞回用户手里。这套基于 Flask + ChatGPT + Docker 的微信公众号聊天机器人源码,解决的正是这件事:它用 Flask 起一个 webhook 服务接收微信推送,把用户消息转给 ChatGPT,再把回复按微信要求的 XML 格式吐回去,同时用 Docker 把整个运行环境封成一个镜像,省掉在服务器上逐个装依赖的功夫。
适合谁看:手上已经有一个认证过的公众号(服务号或订阅号都行,订阅号需要服务器配置权限)、一台能跑 Docker 的机器、一个能调通的 OpenAI API Key。如果你只是想本地跑个 demo 看看效果,这套东西也能用,但公众号那部分得靠内网穿透或者干脆跳过,只测 Flask 接口。下面按"资源里有什么 → 怎么把它跑起来 → 参数怎么调 → 哪里会翻车"的顺序拆一遍,源码包里的app.py、chatgpt.py、docker-compose.yml是三个最该先看的文件。
2. 拆开压缩包先看什么:Flask 入口、ChatGPT 封装与 Docker 三件套
拿到wechat-chatgpt-master这个目录,别急着pip install。先花十分钟把文件结构过一遍,能省掉后面一半的调试时间。这套项目的分层其实很清晰:app.py是 Flask 的 webhook 入口,chatgpt.py封装了和 OpenAI 的交互,log.py管日志,Dockerfile和docker-compose.yml负责环境,Pipfile/Pipfile.lock锁依赖版本。部署文档有两份,一份讲 Python 系统部署,一份讲 Flask 系统部署,内容有重叠,看 Flask 那份就够。
2.1 app.py 里的微信签名校验与消息路由
微信服务器配置 URL 时,会先发一个 GET 请求带signature、timestamp、nonce、echostr四个参数来验证你的服务器是不是活的。校验逻辑是把token、timestamp、nonce三个值按字典序排序后拼接,做一次 SHA1,再和signature比对。这一步错了,公众号后台点"提交"永远提示"token验证失败"。
# app.py 中微信接入验证的核心逻辑(示意,以实际源码为准) import hashlib WECHAT_TOKEN = "your_token_here" # 必须和公众号后台填的 Token 完全一致 def check_signature(signature, timestamp, nonce): # 1. 三个值按字典序排序 sort_list = sorted([WECHAT_TOKEN, timestamp, nonce]) # 2. 拼接成一个字符串 sort_str = "".join(sort_list) # 3. SHA1 加密 hash_str = hashlib.sha1(sort_str.encode("utf-8")).hexdigest() # 4. 与微信传来的 signature 比对 return hash_str == signature逻辑说明:WECHAT_TOKEN是你在公众号后台"服务器配置"里自己填的一串字符,必须和代码里完全一致,大小写敏感。sorted()默认按字符编码排序,微信官方文档要求的就是字典序,别改成按时间排序。SHA1 输出是十六进制小写,hexdigest()正好对上。参数说明:signature是微信算好的签名,你算出来的hash_str和它相等才算通过。常见错误是把token写成了access_token,那是两码事。
消息路由部分,微信 POST 过来的是 XML,不是 JSON。Flask 里得用request.data拿原始字节再解析,用request.json会直接报错。解析出MsgType后判断是text还是event,文本消息取Content字段,事件消息(比如关注)取Event字段。回复也得是 XML,格式错了微信不认,用户那边就是"该公众号暂时无法提供服务"。
2.2 chatgpt.py 的请求封装与超时控制
chatgpt.py干的事是把用户文本发给 OpenAI 的接口,拿回回复文本。这里有两个参数直接决定用户体验:model和timeout。模型选gpt-3.5-turbo这类对话模型,别用text-davinci那种补全模型,后者对中文对话场景不友好。超时时间建议设 4 秒以内,因为微信那边 5 秒就断,留 1 秒给网络和 XML 组装。
# chatgpt.py 请求封装示意 import openai openai.api_key = "sk-xxxxxxxx" # 换成你自己的 Key openai.api_base = "https://api.openai.com/v1" # 有中转就改这里 def get_gpt_reply(user_text, timeout=4): try: resp = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": user_text}], timeout=timeout, max_tokens=500 ) return resp.choices[0].message.content.strip() except Exception as e: # 超时或报错时给用户一个兜底回复,别让公众号卡死 return "稍等,我这边有点忙,请再发一次"逻辑说明:messages用列表结构,role可以是system、user、assistant,想加人设就加一条system。max_tokens控制回复长度,设太大不仅慢还费额度。timeout是 requests 层面的超时,不是模型推理时间,设 4 秒意味着 4 秒内没拿到完整响应就抛异常。参数说明:api_base如果你用的是官方地址就保持默认,用第三方中转就改成对应地址,但注意中转的稳定性和合规性自己把握。异常处理里返回兜底文案,比让用户干等 5 秒然后收到"该公众号暂时无法提供服务"体验好得多。
2.3 Dockerfile 与 docker-compose.yml 的环境固化
Dockerfile一般基于python:3.7-slim或python:3.9-slim,把Pipfile拷进去,用pipenv install --system --deploy装依赖,再COPY . .,最后CMD ["python", "app.py"]。docker-compose.yml则把端口映射、环境变量、重启策略写在一起,一条docker-compose up -d就能起服务。
# docker-compose.yml 结构示意 version: "3" services: wechat-bot: build: . ports: - "80:5000" # 宿主机 80 映射容器 5000,微信要求 80 或 443 environment: - WECHAT_TOKEN=your_token_here - OPENAI_API_KEY=sk-xxxxxxxx restart: always # 崩了自动拉起,生产环境必加逻辑说明:微信服务器配置只接受 80 或 443 端口,所以宿主机映射必须用这两个之一,容器内部 Flask 默认跑 5000 就映射成80:5000。环境变量把 Token 和 Key 从代码里抽出来,改配置不用重新 build 镜像。restart: always是血泪经验,Flask 单进程遇到未捕获异常会退出,没这行半夜挂了都不知道。参数说明:build: .表示用当前目录的 Dockerfile 构建,如果你要拉现成镜像就换成image: xxx。
3. 从零跑通:Python 本地调试与 Docker 一键部署两条路
这套资源给了两种跑法,本地 Python 直跑适合调试,Docker 适合上线。建议先在本地把逻辑跑通,确认签名校验和 GPT 调用都没问题,再上 Docker。直接上 Docker 的话,日志在容器里,排查起来多一层。
3.1 本地 Python 环境准备与依赖安装
Python 版本按文档要求 3.7 或以上,实测 3.9 和 3.10 也能跑,但 3.11 之后某些老依赖可能编译报错。用 Pipenv 的话,进目录先pipenv install,它会读Pipfile装依赖并生成虚拟环境。不想用 Pipenv 就直接pip install flask openai requests,但版本可能和Pipfile.lock锁的不一致,遇到玄学问题优先怀疑版本。
# 方式一:用 Pipenv(推荐,和项目锁文件一致) cd wechat-chatgpt-master pip install pipenv pipenv install pipenv shell # 进入虚拟环境 # 方式二:直接 pip(快,但版本可能漂移) pip install flask openai requests # 启动服务 python app.py # 看到 Running on http://0.0.0.0:5000 就算起来了逻辑说明:pipenv shell进虚拟环境后,所有命令都在隔离环境里跑,不会污染系统 Python。python app.py启动的是 Flask 自带的开发服务器,够调试用,但生产环境必须换 gunicorn 或 uwsgi,Flask 自带服务器扛不住并发。参数说明:如果app.py里写死了app.run(debug=True),上线前记得关掉 debug,否则报错页面会暴露源码路径。
本地跑起来后,公众号那边还没法直接连,因为微信要求公网可访问的 80/443。调试阶段常见做法是用内网穿透工具把本地 5000 映射出去,拿到一个临时域名填到公众号后台。但注意,穿透工具的域名每次重启会变,公众号配置也得跟着改,比较折腾。更稳的做法是直接在一台有公网 IP 的云服务器上跑。
3.2 Docker 镜像构建与容器编排
服务器上装好 Docker 和 docker-compose 后,把整个目录传上去,改好docker-compose.yml里的环境变量,两条命令搞定。
# 构建并后台启动 docker-compose up -d --build # 查看日志,确认没有报错 docker-compose logs -f # 查看容器状态,STATUS 要是 Up docker ps # 改完代码重新部署 docker-compose down docker-compose up -d --build逻辑说明:--build强制重新构建镜像,改了代码不加这个参数容器里还是旧的。logs -f实时跟日志,Ctrl+C 退出但不影响容器运行。docker ps看STATUS列,如果是Restarting说明启动就崩了,得看日志找原因。参数说明:-d是后台运行,不加的话终端一关服务就停。down会删容器但保留镜像,up -d --build重新构建启动,这是改代码后的标准动作。
部署完在公众号后台填服务器地址,格式是http://你的域名/wechat(路径看app.py里路由怎么写的),Token 填和代码里一致的,点提交。如果提示成功,再给公众号发条消息,几秒内应该能收到 GPT 的回复。收不到就按下一章的排查顺序走。
3.3 公众号后台服务器配置与消息联调
公众号后台"开发 → 基本配置 → 服务器配置"里,URL 填公网地址,Token 填自定义字符串,EncodingAESKey 随机生成,消息加解密方式选"明文模式"最省事。选"安全模式"的话代码里还得处理 AES 解密,这套源码不一定带了,别给自己找麻烦。
提交时微信会发 GET 验证请求,通过后配置才生效。生效后发消息,微信 POST 到你的 URL,你的服务返回 XML。联调阶段最容易卡在三个地方:签名校验失败、XML 格式不对、5 秒超时。签名失败看 Token 是否一致;XML 格式不对看返回的Content字段有没有转义特殊字符;超时看 GPT 调用是不是太慢。建议在app.py里把收到的原始 XML 和返回的 XML 都打日志,出问题一眼就能定位。
4. 参数调优与避坑:Token、超时、模型选择里的那些坑
这套东西跑起来不难,难的是稳定跑下去。下面几条是实际部署时最常撞上的,每条按"现象 → 原因 → 解决"写,照着排查能省不少时间。
4.1 避坑一:公众号提示"该公众号暂时无法提供服务"
现象:后台配置提交成功,但一发消息就提示无法提供服务,日志里可能连请求都没收到。
原因:最常见的是返回的 XML 格式不对。微信要求回复的 XML 根节点是<xml>,里面<ToUserName>和<FromUserName>的位置和请求里是反的——请求里ToUserName是你的公众号,回复里ToUserName要填用户的 OpenID。搞反了微信直接丢弃。
解决:对照微信官方文档的回复 XML 模板逐字段核对,特别注意ToUserName和FromUserName互换,CreateTime是秒级时间戳不是毫秒。另外Content里如果有<、>、&这些字符,要做 XML 转义,否则解析失败。
4.2 避坑二:GPT 回复慢导致微信重试三次
现象:用户发一条消息,公众号回了三条一样的,或者回复内容对不上。
原因:微信 5 秒没收到响应就重试,重试三次。你的服务如果每次都调 GPT 且耗时超过 5 秒,微信就认为你没收到,重复推送。而你的服务每次收到都调一次 GPT,就回了多次。
解决:两个方向。一是把 GPT 调用超时压到 4 秒以内,超时就返回兜底文案,保证 5 秒内必回。二是做消息去重,用MsgId做键,处理过的直接返回空串或缓存结果,微信收到空串不会重试。生产环境建议两个都做。
4.3 避坑三:Docker 容器起来了但外部访问不通
现象:docker ps显示 Up,但浏览器访问域名超时,公众号也连不上。
原因:端口映射写错,或者服务器安全组没放行。docker-compose.yml里80:5000前面是宿主机端口后面是容器端口,写反了外部访问 5000 但容器里 Flask 监听的是 5000,宿主机 80 没映射上。另外云服务器默认安全组只开 22,80 和 443 要手动放行。
解决:docker port 容器名看实际映射关系,确认宿主机端口对。安全组在云控制台加规则,放行 80 和 443 的入方向。本地curl http://localhost:80能通但外网不通,基本就是安全组问题。
4.4 避坑四:API Key 泄露或额度被刷爆
现象:账单突然暴涨,或者 Key 被禁用。
原因:Key 硬编码在chatgpt.py里,代码传到公开仓库或者镜像被拉走,Key 就泄露了。另外没有做频率限制,被人恶意刷接口。
解决:Key 一律走环境变量,docker-compose.yml里用environment注入,代码里os.environ.get("OPENAI_API_KEY")读。加一层简单的频率限制,比如同一个 OpenID 每分钟最多 5 次,用内存字典或 Redis 都行。公开仓库前务必检查有没有把 Key 提交上去,提交过的话换 Key,改历史记录没用。
4.5 避坑五:Python 版本和依赖版本不匹配
现象:pipenv install报编译错误,或者跑起来ImportError。
原因:Pipfile.lock锁的是特定版本,Python 3.11 之后某些包(比如老版本cryptography)没有预编译 wheel,要现场编译,缺系统库就报错。或者系统里同时有多个 Python,pip装到了 A 环境,python跑的是 B 环境。
解决:优先用文档要求的 Python 3.7~3.10。确认which python和which pip指向同一个环境。Docker 部署能规避大部分这类问题,因为镜像里 Python 版本是固定的,所以生产环境强烈建议走 Docker。
5. 进阶:把单轮问答改成带上下文的多轮对话
默认这套源码是单轮的,用户每发一句都是独立请求,GPT 不记得上一句说了什么。想改成多轮,核心是维护一个会话历史。最简单的做法是用字典按 OpenID 存最近几轮对话,每次请求把历史一起发给 GPT。
# 多轮对话维护示意 from collections import defaultdict # 每个用户最多保留 5 轮,防止 token 超限 history = defaultdict(list) MAX_ROUNDS = 5 def get_reply_with_context(openid, user_text): history[openid].append({"role": "user", "content": user_text}) # 只保留最近 MAX_ROUNDS 轮 if len(history[openid]) > MAX_ROUNDS * 2: history[openid] = history[openid][-MAX_ROUNDS * 2:] resp = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=history[openid], timeout=4 ) reply = resp.choices[0].message.content.strip() history[openid].append({"role": "assistant", "content": reply}) return reply逻辑说明:history用defaultdict(list),每个 OpenID 一个列表,存user和assistant交替的消息。MAX_ROUNDS * 2是因为一轮对话包含一问一答两条消息。超出就切片保留最近的,防止消息列表无限增长把 token 撑爆。参数说明:MAX_ROUNDS设 5 意味着记得最近 5 轮,设太大不仅慢还费钱,设太小用户觉得"失忆"。这个方案的内存版重启就丢历史,要持久化就换 Redis,键用 OpenID,值存 JSON 序列化的消息列表,加个过期时间比如 30 分钟。
验证多轮是否生效,发"我叫小明",再发"我叫什么",能答出"小明"就对了。如果答不出,检查history是不是每次请求都新建了,或者 OpenID 取错了字段。另外注意,多轮会让每次请求的 token 数增加,响应时间变长,4 秒超时可能不够,得根据实际网络情况调,必要时把兜底文案改成"正在思考,请稍等再发一次"。
从那以后我每次部署这类 webhook 服务,都强制先本地打日志跑通签名和 XML 往返,再上 Docker,最后才配公众号后台——顺序反了,排查起来就是黑匣子。希望帮到你。
本文还有配套的精品资源,点击获取