跳到主要内容
极客日志极客日志面向AI+效率的开发者社区
首页博客GitHub 精选镜像AI 生图工具UI配色美学隐私政策关于联系
搜索内容 / 工具 / 仓库 / 镜像...⌘K搜索
注册
博客列表
PythonAI

Gemini QQ 机器人部署指南

基于 Ubuntu 服务器部署 Gemini QQ 机器人,采用 OneBot V11 (NapCat) 协议端与 NoneBot2 框架。流程涵盖 Docker 环境搭建、NapCat 容器启动、Python 虚拟环境配置及核心代码编写。集成 Google Generative AI 实现群聊互动、记忆压缩及打卡统计功能。通过 Systemd 管理服务进程,支持 WebUI 扫码登录与 WebSocket 通信。提供管理员指令修改参数及查看排行榜等管理功能。

MongoKing发布于 2026/4/7更新于 2026/7/2331 浏览

Gemini QQ 机器人部署指南

核心架构:OneBot V11 (NapCat) + NoneBot2 + Gemini Flash 适用系统:Ubuntu 22.04 LTS (阿里云/腾讯云)


第一阶段:基础设施准备

SSH 连接服务器后,复制以下命令执行。

安装必要软件 (Docker + Python)

# 更新软件源
sudo apt update && sudo apt upgrade -y
# 安装 Docker
curl -fsSL https://get.docker.com | bash
# 安装 Python3 及虚拟环境工具
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 的问题。

  1. 放行防火墙端口
    • 请去阿里云/腾讯云控制台的'防火墙'或'安全组',放行 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 库,这是目前配合此代码最稳定的版本,没有任何属性报错。

  1. 创建 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
# 恢复旧版导入(关键!)
import google.generativeai as genai
import asyncio
import datetime
import pytz
import pickle
import os
import random

# ================= 核心配置区 =================
API_KEY = "在这里粘贴你的 Google_API_Key"
# 必须替换!
DATA_FILE = "bot_memory.pkl"
# ============================================

# 旧版兼容写法(0.3.2 支持)
genai.configure(api_key=API_KEY)
chat_model = genai.GenerativeModel('gemini-2.5-flash')

