#!/usr/bin/env python3 """Run Jev (and optional current-model) on the frozen T0 samples. Offline. Pins `jev-1.13.0`. Reads TYPESAFE_API_KEY from the environment only. Source B rows are aggregated and never written as raw text. """ from __future__ import annotations import argparse import json import os import sys import time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any, Mapping, Sequence ROOT = Path(__file__).resolve().parents[2] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from scripts.research.jev_intent_current import ( # noqa: E402 call_current_retry, current_model_id, stratified_sample, ) from scripts.research.jev_intent_questions import ( # noqa: E402 JEV_MODEL, build_state, enforce_combo, parse_jev_answers, sdk_questions, ) SAMPLES_DIR = ROOT / "scripts" / "research" / "jev_intent_samples" CACHE_DIR = Path(r"G:\Ferti\Jyotisha\.cache\jev_intent") REPORT_JSON = ROOT / "docs" / "research" / "jev_intent_2026_09_19.json" REPORT_MD = ROOT / "docs" / "research" / "jev_intent_2026_09_19.md" def load_jsonl(path: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] if not path.is_file(): return rows for line in path.read_text(encoding="utf-8").splitlines(): if line.strip(): rows.append(json.loads(line)) return rows def gold_of(sample: Mapping[str, Any]) -> dict[str, Any]: gold = sample.get("gold") or {} return enforce_combo(gold.get("intent"), gold.get("answer_class"), gold.get("has_new_dated_event")) def call_jev(sample: Mapping[str, Any], *, timeout: float = 60.0) -> dict[str, Any]: from typesafe_sdk import TypeSafeClient started = time.perf_counter() with TypeSafeClient() as client: response = client.system_one( state=build_state(sample), questions=sdk_questions(sample), model=JEV_MODEL, timeout=timeout, ) elapsed_ms = int((time.perf_counter() - started) * 1000) payload = response.model_dump() if hasattr(response, "model_dump") else dict(response) answers = payload.get("answers") or getattr(response, "answers", {}) or {} if hasattr(answers, "items") and answers and not isinstance(next(iter(answers.values()), None), dict): # SDK objects → dicts converted = {} for key, value in answers.items(): converted[key] = value.model_dump() if hasattr(value, "model_dump") else { "type": getattr(value, "type", None), "choice": getattr(value, "choice", None), "probabilities": getattr(value, "probabilities", None), "confidence": getattr(value, "confidence", None), "noul": getattr(value, "noul", None), } answers = converted parsed = parse_jev_answers(answers) usage = payload.get("usage") or {} return { "ok": True, "unavailable": False, "model": payload.get("model") or JEV_MODEL, "intent": parsed["intent"], "answer_class": parsed["answer_class"], "has_new_dated_event": parsed["has_new_dated_event"], "confidence": parsed["confidence"], "raw": parsed.get("raw"), "input_tokens": usage.get("input_tokens"), "output_tokens": usage.get("output_tokens"), "elapsed_ms": elapsed_ms, } def call_jev_retry(sample: Mapping[str, Any], *, retries: int = 4) -> dict[str, Any]: last_error = "" delay = 1.0 for attempt in range(retries): try: return call_jev(sample) except Exception as exc: # noqa: BLE001 last_error = type(exc).__name__ name = type(exc).__name__ if "RateLimit" in name or "Overload" in name or "Timeout" in name: time.sleep(delay) delay = min(delay * 2, 16) continue if "Authentication" in name or "Permission" in name: break time.sleep(delay) delay = min(delay * 2, 16) return { "ok": False, "unavailable": True, "model": JEV_MODEL, "intent": None, "answer_class": None, "has_new_dated_event": None, "confidence": None, "raw": None, "input_tokens": None, "output_tokens": None, "elapsed_ms": None, "error": last_error, } def match_fields(pred: Mapping[str, Any] | None, gold: Mapping[str, Any]) -> dict[str, bool]: if not pred or pred.get("unavailable"): return {"intent": False, "answer_class": False, "has_new_dated_event": False, "all": False} intent_ok = pred.get("intent") == gold.get("intent") class_needed = gold.get("intent") == "answer_current_focus" class_ok = (pred.get("answer_class") == gold.get("answer_class")) if class_needed else True dated_ok = bool(pred.get("has_new_dated_event")) == bool(gold.get("has_new_dated_event")) return { "intent": intent_ok, "answer_class": class_ok, "has_new_dated_event": dated_ok, "all": intent_ok and class_ok and dated_ok, } def same_output(a: Mapping[str, Any] | None, b: Mapping[str, Any] | None) -> bool: if not a or not b: return False if a.get("unavailable") or b.get("unavailable"): return a.get("unavailable") is True and b.get("unavailable") is True return ( a.get("intent") == b.get("intent") and a.get("answer_class") == b.get("answer_class") and bool(a.get("has_new_dated_event")) == bool(b.get("has_new_dated_event")) ) def percentile(values: Sequence[float], p: float) -> float | None: if not values: return None ordered = sorted(values) if len(ordered) == 1: return float(ordered[0]) idx = int(round((len(ordered) - 1) * p)) return float(ordered[idx]) def layer_metrics(rows: Sequence[Mapping[str, Any]], *, pred_key: str) -> dict[str, Any]: n = len(rows) unavailable = sum(1 for row in rows if (row.get(pred_key) or {}).get("unavailable")) intent_ok = 0 class_ok = 0 class_n = 0 dated_ok = 0 all_ok = 0 high_conf_err = 0 low_conf = 0 wrong = 0 wrong_low = 0 no_unsure = 0 confidences: list[float] = [] elapsed: list[float] = [] tokens: list[float] = [] for row in rows: gold = row["gold"] pred = row.get(pred_key) or {} flags = match_fields(pred, gold) if pred.get("unavailable"): wrong += 1 continue intent_ok += int(flags["intent"]) if gold.get("intent") == "answer_current_focus": class_n += 1 class_ok += int(flags["answer_class"]) dated_ok += int(flags["has_new_dated_event"]) all_ok += int(flags["all"]) conf = pred.get("confidence") if isinstance(conf, (int, float)): confidences.append(float(conf)) if conf >= 0.8 and not flags["all"]: high_conf_err += 1 if conf < 0.5: low_conf += 1 if not flags["all"]: wrong += 1 if conf < 0.5: wrong_low += 1 elif not flags["all"]: wrong += 1 if ( gold.get("intent") == "answer_current_focus" and gold.get("answer_class") in {"no", "unsure"} and pred.get("answer_class") in {"no", "unsure"} and pred.get("answer_class") != gold.get("answer_class") ): no_unsure += 1 if isinstance(pred.get("elapsed_ms"), (int, float)): elapsed.append(float(pred["elapsed_ms"])) if isinstance(pred.get("input_tokens"), (int, float)): tokens.append(float(pred["input_tokens"])) denom = max(n, 1) return { "n": n, "unavailable": unavailable, "intent_acc": intent_ok / denom, "answer_class_acc": (class_ok / class_n) if class_n else None, "answer_class_n": class_n, "dated_acc": dated_ok / denom, "all_acc": all_ok / denom, "high_conf_error_rate": high_conf_err / denom, "low_conf_coverage": low_conf / denom, "low_conf_recall": (wrong_low / wrong) if wrong else None, "no_vs_unsure": no_unsure, "median_ms": percentile(elapsed, 0.5), "p95_ms": percentile(elapsed, 0.95), "mean_input_tokens": (sum(tokens) / len(tokens)) if tokens else None, "cost_usd": ((sum(tokens) / 1_000_000) * 0.042) if tokens else None, } def self_consistency(rows: Sequence[Mapping[str, Any]], a: str, b: str) -> float | None: if not rows: return None hits = sum(1 for row in rows if same_output(row.get(a), row.get(b))) return hits / len(rows) def curve_theta(rows: Sequence[Mapping[str, Any]]) -> dict[str, Any]: """Pick θ on the confidence–accuracy curve for Jev-first fallback.""" points = [] for theta in (0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9): used = 0 correct = 0 fallback = 0 high_err = 0 for row in rows: pred = row.get("jev_1") or {} gold = row["gold"] conf = pred.get("confidence") if pred.get("unavailable") or not isinstance(conf, (int, float)) or conf < theta: fallback += 1 continue used += 1 flags = match_fields(pred, gold) correct += int(flags["all"]) if not flags["all"] and conf >= 0.8: high_err += 1 points.append({ "theta": theta, "jev_share": used / max(len(rows), 1), "fallback_share": fallback / max(len(rows), 1), "jev_acc": (correct / used) if used else None, "high_conf_errors_in_used": high_err, }) # Prefer θ whose used-set accuracy is high and fallback is not everything. ranked = [p for p in points if (p["jev_acc"] or 0) >= 0.85 and p["jev_share"] >= 0.2] recommended = ranked[0]["theta"] if ranked else 0.8 if ranked: recommended = max(ranked, key=lambda p: (p["jev_acc"] or 0, p["jev_share"]))["theta"] return {"points": points, "recommended_theta": recommended} def error_portrait(rows: Sequence[Mapping[str, Any]], *, limit: int = 3) -> dict[str, list[dict[str, Any]]]: buckets = {"cjk_colloquial": [], "literal": [], "other": []} for row in rows: gold = row["gold"] pred = row.get("jev_1") or {} current = row.get("current_1") or {} jev_flags = match_fields(pred, gold) cur_flags = match_fields(current, gold) if current else {"all": False} if jev_flags["all"] or not current or not cur_flags["all"]: if jev_flags["all"] or pred.get("unavailable"): continue # still portrait Jev-wrong even if current didn't run message = str(row.get("user_message") or "") rewritten = message if len(rewritten) > 40: rewritten = rewritten[:18] + "…" + rewritten[-12:] kind = "other" if any(token in message for token in ("嗯", "吧", "咋", "噻", "emmm", "hahaha", "额")): kind = "cjk_colloquial" elif gold.get("intent") != pred.get("intent"): kind = "literal" if len(buckets[kind]) >= limit: continue buckets[kind].append({ "id": row.get("id"), "layer": row.get("layer"), "rewritten_message": rewritten, "gold": gold, "jev": { "intent": pred.get("intent"), "answer_class": pred.get("answer_class"), "has_new_dated_event": pred.get("has_new_dated_event"), "confidence": pred.get("confidence"), }, }) return buckets def run_batch( samples: Sequence[Mapping[str, Any]], *, workers: int, run_id: str, cache: dict[str, Any], call_fn=None, ) -> None: caller = call_fn or call_jev_retry pending = [] for sample in samples: key = f"{run_id}:{sample['id']}" if key in cache: sample[run_id] = cache[key] else: pending.append(sample) if not pending: return print(f"{run_id}: {len(pending)} calls, {len(samples) - len(pending)} cached", flush=True) with ThreadPoolExecutor(max_workers=workers) as pool: futures = {pool.submit(caller, sample): sample for sample in pending} done = 0 for future in as_completed(futures): sample = futures[future] result = future.result() sample[run_id] = result cache[f"{run_id}:{sample['id']}"] = result done += 1 if done % 50 == 0 or done == len(pending): print(f" {run_id} {done}/{len(pending)}", flush=True) def load_cache() -> dict[str, Any]: path = CACHE_DIR / "jev_runs.json" if not path.is_file(): return {} try: return json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError: return {} def save_cache(cache: dict[str, Any]) -> None: CACHE_DIR.mkdir(parents=True, exist_ok=True) path = CACHE_DIR / "jev_runs.json" path.write_text(json.dumps(cache, ensure_ascii=False), encoding="utf-8") def flatten_for_report(sample: Mapping[str, Any], *, include_message: bool) -> dict[str, Any]: row = { "id": sample.get("id"), "source": sample.get("source"), "layer": sample.get("layer"), "gold": sample.get("gold"), "jev_1": sample.get("jev_1"), "jev_2": sample.get("jev_2"), "current_1": sample.get("current_1"), "current_2": sample.get("current_2"), } if include_message: row["user_message"] = sample.get("user_message") row["persona"] = sample.get("persona") return row def write_markdown(report: Mapping[str, Any]) -> None: layers = report["metrics"]["jev_by_layer"] current = report["metrics"].get("current_by_layer") or {} conclusion = report["conclusion"] lines = [ "# TypeSafe Jev 接管校正意图分类 · 离线对照（2026-09-19）", "", f"- 任务：`docs/tasks/TASK-rectification-jev-intent-classifier-research-20260919.md`", f"- 基线：`origin/staging` @ `{report['meta']['baseline']}`", f"- Jev 模型：`{JEV_MODEL}`（不用 jev-latest）", f"- 生成器 / 复核：`{report['meta']['generator']}` / `{report['meta']['reviewer']}`", f"- 真值 sha256：synthetic `{report['meta']['sha256']['synthetic']}`；simulated `{report['meta']['sha256']['simulated']}`；disputed `{report['meta']['sha256']['disputed']}`", f"- 来源 B：{report['meta']['source_b_n']} 条（本地，不提交原文）", "", "## 结论", "", f"**{conclusion['verdict']}**。{conclusion['reason']}", "", f"若接：{conclusion['if_connect']}", "", "## T3 指标（Jev，来源 C）", "", "| 层 | n | intent | answer_class | dated | 高置信错误 | 低置信覆盖 | 低置信召回 | no/unsure 互判 | 自洽率 | 中位 ms | P95 ms | 次均 input tok |", "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", ] for layer in ("choice", "collect", "none"): m = layers[layer] cons = report["metrics"]["jev_self_consistency"].get(layer) def pct(value: float | None) -> str: return "—" if value is None else f"{value:.1%}" def ms(value: float | None) -> str: return "—" if value is None else f"{value:.0f}" lines.append( f"| {layer} | {m['n']} | {pct(m['intent_acc'])} | {pct(m['answer_class_acc'])} | {pct(m['dated_acc'])} | " f"{pct(m['high_conf_error_rate'])} | {pct(m['low_conf_coverage'])} | {pct(m['low_conf_recall'])} | " f"{m['no_vs_unsure']} | {pct(cons)} | {ms(m['median_ms'])} | {ms(m['p95_ms'])} | " f"{(m['mean_input_tokens'] or 0):.0f} |" ) source_a_m = report["metrics"].get("source_a") if source_a_m: acc = source_a_m.get("answer_class_acc") acc_txt = "—" if acc is None else f"{acc:.1%}" lines += [ "", f"来源 A（测试夹具，n={source_a_m['n']}）intent {source_a_m['intent_acc']:.1%}，answer_class {acc_txt}，高置信错误 {source_a_m['high_conf_error_rate']:.1%}。", ] lines += [ "", "## 现行模型对照", "", ] if report["meta"].get("current_model"): note = report["meta"].get("current_model_note") or "" lines.append(f"模型：`{report['meta']['current_model']}`。{note}") lines.append("") lines.append("| 层 | n | intent | answer_class | dated | 自洽率 | 中位 ms |") lines.append("| --- | ---: | ---: | ---: | ---: | ---: | ---: |") for layer in ("choice", "collect", "none"): m = current.get(layer) or {} cons = (report["metrics"].get("current_self_consistency") or {}).get(layer) def pct(value: float | None) -> str: return "—" if value is None else f"{value:.1%}" def ms(value: float | None) -> str: return "—" if value is None else f"{value:.0f}" lines.append( f"| {layer} | {m.get('n', 0)} | {pct(m.get('intent_acc'))} | {pct(m.get('answer_class_acc'))} | " f"{pct(m.get('dated_acc'))} | {pct(cons)} | {ms(m.get('median_ms'))} |" ) paired = report["metrics"].get("paired_sample") or {} if paired: lines.append("") lines.append("同一样本上 Jev vs 现行（相对门槛用这一表）：") lines.append("") lines.append("| 层 | n | Jev intent | 现行 intent | 差（Jev−现行） | 门槛现行−3pp |") lines.append("| --- | ---: | ---: | ---: | ---: | ---: |") for layer in ("choice", "collect", "none"): cell = paired.get(layer) or {} def pct(value: float | None) -> str: return "—" if value is None else f"{value:.1%}" jev = cell.get("jev_intent") cur = cell.get("current_intent") delta = None if jev is None or cur is None else jev - cur gate = None if cur is None else cur - 0.03 lines.append( f"| {layer} | {cell.get('n', 0)} | {pct(jev)} | {pct(cur)} | {pct(delta)} | {pct(gate)} |" ) else: lines.append(report["meta"].get("current_model_note") or "未跑现行模型。") lines += [ "", "## 代表性检验（来源 B vs 来源 C）", "", report["metrics"]["representativeness"]["note"], "", "## 置信度–准确率曲线与 θ", "", f"推荐 θ = {report['metrics']['theta']['recommended_theta']}。点：", "", "```json", json.dumps(report["metrics"]["theta"]["points"], ensure_ascii=False, indent=2), "```", "", "## 错例画像（改写后，无法对应真实会话）", "", ] portrait = report["metrics"]["error_portrait"] for kind, rows in portrait.items(): lines.append(f"### {kind}") lines.append("") if not rows: lines.append("无。") lines.append("") continue for item in rows: lines.append( f"- `{item['id']}` / {item['layer']} / 「{item['rewritten_message']}」" f" gold={item['gold']} jev={item['jev']}" ) lines.append("") lines += [ "## T1 判据对照", "", "见 `scripts/research/jev_intent_questions.py` 的 `CRITERION_MAP`。丢掉的语义：生产提示里的「通常」「不要按 A/B/C/D 猜」「不要按关键词表」；结构不变量改由 `enforce_combo` 强制。", "", "| 现行提示原句 | Jev 落点 | 丢失 |", "| --- | --- | --- |", ] from scripts.research.jev_intent_questions import CRITERION_MAP for row in CRITERION_MAP: lost = row["lost"].replace("|", "\\|") if row["lost"] else "—" lines.append(f"| {row['production']} | {row['jev']} | {lost} |") lines += [ "", "## 回退", "", "任何上线方案必须保留回退到现行会话模型的路径。官方限流会动态调整。", "", ] REPORT_MD.write_text("\n".join(lines) + "\n", encoding="utf-8") def decide_verdict(report: dict[str, Any]) -> dict[str, str]: layers = report["metrics"]["jev_by_layer"] current = report["metrics"].get("current_by_layer") or {} theta = report["metrics"]["theta"]["recommended_theta"] source_b_n = report["meta"]["source_b_n"] high_err = max(layers[layer]["high_conf_error_rate"] for layer in layers) low_recall = [layers[layer]["low_conf_recall"] for layer in layers if layers[layer]["low_conf_recall"] is not None] recall_ok = bool(low_recall) and all(v is not None and v >= 0.60 for v in low_recall) choice_class = layers.get("choice", {}).get("answer_class_acc") if source_b_n < 30: represent = "来源 B 不足 30 条，代表性检验只报数、不判定。" else: represent = report["metrics"]["representativeness"]["note"] absolute_fail: list[str] = [] if high_err > 0.03: absolute_fail.append(f"高置信错误率 {high_err:.1%} > 3%") if not recall_ok: recall_txt = " / ".join(f"{v:.1%}" for v in low_recall) absolute_fail.append(f"低置信召回 {recall_txt} < 60%（错了却仍高置信）") if choice_class is not None and choice_class < 0.80: absolute_fail.append(f"点选题 answer_class 准确率 {choice_class:.1%}（写库字段）") paired = report["metrics"].get("paired_sample") or {} relative_note = "" if paired: bits = [] for layer in ("choice", "collect", "none"): cell = paired.get(layer) or {} jev = cell.get("jev_intent") cur = cell.get("current_intent") if jev is None or cur is None: continue ok = round(jev * 100, 1) >= round((cur - 0.03) * 100, 1) bits.append( f"{layer} Jev {jev:.1%} vs 现行 {cur:.1%}（门槛 {cur-0.03:.1%}，{'过' if ok else '未过'}）" ) if bits: relative_note = " 相对 −3pp（同一样本）：" + "；".join(bits) + "。" if absolute_fail: return { "verdict": "不可接", "reason": ( "来源 C 上 Jev 的绝对门槛未过：" + "；".join(absolute_fail) + "。" + represent + relative_note + (" 现行模型对照未跑，相对 −3pp 门槛无法计算。" if not current and not paired else "") ), "if_connect": ( "不接。现行分类器继续用会话选定的贵模型。" "若还要观察中文口语，只允许 (b) 影子双跑只记日志，不得按 confidence 写库。" f"曲线上 θ={theta} 时高置信错误仍未清零。" ), } paired = report["metrics"].get("paired_sample") or {} if not current and not paired: return { "verdict": "缺数据", "reason": ( "绝对门槛未破，但本机没有会话模型凭据，现行分类器对照未跑，" "无法检验「各层 ≥ 现行 − 3 个百分点」。" f"{represent}" ), "if_connect": ( f"不得上线。若只做影子，用 (b) 双跑只记日志。" f"若将来现行对照过门，优先 (a) Jev 先判、confidence < {theta} 回退现行模型。" ), } gaps = [] for layer in ("choice", "collect", "none"): cell = paired.get(layer) or {} jev = cell.get("jev_intent") cur = cell.get("current_intent") if jev is None: jev = layers[layer]["intent_acc"] if cur is None: cur = (current.get(layer) or {}).get("intent_acc") if cur is None: continue gaps.append((layer, jev, cur, round(jev * 100, 1) - round((cur - 0.03) * 100, 1))) failed = [g for g in gaps if g[3] < 0] if failed: return { "verdict": "不可接", "reason": f"相对门槛未过：失败层={failed}。{represent}", "if_connect": "不接。现行分类器继续用会话选定的贵模型。", } return { "verdict": "可接", "reason": f"来源 C 三层 intent 均 ≥ 现行 − 3pp，高置信错误 ≤3%，低置信召回 ≥60%。{represent}", "if_connect": f"(a) Jev 先判，confidence < {theta} 回退现行模型。必须保留回退路径。", } def _merge_report_preds(samples: list[dict[str, Any]]) -> None: if not REPORT_JSON.is_file(): return try: existing = json.loads(REPORT_JSON.read_text(encoding="utf-8")) except json.JSONDecodeError: return by_id = {row.get("id"): row for row in existing.get("rows") or [] if row.get("id")} for sample in samples: prior = by_id.get(sample.get("id")) or {} for key in ("jev_1", "jev_2", "current_1", "current_2"): if key not in sample and prior.get(key): sample[key] = prior[key] def paired_layer_metrics(rows: Sequence[Mapping[str, Any]]) -> dict[str, Any]: out: dict[str, Any] = {} for layer in ("choice", "collect", "none"): subset = [ row for row in rows if row.get("source") == "C" and row.get("layer") == layer and row.get("current_1") and not (row.get("current_1") or {}).get("unavailable") ] if not subset: out[layer] = {"n": 0, "jev_intent": None, "current_intent": None} continue jev = layer_metrics(subset, pred_key="jev_1") cur = layer_metrics(subset, pred_key="current_1") out[layer] = { "n": len(subset), "jev_intent": jev["intent_acc"], "current_intent": cur["intent_acc"], "jev_answer_class": jev["answer_class_acc"], "current_answer_class": cur["answer_class_acc"], } return out def main(argv: Sequence[str] | None = None) -> int: parser = argparse.ArgumentParser() parser.add_argument("--workers", type=int, default=8) parser.add_argument("--limit", type=int, default=0) parser.add_argument("--skip-second", action="store_true") parser.add_argument("--second-fraction", type=float, default=1.0) parser.add_argument("--current-only", action="store_true") parser.add_argument("--sample-fraction", type=float, default=1.0) parser.add_argument("--sample-seed", type=int, default=20260919) args = parser.parse_args(argv) if not args.current_only and not os.environ.get("TYPESAFE_API_KEY"): print("TYPESAFE_API_KEY missing", file=sys.stderr) return 2 if args.current_only and not os.environ.get("DEEPSEEK_API_KEY"): print("DEEPSEEK_API_KEY missing", file=sys.stderr) return 2 synthetic = load_jsonl(SAMPLES_DIR / "synthetic.jsonl") simulated = load_jsonl(SAMPLES_DIR / "simulated.jsonl") source_b = load_jsonl(CACHE_DIR / "source_b.jsonl") samples = [row for row in synthetic + simulated if row.get("source") in {"A", "C"}] if args.limit: samples = samples[: args.limit] _merge_report_preds(samples) cache = load_cache() current_sample: list[dict[str, Any]] = [] try: if not args.current_only: run_batch(samples, workers=args.workers, run_id="jev_1", cache=cache) save_cache(cache) second = samples if args.skip_second: second = [] elif args.second_fraction < 1: n = max(1, int(len(samples) * args.second_fraction)) second = [row for row in samples if row.get("source") == "C"][:n] if second: run_batch(second, workers=args.workers, run_id="jev_2", cache=cache) save_cache(cache) if source_b: run_batch(source_b, workers=args.workers, run_id="jev_1", cache=cache) save_cache(cache) source_c_rows = [row for row in samples if row.get("source") == "C"] if args.current_only or os.environ.get("DEEPSEEK_API_KEY"): if args.sample_fraction < 1: current_sample = stratified_sample( source_c_rows, fraction=args.sample_fraction, seed=args.sample_seed, ) current_sample.extend(row for row in samples if row.get("source") == "A") else: current_sample = list(samples) run_batch( current_sample, workers=args.workers, run_id="current_1", cache=cache, call_fn=call_current_retry, ) save_cache(cache) second_current = current_sample if args.skip_second: second_current = [] elif args.second_fraction < 1: second_current = stratified_sample( [row for row in current_sample if row.get("source") == "C"], fraction=args.second_fraction, seed=args.sample_seed + 1, ) if second_current: run_batch( second_current, workers=args.workers, run_id="current_2", cache=cache, call_fn=call_current_retry, ) save_cache(cache) by_id = {row["id"]: row for row in current_sample} for row in second_current: dest = by_id.setdefault(row["id"], row) if row.get("current_1"): dest["current_1"] = row["current_1"] if row.get("current_2"): dest["current_2"] = row["current_2"] for sample in samples: extra = by_id.get(sample["id"]) if extra: if extra.get("current_1"): sample["current_1"] = extra["current_1"] if extra.get("current_2"): sample["current_2"] = extra["current_2"] finally: save_cache(cache) source_c = [row for row in samples if row.get("source") == "C"] source_a = [row for row in samples if row.get("source") == "A"] by_layer = {layer: [row for row in source_c if row.get("layer") == layer] for layer in ("choice", "collect", "none")} jev_metrics = {layer: layer_metrics(rows, pred_key="jev_1") for layer, rows in by_layer.items()} jev_cons = {layer: self_consistency(rows, "jev_1", "jev_2") for layer, rows in by_layer.items()} source_a_metrics = layer_metrics(source_a, pred_key="jev_1") if source_a else None source_b_metrics = layer_metrics(source_b, pred_key="jev_1") if source_b else None represent_note = "来源 B 不足 30 条，代表性检验只报数、不判定。" if source_b_metrics and source_b_metrics["n"] >= 30: deltas = [] for layer in ("choice", "collect", "none"): # source B may not be layered the same; compare overall intent acc pass delta = abs((source_b_metrics["intent_acc"] or 0) - ( sum(jev_metrics[layer]["intent_acc"] * jev_metrics[layer]["n"] for layer in jev_metrics) / max(sum(jev_metrics[layer]["n"] for layer in jev_metrics), 1) )) if delta > 0.10: represent_note = f"来源 B 与来源 C 的 intent 准确率相差 {delta:.1%} > 10pp，模拟语料不代表真人，来源 C 门槛结论降为缺数据。" else: represent_note = f"来源 B 与来源 C intent 准确率相差 {delta:.1%}，未超过 10pp。" sha = { "synthetic": __import__("hashlib").sha256((SAMPLES_DIR / "synthetic.jsonl").read_bytes()).hexdigest(), "simulated": __import__("hashlib").sha256((SAMPLES_DIR / "simulated.jsonl").read_bytes()).hexdigest(), "disputed": __import__("hashlib").sha256((SAMPLES_DIR / "disputed.jsonl").read_bytes()).hexdigest() if (SAMPLES_DIR / "disputed.jsonl").is_file() else "", } report: dict[str, Any] = { "meta": { "baseline": "d9722d56", "jev_model": JEV_MODEL, "generator": "agent-template-v1", "reviewer": "agent-rule-v1", "sha256": sha, "source_b_n": len(source_b), "current_model": current_model_id() if any(row.get("current_1") for row in samples) else None, "current_model_note": ( ( f"DeepSeek Flash 顶现行 `classifyRectificationTurnIntent` 提示词；" f"来源 C 分层随机 {args.sample_fraction:.0%}（seed {args.sample_seed}），" f"来源 A 全量。不是线上会话模型。" if any(row.get("current_1") for row in samples) else ( "本机无会话模型目录凭据（模型 key 在数据库加密配置里）。" "现行 `classifyRectificationTurnIntent` 对照未跑。" ) ) ), }, "metrics": { "jev_by_layer": jev_metrics, "jev_self_consistency": jev_cons, "current_by_layer": ( {layer: layer_metrics( [row for row in samples if row.get("source") == "C" and row.get("layer") == layer and row.get("current_1")], pred_key="current_1", ) for layer in ("choice", "collect", "none")} if any(row.get("current_1") for row in samples) else None ), "current_self_consistency": ( {layer: self_consistency( [row for row in samples if row.get("source") == "C" and row.get("layer") == layer and row.get("current_2")], "current_1", "current_2", ) for layer in ("choice", "collect", "none")} if any(row.get("current_2") for row in samples) else None ), "paired_sample": paired_layer_metrics(samples) if any(row.get("current_1") for row in samples) else {}, "source_a": source_a_metrics, "source_b": source_b_metrics, "representativeness": {"note": represent_note}, "theta": curve_theta([row for row in samples if row.get("source") == "C"]), "error_portrait": error_portrait(samples), }, "rows": [flatten_for_report(row, include_message=True) for row in samples], "source_b_aggregate": source_b_metrics, } report["conclusion"] = decide_verdict(report) REPORT_JSON.parent.mkdir(parents=True, exist_ok=True) REPORT_JSON.write_text(json.dumps({ **report, "rows": [flatten_for_report(row, include_message=row.get("source") != "B") for row in samples], }, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") write_markdown(report) print(json.dumps({ "conclusion": report["conclusion"], "jev_by_layer": {k: { "n": v["n"], "intent": v["intent_acc"], "high_conf_err": v["high_conf_error_rate"], "low_conf_recall": v["low_conf_recall"], "unavailable": v["unavailable"], } for k, v in jev_metrics.items()}, "self_consistency": jev_cons, "source_b_n": len(source_b), }, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())