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) <noreply@anthropic.com>
2026-07-14 06:05:23 -04:00
parent b9616de21e
commit 36c712defa
4 changed files with 190 additions and 164 deletions
@@ -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)
    )
@@ -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
@@ -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(
@@ -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)