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 协议兼容框架为例流程是这样的登录着 QQ 账号的框架进程持续运行。当账号收到一条新消息框架把消息事件包装成固定格式的 JSON通过 WebSocket 推送给你的业务进程。你的业务代码解析 JSON提取出消息内容、发送者、群 ID 等信息。你的代码调用 AI 模型 API得到回复文本。你的代码再构造一个发送消息的请求 JSON用 WebSocket 发回给框架。框架把消息发送到对应的私聊或群聊会话。整个链路中你只需要聚焦在第二步到第五步。一个典型的 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 均可Python3.8 及以上QQ 账号一个用于机器人的账号建议使用小号机器人框架任选一款支持 OneBot 协议、支持 WebSocket 服务端的框架AI API任意 OpenAI 兼容接口的 Key这里给一个诚意提醒机器人登录的账号不要用主号。无论使用哪种框架第三方协议登录都存在账号风控风险。用一个小号即使出现问题也不会影响日常社交。3.2 安装 Python 依赖创建一个项目目录并创建虚拟环境避免依赖污染系统 Pythonmkdir 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-client1.6.0 requests2.31.0 python-dotenv1.0.03.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。你需要先做几个判断事件类型是否为message过滤掉通知、请求等其他事件。消息是群聊还是私聊决定回复发往哪里。是否命中了指令前缀例如/ai避免机器人回复群内所有无关消息。这只是最基础的一个过滤但已经能避免绝大多数失控情况。很多新手写的机器人一上线就疯狂回复就是因为没有做事件过滤。4.3 调用 AI 模型的注意事项调用模型接口属于耗时操作网络延迟可能在 1 到 10 秒甚至更久。因此不要让主线程卡在 HTTP 请求上。本文示例中会把每条消息的处理放到独立线程里保证收到多条消息时能并发处理。此外AI 接口不是一定成功的。网络抖动、限流、超时都可能导致异常。代码里必须捕获异常并给用户一个可读的提示而不是让程序直接崩溃。5. 完整代码实现让机器人学会调用 AI5.1 配置文件 .env在项目根目录创建.env文件WS_URLws://127.0.0.1:3001 AI_API_URLhttps://api.openai.com/v1/chat/completions AI_API_KEY你的APIKey AI_MODELgpt-3.5-turbo参数说明配置项作用WS_URL机器人框架的 WebSocket 服务端地址AI_API_URLAI 模型服务的接口地址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: fBearer {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, headersheaders, jsonpayload, timeout60) 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 fAI 调用失败{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(targethandle_message, args(ws, message), daemonTrue).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_messageon_message, on_openon_open, on_erroron_error, on_closeon_close, ) ws.run_forever() if __name__ __main__: main()5.3 代码逻辑拆解先看call_ai函数。它做了三件事构造带鉴权信息的请求头、组装符合 OpenAI 规范的 payload、把响应中的choices[0].message.content提取出来。这里有一个容易忽略的细节timeout60。模型接口在高峰期响应慢超时时间太短会导致频繁失败太长又会让用户等待过久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 失败时的第一排查原则如果机器人没有任何反应不要先怀疑代码。按下面顺序排查先看框架端机器人 QQ 是否在线框架控制台是否显示 WebSocket 客户端已连接。再看程序端控制台是否打印WebSocket 已连接。手动用一个 WebSocket 调试工具连接WS_URL发送一个测试消息确认框架是否正常推送事件。这三步能把问题从代码错误和环境错误两个大类里快速定位出来。7. 常见问题与排查思路下面整理了 QQ AI 机器人开发里出现频率最高的几个问题。问题现象可能原因排查方式解决方案程序无法连接 WebSocketWS_URL 配置错误或框架未开启 WebSocket 服务检查框架控制台设置确认端口号一致将 WS_URL 修改为框架实际监听地址QQ 收到消息但机器人无反应消息事件格式与 OneBot 标准不同在handle_message里打印完整消息 JSON根据实际字段名调整解析逻辑AI 调用总是超时网络不通或接口响应慢单独写脚本用 requests 测试接口连通性更换网络环境或把 timeout 调大机器人回复内容过长被拆分平台消息长度限制观察返回内容长度和类型对回复做截断或使用分段发送频繁出现操作频繁提示机器人发送消息太密集检查是否有并发循环触发增加消息发送频率限制做全局节流API Key 泄漏到公开仓库.gitignore没配置检查日志和 Git 历史立即吊销 Key重新生成所有文件补加.gitignore7.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触发的指令与 promptAI 接口响应耗时异常堆栈发送结果上线前可以先把日志接入文本文件后续需要再对接日志采集系统。8.5 上线前的安全检查机器人如果要在群里长期运行必须明确它的安全边界不要把系统命令、文件读取等能力暴露给群友。对 AI 返回内容做长度和关键词过滤避免意外发送违规内容。为机器人设置指令白名单避免任何人通过自然语言操作危险功能。定期检查 API 调用记录发现异常使用及时处理。如果允许群友让机器人执行代码或脚本这已经属于高风险功能需要更强的权限隔离不建议在 QQ 机器人里开放。8.6 部署到云服务器的建议本地电脑不可能 24 小时在线正式使用建议部署到云服务器。部署时注意使用 systemd 或 Docker 托管 Python 进程崩溃后自动重启。机器人框架和业务代码尽可能在同一内网环境减少网络抖动。定期备份配置文件但绝不能备份 API Key 到不安全的地方。9. 总结与后续学习方向这个项目的本质并不复杂框架解决 QQ 协议问题业务代码解决 AI 调用问题两者通过标准 JSON 消息通信。当你理解了这条数据流就会发现所谓的QQ AI 机器人并不是一个神秘产品而是一个消息转发服务加上一个模型调用接口的组合。接下来值得继续深入的方向有三个一是把单轮问答扩展成多轮对话加入上下文管理和用户会话隔离二是设计一套指令系统让机器人支持多个指令而不是只响应/ai三是引入消息队列和本地缓存提升并发处理能力和响应速度。跑通一个最小版本很容易把它运行得稳定、安全、有边界才是真正的挑战。建议你先在纯私聊环境里测试再逐步开放到群聊。每一步改动都先在测试 QQ 号上验证确认没有异常后再用到正式环境。上述方案基于 OneBot 协议通用的 WebSocket 接口实现不绑定特定框架。如果你现在使用的是其他协议方案只要消息事件格式相近这套代码的改造成本也不会太高。

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

相关推荐

C++ std::map反向遍历全攻略：从rbegin到正迭代器模拟

高速可回收模块化靶标无人机总体设计方案解析

船岸一体边缘智能感知网络：摄像头接入与自定义算法加载全解析

更多资讯

SystemVerilog接口与功能覆盖率实战：从modport到covergroup

零基础学C语言：编译、指针与内存实战指南

树莓派 Pico ADC 探秘：从寄存器到 FIFO、DMA 与过采样

AI PCB高阶升级：从材料到工艺的全链条解析

用C#构建工业级文档管理系统：从架构设计到实践

eLLM：长程推理场景下CPU如何逆袭GPU？技术拆解与实测

今日更新

在S3C2410上移植Nucleus RTOS：启动、驱动与实时性调优实战

2026年Turnitin降重降AI二合一红黑榜：能过和别碰

2026期刊投稿AI写作工具实测：6款打分谁更值得用

本周精选

超人会飞不算本事：系统稳定依赖清晰规则与边界设计

超人VS蜘蛛侠：拆解超级IP的影响力与传播方法论

基于CNN的调制信号识别：MATLAB实现时频图分类实战

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？