wechat-bot 防封实战:基于 WeChaty 的行为整形、限速与运行期自救
【免费下载链接】wechat-bot🤖 Multi-platform IM AI Agent for Telegram, WhatsApp, Lark, and WeChat. Connects ChatGPT / Claude / Kimi / DeepSeek / Ollama / Pi for auto-replies, community analysis, contact management, and inactive-friend detection.项目地址: https://gitcode.com/GitHub_Trending/we/wechat-bot
你跑起来的 wechat-bot,主号会话常常两三周失效。本文在现有 WeChaty 流程上补四道护栏:步幅整形、内容整形、授权收敛、运行期自救,让微信机器人防封方案落到真实的 sendMessage.js 上,压缩行为与内容两类线索。
先看机制:风控靠哪三类线索判定机器人
风控盯的不是"机器人"三个字,而是行为节奏、内容统计、会话环境这三类可量化线索。当前仓库恰好在前两类上完全裸奔,第三类则决定了这套方案的能力边界。
| 线索维度 | 风控观察到的异常 | 本仓库现状 |
|---|---|---|
| 行为节奏 | 发送间隔过于规律、全天无作息 | getReply()返回后立刻say(),中间零延迟 |
| 内容统计 | 高重复率、模板化输出 | 同一模型加同一提示词,回复指纹高度一致 |
| 会话环境 | 非官方客户端协议、会话频繁失效 | puppet 为wechaty-puppet-wechat4u,走网页协议 |
sendMessage.js 里唯一的一层闸门是白名单判定,摘出来看很短:
// src/wechaty/sendMessage.js(节选) const isRoom = roomWhiteList.includes(roomName) && content.includes(`${botName}`) const isAlias = aliasWhiteList.includes(remarkName) || aliasWhiteList.includes(name) const isAuthorizedCommand = (room && isRoom) || (!room && isAlias) if (!isAuthorizedCommand) return const response = await getReply(question) await room.say(response)这段代码说明了三件事:
- 授权判定只发生在"收"的一侧,白名单之外静默丢弃,这一点是对的,应保留。
- 从
getReply()到say()之间没有任何间隔、去重、时段判断,回复速度和密度完全由 AI 服务的响应时间决定,比真人反应快且规律。 - 白名单收敛的是"谁触发",不收敛"发多少、发什么",后两者正是行为节奏和内容统计两类线索的来源。
需要说清的边界:第三类线索(网页协议、扫码会话)改业务代码无解,本文只能靠"低频、小会话、慢养"去压低它的暴露面,不要把代码方案当成对协议风险的承诺。
总体蓝图:四层护栏插在消息事件与 say() 之间
四道护栏以漏斗形式串在现有链路里,每一层都能独立开关。
数据流一句话概括:消息事件先过闸门,再等步幅,取回复后整形,最后才发送;每次say产生的self记录回写 messageStore.js 维护的messages.jsonl,哨兵每 5 分钟回读该文件,按级别反向调节步幅系数与闸门口径。选这条回路是因为WECHAT_STORE_MESSAGES默认就是开启的,护栏直接复用现有落盘,不用新造日志体系。
分模块落地:三个模块互不依赖,可逐个上线
三个模块各自一个文件、一个开关,出问题时回滚粒度是单个文件。以下代码均为新增建议,不改动仓库现有文件的调用方式。
给发送加上步幅整形
设计思路:真人的特征是间隔服从一个宽分布,而不是恒定值。所以核心是"随机区间 + 消息长度加成 + 最小间隔兜底",再叠一个静默时段直接放弃部分回复,制造"人下班了"的缺口。刻意不用固定 1 秒这类整数延迟,因为规律性本身就是特征。
核心代码:
// src/guard/pacer.js const GAP = { dm: [1200, 3800], // 私聊 [min, max],毫秒 room: [2500, 7000], // 群聊,更稀疏 } const QUIET = { weekday: [[0, 9], [22, 24]], weekend: [[0, 10], [23, 24]], } export class SendPacer { constructor({ minGapMs = 600, quietDrop = 0.08, gapFactor = 1 } = {}) { this.minGapMs = minGapMs this.quietDrop = quietDrop this.gapFactor = gapFactor // 哨兵可调,默认 1 } inQuietWindow(now = new Date()) { const day = now.getDay() const table = day >= 1 && day <= 5 ? QUIET.weekday : QUIET.weekend const hour = now.getHours() return table.some(([from, to]) => hour >= from && hour < to) } async nextSlot(kind, textLength) { if (this.inQuietWindow() && Math.random() < this.quietDrop) return null const [lo, hi] = GAP[kind] || GAP.dm const lengthBoost = Math.min(textLength / 200, 2) * 800 const target = (lo + Math.random() * (hi - lo) + lengthBoost) * this.gapFactor const wait = Math.max(target - (Date.now() - (this.lastOutAt || 0)), this.minGapMs) await new Promise((resolve) => setTimeout(resolve, wait)) this.lastOutAt = Date.now() return true } }接入点在 bot.js 的message事件里,把defaultMessage包一层:nextSlot返回null就放弃本轮。
关键参数与边界:
- 群成员多、回复密度大时把
GAP.room上限拉到 12000 以上,群聊是重复率最容易爆的场景。 quietDrop是"放弃回复"的概率而非开关,小号可提到 0.2;主号承载真人会话时必须设 0,否则会漏答真人消息,代价是白天节奏失真。- 机器时钟漂移会直接污染间隔分布,先保证 NTP 正常再谈调参。
给回复加上内容整形
设计思路:模型输出天然趋同,同一问题换个问法也给出近似文本。整形分三步:敏感词打码、句式微调、与历史self记录做 n-gram 重叠率比对,超标即弃发。比对对象直接用仓库已有的loadWechatMessages,不额外维护一份历史。
核心代码:
// src/guard/shaper.js import { loadWechatMessages } from '../platforms/wechat/messageStore.js' const OPENERS = ['好,', '嗯,', '查了下,', ''] export class ReplyShaper { constructor({ dataDir = '.data/wechat', repeatThreshold = 0.82, words = [] } = {}) { this.dataDir = dataDir this.repeatThreshold = repeatThreshold this.words = words } mask(raw) { return this.words.reduce( (text, word) => text.split(word).join('*'.repeat(word.length)), raw ) } vary(raw) { const opener = OPENERS[Math.floor(Math.random() * OPENERS.length)] return raw.trim() ? `${opener}${raw.trim()}` : raw } overlapWith(selfText, candidate, n = 8) { const grams = (text) => { const set = new Set() for (let i = 0; i + n <= text.length; i++) set.add(text.slice(i, i + n)) return set } const base = grams(selfText) if (!base.size) return 0 let hit = 0 for (const gram of grams(candidate)) if (base.has(gram)) hit += 1 return hit / base.size } looksRepeated(candidate) { const recent = loadWechatMessages({ dataDir: this.dataDir, limit: 200 }) .filter((record) => record.self && record.text) return recent.some((record) => this.overlapWith(record.text, candidate) >= this.repeatThreshold) } }关键参数与边界:
repeatThreshold默认 0.82,如果你的系统提示词让模型大量输出固定模板(固定开头、固定落款),把它降到 0.7,否则去重会形同虚设。- 8-gram 对短句(少于 16 字符)无区分度,
overlapWith返回 0 直接放行,短回复主要靠vary的前缀变化兜底。 - 敏感词表建议只放合规底线词,不放业务词,误伤率会很难看。
把授权闸门抽成独立 gate
设计思路:现在的判定逻辑散在 sendMessage.js 里,和发送动作耦合。抽成纯函数后,哨兵可以在运行期收口闸门口径(比如危险级别时只放行白名单私聊),而不用动消息处理主流程。
核心代码:
// src/guard/gate.js import { getWechatRuntimeConfig } from '../config/env.js' export function shouldReply({ roomName, alias, name, content, botName }) { const { roomWhiteList, aliasWhiteList, autoReplyPrefix } = getWechatRuntimeConfig() if (!roomName) { const inList = aliasWhiteList.includes(alias) || aliasWhiteList.includes(name) const prefixOk = !autoReplyPrefix || content.trimStart().startsWith(autoReplyPrefix) return inList && prefixOk } const stripped = content.replace(botName, '').trimStart() const mentioned = content.includes(botName) const inList = roomWhiteList.includes(roomName) const prefixOk = !autoReplyPrefix || stripped.startsWith(autoReplyPrefix) return mentioned && inList && prefixOk }关键参数与边界:
AUTO_REPLY_PREFIX留空等价于"名单内全量回复",首周务必留空观察基线,第二周再设成AI之类的短前缀收口。botName若含正则特殊字符(如.、(),includes不受影响,但后续若换成正则匹配记得先转义。- 这个函数只回答"回不回","怎么回、何时回"分别归 Shaper 和 Pacer 管,三者在
defaultMessage里串联,互不感知。
部署与调参:先闸门、后步幅、最后哨兵,每步隔一个整天
环境变量按 env.js 的既有模式扩展,原有项参考仓库.env.example,新增护栏项如下:
# .env:护栏新增项(原有 BOT_NAME / 白名单 / AUTO_REPLY_PREFIX 照旧) WECHAT_DATA_DIR='.data/wechat' WECHAT_STORE_MESSAGES='true' GUARD_MAX_OUT_PER_HOUR=40 GUARD_MIN_GAP_MS=600 GUARD_QUIET_DROP=0.08 GUARD_REPEAT_THRESHOLD=0.82| 参数 | 推荐值 | 说明 |
|---|---|---|
| GUARD_MAX_OUT_PER_HOUR | 40 | 单小时发送上限,账龄不足 3 年先减半 |
| GUARD_MIN_GAP_MS | 600 | 两条消息最小间隔,防突发连发 |
| GUARD_QUIET_DROP | 0.08 | 静默时段放弃回复概率,主号设 0 |
| GUARD_REPEAT_THRESHOLD | 0.82 | 8-gram 重叠率弃发阈值,模板化输出多则降到 0.7 |
| AUTO_REPLY_PREFIX | AI | 命中前缀才回复,留空为名单内全量 |
| WECHAT_STORE_MESSAGES | true | 必须开启,Shaper 与 Guardian 依赖该记录 |
分步部署清单:
- 克隆仓库并配置
.env:git clone https://gitcode.com/GitHub_Trending/we/wechat-bot,填入BOT_NAME与两个白名单,AUTO_REPLY_PREFIX留空。 npm run start(或wb start -s <服务类型>)扫码登录,首日只跑现有白名单闸门,不动任何新模块,确认messages.jsonl正常增长。- 接入
SendPacer,观察一天,检查日志间隔呈分布而非定值。 - 接入
ReplyShaper,观察一天,确认弃发只发生在高重复场景。 - 启动
Guardian,用脚本短时间刷消息验证 L1 能正常触发后再交回日常。 - 全程用一个小号验证,确认无误后再切主号,且主号每周只扩一次白名单。
运行期保障:护栏要能自我降速,否则第一个风险信号会漏掉
哨兵的职责只有一个:在风险信号出现时自动收紧步幅与闸门,而不是等人发现。指标全部取自messages.jsonl的self记录,不引入新存储。
监控指标:
- 小时发送量及其占
GUARD_MAX_OUT_PER_HOUR的比例。 - 连续突发次数:相邻两条
self记录间隔小于GUARD_MIN_GAP_MS的频次。 - 高重复回复占比:Shaper 弃发次数除以总回复次数。
- AI 服务连续失败次数:连续 5 次失败视为通道异常,停止排队发送。
- 会话存活天数:
login到logout的间隔,bot.js 两个事件里各记一笔即可。
| 级别 | 触发条件 | 自动动作 |
|---|---|---|
| L1 | 小时量 ≥ 0.7 倍上限,或连续突发 ≥ 3 次 | 步幅系数 ×1.5 |
| L2 | 小时量打满上限,或高重复占比 > 0.3 | 步幅系数 ×3,群聊回复关闭 |
| L3 | L2 持续 30 分钟,或 AI 连续失败 5 次 | 冻结发送 2 小时,仅落盘记录 |
| L4 | 收到 logout 事件,或人工触发 | 停止进程,保留.data会话待重扫 |
核心模块代码:
// src/guard/guardian.js import fs from 'fs' import { getMessageStorePath } from '../platforms/wechat/messageStore.js' const LEVELS = ['L0', 'L1', 'L2', 'L3', 'L4'] export class Guardian { constructor({ maxPerHour, pacer, dataDir = '.data/wechat' } = {}) { this.maxPerHour = maxPerHour this.pacer = pacer this.dataDir = dataDir this.level = 'L0' } selfOutbound(sinceMs) { const file = getMessageStorePath(this.dataDir) if (!fs.existsSync(file)) return 0 return fs .readFileSync(file, 'utf8') .split('\n') .filter(Boolean) .filter((line) => { try { const record = JSON.parse(line) return record.self && Date.parse(record.timestamp) >= sinceMs } catch { return false } }).length } tick() { const count = this.selfOutbound(Date.now() - 3600_000) const ratio = count / this.maxPerHour const next = ratio >= 1 ? 'L2' : ratio >= 0.7 ? 'L1' : 'L0' if (LEVELS.indexOf(next) > LEVELS.indexOf(this.level)) this.apply(next) } apply(level) { this.level = level const factor = { L0: 1, L1: 1.5, L2: 3, L3: Infinity, L4: Infinity }[level] if (this.pacer) this.pacer.gapFactor = factor console.log(`[guardian] level=${level} gapFactor=${factor}`) } } export function startGuardian(options) { const guardian = new Guardian(options) return setInterval(() => guardian.tick(), 5 * 60 * 1000) }三点说明:
tick只升不降,降速靠gapFactor在下一个自然低峰期人工复位或加时间窗回退,避免"降速—恢复—再降速"的抖动被风控当成新的规律性特征。- L2 的"群聊关闭"不是在这里实现的,而是在
shouldReply里读guardian.level,L2及以上直接让群聊分支返回 false,改动一行即可。 L3/L4用Infinity表示完全停发:gapFactor为无穷时nextSlot的等待时间溢出,调用方应把Infinity特判为"不发送",落地时补一个if (!Number.isFinite(this.pacer.gapFactor)) return null即可。
带走清单
.env中BOT_NAME带@前缀,两个白名单非空,AUTO_REPLY_PREFIX首周留空- 确认
WECHAT_STORE_MESSAGES=true,且.data/wechat/messages.jsonl里有self: true记录 - 接入
SendPacer后检查日志间隔呈宽分布,出现连续固定值立即回滚 - 为
ReplyShaper配好敏感词表,并把模板化输出多的场景阈值降到 0.7 - 启动
Guardian后用脚本刷量验证 L1 在 0.7 倍上限处触发 - 账龄不足 3 年的账号把
GUARD_MAX_OUT_PER_HOUR减半,小号跑满一周再切主号
护栏买不来"不封",它只做两件事:让你在被判之前先看见信号,并把行为与内容两类线索压到接近真人基线。
【免费下载链接】wechat-bot🤖 Multi-platform IM AI Agent for Telegram, WhatsApp, Lark, and WeChat. Connects ChatGPT / Claude / Kimi / DeepSeek / Ollama / Pi for auto-replies, community analysis, contact management, and inactive-friend detection.项目地址: https://gitcode.com/GitHub_Trending/we/wechat-bot
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考