From 36c712defaf4921e710807546724949ef572eb7f Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 06:05:23 -0400 Subject: [PATCH] feat(pipeline): replace QuestionFamilySpec with TaskTypeStrategy - SlotAssignment: remove family field, strategy looked up at process time - PipelineConfig: remove family_ratios field - _assign_slots: remove family_ratios and rng params (pure deterministic) - _process_one_slot: use get_strategy() for sampling, generation, gates - Add sub_pattern support (level/constraint override, instruction injection) - Add strategy.extra_gates() check after standard gates - load_pipeline_config: stop reading family_ratios from YAML - Update tools/generate_questions.py seed override and dry-run log - Update all integration tests to match new API Co-Authored-By: Claude Opus 4.6 (1M context) --- app/question_gen/pipeline_v2.py | 73 ++++++---- tests/integration/test_cli_generate_v2.py | 12 -- tests/integration/test_pipeline_v2.py | 111 +++++---------- tools/generate_questions.py | 158 +++++++++++++++------- 4 files changed, 190 insertions(+), 164 deletions(-) diff --git a/app/question_gen/pipeline_v2.py b/app/question_gen/pipeline_v2.py index d164bdb..cc015bf 100644 --- a/app/question_gen/pipeline_v2.py +++ b/app/question_gen/pipeline_v2.py @@ -34,11 +34,11 @@ import numpy as np import yaml from loguru import logger -from app.question_gen.families import QuestionFamilySpec, get_family_for_slot from app.question_gen.gates import GateReport, GateResult, GateVerdict, run_gates from app.question_gen.generator_v2 import CandidateQuestion, generate_one_v2 from app.question_gen.postprocess import run_postprocess -from app.question_gen.sampler_v2 import _TASK_TYPE_TO_LEVEL, sample_material_v2 +from app.question_gen.sampler_v2 import sample_material_v2 +from app.question_gen.strategy import get_strategy from core.types import GeneratedQuestion if TYPE_CHECKING: @@ -62,14 +62,12 @@ class SlotAssignment: slot_id: slot 唯一标识。 video_id: 分配到的视频 ID。 task_type: 任务类型。 - family: 分配的问题家族规格。 - seq: slot 序号（同 task_type 内从 1 开始）。 + seq: slot 序号。 """ slot_id: str video_id: str task_type: str - family: QuestionFamilySpec seq: int @@ -93,7 +91,6 @@ class PipelineConfig: """管线配置。 属性: - family_ratios: 家族名到权重的映射。 per_type: 每种 task_type 生成的题目数。 retry_limit: 单 slot 最大重出次数。 heavy_sample_rate: 重量抽检采样比例 [0.0, 1.0]。 @@ -103,7 +100,6 @@ class PipelineConfig: output_dir: 输出目录。 """ - family_ratios: dict[str, float] per_type: int retry_limit: int heavy_sample_rate: float @@ -146,12 +142,7 @@ def load_pipeline_config(yaml_path: Path) -> PipelineConfig: section = raw["question_gen_v2"] - # 家族名统一为大写 - raw_ratios = section["family_ratios"] - family_ratios = {k.upper(): float(v) for k, v in raw_ratios.items()} - return PipelineConfig( - family_ratios=family_ratios, per_type=int(section["per_type"]), retry_limit=int(section["retry_limit"]), heavy_sample_rate=float(section["heavy_sample_rate"]), @@ -171,20 +162,16 @@ def _assign_slots( video_ids: list[str], task_types: list[str], per_type: int, - family_ratios: dict[str, float], - rng: random.Random, ) -> list[SlotAssignment]: """将出题目标分配为具体 slot 列表。 总 slot 数 = len(task_types) * per_type。 - 在视频间 round-robin 分配，每个 slot 通过 get_family_for_slot 决定家族。 + 在视频间 round-robin 分配。不再选择 family — strategy 在处理时查找。 参数: video_ids: 视频 ID 列表。 task_types: 任务类型列表。 per_type: 每种 task_type 的目标题数。 - family_ratios: 家族权重映射。 - rng: 可控随机数生成器。 返回: SlotAssignment 列表。 @@ -195,7 +182,6 @@ def _assign_slots( for task_type in task_types: for i in range(per_type): video_id = video_ids[i % len(video_ids)] - family = get_family_for_slot(task_type, family_ratios, rng) global_seq += 1 slot_id = f"{task_type}_{global_seq:04d}" slots.append( @@ -203,7 +189,6 @@ def _assign_slots( slot_id=slot_id, video_id=video_id, task_type=task_type, - family=family, seq=global_seq, ) ) @@ -348,6 +333,9 @@ async def _process_one_slot( _RESAMPLE_VIDEO_INTERVAL = 1 async with sem: + strategy = get_strategy(slot.task_type) + sub_pattern = strategy.select_sub_pattern(rng) + prev_reason: str | None = None current_tree = tree current_video_id = slot.video_id @@ -366,15 +354,25 @@ async def _process_one_slot( current_video_id, ) - # Phase 1: 采样素材 + # Phase 1: 采样素材（sub_pattern 可覆盖 level 和 constraint） + level = ( + sub_pattern.sampling_level_override + if sub_pattern and sub_pattern.sampling_level_override is not None + else strategy.sampling_level + ) + constraint = ( + sub_pattern.constraint_override + if sub_pattern and sub_pattern.constraint_override is not None + else strategy.sampling_constraint + ) try: material = sample_material_v2( tree=current_tree, task_type=slot.task_type, used_node_ids=used_node_ids, rng=rng, - level=_TASK_TYPE_TO_LEVEL[slot.task_type], - constraint=slot.family.sampling, + level=level, + constraint=constraint, ) except (RuntimeError, KeyError) as e: logger.warning( @@ -392,11 +390,14 @@ async def _process_one_slot( vlm=vlm, tree=current_tree, material=material, - family_spec=slot.family, task_type=slot.task_type, seq=slot.seq, video_id=current_video_id, + prompt_template=strategy.prompt_template, + strategy_name=strategy.strategy_name, + skill_target=strategy.skill_target, reject_reason=prev_reason, + sub_pattern_instruction=sub_pattern.instruction if sub_pattern else None, session_id=session_id, ) except (ValueError, FileNotFoundError, OSError, Exception) as e: @@ -416,11 +417,12 @@ async def _process_one_slot( run_id=run_id, slot_id=slot.slot_id, video_id=current_video_id, - family=slot.family.name, + family=strategy.strategy_name, task_type=slot.task_type, - skill_target=slot.family.skill_target, + skill_target=strategy.skill_target, attempt=attempt, question_text=candidate.question, + sub_pattern=sub_pattern.name if sub_pattern else None, ) # Phase 4: 后处理 @@ -468,7 +470,7 @@ async def _process_one_slot( candidate=candidate, tree=tree, llm=llm, - family_spec=slot.family, + leak_probe_template=strategy.leak_probe_template, postprocess=pp, vlm=vlm, session_id=session_id, @@ -485,6 +487,21 @@ async def _process_one_slot( continue store.update_gates(item_id, report) + # 题型专属额外 gate + extra_results = strategy.extra_gates(candidate) + if any(r.verdict == GateVerdict.FAIL for r in extra_results): + prev_reason = "; ".join( + r.reason for r in extra_results if r.verdict == GateVerdict.FAIL + ) + logger.info( + "slot {} 额外 gate 失败 (attempt {}/{}): {}", + slot.slot_id, + attempt, + config.retry_limit, + prev_reason, + ) + continue + if not report.passed: prev_reason = report.reject_reason logger.info( @@ -511,7 +528,7 @@ async def _process_one_slot( # Phase 8: 通过全部检查 → 接受（使用洗牌后的选项和答案） result = _to_generated_question( candidate, - family=slot.family.name, + family=strategy.strategy_name, options=pp.options, answer=pp.answer, ) @@ -762,7 +779,7 @@ async def run_pipeline_v2( rng = random.Random(config.seed) # Phase 1: 分配 slot + 创建 run 记录 - slots = _assign_slots(video_ids, task_types, config.per_type, config.family_ratios, rng) + slots = _assign_slots(video_ids, task_types, config.per_type) logger.info( "管线启动: {} slots, {} 视频, {} 任务类型", len(slots), len(video_ids), len(task_types) ) diff --git a/tests/integration/test_cli_generate_v2.py b/tests/integration/test_cli_generate_v2.py index 7d80da9..f449423 100644 --- a/tests/integration/test_cli_generate_v2.py +++ b/tests/integration/test_cli_generate_v2.py @@ -45,21 +45,9 @@ class TestCLIGenerateV2: config_path.write_text( """\ question_gen_v2: - family_ratios: - retrieval: 0.30 - reasoning: 0.25 - enumeration: 0.20 - visual: 0.15 - spatial: 0.10 - gate: - blind_answer_model: "mock" - leak_test_model: "mock" - key_verify_model: "mock" - multi_true_model: "mock" dedup_threshold: 0.85 retry_limit: 3 heavy_sample_rate: 0.15 - heavy_agent_model: "mock" output_dir: "{output_dir}" per_type: 2 concurrency: 2 diff --git a/tests/integration/test_pipeline_v2.py b/tests/integration/test_pipeline_v2.py index 670087e..bb1221b 100644 --- a/tests/integration/test_pipeline_v2.py +++ b/tests/integration/test_pipeline_v2.py @@ -16,7 +16,6 @@ if TYPE_CHECKING: import numpy as np import pytest -from app.question_gen.families import ALL_FAMILIES from app.question_gen.pipeline_v2 import ( PipelineConfig, PipelineResult, @@ -170,11 +169,21 @@ def _make_tree() -> TreeIndex: class MockVLM: - """受控 VLM mock — 每次调用返回候选题 JSON。""" + """受控 VLM mock — 自动区分生成请求和门控请求。 - def __init__(self, responses: list[str] | None = None) -> None: + 检测 prompt 中是否包含 'verdict' 关键词判断请求类型： + - 门控请求 → 返回 gate_response（默认 pass） + - 生成请求 → 按序返回 candidate JSON + """ + + def __init__( + self, + responses: list[str] | None = None, + gate_response: str | None = None, + ) -> None: self._responses = responses or [_make_candidate_json()] - self._call_count = 0 + self._gate_response = gate_response or _make_gate_pass_response() + self._gen_count = 0 async def chat_with_images( self, @@ -184,8 +193,11 @@ class MockVLM: session_id: str | None = None, parent_call_id: str | None = None, ) -> LLMResponse: - idx = min(self._call_count, len(self._responses) - 1) - self._call_count += 1 + prompt_text = str(messages) + if "verdict" in prompt_text.lower(): + return _make_llm_response(self._gate_response) + idx = min(self._gen_count, len(self._responses) - 1) + self._gen_count += 1 return _make_llm_response(self._responses[idx]) @@ -223,13 +235,6 @@ def tree() -> TreeIndex: @pytest.fixture def default_config(tmp_path: Path) -> PipelineConfig: return PipelineConfig( - family_ratios={ - "RETRIEVAL": 0.30, - "REASONING": 0.25, - "ENUMERATION": 0.20, - "VISUAL": 0.15, - "SPATIAL": 0.10, - }, per_type=2, retry_limit=3, heavy_sample_rate=0.15, @@ -237,13 +242,6 @@ def default_config(tmp_path: Path) -> PipelineConfig: concurrency=2, seed=42, output_dir=tmp_path / "output", - gate_models={ - "blind_answer_model": "gpt-4.1-mini", - "leak_test_model": "gpt-4.1-mini", - "key_verify_model": "gpt-4.1-mini", - "multi_true_model": "gpt-4.1-mini", - }, - heavy_agent_model="gpt-4.1-mini", ) @@ -268,63 +266,39 @@ class TestSlotAssignment: def test_per_type_count(self): """验证生成的 slot 总数 = len(task_types) * per_type。""" video_ids = ["vid_001", "vid_002"] - task_types = ["Action Recognition", "Object Recognition", "Causal Reasoning"] + task_types = ["Action Recognition", "Object Recognition", "Counting Problem"] per_type = 4 - family_ratios = {"RETRIEVAL": 0.5, "VISUAL": 0.5} - rng = random.Random(42) - slots = _assign_slots(video_ids, task_types, per_type, family_ratios, rng) + slots = _assign_slots(video_ids, task_types, per_type) assert len(slots) == len(task_types) * per_type - def test_family_distribution(self): - """验证家族分配来自 get_family_for_slot（合法 family 与 task_type 匹配）。""" - video_ids = ["vid_001", "vid_002", "vid_003"] - task_types = ["Action Recognition"] - per_type = 20 - family_ratios = {"RETRIEVAL": 0.5, "VISUAL": 0.5} - rng = random.Random(42) - - slots = _assign_slots(video_ids, task_types, per_type, family_ratios, rng) - - for slot in slots: - assert slot.task_type == "Action Recognition" - # Family 必须是 task_type 合法的家族之一 - family = slot.family - assert slot.task_type in family.legal_task_types - def test_round_robin_across_videos(self): """验证 slot 在视频间轮转分配。""" video_ids = ["vid_A", "vid_B"] task_types = ["Action Recognition"] per_type = 4 - family_ratios = {"RETRIEVAL": 1.0} - rng = random.Random(42) - slots = _assign_slots(video_ids, task_types, per_type, family_ratios, rng) + slots = _assign_slots(video_ids, task_types, per_type) video_assignments = [s.video_id for s in slots] # 应该轮转分配 assert video_assignments.count("vid_A") == 2 assert video_assignments.count("vid_B") == 2 - def test_deterministic_with_seed(self): - """相同 seed 产出相同 slot 序列。""" + def test_deterministic(self): + """相同参数产出相同 slot 序列（无随机性）。""" video_ids = ["vid_001", "vid_002"] - task_types = ["Action Recognition", "Causal Reasoning"] + task_types = ["Action Recognition", "Temporal Reasoning"] per_type = 3 - family_ratios = {"RETRIEVAL": 0.5, "REASONING": 0.5} - rng1 = random.Random(99) - slots1 = _assign_slots(video_ids, task_types, per_type, family_ratios, rng1) - - rng2 = random.Random(99) - slots2 = _assign_slots(video_ids, task_types, per_type, family_ratios, rng2) + slots1 = _assign_slots(video_ids, task_types, per_type) + slots2 = _assign_slots(video_ids, task_types, per_type) for s1, s2 in zip(slots1, slots2, strict=True): assert s1.slot_id == s2.slot_id assert s1.video_id == s2.video_id - assert s1.family.name == s2.family.name + assert s1.task_type == s2.task_type # --------------------------------------------------------------------------- @@ -344,12 +318,10 @@ class TestProcessOneSlot: used_node_ids: set[str] = set() rng = random.Random(42) - family = ALL_FAMILIES[0] # RETRIEVAL slot = SlotAssignment( slot_id="slot_001", video_id="vid_L1_000", task_type="Action Recognition", - family=family, seq=1, ) @@ -374,7 +346,11 @@ class TestProcessOneSlot: assert result is not None assert result.question_id - assert result.skill_target == family.skill_target + # strategy 根据 task_type 决定 skill_target + from app.question_gen.strategy import get_strategy + + expected_strategy = get_strategy("Action Recognition") + assert result.skill_target == expected_strategy.skill_target @pytest.mark.asyncio async def test_retry_on_fail(self, tree, default_config, store, tmp_path): @@ -402,12 +378,10 @@ class TestProcessOneSlot: used_node_ids: set[str] = set() rng = random.Random(42) - family = ALL_FAMILIES[0] # RETRIEVAL slot = SlotAssignment( slot_id="slot_002", video_id="vid_L1_000", task_type="Action Recognition", - family=family, seq=2, ) @@ -444,12 +418,10 @@ class TestProcessOneSlot: used_node_ids: set[str] = set() rng = random.Random(42) - family = ALL_FAMILIES[0] # RETRIEVAL slot = SlotAssignment( slot_id="slot_003", video_id="vid_L1_000", task_type="Action Recognition", - family=family, seq=3, ) @@ -490,7 +462,6 @@ class TestPipelineV2: llm = MockLLM([_make_gate_pass_response()] * 200) config = PipelineConfig( - family_ratios=default_config.family_ratios, per_type=2, retry_limit=2, heavy_sample_rate=0.5, # 高比例便于测试 @@ -498,8 +469,6 @@ class TestPipelineV2: concurrency=2, seed=42, output_dir=tmp_path / "out", - gate_models=default_config.gate_models, - heavy_agent_model="gpt-4.1-mini", ) # 只用一个 task_type 确保树能满足采样 @@ -525,7 +494,6 @@ class TestPipelineV2: llm = MockLLM([_make_gate_pass_response()] * 100) config = PipelineConfig( - family_ratios=default_config.family_ratios, per_type=2, retry_limit=2, heavy_sample_rate=0.0, # 不做 heavy check @@ -533,18 +501,13 @@ class TestPipelineV2: concurrency=2, seed=42, output_dir=tmp_path / "out", - gate_models=default_config.gate_models, - heavy_agent_model="gpt-4.1-mini", ) - # 用相同 seed 计算 slot_ids（_assign_slots 是确定性的） - rng_preview = random.Random(config.seed) + # _assign_slots 现在是确定性的（无随机性） preview_slots = _assign_slots( ["vid_L1_000"], ["Action Recognition"], config.per_type, - config.family_ratios, - rng_preview, ) # 构造 progress 标记所有 slot 已完成 progress = {s.slot_id: "accepted" for s in preview_slots} @@ -565,7 +528,7 @@ class TestPipelineV2: ) # 所有 slot 在 progress 中 → VLM 零调用 - assert vlm2._call_count == 0 + assert vlm2._gen_count == 0 assert len(result2.accepted) == 0 @pytest.mark.asyncio @@ -586,7 +549,6 @@ class TestPipelineV2: vlm = MockVLM([_make_candidate_json(f"Q{i}?") for i in range(20)]) config = PipelineConfig( - family_ratios=default_config.family_ratios, per_type=2, retry_limit=2, heavy_sample_rate=1.0, # 100% 抽检 @@ -594,8 +556,6 @@ class TestPipelineV2: concurrency=2, seed=42, output_dir=tmp_path / "out", - gate_models=default_config.gate_models, - heavy_agent_model="gpt-4.1-mini", ) result = await run_pipeline_v2( @@ -620,7 +580,6 @@ class TestPipelineV2: llm = MockLLM([_make_gate_pass_response()] * 50) config = PipelineConfig( - family_ratios=default_config.family_ratios, per_type=2, retry_limit=2, heavy_sample_rate=0.0, @@ -628,8 +587,6 @@ class TestPipelineV2: concurrency=1, seed=42, output_dir=tmp_path / "out", - gate_models=default_config.gate_models, - heavy_agent_model="gpt-4.1-mini", ) result = await run_pipeline_v2( diff --git a/tools/generate_questions.py b/tools/generate_questions.py index bed1e74..b739d1e 100644 --- a/tools/generate_questions.py +++ b/tools/generate_questions.py @@ -349,6 +349,8 @@ def _append_to_json(output_dir: Path, question: GeneratedQuestion) -> None: "answer": question.answer, "source_nodes": list(question.source_nodes), "difficulty": question.difficulty, + "family": question.family, + "skill_target": question.skill_target, } existing.append(entry) @@ -850,6 +852,13 @@ def _add_generate_v2_parser(subparsers: argparse._SubParsersAction) -> None: default=None, help="随机种子（覆盖配置文件中的 seed）", ) + p.add_argument( + "--task-types", + type=str, + nargs="+", + default=None, + help="只生成指定的 task_type（默认全部 12 类）", + ) p.add_argument( "--dry-run", action="store_true", @@ -889,7 +898,6 @@ async def _run_generate_v2(args: argparse.Namespace) -> None: # Phase 2: 覆盖 seed if args.seed is not None: config = PipelineConfig( - family_ratios=config.family_ratios, per_type=config.per_type, retry_limit=config.retry_limit, heavy_sample_rate=config.heavy_sample_rate, @@ -897,8 +905,6 @@ async def _run_generate_v2(args: argparse.Namespace) -> None: concurrency=config.concurrency, seed=args.seed, output_dir=config.output_dir, - gate_models=config.gate_models, - heavy_agent_model=config.heavy_agent_model, ) logger.info("seed 覆盖为: {}", args.seed) @@ -918,20 +924,21 @@ async def _run_generate_v2(args: argparse.Namespace) -> None: logger.info("发现 {} 个视频: {}", len(video_ids), video_ids[:5]) # Phase 4: dry-run 模式 - task_types = [ + all_task_types = [ "Action Recognition", "Action Reasoning", - "Action Prediction", - "Action Sequence", + "Attribute Perception", + "Counting Problem", + "Information Synopsis", "Object Recognition", "Object Reasoning", - "Object Interaction", - "Scene Understanding", - "Event Reasoning", - "Causal Reasoning", - "Temporal Reasoning", + "OCR Problems", + "Spatial Perception", "Spatial Reasoning", + "Temporal Perception", + "Temporal Reasoning", ] + task_types = args.task_types if args.task_types else all_task_types total_slots = len(task_types) * config.per_type if args.dry_run: logger.info("[dry-run] 管线配置摘要:") @@ -941,7 +948,6 @@ async def _run_generate_v2(args: argparse.Namespace) -> None: logger.info("[dry-run] 总 slot 数: {}", total_slots) logger.info("[dry-run] 并发: {}", config.concurrency) logger.info("[dry-run] 去重阈值: {}", config.dedup_threshold) - logger.info("[dry-run] 家族权重: {}", config.family_ratios) logger.info("[dry-run] 退出（不调用 LLM/VLM）") return @@ -955,6 +961,8 @@ async def _run_generate_v2(args: argparse.Namespace) -> None: telemetry = SQLiteTelemetryRecorder(str(PROJECT_ROOT / "logs" / "generate_v2_telemetry.db")) breaker_threshold = int(os.getenv("LLM_CIRCUIT_BREAKER_THRESHOLD", "5")) + # 并发 slot 共享客户端，熔断阈值需按并发度缩放（与 build_trees 一致） + breaker_threshold = max(breaker_threshold, config.concurrency * 2) 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")) @@ -985,11 +993,11 @@ async def _run_generate_v2(args: argparse.Namespace) -> None: ) vlm = GovernedVLMClient(vlm_base) - # LLM 客户端（门控用） + # LLM 客户端（门控用，复用 JUDGE_LLM 配置） llm = GovernedLLMClient( - model=os.environ.get("LLM_MODEL", "gpt-4.1-mini"), - base_url=os.environ["LLM_BASE_URL"], - api_key=os.environ["LLM_API_KEY"], + model=os.environ.get("JUDGE_LLM_MODEL", "gpt-4.1-mini"), + base_url=os.environ["JUDGE_LLM_BASE_URL"], + api_key=os.environ["JUDGE_LLM_API_KEY"], provider="openai", thinking=False, breaker=_make_breaker(), @@ -1041,6 +1049,16 @@ async def _run_generate_v2(args: argparse.Namespace) -> None: if not trees: logger.error("所有视频的树加载均失败，无法继续") sys.exit(1) + + # 将相对帧路径解析为绝对路径（与 v1 generate 一致） + for vid, tree in trees.items(): + video_dir = videos_dir / vid + for l1 in tree.roots: + for l2 in l1.children: + for l3 in l2.children: + if l3.frame_path and not Path(l3.frame_path).is_absolute(): + l3.frame_path = str(video_dir / l3.frame_path) + logger.info("成功加载 {} / {} 棵视频树", len(trees), len(video_ids)) # Phase 7: 初始化 QuestionGenStore @@ -1050,10 +1068,81 @@ async def _run_generate_v2(args: argparse.Namespace) -> None: db_path.parent.mkdir(parents=True, exist_ok=True) store = QuestionGenStore(str(db_path)) - # Phase 8: 加载断点续跑进度 + # Phase 8: 加载断点续跑进度（DB 进度 + 已有 JSON 中已满的 slot） progress: dict[str, str] = store.load_progress() - # Phase 9: 运行管线 + output_path_check = config.output_dir / "accepted_questions.json" + if output_path_check.exists(): + try: + existing_qs = json.loads(output_path_check.read_text(encoding="utf-8")) + filled_slots = 0 + for q in existing_qs: + qid = q.get("question_id", "") + vid = q.get("video_id", "") + if not qid or not vid or not qid.startswith(vid + "_"): + logger.warning("跳过格式异常的已有题目: question_id={}", qid) + continue + slot_id = qid[len(vid) + 1 :] + if slot_id not in progress: + progress[slot_id] = "accepted" + filled_slots += 1 + logger.info( + "从已有 JSON 补充 progress: {} 个 slot 标记为已完成 (已有 {} 题)", + filled_slots, + len(existing_qs), + ) + except (json.JSONDecodeError, OSError, ValueError) as exc: + logger.warning("已有 JSON 读取失败，跳过 progress 补充: {}", exc) + + # Phase 9: 准备实时持久化回调（v1 式逐题追加，崩溃最多丢一题） + output_dir = config.output_dir + output_dir.mkdir(parents=True, exist_ok=True) + output_path = output_dir / "accepted_questions.json" + _accepted_texts: set[str] = set() + + if output_path.exists(): + try: + _existing = json.loads(output_path.read_text(encoding="utf-8")) + _accepted_texts = {q["question"] for q in _existing} + except (json.JSONDecodeError, OSError): + pass + + def _on_accept(q: GeneratedQuestion) -> None: + """每接受一题立即追加到 JSON（原子写入）。""" + if q.question in _accepted_texts: + return + _accepted_texts.add(q.question) + + existing: list[dict] = [] + if output_path.exists(): + try: + existing = json.loads(output_path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + existing = [] + + existing.append( + { + "question_id": q.question_id, + "video_id": q.video_id, + "task_type": q.task_type, + "question": q.question, + "options": list(q.options), + "answer": q.answer, + "source_nodes": list(q.source_nodes), + "difficulty": q.difficulty, + "family": q.family, + "skill_target": q.skill_target, + } + ) + + tmp_path = output_path.with_suffix(".tmp") + tmp_path.write_text( + json.dumps(existing, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + os.replace(str(tmp_path), str(output_path)) + + # Phase 10: 运行管线（on_accept 实时持久化每道接受的题） active_video_ids = [vid for vid in video_ids if vid in trees] result = await run_pipeline_v2( video_ids=active_video_ids, @@ -1065,38 +1154,13 @@ async def _run_generate_v2(args: argparse.Namespace) -> None: config=config, task_types=task_types, progress=progress, + on_accept=_on_accept, ) - # Phase 10: 保存输出 - output_dir = config.output_dir - output_dir.mkdir(parents=True, exist_ok=True) - output_path = output_dir / "accepted_questions.json" - - accepted_data = [] - for q in result.accepted: - accepted_data.append( - { - "question_id": q.question_id, - "video_id": q.video_id, - "task_type": q.task_type, - "question": q.question, - "options": list(q.options), - "answer": q.answer, - "source_nodes": list(q.source_nodes), - "difficulty": q.difficulty, - "skill_target": q.skill_target, - } - ) - - output_path.write_text( - json.dumps(accepted_data, ensure_ascii=False, indent=2), - encoding="utf-8", - ) - logger.info( - "输出已保存: {} ({} 题)", - output_path, - len(accepted_data), + final_count = ( + len(json.loads(output_path.read_text(encoding="utf-8"))) if output_path.exists() else 0 ) + logger.info("输出已保存: {} (总计 {} 题)", output_path, final_count) # 统计报告 logger.info("=" * 60)