Python打造QQ AI机器人：OneBot协议+WebSocket接入大模型

如果你是一个 QQ 群的群主或管理员，大概率遇到过这类场景：群成员反复问同一个问题，你想找一个能自动回复、查资料、甚至帮忙写代码的机器人，但查了一圈资料后，发现要么是已经失效的旧教程，要么是封装得过于黑盒的商业产品。

自己动手做一个 QQ 机器人，在很多人的印象里是技术宅专属玩具，需要搞懂复杂的协议、绕过各种风控规则、还要有一台二十四小时在线的服务器。实际上，这个门槛已经被大幅压缩了。如今成熟的机器人框架承担了绝大部分通信和协议工作，你真正需要写的，只是一段接收消息、调用 AI 接口、再发送回复的业务代码。

这篇文章要做的就是把这个过程拆开：用 Python 实现一个最小可用的 QQ AI 机器人，核心逻辑是以 WebSocket 方式连接一个 OneBot 协议兼容的机器人框架，然后把收到的群聊或私聊消息转发给大模型接口，拿到回答后再发回 QQ。准备比较充分的前提下，5 分钟可以跑通；从零开始安装环境、申请模型 API，半小时内也能完成。

读完这篇文章，你会得到一个能实际对话的私聊机器人，同时理解 QQ 机器人接入 AI 模型的完整数据流，以及把它扩展成多轮对话、权限管理、正式部署时的关键思路。

1. 为什么现在做 QQ 机器人变得简单了

先说结论：今天做 QQ 机器人，真正的难点已经不是写代码，而是理解消息事件与 AI 调用之间的数据流。

过去开发者要面对的是一堆历史遗留问题：QQ 协议不公开、第三方库容易被风控、消息格式混乱。每出一个新版本，可能就要重新适配。这也导致很多人一想到 QQ 机器人，第一反应就觉得"太麻烦"。

现在不一样了。社区里已经沉淀出不少基于 OneBot 协议标准的机器人框架。这类框架负责完成 QQ 账号的登录、消息接收、事件上报和消息发送，对外提供统一的 WebSocket 或 HTTP 接口。你的业务代码只需要按照 OneBot 规范收发 JSON 消息，不需要关心 QQ 协议内部细节。

AI 接入这一侧同样在被标准化。大多数大模型服务商都提供了 OpenAI 兼容的 HTTP 接口：你只要构造一个 messages 数组，POST 给指定 URL，就能拿到模型的文本回复。这意味着，你的机器人代码可以在不同模型之间切换，只需要修改配置，而不需要改动业务逻辑。

所以，"5 分钟"这个说法是有明确前提的：框架已经下载好、账号已经登录、模型 API 已经可用。剩下的就是把两块能力粘起来，这也就是文章标题里"AI 模型接入"要解决的问题。

2. 核心概念：OneBot 协议与 AI 模型接入原理

2.1 QQ 机器人是如何收到消息的

要理解项目原理，先搞清一条消息从 QQ 群到你的代码之间发生了什么。

以 OneBot 协议兼容框架为例，流程是这样的：

1. 登录着 QQ 账号的框架进程持续运行。
2. 当账号收到一条新消息，框架把消息事件包装成固定格式的 JSON，通过 WebSocket 推送给你的业务进程。
3. 你的业务代码解析 JSON，提取出消息内容、发送者、群 ID 等信息。
4. 你的代码调用 AI 模型 API，得到回复文本。
5. 你的代码再构造一个"发送消息"的请求 JSON，用 WebSocket 发回给框架。
6. 框架把消息发送到对应的私聊或群聊会话。

整个链路中，你只需要聚焦在第二步到第五步。

一个典型的 OneBot 消息事件 JSON 长这样：

{ "post_type": "message", "message_type": "group", "group_id": 123456789, "user_id": 987654321, "raw_message": "/ai 用Python写一个快速排序", "self_id": 555666777 }

字段含义很直观：

字段	含义
post_type	事件类型，message 表示消息事件
message_type	消息类型，group 表示群聊，private 表示私聊
group_id	群号
user_id	发送者 QQ 号
raw_message	原始消息文本
self_id	机器人自己的 QQ 号

