ChatGPT公众号开发实战：从接入到优化的全链路解决方案

1次阅读

共计 2005 个字符，预计需要花费 6 分钟才能阅读完成。

微信公众号接入 ChatGPT 时，开发者常遇到以下几个核心问题：

API 调用限制：ChatGPT API 有严格的速率限制（如每分钟 3 - 5 次请求），而公众号消息需在 5 秒内响应，直接调用易触发限流。
消息处理延迟：用户连续发送消息时，若前一个 ChatGPT 请求未完成，会导致响应堆积或超时。
多轮对话管理：默认 API 无状态，需要自行维护会话上下文，否则每次交互都是独立问答。
异常恢复：网络波动或 API 故障时，缺乏重试机制会导致用户体验中断。

直接同步调用
优点：实现简单，代码量少
缺点：无法应对高并发，易超时或被限流
异步回调 + 数据库存储
优点：缓解瞬时压力
缺点：依赖数据库性能，上下文管理复杂
消息队列 + 内存缓存（推荐）
优点：削峰填谷，支持优先级处理
缺点：需要额外维护队列服务

flowchart LR
    A[微信服务器] -->| 推送消息 | B[消息队列]
    B --> C[工作进程]
    C -->| 调用 | D[ChatGPT API]
    D -->| 存储 | E[Redis 缓存]
    E --> C
    C -->| 返回 | A

# 使用 Celery 处理异步任务
@app.route('/wechat', methods=['POST'])
def wechat_handler():
    msg = parse_wechat_msg(request.data)  # 解析微信 XML
    task = process_msg.delay(msg)         # 投递异步任务
    return ""  # 先返回空响应避免超时

@celery.task(bind=True)
def process_msg(self, msg):
    try:
        reply = generate_reply(msg)
        send_wechat_reply(msg.FromUserName, reply)
    except Exception as e:
        self.retry(exc=e, countdown=60)  # 失败后延迟重试

def generate_reply(msg):
    cache_key = f"conv:{msg.FromUserName}"
    history = redis.lrange(cache_key, 0, 4)  # 保留最近 5 轮对话

    prompt = build_prompt(msg.Content, history)
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=prompt
    )

    # 更新对话历史
    redis.lpush(cache_key, msg.Content, response.choices[0].message)
    redis.expire(cache_key, 1800)  # 30 分钟过期
    return response.choices[0].message

指数退避重试：首次失败立即重试，后续每次间隔时间倍增（1s, 2s, 4s…）
熔断机制：连续 5 次失败后暂停请求 1 分钟
降级回复：异常时返回预设话术（如 ” 服务繁忙，请稍后再试 ”）

请求缓存：相同问题直接返回缓存答案（MD5 哈希作为 key）
模板缓存：常见问题的回复模板预生成

IO 非阻塞：使用 aiohttp 替代 requests 发起 API 调用
批量处理：合并短时间内多个消息为一组 prompt

# 使用令牌桶算法
bucket = TokenBucket(capacity=3, fill_rate=1/60)  # 每分钟 3 次

def call_chatgpt(prompt):
    if not bucket.consume(1):
        raise RateLimitError
    # 正常调用 API...