wechat_reply_policy.py
8.37 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
#!/usr/bin/env python3
"""环节4 · 策略判定 —— 确定性安全闸（自动发 / 进审核 / 跳过）
为什么是代码不是让 LLM 判：环节3 已验证「SOUL.md 里写的软约束会漏」(约饭案例模型仍
主动接话)。真正的自动/审核安全门必须是确定性、可审计、LLM 说服不动的硬规则。
输入：watcher 事件 + Hermes 生成的草稿 + (可选)模型置信/should_reply 信号
输出：{action: auto|review|skip, reason}
skip —— 压根不该回（群里没@我、公众号、系统消息、自己发的、模型说别回）
auto —— 白名单 + 文本 + 无敏感词 + 总开关开 → 直接发（环节5）
review —— 其余一律进审核队列 pending.jsonl，人工在 wechat viewer 里点发送
设计=default-deny：不在白名单、命中敏感词、低置信、拿不准 → 全部 review。
先保守起步（auto_send_enabled 默认 False = 全审核），跑顺了再对白名单放开自动。
配置 ~/.hermes/wechat_reply/policy_config.json（首次自动生成默认值）。
"""
import json
import os
import time
WORK_DIR = os.environ.get(
"WECHAT_REPLY_DIR",
"/data/data/com.termux/files/home/.hermes/wechat_reply")
CONFIG_FILE = f"{WORK_DIR}/policy_config.json"
PENDING_FILE = f"{WORK_DIR}/pending.jsonl" # 审核队列（环节5 viewer 读）
DECISION_LOG = f"{WORK_DIR}/decisions.jsonl"
DEFAULT_CONFIG = {
"auto_send_enabled": False, # 总开关：False = 一切都进审核（保守起步）
"whitelist_contacts": [], # 允许自动发的 wxid（私聊）
"whitelist_groups": [], # 允许 @我 时自动发的 chatroom id
"min_confidence": 0.75, # 低于此→审核（模型给了置信才生效）
# 命中即强制 review（即使白名单也拉回人工）——钱/承诺/约时间/隐私
"block_keywords": [
"钱", "转账", "红包", "借", "还款", "付款", "打款", "汇款", "定金", "报价",
"价格", "多少钱", "合同", "签", "承诺", "保证", "担保",
"几点", "明天", "后天", "周末", "约", "见面", "面基", "碰头", "地址在",
"身份证", "银行卡", "验证码", "密码", "手机号", "住址",
],
}
# 非文本消息类型（草稿通常无意义）
NON_TEXT_TYPES = {3, 34, 43, 47, 42, 48, 49, 62, 10000}
SYSTEM_MARKERS = ("撤回了一条消息", "拍了拍", "拍一拍", "添加到通讯录", "以上是打招呼的内容")
def load_config():
try:
with open(CONFIG_FILE) as f:
cfg = json.load(f)
return {**DEFAULT_CONFIG, **cfg}
except (OSError, ValueError):
os.makedirs(WORK_DIR, exist_ok=True)
with open(CONFIG_FILE, "w") as f:
json.dump(DEFAULT_CONFIG, f, ensure_ascii=False, indent=2)
return dict(DEFAULT_CONFIG)
def _hit_keyword(text, keywords):
for k in keywords:
if k and k in text:
return k
return None
def decide(event, draft, confidence=None, should_reply=True, self_wxid="", cfg=None):
"""核心判定。返回 {action, reason, hit?}。纯函数、无副作用（写队列在 route()）。"""
cfg = cfg or load_config()
kind = event.get("kind", "")
sender = event.get("sender") or event.get("talker") or ""
talker = event.get("talker", "")
body = event.get("body", "") or ""
mtype = event.get("type", 0)
# ── skip 区：压根不回 ──
if sender and self_wxid and sender == self_wxid:
return {"action": "skip", "reason": "self:自己发的"}
if kind == "group_other":
return {"action": "skip", "reason": "group:群里没@我也没引用我"}
if str(talker).startswith("gh_") or str(sender).startswith("gh_"):
return {"action": "skip", "reason": "official:公众号/服务号推送"}
if any(m in body for m in SYSTEM_MARKERS):
return {"action": "skip", "reason": "system:系统消息(撤回/拍一拍/加好友)"}
if not should_reply:
return {"action": "skip", "reason": "model:模型判定无需回复"}
if not (draft or "").strip():
if mtype in NON_TEXT_TYPES:
return {"action": "skip", "reason": f"nontext:非文本(type={mtype})且无草稿"}
return {"action": "skip", "reason": "empty:草稿为空"}
# ── 硬安全override：命中敏感词一律 review（白名单也拉回）──
hit = _hit_keyword(body + "\n" + (draft or ""), cfg["block_keywords"])
if hit:
return {"action": "review", "reason": f"sensitive:命中敏感词「{hit}」", "hit": hit}
# ── 低置信 → review ──
if confidence is not None and confidence < cfg["min_confidence"]:
return {"action": "review", "reason": f"lowconf:置信{confidence}<{cfg['min_confidence']}"}
# ── 白名单 + 总开关 → auto ──
is_group = event.get("isGroup") or str(talker).endswith("@chatroom")
whitelisted = (
(not is_group and sender in cfg["whitelist_contacts"]) or
(is_group and talker in cfg["whitelist_groups"])
)
if whitelisted and cfg["auto_send_enabled"]:
return {"action": "auto", "reason": "whitelist:白名单+总开关开→直接发"}
# ── default-deny：其余全部人工审核 ──
if not cfg["auto_send_enabled"]:
return {"action": "review", "reason": "conservative:总开关关(全审核)"}
return {"action": "review", "reason": "notwhitelist:不在白名单→人工"}
def route(event, draft, decision, meta=None):
"""按判定落地：review→写审核队列 pending.jsonl；一切→记 decisions.jsonl 审计。"""
os.makedirs(WORK_DIR, exist_ok=True)
rec = {
"ts": int(time.time()),
"msgId": event.get("msgId"),
"talker": event.get("talker"),
"sender": event.get("sender"),
"kind": event.get("kind"),
"action": decision["action"],
"reason": decision["reason"],
"draft": draft,
**(meta or {}),
}
with open(DECISION_LOG, "a") as f:
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
if decision["action"] == "review":
item = {**rec, "status": "pending"}
with open(PENDING_FILE, "a") as f:
f.write(json.dumps(item, ensure_ascii=False) + "\n")
return rec
# ── 自检 demo（lazy 规则：留一个可跑的检查）──
def _demo():
cfg = dict(DEFAULT_CONFIG)
cfg["whitelist_contacts"] = ["wxid_friend"]
cfg["auto_send_enabled"] = True
cases = [
("陌生猎头(非白名单)", {"kind": "private", "talker": "wxid_x", "sender": "wxid_x",
"body": "有急招岗位吗", "type": 1}, "暂不考虑，谢谢", None, "review"),
("白名单好友闲聊", {"kind": "private", "talker": "wxid_friend", "sender": "wxid_friend",
"body": "在干嘛", "type": 1}, "摸鱼呢哈哈", None, "auto"),
("白名单好友约饭(敏感词)", {"kind": "private", "talker": "wxid_friend", "sender": "wxid_friend",
"body": "明天中午一起吃饭？", "type": 1}, "可以啊几点", None, "review"),
("白名单好友谈钱", {"kind": "private", "talker": "wxid_friend", "sender": "wxid_friend",
"body": "借我5000周转下", "type": 1}, "行", None, "review"),
("群里没@我", {"kind": "group_other", "talker": "x@chatroom", "sender": "wxid_a",
"isGroup": True, "body": "哈哈", "type": 1}, "草稿", None, "skip"),
("公众号推送", {"kind": "private", "talker": "gh_abc", "sender": "gh_abc",
"body": "<msg>文章</msg>", "type": 49}, "", None, "skip"),
("撤回系统消息", {"kind": "group_at_me", "talker": "x@chatroom", "sender": "wxid_a",
"body": "\"张三\" 撤回了一条消息", "type": 10000}, "", None, "skip"),
("低置信", {"kind": "private", "talker": "wxid_friend", "sender": "wxid_friend",
"body": "你觉得呢", "type": 1}, "嗯", 0.4, "review"),
]
ok = True
for name, ev, draft, conf, expect in cases:
d = decide(ev, draft, confidence=conf, cfg=cfg)
flag = "OK " if d["action"] == expect else "FAIL"
if d["action"] != expect:
ok = False
print(f"[{flag}] {name}: 期望={expect} 实得={d['action']} ({d['reason']})")
print("=== ALL PASS ===" if ok else "=== 有 FAIL ===")
return ok
if __name__ == "__main__":
import sys
if "--demo" in sys.argv:
sys.exit(0 if _demo() else 1)
print("用法: wechat_reply_policy.py --demo (或作为库 import decide/route)")