之前在做群机器人时,一直想找一个上下文能力强、回答质量高的模型来驱动 QQ 机器人。试过好几个方案,要么上下文太短,多聊几句就“失忆”,要么回答太模板化,放在群聊里显得有些生硬。后来把 Grok 模型接入 QQ 机器人后,对话体验提升了不少,支持长上下文、可调教空间大,而且调用方式走的是标准的 OpenAI 兼容接口,接入成本并不高。这篇文章就把整个接入过程完整拆解一遍,从环境准备、OneBot 协议连接、Python 代码实现,到系统提示词调教、常见报错排查、生产环境注意事项,全部覆盖。想给自己 QQ 群加一个 AI 助手的新手,或者想快速验证 Grok 模型在 IM 场景下效果的同学,都可以照着操作。
1. 背景与核心概念
1.1 Grok 模型是什么
Grok 是 xAI 推出的对话式大模型,设计上强调逻辑推理、代码生成和长文本理解。和普通聊天机器人不同,Grok 在多轮对话中的上下文保持能力比较强,适合用来做需要“记住前文”的 QQ 机器人。
本文标题里提到的 Grok 4.3,你可以把它当成一个具体的模型版本示例。实际上,模型版本更新很快,不同时间点开放的模型名称、上下文长度、价格策略都不一样。本文的核心思路是通用的:只要你的账号能调用某个 Grok 模型,并且拿到 API Key,就可以通过下面的方式接入 QQ 机器人。
1.2 QQ 机器人的几种实现方式
QQ 机器人的实现方案大致分为两类:
第一类是官方机器人接口。优点是合规、稳定,但需要通过官方平台申请,审核流程较长,支持的能力也受平台约束。
第二类是基于 OneBot 协议的非官方实现,例如 NapCat、Lagrange、go-cqhttp 等。这类方案部署灵活,可以自建协议端,适合学习、测试和内部工具。本文演示的就是这种方案。
这里需要说清楚:非官方协议端存在账号风控风险,建议使用小号进行学习和测试,不要在生产环境直接使用主账号,更不要用于违反平台规则的活动。
1.3 整体接入链路
Grok 接入 QQ 机器人的完整数据链路如下:
QQ 群/私聊消息 ↓ OneBot 协议端(NapCat 等) ↓ WebSocket 推送消息事件 ↓ Python 机器人程序接收 ↓ 调用 Grok API(OpenAI 兼容接口) ↓ 返回回复内容 ↓ 通过 WebSocket 发送到 QQ简单来说,我们并不需要直接和 QQ 底层协议打交道,只要让 Python 程序连接 OneBot 协议端的 WebSocket 服务,订阅消息事件,再把收到的消息交给 Grok API 处理,最后把回复发回群聊即可。
2. 环境准备与版本说明
2.1 运行环境
本文示例使用 Python 3 编写,建议使用 Python 3.9 或更高版本。操作系统不限,Windows、Linux、macOS 都可以。如果你的服务器在国内,注意确保运行环境能够正常访问 Grok API 的域名,否则会出现超时报错。
版本需要根据你的项目实际情况调整,本文示例以常见环境为例,重点演示配置思路。
2.2 安装依赖
需要安装的 Python 库有:
websockets:连接 OneBot 协议端的 WebSocket 服务。httpx:调用 Grok API,支持异步请求。python-dotenv:读取.env环境变量文件,管理密钥。
安装命令:
pip install websockets httpx python-dotenv也可以把依赖写入requirements.txt:
websockets>=12.0 httpx>=0.27.0 python-dotenv>=1.0.0然后执行:
pip install -r requirements.txt2.3 准备 OneBot 协议端
本文以 NapCat 为例。NapCat 是目前社区使用较多的 OneBot 协议实现,安装后可以在本地开启 WebSocket 服务。
安装完成并登录 QQ 账号后,需要做以下配置:
- 在 NapCat 管理界面中找到网络配置。
- 开启 WebSocket 服务器,记下监听端口,默认通常是
3001。 - 确认可以访问类似
ws://127.0.0.1:3001的地址。
不同版本的 NapCat 配置入口名称可能略有差异,但核心思路一致:只要协议端能提供一个 WebSocket 地址供我们连接即可。
2.4 准备 Grok API Key
登录 Grok 模型对应的开放平台,在控制台中创建 API Key。创建后请立即复制保存,因为密钥通常只会完整显示一次。
如果你还没有开通,需要先完成实名认证和余额充值。具体开通流程以平台页面为准。
获得 API Key 后,建议写入项目根目录的.env文件,而不是直接写在代码里:
XAI_API_KEY=你的-api-key XAI_BASE_URL=https://api.x.ai/v1 XAI_MODEL=grok-4.3 ONEBOT_WS_URL=ws://127.0.0.1:3001注意:XAI_MODEL这个值一定要以你账号真实可用的模型名称为准。如果控制台显示的模型名称不是grok-4.3,请改成实际名称,否则调用时会报模型不存在。
3. 核心原理拆解
3.1 OneBot 协议的消息推送
OneBot 协议规定了机器人端和协议端之间的通信格式。协议端收到 QQ 消息后,会通过 WebSocket 推送一个 JSON 事件对象。
一个典型的群消息事件如下:
{ "post_type": "message", "message_type": "group", "group_id": 123456789, "user_id": 987654321, "raw_message": "你好", "message": [ { "type": "text", "data": { "text": "你好" } } ] }我们需要关注几个字段:
post_type:事件类型,message表示消息事件。message_type:消息类型,group表示群聊,private表示私聊。group_id:群号。user_id:发送者 QQ 号。raw_message:原始消息文本。
3.2 发送消息的动作
要向 QQ 发送消息,机器人端需要向协议端发送一个“动作”请求。动作名通常是send_group_msg或send_private_msg。
例如:
{ "action": "send_group_msg", "params": { "group_id": 123456789, "message": "Hello from Grok" }, "echo": "reply" }echo字段用于判断响应对应哪一次请求,示例中可以简单处理。
3.3 OpenAI 兼容接口的调用方式
Grok API 走的是 OpenAI 兼容格式,核心接口是:
POST {XAI_BASE_URL}/chat/completions请求体格式:
{ "model": "grok-4.3", "messages": [ { "role": "system", "content": "你是一个友好的QQ机器人助手" }, { "role": "user", "content": "你好" } ], "temperature": 0.7, "max_tokens": 2048 }其中messages数组就是对话上下文:
system:系统提示词,用来设定人设和行为规范。user:用户消息。assistant:模型之前的回复。
多轮对话的本质,就是把历史消息全部放进messages数组一起发送给模型。上下文越长,模型越能“记住”之前聊过什么,但消耗的 token 也越多。
3.4 上下文管理策略
每个 QQ 用户可以建立独立的上下文会话。简单做法是:
- 用
group_id + user_id作为会话 Key。 - 每个会话维护一个消息队列。
- 设置最大轮数,超过上限就丢弃最早的记录。
这样做既能保证多轮对话效果,又不会让请求体无限膨胀。
4. 完整实战案例
下面进入完整代码实现。项目结构如下:
grok_qq_bot/ ├── .env # 环境变量配置文件 ├── requirements.txt # Python 依赖 ├── config.py # 配置读取 ├── grok_api.py # Grok API 调用封装 └── bot.py # QQ 机器人主程序4.1 编写配置文件
文件路径:config.py
import os from dotenv import load_dotenv load_dotenv() # Grok API 配置 XAI_API_KEY = os.getenv("XAI_API_KEY", "") XAI_BASE_URL = os.getenv("XAI_BASE_URL", "https://api.x.ai/v1") XAI_MODEL = os.getenv("XAI_MODEL", "grok-4.3") # OneBot WebSocket 地址 ONEBOT_WS_URL = os.getenv("ONEBOT_WS_URL", "ws://127.0.0.1:3001") # 对话参数 DEFAULT_TEMPERATURE = 0.7 DEFAULT_MAX_TOKENS = 2048 # 上下文保留轮数(每轮包含 user 和 assistant 两条消息) MAX_CONTEXT_TURNS = 20 # 系统提示词 SYSTEM_PROMPT = ( "你是一个友善、幽默、乐于助人的QQ机器人助手。" "请使用简体中文回复,回答要清晰、简洁、准确。" "如果遇到无法确认的问题,请直接说明你不确定,不要编造信息。" "当用户提到违法、暴力、色情等敏感内容时,请礼貌拒绝回答。" )这里把系统提示词单独放在配置里,方便后续调教,不需要改动代码逻辑。
4.2 编写 Grok API 调用模块
文件路径:grok_api.py
import httpx import config async def chat_with_grok( messages, temperature=None, max_tokens=None, ): """ 调用 Grok 的 Chat Completions 接口。 messages 是 OpenAI 兼容格式的消息列表。 """ if temperature is None: temperature = config.DEFAULT_TEMPERATURE if max_tokens is None: max_tokens = config.DEFAULT_MAX_TOKENS url = f"{config.XAI_BASE_URL}/chat/completions" headers = { "Authorization": f"Bearer {config.XAI_API_KEY}", "Content-Type": "application/json", } payload = { "model": config.XAI_MODEL, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, } async with httpx.AsyncClient(timeout=90) as client: resp = await client.post(url, json=payload, headers=headers) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]["content"]这里使用httpx.AsyncClient发起异步请求,避免在 WebSocket 事件循环中同步阻塞。timeout=90表示最长等待 90 秒,防止模型推理较慢时提前超时。
4.3 编写 QQ 机器人主程序
文件路径:bot.py
import asyncio import json import re from collections import defaultdict, deque import websockets import config from grok_api import chat_with_grok # 过滤消息中的 CQ 码,例如图片、At 等 CQ_CODE_PATTERN = re.compile(r"\[CQ:\w+[^\]]*\]") # 会话上下文存储 # key: "group_群号_QQ号" 或 "private_QQ号" # value: deque,保存最近若干轮消息 sessions = defaultdict( lambda: deque(maxlen=config.MAX_CONTEXT_TURNS * 2) ) def get_session_key(event): """根据事件生成会话 Key。""" if event.get("group_id"): return f"group_{event['group_id']}_{event['user_id']}" if event.get("user_id"): return f"private_{event['user_id']}" return None def build_messages(session_key, user_message): """构造发送给模型的完整消息列表。""" messages = [{"role": "system", "content": config.SYSTEM_PROMPT}] for item in sessions[session_key]: messages.append(item) messages.append({"role": "user", "content": user_message}) return messages async def send_message(ws, event, reply): """向 QQ 发送回复消息。""" if event.get("group_id"): action = "send_group_msg" params = {"group_id": event["group_id"], "message": reply} else: action = "send_private_msg" params = {"user_id": event["user_id"], "message": reply} await ws.send( json.dumps( { "action": action, "params": params, "echo": "reply", } ) ) async def handle_event(ws, event): """处理一条来自 OneBot 协议端的事件。""" if event.get("post_type") != "message": return raw_message = event.get("raw_message", "") or event.get("message", "") plain_text = CQ_CODE_PATTERN.sub("", raw_message).strip() if not plain_text: return session_key = get_session_key(event) if not session_key: return group_id = event.get("group_id") # 群聊中使用指令前缀触发,避免机器人回复过多消息导致刷屏 if group_id and not plain_text.startswith("/grok"): return if group_id: plain_text = plain_text.replace("/grok", "", 1).strip() if not plain_text: await send_message(ws, event, "请直接对我说你想聊的内容~") return messages = build_messages(session_key, plain_text) try: reply = await chat_with_grok(messages) except Exception as exc: print(f"[Grok调用异常] {exc}") await send_message(ws, event, "我暂时开小差了,请稍后再试。") return # 调用成功后再写入上下文,避免失败消息污染历史 sessions[session_key].append( {"role": "user", "content": plain_text} ) sessions[session_key].append( {"role": "assistant", "content": reply} ) await send_message(ws, event, reply) async def main(): print(f"正在连接 OneBot WebSocket: {config.ONEBOT_WS_URL}") async with websockets.connect(config.ONEBOT_WS_URL) as ws: print("连接成功,等待消息...") async for raw in ws: try: event = json.loads(raw) await handle_event(ws, event) except Exception as exc: print(f"[消息处理异常] {exc}") if __name__ == "__main__": asyncio.run(main())这段代码实现了几个关键功能:
第一,过滤 CQ 码。QQ 群消息里可能包含[CQ:image,file=xxx]、[CQ:at,qq=xxx]这样的代码块,直接发给模型会导致理解混乱,所以需要用正则去掉。
第二,群聊指令前缀。群聊里如果机器人响应所有消息,会非常吵。示例中只有以/grok开头的消息才会被机器人处理,私聊则全部响应。
第三,上下文独立存储。每个用户在自己的会话中聊天,互不干扰。
第四,调用失败不影响历史记录。只有模型成功返回后,才把当前对话写入上下文队列,避免把异常情况带入下一轮。
4.4 运行与验证
启动机器人前,先确认 NapCat 已经运行,并且 WebSocket 地址正确。然后执行:
python bot.py看到如下输出表示连接成功:
正在连接 OneBot WebSocket: ws://127.0.0.1:3001 连接成功,等待消息...此时在 QQ 群聊中发送:
/grok 用一句话介绍量子计算机器人应该会调用 Grok API,然后把回复发送到群里。
私聊中直接发送任何消息即可触发。
4.5 结果说明
如果一切正常,你会看到模型返回的内容完整出现在 QQ 消息中。由于代码没有做长消息分段处理,当回复超过 QQ 单条消息长度限制时,可能会被截断或发送失败。
这个问题可以在生产环境中处理:超过一定长度就按固定字符数分段发送,或者用“继续”机制让模型分多次输出。后续章节会给出建议。
5. 丰富调教内容与方法
接入只是第一步,真正让机器人好用的是“调教”。调教主要体现在系统提示词、生成参数和上下文策略三个方面。
5.1 系统提示词设计
系统提示词决定机器人的“人设”。同一个模型,提示词不同,表现可能完全不同。
下面是一个“群聊百科助手”风格的提示词示例:
你是一个群聊百科助手,名叫小格。 你的特点: 1. 回答简洁,单次回复一般不超过100字。 2. 遇到技术问题,给出可以操作的具体建议。 3. 面对无意义刷屏时,温和地提醒用户换一个话题。 4. 不知道的内容直接说不知道,不要编造。 5. 不参与争吵,不输出攻击性言论。如果希望机器人偏向代码辅助,可以把提示词换成:
你是群里的代码助手,擅长 Python、Java、SQL 等语言。 当用户提出编程问题时,优先给出可运行的代码示例,并解释关键点。 如果用户没有提供完整背景,先询问必要信息再回答。建议把提示词放到.env文件旁边的单独配置文件里,或者使用配置中心管理。这样调整人设时不需要重新部署代码。
5.2 生成参数调节
OpenAI 兼容接口支持多个生成参数,其中比较重要的是:
temperature:控制随机性。值越低回复越稳定,越高越有创造性。建议工具类机器人使用0.3到0.5,闲聊机器人使用0.7到0.9。max_tokens:限制最大输出长度。群聊场景建议 500 到 1024,避免回复过长刷屏。top_p:核采样参数,一般保持默认,不需要频繁调整。
在示例代码中,这些参数已经在config.py里配置,你可以按需修改。
5.3 上下文长度控制
超长上下文是 Grok 的优势,但也要注意成本。每轮对话都携带全部历史消息,token 消耗会随着对话进行快速增长。
建议策略:
- 限制每个会话的最大轮数,示例中
MAX_CONTEXT_TURNS = 20。 - 定期清理长时间不活跃的会话,节省内存。
- 对敏感信息进行脱敏后再送入模型。
- 如果只做单轮问答机器人,可以不保留上下文,每次只发送当前消息。
5.4 敏感内容过滤
作为 QQ 群机器人,调教时一定要加入安全约束。在系统提示词中明确要求模型拒绝回答违法、暴力、色情等敏感内容,是最基础的一层防护。
如果对安全要求较高,还可以在代码层增加关键词过滤,拦截明确违规的输入和输出。例如:
BLOCK_WORDS = ["xxx关键词1", "xxx关键词2"] def check_block_words(text): for word in BLOCK_WORDS: if word in text: return True return False在调用 API 之前检查用户消息,在发送回复之前检查模型输出,命中关键词就不回复或回复固定提示语。需要注意的是,关键词过滤只是辅助手段,不能完全替代模型自身的安全对齐。
6. 常见问题与排查思路
实际运行过程中,可能会遇到各种报错。下面整理了一份常见问题对照表。
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
| 连接不上 OneBot WebSocket | 地址或端口写错,或者协议端没有开启 WebSocket 服务 | 检查ONEBOT_WS_URL,确认 NapCat 已运行 |
| 连接后收不到任何消息 | 协议端配置了 Access Token,客户端未携带 | 在 WebSocket 连接 URL 中加上 token,或关闭鉴权 |
报错401 | API Key 错误或已过期 | 重新生成 API Key,检查.env配置 |
报错404,提示模型不存在 | XAI_MODEL名称不对 | 登录开放平台控制台,确认当前可用的模型名称 |
| 请求超时 | 网络不稳定,或模型推理时间过长 | 增大timeout,增加重试机制 |
| 群聊中 bot 回复所有消息,太吵 | 没有设置触发前缀 | 只在/grok指令下响应,或只响应 at 消息 |
| 回复内容被 QQ 截断 | 单条回复过长 | 按 200 到 400 字符分段发送 |
| Windows 控制台输出乱码 | 编码问题 | 设置PYTHONIOENCODING=utf-8 |
| 模型上下文“记不住” | 上下文轮数太少,或会话 Key 设计不合理 | 增大MAX_CONTEXT_TURNS,统一会话 Key 规则 |
这里单独说明一下 WebSocket 鉴权问题。部分 OneBot 协议端允许设置 Access Token,如果设置了,连接时需要添加请求头或 URL 参数。具体格式取决于协议端版本,建议查阅对应文档,通常是通过Authorization: Bearer {token}或 URL 查询参数传入。
7. 最佳实践与工程建议
7.1 密钥安全管理
绝对不要把 API Key 硬编码在代码里,也不要提交到 Git 仓库。推荐使用.env文件配合.gitignore,或者使用服务器的环境变量。
.gitignore至少包含:
.env __pycache__/ *.pyc venv/7.2 错误重试与限流
Grok API 在高峰期可能返回 429(请求过多)或 5xx(服务端错误)。生产环境建议实现简单的重试逻辑,例如失败后等待 1 秒、3 秒、5 秒重试三次。
同时要注意单用户调用频率。QQ 群聊中如果多个用户同时触发,可能出现并发请求。建议使用线程安全的队列或信号量控制最大并发数。
7.3 日志记录
记录完整的运行日志对排查问题非常重要。至少记录以下信息:
- 收到消息的时间、群号、用户 ID、消息内容摘要。
- 调用 Grok API 的耗时和返回状态。
- 发送消息失败的异常堆栈。
可以使用 Python 标准库logging实现,避免把所有输出都打印到控制台。
7.4 长消息分段发送
QQ 单条消息长度有限。当模型回复较长时,建议先判断长度,超过阈值就分段发送。
分段示例:
def split_message(text, limit=200): return [text[i:i+limit] for i in range(0, len(text), limit)]分段时要注意不要在代码块中间断开,否则会影响阅读体验。
7.5 账号安全与合规
再次提醒:非官方协议端存在账号风控风险,务必使用小号测试。正式场景中如果确实需要在 QQ 生态做机器人,优先了解官方机器人开放能力。
同时,机器人输出内容必须遵守法律法规和平台规定。不要诱导模型生成违规内容,不要将机器人用于刷量、骚扰、诈骗等任何违法用途。
7.6 生产环境架构建议
本文示例是一个最小可运行版本,便于理解原理。如果要做成正式服务,建议:
- 使用成熟框架如 NoneBot2,它提供了插件管理、权限控制、会话管理能力。
- 将 Grok API 调用抽成独立服务,方便扩展和复用。
- 使用消息队列削峰,避免高并发时直接压垮协议端。
- 增加监控告警,例如调用失败率超过阈值时通知维护人员。
8. 总结与下一步
到这里,整个 Grok 模型接入 QQ 机器人的流程就完整跑通了。我们一起实现了从 OneBot 协议端接收消息、过滤 CQ 码、调用 Grok 接口、维护多轮上下文、发送回复到 QQ 的全链路代码,并且分析了系统提示词、生成参数、上下文长度控制等调教方法。对于常见的连接失败、鉴权错误、模型名称错误等问题,也整理了排查方向。
这套代码虽然简单,但是一个非常好的学习样板。理解了它,你再去看 NoneBot2 等成熟框架的源码,会发现思路是相通的:无非是协议适配、消息处理、模型调用、上下文管理这几个模块。
下一步,你可以尝试做这几件事:
第一,把项目里的系统提示词改成自己真正需要的场景,比如编程助手、学习辅导、群聊百科。
第二,尝试接入更多模型平台,因为示例里用的是 OpenAI 兼容接口,很多模型服务都可以用同样的代码接入。
第三,完善重试、限流、日志、分段发送等生产级能力,把这个最小示例打磨成一个稳定的服务。
动手实践是掌握这项技术最好的方式。建议先跑通最小示例,再逐步加功能。如果过程中遇到问题,欢迎对照本文的常见问题章节逐一排查。