class GroupMemory:
    def __init__(self, group_id):
        self.group_id = group_id
        self.summary = "暂无早期历史记录。"
        self.buffer = []
        self.lock = asyncio.Lock()
        # 异步锁(无法序列化)
        # 状态追踪
        self.last_active_time = 0
        self.last_msg_content = ""
        self.repeat_count = 0

    # ===== 每个群独立的参数配置 =====
    self.bot_config = {
        "COOLDOWN_SECONDS": 0,  # 主动说话冷却 (0 分钟)
        "TRIGGER_PROBABILITY": 0.05,  # 闲聊插嘴概率 (5%)
        "REPEATER_THRESHOLD": 2,  # 复读检测阈值 (至少 2 人)
        "REPEATER_EXEC_PROB": 0.1,  # 复读执行概率 (10%,防止刷屏)
        "MEMORY_LIMIT_CHARS": 10000,  # 缓存字数上限
        "MEMORY_KEEP_COUNT": 100  # 压缩时保留最近 100 条
    }

    # ===== 修复:打卡数据存储(移除 lambda,改用普通字典)=====
    # 结构:{用户 ID: {"name": 用户名,"total_stars": 总星星数,"records": {打卡内容:次数}}}
    self.checkin_data = {}  # 普通字典,不再用 defaultdict(lambda)

    # ===== 新增:自定义序列化/反序列化方法(跳过 Lock)=====
    def __getstate__(self):
        # 序列化时,排除 lock 对象,返回其他需要保存的属性
        state = self.__dict__.copy()
        del state['lock']  # 移除无法序列化的 Lock
        return state

    def __setstate__(self, state):
        # 反序列化时,恢复所有属性,并重新初始化 Lock
        self.__dict__.update(state)
        self.lock = asyncio.Lock()  # 重新创建 Lock 对象

    # ========== 修复:打卡功能方法(手动初始化默认值)==========
    def checkin(self, user_id, user_name, content):
        """执行打卡:增加记录 + 总星星"""
        # 手动初始化用户数据(无 lambda,兼容 pickle)
        if user_id not in self.checkin_data:
            self.checkin_data[user_id] = {"name": "", "total_stars": 0, "records": {}}  # 普通字典存打卡内容
        user_data = self.checkin_data[user_id]
        user_data["name"] = user_name  # 更新最新昵称
        # 手动初始化打卡内容
        if content not in user_data["records"]:
            user_data["records"][content] = 0
        user_data["records"][content] += 1
        user_data["total_stars"] += 1  # 1 次打卡=1 颗星
        save_data()  # 立即持久化

    def delete_checkin(self, user_id, content):
        """删除打卡:减少记录 + 总星星(返回:是否成功,提示语)"""
        if user_id not in self.checkin_data:
            return False, "你还没有任何打卡记录哦~"
        user_data = self.checkin_data[user_id]
        if content not in user_data["records"] or user_data["records"][content] == 0:
            return False, f"你没有「{content}」的打卡记录~"
        # 减少次数
        user_data["records"][content] -= 1
        user_data["total_stars"] -= 1
        # 内容次数为 0 则移除该记录
        if user_data["records"][content] == 0:
            del user_data["records"][content]
        # 无任何记录则移除用户
        if not user_data["records"]:
            del self.checkin_data[user_id]
        save_data()
        return True, f"已删除「{content}」的 1 次打卡记录~"

    # ========== 管理员管理他人打卡记录的方法 ==========
    def admin_checkin(self, target_user_id, target_user_name, content, count, operation):
        """ 管理员操作他人打卡记录
        operation: "add"(增加) / "delete"(删除)
        返回:是否成功,提示语
        """
        # 1. 初始化目标用户数据
        if target_user_id not in self.checkin_data:
            self.checkin_data[target_user_id] = {"name": target_user_name, "total_stars": 0, "records": {}}
        user_data = self.checkin_data[target_user_id]
        user_data["name"] = target_user_name  # 更新目标用户昵称
        # 2. 初始化打卡内容
        if content not in user_data["records"]:
            user_data["records"][content] = 0
        # 3. 执行增加/删除操作
        if operation == "add":
            user_data["records"][content] += count
            user_data["total_stars"] += count
            save_data()
            return True, f"已为用户【{target_user_name}】增加「{content}」打卡记录 {count} 次~"
        elif operation == "delete":
            if user_data["records"][content] < count:
                return False, f"用户【{target_user_name}】的「{content}」打卡记录不足 {count} 次!"
            user_data["records"][content] -= count
            user_data["total_stars"] -= count
            # 内容次数为 0 则移除该记录
            if user_data["records"][content] == 0:
                del user_data["records"][content]
            # 无任何记录则移除用户
            if not user_data["records"]:
                del self.checkin_data[target_user_id]
            save_data()
            return True, f"已为用户【{target_user_name}】删除「{content}」打卡记录 {count} 次~"
        else:
            return False, "操作类型错误!仅支持「增加」或「删除」"

    def calculate_medals(self, stars):
        """计算星星对应的勋章:4 星=1 月,4 月=1 日,4 日=1 皇冠"""
        crown = stars // (4 * 4 * 4)  # 皇冠:64 星
        stars %= (4 * 4 * 4)
        sun = stars // (4 * 4)  # 太阳:16 星
        stars %= (4 * 4)
        moon = stars // 4  # 月亮:4 星
        star = stars % 4  # 剩余星星
        medals = ""
        if crown > 0:
            medals += "👑" * crown
        if sun > 0:
            medals += "☀️" * sun
        if moon > 0:
            medals += "🌙" * moon
        if star > 0 or (crown + sun + moon + star == 0):
            medals += "⭐" * star
        return medals

    def generate_checkin_ranking(self):
        """生成打卡排行榜(按总星星降序,子项也显示勋章)"""
        if not self.checkin_data:
            return "当前还没有打卡记录哦~快叫群友一起打卡吧!"
        # 按总星星排序(兼容普通字典)
        sorted_users = sorted(
            self.checkin_data.values(), key=lambda x: x["total_stars"], reverse=True
        )
        # 格式化输出
        ranking = "🏆 打卡记录排行榜 🏆\n"
        for idx, user in enumerate(sorted_users, 1):
            total_medals = self.calculate_medals(user["total_stars"])
            ranking += f"{idx}. {user['name']}{total_medals}\n"
            # 子项显示对应的勋章
            for content, count in user["records"].items():
                content_medals = self.calculate_medals(count)
                ranking += f" · {content}{content_medals}\n"
        return ranking

    # 原有方法(不变)
    def add_message(self, name, msg, time_str):
        entry = f"[{time_str}] 【{name}】: {msg}"
        self.buffer.append(entry)
        clean_msg = msg.strip()
        if clean_msg == self.last_msg_content and clean_msg:
            self.repeat_count += 1
        else:
            self.last_msg_content = clean_msg
            self.repeat_count = 1

    def estimate_length(self):
        return sum(len(m) for m in self.buffer)

    async def check_and_compress(self):
        async with self.lock:
            if self.estimate_length() > self.bot_config["MEMORY_LIMIT_CHARS"] and len(self.buffer) > self.bot_config["MEMORY_KEEP_COUNT"]:
                to_compress = self.buffer[:-self.bot_config["MEMORY_KEEP_COUNT"]]
                text_block = "\n".join(to_compress)
                prompt = f"""
 你是记忆整理员,更新长期记忆摘要(800 字内):
 【长期记忆】:{self.summary}
 【待归档对话】:{text_block}
 """
                try:
                    loop = asyncio.get_running_loop()
                    resp = await loop.run_in_executor(None, lambda: chat_model.generate_content(prompt))
                    if resp.candidates[0].content.parts[0].text:
                        self.summary = resp.candidates[0].content.parts[0].text
                        self.buffer = self.buffer[-self.bot_config["MEMORY_KEEP_COUNT"]:]
                        save_data()
                except Exception as e:
                    print(f"压缩记忆失败:{e}")
                    pass

    async def generate_reply(self, current_question, sender_name, is_active_interrupt=False):
        tz = pytz.timezone('Asia/Shanghai')
        context_str = "\n".join(self.buffer)
        # 构建回复提示词(补充'不要加前缀'的限制)
        if is_active_interrupt:
            task_prompt = "没人艾特你,自然加入讨论,简短、像群友,可复读/玩梗,直接说话不加称呼,绝对不要输出任何时间戳/日期/时间格式,不要添加自己的名字前缀(如【Gemini】)。"
        else:
            if not current_question.strip():
                task_prompt = f"用户【{sender_name}】艾特了你,自然接话,绝对不要输出任何时间戳/日期/时间格式,不要添加自己的名字前缀(如【Gemini】)。"
            else:
                task_prompt = f"用户【{sender_name}】提问:{current_question},请回答,绝对不要输出任何时间戳/日期/时间格式,不要添加自己的名字前缀(如【Gemini】)。"
        final_prompt = f"""
 你是 QQ 群里的 AI 伙伴 Gemini。
 【长期记忆】:{self.summary}
 【近期对话】:{context_str}
 【任务】:{task_prompt}
 【规则】:
 1. 问时间时只说当前北京时间(比如'现在是下午 3 点 20 分'),但不要输出任何括号/时间戳/数字格式的时间;
 2. 问新闻/数据自动联网,风格像群友而非客服;
 3. 回复时禁止添加自己的名字或任何标识前缀(如【Gemini】),直接输出回复内容。
 """
        try:
            loop = asyncio.get_running_loop()
            resp = await loop.run_in_executor(None, lambda: chat_model.generate_content(final_prompt))
            return resp.candidates[0].content.parts[0].text if resp.candidates[0].content.parts[0].text else "(暂时无法回复)"
        except Exception as e:
            return f"(回复失败:{str(e)})"

    async def check_active_intervention(self, current_msg):
        import time
        now = time.time()
        if now - self.last_active_time < self.bot_config["COOLDOWN_SECONDS"]:
            return False, None
        if (self.repeat_count >= self.bot_config["REPEATER_THRESHOLD"] and random.random() < self.bot_config["REPEATER_EXEC_PROB"] and current_msg.strip()):
            try:
                loop = asyncio.get_running_loop()
                check_resp = await loop.run_in_executor(None, lambda: chat_model.generate_content(f"判断'{current_msg}'是否是正常梗(非广告/脏话):是回 YES,否回 NO"))
                if "YES" in check_resp.candidates[0].content.parts[0].text.strip().upper():
                    self.last_active_time = now
                    return True, current_msg
            except:
                pass
        return False, None

        if "gemini" not in current_msg.lower() and random.random() > self.bot_config["TRIGGER_PROBABILITY"]:
            return False, None

        try:
            loop = asyncio.get_running_loop()
            decision_resp = await loop.run_in_executor(None, lambda: chat_model.generate_content(f"""
 最近消息:{chr(10).join(self.buffer[-15:])}
 当前消息:{current_msg}
 没被艾特,是否有必要说话?满足(客观问题无人答/讨论你/有趣的氛围/有趣的话题)回 YES,否则 NO
 """))
            if "YES" in decision_resp.candidates[0].content.parts[0].text.strip().upper():
                reply = await self.generate_reply("", "群友", is_active_interrupt=True)
                self.last_active_time = now
                return True, reply
        except Exception as e:
            print(f"主动介入失败:{e}")
            pass
        return False, None

