简介这是一款基于Hook机制实现的微信机器人项目面向具备一定Python基础、希望自建微信自动化工具的开发者与运维爱好者可用于群管理、消息自动回复与AI对话等场景。项目经过一年迭代功能更贴近日常使用并保持不定期维护有代码能力者也可自行二次开发。压缩包共76个文件约4.93MB以34个png截图、12个py源码、9个pyc编译文件为主另含xml配置、yaml参数文件及README说明文档目录按Bot_Server、Recv_Msg_Dispose、Api_Server、Push_Server、Config等模块划分结构清晰便于理解消息接收、分发与推送的完整链路。目前已支持接入GPT、星火、千帆等AI能力以及关键词拉人进群、自动群发、入群欢迎等功能。已有846人学习下载适合想研究Hook机制、微信协议交互与机器人工程化落地的读者参考借鉴。1. 基于HOOK机制的微信机器人从源码包到跑通第一条自动回复微信生态里做自动化最怕的就是协议不稳定、账号被风控。市面上不少方案要么依赖网页版协议早就被限制得死死的要么走模拟点击这种笨路子延迟高还容易出错。NGCBot 这个项目走的是另一条路——基于 HOOK 机制直接拦截微信客户端的消息收发接口相当于在微信进程内部装了个“监听器”消息一来就能拿到回复也能直接注入。这套思路的好处是响应快、支持的功能全像 AI 对话、关键词拉群、自动群发、入群欢迎这些都能做。它适合有一定 Python 基础、想自己搭一个可控机器人的开发者也适合想研究 HOOK 通信机制的技术爱好者。源码包结构清晰按服务端、消息处理、API 接口、配置、推送几个模块拆开下面从目录结构开始拆。2. 源码目录拆解五个服务模块各管什么拿到 NGCBot-master.zip 解压后第一眼看到一堆文件夹和 .py 文件容易懵。其实整个项目是按“服务”来划分职责的每个服务跑一个独立进程通过本地端口或文件锁来协调。理解了这个分层后面改配置、加功能、排查问题都会顺很多。2.1 核心服务进程与启动顺序项目根目录下有几个关键的入口文件Main_Server.py主服务入口负责拉起其他子服务、加载配置、初始化数据库连接。Bot_Server/机器人核心逻辑所在里面main.py是消息接收和分发的起点。Recv_Msg_Dispose/消息处理层分Room_Msg_Dispose.py群消息和Friend_Msg_Dispose.py好友消息。Api_Server/对外接口层SparkApi.py是星火大模型的调用封装Api_Main_Server.py负责路由。Db_Server/数据库服务Db_Main_Server.py管连接池Db_Point_Server.py管积分相关的读写。Push_Server/推送服务Push_Main_Server.py处理主动消息推送。Cache/缓存服务Cache_Main_Server.py管理内存缓存和临时数据。Config/config.yaml全局配置文件所有可调参数都在这里。OutPut/OutPut.py日志和输出管理。启动顺序上常见做法是先跑Db_Server再跑Cache然后Api_Server最后Main_Server.py把所有服务串起来。不过项目里Main_Server.py一般会按依赖顺序自动拉起所以直接运行主入口就行。注意.wcf.lock这个文件它是 HOOK 框架的锁文件如果上次没正常退出这个文件残留会导致启动失败删掉再跑即可。2.2 消息从接收到回复的完整链路一条消息进来大致走这么几步HOOK 层拦截到微信客户端的消息回调写入本地缓存或直接推给Bot_Server/main.py。main.py根据消息类型群聊/私聊分发给Recv_Msg_Dispose下对应的处理脚本。处理脚本解析消息内容判断是否命中关键词、是否触发 AI 回复、是否需要拉人进群等。如果需要调用 AI走Api_Server里的封装把用户消息拼成 prompt 发给大模型拿到回复。回复内容再通过 HOOK 层注入微信发送接口完成自动回复。同时Db_Server会记录消息日志、积分变动等数据Push_Server负责定时或条件触发的主动推送。这个链路里Config/config.yaml控制着哪些功能开关打开、AI 用哪个平台、关键词列表是什么。下面重点看配置怎么改。2.3 config.yaml 关键参数逐项说明配置文件是 YAML 格式缩进敏感改的时候别用 Tab。常见参数分几块# 机器人基础配置 bot: name: NGCBot # 机器人昵称用于日志标识 admin_wxid: wxid_xxxxx # 管理员微信ID用于接收告警和特权命令 auto_start: true # 主服务启动时是否自动拉起子服务 # HOOK 相关 hook: lock_file: .wcf.lock # 锁文件路径异常退出后需手动删除 recv_timeout: 30 # 消息接收超时秒数网络差可调大 # AI 平台配置 ai: enable: true provider: spark # 可选 spark / gpt / qianfan spark: app_id: 你的app_id api_key: 你的api_key api_secret: 你的api_secret model: generalv3 # 星火模型版本 gpt: api_key: sk-xxxx base_url: https://api.openai.com/v1 model: gpt-3.5-turbo qianfan: ak: 你的ak sk: 你的sk # 群管理功能 group: welcome_enable: true # 入群欢迎开关 welcome_msg: 欢迎新朋友请先看群公告 keyword_invite: enable: true keywords: [拉我, 进群, 邀请] target_group: 群ID或群名 auto_broadcast: enable: false cron: 0 9 * * * # 每天9点群发cron表达式 msg: 早上好今日资讯已更新 # 数据库 database: type: sqlite # 默认sqlite可换mysql path: ./data/ngcbot.db pool_size: 5几个容易踩坑的参数admin_wxid必须填对否则管理员命令不生效而且收不到异常告警。provider切换后对应平台的 key 必须填全缺一个就报鉴权失败。keyword_invite.keywords是列表匹配方式是包含匹配不是完全相等。比如用户发“能拉我进群吗”包含“拉我”就会触发。auto_broadcast.cron用的是标准 cron 表达式分 时 日 月 周别写反。database.path如果是相对路径相对于Main_Server.py所在目录不是项目根目录这个容易搞混。改完配置后建议先用python -c import yaml; yaml.safe_load(open(Config/config.yaml))检查语法没问题再启动。3. 环境搭建与首次运行依赖、数据库、AI 接口三件套配置看懂了接下来动手跑起来。这一章按顺序把环境依赖、数据库初始化、AI 接口调试三件事做完每一步都有可复制的命令和代码。3.1 Python 依赖安装与版本选择项目根目录有requirements.txt先看里面有没有版本锁。常见依赖包括pyyaml、requests、websockets、sqlalchemy等。建议用 Python 3.8 到 3.10太新的版本某些 HOOK 库可能不兼容。# 创建虚拟环境避免污染全局 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 安装依赖建议加国内镜像加速 pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple # 如果 requirements.txt 里没锁版本可以手动固定几个关键库 pip install pyyaml6.0 requests2.31.0 sqlalchemy2.0.0安装完用pip list核对一下重点看pyyaml和requests有没有装上。如果报Microsoft Visual C 14.0 is required说明某个库需要编译去装一下 VC 构建工具或者找对应的 whl 包手动装。3.2 数据库初始化与表结构确认项目默认用 SQLite开箱即用不用额外装数据库服务。第一次运行Db_Server/Db_Main_Server.py时会自动建表。也可以手动跑一次初始化# 手动初始化数据库在项目根目录执行 from Db_Server.Db_Main_Server import init_db # init_db 会读取 config.yaml 里的 database 配置 # 然后创建消息记录表、积分表、用户表等 init_db() print(数据库初始化完成)执行后检查data/ngcbot.db文件是否生成。用 SQLite 命令行工具看一下表结构sqlite3 data/ngcbot.db .tables # 应该能看到 msg_log、user_point、group_config 等表 sqlite3 data/ngcbot.db .schema msg_log # 查看具体字段确认有 wxid、content、timestamp 等列如果要用 MySQL改config.yaml里的database.type为mysql并补上 host、port、user、password、dbname 几个字段。注意 MySQL 需要提前建好库SQLite 不用。3.3 AI 接口连通性测试AI 回复是核心功能之一跑通之前先单独测接口别等机器人跑起来才发现 key 不对。以星火为例# 测试星火接口连通性 import requests import json from Config.config_loader import load_config cfg load_config() spark_cfg cfg[ai][spark] url https://spark-api-open.xf-yun.com/v1/chat/completions headers { Authorization: fBearer {spark_cfg[api_key]}:{spark_cfg[api_secret]}, Content-Type: application/json } data { model: spark_cfg[model], messages: [{role: user, content: 你好测试一下}] } resp requests.post(url, headersheaders, jsondata, timeout10) print(resp.status_code) print(resp.json())返回 200 且 JSON 里有choices字段说明接口通了。如果返回 401检查 api_key 和 api_secret 是否填反返回 404检查 model 名称是否拼错。GPT 和千帆的测试方式类似换对应的 url 和鉴权头即可。三个都通了之后直接运行python Main_Server.py看到日志里出现“所有服务启动完成”就说明机器人跑起来了。这时候用另一个微信号给机器人发条消息看能不能收到自动回复。4. 避坑与排查锁文件、编码、风控、AI 超时跑起来只是第一步实际用起来会遇到各种奇怪问题。这一章列几个高频翻车场景每条按现象、原因、解决来写都是血泪经验。4.1 启动报错 .wcf.lock 被占用现象运行Main_Server.py直接退出日志提示lock file exists或another instance is running。原因上次机器人没正常关闭.wcf.lock文件残留。HOOK 框架用这个文件防止多实例冲突但异常退出时不会自动清理。解决手动删掉项目根目录和Bot_Server/下的.wcf.lock再重新启动。如果经常出现可以在启动脚本里加一段清理逻辑import os lock_files [.wcf.lock, Bot_Server/.wcf.lock] for lf in lock_files: if os.path.exists(lf): os.remove(lf) print(f已清理残留锁文件: {lf})4.2 中文消息乱码或回复内容截断现象收到的消息里中文变成问号或者 AI 回复只显示一半。原因编码不统一。HOOK 层传过来的可能是 GBK 编码而 Python 脚本默认 UTF-8 处理中间没做转换。另外消息长度超过微信单条限制也会被截断。解决在消息处理入口加编码转换并对长回复做分片发送。def safe_decode(raw): 尝试多种编码解码返回可读字符串 if isinstance(raw, bytes): for enc in [utf-8, gbk, gb2312]: try: return raw.decode(enc) except UnicodeDecodeError: continue return raw.decode(utf-8, errorsignore) return str(raw) def split_long_msg(msg, max_len500): 长消息分片避免被微信截断 return [msg[i:imax_len] for i in range(0, len(msg), max_len)]4.3 频繁回复触发风控现象机器人用了一段时间后发消息失败或者账号被限制部分功能。原因回复频率太高、内容重复度高、短时间内大量加好友或拉群都会触发微信的风控策略。解决加发送间隔和随机延迟控制单位时间内的消息量。import time import random def send_with_delay(send_func, msg, min_delay1.5, max_delay4.0): 带随机延迟的发送降低风控概率 time.sleep(random.uniform(min_delay, max_delay)) return send_func(msg)另外auto_broadcast群发功能建议只在低峰期用且内容每次微调别一模一样。4.4 AI 接口超时导致消息堆积现象用户发消息后很久没回复日志里一堆timeout后面消息越积越多。原因AI 接口响应慢或网络抖动同步调用阻塞了消息处理线程。解决给 AI 调用设超时并加降级回复。import requests def call_ai(prompt, timeout8): try: resp requests.post(AI_URL, json{prompt: prompt}, timeouttimeout) return resp.json().get(reply, 我暂时想不出来稍后再试) except requests.Timeout: return 网络有点慢请再发一次 except Exception as e: print(fAI调用异常: {e}) return 服务开小差了稍后恢复超时时间别设太长8 到 10 秒足够再长用户也等不了。4.5 数据库锁导致写入失败现象日志报database is locked消息记录写不进去。原因SQLite 并发写入能力弱多个服务同时写同一张表就会锁。解决要么换 MySQL要么在 SQLite 连接上加timeout参数并减少并发写入。from sqlalchemy import create_engine engine create_engine( sqlite:///data/ngcbot.db, connect_args{timeout: 15, check_same_thread: False} )check_same_threadFalse允许跨线程使用连接timeout15表示锁等待 15 秒后超时。如果写入量确实大还是建议上 MySQL。5. 进阶玩法关键词拉群与定时群发的参数调优基础功能跑通后关键词拉群和定时群发是最能体现机器人价值的两个点但参数没调好容易翻车。这一章把这两个功能的实现逻辑和调优技巧拆开讲。5.1 关键词拉群的匹配策略与防重复关键词拉群的逻辑是监听群消息如果内容命中配置的关键词列表就自动把发送者拉进目标群。听起来简单但实际用起来有几个细节匹配方式默认是包含匹配用户发“求拉我进群”能命中“拉我”。但如果关键词太短比如“群”几乎每条消息都命中会疯狂拉人。建议关键词至少 2 个字且用组合词。防重复同一个人短时间内多次触发应该只拉一次。可以用缓存记录最近触发过的 wxid。from Cache.Cache_Main_Server import cache def handle_keyword_invite(wxid, msg, keywords, target_group): 关键词拉群处理 # 检查是否已触发过缓存有效期 300 秒 cache_key finvite_{wxid} if cache.get(cache_key): return # 已拉过跳过 for kw in keywords: if kw in msg: # 执行拉群操作 invite_to_group(wxid, target_group) # 写入缓存防止重复 cache.set(cache_key, True, expire300) print(f已拉 {wxid} 进群触发词: {kw}) breakcache.set的expire参数控制防重复时间窗口300 秒够用。如果群人数多可以调大到 600 秒。5.2 定时群发的 cron 表达式与内容轮换定时群发用 cron 控制时间常见配置表达式含义0 9 * * *每天 9:000 9,18 * * *每天 9:00 和 18:000 9 * * 1-5工作日 9:00*/30 * * * *每 30 分钟内容轮换是为了避免被判定为垃圾消息。可以把多条文案存列表每次随机选一条或者按顺序轮换。import random broadcast_msgs [ 早上好今日天气不错记得吃早餐, 新的一天开始了有什么想聊的, 群内资讯已更新欢迎查看。 ] def do_broadcast(group_list): msg random.choice(broadcast_msgs) for g in group_list: send_with_delay(send_group_msg, msg, g) print(f已群发到 {g})random.choice每次随机选一条降低重复度。如果群特别多建议分批发送每批间隔几分钟。5.3 用日志反查消息处理链路功能多了之后出问题不好定位。OutPut/OutPut.py里一般会封装日志方法建议在关键节点都打上日志并带上消息 ID 或 wxid方便串联。from OutPut.OutPut import log_info, log_error def process_message(msg_id, wxid, content): log_info(f[{msg_id}] 收到消息 from {wxid}: {content[:50]}) try: # 处理逻辑 reply generate_reply(content) log_info(f[{msg_id}] 生成回复: {reply[:50]}) return reply except Exception as e: log_error(f[{msg_id}] 处理失败: {e}) return None日志里带上msg_id从接收到回复整条链路都能追。如果某个用户反馈没收到回复直接搜 wxid 就能看到卡在哪一步。5.4 验证机器人是否正常工作的检查清单最后给一个快速自检清单每次改完配置或更新代码后跑一遍检查项方法预期结果锁文件已清理ls .wcf.lock文件不存在配置文件语法python -c import yaml; yaml.safe_load(open(Config/config.yaml))无报错数据库可写手动插入一条测试记录插入成功AI 接口通跑 3.3 的测试脚本返回 200消息收发正常给机器人发“测试”收到回复关键词拉群发含关键词的消息被拉进目标群日志无 ERRORgrep ERROR logs/*.log无输出这套流程走下来基本能覆盖日常维护的大部分场景。从那以后我每次改完配置都强制走一遍这个清单省得半夜被报警叫醒。希望帮到你。本文还有配套的精品资源点击获取