From 93c9be8bfa792bc73fd61d19dc4bf75a11e1a9c5 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Thu, 9 Jul 2026 05:47:52 -0400 Subject: [PATCH] =?UTF-8?q?feat(tools):=20generate=5Fquestions.py=20calibr?= =?UTF-8?q?ate=20=E5=AD=90=E5=91=BD=E4=BB=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fisher exact test + effect size 组合判定（PASS/WARN/FAIL） - 按 video_id 分组推理，避免跨视频树错用 - baseline 支持从 DB 读取或自动跑推理 - 对比表输出 + 退出码控制 Co-Authored-By: Claude Opus 4.6 (1M context) --- tests/unit/test_generate_questions.py | 95 +++++ tools/generate_questions.py | 515 +++++++++++++++++++++++++- 2 files changed, 606 insertions(+), 4 deletions(-) diff --git a/tests/unit/test_generate_questions.py b/tests/unit/test_generate_questions.py index 3d608af..eebd23f 100644 --- a/tests/unit/test_generate_questions.py +++ b/tests/unit/test_generate_questions.py @@ -11,6 +11,7 @@ import sys from pathlib import Path import numpy as np +import pytest # 确保项目根目录在 sys.path 中 PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent @@ -20,10 +21,13 @@ if str(PROJECT_ROOT) not in sys.path: from core.types import GeneratedQuestion from tools.generate_questions import ( _append_to_json, + _calibrate_exit_code, + _judge_task_type, _load_or_init_progress, _rebuild_embedding_pool, _save_progress, _select_exemplars, + _validate_calibrate_args, ) # --------------------------------------------------------------------------- @@ -320,3 +324,94 @@ class TestAppendToJson: v2_data = json.loads((tmp_path / "v2.json").read_text()) assert len(v1_data) == 1 assert len(v2_data) == 1 + + +# --------------------------------------------------------------------------- +# TestCalibrateJudgment +# --------------------------------------------------------------------------- + + +class TestCalibrateJudgment: + """_judge_task_type 校准判定测试。""" + + def test_pass_when_delta_small(self) -> None: + """差值在容忍范围内判定为 PASS。""" + verdict = _judge_task_type( + bench_correct=60, + bench_total=100, + gen_correct=12, + gen_total=20, + tolerance=0.10, + alpha=0.05, + ) + assert verdict == "PASS" + + def test_fail_when_delta_large_and_significant(self) -> None: + """差值超阈值且统计显著判定为 FAIL。""" + verdict = _judge_task_type( + bench_correct=144, + bench_total=240, + gen_correct=6, + gen_total=20, + tolerance=0.10, + alpha=0.05, + ) + assert verdict == "FAIL" + + def test_warn_when_delta_large_but_not_significant(self) -> None: + """差值超阈值但不统计显著判定为 WARN。""" + verdict = _judge_task_type( + bench_correct=2, + bench_total=3, + gen_correct=8, + gen_total=20, + tolerance=0.10, + alpha=0.05, + ) + assert verdict == "WARN" + + +# --------------------------------------------------------------------------- +# TestCalibrateIntegration +# --------------------------------------------------------------------------- + + +class TestCalibrateIntegration: + """calibrate 辅助函数集成测试。""" + + def test_baseline_params_must_be_paired(self) -> None: + """baseline 参数必须成对出现。""" + with pytest.raises(ValueError, match="成对"): + _validate_calibrate_args(baseline_db="some.db", baseline_run_id=None) + + def test_baseline_params_both_none_ok(self) -> None: + """两个参数都为 None 不报错。""" + _validate_calibrate_args(baseline_db=None, baseline_run_id=None) + + def test_baseline_params_both_provided_ok(self) -> None: + """两个参数都提供不报错。""" + _validate_calibrate_args(baseline_db="some.db", baseline_run_id="run-001") + + def test_baseline_run_id_only_raises(self) -> None: + """只提供 run_id 也报错。""" + with pytest.raises(ValueError, match="成对"): + _validate_calibrate_args(baseline_db=None, baseline_run_id="run-001") + + def test_has_fail_returns_exit_code_1(self) -> None: + """存在 FAIL 时返回退出码 1。""" + verdicts = {"Object Recognition": "PASS", "Action Reasoning": "FAIL"} + assert _calibrate_exit_code(verdicts) == 1 + + def test_all_pass_or_warn_returns_exit_code_0(self) -> None: + """全部 PASS 或 WARN 时返回退出码 0。""" + verdicts = {"Object Recognition": "PASS", "Spatial Perception": "WARN"} + assert _calibrate_exit_code(verdicts) == 0 + + def test_all_pass_returns_exit_code_0(self) -> None: + """全部 PASS 时返回退出码 0。""" + verdicts = {"Object Recognition": "PASS", "Action Reasoning": "PASS"} + assert _calibrate_exit_code(verdicts) == 0 + + def test_empty_verdicts_returns_exit_code_0(self) -> None: + """空 verdicts 时返回退出码 0。""" + assert _calibrate_exit_code({}) == 0 diff --git a/tools/generate_questions.py b/tools/generate_questions.py index 741de37..a646027 100644 --- a/tools/generate_questions.py +++ b/tools/generate_questions.py @@ -17,6 +17,8 @@ import json import os import random import sys +import uuid +from collections import defaultdict from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parent.parent @@ -28,7 +30,9 @@ from loguru import logger load_dotenv(PROJECT_ROOT / ".env") import numpy as np +from scipy.stats import fisher_exact +from app.harness.log import HarnessLog from app.question_gen.loader import load_benchmark from app.question_gen.synthesizer import ( TASK_TYPE_LEVEL_MAP, @@ -355,6 +359,438 @@ def _append_to_json(output_dir: Path, question: GeneratedQuestion) -> None: os.replace(str(tmp), str(json_path)) +# --------------------------------------------------------------------------- +# calibrate 辅助函数 +# --------------------------------------------------------------------------- + + +def _judge_task_type( + bench_correct: int, + bench_total: int, + gen_correct: int, + gen_total: int, + tolerance: float, + alpha: float, +) -> str: + """判定单个题型的校准结果。 + + 根据 benchmark 和生成题正确率差值 + Fisher 精确检验决定判定。 + + 参数: + bench_correct: benchmark 答对数。 + bench_total: benchmark 总题数。 + gen_correct: 生成题答对数。 + gen_total: 生成题总题数。 + tolerance: 正确率差值容忍阈值。 + alpha: Fisher 检验显著性水平。 + + 返回: + "PASS" — 差值在容忍范围内。 + "FAIL" — 差值超阈值且统计显著。 + "WARN" — 差值超阈值但不显著。 + """ + delta = abs(gen_correct / gen_total - bench_correct / bench_total) + if delta <= tolerance: + return "PASS" + + table = [ + [bench_correct, bench_total - bench_correct], + [gen_correct, gen_total - gen_correct], + ] + _, p = fisher_exact(table) + + if p < alpha and delta > tolerance: + return "FAIL" + return "WARN" + + +def _validate_calibrate_args( + baseline_db: str | None, + baseline_run_id: str | None, +) -> None: + """校验 baseline 参数必须成对出现。 + + 参数: + baseline_db: 基线数据库路径。 + baseline_run_id: 基线运行标识。 + + 异常: + ValueError: 只提供了一个而非两个参数。 + """ + has_db = baseline_db is not None + has_run_id = baseline_run_id is not None + if has_db != has_run_id: + raise ValueError("--baseline-db 和 --baseline-run-id 必须成对出现") + + +def _calibrate_exit_code(verdicts: dict[str, str]) -> int: + """根据所有题型的判定结果决定进程退出码。 + + 参数: + verdicts: {题型: "PASS"|"WARN"|"FAIL"} 映射。 + + 返回: + 存在任一 FAIL → 1，否则 → 0。 + """ + if any(v == "FAIL" for v in verdicts.values()): + return 1 + return 0 + + +def _read_baseline_per_task_type( + db_path: str, + run_id: str, +) -> dict[str, dict]: + """从已有 HarnessLog DB 中读取指定 run 的 per_task_type 正确率。 + + 参数: + db_path: SQLite 数据库路径。 + run_id: 运行标识。 + + 返回: + {task_type: {"accuracy": float, "total": int, "correct": int}}。 + + 异常: + FileNotFoundError: 数据库文件不存在。 + ValueError: 未找到指定 run_id 的预测记录。 + """ + import sqlite3 + + if not Path(db_path).exists(): + raise FileNotFoundError(f"基线数据库不存在: {db_path}") + + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + try: + rows = conn.execute( + "SELECT task_type, prediction, answer FROM predictions WHERE run_id = ?", + (run_id,), + ).fetchall() + finally: + conn.close() + + if not rows: + raise ValueError(f"未找到 run_id={run_id} 的预测记录") + + groups: dict[str, list[dict]] = defaultdict(list) + for row in rows: + groups[dict(row)["task_type"]].append(dict(row)) + + result: dict[str, dict] = {} + for task_type, records in groups.items(): + total = len(records) + correct = sum(1 for r in records if r["prediction"] == r["answer"]) + result[task_type] = { + "accuracy": correct / total, + "total": total, + "correct": correct, + } + return result + + +def _build_llm_client(): + """构建 GovernedLLMClient（推理用 LLM）。 + + 从 .env 读取 SEARCH_LLM_MODEL / SEARCH_LLM_BASE_URL / SEARCH_LLM_API_KEY + 和 LLM 韧性参数。 + + 返回: + GovernedLLMClient 实例。 + """ + from adapters.breaker import CircuitBreaker + from adapters.llm import GovernedLLMClient + from adapters.telemetry import SQLiteTelemetryRecorder + + (PROJECT_ROOT / "logs").mkdir(exist_ok=True) + telemetry = SQLiteTelemetryRecorder(str(PROJECT_ROOT / "logs" / "calibrate_telemetry.db")) + + breaker_threshold = int(os.getenv("LLM_CIRCUIT_BREAKER_THRESHOLD", "5")) + breaker_cooldown = int(os.getenv("LLM_CIRCUIT_BREAKER_COOLDOWN", "60")) + timeout_s = float(os.getenv("LLM_TIMEOUT", "120")) + max_retries = int(os.getenv("LLM_MAX_RETRIES", "3")) + base_delay = float(os.getenv("LLM_RETRY_BASE_DELAY", "2.0")) + max_delay = float(os.getenv("LLM_RETRY_MAX_DELAY", "30.0")) + ttft = float(os.getenv("LLM_TTFT_TIMEOUT", "30")) + inter_token = float(os.getenv("LLM_INTER_TOKEN_TIMEOUT", "15")) + + return GovernedLLMClient( + model=os.environ["SEARCH_LLM_MODEL"], + base_url=os.environ["SEARCH_LLM_BASE_URL"], + api_key=os.environ["SEARCH_LLM_API_KEY"], + provider="deepseek", + thinking=False, + breaker=CircuitBreaker(fail_threshold=breaker_threshold, cooldown_s=breaker_cooldown), + cache=None, + telemetry=telemetry, + timeout_s=timeout_s, + ttft_timeout_s=ttft, + inter_token_timeout_s=inter_token, + max_retries=max_retries, + retry_base_delay_s=base_delay, + retry_max_delay_s=max_delay, + ) + + +async def _run_inference_for_questions( + questions: list, + *, + store_dir: Path, + prompts_dir: Path, + db_path: str, + run_id: str, + concurrency: int, + max_steps: int, + skill_mode: str, + llm, + vlm, + embed_provider, +) -> dict[str, dict]: + """对题目列表运行推理，返回 per_task_type 指标。 + + 按 video_id 分组，逐组构建推理依赖并执行推理， + 最后合并所有组的 per_task_type 结果。 + + 参数: + questions: 待推理的题目列表。 + store_dir: store 根目录。 + prompts_dir: prompt 文件目录。 + db_path: SQLite 数据库路径。 + run_id: 运行标识。 + concurrency: 最大并发数。 + max_steps: AgentLoop 单题最大步数。 + skill_mode: skill 模式。 + llm: LLMProvider 实例。 + vlm: VLMProvider 实例。 + embed_provider: EmbeddingProvider 实例。 + + 返回: + {task_type: {"accuracy": float, "total": int, "correct": int}}。 + """ + from app.harness.factory import build_inference_deps + from app.harness.inference import run_inference + + # Phase 1: 按 video_id 分组 + by_video: dict[str, list] = defaultdict(list) + for q in questions: + by_video[q.video_id].append(q) + + # Phase 2: 逐组推理 + all_per_task: dict[str, dict] = {} + skills_dir = store_dir / "skills" + if not skills_dir.exists(): + skills_dir = None + + with HarnessLog(db_path, run_id) as log: + for video_id, group in by_video.items(): + deps = build_inference_deps( + store_dir=store_dir, + video_id=video_id, + prompts_dir=prompts_dir, + skills_dir=skills_dir, + skill_mode=skill_mode, + embed_provider=embed_provider, + llm=llm, + vlm=vlm, + ocr=None, + verify_vision=False, + anchor=False, + assemble_mode="plain", + ) + result = await run_inference( + group, + llm=deps.llm, + tool_dispatch_fn=deps.tool_dispatch_fn, + prompt_builder=deps.prompt_builder, + log=log, + run_id=run_id, + concurrency=concurrency, + max_steps=max_steps, + skill_mode=skill_mode, + ) + # Phase 3: 合并 per_task_type + for task_type, metrics in result.per_task_type.items(): + if task_type in all_per_task: + existing = all_per_task[task_type] + merged_total = existing["total"] + metrics["total"] + merged_correct = existing["correct"] + metrics["correct"] + all_per_task[task_type] = { + "accuracy": merged_correct / merged_total, + "total": merged_total, + "correct": merged_correct, + } + else: + all_per_task[task_type] = dict(metrics) + + return all_per_task + + +def _format_comparison_table( + bench_per_task: dict[str, dict], + gen_per_task: dict[str, dict], + verdicts: dict[str, str], + p_values: dict[str, float], +) -> str: + """格式化校准比较表。 + + 参数: + bench_per_task: benchmark 各题型指标。 + gen_per_task: 生成题各题型指标。 + verdicts: 各题型判定结果。 + p_values: 各题型 Fisher 检验 p 值。 + + 返回: + 格式化的比较表字符串。 + """ + verdict_symbols = {"PASS": "✓ PASS", "WARN": "⚠ WARN", "FAIL": "✗ FAIL"} + all_types = sorted(set(bench_per_task) | set(gen_per_task)) + + header = f"{'题型':<20s} | {'bench':>6s} | {'gen':>6s} | {'Δ':>7s} | {'p-value':>7s} | 判定" + sep = "-" * 19 + "-|" + "-" * 8 + "|" + "-" * 8 + "|" + "-" * 9 + "|" + "-" * 9 + "|" + "-" * 8 + lines = [header, sep] + + for task_type in all_types: + b = bench_per_task.get(task_type, {"accuracy": 0.0, "total": 0, "correct": 0}) + g = gen_per_task.get(task_type, {"accuracy": 0.0, "total": 0, "correct": 0}) + delta = g["accuracy"] - b["accuracy"] + p_val = p_values.get(task_type, float("nan")) + verdict = verdicts.get(task_type, "N/A") + symbol = verdict_symbols.get(verdict, verdict) + + lines.append( + f"{task_type:<20s} | {b['accuracy']:>5.1%} | {g['accuracy']:>5.1%} " + f"| {delta:>+6.1%} | {p_val:>7.3f} | {symbol}" + ) + + return "\n".join(lines) + + +async def _run_calibrate(args: argparse.Namespace) -> None: + """calibrate 子命令主流程。 + + 对比 benchmark 和生成题在 Agent 推理下的正确率， + 逐题型 Fisher 精确检验判定校准质量。 + + 参数: + args: CLI 参数。 + """ + _validate_calibrate_args( + getattr(args, "baseline_db", None), + getattr(args, "baseline_run_id", None), + ) + + generated_dir = Path(args.generated_dir) + benchmark_dir = Path(args.benchmark_dir) + store_dir = Path(args.store_dir) + db_path = args.db_path + prompts_dir = Path(args.prompts_dir) + concurrency = args.concurrency + max_steps = args.max_steps + skill_mode = args.skill_mode + tolerance = args.tolerance + alpha = args.alpha + + # Phase 1: 加载题目 + logger.info("加载生成题目: {}", generated_dir) + gen_questions = load_benchmark(generated_dir) + logger.info("加载 benchmark 题目: {}", benchmark_dir) + bench_questions = load_benchmark(benchmark_dir) + logger.info("生成题 {} 道, benchmark {} 道", len(gen_questions), len(bench_questions)) + + # Phase 2: 获取 benchmark baseline + baseline_db = getattr(args, "baseline_db", None) + baseline_run_id = getattr(args, "baseline_run_id", None) + + if baseline_db and baseline_run_id: + logger.info("从基线 DB 读取 benchmark 指标: db={}, run_id={}", baseline_db, baseline_run_id) + bench_per_task = _read_baseline_per_task_type(baseline_db, baseline_run_id) + else: + logger.info("运行 benchmark 推理以获取基线指标") + llm = _build_llm_client() + vlm = _build_vlm_client() + embed_provider = _build_embed_provider() + bench_run_id = f"calibrate-bench-{uuid.uuid4().hex[:8]}" + bench_per_task = await _run_inference_for_questions( + bench_questions, + store_dir=store_dir, + prompts_dir=prompts_dir, + db_path=db_path, + run_id=bench_run_id, + concurrency=concurrency, + max_steps=max_steps, + skill_mode=skill_mode, + llm=llm, + vlm=vlm, + embed_provider=embed_provider, + ) + + # Phase 3: 运行生成题推理 + logger.info("运行生成题推理") + if not baseline_db: + # 客户端已在 Phase 2 构建 + pass + else: + llm = _build_llm_client() + vlm = _build_vlm_client() + embed_provider = _build_embed_provider() + + gen_run_id = f"calibrate-gen-{uuid.uuid4().hex[:8]}" + gen_per_task = await _run_inference_for_questions( + gen_questions, + store_dir=store_dir, + prompts_dir=prompts_dir, + db_path=db_path, + run_id=gen_run_id, + concurrency=concurrency, + max_steps=max_steps, + skill_mode=skill_mode, + llm=llm, + vlm=vlm, + embed_provider=embed_provider, + ) + + # Phase 4: 逐题型判定 + all_types = sorted(set(bench_per_task) | set(gen_per_task)) + verdicts: dict[str, str] = {} + p_values: dict[str, float] = {} + + for task_type in all_types: + b = bench_per_task.get(task_type) + g = gen_per_task.get(task_type) + if b is None or g is None or b["total"] == 0 or g["total"] == 0: + verdicts[task_type] = "WARN" + p_values[task_type] = float("nan") + continue + + verdicts[task_type] = _judge_task_type( + bench_correct=b["correct"], + bench_total=b["total"], + gen_correct=g["correct"], + gen_total=g["total"], + tolerance=tolerance, + alpha=alpha, + ) + + # 计算 p-value 供表格显示 + table = [ + [b["correct"], b["total"] - b["correct"]], + [g["correct"], g["total"] - g["correct"]], + ] + _, p_val = fisher_exact(table) + p_values[task_type] = p_val + + # Phase 5: 输出比较表 + table_str = _format_comparison_table(bench_per_task, gen_per_task, verdicts, p_values) + logger.info("校准比较表:\n{}", table_str) + + # Phase 6: 退出 + exit_code = _calibrate_exit_code(verdicts) + if exit_code == 0: + logger.info("校准通过: 所有题型 PASS 或 WARN") + else: + logger.error("校准失败: 存在 FAIL 题型") + sys.exit(exit_code) + + # --------------------------------------------------------------------------- # generate 主流程 # --------------------------------------------------------------------------- @@ -615,8 +1051,80 @@ def _parse_args() -> argparse.Namespace: help="随机种子", ) - # calibrate 子命令（占位，后续任务实现） - subparsers.add_parser("calibrate", help="校准题目难度（待实现）") + # calibrate 子命令 + cal_parser = subparsers.add_parser("calibrate", help="校准生成题与 benchmark 难度一致性") + cal_parser.add_argument( + "--generated-dir", + type=str, + required=True, + help="生成题目目录", + ) + cal_parser.add_argument( + "--benchmark-dir", + type=str, + required=True, + help="benchmark 题目目录", + ) + cal_parser.add_argument( + "--store-dir", + type=str, + required=True, + help="store 根目录", + ) + cal_parser.add_argument( + "--db-path", + type=str, + required=True, + help="校准 SQLite 数据库路径", + ) + cal_parser.add_argument( + "--prompts-dir", + type=str, + required=True, + help="prompt 文件目录", + ) + cal_parser.add_argument( + "--concurrency", + type=int, + required=True, + help="推理并发数", + ) + cal_parser.add_argument( + "--max-steps", + type=int, + required=True, + help="AgentLoop 单题最大步数", + ) + cal_parser.add_argument( + "--skill-mode", + type=str, + required=True, + help="skill 模式 (auto/manual/none)", + ) + cal_parser.add_argument( + "--tolerance", + type=float, + required=True, + help="正确率差值容忍阈值", + ) + cal_parser.add_argument( + "--alpha", + type=float, + required=True, + help="Fisher 检验显著性水平", + ) + cal_parser.add_argument( + "--baseline-db", + type=str, + default=None, + help="基线数据库路径（可选，须与 --baseline-run-id 成对）", + ) + cal_parser.add_argument( + "--baseline-run-id", + type=str, + default=None, + help="基线运行标识（可选，须与 --baseline-db 成对）", + ) return parser.parse_args() @@ -629,8 +1137,7 @@ def main() -> None: if args.command == "generate": asyncio.run(_run_generate(args)) elif args.command == "calibrate": - logger.error("calibrate 子命令尚未实现") - sys.exit(1) + asyncio.run(_run_calibrate(args)) if __name__ == "__main__":