feat: --retry-uncertain re-diagnoses uncertain rows; fix docstring

This commit is contained in:
2026-07-16 05:41:19 -04:00
parent c68de849da
commit b307f51340
5 changed files with 73 additions and 10 deletions
+14 -4
View File
@@ -106,21 +106,31 @@ class SqliteDiagnosisSignalStore:
)
self._conn.commit()
def done_question_ids(self, baseline_run_id: str, diag_fingerprint: str) -> set[str]:
def done_question_ids(
self,
baseline_run_id: str,
diag_fingerprint: str,
*,
retry_uncertain: bool = False,
) -> set[str]:
"""查询指定 run 与诊断指纹下已完成的 question_id 集合。
参数:
baseline_run_id: baseline run 标识。
diag_fingerprint: 诊断口径指纹。
retry_uncertain: True 时追加 `AND tier != 'uncertain'`,把 uncertain
(信号不可信降级)题排除出已完成集,令其被重新诊断;默认 False。
返回:
已落盘信号的 question_id 集合;无匹配时为空集,供断点续跑跳过。
"""
cursor = self._conn.execute(
sql = (
"SELECT DISTINCT question_id FROM baseline_diagnosis"
" WHERE baseline_run_id = ? AND diag_fingerprint = ?",
(baseline_run_id, diag_fingerprint),
" WHERE baseline_run_id = ? AND diag_fingerprint = ?"
)
if retry_uncertain:
sql += " AND tier != 'uncertain'"
cursor = self._conn.execute(sql, (baseline_run_id, diag_fingerprint))
return {r["question_id"] for r in cursor.fetchall()}
def load(self, baseline_run_id: str, diag_fingerprint: str) -> list[DiagnosisSignalRow]:
+13 -5
View File
@@ -6,7 +6,9 @@
包装内层 RunLog,兼容 traces 未落表的历史 run);
3. 把 error_attributions / infra / degraded 三类产物确定性投影为
DiagnosisSignalRowtier 由 split_selection.score_signal 判定);
4. 逐行 store.upsert 落盘,单行单事务 → 崩溃最多丢正在写的一行。
4. run 末(Phase 3逐行 store.upsert 落库——诊断在 Phase 2 全部跑完后才落库,
故崩溃丢本次 run 未落库的全部结果(不是"仅一行");靠 GovernedLLMClient 的
Redis 缓存缓解重跑时的 LLM 重烧,下次调用命中缓存直接续。
错误处理诚实标注(不谎称全传播):
- run_diagnosis 的 C1/C2 阶段(指标计算、错误归因)网络/API 失败经
@@ -70,8 +72,9 @@ async def run_baseline_diagnosis(
questions: dict[str, GeneratedQuestion],
store: DiagnosisSignalStore,
deps: DiagnosisDeps,
retry_uncertain: bool = False,
) -> None:
"""对 baseline run 的错题跑离线诊断并把信号逐行落库(断点续跑幂等)。
"""对 baseline run 的错题跑离线诊断并把信号落库(断点续跑幂等)。
参数:
baseline_run_id: baseline run 标识(如 "infer_adhoc"),信号行主键之一。
@@ -79,8 +82,10 @@ async def run_baseline_diagnosis(
wrong_ids: 本次待诊断的可诊断错题 question_id 列表(保序)。
questions: question_id → GeneratedQuestion 映射,需覆盖 wrong_ids 全部题
及 run_diagnosis 返回的所有 infra/degraded 题(用于取 video_id/task_type)。
store: 诊断信号存储端口,逐行 upsert 落盘并提供 done_question_ids 续跑查询。
store: 诊断信号存储端口,upsert 落盘并提供 done_question_ids 续跑查询。
deps: 外部依赖束(见 DiagnosisDeps)。
retry_uncertain: True 时把已落 tier='uncertain'(信号不可信降级)的题也纳入
remaining 重新诊断,透传给 store.done_question_ids;默认 False。
返回:
None。副作用为把逐题 DiagnosisSignalRow 写入 store。
@@ -93,7 +98,9 @@ async def run_baseline_diagnosis(
T0)、degraded_question_idsuncertain)。
"""
# Phase 1: 算 remaining(续跑幂等)
done = store.done_question_ids(baseline_run_id, diag_fingerprint)
done = store.done_question_ids(
baseline_run_id, diag_fingerprint, retry_uncertain=retry_uncertain
)
remaining = [qid for qid in wrong_ids if qid not in done]
if not remaining:
logger.info(
@@ -169,7 +176,8 @@ def _project_and_persist(
{tier: 行数} 计数字典(T2/T1/T0/uncertain),供上层日志与 manifest。
关键实现:
逐行 upsert(单行单事务),中途崩溃最多丢正在写的一行。三桶**非互斥**:
本函数在 run 末(Phase 3)逐行 upsert(单行单事务);诊断已在 Phase 2 全部
跑完,故本阶段中途崩溃丢本次 run 未落库的余下行。三桶**非互斥**:
同一 degraded 错题可能同时出现在 error_attributionsjudge 解析失败仍建
attribution)里,故按 **degraded > infra > attribution** 优先级去重——先落
degraded/infra,再在 attribution 循环跳过已落题,保证**每题恰写一行、
+10
View File
@@ -522,6 +522,7 @@ async def run_pipeline(
out_dir: Path,
generated_at: str,
force: bool = False,
retry_uncertain: bool = False,
) -> SplitBuildResult:
"""内联三阶段:Phase 0 INFRA T0 补录 → Phase 1 诊断 → Phase 2 冻结切分 → McNemar 护栏。
@@ -538,6 +539,7 @@ async def run_pipeline(
out_dir: 冻结产物目录(pools.json + split_manifest.json)。
generated_at: 生成时间戳(ISO 字符串,由调用方传入;见模块 C-2 复现锚点约定)。
force: 覆盖已存在冻结产物开关,透传给 build_split(False 时已存在即报错)。
retry_uncertain: 透传给 run_baseline_diagnosis,令已落 uncertain 题被重新诊断。
返回:
SplitBuildResult(冻结三池 + manifest + assignment)。
@@ -557,6 +559,7 @@ async def run_pipeline(
questions=questions,
store=signal_store,
deps=diagnosis_deps,
retry_uncertain=retry_uncertain,
)
# Phase 2: 冻结切分(读诊断信号 → 贪心选择 → 视频组原子切三池 → 冻结 + 六条断言)。
@@ -656,6 +659,7 @@ def _execute_real(config: VideoSplitConfig, fingerprint: str, args: argparse.Nam
out_dir=out_dir,
generated_at=generated_at,
force=args.force,
retry_uncertain=args.retry_uncertain,
)
)
finally:
@@ -784,6 +788,12 @@ def build_arg_parser() -> argparse.ArgumentParser:
action="store_true",
help="覆盖已存在的冻结 pools.json/manifest(旧产物备份为 .bak.*",
)
parser.add_argument(
"--retry-uncertain",
action="store_true",
dest="retry_uncertain",
help="把已落 tier='uncertain'(信号不可信降级)的题重新诊断,而非当作已完成跳过",
)
return parser
+9 -1
View File
@@ -127,12 +127,20 @@ class DiagnosisSignalStore(Protocol):
"""
...
def done_question_ids(self, baseline_run_id: str, diag_fingerprint: str) -> set[str]:
def done_question_ids(
self,
baseline_run_id: str,
diag_fingerprint: str,
*,
retry_uncertain: bool = False,
) -> set[str]:
"""查询指定 run 与诊断指纹下已完成的 question_id 集合。
参数:
baseline_run_id: baseline run 标识。
diag_fingerprint: 诊断口径指纹。
retry_uncertain: True 时把 tier='uncertain'(信号不可信降级)题视为
未完成,令其被重新诊断;默认 False(uncertain 也算完成,不重诊)。
返回:
已落盘信号的 question_id 集合,用于断点续跑跳过。
@@ -85,6 +85,33 @@ def test_null_fields_for_infra_row(tmp_path):
assert rows[0].error_type is None and rows[0].evolution_target is None and rows[0].infra is True
def test_done_question_ids_retry_uncertain_excludes(tmp_path):
"""retry_uncertain=True 时 uncertain 题不算完成(会被重诊),非 uncertain 仍算完成。"""
s = _store(tmp_path)
# T2defect)题:已完成
s.upsert(
DiagnosisSignalRow(
"t2_qid", "v1", "run", "fp", "Counting",
"search_failure", "defect", "T2", "skill",
degraded=False, infra=False, session_id="s",
)
)
# uncertaindegraded)题:信号不可信
s.upsert(
DiagnosisSignalRow(
"uncertain_qid", "v2", "run", "fp", "Counting",
None, None, "uncertain", None,
degraded=True, infra=False, session_id=None,
)
)
# 默认:两者都算完成
assert s.done_question_ids("run", "fp") == {"t2_qid", "uncertain_qid"}
# retry_uncertainuncertain 被排除,T2 仍算完成
done = s.done_question_ids("run", "fp", retry_uncertain=True)
assert "uncertain_qid" not in done
assert "t2_qid" in done
def test_context_manager_closes_connection(tmp_path):
# with 块退出后连接关闭,再操作应报 ProgrammingError
with SqliteDiagnosisSignalStore(str(tmp_path / "h.db")) as s: