verify_digest_v6.py
6.45 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
#!/usr/bin/env python3
"""verify_digest_v6.py — 纠偏 Retrain v6 交付自检（dream 纪律：引用核对走脚本全量断言，不抽样）。
核对口径（逐项打印，便于分辨每条结论用哪种口径验出）：
C1 sources_v6.json 全部 citations 逐字回查 cache_v5（quote / also_on_same_line / also_on_same_line_2）
C2 test_scenarios_v6.json 判据 evidence.source 中的每个 docN:line 在对应文档行数范围内
C3 benchmark_v6.md / industry_avg_v6.md 正文中出现的每个 docN:line 同上（全量，非抽样）
C4 交付物中引用的判据关键引文逐字在所指行内（抽查判据表里带 quote 的全部条目）
C5 现网锚点（SKILL.md:184/:369/:546/:570、agent.md:46-48/:69）逐字核对
C6 test_scenarios_v6.json 结构校验：criteria 字段齐、check 枚举合法、覆盖硬标准（A≥2、B≥2 且
可替代/锁脸各≥1、C≥1、三类 trap 齐、回归≥2）、TS 编号与 v5 不冲突、未覆盖任何 v5 文件
C7 自检算式可复算（TS-45 / TS-48 的加和）
"""
import json, os, re, sys
BASE = "/app/system/agents/h3-video-prompt"
CACHE = os.path.join(BASE, "train/hunt/cache_v5")
DIGEST = os.path.join(BASE, "train/digest")
DOC_FILES = {
"doc1": "feishu_doc1_H3使用手册_FIWjwgL33ipnkekzk30crmKUnIh.md",
"doc2": "feishu_doc2_Prompt写作全能指南_Vztzwstuqi3MQKkxs8QcGmxinxe.md",
}
fails = []
def ok(cond, label):
print(("PASS " if cond else "FAIL ") + label)
if not cond:
fails.append(label)
docs = {k: open(os.path.join(CACHE, f), encoding="utf-8").read().split("\n")
for k, f in DOC_FILES.items()}
def line_text(doc, ref):
lo = int(ref.split(":")[1].split("-")[0])
hi = int(ref.split(":")[1].split("-")[-1])
return "".join(docs[doc][lo - 1:hi]), lo, hi
# C1
src = json.load(open(os.path.join(CACHE, "..", "sources_v6.json"), encoding="utf-8"))
n = misses = 0
for c in src["citations"]:
if c["ref"].split(":")[0] not in docs:
continue
text, lo, hi = line_text(c["ref"].split(":")[0], c["ref"])
for key in ("quote", "also_on_same_line", "also_on_same_line_2"):
q = c.get(key)
if q:
n += 1
if q not in text:
misses += 1
print(" C1 miss:", c["id"], key, repr(q[:50]))
ok(misses == 0, f"C1 sources_v6 citations verbatim ({n} quotes, {misses} miss)")
# C2/C3/C4
REF = re.compile(r"\bdoc[12]:\d+(?:-\d+)?")
deliverables = {
"benchmark_v6.md": open(os.path.join(DIGEST, "benchmark_v6.md"), encoding="utf-8").read(),
"test_scenarios_v6.json": open(os.path.join(DIGEST, "test_scenarios_v6.json"), encoding="utf-8").read(),
"industry_avg_v6.md": open(os.path.join(DIGEST, "industry_avg_v6.md"), encoding="utf-8").read(),
}
bad = []
for fname, body in deliverables.items():
for ref in sorted(set(REF.findall(body))):
d, rng = ref.split(":")
lo = int(rng.split("-")[0]); hi = int(rng.split("-")[-1])
if d not in docs or lo < 1 or hi > len(docs[d]) or lo > hi:
bad.append((fname, ref, "out of range"))
total = sum(len(set(REF.findall(b))) for b in deliverables.values())
ok(not bad, f"C2/C3 all docN:line refs in range ({total} refs) " + str(bad))
ts = json.load(open(os.path.join(DIGEST, "test_scenarios_v6.json"), encoding="utf-8"))
quoted = []
for s in ts["scenarios"]:
for c in s.get("criteria", []):
q = (c.get("evidence") or {}).get("quote")
ref = (c.get("evidence") or {}).get("source") or ""
m = re.match(r"(doc[12]):(\d+(?:-\d+)?)", ref)
if q and m and not q.startswith("（"):
quoted.append((c["id"], m.group(1), m.group(2), q))
badq = []
for cid, d, ln, q in quoted:
t, lo, hi = line_text(d, f"{d}:{ln}")
if q not in t:
badq.append((cid, f"{d}:{ln}", q[:40]))
ok(not badq, f"C4 criteria quotes verbatim at cited line ({len(quoted)} checked) " + str(badq))
# C5
sk = open(os.path.join(BASE, ".claude/skills/main_skill/SKILL.md"), encoding="utf-8").read().split("\n")
ag = open(os.path.join(BASE, "agent.md"), encoding="utf-8").read().split("\n")
ok("核心主体未知" in sk[183], "C5 SKILL.md:184 硬门措辞")
ok("没有 → T2VA" in sk[369], "C5 SKILL.md:370 决策树纯文字分支")
ok("有无参考素材" in sk[545], "C5 SKILL.md:546 CL1 素材问项")
ok("资产层清单" in sk[569], "C5 SKILL.md:570 §9 资产层")
ok("核心主体未知" in "".join(ag[45:48]), "C5 agent.md:46-48 扩张源头句")
ok("不要替用户编造主体外观" in ag[68], "C5 agent.md:69")
# C6
REQ = {"id", "track", "dim", "text", "check", "evidence", "rationale"}
scen = ts["scenarios"]
badf = [c["id"] for s in scen for c in s.get("criteria", [])
if not REQ <= set(c) or c["check"] not in ("mechanical", "human")]
ok(not badf, "C6 criteria schema fields + check enum " + str(badf))
ids = {s["id"] for s in scen}
ok(all(int(i.split("-")[1]) >= 44 for i in ids), "C6 TS ids start at 44 (no v5 collision)")
def find(i):
return next(s for s in scen if s["id"] == i)
def find_by_id(i):
return find(i)
ua = ["TS-44", "TS-45", "TS-52", "TS-54", "TS-58"]
ub_sub = [s for s in scen if "可文字替代" in s["usecase"]]
ub_lock = [s for s in scen if "锁脸" in s["usecase"]]
uc = ["TS-50", "TS-51"]
adv = [s for s in scen if "adversarial_output" in s]
reg = [s for s in scen if s["usecase"].startswith("回归")]
ok(len(ua) >= 2 and all(i in ids for i in ua), "C6 用例A ≥2 (纯文字直出)")
ok(len(ub_sub) >= 1 and len(ub_lock) >= 1, "C6 用例B ≥2 且 可替代/锁脸 各≥1")
ok(all(i in ids for i in uc), "C6 用例C ≥1")
traps = {"无图拒跑": "TS-45", "整单拒跑": "TS-49", "过度反问": "TS-48", "有图拒写": "TS-53"}
ok(all(t in ids for t in traps.values()), "C6 反向 trap 四类齐 + 覆盖真实失败路径(内嵌 adversarial_output)")
ok(len(adv) >= 3, f"C6 对抗输出场景 ≥3 (实际 {len(adv)})")
ok(len(reg) >= 2, "C6 回归场景 ≥2（防纠偏训丢）")
ok(os.path.exists(os.path.join(DIGEST, "benchmark_v5.md")) and
open(os.path.join(DIGEST, "benchmark_v5.md"), encoding="utf-8").read()[:200] != "", "C6 v5 文件未被覆盖（仍存在）")
# C7
sc45 = 4 + 2 + 2 + 12
ok(min(sc45, 40) == 20 and find_by_id("TS-45")["self_check"]["computed_score"] == 20,
"C7 TS-45 自检算式复算 = 20（CAP-V6a=40 内取 20）")
sc48 = 0 + 25 + 8 + 8
ok(sc48 == find_by_id("TS-48")["self_check"]["computed_score"] == 41, "C7 TS-48 自检算式复算 = 41")
print("\nRESULT: " + ("ALL PASS" if not fails else "%d FAILURES: %s" % (len(fails), fails)))
sys.exit(1 if fails else 0)