2.2 AI 模型接入的通用方式

所谓"AI 模型接入"，本质就是把上面解析出来的消息文本，作为 prompt 发给模型接口，拿到响应后再发回 QQ。

当前主流的模型服务商大多提供 OpenAI 兼容的 Chat Completions 接口。核心请求体长这样：

{ "model": "gpt-3.5-turbo", "messages": [ {"role": "user", "content": "你好"} ] }

调用后返回：

{ "choices": [ { "message": { "role": "assistant", "content": "你好，有什么可以帮你？" } } ] }

这个接口模式有很强的通用性。不管底层是 GPT、通义、文心还是其他兼容 OpenAI 规范的服务，只要 base_url、API Key 和模型名配置正确，代码几乎不用改。

2.3 为什么选择 WebSocket 方式

机器人框架与业务代码之间有两种常见通信方式：HTTP 反向上报和 WebSocket。这篇文章采用 WebSocket 直连，理由有三点：

• 长连接减少了 HTTP 握手开销。
• 框架主动推送事件，代码结构更贴近回调模型。
• 本地调试时能直观看到消息输出。

理解了这个基础架构，后面写代码就只是机械工作了。

3. 环境准备与前置条件

先说明：为了让教程聚焦在 AI 接入本身，机器人框架部分不展开配置细节。下面列出的是本文示例代码需要的最小环境。

3.1 软硬件要求

项目	要求
操作系统	Windows / macOS / Linux 均可
Python	3.8 及以上
QQ 账号	一个用于机器人的账号，建议使用小号
机器人框架	任选一款支持 OneBot 协议、支持 WebSocket 服务端的框架
AI API	任意 OpenAI 兼容接口的 Key

这里给一个诚意提醒：机器人登录的账号不要用主号。无论使用哪种框架，第三方协议登录都存在账号风控风险。用一个小号，即使出现问题也不会影响日常社交。

3.2 安装 Python 依赖

创建一个项目目录，并创建虚拟环境，避免依赖污染系统 Python：

mkdir qq-ai-bot cd qq-ai-bot python -m venv venv

激活虚拟环境：

# Windows venv\Scripts\activate # macOS / Linux source venv/bin/activate

然后安装依赖：

pip install websocket-client requests python-dotenv

三个库的用途分别是：

• websocket-client：连接机器人框架的 WebSocket 服务端。
• requests：调用 AI 模型的 HTTP 接口。
• python-dotenv：从.env文件读取配置，避免把 API Key 写死在代码里。

依赖安装完成后，可以新建一个requirements.txt方便以后复现：

websocket-client>=1.6.0 requests>=2.31.0 python-dotenv>=1.0.0

3.3 准备 AI 模型 API Key

这一步取决于你使用哪家模型服务。通用的做法是在服务商控制台创建一个 API Key，并拿到接口的 base_url 和模型名称。

需要注意：API Key 属于敏感凭证，不要提交到公开仓库，也不要写在代码里。本文的做法是放在.env文件中，并将.env加入.gitignore。

4. 核心流程拆解：从消息到回复的完整链路

在写代码之前，先看清整个项目的文件结构和消息走向，这会直接影响后面排错效率。

4.1 项目文件结构

qq-ai-bot/ ├── venv/ # Python 虚拟环境 ├── .env # 环境配置，不入库 ├── requirements.txt # 依赖清单 └── bot.py # 业务逻辑主程序

4.2 消息处理的关键分支

程序收到 WebSocket 推送后，不能直接一股脑全部转发给 AI。你需要先做几个判断：

1. 事件类型是否为message，过滤掉通知、请求等其他事件。
2. 消息是群聊还是私聊，决定回复发往哪里。
3. 是否命中了指令前缀，例如/ai，避免机器人回复群内所有无关消息。

这只是最基础的一个过滤，但已经能避免绝大多数失控情况。很多新手写的机器人一上线就疯狂回复，就是因为没有做事件过滤。

4.3 调用 AI 模型的注意事项

调用模型接口属于耗时操作，网络延迟可能在 1 到 10 秒甚至更久。因此不要让主线程卡在 HTTP 请求上。本文示例中会把每条消息的处理放到独立线程里，保证收到多条消息时能并发处理。

