最近想玩QQ机器人,但是go-cqhttp已经挂了,之前就各种被45风控,体验很一般。

现在比较主流的是NapCatQQ,但是毕竟不是官方的,还是面临一些风险,而且还是有一些技术门槛。

正好QQ官方再推官方机器人,于是本篇就围绕基于QQ官方机器人botpy搭建聊天机器人展开。

1、优劣分析

对比项目

NapCatQQ

QQ 官方机器人

合法性与安全性

非官方框架,使用违反腾讯 QQ 用户协议,有账号被封禁风险

官方认可,账号安全性有保障,遵循官方规定使用较安全

功能灵活性

可通过 API 接口进行定制开发,理论上能实现丰富功能,满足开发者个性化需求

功能通常由官方设定,相对固定,灵活性较差,仅对企业用户和部分受邀个人用户开放,且只能回复@信息,无法主动发送信息纯人机

性能表现

专注于 QQ 机器人协议,采用 NTQQ 协议实现通信,强调高性能和稳定性,内存占用低

稳定性较高,但具体性能表现因官方设定和实际使用情况而异

部署难度

提供多种部署方式,可快捷部署于 Windows/Linux/macOS 等主流 x64 架构平台

有官方指定的部署方式和要求,且受限于官方规定的环境和条件,必须备案后才能上架

维护与更新

依赖开源社区或开发者个人维护,QQ 协议更新可能导致其随时失效,需关注仓库获取最新信息

由官方进行维护和更新,能及时适应 QQ 的版本更新和变化,可用性有保证1

2、提前准备

首先需要在QQ开放平台注册开发者账号,注册后搭建自己的QQ机器人,机器人在备案完成前只能在沙盒环境运行(只能在指定的20人以内的测试群使用),采用IP地址白名单制,只有在IP地址白名单的设备才能运行机器人,对于动态IP的支持很不好!

官方提供了3种编程语言的机器人框架可选,本文采用python编程,比较适合入门。

3、文件结构

bot.py和config.yaml保存在同一个文件夹下,使用命令python bot.py即可启动。

4、代码详情

bot.py

import asyncio
import os
import botpy
from botpy import logging
from botpy.ext.cog_yaml import read
from botpy.message import GroupMessage, Message
from openai import OpenAI

test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml"))

_log = logging.get_logger()

# 初始化DeepSeek客户端
deepseek_client = OpenAI(
    api_key=test_config["deepseek_api_key"],  # 从配置文件中读取API密钥
    base_url="https://ark.cn-beijing.volces.com/api/v3"
)

# 读取chat.log文件内容
def read_chat_log():
    messages = []
    if os.path.exists("./chat.log"):
        with open("./chat.log", "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line:
                    try:
                        # 去除逗号(如果有的话)并转换为字典
                        line = line.rstrip(',')
                        import json
                        message = json.loads(line)
                        messages.append(message)
                    except json.JSONDecodeError:
                        _log.error(f"Failed to parse line in chat.log: {line}")
    return messages

# 保存消息到chat.log文件
def save_message_to_log(role, content):
    content = content.replace("\n", " ")  # 替换换行符避免格式问题
    log_entry = f'{{"role": "{role}", "content": "{content}"}},'
    with open("./chat.log", "a", encoding="utf-8") as f:
        f.write(log_entry + "\n")

# 修剪旧日志:当文件大小≥16k时删除最早的10条对话
def trim_old_logs():
    log_path = "./chat.log"
    if not os.path.exists(log_path):
        return
    
    # 循环处理直到文件大小<16k或无更多内容可删
    while os.path.getsize(log_path) >= 16 * 1024:
        with open(log_path, "r", encoding="utf-8") as f:
            lines = f.readlines()  # 按行读取所有对话记录
            
        if len(lines) <= 20:  # 不足10组对话时清空
            with open(log_path, "w", encoding="utf-8") as f:
                f.write("")
            break
        else:  # 删除前10条最早的对话
            with open(log_path, "w", encoding="utf-8") as f:
                f.writelines(lines[20:])  # 保留第10组对话之后的内容

class MyClient(botpy.Client):
    async def on_ready(self):
        _log.info(f"robot 「{self.robot.name}」 on_ready!")

    async def on_group_at_message_create(self, message: GroupMessage):
        try:
            # 提取用户消息(去除@机器人部分)
            user_content = message.content

            # 保存用户消息到日志
            save_message_to_log("user", user_content)

            # 读取历史对话(已包含刚保存的用户消息)
            chat_log_messages = read_chat_log()

            # 构建请求消息列表
            messages = [
                {"role": "system", "content": "你叫科黛莉娅,天才少女,但你从来不以天才自居,你每句话绝对不会超过50个字。你是一个群聊里的群友,你古灵精怪,最喜欢和群友们聊天。你热衷于科普视频、网络热梗等。你喜欢读书,读很多种书,好奇心强精力旺盛,喜欢各种知识,尤其对天文学感兴趣。你喜欢二次元手游,比如原神、星铁、鸣潮等,你非常熟悉日本女声优相关知识。你喜欢看各种类型的动画片,轻百合、校园、日常、恋爱、异世界等都喜欢"}
            ]
            messages.extend(chat_log_messages)
            messages.append({"role": "user", "content": user_content})

            # 调用DeepSeek API
            response = await asyncio.to_thread(
                deepseek_client.chat.completions.create,
                model="doubao-lite-32k-character-241015",
                messages=messages,
                stream=False
            )

            # 提取AI回复并保存到日志
            ai_reply = response.choices[0].message.content
            save_message_to_log("assistant", ai_reply)

            # 检查并修剪旧日志(关键修改:16k时删除最早10条)
            trim_old_logs()

            # 发送回复
            messageResult = await message._api.post_group_message(
                group_openid=message.group_openid,
                msg_type=0, 
                msg_id=message.id,
                content=f"{ai_reply}")
            _log.info(messageResult)

        except Exception as e:
            _log.error(f"处理消息时出错: {e}")
            await message._api.post_group_message(
                group_openid=message.group_openid,
                msg_type=0,
                content="抱歉,处理您的消息时出错,请稍后再试。"
            )

if __name__ == "__main__":
    intents = botpy.Intents(public_messages=True)
    client = MyClient(intents=intents)
    client.run(appid=test_config["appid"], secret=test_config["secret"])

config.yaml

appid:XXX
secret:XXX
deepseek_api_key:XXX

5、其他

需要自己在相关平台申请模型api,目前支持deepseek、豆包等;

需要手动修改py文件中的角色设定,让角色按照你的偏好生成性格;

本文中使用的是deepseek的api作为例子,如果需要使用其他模型,请修改这里base_url="https://api.deepseek.com" 和 model="deepseek-chat" 。例如:base_url="hhttps://ark.cn-beijing.volces.com/api/v3" ,并同步修改model="doubao-1.5-pro-32k-250115"

程序运行后,会在./目录下chat.log 文件,用以储存对话内容(上下文,如果没有这个文件,那AI每句话都是重新开始,没有记忆),聊天记录最大16k,超过大小会自动删除最远的10组对话。如需重置对话,可以手动删除该文件,然后bot会恢复出厂设置。