跳到主要内容Gemini 全能 QQ 机器人部署手册 | 极客日志PythonAI算法
Gemini 全能 QQ 机器人部署手册
基于 OneBot V11 (NapCat)、NoneBot2 及 Gemini Flash 模型的全能 QQ 机器人部署流程。涵盖 Ubuntu 22.04 环境下的 Docker 安装、NapCat 容器配置、Python 虚拟环境搭建、核心代码编写(含打卡与记忆功能)、Systemd 服务管理及 WebSocket 连接设置。提供管理员指令配置及日志查看方法,确保机器人稳定运行于云服务器。
岁月神偷1 浏览 Gemini 全能 QQ 机器人部署手册
核心架构:OneBot V11 (NapCat) + NoneBot2 + Gemini Flash
适用系统:Ubuntu 22.04 LTS (阿里云/腾讯云)
🟢 第一阶段:基础设施准备
SSH 连接服务器后,复制以下命令执行。
安装必要软件 (Docker + Python)
sudo apt update && sudo apt upgrade -y
curl -fsSL https://get.docker.com | bash
sudo apt install python3-pip python3-venv -y
mkdir -p /root/gemini_bot
cd /root/gemini_bot
🔌 第二阶段:部署 QQ 协议端 (NapCat)
【避坑重点】:这里我们使用 --net=host 模式启动。这样容器和宿主机共用网络,配置 WebSocket 时直接填 127.0.0.1 即可,彻底解决连接不上、找不到 IP 的问题。
- 放行防火墙端口
- 请去阿里云/腾讯云控制台的'防火墙'或'安全组',放行 TCP 6099 和 TCP 8080。
启动 NapCat 容器
(请将下方命令中的 你的 QQ 号 替换为实际数字)
docker run -d\
--name napcat \
--net=host \
--restart always \
-e ACCOUNT=你的 QQ 号 \
-v /root/gemini_bot/napcat/config:/app/config \
mlikiowa/napcat-docker:latest
🧠 第三阶段:部署核心代码
【避坑重点】:这里我们锁定安装 0.3.2 版本的 Google 库,这是目前配合此代码最稳定的版本,没有任何属性报错。
- 创建 memory.py
- 使用
nano memory.py 创建文件,填入你的 API Key:
配置 Python 环境
cd /root/gemini_bot
python3 -m venv venv
source venv/bin/activate
pip install nonebot2[fastapi] nonebot-adapter-onebot google-generativeai==0.3.2 pytz
google.generativeai genai
asyncio
datetime
pytz
pickle
os
random
API_KEY =
DATA_FILE =
genai.configure(api_key=API_KEY)
chat_model = genai.GenerativeModel()
:
():
.group_id = group_id
.summary =
.buffer = []
.lock = asyncio.Lock()
.last_active_time =
.last_msg_content =
.repeat_count =
.bot_config = {
: ,
: ,
: ,
: ,
: ,
:
}
.checkin_data = {}
():
state = .__dict__.copy()
state[]
state
():
.__dict__.update(state)
.lock = asyncio.Lock()
():
user_id .checkin_data:
.checkin_data[user_id] = {: , : , : {}}
user_data = .checkin_data[user_id]
user_data[] = user_name
content user_data[]:
user_data[][content] =
user_data[][content] +=
user_data[] +=
save_data()
():
user_id .checkin_data:
,
user_data = .checkin_data[user_id]
content user_data[] user_data[][content] == :
,
user_data[][content] -=
user_data[] -=
user_data[][content] == :
user_data[][content]
user_data[]:
.checkin_data[user_id]
save_data()
,
():
target_user_id .checkin_data:
.checkin_data[target_user_id] = {: target_user_name, : , : {}}
user_data = .checkin_data[target_user_id]
user_data[] = target_user_name
content user_data[]:
user_data[][content] =
operation == :
user_data[][content] += count
user_data[] += count
save_data()
,
operation == :
user_data[][content] < count:
,
user_data[][content] -= count
user_data[] -= count
user_data[][content] == :
user_data[][content]
user_data[]:
.checkin_data[target_user_id]
save_data()
,
:
,
():
crown = stars // ( * * )
stars %= ( * * )
sun = stars // ( * )
stars %= ( * )
moon = stars //
star = stars %
medals =
crown > :
medals += * crown
sun > :
medals += * sun
moon > :
medals += * moon
star > (crown + sun + moon + star == ):
medals += * star
medals
():
.checkin_data:
sorted_users = (
.checkin_data.values(), key= x: x[], reverse=
)
ranking =
idx, user (sorted_users, ):
total_medals = .calculate_medals(user[])
ranking +=
content, count user[].items():
content_medals = .calculate_medals(count)
ranking +=
ranking
():
entry =
.buffer.append(entry)
clean_msg = msg.strip()
clean_msg == .last_msg_content clean_msg:
.repeat_count +=
:
.last_msg_content = clean_msg
.repeat_count =
():
((m) m .buffer)
():
.lock:
.estimate_length() > .bot_config[] (.buffer) > .bot_config[]:
to_compress = .buffer[:-.bot_config[]]
text_block = .join(to_compress)
prompt =
:
loop = asyncio.get_running_loop()
resp = loop.run_in_executor(, : chat_model.generate_content(prompt))
resp.candidates[].content.parts[].text:
.summary = resp.candidates[].content.parts[].text
.buffer = .buffer[-.bot_config[]:]
save_data()
Exception e:
()
():
tz = pytz.timezone()
context_str = .join(.buffer)
is_active_interrupt:
task_prompt =
:
current_question.strip():
task_prompt =
:
task_prompt =
final_prompt =
:
loop = asyncio.get_running_loop()
resp = loop.run_in_executor(, : chat_model.generate_content(final_prompt))
resp.candidates[].content.parts[].text resp.candidates[].content.parts[].text
Exception e:
():
time
now = time.time()
now - .last_active_time < .bot_config[]:
,
(.repeat_count >= .bot_config[] random.random() < .bot_config[] current_msg.strip()):
:
loop = asyncio.get_running_loop()
check_resp = loop.run_in_executor(, : chat_model.generate_content())
check_resp.candidates[].content.parts[].text.strip().upper():
.last_active_time = now
, current_msg
:
,
current_msg.lower() random.random() > .bot_config[]:
,
:
loop = asyncio.get_running_loop()
decision_resp = loop.run_in_executor(, : chat_model.generate_content())
decision_resp.candidates[].content.parts[].text.strip().upper():
reply = .generate_reply(, , is_active_interrupt=)
.last_active_time = now
, reply
Exception e:
()
,
memories = {}
():
:
(DATA_FILE, ) f:
pickle.dump(memories, f)
Exception e:
()
():
memories
os.path.exists(DATA_FILE):
:
(DATA_FILE, ) f:
memories = pickle.load(f)
:
memories = {}
():
group_id memories:
memories[group_id] = GroupMemory(group_id)
memories[group_id]
load_data()
import
as
import
import
import
import
import
import
"在这里粘贴你的 Google_API_Key"
"bot_memory.pkl"
'gemini-2.5-flash'
class
GroupMemory
def
__init__
self, group_id
self
self
"暂无早期历史记录。"
self
self
self
0
self
""
self
0
self
"COOLDOWN_SECONDS"
0
"TRIGGER_PROBABILITY"
0.05
"REPEATER_THRESHOLD"
2
"REPEATER_EXEC_PROB"
0.1
"MEMORY_LIMIT_CHARS"
10000
"MEMORY_KEEP_COUNT"
100
self
def
__getstate__
self
self
del
'lock'
return
def
__setstate__
self, state
self
self
def
checkin
self, user_id, user_name, content
"""执行打卡:增加记录 + 总星星"""
if
not
in
self
self
"name"
""
"total_stars"
0
"records"
self
"name"
if
not
in
"records"
"records"
0
"records"
1
"total_stars"
1
def
delete_checkin
self, user_id, content
"""删除打卡:减少记录 + 总星星(返回:是否成功,提示语)"""
if
not
in
self
return
False
"你还没有任何打卡记录哦~"
self
if
not
in
"records"
or
"records"
0
return
False
f"你没有「{content}」的打卡记录~"
"records"
1
"total_stars"
1
if
"records"
0
del
"records"
if
not
"records"
del
self
return
True
f"已删除「{content}」的 1 次打卡记录~"
def
admin_checkin
self, target_user_id, target_user_name, content, count, operation
""" 管理员操作他人打卡记录 operation: "add"(增加) / "delete"(删除)
返回:是否成功,提示语 """
if
not
in
self
self
"name"
"total_stars"
0
"records"
self
"name"
if
not
in
"records"
"records"
0
if
"add"
"records"
"total_stars"
return
True
f"已为用户【{target_user_name}】增加「{content}」打卡记录 {count} 次~"
elif
"delete"
if
"records"
return
False
f"用户【{target_user_name}】的「{content}」打卡记录不足 {count} 次!"
"records"
"total_stars"
if
"records"
0
del
"records"
if
not
"records"
del
self
return
True
f"已为用户【{target_user_name}】删除「{content}」打卡记录 {count} 次~"
else
return
False
"操作类型错误!仅支持「增加」或「删除」"
def
calculate_medals
self, stars
"""计算星星对应的勋章:4 星=1 月,4 月=1 日,4 日=1 皇冠"""
4
4
4
4
4
4
4
4
4
4
4
4
""
if
0
"👑"
if
0
"☀️"
if
0
"🌙"
if
0
or
0
"⭐"
return
def
generate_checkin_ranking
self
"""生成打卡排行榜(按总星星降序,子项也显示勋章)"""
if
not
self
return
"当前还没有打卡记录哦~快叫群友一起打卡吧!"
sorted
self
lambda
"total_stars"
True
"🏆 打卡记录排行榜 🏆\n"
for
in
enumerate
1
self
"total_stars"
f"{idx}. {user['name']}{total_medals}\n"
for
in
"records"
self
f" · {content}{content_medals}\n"
return
def
add_message
self, name, msg, time_str
f"[{time_str}] 【{name}】: {msg}"
self
if
self
and
self
1
else
self
self
1
def
estimate_length
self
return
sum
len
for
in
self
async
def
check_and_compress
self
async
with
self
if
self
self
"MEMORY_LIMIT_CHARS"
and
len
self
self
"MEMORY_KEEP_COUNT"
self
self
"MEMORY_KEEP_COUNT"
"\n"
f""" 你是记忆整理员,更新长期记忆摘要(800 字内):
【长期记忆】:{self.summary}
【待归档对话】:{text_block}"""
try
await
None
lambda
if
0
0
self
0
0
self
self
self
"MEMORY_KEEP_COUNT"
except
as
print
f"压缩记忆失败:{e}"
pass
async
def
generate_reply
self, current_question, sender_name, is_active_interrupt=False
'Asia/Shanghai'
"\n"
self
if
"没人艾特你,自然加入讨论,简短、像群友,可复读/玩梗,直接说话不加称呼,绝对不要输出任何时间戳/日期/时间格式,不要添加自己的名字前缀(如【Gemini】)。"
else
if
not
f"用户【{sender_name}】艾特了你,自然接话,绝对不要输出任何时间戳/日期/时间格式,不要添加自己的名字前缀(如【Gemini】)。"
else
f"用户【{sender_name}】提问:{current_question},请回答,绝对不要输出任何时间戳/日期/时间格式,不要添加自己的名字前缀(如【Gemini】)。"
f""" 你是 QQ 群里的 AI 伙伴 Gemini。
【长期记忆】:{self.summary}
【近期对话】:{context_str}
【任务】:{task_prompt}
【规则】:
1. 问时间时只说当前北京时间(比如'现在是下午 3 点 20 分'),但不要输出任何括号/时间戳/数字格式的时间;
2. 问新闻/数据自动联网,风格像群友而非客服;
3. 回复时禁止添加自己的名字或任何标识前缀(如【Gemini】),直接输出回复内容。 """
try
await
None
lambda
return
0
0
if
0
0
else
"(暂时无法回复)"
except
as
return
f"(回复失败:{str(e)})"
async
def
check_active_intervention
self, current_msg
import
if
self
self
"COOLDOWN_SECONDS"
return
False
None
if
self
self
"REPEATER_THRESHOLD"
and
self
"REPEATER_EXEC_PROB"
and
try
await
None
lambda
f"判断'{current_msg}'是否是正常梗(非广告/脏话):是回 YES,否回 NO"
if
"YES"
in
0
0
self
return
True
except
pass
return
False
None
if
"gemini"
not
in
and
self
"TRIGGER_PROBABILITY"
return
False
None
try
await
None
lambda
f""" 最近消息:{chr(10).join(self.buffer[-15:])}
当前消息:{current_msg}
没被艾特,是否有必要说话?满足(客观问题无人答/讨论你/有趣的氛围/有趣的话题)回 YES,否则 NO """
if
"YES"
in
0
0
await
self
""
"群友"
True
self
return
True
except
as
print
f"主动介入失败:{e}"
pass
return
False
None
def
save_data
try
with
open
'wb'
as
except
as
print
f"保存记忆失败:{e}"
def
load_data
global
if
try
with
open
'rb'
as
except
def
get_memory
group_id
if
not
in
return
import nonebot
from nonebot import on_message
from nonebot.adapters.onebot.v11 import Adapter, Bot, GroupMessageEvent
from memory import get_memory, save_data
import atexit
import datetime
import pytz
ADMIN_QQ = 1234567
nonebot.init()
driver = nonebot.get_driver()
driver.register_adapter(Adapter)
atexit.register(save_data)
def unix_to_beijing(timestamp):
"""将时间戳转换为北京时间字符串"""
dt_utc = datetime.datetime.fromtimestamp(timestamp, pytz.utc)
dt_bj = dt_utc.astimezone(pytz.timezone('Asia/Shanghai'))
return dt_bj.strftime("%H:%M:%S")
def get_current_bj_time():
"""获取当前北京时间字符串"""
return datetime.datetime.now(pytz.timezone('Asia/Shanghai')).strftime("%H:%M:%S")
def is_at_me(bot: Bot, event: GroupMessageEvent) -> bool:
"""判断是否@了机器人"""
if event.is_tome():
return True
for seg in event.message:
if seg.type == "at" and str(seg.data.get("qq")) == str(bot.self_id):
return True
return False
handler = on_message(priority=1, block=True)
@handler.handle()
async def _(bot: Bot, event: GroupMessageEvent):
user_id = event.user_id
group_id = event.group_id
msg = event.get_plaintext().strip()
mem = get_memory(group_id)
try:
info = await bot.get_group_member_info(group_id=group_id, user_id=user_id)
name = info.get('card') or info.get('nickname') or str(user_id)
except:
name = "群友"
msg_lower = msg.lower()
def get_cmd_content(prefix):
prefix_len = len(prefix)
if msg.startswith(prefix):
return msg[prefix_len:].strip()
elif msg_lower.startswith(prefix.lower()):
for i in range(len(msg)):
if msg[i:i+len(prefix)].lower() == prefix.lower():
return msg[i+len(prefix):].strip()
return ""
if msg_lower.startswith("gemini 设置 "):
cmd_content = get_cmd_content("gemini 设置 ")
mem.add_message(name, msg, get_current_bj_time())
save_data()
reply_msg = ""
if user_id != ADMIN_QQ:
reply_msg = "❌ 你没有权限执行此操作!仅管理员可用~"
else:
try:
cmd_parts = msg.split(" ", 3)
if len(cmd_parts) != 4:
reply_msg = "❌ 指令格式错误!\n正确格式:gemini 设置 参数名 值\n示例:gemini 设置 冷却时间 600"
else:
param_map = {
"冷却时间": "COOLDOWN_SECONDS",
"插嘴概率": "TRIGGER_PROBABILITY",
"复读阈值": "REPEATER_THRESHOLD",
"复读概率": "REPEATER_EXEC_PROB",
"缓存字数上限": "MEMORY_LIMIT_CHARS",
"保留消息数": "MEMORY_KEEP_COUNT"
}
param_cn = cmd_parts[2]
param_value = cmd_parts[3]
if param_cn not in param_map:
valid_params = "、".join(param_map.keys())
reply_msg = f"❌ 参数名错误!\n支持的参数:{valid_params}"
else:
param_en = param_map[param_cn]
if param_en in ["COOLDOWN_SECONDS", "REPEATER_THRESHOLD", "MEMORY_LIMIT_CHARS", "MEMORY_KEEP_COUNT"]:
new_value = int(param_value)
else:
new_value = float(param_value)
mem.bot_config[param_en] = new_value
save_data()
reply_msg = f"✅ 【{group_id}群】参数修改成功!\n{param_cn} = {new_value}"
except ValueError:
reply_msg = "❌ 参数值格式错误!\n整数填数字(如 600),概率填小数(如 0.02)"
except Exception as e:
reply_msg = f"❌ 修改失败:{str(e)}"
mem.add_message("Gemini", reply_msg, get_current_bj_time())
save_data()
await handler.finish(reply_msg)
elif msg_lower.startswith("gemini 打卡 "):
content = get_cmd_content("gemini 打卡 ")
if not content:
reply_msg = "❌ 打卡内容不能为空哦~比如:「gemini 打卡 篮球」"
else:
mem.checkin(str(user_id), name, content)
ranking = mem.generate_checkin_ranking()
reply_msg = f"✅ {name} 打卡「{content}」成功!\n{ranking}"
mem.add_message(name, msg, get_current_bj_time())
mem.add_message("Gemini", reply_msg, get_current_bj_time())
save_data()
await handler.finish(reply_msg)
elif msg_lower.startswith("gemini 删除 "):
content = get_cmd_content("gemini 删除 ")
if not content:
reply_msg = "❌ 删除内容不能为空哦~比如:「gemini 删除 篮球」"
else:
success, result = mem.delete_checkin(str(user_id), content)
if success:
ranking = mem.generate_checkin_ranking()
reply_msg = f"{result}\n{ranking}"
else:
reply_msg = result
mem.add_message(name, msg, get_current_bj_time())
mem.add_message("Gemini", reply_msg, get_current_bj_time())
save_data()
await handler.finish(reply_msg)
elif msg_lower.strip() == "gemini 排行榜":
ranking = mem.generate_checkin_ranking()
reply_msg = f"{ranking}"
mem.add_message(name, msg, get_current_bj_time())
mem.add_message("Gemini", reply_msg, get_current_bj_time())
save_data()
await handler.finish(reply_msg)
elif msg_lower.startswith("gemini 管理打卡 "):
if user_id != ADMIN_QQ:
reply_msg = "❌ 你没有权限执行此操作!仅管理员可用~"
mem.add_message(name, msg, get_current_bj_time())
mem.add_message("Gemini", reply_msg, get_current_bj_time())
save_data()
await handler.finish(reply_msg)
cmd_content = get_cmd_content("gemini 管理打卡 ")
if not cmd_content:
reply_msg = "❌ 指令格式错误!\n正确格式:\ngemini 管理打卡 增加 目标 QQ 号 内容 次数\n示例:gemini 管理打卡 增加 123456 跑步 3\n\ngemini 管理打卡 删除 目标 QQ 号 内容 次数\n示例:gemini 管理打卡 删除 123456 跑步 1"
mem.add_message(name, msg, get_current_bj_time())
mem.add_message("Gemini", reply_msg, get_current_bj_time())
save_data()
await handler.finish(reply_msg)
cmd_parts = cmd_content.split()
if len(cmd_parts) != 4:
reply_msg = "❌ 指令参数不足!\n正确格式:gemini 管理打卡 增加/删除 目标 QQ 号 内容 次数"
mem.add_message(name, msg, get_current_bj_time())
mem.add_message("Gemini", reply_msg, get_current_bj_time())
save_data()
await handler.finish(reply_msg)
operation = cmd_parts[0]
target_qq = cmd_parts[1]
content = cmd_parts[2]
count_str = cmd_parts[3]
if operation not in ["增加", "删除"]:
reply_msg = "❌ 操作类型错误!仅支持「增加」或「删除」"
mem.add_message(name, msg, get_current_bj_time())
mem.add_message("Gemini", reply_msg, get_current_bj_time())
save_data()
await handler.finish(reply_msg)
if not target_qq.isdigit():
reply_msg = "❌ 目标 QQ 号必须是纯数字!"
mem.add_message(name, msg, get_current_bj_time())
mem.add_message("Gemini", reply_msg, get_current_bj_time())
save_data()
await handler.finish(reply_msg)
if not count_str.isdigit() or int(count_str) <= 0:
reply_msg = "❌ 次数必须是正整数!"
mem.add_message(name, msg, get_current_bj_time())
mem.add_message("Gemini", reply_msg, get_current_bj_time())
save_data()
await handler.finish(reply_msg)
count = int(count_str)
try:
target_info = await bot.get_group_member_info(group_id=group_id, user_id=target_qq)
target_name = target_info.get('card') or target_info.get('nickname') or target_qq
except:
target_name = target_qq
op_map = {"增加": "add", "删除": "delete"}
success, result = mem.admin_checkin(
target_user_id=str(target_qq),
target_user_name=target_name,
content=content,
count=count,
operation=op_map[operation]
)
if success:
ranking = mem.generate_checkin_ranking()
reply_msg = f"{result}\n{ranking}"
else:
reply_msg = result
mem.add_message(name, msg, get_current_bj_time())
mem.add_message("Gemini", reply_msg, get_current_bj_time())
save_data()
await handler.finish(reply_msg)
if not msg:
msg = " "
mem.add_message(name, msg, unix_to_beijing(event.time))
save_data()
await mem.check_and_compress()
final_reply = None
if is_at_me(bot, event):
final_reply = await mem.generate_reply(msg, name, is_active_interrupt=False)
else:
should, content = await mem.check_active_intervention(msg)
if should and content:
final_reply = content
if final_reply:
mem.add_message("Gemini", final_reply, get_current_bj_time())
save_data()
await handler.finish(final_reply)
if __name__ == "__main__":
nonebot.run(host="0.0.0.0", port=8080)
⚙️ 第四阶段:配置系统服务 (Systemd)
为了实现您要求的命令管理,我们需要创建一个系统服务文件。
sudo systemctl daemon-reload
sudo systemctl enable gemini-bot
粘贴内容
(注意:ExecStart 指向了虚拟环境中的 python,确保路径准确)
[Unit]
Description=Gemini QQ Bot Service
After=network.target
[Service]
User=root
WorkingDirectory=/root/gemini_bot
ExecStart=/root/gemini_bot/venv/bin/python bot.py
Restart=always
RestartSec=5
[Install]
WantedBy=multi-user.target
sudo nano /etc/systemd/system/gemini-bot.service
🖥️ 第五阶段:NapCat 连接 (WebUI)
【避坑重点】:因为我们用了 host 模式,这里的 WS 地址必须填 127.0.0.1。
- 浏览器访问 NapCat
访问 http://服务器 IP:6099/webui
(如果没有 Token,输入 napcat 或查看日志 docker logs napcat)
- 扫码登录
使用手机 QQ 扫码。
- 配置 WebSocket (Reverse)
- 点击左侧 网络配置 -> WebSocket (Reverse)
- URL 填写:
ws://127.0.0.1:8080/onebot/v11/ws
- 启用:打钩
- 点击 保存。
systemctl start gemini-bot
🎮 最终:管理命令速查表
恭喜!部署完成。以后您只需要使用以下命令管理您的机器人:
| 功能 | 命令 |
|---|
| 启动机器人 | systemctl start gemini-bot |
| 停止机器人 | systemctl stop gemini-bot |
| 改代码后重启 | systemctl restart gemini-bot |
| 查看运行状态 | systemctl status gemini-bot |
| 实时查看日志 | journalctl -u gemini-bot -f |
(日志中若看到 Connect to 127.0.0.1:8080 或 Received message 即为成功)
相关免费在线工具
- 加密/解密文本
使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
- RSA密钥对生成器
生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online
- Mermaid 预览与可视化编辑
基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online
- curl 转代码
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
- Base64 字符串编码/解码
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
- Base64 文件转换器
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online