此外，AI 接口不是一定成功的。网络抖动、限流、超时都可能导致异常。代码里必须捕获异常并给用户一个可读的提示，而不是让程序直接崩溃。

5. 完整代码实现：让机器人学会调用 AI

5.1 配置文件 .env

在项目根目录创建.env文件：

WS_URL=ws://127.0.0.1:3001 AI_API_URL=https://api.openai.com/v1/chat/completions AI_API_KEY=你的APIKey AI_MODEL=gpt-3.5-turbo

参数说明：

配置项	作用
WS_URL	机器人框架的 WebSocket 服务端地址
AI_API_URL	AI 模型服务的接口地址
AI_API_KEY	调用模型接口的密钥
AI_MODEL	使用的模型名称

如果你使用的是国内可访问的 OpenAI 兼容服务，把AI_API_URL和AI_API_KEY替换成对应服务的地址和密钥即可。

5.2 主程序 bot.py

下面是完整的机器人业务代码：

# 文件路径：bot.py import json import os import threading import requests import websocket from dotenv import load_dotenv load_dotenv() # 读取环境配置 WS_URL = os.getenv("WS_URL", "ws://127.0.0.1:3001") AI_API_URL = os.getenv("AI_API_URL", "https://api.openai.com/v1/chat/completions") AI_API_KEY = os.getenv("AI_API_KEY", "") AI_MODEL = os.getenv("AI_MODEL", "gpt-3.5-turbo") def call_ai(prompt: str) -> str: """调用大模型接口，返回回复文本""" headers = { "Authorization": f"Bearer {AI_API_KEY}", "Content-Type": "application/json", } payload = { "model": AI_MODEL, "messages": [{"role": "user", "content": prompt}], "temperature": 0.7, } resp = requests.post(AI_API_URL, headers=headers, json=payload, timeout=60) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]["content"] def handle_message(ws: websocket.WebSocketApp, message: str): """处理一条收到的消息""" msg = json.loads(message) print("[收到消息]", msg) # 只处理消息事件 if msg.get("post_type") != "message": return message_type = msg.get("message_type", "private") user_id = msg.get("user_id") group_id = msg.get("group_id") raw_message = msg.get("raw_message", "") # 指令过滤，只处理以 /ai 开头的消息 if not raw_message.strip().startswith("/ai"): return prompt = raw_message.strip()[3:].strip() if not prompt: prompt = "你好，请介绍一下你自己" try: reply = call_ai(prompt) except Exception as e: print("[AI 调用失败]", e) reply = f"AI 调用失败：{e}" # 构造发送消息的请求 send_msg = { "action": "send_group_msg" if message_type == "group" else "send_private_msg", "params": { "user_id": user_id, "group_id": group_id, "message": reply, }, } ws.send(json.dumps(send_msg)) print("[已发送回复]", reply) def on_message(ws: websocket.WebSocketApp, message: str): """WebSocket 收到消息后，开新线程处理，避免阻塞""" threading.Thread(target=handle_message, args=(ws, message), daemon=True).start() def on_open(ws: websocket.WebSocketApp): print("WebSocket 已连接，等待 QQ 消息...") def on_error(ws: websocket.WebSocketApp, error): print("WebSocket 错误", error) def on_close(ws: websocket.WebSocketApp, close_status_code, close_msg): print("WebSocket 连接关闭") def main(): ws = websocket.WebSocketApp( WS_URL, on_message=on_message, on_open=on_open, on_error=on_error, on_close=on_close, ) ws.run_forever() if __name__ == "__main__": main()

5.3 代码逻辑拆解

先看call_ai函数。它做了三件事：构造带鉴权信息的请求头、组装符合 OpenAI 规范的 payload、把响应中的choices[0].message.content提取出来。这里有一个容易忽略的细节：timeout=60。模型接口在高峰期响应慢，超时时间太短会导致频繁失败，太长又会让用户等待过久，60 秒是个相对稳妥的初始值。

再看handle_message函数。第一层过滤是post_type，这可以拦截掉好友申请、群成员变动等非消息事件。第二层过滤是/ai前缀，防止机器人回复群里所有聊天内容。群聊场景下这一步是刚需，否则机器人会被刷屏，也容易触发平台的频繁操作限制。

