最近想玩QQ机器人,但是go-cqhttp已经挂了,之前就各种被45风控,体验很一般。
现在比较主流的是NapCatQQ,但是毕竟不是官方的,还是面临一些风险,而且还是有一些技术门槛。
正好QQ官方再推官方机器人,于是本篇就围绕基于QQ官方机器人botpy搭建聊天机器人展开。
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会恢复出厂设置。
评论