从零开始：QQ机器人接入ChatGPT的完整实现指南

1次阅读

没有评论

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

QQ 机器人接入 AI 助手的典型使用场景包括：

社群管理：自动回复常见问题、踢出广告账号、新成员欢迎
智能客服：7×24 小时解答产品咨询、订单查询
娱乐互动：成语接龙、占卜算命、歌词生成等趣味功能
教育辅助：编程问题解答、外语翻译、知识点查询

优点：无需企业资质，个人开发者可直接使用
缺点：存在封号风险，协议稳定性依赖第三方维护

优点：官方支持，消息收发稳定
缺点：需要企业资质认证，审核周期较长

优点：封装完善，社区生态丰富
缺点：存在版本兼容性问题

推荐选择 aiocqhttp+OneBot 协议的组合方案，平衡了开发效率与合规性。

使用 aiocqhttp 建立 WebSocket 长连接：

from aiocqhttp import CQHttp

bot = CQHttp(api_root='http://127.0.0.1:5700')

@bot.on_message('private')
async def handle_private(context):
    # 消息预处理逻辑
    filtered_msg = sensitive_filter(context['message'])
    if not filtered_msg:
        return {'reply': '包含敏感内容'}

    # 调用对话服务
    response = await chat_service.generate_response(user_id=context['user_id'],
        message=filtered_msg
    )

    # 返回格式化响应
    return {'reply': response}

实现带流式响应的 API 调用：

import openai
from tenacity import retry, stop_after_attempt

class ChatService:
    def __init__(self):
        self.conversation_manager = RedisManager()

    @retry(stop=stop_after_attempt(3))
    async def generate_response(self, user_id, message):
        # 获取历史上下文
        history = self.conversation_manager.get_context(user_id)

        try:
            response = await openai.ChatCompletion.acreate(
                model="gpt-3.5-turbo",
                messages=[{"role": "system", "content": "你是有礼貌的助手"},
                    *history,
                    {"role": "user", "content": message}
                ],
                temperature=0.7,
                stream=True  # 启用流式响应
            )

            # 处理流式数据
            full_reply = ""
            async for chunk in response:
                delta = chunk.choices[0].delta
                if delta.get("content"):
                    full_reply += delta["content"]

                    # 敏感内容实时检测
                    if sensitive_filter.check(delta["content"]):
                        full_reply = "回复触发安全规则"
                        break

            # 更新对话历史
            self.conversation_manager.save_context(
                user_id, 
                user_msg=message,
                bot_msg=full_reply
            )

            return full_reply

        except openai.error.RateLimitError:
            # 监控埋点
            monitor.log('rate_limit')
            return "服务繁忙，请稍后再试"

Redis 存储结构示例：

import redis
from json import dumps, loads

class RedisManager:
    def __init__(self):
        self.client = redis.StrictRedis(
            host='localhost', 
            port=6379,
            decode_responses=True
        )

    def get_context(self, user_id, max_turns=5):
        """获取最近 N 轮对话历史"""
        key = f"qqbot:{user_id}:history"
        raw = self.client.lrange(key, 0, max_turns*2-1)
        return [loads(item) for item in raw[::-1]]

    def save_context(self, user_id, user_msg, bot_msg):
        """保存最新对话记录"""
        key = f"qqbot:{user_id}:history"

        # 使用 Redis 链表存储
        pipe = self.client.pipeline()
        pipe.rpush(key, dumps({"role": "user", "content": user_msg}))
        pipe.rpush(key, dumps({"role": "assistant", "content": bot_msg}))
        pipe.ltrim(key, -10, -1)  # 保留最近 5 轮对话
        pipe.expire(key, 3600*24) # 设置 TTL
        pipe.execute()

QQ 消息频率限制：个人账号每分钟 20 条，群聊每分钟 3 条
实现策略：
使用令牌桶算法控制发送速率
重要消息优先队列
错误自动退避重试

对话历史摘要：对长上下文生成摘要而非完整保存
模型选择：简单场景使用 text-davinci-003 替代 ChatGPT
缓存机制：对常见问题答案建立本地缓存

三级过滤体系：
前置关键词黑名单
中置 AI 内容审核 API
后置人工审核队列
紧急熔断机制：当触发敏感词时立即终止会话

实现带工具调用的多轮对话：

from langchain.chains import ConversationChain
from langchain.memory import RedisChatMessageHistory

class EnhancedChatService:
    def __init__(self):
        self.llm = ChatOpenAI(temperature=0)

    def get_chain(self, user_id):
        history = RedisChatMessageHistory(
            session_id=user_id,
            url="redis://localhost:6379/0"
        )

        return ConversationChain(
            llm=self.llm,
            memory=history,
            prompt=PROMPT_TEMPLATE
        )