最后，on_message里用threading.Thread处理每条消息。一旦变成多线程，要注意共享连接对象ws的并发写。WebSocket 客户端库本身会处理发送排队，但如果你在多个线程里修改同一个对象的状态，仍然可能出现竞态。本文示例只调用ws.send，是安全的；如果你要维护共享会话上下文，就一定要加锁。

5.4 启动脚本

为了方便，创建一个简单的启动入口：

python bot.py

正常情况下，你会看到控制台输出：

WebSocket 已连接，等待 QQ 消息...

这说明程序已经成功连接上了机器人框架。

6. 运行结果与效果验证

6.1 测试场景设计

程序启动后，用另一个 QQ 号给机器人账号发一条私聊消息：

/ai 用Python写一个判断回文数的函数

如果一切正常，机器人会调用模型接口，然后在 QQ 对话框里返回类似下面的内容：

def is_palindrome(x): s = str(x) return s == s[::-1]

同时，运行程序的终端会打印两行日志：

[收到消息] {'post_type': 'message', 'message_type': 'private', 'user_id': 12345, 'raw_message': '/ai 用Python写一个判断回文数的函数', ...} [已发送回复] def is_palindrome(x): s = str(x) return s == s[::-1]

日志中的 JSON 内容可能比上面更复杂，这是正常的，因为不同框架上报的字段不完全一致。你只需要关注这几个关键字段是否出现在日志里。

6.2 群聊测试

在 QQ 群里 @机器人并发送：

/ai 推荐三本入门Python的书

机器人会识别到message_type为group，把回复发送到当前群。

这里要特别提醒：不要把机器人群聊权限设置成"任何人可触发"。在真实项目中，至少要加一个群号白名单，否则任何群里都能通过你的机器人消耗模型 API 额度。

6.3 失败时的第一排查原则

如果机器人没有任何反应，不要先怀疑代码。按下面顺序排查：

1. 先看框架端：机器人 QQ 是否在线，框架控制台是否显示 WebSocket 客户端已连接。
2. 再看程序端：控制台是否打印WebSocket 已连接。
3. 手动用一个 WebSocket 调试工具连接WS_URL，发送一个测试消息，确认框架是否正常推送事件。

这三步能把问题从"代码错误"和"环境错误"两个大类里快速定位出来。

7. 常见问题与排查思路

下面整理了 QQ AI 机器人开发里出现频率最高的几个问题。

问题现象	可能原因	排查方式	解决方案
程序无法连接 WebSocket	WS_URL 配置错误，或框架未开启 WebSocket 服务	检查框架控制台设置，确认端口号一致	将 WS_URL 修改为框架实际监听地址
QQ 收到消息但机器人无反应	消息事件格式与 OneBot 标准不同	在handle_message里打印完整消息 JSON	根据实际字段名调整解析逻辑
AI 调用总是超时	网络不通或接口响应慢	单独写脚本用 requests 测试接口连通性	更换网络环境，或把 timeout 调大
机器人回复内容过长被拆分	平台消息长度限制	观察返回内容长度和类型	对回复做截断，或使用分段发送
频繁出现"操作频繁"提示	机器人发送消息太密集	检查是否有并发循环触发	增加消息发送频率限制，做全局节流
API Key 泄漏到公开仓库	.gitignore没配置	检查日志和 Git 历史	立即吊销 Key，重新生成；所有文件补加.gitignore

7.1 关于 JSON 解析失败

这是新手最容易遇到的问题。OneBot 框架版本较多，不同框架、不同版本的字段存在差异。比如有的框架上报的message字段是纯文本，有的则是包含type和data的数组。

一个稳妥做法是打印原始message字段，根据实际格式调整解析函数。千万不要假设所有框架上报格式完全一致。

7.2 关于账号风控

任何基于第三方协议的机器人都有账号风险。控制风险的关键是：控制消息频率，避免机器人短时间内大量主动发消息；不要在第一天上线就高频运行；准备好备用 QQ 号。如果账号被限制，第一时间停止机器人进程。

8. 最佳实践与工程建议

