From 4af99b700b40958e5cbedce27c92ffaaa029dcc3 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 16:08:13 -0400 Subject: [PATCH] feat: add mirror question generation with canonical distinctness check --- app/question_gen/adversarial_filter.py | 135 +++++++++++++++++- .../question_gen/ar_mirror_question.md | 25 ++++ tests/unit/test_adversarial_mirror.py | 75 ++++++++++ 3 files changed, 234 insertions(+), 1 deletion(-) create mode 100644 store/prompts/question_gen/ar_mirror_question.md create mode 100644 tests/unit/test_adversarial_mirror.py diff --git a/app/question_gen/adversarial_filter.py b/app/question_gen/adversarial_filter.py index ec8eef3..ce07e46 100644 --- a/app/question_gen/adversarial_filter.py +++ b/app/question_gen/adversarial_filter.py @@ -12,14 +12,20 @@ from __future__ import annotations import enum import hashlib import json +from pathlib import Path from typing import TYPE_CHECKING, Protocol +from json_repair import repair_json from loguru import logger +from core.types import GeneratedQuestion + if TYPE_CHECKING: from app.question_gen.adversarial_config import AdversarialFilterConfig from app.question_gen.run_store import QuestionGenStore - from core.types import GeneratedQuestion + from app.question_gen.sampler_v2 import MaterialContext + from app.tree.index import TreeIndex + from core.protocols import VLMProvider _VALID_LETTERS = ("A", "B", "C", "D") @@ -213,3 +219,130 @@ async def run_cheater_gate( len(questions), len(completed), len(todo), len(survivors), ) return survivors + + +_PROMPTS_DIR = Path(__file__).resolve().parent.parent.parent / "store" / "prompts" / "question_gen" + + +def _rebuild_material(tree: TreeIndex, source_nodes: tuple[str, ...]) -> MaterialContext: + """从 source_nodes 重建镜像生成所需素材（字幕 + 帧）。 + + 复用 sampler_v2 的采集辅助；anchor/cross_l2_texts 镜像生成不需要，置空。 + + 参数: + tree: 三层树索引（Phase B 持树）。 + source_nodes: 原题来源节点 ID 元组。 + + 返回: + 仅含 subtitle_sentences / frame_paths 的 MaterialContext。 + """ + from app.question_gen.sampler_v2 import ( + MaterialContext as _MaterialContext, + ) + from app.question_gen.sampler_v2 import ( + _collect_frame_paths, + _collect_subtitle_sentences, + ) + + subtitles = _collect_subtitle_sentences(tree, source_nodes) + frames: list[str] = [] + for nid in source_nodes: + frames.extend(_collect_frame_paths(tree, nid)) + return _MaterialContext( + anchor=None, # 镜像 prompt 不用 anchor + source_nodes=source_nodes, + subtitle_sentences=subtitles, + frame_paths=frames, + cross_l2_texts=[], + ) + + +def _parse_mirror(raw: str) -> dict | None: + """解析 VLM 镜像响应；{"mirror": null} 或解析失败 → None。 + + 参数: + raw: VLM 原始文本响应。 + + 返回: + mirror 字典；null / 非法结构 / 解析失败一律 None。 + """ + content = raw.strip() + if "```" in content: + for part in content.split("```"): + s = part.strip() + if s.startswith("json"): + s = s[4:].strip() + if s.startswith("{"): + content = s + break + try: + data = json.loads(repair_json(content, return_objects=False)) + except (json.JSONDecodeError, TypeError, ValueError): + # 畸形 VLM 响应绝不中断本轮：解析失败 → None（上游按 flip_skipped 处理，设计 §4.2） + return None + if not isinstance(data, dict): + return None + mirror = data.get("mirror") + return mirror if isinstance(mirror, dict) else None + + +async def generate_mirror_question( + question: GeneratedQuestion, + *, + flip_axis: str, + vlm: VLMProvider, + material: MaterialContext, + session_id: str, +) -> GeneratedQuestion | None: + """VLM 生成翻转 flip_axis 的镜像题；正解 canonical 与原题相同则返 None。 + + 参数: + question: 原题。 + flip_axis: 翻转轴（"before/after" | "first/last"）。 + vlm: VLM 端口。 + material: 重建素材（frame_paths / subtitles）。 + session_id: 遥测会话 ID。 + + 返回: + 镜像 GeneratedQuestion（question_id 加 "_mirror" 后缀，不进题库）； + 无法造出有效对（null / 正解相同 / 解析失败）返回 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 = ( + f"## Original Question\n{question.question}\n" + "## Options\n" + "\n".join(question.options) + "\n" + f"## Correct Answer\n{question.answer}\n" + f"## Flip Axis\n{flip_axis}\n" + 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 + ) + 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): + 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) + if p_text is None or q_text is None or p_text.strip() == q_text.strip(): + return None + return mirror_q diff --git a/store/prompts/question_gen/ar_mirror_question.md b/store/prompts/question_gen/ar_mirror_question.md new file mode 100644 index 0000000..1ceb1c1 --- /dev/null +++ b/store/prompts/question_gen/ar_mirror_question.md @@ -0,0 +1,25 @@ +You generate a MIRROR (axis-flipped) version of a video Action Recognition +multiple-choice question, using the SAME video material. + +## Given +- The original question, its four options, and the correct answer. +- The flip axis (e.g. "before/after" or "first/last"). +- Subtitle context and video frames. + +## Rules +- Flip ONLY the given axis: turn "before X" into "after X", "first" into + "last", etc. Everything else (subject, granularity, style) stays identical. +- The mirror question MUST have a genuinely DIFFERENT correct answer than the + original — it asks about the opposite side of the same axis. +- Reuse the SAME candidate option texts where possible, re-shuffled; the letter + of the correct option WILL differ from the original. +- If the axis cannot be flipped into a well-formed question with a distinct + correct answer (e.g. list-style or "cannot determine" answers), output + {"mirror": null}. + +## Output +Respond with ONLY a JSON object: +```json +{"mirror": {"question": "...", "options": ["A. ...", "B. ...", "C. ...", "D. ..."], "answer": "C"}} +``` +Or {"mirror": null} if no valid mirror exists. diff --git a/tests/unit/test_adversarial_mirror.py b/tests/unit/test_adversarial_mirror.py new file mode 100644 index 0000000..8df9274 --- /dev/null +++ b/tests/unit/test_adversarial_mirror.py @@ -0,0 +1,75 @@ +"""镜像生成：成功造出正解相反的镜像；正解相同/生成 null → 返回 None。""" + +import pytest + +from app.question_gen.adversarial_filter import generate_mirror_question +from core.types import GeneratedQuestion, LLMResponse + + +class _FakeVLM: + def __init__(self, content: str): + self._content = content + + async def chat_with_images(self, messages, images, *, session_id=None, parent_call_id=None): + return LLMResponse( + content=self._content, thinking="", model="fake", provider="fake", + prompt_tokens=0, completion_tokens=0, latency_ms=0, + ttft_ms=None, max_inter_token_ms=None, cache_hit=False, call_id="c", + ) + + +def _q(): + return GeneratedQuestion( + question_id="q1", video_id="v1", task_type="Action Recognition", + question="X 之前做了什么？", options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), + answer="A", source_nodes=("n1",), difficulty="hard", + sub_pattern="temporal_reasoning_failure", + ) + + +@pytest.mark.asyncio +async def test_mirror_distinct_correct_ok(): + vlm = _FakeVLM('{"mirror": {"question": "X 之后做了什么？", ' + '"options": ["A. 炒", "B. 蒸", "C. 煮", "D. 炸"], "answer": "A"}}') + mirror = await generate_mirror_question( + _q(), flip_axis="before/after", vlm=vlm, material=_FakeMaterial(), session_id="s", + ) + assert mirror is not None + # 原正解 canonical="蒸"，镜像正解 canonical="炒" → 相异，有效 + assert mirror.answer == "A" + assert mirror.options[0] == "A. 炒" + + +@pytest.mark.asyncio +async def test_mirror_same_correct_rejected(): + # 镜像正解 canonical 仍是"蒸" → 造不出有效对 → None + vlm = _FakeVLM('{"mirror": {"question": "X 之后？", ' + '"options": ["A. 蒸", "B. 炒", "C. 煮", "D. 炸"], "answer": "A"}}') + mirror = await generate_mirror_question( + _q(), flip_axis="before/after", vlm=vlm, material=_FakeMaterial(), session_id="s", + ) + assert mirror is None + + +@pytest.mark.asyncio +async def test_mirror_null_returns_none(): + vlm = _FakeVLM('{"mirror": null}') + mirror = await generate_mirror_question( + _q(), flip_axis="before/after", vlm=vlm, material=_FakeMaterial(), session_id="s", + ) + assert mirror is None + + +@pytest.mark.asyncio +async def test_mirror_malformed_response_returns_none(): + # 畸形 VLM 响应（连 json_repair 都救不回）不得抛异常中断本轮，须返 None + vlm = _FakeVLM("对不起，我无法完成这个请求。") + mirror = await generate_mirror_question( + _q(), flip_axis="before/after", vlm=vlm, material=_FakeMaterial(), session_id="s", + ) + assert mirror is None + + +class _FakeMaterial: + subtitle_sentences = ["先炒后蒸"] + frame_paths = ["/f1.jpg"]