之前聊过 Trie 上下文窗口的关键词监测方案但那个方案对小团队单账号来说偏重——Trie 要内存、上下文判断要规则、多线程要 CPU。这篇给一个轻量方案不用 Trie简单前缀匹配 LRU 消息去重 单线程单轮询资源开销比 Trie 方案低 80%。一、简单前缀匹配一个字一个字扫不用建树直接遍历敏感词列表对每条群消息做 in 判断——关键词 in 消息内容。对 95% 的场景够用明确的敏感词广告、赌博、政治。上下文判断这一步在轻量方案里省略靠运营在后台做误报清理迭代词库。词库维护在本地 JSON 文件里支持热加载每隔 30 秒读一次文件有变更就重新加载。二、LRU 消息去重避免回调重推的重复告警平台回调会因为网络原因重复推送同一条消息——如果不做去重同一条敏感消息会触发好几次告警。用 LRU最近最少使用缓存存最近 1000 个 msg_id进来先查缓存、命中就跳过没命中才处理并存缓存。LRU 缓存用 OrderedDict 实现1000 条内存不到 1MB单线程无锁开销极低。三、单线程单轮询每秒 1 次够用小团队单账号群消息量一般在每秒 0-5 条单线程每秒轮询 1 次足够覆盖延迟可接受最多 1 秒。轮询拿到新消息列表后逐条处理处理完进入下一轮。单线程无锁、异步写告警日志整个进程 CPU 占用稳定在 5% 以内跑在低配 VPS 或树莓派上都没问题。轻量方案对照组件做什么资源开销适用场景前缀匹配词库 in 消息O(W×N)词库 50 个词LRU 去重msg_id 查缓存内存 1MB避免重复告警单线程轮询1 秒 1 次CPU 5%单账号小群量轻量关键词监测实现import time, threading, json, os from collections import OrderedDict class LightweightMonitor: LRU_SIZE 1000 # 去重缓存容量 POLL_INTERVAL 1 # 轮询间隔 1 秒 HOT_RELOAD_SEC 30 # 词库热加载间隔 def __init__(self, api, store, word_file): self.api api self.store store self.word_file word_file self.words self._load_words() self.lru OrderedDict() self.last_msg_id store.last_msg_id() threading.Thread(targetself.run, daemonTrue).start() threading.Thread(targetself.hot_reload, daemonTrue).start() def _load_words(self): if not os.path.exists(self.word_file): return [] with open(self.word_file, encodingutf-8) as f: data json.load(f) return data.get(sensitive, []) def hot_reload(self): while True: time.sleep(self.HOT_RELOAD_SEC) new_words self._load_words() if new_words ! self.words: self.words new_words self.store.log(words_reloaded, len(self.words)) def run(self): while True: events self.api.get(/group/message/poll, params{ since: self.last_msg_id}).json()[items] for ev in events: self.last_msg_id ev[msg_id] # 1. LRU 去重 if ev[msg_id] in self.lru: continue self._lru_add(ev[msg_id]) # 2. 前缀匹配 hit self._scan(ev[content]) if hit: self._alert(ev, hit) self.api.ack(self.last_msg_id) time.sleep(self.POLL_INTERVAL) def _lru_add(self, msg_id): if msg_id in self.lru: self.lru.move_to_end(msg_id) self.lru[msg_id] True if len(self.lru) self.LRU_SIZE: self.lru.popitem(lastFalse) def _scan(self, content): for word in self.words: if word in content: return word return None def _alert(self, ev, hit): # 异步告警写日志 推给管理员 threading.Thread(targetself._do_alert, args(ev, hit), daemonTrue).start() def _do_alert(self, ev, hit): self.store.log_scan(ev, hit) owner self.store.group_owner(ev[group_id]) self.api.post(/send, { receiver: owner, content: (f群 {ev[group_id]} 命中敏感词「{hit}」 f发送人 {ev[from]} f内容 {ev[content][:100]})[:500]}).json()落地建议群关键词监测不一定要上 Trie 上下文窗口那套重型方案——小团队单账号用简单前缀匹配 LRU 去重 单线程轮询就够了整个进程 CPU 占用 5% 以内、内存不到 1MB。LRU 去重避免回调重推的重复告警词库热加载支持运营随时更新不用重部署。重词库≥50 个或多账号大群量再升级到 Trie 方案。 
长河君 · 2026/10/30 13:20:16 · 阅读 1.2k 不用天天盯着手机！个人微信二次开发实现微信群关键词自动监测
之前聊过 Trie 上下文窗口的关键词监测方案但那个方案对小团队单账号来说偏重——Trie 要内存、上下文判断要规则、多线程要 CPU。这篇给一个轻量方案不用 Trie简单前缀匹配 LRU 消息去重 单线程单轮询资源开销比 Trie 方案低 80%。一、简单前缀匹配一个字一个字扫不用建树直接遍历敏感词列表对每条群消息做 in 判断——关键词 in 消息内容。对 95% 的场景够用明确的敏感词广告、赌博、政治。上下文判断这一步在轻量方案里省略靠运营在后台做误报清理迭代词库。词库维护在本地 JSON 文件里支持热加载每隔 30 秒读一次文件有变更就重新加载。二、LRU 消息去重避免回调重推的重复告警平台回调会因为网络原因重复推送同一条消息——如果不做去重同一条敏感消息会触发好几次告警。用 LRU最近最少使用缓存存最近 1000 个 msg_id进来先查缓存、命中就跳过没命中才处理并存缓存。LRU 缓存用 OrderedDict 实现1000 条内存不到 1MB单线程无锁开销极低。三、单线程单轮询每秒 1 次够用小团队单账号群消息量一般在每秒 0-5 条单线程每秒轮询 1 次足够覆盖延迟可接受最多 1 秒。轮询拿到新消息列表后逐条处理处理完进入下一轮。单线程无锁、异步写告警日志整个进程 CPU 占用稳定在 5% 以内跑在低配 VPS 或树莓派上都没问题。轻量方案对照组件做什么资源开销适用场景前缀匹配词库 in 消息O(W×N)词库 50 个词LRU 去重msg_id 查缓存内存 1MB避免重复告警单线程轮询1 秒 1 次CPU 5%单账号小群量轻量关键词监测实现import time, threading, json, os from collections import OrderedDict class LightweightMonitor: LRU_SIZE 1000 # 去重缓存容量 POLL_INTERVAL 1 # 轮询间隔 1 秒 HOT_RELOAD_SEC 30 # 词库热加载间隔 def __init__(self, api, store, word_file): self.api api self.store store self.word_file word_file self.words self._load_words() self.lru OrderedDict() self.last_msg_id store.last_msg_id() threading.Thread(targetself.run, daemonTrue).start() threading.Thread(targetself.hot_reload, daemonTrue).start() def _load_words(self): if not os.path.exists(self.word_file): return [] with open(self.word_file, encodingutf-8) as f: data json.load(f) return data.get(sensitive, []) def hot_reload(self): while True: time.sleep(self.HOT_RELOAD_SEC) new_words self._load_words() if new_words ! self.words: self.words new_words self.store.log(words_reloaded, len(self.words)) def run(self): while True: events self.api.get(/group/message/poll, params{ since: self.last_msg_id}).json()[items] for ev in events: self.last_msg_id ev[msg_id] # 1. LRU 去重 if ev[msg_id] in self.lru: continue self._lru_add(ev[msg_id]) # 2. 前缀匹配 hit self._scan(ev[content]) if hit: self._alert(ev, hit) self.api.ack(self.last_msg_id) time.sleep(self.POLL_INTERVAL) def _lru_add(self, msg_id): if msg_id in self.lru: self.lru.move_to_end(msg_id) self.lru[msg_id] True if len(self.lru) self.LRU_SIZE: self.lru.popitem(lastFalse) def _scan(self, content): for word in self.words: if word in content: return word return None def _alert(self, ev, hit): # 异步告警写日志 推给管理员 threading.Thread(targetself._do_alert, args(ev, hit), daemonTrue).start() def _do_alert(self, ev, hit): self.store.log_scan(ev, hit) owner self.store.group_owner(ev[group_id]) self.api.post(/send, { receiver: owner, content: (f群 {ev[group_id]} 命中敏感词「{hit}」 f发送人 {ev[from]} f内容 {ev[content][:100]})[:500]}).json()落地建议群关键词监测不一定要上 Trie 上下文窗口那套重型方案——小团队单账号用简单前缀匹配 LRU 去重 单线程轮询就够了整个进程 CPU 占用 5% 以内、内存不到 1MB。LRU 去重避免回调重推的重复告警词库热加载支持运营随时更新不用重部署。重词库≥50 个或多账号大群量再升级到 Trie 方案。