跑通一个最小机器人只是开始。如果这个机器人要进入真实使用场景，下面这些工程实践能帮你少踩很多坑。

8.1 把配置和代码分离

文章示例已经把 API Key 放进了.env，这是最低要求。更进一步，建议把以下内容全部配置化：

• 指令前缀
• 允许触发的群号列表
• 允许使用的用户白名单
• 模型名称与 temperature 参数
• 最大回复长度

配置化能让你在不改代码的前提下调整机器人的行为。对于没有自动化运维经验的团队，这一步尤其重要。

8.2 为多轮对话设计上下文管理

当前示例只会把单条用户消息发给模型。如果要实现多轮对话，需要保存每个会话的 messages 历史，并把历史一起传给模型。这里要注意两点：

• 不同用户、不同群的上下文要隔离，用group_id + user_id作为会话 key。
• 上下文的长度不能无限增长，超过模型上下文窗口时，要采用滑动窗口策略，只保留最近 N 条消息。

例如用一个简单的字典加列表来维护上下文：

sessions = {} def get_session(key: str) -> list: if key not in sessions: sessions[key] = [] return sessions[key] def update_session(key: str, messages: list, max_len: int = 10): sessions[key] = messages[-max_len:]

注意，多线程环境下操作sessions字典时要加锁，防止并发写导致数据错乱。

8.3 频率限制与额度保护

模型 API 是按调用量计费的，机器人越活跃，消耗越大。因此线上机器人必须有频率限制策略。常见做法：

• 每个用户每分钟最多调用 N 次。
• 每个群每分钟最多调用 M 次。
• 对超过限制的请求直接返回提示，不再调用模型。

可以借助简单的时间窗口计数实现，不需要引入额外中间件。代码量不大，但能避免 API 额度被恶意耗尽。

8.4 日志与可观测性

不要只看print输出。把日志改为结构化输出后，排查问题的效率会提升很多。建议至少记录以下几项：

• 完整的消息事件 JSON
• 触发的指令与 prompt
• AI 接口响应耗时
• 异常堆栈
• 发送结果

上线前可以先把日志接入文本文件，后续需要再对接日志采集系统。

8.5 上线前的安全检查

机器人如果要在群里长期运行，必须明确它的安全边界：

• 不要把系统命令、文件读取等能力暴露给群友。
• 对 AI 返回内容做长度和关键词过滤，避免意外发送违规内容。
• 为机器人设置指令白名单，避免任何人通过自然语言操作危险功能。
• 定期检查 API 调用记录，发现异常使用及时处理。

如果允许群友让机器人执行代码或脚本，这已经属于高风险功能，需要更强的权限隔离，不建议在 QQ 机器人里开放。

8.6 部署到云服务器的建议

本地电脑不可能 24 小时在线，正式使用建议部署到云服务器。部署时注意：

• 使用 systemd 或 Docker 托管 Python 进程，崩溃后自动重启。
• 机器人框架和业务代码尽可能在同一内网环境，减少网络抖动。
• 定期备份配置文件，但绝不能备份 API Key 到不安全的地方。

9. 总结与后续学习方向

这个项目的本质并不复杂：框架解决 QQ 协议问题，业务代码解决 AI 调用问题，两者通过标准 JSON 消息通信。当你理解了这条数据流，就会发现所谓的"QQ AI 机器人"并不是一个神秘产品，而是一个消息转发服务加上一个模型调用接口的组合。

接下来值得继续深入的方向有三个：一是把单轮问答扩展成多轮对话，加入上下文管理和用户会话隔离；二是设计一套指令系统，让机器人支持多个指令，而不是只响应/ai；三是引入消息队列和本地缓存，提升并发处理能力和响应速度。

跑通一个最小版本很容易，把它运行得稳定、安全、有边界才是真正的挑战。建议你先在纯私聊环境里测试，再逐步开放到群聊。每一步改动都先在测试 QQ 号上验证，确认没有异常后再用到正式环境。

上述方案基于 OneBot 协议通用的 WebSocket 接口实现，不绑定特定框架。如果你现在使用的是其他协议方案，只要消息事件格式相近，这套代码的改造成本也不会太高。