Gemini 全能 QQ 机器人部署手册 (V1.0 Release)

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

🟢 第一阶段：基础设施准备

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

安装必要软件 (Docker + Python)

# 更新软件源sudoapt update &&sudoapt upgrade -y# 安装 Dockercurl-fsSL https://get.docker.com |bash# 安装 Python3 及虚拟环境工具sudoaptinstall python3-pip python3-venv -y# 创建项目文件夹mkdir-p /root/gemini_bot  /root/gemini_bot

# 恢复旧版导入（关键！）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')classGroupMemory: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']# 移除无法序列化的Lockreturn state def__setstate__(self, state):# 反序列化时，恢复所有属性，并重新初始化Lock self.__dict__.update(state) self.lock = asyncio.Lock()# 重新创建Lock对象# ========== 修复：打卡功能方法（手动初始化默认值）==========defcheckin(self, user_id, user_name, content):"""执行打卡：增加记录+总星星"""# 手动初始化用户数据（无lambda，兼容pickle）if user_id notin 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 notin user_data["records"]: user_data["records"][content]=0 user_data["records"][content]+=1 user_data["total_stars"]+=1# 1次打卡=1颗星 save_data()# 立即持久化defdelete_checkin(self, user_id, content):"""删除打卡：减少记录+总星星（返回：是否成功, 提示语）"""if user_id notin self.checkin_data:returnFalse,"你还没有任何打卡记录哦~" user_data = self.checkin_data[user_id]if content notin user_data["records"]or user_data["records"][content]==0:returnFalse,f"你没有「{content}」的打卡记录~"# 减少次数 user_data["records"][content]-=1 user_data["total_stars"]-=1# 内容次数为0则移除该记录if user_data["records"][content]==0:del user_data["records"][content]# 无任何记录则移除用户ifnot user_data["records"]:del self.checkin_data[user_id] save_data()returnTrue,f"已删除「{content}」的1次打卡记录~"# ========== 管理员管理他人打卡记录的方法 ==========defadmin_checkin(self, target_user_id, target_user_name, content, count, operation):""" 管理员操作他人打卡记录 operation: "add"（增加） / "delete"（删除）返回：是否成功, 提示语 """# 1. 初始化目标用户数据if target_user_id notin 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 notin user_data["records"]: user_data["records"][content]=0# 3. 执行增加/删除操作if operation =="add": user_data["records"][content]+= count user_data["total_stars"]+= count save_data()returnTrue,f"已为用户【{target_user_name}】增加「{content}」打卡记录 {count} 次~"elif operation =="delete":if user_data["records"][content]< count:returnFalse,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]# 无任何记录则移除用户ifnot user_data["records"]:del self.checkin_data[target_user_id] save_data()returnTrue,f"已为用户【{target_user_name}】删除「{content}」打卡记录 {count} 次~"else:returnFalse,"操作类型错误！仅支持「增加」或「删除」"defcalculate_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 >0or(crown+sun+moon+star ==0): medals +="⭐"* star return medals defgenerate_checkin_ranking(self):"""生成打卡排行榜（按总星星降序，子项也显示勋章）"""ifnot self.checkin_data:return"当前还没有打卡记录哦~快叫群友一起打卡吧！"# 按总星星排序（兼容普通字典） sorted_users =sorted( self.checkin_data.values(), key=lambda x: x["total_stars"], reverse=True)# 格式化输出 ranking ="🏆 打卡记录排行榜 🏆\n"for idx, user inenumerate(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 # 原有方法（不变）defadd_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 +=1else: self.last_msg_content = clean_msg self.repeat_count =1defestimate_length(self):returnsum(len(m)for m in self.buffer)asyncdefcheck_and_compress(self):asyncwith self.lock:if self.estimate_length()> self.bot_config["MEMORY_LIMIT_CHARS"]andlen(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}")passasyncdefgenerate_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:ifnot 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:returnf"（回复失败：{str(e)}）"asyncdefcheck_active_intervention(self, current_msg):import time now = time.time()if now - self.last_active_time < self.bot_config["COOLDOWN_SECONDS"]:returnFalse,Noneif(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 returnTrue, current_msg except:passreturnFalse,Noneif"gemini"notin current_msg.lower()and random.random()> self.bot_config["TRIGGER_PROBABILITY"]:returnFalse,Nonetry: 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 returnTrue, reply except Exception as e:print(f"主动介入失败：{e}")passreturnFalse,None# --- 记忆持久化（不变）--- memories ={}defsave_data():try:withopen(DATA_FILE,'wb')as f: pickle.dump(memories, f)except Exception as e:print(f"保存记忆失败：{e}")defload_data():global memories if os.path.exists(DATA_FILE):try:withopen(DATA_FILE,'rb')as f: memories = pickle.load(f)except: memories ={}defget_memory(group_id):if group_id notin memories: memories[group_id]= GroupMemory(group_id)return memories[group_id] load_data()

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)defunix_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")defget_current_bj_time():"""获取当前北京时间字符串"""return datetime.datetime.now(pytz.timezone('Asia/Shanghai')).strftime("%H:%M:%S")defis_at_me(bot: Bot, event: GroupMessageEvent)->bool:"""判断是否@了机器人"""if event.is_tome():returnTruefor seg in event.message:if seg.type=="at"andstr(seg.data.get("qq"))==str(bot.self_id):returnTruereturnFalse# === 单核处理器 === handler = on_message(priority=1, block=True)@handler.handle()asyncdef_(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')orstr(user_id)except: name ="群友"# ===== 核心：统一转小写判断指令前缀（不区分大小写）===== msg_lower = msg.lower()# 转小写，仅用于指令判断# 提取指令内容（保留原大小写）defget_cmd_content(prefix): prefix_len =len(prefix)# 精确匹配前缀（原大小写）if msg.startswith(prefix):return msg[prefix_len:].strip()# 兼容大小写的前缀匹配elif msg_lower.startswith(prefix.lower()):for i inrange(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)iflen(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 notin 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 打卡 ")ifnot 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 删除 ")ifnot 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 管理打卡 ")ifnot 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()iflen(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 notin["增加","删除"]: 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号为纯数字ifnot 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)# 校验次数为正整数ifnot count_str.isdigit()orint(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. 普通消息处理逻辑 =====ifnot msg: msg =" "# 记录普通消息 mem.add_message(name, msg, unix_to_beijing(event.time)) save_data()# 消息压缩逻辑await mem.check_and_compress()# 主动回复/插话逻辑 final_reply =Noneif 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)

功能	命令
启动机器人	`systemctl start gemini-bot`
停止机器人	`systemctl stop gemini-bot`
改代码后重启	`systemctl restart gemini-bot`
查看运行状态	`systemctl status gemini-bot`
实时查看日志	`journalctl -u gemini-bot -f`

Gemini 全能 QQ 机器人部署手册 (V1.0 Release)