QQ机器人接入ChatGPT实战：从鉴权到消息处理的完整解决方案

1次阅读

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

在开发 QQ 机器人接入 ChatGPT 时，开发者通常会遇到几个关键问题：

鉴权流程复杂 ：QQ 开放平台的 API 鉴权涉及多层加密和签名验证，对于新手来说容易出错。
上下文丢失 ：在长对话场景中，如果没有妥善管理对话状态，ChatGPT 无法记住之前的交流内容。
性能瓶颈 ：同步处理消息会导致响应延迟，尤其在高峰期容易引发超时或消息丢失。

我们选择 CoolQ HTTP 插件作为通信桥梁，它允许 QQ 机器人和我们的后端服务通过 Webhook 进行双向通信。

使用 JWT（JSON Web Token）对请求进行签名验证，确保消息来源的合法性。以下是 Python 示例代码：

from datetime import datetime, timedelta
import jwt

def generate_jwt(api_key: str, secret: str) -> str:
    payload = {
        'api_key': api_key,
        'exp': datetime.utcnow() + timedelta(minutes=30)
    }
    return jwt.encode(payload, secret, algorithm='HS256')

使用 Redis 存储用户对话上下文，并设置 TTL（Time To Live）自动清理过期会话：

import redis

r = redis.Redis(host='localhost', port=6379, db=0)

def save_context(user_id: str, context: dict, ttl: int = 3600):
    r.setex(f'chat:{user_id}', ttl, json.dumps(context))

使用 Celery 异步任务队列处理 ChatGPT API 调用，避免阻塞主线程：

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task
def call_chatgpt(prompt: str) -> str:
    # ChatGPT API 调用逻辑
    return response

QQ 消息传输需要使用 AES 加密，以下是加解密实现：

from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
import base64

def encrypt_message(key: bytes, iv: bytes, message: str) -> str:
    cipher = AES.new(key, AES.MODE_CBC, iv)
    ct_bytes = cipher.encrypt(pad(message.encode(), AES.block_size))
    return base64.b64encode(ct_bytes).decode()

对话状态机管理用户会话流程，状态转换图可简化为：

stateDiagram
    [*] --> 等待输入
    等待输入 --> 处理中: 收到消息
    处理中 --> 等待输入: 回复完成
    处理中 --> 错误处理: 请求失败
    错误处理 --> 等待输入: 重试成功

实现指数退避算法处理暂时性故障：

import time
import math

def exponential_backoff(retry_count: int, max_wait: int = 60):
    wait = min(math.pow(2, retry_count), max_wait)
    time.sleep(wait)