Files
Jyotisha/scripts/research/jev_intent_probe.py
T
jesse-ux 507ef959f5
Independent Staging Quality Gate / validate (push) Failing after 9m3s
Independent Staging Quality Gate / publish (push) Skipped
research(jev-intent): fix2 把来源 B 现行与高置信错误补进报告
离线从 cache 聚合,不重跑模型。无焦点层 Flash 69.7% 低于 Jev 78.8%。采集层相对门槛标不可判。
2026-09-19 12:28:43 +08:00

1128 lines
49 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}"
NONE_CONFUSION_LABELS = (
"provide_new_evidence",
"stop_rectification",
"ask_about_result",
"unclear",
"answer_current_focus",
)
def attach_from_cache(
samples: Sequence[dict[str, Any]],
cache: Mapping[str, Any],
run_ids: Sequence[str],
) -> dict[str, int]:
missing = {rid: 0 for rid in run_ids}
for sample in samples:
for rid in run_ids:
key = cache_key(rid, sample)
if key in cache:
sample[rid] = cache[key]
else:
missing[rid] += 1
return missing
def strip_confidence(metrics: Mapping[str, Any] | None) -> dict[str, Any] | None:
if metrics is None:
return None
out = dict(metrics)
out["high_conf_error_rate"] = None
out["low_conf_coverage"] = None
out["low_conf_recall"] = None
return out
def confusion_counts(
rows: Sequence[Mapping[str, Any]],
pred_key: str,
labels: Sequence[str] = NONE_CONFUSION_LABELS,
) -> dict[str, Any]:
counts = {gold: {pred: 0 for pred in labels} for gold in labels}
other = 0
for row in rows:
gold = (row.get("gold") or {}).get("intent")
pred = (row.get(pred_key) or {}).get("intent")
if gold in counts and pred in counts[gold]:
counts[gold][pred] += 1
else:
other += 1
return {"labels": list(labels), "counts": counts, "other": other, "n": len(rows)}
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
if layer == "collect":
verdict = "不可判(复核与对照同源)"
elif jev is None or cur is None:
verdict = "—"
elif round(jev * 100, 1) >= round((cur - 0.03) * 100, 1):
verdict = "过"
else:
verdict = "未过"
lines.append(
f"| {layer} | {cell.get('n', 0)} | {pct(jev)} | {pct(cur)} | {pct(delta)} | {pct(gate)} | {verdict} |"
)
else:
lines.append(report["meta"].get("current_model_note") or "未跑现行模型。")
lines += [
"",
"## 代表性检验(来源 B vs 来源 C)",
"",
report["metrics"]["representativeness"]["note"],
"",
]
b_jev = report["metrics"].get("source_b")
b_jev_layer = report["metrics"].get("source_b_by_layer") or {}
b_cur = report["metrics"].get("current_source_b")
b_cur_layer = report["metrics"].get("current_source_b_by_layer") or {}
b_cons = report["metrics"].get("source_b_jev_self_consistency") or {}
if b_jev:
def pct(value: float | None) -> str:
return "—" if value is None else f"{value:.1%}"
lines += [
"来源 B 是真人 + 人工标注 + 线上模型三者齐备的唯一一组。现行无置信度,置信度三列为空。",
"",
"| 范围 | n | Jev intent | 现行 intent | Jev 高置信错误 | Jev 低置信召回 | Jev 自洽 |",
"| --- | ---: | ---: | ---: | ---: | ---: | ---: |",
]
rows_spec = [("全集", b_jev, b_cur, b_cons.get("all"))]
for layer in ("choice", "collect", "none"):
rows_spec.append((
layer,
b_jev_layer.get(layer) or {},
b_cur_layer.get(layer) or {},
b_cons.get(layer),
))
for name, jev_m, cur_m, cons in rows_spec:
lines.append(
f"| {name} | {jev_m.get('n', 0)} | {pct(jev_m.get('intent_acc'))} | {pct((cur_m or {}).get('intent_acc'))} | "
f"{pct(jev_m.get('high_conf_error_rate'))} | {pct(jev_m.get('low_conf_recall'))} | {pct(cons)} |"
)
lines.append("")
none_cur = (b_cur_layer.get("none") or {}).get("intent_acc")
none_jev = (b_jev_layer.get("none") or {}).get("intent_acc")
if none_cur is not None and none_jev is not None:
lines.append(
f"无焦点层现行 intent {none_cur:.1%}、Jev {none_jev:.1%}。"
"现行更低,说明 78.8% 主要是这 33 条本身难,不是单 Jev 不行。"
)
lines.append("")
confusion = report["metrics"].get("source_b_none_confusion") or {}
if confusion:
lines += [
"无焦点层 gold × 预测混淆计数(只有计数,无原文)。预测出现 `answer_current_focus` 是因为模型把无焦点句当成在回答采集题。",
"",
]
for title, key in (("gold × Jev", "jev"), ("gold × 现行", "current")):
table = confusion.get(key) or {}
labels = table.get("labels") or list(NONE_CONFUSION_LABELS)
counts = table.get("counts") or {}
lines.append(f"### {title}(n={table.get('n', 0)})")
lines.append("")
header = "| gold \\ pred | " + " | ".join(labels) + " |"
sep = "| --- | " + " | ".join("---:" for _ in labels) + " |"
lines.append(header)
lines.append(sep)
for gold in labels:
row_counts = counts.get(gold) or {}
cells = " | ".join(str(row_counts.get(pred, 0)) for pred in labels)
lines.append(f"| {gold} | {cells} |")
lines.append("")
lines += [
"## 置信度–准确率曲线与 θ",
"",
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 += [
"",
"## 限制",
"",
"1. **复核 ≈ 生产提示,且复核模型 = 对照模型。** `REVIEW_RUBRIC` 与生产 `COLLECT_INSTRUCTIONS` 逐句对应;生成 / 复核 / 对照都是 `deepseek-flash`。进入测试集的 900 条是「Flash 用近生产提示能答对目标标签」的那 900 条,被剔的 36 条恰是 Flash 不同意的。因此来源 C 上现行 97.5% / 98.8% / 94.0% 是构造出来的上界。采集层「Jev 92.2% 未过相对门槛 94.5%」**不可当作 Jev 输给现行的证据**。",
"2. **采集层 intent 错例的 gold 有争议。** 来源 C 采集层 Jev intent 错例 31 条:`provide_new_evidence → answer_current_focus` 17、`unclear → answer_current_focus` 9、`stop → unclear` 4、`ask → unclear` 1。17 条 pne 几乎全是「另外 2019 年我换工作搬了家」句式,若干尾句落在当前题域,按生产提示可读成 `answer_current_focus + unsure`。9 条 unclear(「一时半会儿真捋不明白」)按「记不清 → unsure」也读得通。两类合计 ≥ 20 条,占该层 intent 错例约 2/3。gold 来自「生成目标 + Flash 复核同意」,不等于人工真值。改写示例:",
" - 「另外 2019 年换过工作,感情那会儿真没细想。」gold=provide_new_evidence;可读成在回答感情采集题。",
" - 「另外 2019 年搬了家,工作那摊子反而没顾上细想。」gold=provide_new_evidence;可读成在回答工作采集题。",
" - 「这事我一时真说不上来。」gold=unclear;按生产提示是 unsure。",
"3. **语料仍不像真人。** 修复单 1 只把长度和人设写成硬红线(已过)。原单还要求按来源 B 的标点 / 语气词比例约束,未进红线:",
"",
"| 指标 | 来源 B(真人) | 来源 C(模拟) |",
"| --- | ---: | ---: |",
"| 含标点 | 13% | 98.9% |",
"| 含语气词(吧/呢/啊/嗯/哦/额/emm) | 4% | 25.6% |",
"| 含年份或月份 | 81% | 47.8% |",
"| 带年份句里写「2019」 | — | 236 / 429 = 55% |",
"| `provide_new_evidence` 里是搬家/换工作 | — | 152 / 157 |",
"",
"根因:生成脚本 `ALT_EVENT_HINTS` 给七个领域的「另一件事」全是搬家/换工作,年份未约束,模型收敛到「另外 2019 年搬过家」。这解释了模拟语料不代表真人的一部分,也解释了采集层错例为何长得一样。本单不修,留给产品决定是否再造一轮。",
"",
"## 回退",
"",
"任何上线方案必须保留回退到现行会话模型的路径。官方限流会动态调整。",
"",
]
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)
mark = "不可判(复核与对照同源)" if layer == "collect" else ("过" if ok else "未过")
bits.append(
f"{layer} Jev {jev:.1%} vs 现行 {cur:.1%}(门槛 {cur-0.03:.1%},{mark})"
)
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)
parser.add_argument("--offline", action="store_true", help="aggregate from cache only, no API calls")
args = parser.parse_args(argv)
if not args.offline and not args.current_only and not os.environ.get("TYPESAFE_API_KEY"):
print("TYPESAFE_API_KEY missing", file=sys.stderr)
return 2
if not args.offline and 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]] = []
second_current: list[dict[str, Any]] = []
if args.offline:
miss_c = attach_from_cache(samples, cache, ("jev_1", "jev_2", "current_1", "current_2"))
miss_b = attach_from_cache(source_b, cache, ("jev_1", "jev_2", "current_1"))
print(json.dumps({"offline_missing": {"samples": miss_c, "source_b": miss_b}}, ensure_ascii=False), flush=True)
try:
if not args.offline and 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 not args.offline and (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 {}
current_source_b = strip_confidence(
layer_metrics(source_b, pred_key="current_1") if source_b and any(row.get("current_1") for row in source_b) else None
)
current_source_b_by_layer = {
layer: strip_confidence(layer_metrics(
[row for row in source_b if row.get("layer") == layer and row.get("current_1")],
pred_key="current_1",
))
for layer in ("choice", "collect", "none")
} if source_b and any(row.get("current_1") for row in source_b) else {}
source_b_jev_self_consistency = (
{
"all": self_consistency(source_b, "jev_1", "jev_2"),
**{
layer: self_consistency(
[row for row in source_b if row.get("layer") == layer],
"jev_1",
"jev_2",
)
for layer in ("choice", "collect", "none")
},
}
if source_b else {}
)
none_b = [row for row in source_b if row.get("layer") == "none"]
source_b_none_confusion = (
{
"jev": confusion_counts(none_b, "jev_1"),
"current": confusion_counts(none_b, "current_1"),
}
if none_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,
"current_source_b": current_source_b,
"current_source_b_by_layer": current_source_b_by_layer,
"source_b_jev_self_consistency": source_b_jev_self_consistency,
"source_b_none_confusion": source_b_none_confusion,
"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),
"current_source_b": {
"n": (current_source_b or {}).get("n"),
"intent": (current_source_b or {}).get("intent_acc"),
},
"source_b_none_confusion_n": (source_b_none_confusion.get("jev") or {}).get("n"),
}, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())