# --- 记忆持久化(不变)---
memories = {}

def save_data():
    try:
        with open(DATA_FILE, 'wb') as f:
            pickle.dump(memories, f)
    except Exception as e:
        print(f"保存记忆失败:{e}")

def load_data():
    global memories
    if os.path.exists(DATA_FILE):
        try:
            with open(DATA_FILE, 'rb') as f:
                memories = pickle.load(f)
        except:
            memories = {}

def get_memory(group_id):
    if group_id not in memories:
        memories[group_id] = GroupMemory(group_id)
    return memories[group_id]

load_data()
  1. 创建 bot.py
    • 使用 nano bot.py 创建:
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  # 管理员 QQ 号(替换成实际的)
# ==============================================

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)  # 获取用户昵称(优先群名片,其次 QQ 昵称,最后用户 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 ""

    # ===== 1. 管理员专属:修改机器人参数 =====
    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)

    # ===== 2. 全员可用:打卡指令 =====
    elif msg_lower.startswith("gemini 打卡 "):
        content = get_cmd_content("gemini 打卡 ")
        if not content:
            reply_msg = "❌ 打卡内容不能为空哦~比如:「gemini 打卡 篮球」"
        else:
            # 执行打卡(user_id 转字符串确保唯一性)
            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)

    # ===== 3. 全员可用:删除打卡指令 =====
    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)

    # ===== 4. 全员可用:查询排行榜 =====
    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)

    # ===== 5. 管理员专属:管理他人打卡记录(QQ 号操作)=====
    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)

        # 校验 QQ 号为纯数字
        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)

        # 获取目标用户昵称(优先群名片,否则用 QQ 号)
        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  # 非群成员则显示 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)

    # ===== 6. 普通消息处理逻辑 =====
    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]
