从零到一：用Python和WebSocket打造AI驱动的QQ机器人

最近在开发者社群里越来越多人在问同一个问题能不能做一个 QQ 机器人把大模型接进去让单聊或群聊里的小伙伴直接 机器人提问、让它写文案、查资料、做翻译答案是可以而且整个链路并不复杂。网上关于 QQ 机器人的资料大多比较零散要么是几年前的旧方案要么只讲了半截卡在鉴权、事件订阅或者消息格式上就没下文了。本文会从零开始带你走一遍完整的接入流程包含环境准备、QQ 开放平台配置、WebSocket 事件订阅、AI 模型接口调用以及完整的 Python 示例代码。只要你的账号和配置都齐全按步骤操作5 分钟内就能跑通一个能对话的 AI QQ 机器人。1. 背景与核心概念1.1 QQ 机器人是什么QQ 机器人本质上是一个运行在服务端的程序它通过 QQ 官方提供的开放接口接入 QQ 的单聊、群聊等场景。用户给机器人发消息或者在某些特定场景下触发事件机器人的服务端就会收到一条事件通知然后根据业务逻辑做出响应比如回复一段文本、发送一张图片、执行一个指令。早期的 QQ 机器人大多是“关键词回复机”配置一些固定的规则比如用户发“天气”就返回天气信息。但随着大模型能力的普及机器人已经从“规则引擎”进化成“AI 助手”用户只需要发送自然语言机器人把消息转发给大模型再把模型返回的结果带回聊天窗口。这种模式让机器人具备了很强的泛化能力不需要针对每句话写死规则也不需要维护大量的问答库。1.2 为什么选择官方机器人平台做 QQ 机器人目前社区里主要有两条路线。第一条是 QQ 官方开放平台路线。你需要在 QQ 开放平台注册开发者账号、创建机器人应用拿到 AppID 和 Token然后通过官方提供的 WebSocket 或 Webhook 方式接收事件再调用官方 API 发送消息。官方路线的优势是稳定、合规、功能边界清晰适合希望正式上线、被更多用户使用的场景。第二条是使用第三方协议库通过模拟客户端登录的方式接入 QQ。这类方案功能灵活但存在账号安全和平台封禁风险而且本质上绕过了官方接口约束不适合正式项目和商业场景。本文推荐并演示官方路线。整个开发过程可以拆成三个环节接收用户消息、调用 AI 模型接口、把结果发送到聊天窗口。其中接收和发送都依赖 QQ 官方机器人 APIAI 能力则来自你选择的大模型服务。1.3 AI 模型接入的通用链路AI 模型接入不限定具体某个大模型。目前市面上主流的大模型服务商基本都提供了 OpenAI 兼容的 HTTP 接口也就是POST /v1/chat/completions这种协议。这意味着我们的代码只需要写一次就可以通过替换接口地址、API Key 和模型名切换到不同的大模型服务上。核心链路如下用户向机器人发送一条消息。机器人服务端通过 WebSocket 收到消息事件解析出发送者、群组信息和消息内容。机器人把消息内容整理成 Prompt提示词调用大模型接口。大模型返回回复内容。机器人调用 QQ 官方发送消息的 API把回复内容发送到对应的聊天窗口。这个链路在任何 AI 机器人项目中都是通用的区别只在于“消息从哪来、回复到哪去”。2. 环境准备与账号申请2.1 开发环境说明本文示例代码使用 Python 编写。建议使用 Python 3.9 或更高版本如果你本机已经安装了 Python 环境可以直接使用如果还没有安装请先到 Python 官网下载对应操作系统的安装包。需要安装的依赖有两个websockets用于和 QQ 官方服务器建立 WebSocket 长连接接收消息事件。httpx或requests用于调用 AI 大模型接口和 QQ 发送消息 API。考虑到性能推荐使用支持异步的httpx。安装命令如下pip install websockets httpxWindows 用户如果遇到安装问题可以检查 Python 是否已加入 PATH或者使用python -m pip install websockets httpx。2.2 注册开发者账号并创建机器人应用打开 QQ 开放平台q.qq.com使用 QQ 扫码登录。首次使用需要完成开发者认证按照页面提示填写基本信息即可。登录后进入“机器人”管理页面点击“创建机器人”。这里需要填写机器人的头像、名称、功能介绍等信息。创建完成后你会进入机器人的管理后台里面有这个机器人的基本配置和凭证信息。请注意创建机器人的审核可能需要一些时间。在等待审核期间你可以先把本地开发环境准备好。本文提到的“5 分钟跑通”前提是账号、机器人应用和凭证都已经准备好真正从零到代码启动所花的时间。2.3 获取 AppID 与 Token在机器人管理后台你能看到两个关键凭证AppID机器人的唯一标识。AppSecret / Token调用接口时用于鉴权的密钥。在官方机器人平台上WebSocket 连接时使用的 Token 格式通常为QQBot {AppID}:{AppSecret}中间用冒号连接。发送消息调用 REST API 时则需要把 Token 放在 HTTP 的Authorization请求头中。需要特别注意Token 等同于机器人的密码不要把它硬编码在前端代码中也不要提交到公开的 Git 仓库里。建议通过环境变量或者本地的配置文件加载。2.4 准备一个可调用的 AI 模型接口AI 大模型的部分你需要提前准备好以下三项信息API 接口地址例如你的大模型服务商提供的兼容 OpenAI 协议的接口地址。API Key你的私有密钥调用接口时放在请求头中。模型名称例如你想要使用的对话模型标识。如果你暂时没有可用的 API Key可以先用一个简单的本地测试函数作为替代代码结构完全一样只是call_ai()函数返回固定文本。这样同样能验证 QQ 机器人链路是否通畅。3. 核心原理WebSocket 事件订阅与消息发送3.1 为什么使用 WebSocket 而不是轮询QQ 机器人接入官方平台时有两种接收事件的方式Webhook 和 WebSocket。Webhook 模式需要你提供一个公网可访问的 HTTPS 地址QQ 服务器把事件 POST 到你的接口上。这种模式适合已经有公网服务器和域名备案的开发者。WebSocket 模式则是由你的程序主动去连接 QQ 服务器建立一条长连接消息事件会实时推送到这条连接上。它不需要公网 IP本地开发也能测试非常适合新手和快速验证场景。本文采用 WebSocket 模式。3.2 连接鉴权与事件声明WebSocket 连接建立之后客户端需要发送一个identify包这个包中包含机器人的 Token 以及需要订阅的事件类型。QQ 服务器校验通过后会返回一个ready事件表示连接成功。在我们这个场景中需要订阅的是消息事件C2C_MESSAGE_CREATE用户单聊机器人时触发。GROUP_AT_MESSAGE_CREATE用户在群里 机器人时触发。这些事件类型的宏定义值以官方文档为准代码中只需要在 identify 时声明对应的 intents 即可。3.3 消息事件的数据结构当用户在 QQ 中向机器人发消息时WebSocket 会收到一个事件数据包。其中核心字段包括消息 IDid每条消息的唯一标识发送被动回复时需要携带。消息内容content用户发来的文本内容。发送者信息author包含用户的 openid。频道信息对于群聊包含group_openid对于单聊包含user_openid。我们只需要从事件中取出这些字段就能完成一次完整的对话闭环。4. 完整实战5 分钟跑通一个 AI 对话机器人4.1 初始化项目结构在本地创建一个项目目录比如命名为qq-ai-bot目录结构如下qq-ai-bot/ ├── bot.py ├── config.py └── requirements.txtrequirements.txt内容websockets httpx在config.py中集中管理所有配置项。这样做的目的是把密钥和代码分离方便后期维护。# 文件路径config.py import os # QQ 机器人配置 APP_ID os.getenv(QQ_APP_ID, 你的AppID) APP_TOKEN os.getenv(QQ_APP_TOKEN, 你的AppSecret) # AI 大模型配置 AI_API_URL os.getenv(AI_API_URL, https://api.example.com/v1/chat/completions) AI_API_KEY os.getenv(AI_API_KEY, 你的大模型API Key) AI_MODEL os.getenv(AI_MODEL, 你的模型名称) # WebSocket 地址以官方文档为准 WS_URL os.getenv(QQ_WS_URL, wss://api.sgroup.qq.com/websocket)4.2 编写 AI 调用模块AI 调用模块负责把用户的 Prompt 发送给大模型接口并返回模型生成的文本。# 文件路径ai_client.py import httpx async def call_ai(prompt: str) - str: 调用大模型接口返回模型回复的文本。 使用 OpenAI 兼容格式的 chat/completions 接口。 headers { Authorization: fBearer {AI_API_KEY}, Content-Type: application/json, } payload { model: AI_MODEL, messages: [ {role: system, content: 你是一个友善的智能助手请用简洁准确的中文回答用户问题。}, {role: user, content: prompt}, ], temperature: 0.7, } async with httpx.AsyncClient(timeout30) as client: resp await client.post(AI_API_URL, jsonpayload, headersheaders) resp.raise_for_status() data resp.json() return data[choices][0][message][content]这里将整个请求封装成一个异步函数方便在 WebSocket 事件处理中直接调用。超时设置为 30 秒避免大模型响应过慢导致连接卡死。接口返回格式是 OpenAI 兼容结构如果你的平台返回格式不同只需要修改解析部分。4.3 编写消息发送模块QQ 机器人发送消息需要使用官方 REST API。单聊和群聊的接口路径不同但请求方式一致。# 文件路径qq_api.py import httpx async def send_private_message(user_openid: str, content: str, msg_id: str) - None: 发送单聊消息 url fhttps://api.sgroup.qq.com/v2/users/{user_openid}/messages headers { Authorization: fQQBot {APP_ID}:{APP_TOKEN}, Content-Type: application/json, } payload { content: content, msg_type: 0, msg_id: msg_id, } async with httpx.AsyncClient(timeout10) as client: resp await client.post(url, jsonpayload, headersheaders) resp.raise_for_status() async def send_group_message(group_openid: str, content: str, msg_id: str) - None: 发送群聊消息 url fhttps://api.sgroup.qq.com/v2/groups/{group_openid}/messages headers { Authorization: fQQBot {APP_ID}:{APP_TOKEN}, Content-Type: application/json, } payload { content: content, msg_type: 0, msg_id: msg_id, } async with httpx.AsyncClient(timeout10) as client: resp await client.post(url, jsonpayload, headersheaders) resp.raise_for_status()接口地址和字段可能会随官方版本调整建议以 QQ 开放平台文档中“发送消息”章节的最新说明为准。这里的msg_id是被动回复消息时需要的原始消息 ID用来关联上下文避免机器人被判定为主动骚扰消息。4.4 编写 WebSocket 主程序接下来是核心部分建立 WebSocket 连接、接收事件、解析消息、调用 AI、回发结果。# 文件路径bot.py import asyncio import json import websockets from config import APP_ID, APP_TOKEN, WS_URL from ai_client import call_ai from qq_api import send_private_message, send_group_message # 计算 identify 所需的 intents 值 # 建议直接按官方文档给的事件宏定义换算 INTENTS 0 # 根据你订阅的事件计算示例中先用占位 def build_identify_payload(): return { op: 2, d: { token: fQQBot {APP_ID}:{APP_TOKEN}, intents: INTENTS, shard: [0, 1], }, } async def handle_event(event: dict): 解析 WebSocket 推送的事件区分单聊和群聊消息。 t event.get(t) # 事件类型 d event.get(d, {}) # 事件数据 if t C2C_MESSAGE_CREATE: content d.get(content, ) msg_id d.get(id, ) author d.get(author, {}) user_openid author.get(user_openid, ) if not content or not user_openid: return reply await call_ai(content) await send_private_message(user_openid, reply, msg_id) elif t GROUP_AT_MESSAGE_CREATE: content d.get(content, ) msg_id d.get(id, ) group_openid d.get(group_openid, ) author d.get(author, {}) user_openid author.get(user_openid, ) if not content or not group_openid: return # 实际群聊消息中可能包含 机器人 的占位文本可以在这里做清理 prompt content reply await call_ai(prompt) await send_group_message(group_openid, reply, msg_id) async def heartbeat(ws): 定时发送心跳包保持连接存活。 while True: await asyncio.sleep(30) await ws.send(json.dumps({op: 1, d: None})) async def main(): # 这里需要将 INTENTS 替换为真实值后面说明 global INTENTS # C2C_MESSAGE_CREATE 和 GROUP_AT_MESSAGE_CREATE 对应的 intents 位 INTENTS (1 0) | (1 10) # 需要根据官方宏定义确认具体值 async with websockets.connect(WS_URL) as ws: print(已连接到 QQ WebSocket 服务器) # 发送鉴权包 await ws.send(json.dumps(build_identify_payload())) # 启动心跳任务 asyncio.create_task(heartbeat(ws)) # 持续接收消息 async for raw_msg in ws: data json.loads(raw_msg) op data.get(op, 0) if op 0: # 收到服务端推送的事件 asyncio.create_task(handle_event(data)) elif op 11: # 心跳确认无需处理 pass if __name__ __main__: asyncio.run(main())这段代码中INTENTS的写法是一个简化示例不同版本的事件宏定义可能不同。建议去官方文档找到具体的事件 intents 值并填入常量。核心逻辑已经完整发送 identify 后收到READY事件然后持续处理消息事件。4.5 启动机器人并验证在项目目录中运行python bot.py如果一切正常控制台会输出“已连接到 QQ WebSocket 服务器”随后程序保持运行状态。此时打开你的 QQ找到已经添加为好友的测试机器人发送一条消息比如“你好请介绍一下你自己”。机器人收到消息后会先调用大模型接口再把结果回复给你。如果是群聊场景你需要先在 QQ 开放平台后台把机器人添加到某个群然后在群里 机器人并发送消息。注意官方机器人默认只响应 消息不会对群里所有聊天内容进行回复这样设计主要是为了避免机器人刷屏。整个验证链路分为四步观察控制台是否输出了连接成功信息。在 QQ 中向机器人发消息。观察控制台是否打印了收到的事件日志如果你加了print调试。等待 AI 返回结果。如果控制台没有输出优先检查 Token 格式是否正确、事件订阅是否配置完整。5. 常见问题与排查思路5.1 WebSocket 连接失败或鉴权失败这是新手最容易遇到的问题。控制台常见的报错是连接直接被关闭或者收到一个包含错误码的事件。排查步骤如下检查APP_ID和APP_TOKEN是否填写正确注意不要有额外的空格或换行。确认 Token 拼接格式是QQBot AppID:AppSecret冒号是英文冒号。查看机器人后台是否有“已发布”“沙箱环境”等状态不同状态会影响事件接收范围。5.2 能连接但不收到任何消息这种情况通常不是连接本身的问题而是事件订阅范围或测试环境的问题。常见原因有INTENTS常量没有正确声明导致服务器没有推送对应事件。机器人还处于沙箱环境中只能接收添加了指定测试人员的消息。群聊场景没有 机器人或者机器人没有被添加到目标群。建议在handle_event函数入口加一行print(json.dumps(event, ensure_asciiFalse))先看看是否真的收到了事件。如果确实没有事件进来再回头检查订阅配置。5.3 发送消息接口报错发送消息时如果收到 HTTP 4xx 错误需要关注几个点问题现象常见原因解决思路401 未授权Token 错误或过期重新复制 AppSecret检查格式400 参数错误openid或group_openid为空检查事件解析逻辑429 请求太多触发了频率限制在发送逻辑里加延时或重试200 但消息没发出去被动回复 msg_id 过期消息事件触发后尽快回复特别是被动回复消息官方对时效有要求。如果机器人需要调用 AI 大模型这个过程有时候会比较慢建议在收到事件后先回一条“思考中”或者“正在处理”的提示然后再异步调用 AI 返回结果避免超时。5.4 AI 接口调用慢或超时大模型接口的响应时间通常在几秒到几十秒之间如果你的网络状况不稳定超时概率会更高。解决方式将httpx的timeout参数适当调大。在调用 AI 之前先通过 QQ 接口发送一条“已收到正在思考中”的占位消息。对于超时情况捕获异常后给用户回复“抱歉我暂时没有响应请稍后再试”。6. 最佳实践与工程建议6.1 密钥与配置管理绝对不要把 AppSecret 和 API Key 写死在代码中。推荐使用环境变量或者独立的配置文件并且将包含密钥的文件加入.gitignore避免泄露。示例.env文件不要提交到仓库QQ_APP_ID你的AppID QQ_APP_TOKEN你的AppSecret AI_API_KEY你的大模型API Key AI_API_URLhttps://api.example.com/v1/chat/completions AI_MODEL你的模型名称如果你的项目使用 Git 管理请在仓库中添加.gitignore并忽略.env、config.py等敏感文件。6.2 事件处理中的并发控制在handle_event中使用asyncio.create_task是合理的但如果消息量突然增大可能会同时发起大量 AI 请求。建议使用asyncio.Semaphore限制最大并发数比如同时最多处理 10 条消息避免把 API Key 的额度消耗完也避免被 QQ 服务器判定为异常行为。6.3 机器人回复的合规与安全接入 AI 模型后机器人可能输出一些不确定内容。在生产环境中建议在系统 Prompt 中明确约束机器人的角色和回答边界。对用户输入做长度限制超过一定长度直接截断。对 AI 输出做敏感词过滤。设置单用户或单群的调用频率限制避免被恶意刷量。如果你要把机器人公开发布到 QQ 平台需要遵守官方规则不要在机器人功能中设计任何诱导分享、诱导关注、色情低俗、违法违规、绕过平台限制的内容。审核流程可能比较严格建议提前阅读官方运营规范。6.4 日志与监控本地开发时可以用print输出关键日志但在生产环境需要更完整的日志体系。建议在以下节点打点WebSocket 连接成功与断开。收到消息事件的内容摘要。调用 AI 接口的耗时和返回状态。发送消息的成功与失败。如果使用云服务器部署可以考虑接入文件日志或云日志服务便于问题追踪。6.5 机器人部署建议本地运行适合测试但不稳定电脑关了就断线。如果需要长期运行建议部署到云服务器并使用 systemd、Docker 或进程守护工具来保证机器人进程崩溃后自动重启。Docker 部署的最小示例FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD [python, bot.py]这样镜像构建完成后在任何支持 Docker 的服务器上都能一键启动。7. 总结与下一步学习方向本文从零开始完成了 QQ 机器人接入 AI 模型的闭环介绍了官方机器人的核心概念完成了环境准备和账号申请讲解了 WebSocket 事件订阅原理并给出了完整可运行的 Python 示例代码最后整理了常见问题和工程优化建议。看完并动手实践后你已经可以做出一个具备基础对话能力的 QQ 机器人。接下来可以继续探索的方向包括给机器人增加多轮记忆能力把用户历史消息保存到 Redis 或数据库中。支持图片生成和发送让 AI 不仅能写文字还能画图。接入更多插件能力比如查询天气、翻译、新闻、定时提醒等。在群里做权限控制只允许特定关键词触发 AI 调用。将机器人拆分成多个服务使用消息队列处理高并发场景。AI 机器人的开发门槛已经比过去低很多关键不是“能不能做”而是你想让它解决什么场景下的问题。从一个简单的对话机器人开始逐步扩展成你真正需要的自动化助手这个过程会带来很多乐趣。如果本文对你有帮助欢迎收藏备用也欢迎在评论区分享你的踩坑经验。