ChatGPT on WeChat 实战：从零搭建 AI 辅助开发环境

ChatGPT on WeChat 实战：从零搭建 AI 辅助开发环境

摘要：本文针对开发者在微信生态中集成 ChatGPT 时遇到的 API 调用限制、消息处理延迟和上下文管理难题，提供了一套完整的解决方案。通过 WeChat Official Account API 与 OpenAI API 的深度整合，开发者可以构建高性能的 AI 对话系统。文章包含 Python 实现代码、会话状态管理技巧和并发优化方案，帮助开发者快速实现生产级部署。

1. 背景痛点：微信生态的“三座大山”

一句话：想优雅地把 ChatGPT 搬进微信，得先翻这三座大山。

2. 技术选型：Serverless 还是常驻进程？

结论：

• 对“秒回”有刚需，选常驻进程。
• Flask 轻量，生态足，本文用 Flask；Django 也行，改路由即可。

3. 核心实现：让微信、GPT、Redis 跳好三人舞

3.1 微信消息加解密 —— wechatpy 一把梭

官方 XML 格式 + AES 加密，手写易踩坑。wechatpy 已封装：

from wechatpy import parse_message, create_reply
from wechatpy.crypto import WeChatCrypto

crypto = WeChatCrypto(token, aes_key, app_id)

def wechat_entry():
signature = request.args.get("signature")
timestamp = request.args.get("timestamp")
nonce = request.args.get("nonce")
encrypt_type = request.args.get("encrypt_type", "raw")

if encrypt_type == "aes":
msg = crypto.decrypt_message(request.data, signature, timestamp, nonce)
else:
msg = request.data

wechat_msg = parse_message(msg)
return dispatch(wechat_msg)

3.2 OpenAI 流式响应 —— 边吐边回

GPT 的 stream=True 把首包时间从 2 s 降到 200 ms，但微信要一次性返回，不能真流。策略：

代码骨架：

import openai, asyncio

async def stream_gpt(prompt):
resp = await openai.ChatCompletion.acreate(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
stream=True,
max_tokens=500,
)
buf = ""
async for chunk in resp:
delta = chunk.choices[0].delta.get("content", "")
buf += delta
if len(buf) >= 140 and first_chunk:
return buf # 立即返回，剩余后台补
return buf

3.3 Redis 会话管理 —— 用 LRU 保记忆

• Key 设计：wechat:{openid}
• Value：JSON 列表，最多保留 8 轮对话，超则左侧弹。
• TTL：30 分钟沉默即销毁，省内存。

import redis, json
r = redis.Redis(decode_responses=True)

MAX_CONTEXT = 8

def get_context(openid):
data = r.get(f"wechat:{openid}")
return json.loads(data) if data else []

def set_context(openid, ctx):
r.setex(f"wechat:{openid}", 1800, json.dumps(ctx[-MAX_CONTEXT:]))

4. 代码示例：一条消息的一生

下面把上面碎片拼成完整中间件，注释比代码多，放心食用。

# wechat_gpt.py
import re, asyncio, openai, redis, time
from flask import Flask, request, make_response
from wechatpy import parse_message, create_reply
from wechatpy.crypto import WeChatCrypto

app = Flask(__name__)
r = redis.Redis(decode_responses=True)

# ========== 配置 ==========
TOKEN = "wx_token"
AES_KEY = "32位Base64"
APP_ID = "wx123456"
openai.api_key = "sk-xxx"
openai.proxy = "http://127.0.0.1:7890" # 国内代理
# ==========================

crypto = WeChatCrypto(TOKEN, AES_KEY, APP_ID)

# 敏感词正则，可自行扩充
SENSITIVE = re.compile(
r"(微信|VX|vx|加群|赌博|发票|枪支)", re.I
)

@app.route("/wx", methods=["GET", "POST"])
def entry():
# 1. 验证签名
if request.method == "GET":
return request.args.get("echostr")

# 2. 解密
msg = crypto.decrypt_message(
request.data,
request.args.get("signature"),
request.args.get("timestamp"),
request.args.get("nonce"),
)
wechat_msg = parse_message(msg)

# 3. 过滤
if SENSITIVE.search(wechat_msg.content):
return reply_text(wechat_msg, "内容敏感，换个话题吧~")

# 4. 异步处理
asyncio.run(handle(wechat_msg))
return "success" # 先回微信，避免 5 秒超时

async def handle(msg):
openid = msg.source
context = get_context(openid)
context.append({"role": "user", "content": msg.content})

try:
gpt_answer = await stream_gpt(context)
except Exception as e:
gpt_answer = "AI 走神了，稍后再试"

# 更新上下文
context.append({"role": "assistant", "content": gpt_answer})
set_context(openid, context)

# 客服消息接口二次推送
from wechatpy import WeChatClient
client = WeChatClient(APP_ID, "app_secret")
client.message.send_text(openid, gpt_answer)

def reply_text(msg, text):
rsp = create_reply(text, msg).render()
return crypto.encrypt_message(rsp, request.args.get("nonce"))

# ========== 工具函数 ==========
def get_context(openid):
data = r.get(f"wechat:{openid}")
return json.loads(data) if data else []

def set_context(openid, ctx, max_len=8):
r.setex(f"wechat:{openid}", 1800, json.dumps(ctx[-max_len:]))

async def stream_gpt(messages):
buf = ""
async for chunk in await openai.ChatCompletion.acreate(
model="gpt-3.5-turbo",
messages=messages,
stream=True,
max_tokens=500,
temperature=0.7,
):
delta = chunk.choices[0].delta.get("content", "")
buf += delta
if len(buf) >= 140: # 首包策略
break
return buf

5. 生产考量：把玩具变成人民币

6. 避坑指南：封号比 Bug 更疼

location /v1 {
proxy_pass https://api.openai.com;
proxy_set_header Host api.openai.com;
proxy_set_header Authorization "Bearer sk-xxx";
}

7. 开放思考：token 这么贵，你怎么省？

• 把“系统提示”做模板化，减少每轮重复指令？
• 用 Embedding 做知识库，让 GPT 只生成“增量”部分？
• 还是干脆把 3.5 换成 4o-mini，精度换成本？

欢迎把你的奇思妙想留言交流。

写完小记：整套流程跑通后，我把公众号丢给产品同事内测，半天就被拉去喝茶——原来用户疯狂问“写周报”，GPT 一滚滚出 800 字，触发了截断提示，用户体验却出奇地好。省 token 的事，看来得继续卷。

如果你也想亲手搭一个能“听”会“说”的 AI 伙伴，不妨看看这个动手实验：从0打造个人豆包实时通话AI。我跟着做了一遍，步骤很细，本地 30 分钟就能跑通，对“实时语音”这条链路会有更立体的体感。至于微信文本方案，把 ASR 换成用户输入即可无缝复用。祝开发愉快，少踩坑，多省钱。