# 指定用户,通常是 root
User=root
# 你的项目路径
WorkingDirectory=/root/gemini_bot
# 使用虚拟环境的 Python 启动 bot.py
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。

  1. 浏览器访问 NapCat
    访问 http://服务器 IP:6099/webui
    (如果没有 Token,输入 napcat 或查看日志 docker logs napcat)
  2. 扫码登录
    使用手机 QQ 扫码。
  3. 配置 WebSocket (Reverse)
    • 点击左侧 网络配置 -> WebSocket (Reverse)
    • URL 填写:ws://127.0.0.1:8080/onebot/v11/ws
    • 启用:打钩
    • 点击 保存。

重启 NapCat 生效

docker restart napcat

启动机器人

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 即为成功)

目录

  1. Gemini QQ 机器人部署指南
  2. 第一阶段:基础设施准备
  3. 更新软件源
  4. 安装 Docker
  5. 安装 Python3 及虚拟环境工具
  6. 创建项目文件夹
  7. 第二阶段:部署 QQ 协议端 (NapCat)
  8. 第三阶段:部署核心代码
  9. 确保在项目目录
  10. 创建虚拟环境
  11. 激活环境
  12. 安装依赖 (锁定版本,拒绝报错)
  13. 恢复旧版导入(关键!)
  14. ================= 核心配置区 =================
  15. 必须替换!
  16. ============================================
  17. 旧版兼容写法(0.3.2 支持)
  18. --- 记忆持久化(不变)---
  19. ================= 管理员配置 =================
  20. ==============================================
  21. === 单核处理器 ===
  22. 第四阶段:配置系统服务 (Systemd)
  23. 指定用户,通常是 root
  24. 你的项目路径
  25. 使用虚拟环境的 Python 启动 bot.py
  26. 自动重启配置
  27. 第五阶段:NapCat 连接 (WebUI)
  28. 最终:管理命令速查表
  • 免费图片AI生成工具免费生成了解详情
  • Magick API 一键接入全球大模型注册送1000万token查看
  • 免费图片视频在线生成30秒,将你的创意变成现实开始设计
  • X/Twitter免费视频下载器免登陆无限额度免费视频解析下载了解详情
  • 100+免费在线小游戏爽一把
