feat: add mirror question generation with canonical distinctness check
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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.
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user