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 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 \-eACCOUNT=你的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
# 恢复旧版导入(关键!)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"# ============================================# 旧版兼容写法(. 支持) genai(api_key=API_KEY) chat_model = genai('gemini--flash')classGroupMemory:(self, group_id): self.group_id = group_id self.summary = self.buffer=[] self.lock = asyncio.()# 异步锁(无法序列化)# 状态追踪 self.last_active_time = self.last_msg_content = self.repeat_count =# ===== 每个群独立的参数配置 ===== self.bot_config ={"COOLDOWN_SECONDS":,# 主动说话冷却 (分钟):,# 闲聊插嘴概率 ():,# 复读检测阈值 (至少人):,# 复读执行概率 (,防止刷屏):,# 缓存字数上限:# 压缩时保留最近 条}# ===== 修复:打卡数据存储(移除lambda,改用普通字典)=====# 结构:{用户ID: {"name": 用户名, : 总星星数, : {打卡内容: 次数}}} self ={}# 普通字典,不再用defaultdict(lambda)# ===== 新增:自定义序列化/反序列化方法(跳过Lock)=====def__getstate__(self):# 序列化时,排除lock对象,返回其他需要保存的属性 state = self.__dict__.()del state[]# 移除无法序列化的Lockreturn state (self, state):# 反序列化时,恢复所有属性,并重新初始化Lock self.__dict__.(state) self.lock = asyncio.()# 重新创建Lock对象# ========== 修复:打卡功能方法(手动初始化默认值)==========(self, user_id, user_name, content):# 手动初始化用户数据(无lambda,兼容pickle)if user_id notin self.checkin_data: self.checkin_data[user_id]={"name":,:,:{}# 普通字典存打卡内容} user_data = self user_data= user_name # 更新最新昵称# 手动初始化打卡内容if notin user_data: user_data[][content]= user_data[][content]+= user_data[]+=# 次打卡=颗星 ()# 立即持久化(self, user_id, content):if user_id notin self.checkin_data:returnFalse, user_data = self.checkin_data[user_id]if content notin user_data[]or user_data[][content]==:returnFalse,f# 减少次数 user_data[][content]-= user_data[]-=# 内容次数为则移除该记录if user_data[][content]==:del user_data[][content]# 无任何记录则移除用户ifnot user_data[]:del self.checkin_data[user_id] ()returnTrue,f# ========== 管理员管理他人打卡记录的方法 ==========(self, target_user_id, target_user_name, content, count, operation):adddelete# . 初始化目标用户数据if target_user_id notin self.checkin_data: self.checkin_data[target_user_id]={"name": target_user_name,:,:{}} user_data = self user_data= target_user_name # 更新目标用户昵称# . 初始化打卡内容if notin user_data: user_data[][content]=# . 执行增加/删除操作if operation ==: user_data[][content]+= count user_data[]+= count ()returnTrue,felif operation ==:if user_data[][content]< count:returnFalse,f user_data[][content]-= count user_data[]-= count # 内容次数为则移除该记录if user_data[][content]==:del user_data[][content]# 无任何记录则移除用户ifnot user_data[]:del self.checkin_data[target_user_id] ()returnTrue,felse:returnFalse,(self, stars): crown = stars //(**)# 皇冠:星 stars %=(**) sun = stars //(*)# 太阳:星 stars %=(*) moon = stars //# 月亮:星 star = stars %# 剩余星星 medals =if crown >: medals +=* crown if sun >: medals +=* sun if moon >: medals +=* moon if star >(crown+sun+moon+star ==): medals +=* star return medals (self):ifnot self.checkin_data:return# 按总星星排序(兼容普通字典) sorted_users =( self.checkin_data.(), key=lambda x: x[], reverse=True)# 格式化输出 ranking =for idx, user (sorted_users,): total_medals = self.(user[]) ranking +=f# 子项显示对应的勋章for content, count in user[].(): content_medals = self.(count) ranking +=freturn ranking # 原有方法(不变)(self, name, msg, time_str): entry =f self.buffer.(entry) clean_msg = msg.()if clean_msg == self.last_msg_content and clean_msg: self.repeat_count +=else: self.last_msg_content = clean_msg self.repeat_count =(self):((m)for m in self.buffer)(self):asyncwith self.lock:if self.()> self.bot_config[](self.buffer)> self.bot_config[]: to_compress = self.buffer[:-self.bot_config[]] text_block =.(to_compress) prompt =ftry: loop = asyncio.() resp =await loop.(None,lambda: chat_model.(prompt))if resp.candidates[].content.parts[].text: self.summary = resp.candidates[].content.parts[].text self.buffer= self.buffer[-self.bot_config[]:] ()except Exception as e:(f)(self, current_question, sender_name, is_active_interrupt=False): tz = pytz.() context_str =.(self.buffer)# 构建回复提示词(补充的限制)if is_active_interrupt: task_prompt =else:ifnot current_question.(): task_prompt =felse: task_prompt =f final_prompt =ftry: loop = asyncio.() resp =await loop.(None,lambda: chat_model.(final_prompt))return resp.candidates[].content.parts[].text if resp.candidates[].content.parts[].text elseexcept Exception as e:returnf(self, current_msg):import time now = time.()if now - self.last_active_time < self.bot_config[]:returnFalse,(self.repeat_count >= self.bot_config[]and random.()< self.bot_config[]and current_msg.()):try: loop = asyncio.() check_resp =await loop.(None,lambda: chat_model.(f))ifin check_resp.candidates[].content.parts[].text.().(): self.last_active_time = now returnTrue, current_msg except:passreturnFalse,Noneifnotin current_msg.()and random.()> self.bot_config[]:returnFalse,Nonetry: loop = asyncio.() decision_resp =await loop.(None,lambda: chat_model.(f))ifin decision_resp.candidates[].content.parts[].text.().(): reply =await self.(,, is_active_interrupt=True) self.last_active_time = now returnTrue, reply except Exception as e:(f)passreturnFalse,None# --- 记忆持久化(不变)--- memories ={}defsave_data():try:(DATA_FILE,)as f: pickle.(memories, f)except Exception as e:(f)():global memories if os.path.(DATA_FILE):try:(DATA_FILE,)as f: memories = pickle.(f)except: memories ={}defget_memory(group_id):if group_id notin memories: memories[group_id]= (group_id)return memories[group_id] ()