极客日志微信公众号二维码

微信扫一扫,关注极客日志

微信公众号「极客日志V2」,在微信中扫描左侧二维码关注。展示文案:极客日志V2 zeeklog

更多推荐文章

查看全部
  • 青少年机器人编程系统化学习路径:从机械启蒙到人工智能
  • 人工智能:自然语言处理在教育领域的应用与实战
  • JavaScript 空值判断工具函数
  • MySQL 常用命令速查表
  • 大模型常见面试题汇总与答案解析
  • LLaMA-Factory 详细安装教程
  • RabbitMQ 与 Spring Boot 集成实战:从 Hello World 到生产配置
  • 从多库并存到一库多能:金仓 KingbaseES 融合架构实践
  • Sora 模型技术报告:世界模拟器与视频生成能力解析
  • 三维人体姿态估计前沿算法与论文案例
  • NFT 元数据去中心化存储与智能合约集成实战
  • Python 空洞卷积网络架构与 PAMAP2 数据集实验分析
  • DeerFlow 2.0 生产级 AI Agent 框架的 Docker 部署与并行编排
  • ZeroClaw 开源:基于 Rust 的轻量级 AI Agent 框架
  • 直流无刷电机 FOC 控制算法
  • Python 爬虫实战:跨境电商数据采集与代理 IP 策略
  • WebGIS、无人机与 AI 融合:构建下一代智能巡检系统
  • 7 系列 FPGA 万兆以太网通信实现方案
  • 前端 WebSocket 实时通信实战:替代轮询方案
  • 基于 7 系列 FPGA 实现万兆网通信

相关免费在线工具

  • RSA密钥对生成器

    生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online

  • Mermaid 预览与可视化编辑

    基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online

  • 随机西班牙地址生成器

    随机生成西班牙地址(支持马德里、加泰罗尼亚、安达卢西亚、瓦伦西亚筛选),支持数量快捷选择、显示全部与下载。 在线工具,随机西班牙地址生成器在线工具,online

  • curl 转代码

    解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online

  • Base64 字符串编码/解码

    将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online

  • Base64 文件转换器

    将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online