Files
Jyotisha/scripts/research/jev_intent_probe.py
T
jesse-ux c2ecbc74cd
Independent Staging Quality Gate / publish (push) Canceled after 0s
Independent Staging Quality Gate / validate (push) Canceled after 1m37s
research(jev-intent): 修复轮重造语料并全量对照
来源 C 改为 DeepSeek Flash 生成+独立复核,撤回模板拼接结论。来源 B 157 条人工标注后跑 Jev x2 与 Flash 全量对照,结论为缺数据。
2026-09-19 11:30:31 +08:00

942 lines
40 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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 hashlib
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 cache_key(run_id: str, sample: Mapping[str, Any]) -> str:
message = str(sample.get("user_message") or "")
digest = hashlib.sha256(message.encode("utf-8")).hexdigest()[:16]
return f"{run_id}:{sample['id']}:{digest}"
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 = cache_key(run_id, sample)
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[cache_key(run_id, sample)] = 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_v2.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_v2.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"- 真值 sha256:synthetic `{report['meta']['sha256']['synthetic']}`;simulated `{report['meta']['sha256']['simulated']}`;disputed `{report['meta']['sha256']['disputed']}`",
f"- 来源 B:{report['meta']['source_b_n']} 条已标注(本地,不提交原文)",
"",
"## 模型",
"",
f"- 生成模型:`{report['meta']['generator']}` / 版本 `{report['meta'].get('generator_version') or report['meta']['generator']}` / 不是线上会话模型(只用于造来源 C)",
f"- 复核模型:`{report['meta']['reviewer']}` / 版本 `{report['meta'].get('reviewer_version') or report['meta']['reviewer']}` / 不是线上会话模型(只用于独立复核,看不到目标标签)",
f"- 对照模型:`{report['meta'].get('current_model') or '—'}` / 版本 `{report['meta'].get('current_model') or '—'}` / **= 线上会话模型**(DeepSeek Flash,顶生产 `classifyRectificationTurnIntent` 提示词)",
f"- Jev:`{JEV_MODEL}` / 不是线上会话模型",
"",
"## 结论",
"",
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")
represent_meta = report["metrics"].get("representativeness") or {}
represent = represent_meta.get("note") or ""
represent_fail = bool(represent_meta.get("fail"))
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 represent_fail:
return {
"verdict": "缺数据",
"reason": (
"来源 B 与来源 C 同层 intent 准确率差 > 10pp,模拟语料不代表真人,来源 C 门槛结论降为缺数据。"
+ represent
+ ((" 同时来源 C 绝对门槛未过:" + ";".join(absolute_fail) + "。") if absolute_fail else "")
+ relative_note
),
"if_connect": "不得上线。先补真机样本或重造更像真人的来源 C,再测。",
}
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 {}
if prior.get("user_message") != sample.get("user_message"):
continue
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("--current-second-fraction", type=float, default=1.0 / 3)
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_all = load_jsonl(CACHE_DIR / "source_b.jsonl")
source_b = [
row for row in source_b_all
if isinstance(row.get("gold"), dict) and row["gold"].get("intent")
]
unlabeled_b = len(source_b_all) - len(source_b)
samples = [row for row in synthetic + simulated if row.get("source") in {"A", "C"}]
sim_ids = {str(row.get("id")) for row in simulated}
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)
run_batch(source_b, workers=args.workers, run_id="jev_2", 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)
if source_b:
run_batch(
source_b, 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.current_second_fraction < 1:
second_current = stratified_sample(
[row for row in current_sample if row.get("source") == "C"],
fraction=args.current_second_fraction,
seed=args.sample_seed + 1,
)
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
source_b_by_layer = {
layer: layer_metrics([row for row in source_b if row.get("layer") == layer], pred_key="jev_1")
for layer in ("choice", "collect", "none")
} if source_b else {}
represent_fail = False
represent_layers: dict[str, Any] = {}
if not source_b or len(source_b) < 30:
represent_note = f"来源 B 已标注 {len(source_b)} 条(未标注 {unlabeled_b} 不进分母),不足 30 条则只报数、不判定。"
else:
bits = []
for layer in ("choice", "collect", "none"):
b_m = source_b_by_layer.get(layer) or {}
c_m = jev_metrics.get(layer) or {}
if not b_m.get("n"):
represent_layers[layer] = {"n": 0, "delta": None}
continue
delta = abs((b_m.get("intent_acc") or 0) - (c_m.get("intent_acc") or 0))
represent_layers[layer] = {
"n": b_m.get("n"),
"source_b_intent": b_m.get("intent_acc"),
"source_c_intent": c_m.get("intent_acc"),
"delta": delta,
}
bits.append(f"{layer} B {b_m.get('intent_acc'):.1%} vs C {c_m.get('intent_acc'):.1%}(差 {delta:.1%},n_B={b_m.get('n')})")
if delta > 0.10:
represent_fail = True
represent_note = "来源 B 与来源 C 同层 intent 准确率:" + ";".join(bits) + "。"
if represent_fail:
represent_note += " 有层差 > 10pp,结论降为缺数据。"
else:
represent_note += " 各层均未超过 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 "",
}
gen_name = (simulated[0].get("generator") if simulated else None) or "deepseek-flash"
rev_name = (simulated[0].get("reviewer") if simulated else None) or "deepseek-flash"
cache_ids = {
key.split(":")[1]
for key in cache
if key.startswith("jev_1:C-")
}
sim_ok = sim_ids <= cache_ids or args.current_only
print(json.dumps({
"id_check": {
"simulated": len(sim_ids),
"jev_1_cache_c": len(cache_ids),
"simulated_subset_of_cache": sim_ids <= cache_ids,
"missing": sorted(sim_ids - cache_ids)[:12],
}
}, ensure_ascii=False), flush=True)
report: dict[str, Any] = {
"meta": {
"baseline": "69a44fe7",
"jev_model": JEV_MODEL,
"generator": gen_name,
"generator_version": gen_name,
"reviewer": rev_name,
"reviewer_version": rev_name,
"sha256": sha,
"source_b_n": len(source_b),
"source_b_unlabeled": unlabeled_b,
"current_model": current_model_id() if any(row.get("current_1") for row in samples) else None,
"current_model_note": (
(
"DeepSeek Flash = 线上会话模型,顶现行 `classifyRectificationTurnIntent` 提示词;"
f"来源 C 全量第一次"
+ (
f",第二次分层 {args.current_second_fraction:.0%}(seed {args.sample_seed + 1})"
if args.current_second_fraction < 1 else ",第二次全量"
)
+ ";来源 A 全量;来源 B 已标注全量一次。"
if any(row.get("current_1") for row in samples)
else (
"本机无会话模型凭据,现行对照未跑。"
)
)
),
"id_check_ok": sim_ok,
},
"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,
"source_b_by_layer": source_b_by_layer,
"representativeness": {
"note": represent_note,
"fail": represent_fail,
"layers": represent_layers,
},
"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())