simonkoson
d91c20a126
- ground-truth v0.5.0: 补完10期narrative_structure标注(MiMo跑批+制片人审定) - summarize.py: 修复无标注期次格式化崩溃 - 003_add_ai_labels.sql: episodes表加7列(6打标+1置信度)迁移文件 - episode.py: Model同步加7个Optional字段 - CLAUDE.md: 项目主控文件入库 - doco/deliverables: 20期终版文稿md入库 - doco/programs: 中间产物从git移除(本地保留), .gitignore统一管理 - note/寄存条: 两个子项目寄存条入库 - .gitignore: 补充doco中间产物/Excel锁文件/临时文件规则 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
193 lines
8.6 KiB
Python
193 lines
8.6 KiB
Python
"""
|
|
summarize.py - 汇总打标结果命中情况
|
|
用法: python summarize.py --model mimo-v2.5-pro
|
|
"""
|
|
|
|
import sys
|
|
sys.stdout.reconfigure(encoding='utf-8')
|
|
sys.stderr.reconfigure(encoding='utf-8')
|
|
|
|
import json
|
|
import glob
|
|
import argparse
|
|
from pathlib import Path
|
|
|
|
BASE_DIR = Path(__file__).parent.parent
|
|
EXPERIMENTS_DIR = BASE_DIR / "experiments"
|
|
GROUND_TRUTH = BASE_DIR / "benchmark-set" / "ground-truth.json"
|
|
|
|
|
|
def load_ground_truth_map():
|
|
"""从 ground-truth.json 源文件加载最新 GT（而非实验文件中的快照）。"""
|
|
try:
|
|
data = json.loads(GROUND_TRUTH.read_text(encoding="utf-8"))
|
|
return {ep["ep"]: ep for ep in data["episodes"]}
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def load_json(path):
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def latest_per_ep(files):
|
|
"""同 ep 有多个文件时取时间戳最新的。"""
|
|
latest = {}
|
|
for f in files:
|
|
name = f.name
|
|
# 文件名格式: 20260611_154037_model_field_ep003.json
|
|
parts = name.replace(".json", "").split("_")
|
|
if len(parts) >= 4:
|
|
ts = parts[0] + parts[1] # yyyymmddHHMMSS
|
|
ep = int(parts[-1].replace("ep", ""))
|
|
key = ep
|
|
if key not in latest or ts > latest[key][1]:
|
|
latest[key] = (f, ts)
|
|
return {ep: info[0] for ep, info in latest.items()}
|
|
|
|
|
|
def run(model, field="narrative"):
|
|
# 文件匹配：新格式有 field 标记，老格式（无 field）向后兼容
|
|
pattern_new = str(EXPERIMENTS_DIR / f"*_{model}_{field}_*.json")
|
|
files_new = sorted(Path(p) for p in glob.glob(pattern_new))
|
|
# 如果是 narrative 且新格式文件为空，尝试匹配老格式（无 field 标记）
|
|
if field == "narrative" and not files_new:
|
|
pattern_old = str(EXPERIMENTS_DIR / f"*_{model}_ep*.json")
|
|
files_old = [f for f in sorted(Path(p) for p in glob.glob(pattern_old))
|
|
if len(f.name.replace(".json", "").split("_")) == 3] # 老格式: ts_model_epNN.json
|
|
files = files_old
|
|
else:
|
|
files = files_new
|
|
|
|
if not files:
|
|
print(f"未找到 {pattern_new}")
|
|
return
|
|
|
|
ep_files = latest_per_ep(files)
|
|
gt_map = load_ground_truth_map() # 从源文件读最新 GT
|
|
rows = []
|
|
parse_fail = 0
|
|
|
|
for ep in sorted(ep_files.keys()):
|
|
data = load_json(ep_files[ep])
|
|
if data is None:
|
|
parse_fail += 1
|
|
rows.append({"ep": ep, "title": "?", "gt": "?", "pred": "解析失败", "hit": False, "conf": "?"})
|
|
continue
|
|
|
|
# 优先用源文件 GT，回退到实验文件中的快照
|
|
gt = gt_map.get(ep, data.get("ground_truth", {}))
|
|
result = data.get("result")
|
|
|
|
title = gt.get("title", "?")
|
|
|
|
if field == "opening_hook":
|
|
gt_val = gt.get("opening_hook")
|
|
pred_val = result.get("opening_hook") if result else None
|
|
conf = result.get("confidence", "?") if result else "?"
|
|
if gt_val is None:
|
|
rows.append({"ep": ep, "title": title, "gt": "(无标注)", "pred": pred_val or "解析失败", "hit": None, "conf": conf})
|
|
else:
|
|
hit = pred_val == gt_val if pred_val is not None else False
|
|
rows.append({"ep": ep, "title": title, "gt": gt_val, "pred": pred_val or "解析失败", "hit": hit, "conf": conf})
|
|
elif field == "narrative":
|
|
gt_val = gt.get("narrative_structure")
|
|
pred_val = result.get("narrative_structure") if result else None
|
|
conf = result.get("confidence", "?") if result else "?"
|
|
if gt_val is None:
|
|
rows.append({"ep": ep, "title": title, "gt": "(无标注)", "pred": pred_val or "解析失败", "hit": None, "conf": conf})
|
|
else:
|
|
hit = pred_val == gt_val if pred_val is not None else False
|
|
rows.append({"ep": ep, "title": title, "gt": gt_val, "pred": pred_val or "解析失败", "hit": hit, "conf": conf})
|
|
elif field == "classification":
|
|
conf = result.get("confidence", "?") if result else "?"
|
|
sub = {}
|
|
for key in ("program_format", "equipment_domain", "scene_tags", "tech_tags"):
|
|
gt_val = gt.get(key)
|
|
pred_val = result.get(key) if result else None
|
|
if gt_val is None or pred_val is None:
|
|
sub[key] = None
|
|
elif isinstance(gt_val, list):
|
|
sub[key] = set(gt_val) == set(pred_val) if isinstance(pred_val, list) else False
|
|
else:
|
|
sub[key] = pred_val == gt_val
|
|
all_hit = all(v is True for v in sub.values() if v is not None)
|
|
rows.append({"ep": ep, "title": title, "sub": sub, "all_hit": all_hit, "conf": conf})
|
|
|
|
if field == "opening_hook":
|
|
for r in rows:
|
|
if r["hit"] is None:
|
|
mark = "-"
|
|
else:
|
|
mark = "✓" if r["hit"] else "✗"
|
|
conf_str = f'置信度:{r["conf"]}' if r["conf"] != "?" else ""
|
|
print(f' ep{r["ep"]:03d} {r["title"]:<12} | 标准:{r["gt"]:<6} | {model}:{r["pred"]:<6} | {mark} {conf_str}')
|
|
|
|
scored = [r for r in rows if r["hit"] is not None]
|
|
hits = sum(1 for r in scored if r["hit"])
|
|
total = len(scored)
|
|
unlabeled = sum(1 for r in rows if r["hit"] is None)
|
|
print(f"\n ===== {model} 命中情况 =====")
|
|
print(f" opening_hook 命中: {hits}/{total} = {hits*100//total if total else 0}%")
|
|
print(f" 无标注(跳过): {unlabeled} 期")
|
|
print(f" 解析失败: {parse_fail} 期")
|
|
elif field == "narrative":
|
|
# 打印每行
|
|
for r in rows:
|
|
if r["hit"] is None:
|
|
mark = "-"
|
|
else:
|
|
mark = "✓" if r["hit"] else "✗"
|
|
conf_str = f'置信度:{r["conf"]}' if r["conf"] != "?" else ""
|
|
print(f' ep{r["ep"]:03d} {r["title"]:<12} | 标准:{r["gt"]:<6} | {model}:{r["pred"]:<6} | {mark} {conf_str}')
|
|
|
|
# 汇总（跳过无标注）
|
|
scored = [r for r in rows if r["hit"] is not None]
|
|
hits = sum(1 for r in scored if r["hit"])
|
|
total = len(scored)
|
|
unlabeled = sum(1 for r in rows if r["hit"] is None)
|
|
hi_conf = [r for r in scored if r["conf"] == "高"]
|
|
mid_low = [r for r in scored if r["conf"] in ("中", "低")]
|
|
hi_hit = sum(1 for r in hi_conf if r["hit"])
|
|
ml_hit = sum(1 for r in mid_low if r["hit"])
|
|
|
|
print(f"\n ===== {model} 命中情况 =====")
|
|
print(f" narrative_structure 命中: {hits}/{total} = {hits*100//total if total else 0}%")
|
|
print(f" 自评\"高\"置信的命中率: {hi_hit}/{len(hi_conf)}")
|
|
print(f" 自评\"中/低\"置信的命中率: {ml_hit}/{len(mid_low)}")
|
|
print(f" 无标注(跳过): {unlabeled} 期")
|
|
print(f" 解析失败: {parse_fail} 期")
|
|
elif field == "classification":
|
|
field_names = ["program_format", "equipment_domain", "scene_tags", "tech_tags"]
|
|
short = {"program_format": "题材", "equipment_domain": "装备域", "scene_tags": "场景", "tech_tags": "技术"}
|
|
for r in rows:
|
|
marks = ""
|
|
for fn in field_names:
|
|
v = r["sub"].get(fn)
|
|
marks += "✓" if v is True else ("✗" if v is False else "-")
|
|
all_mark = "✓" if r["all_hit"] else "✗"
|
|
print(f' ep{r["ep"]:03d} {r["title"]:<12} | {"/".join(short[f] for f in field_names)}={marks} | 全对:{all_mark} 置信度:{r["conf"]}')
|
|
|
|
total = len(rows)
|
|
for fn in field_names:
|
|
scored = [r for r in rows if r["sub"].get(fn) is not None]
|
|
hits = sum(1 for r in scored if r["sub"][fn])
|
|
pct = hits * 100 // len(scored) if scored else 0
|
|
print(f" {short[fn]}命中: {hits}/{len(scored)} = {pct}%")
|
|
all_scored = [r for r in rows if any(v is not None for v in r["sub"].values())]
|
|
all_hits = sum(1 for r in all_scored if r["all_hit"])
|
|
print(f" 4字段全对: {all_hits}/{len(all_scored)} = {all_hits*100//len(all_scored) if all_scored else 0}%")
|
|
print(f" 解析失败: {parse_fail} 期")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--model", required=True, help="模型键名，如 mimo-v2.5-pro / deepseek-v4-pro")
|
|
parser.add_argument("--field", default="narrative", choices=["narrative", "classification", "opening_hook"],
|
|
help="打标字段: narrative(叙事结构) / classification(4分类) / opening_hook(开篇钩子)")
|
|
args = parser.parse_args()
|
|
run(args.model, args.field)
|