diff --git a/app/question_gen/pipeline_v2.py b/app/question_gen/pipeline_v2.py index 5237dae..4072d6e 100644 --- a/app/question_gen/pipeline_v2.py +++ b/app/question_gen/pipeline_v2.py @@ -46,6 +46,8 @@ if TYPE_CHECKING: from collections.abc import Callable from app.question_gen.run_store import QuestionGenStore + from app.question_gen.sampler_v2 import MaterialContext + from app.question_gen.strategy import TaskTypeStrategy from app.tree.index import TreeIndex from core.protocols import LLMProvider, VLMProvider @@ -341,6 +343,80 @@ def _replace_candidate_options( ) +async def _apply_grounded_selector( + candidate: CandidateQuestion, + strategy: TaskTypeStrategy, + material: MaterialContext, + vlm: VLMProvider, + config: PipelineConfig, + store: QuestionGenStore, + item_id: str, + slot_id: str, + attempt: int, + *, + session_id: str, +) -> tuple[CandidateQuestion | None, str | None]: + """对 AR 候选跑 grounded selector，落观测，返回 (candidate, reject_reason)。 + + 完整拥有 Phase 3.5 的分流控制流，使调用方仅需单一失败分支： + - 非 AR 策略（`uses_grounded_selector` 为假）：直接放行，返回 (candidate, None)。 + - 成功：返回 (重组后的 candidate, None)。 + - selector 异常（`build_grounded_options` 内部抛 ValueError/FileNotFoundError）： + 仅置 reason，不落 rejected，返回 (None, "selector_error: ...")。 + - hard-fail：落 observation + mark_item_rejected，返回 (None, reason)。 + + observation 无论成败都落库（供 EOB 退化观测与调参）。 + + 参数: + candidate: 待重组的候选题目。 + strategy: 题型策略（决定是否走 grounded 路径）。 + material: 采样素材（提供 frame_paths / subtitles）。 + vlm: VLM 调用端口。 + config: 管线配置（提供 selector 三参）。 + store: 日志记录器（落 selector 观测与拒绝态）。 + item_id: 当前 item 的唯一 ID。 + slot_id: slot 标识（日志）。 + attempt: 当前重出轮次（日志）。 + session_id: 遥测会话 ID。 + + 返回: + (candidate, None) 放行；(None, reject_reason) 要求调用方重出。 + """ + if not strategy.uses_grounded_selector: + return candidate, None + + from app.question_gen.distractor_selector import SelectorConfig, build_grounded_options + + correct_text = _extract_correct_text(candidate.options, candidate.answer) + selector_cfg = SelectorConfig( + candidate_pool_size=config.candidate_pool_size, + delta_low=config.selector_delta_low, + delta_high=config.selector_delta_high, + ) + try: + outcome = await build_grounded_options( + vlm=vlm, + question=candidate.question, + correct_text=correct_text, + material=material, + config=selector_cfg, + session_id=session_id, + ) + except (ValueError, FileNotFoundError) as e: + logger.warning("slot {} selector 异常 (attempt {}): {}", slot_id, attempt, e) + return None, f"selector_error: {e}" + + # observation 始终落库（含 hard-fail），供 EOB 退化观测与调参 + store.update_selector_scores(item_id, json.dumps(outcome.observation, ensure_ascii=False)) + if outcome.hard_fail: + reason = "grounded 干扰项不足（selector 硬失败）" + store.mark_item_rejected(item_id, reason) + logger.info("slot {} selector 硬失败 (attempt {})", slot_id, attempt) + return None, reason + + return _replace_candidate_options(candidate, outcome.options, outcome.answer), None + + # --------------------------------------------------------------------------- # 单 Slot 处理（重出循环） # --------------------------------------------------------------------------- @@ -393,8 +469,6 @@ async def _process_one_slot( 返回: GeneratedQuestion（通过全部检查）或 None（重出耗尽）。 """ - resample_video_interval = 1 - async with sem: strategy = get_strategy(slot.task_type) sub_pattern = strategy.select_sub_pattern(rng) @@ -404,8 +478,8 @@ async def _process_one_slot( current_video_id = slot.video_id for attempt in range(1, config.retry_limit + 1): - # 连续失败 resample_video_interval 次后换视频 - if attempt > 1 and (attempt - 1) % resample_video_interval == 0 and all_trees: + # 每次重试都换视频（连续失败即切换） + if attempt > 1 and all_trees: alt_ids = [v for v in all_trees if v != current_video_id] if alt_ids: current_video_id = rng.choice(alt_ids) @@ -488,50 +562,22 @@ async def _process_one_slot( sub_pattern=sub_pattern.name if sub_pattern else None, ) - # Phase 3.5: grounded selector（仅 AR 路径） - if strategy.uses_grounded_selector: - from app.question_gen.distractor_selector import ( - SelectorConfig, - build_grounded_options, - ) - - correct_text = _extract_correct_text(candidate.options, candidate.answer) - selector_cfg = SelectorConfig( - candidate_pool_size=config.candidate_pool_size, - delta_low=config.selector_delta_low, - delta_high=config.selector_delta_high, - ) - try: - outcome = await build_grounded_options( - vlm=vlm, - question=candidate.question, - correct_text=correct_text, - material=material, - config=selector_cfg, - session_id=session_id, - ) - except (ValueError, FileNotFoundError) as e: - logger.warning( - "slot {} selector 异常 (attempt {}): {}", slot.slot_id, attempt, e - ) - prev_reason = f"selector_error: {e}" - continue - - # observation 始终落库（含 hard-fail），供 EOB 退化观测与调参 - store.update_selector_scores( - item_id, json.dumps(outcome.observation, ensure_ascii=False) - ) - - if outcome.hard_fail: - prev_reason = "grounded 干扰项不足（selector 硬失败）" - store.mark_item_rejected(item_id, prev_reason) - logger.info("slot {} selector 硬失败 (attempt {})", slot.slot_id, attempt) - continue - - # 用 grounded 四选项替换候选（frozen → 构造新实例） - candidate = _replace_candidate_options( - candidate, outcome.options, outcome.answer - ) + # Phase 3.5: grounded selector（仅 AR 路径；helper 内部完成分流与落库） + candidate, selector_reason = await _apply_grounded_selector( + candidate, + strategy, + material, + vlm, + config, + store, + item_id, + slot.slot_id, + attempt, + session_id=session_id, + ) + if selector_reason is not None: + prev_reason = selector_reason + continue # Phase 4: 后处理 source_texts = list(material.subtitle_sentences)