diff --git a/app/question_gen/adversarial_filter.py b/app/question_gen/adversarial_filter.py index ce07e46..133b725 100644 --- a/app/question_gen/adversarial_filter.py +++ b/app/question_gen/adversarial_filter.py @@ -244,10 +244,12 @@ def _rebuild_material(tree: TreeIndex, source_nodes: tuple[str, ...]) -> Materia _collect_subtitle_sentences, ) - subtitles = _collect_subtitle_sentences(tree, source_nodes) + # 保序去重：source_nodes 若同时含父子节点，两者子树重叠会重复采集，去冗余再喂 VLM。 + subtitles = list(dict.fromkeys(_collect_subtitle_sentences(tree, source_nodes))) frames: list[str] = [] for nid in source_nodes: frames.extend(_collect_frame_paths(tree, nid)) + frames = list(dict.fromkeys(frames)) return _MaterialContext( anchor=None, # 镜像 prompt 不用 anchor source_nodes=source_nodes, @@ -286,6 +288,34 @@ def _parse_mirror(raw: str) -> dict | None: return mirror if isinstance(mirror, dict) else None +def _existing_frames(frame_paths: list[str]) -> list[str]: + """过滤掉磁盘上不存在的帧路径，保序返回有效帧列表（防失效帧喂给 VLM）。""" + return [p for p in frame_paths if Path(p).exists()] + + +def _build_mirror_question( + question: GeneratedQuestion, mirror: dict +) -> GeneratedQuestion | None: + """从解析出的 mirror dict 构造镜像题；字段缺失/类型错误 → None。""" + try: + options = tuple(str(o) for o in mirror["options"]) + answer = str(mirror["answer"]).strip().upper() + m_question = str(mirror["question"]) + except (KeyError, TypeError): + return None + return GeneratedQuestion( + question_id=f"{question.question_id}_mirror", + video_id=question.video_id, + task_type=question.task_type, + question=m_question, + options=options, + answer=answer, + source_nodes=question.source_nodes, + difficulty=question.difficulty, + sub_pattern=question.sub_pattern, + ) + + async def generate_mirror_question( question: GeneratedQuestion, *, @@ -305,8 +335,15 @@ async def generate_mirror_question( 返回: 镜像 GeneratedQuestion（question_id 加 "_mirror" 后缀，不进题库）； - 无法造出有效对（null / 正解相同 / 解析失败）返回 None。 + 素材为空 / null / 正解相同 / 解析失败均返回 None（上层记 flip_skipped）。 """ + # 素材防御：无字幕 AND 无有效帧 → 不调 VLM，直接 flip_skipped（避免空素材空转 VLM）。 + valid_frames = _existing_frames(list(material.frame_paths)) + if not material.subtitle_sentences and not valid_frames: + logger.warning( + "镜像素材为空（source_nodes 可能缺失或帧失效），跳过: {}", question.question_id + ) + return None system = (_PROMPTS_DIR / "ar_mirror_question.md").read_text(encoding="utf-8") subs = "\n".join(f" - {s}" for s in material.subtitle_sentences) user = ( @@ -317,32 +354,20 @@ async def generate_mirror_question( f"## Subtitles\n{subs}\n" ) messages = [{"role": "system", "content": system}, {"role": "user", "content": user}] - resp = await vlm.chat_with_images( - messages, list(material.frame_paths), session_id=session_id - ) + resp = await vlm.chat_with_images(messages, valid_frames, session_id=session_id) mirror = _parse_mirror(resp.content) if mirror is None: return None - try: - options = tuple(str(o) for o in mirror["options"]) - answer = str(mirror["answer"]).strip().upper() - m_question = str(mirror["question"]) - except (KeyError, TypeError): + mirror_q = _build_mirror_question(question, mirror) + if mirror_q is None: return None - mirror_q = GeneratedQuestion( - question_id=f"{question.question_id}_mirror", - video_id=question.video_id, - task_type=question.task_type, - question=m_question, - options=options, - answer=answer, - source_nodes=question.source_nodes, - difficulty=question.difficulty, - sub_pattern=question.sub_pattern, - ) # 镜像正解字面校验：canonical(P) 必须 != canonical(Q)（按选项文本比较，非字母） p_text = canonical_answer_text(question.options, question.answer) - q_text = canonical_answer_text(mirror_q.options, answer) + q_text = canonical_answer_text(mirror_q.options, mirror_q.answer) if p_text is None or q_text is None or p_text.strip() == q_text.strip(): + logger.info( + "镜像正解与原题正解无区分（或无效），跳过翻转门: question_id={}", + question.question_id, + ) return None return mirror_q diff --git a/tests/unit/test_adversarial_mirror.py b/tests/unit/test_adversarial_mirror.py index 8df9274..b6dc434 100644 --- a/tests/unit/test_adversarial_mirror.py +++ b/tests/unit/test_adversarial_mirror.py @@ -2,7 +2,20 @@ import pytest -from app.question_gen.adversarial_filter import generate_mirror_question +from app.question_gen.adversarial_filter import ( + _rebuild_material, + generate_mirror_question, +) +from app.tree.index import ( + IndexMeta, + L1Card, + L1Node, + L2Card, + L2Node, + L3Card, + L3Node, + TreeIndex, +) from core.types import GeneratedQuestion, LLMResponse @@ -73,3 +86,90 @@ async def test_mirror_malformed_response_returns_none(): class _FakeMaterial: subtitle_sentences = ["先炒后蒸"] frame_paths = ["/f1.jpg"] + + +class _NoCallVLM: + """素材为空时被误调用即失败——断言空素材绝不触达 VLM。""" + + async def chat_with_images(self, *args, **kwargs): + raise AssertionError("素材为空时不应调用 VLM") + + +class _EmptyMaterial: + subtitle_sentences: list[str] = [] + frame_paths = ["/does/not/exist/frame_x.jpg"] + + +@pytest.mark.asyncio +async def test_mirror_empty_material_skips_vlm(): + # 无字幕 AND 帧全不存在 → 直接返 None，且绝不调用 VLM + mirror = await generate_mirror_question( + _q(), flip_axis="before/after", vlm=_NoCallVLM(), + material=_EmptyMaterial(), session_id="s", + ) + assert mirror is None + + +def _tiny_tree() -> tuple[TreeIndex, str, str]: + """构建 1×L1→1×L2→2×L3 的最小真实树；返回 (tree, l2_id, first_l3_id)。 + + L2 自带字幕，两个 L3 各带字幕与帧路径，便于验证父子重叠去重。 + """ + l3_nodes = [ + L3Node( + id=f"l2a_l3_{i}", + card=L3Card( + frame_summary=f"帧{i}", + visible_entities=[], + ongoing_actions=[], + visible_text=[], + spatial_layout="居中", + visual_attributes={}, + subtitle=f"字幕{i}", + ), + frame_path=f"frames/l2a_l3_{i}.jpg", + ) + for i in range(2) + ] + l2 = L2Node( + id="l2a", + card=L2Card( + event_description="事件A", + entities=[], + actions=[], + action_subjects=[], + visible_text=[], + spatial_relations="并列", + state_changes=None, + subtitle="L2字幕", + ), + children=l3_nodes, + ) + l1 = L1Node( + id="l1a", + card=L1Card( + scene_summary="场景A", + main_setting="室内", + key_entities=[], + main_actions=[], + topic_keywords=[], + visible_text=[], + temporal_flow="顺序", + ), + children=[l2], + ) + tree = TreeIndex( + metadata=IndexMeta(source_path="/t.mp4", modality="video"), roots=[l1] + ) + return tree, "l2a", "l2a_l3_0" + + +def test_rebuild_material_dedup_parent_child(): + # source_nodes 同含父 L2 与子 L3（子树重叠）→ 素材须保序去重、无冗余 + tree, l2_id, l3_id = _tiny_tree() + material = _rebuild_material(tree, (l2_id, l3_id)) + assert len(material.subtitle_sentences) == len(set(material.subtitle_sentences)) + assert len(material.frame_paths) == len(set(material.frame_paths)) + # L2字幕 + 两条 L3 字幕 = 3 条；两帧 = 2 帧（去重后） + assert material.subtitle_sentences == ["L2字幕", "字幕0", "字幕1"] + assert material.frame_paths == ["frames/l2a_l3_0.jpg", "frames/l2a_l3_1.jpg"]