保姆级教程：用Python和go-cqhttp给QQ群做个自动回复机器人（附完整代码）

从零构建智能QQ群机器人Python与go-cqhttp实战指南在社群运营中自动回复机器人早已成为提升管理效率的利器。无论是处理高频咨询、新人引导还是定时发布公告一个配置得当的机器人能显著减轻管理员负担。本文将带你深入探索如何基于go-cqhttp框架和Python打造功能丰富的QQ群机器人涵盖从环境搭建到高级功能的完整实现路径。1. 环境准备与基础配置1.1 go-cqhttp的获取与初始化首先需要从官方GitHub仓库下载对应系统版本的go-cqhttp可执行文件。对于Windows用户推荐选择go-cqhttp_windows_amd64.exe版本。下载完成后建议将其放置在独立的项目目录中。首次运行时程序会提示选择通信方式请选择您需要创建的通信方式: 0: HTTP通信 1: 正向WebSocket 2: 反向WebSocket输入0选择HTTP通信后程序会自动生成config.yml配置文件。用文本编辑器打开该文件关键配置项包括account: # 账号相关 uin: 123456789 # QQ账号 password: # 密码为空时使用扫码登录 heartbeat: interval: 0 # 关闭心跳检测 servers: - http: host: 127.0.0.1 port: 5700 post: - url: http://127.0.0.1:5701 # 事件上报地址1.2 Python环境准备确保已安装Python 3.7版本并安装必要的依赖库pip install requests fastapi uvicorn python-dotenv建议使用虚拟环境隔离项目依赖python -m venv venv source venv/bin/activate # Linux/Mac venv\Scripts\activate # Windows2. 核心通信机制实现2.1 HTTP API基础交互go-cqhttp通过HTTP接口提供丰富的功能支持。以下是一个基础的Python封装类import requests from typing import Dict, Any class CQHttpAPI: def __init__(self, api_urlhttp://127.0.0.1:5700): self.api_url api_url def call_api(self, endpoint: str, data: Dict[str, Any]) - Dict: 调用go-cqhttp API url f{self.api_url}/{endpoint} headers {Content-Type: application/json} response requests.post(url, jsondata, headersheaders) return response.json() def send_private_msg(self, user_id: int, message: str) - Dict: 发送私聊消息 data { user_id: user_id, message: message, auto_escape: False # 允许CQ码解析 } return self.call_api(send_private_msg, data) def send_group_msg(self, group_id: int, message: str) - Dict: 发送群消息 data { group_id: group_id, message: message, auto_escape: False } return self.call_api(send_group_msg, data)2.2 消息接收与处理go-cqhttp通过配置的post地址上报事件。我们可以使用FastAPI搭建一个简单的接收服务from fastapi import FastAPI, Request from pydantic import BaseModel app FastAPI() class GroupMessage(BaseModel): message_type: str group_id: int user_id: int message: str raw_message: str app.post(/event) async def handle_event(request: Request): data await request.json() if data.get(post_type) message: if data.get(message_type) group: # 处理群消息 msg GroupMessage(**data) print(f收到群{msg.group_id}消息{msg.message}) # 简单回复逻辑 if 你好 in msg.message: api CQHttpAPI() api.send_group_msg(msg.group_id, f[CQ:at,qq{msg.user_id}] 你好呀) return {status: ok}启动服务命令uvicorn main:app --reload --port 57013. 实用功能实现3.1 新人入群欢迎通过监听群成员增加事件实现欢迎功能class GroupIncrease(BaseModel): post_type: str notice_type: str group_id: int user_id: int app.post(/event) async def handle_event(request: Request): data await request.json() # 新人入群处理 if data.get(notice_type) group_increase: event GroupIncrease(**data) welcome_msg ( f[CQ:at,qq{event.user_id}] 欢迎加入本群\n 请先阅读群公告有问题可以随时管理员~ ) api CQHttpAPI() api.send_group_msg(event.group_id, welcome_msg) # ...其他事件处理3.2 关键词自动回复实现基于关键词的智能回复系统KEYWORD_RESPONSES { 帮助: 本群提供以下帮助\n1. 常见问题解答\n2. 技术支持\n3. 活动咨询, 活动: 最新活动\n- 本周六线上分享会\n- 下月初线下见面会, 规则: 群规则\n1. 禁止广告\n2. 禁止人身攻击\n3. 保持友善交流 } app.post(/event) async def handle_event(request: Request): data await request.json() if data.get(post_type) message and data.get(message_type) group: msg GroupMessage(**data) content msg.message.strip() # 关键词匹配 for keyword, response in KEYWORD_RESPONSES.items(): if keyword in content: api CQHttpAPI() reply f[CQ:at,qq{msg.user_id}] {response} api.send_group_msg(msg.group_id, reply) break3.3 定时任务管理使用APScheduler实现定时公告功能from apscheduler.schedulers.background import BackgroundScheduler from datetime import datetime scheduler BackgroundScheduler() def daily_announcement(): now datetime.now().strftime(%Y-%m-%d %H:%M) message f【每日公告】{now}\n今日话题AI技术交流\n请遵守群规理性讨论 api CQHttpAPI() api.send_group_msg(12345678, message) # 替换为你的群号 # 每天上午10点执行 scheduler.add_job(daily_announcement, cron, hour10) scheduler.start()4. 高级功能与优化4.1 消息持久化与上下文使用SQLite存储消息记录实现上下文感知回复import sqlite3 from contextlib import closing def init_db(): with closing(sqlite3.connect(bot.db)) as conn: cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, group_id INTEGER, user_id INTEGER, message TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() def save_message(group_id: int, user_id: int, message: str): with closing(sqlite3.connect(bot.db)) as conn: cursor conn.cursor() cursor.execute( INSERT INTO messages (group_id, user_id, message) VALUES (?, ?, ?), (group_id, user_id, message) ) conn.commit()4.2 敏感词过滤实现基础的敏感词检测功能SENSITIVE_WORDS [广告, 赌博, 诈骗] def contains_sensitive(text: str) - bool: return any(word in text for word in SENSITIVE_WORDS) app.post(/event) async def handle_event(request: Request): data await request.json() if data.get(post_type) message and data.get(message_type) group: msg GroupMessage(**data) if contains_sensitive(msg.message): api CQHttpAPI() warning ( f[CQ:at,qq{msg.user_id}] 您的消息包含敏感内容请注意遵守群规\n 多次违规将被移出群聊。 ) api.send_group_msg(msg.group_id, warning)4.3 性能优化建议异步处理使用httpx替代requests进行异步HTTP请求连接池复用HTTP连接减少开销消息队列高负载场景下引入Redis作为消息队列import httpx async def async_send_group_msg(group_id: int, message: str): async with httpx.AsyncClient() as client: data {group_id: group_id, message: message} await client.post(http://127.0.0.1:5700/send_group_msg, jsondata)5. 部署与维护5.1 生产环境部署推荐使用supervisor管理进程[program:go-cqhttp] command/path/to/go-cqhttp directory/path/to/working/dir autostarttrue autorestarttrue [program:bot] commanduvicorn main:app --host 0.0.0.0 --port 5701 directory/path/to/bot autostarttrue autorestarttrue5.2 日志管理配置go-cqhttp的日志输出# config.yml log: level: info format: json output: file path: logs/go-cqhttp.logPython端使用logging模块import logging from logging.handlers import RotatingFileHandler logger logging.getLogger(qqbot) logger.setLevel(logging.INFO) handler RotatingFileHandler( logs/bot.log, maxBytes5*1024*1024, backupCount3 ) formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s ) handler.setFormatter(formatter) logger.addHandler(handler)5.3 安全建议使用环境变量存储敏感信息限制API访问IP定期更新go-cqhttp版本from dotenv import load_dotenv import os load_dotenv() API_URL os.getenv(CQHTTP_API_URL, http://127.0.0.1:5700)