diff --git a/app/question_gen/synthesizer.py b/app/question_gen/synthesizer.py index fb9ceaf..c070723 100644 --- a/app/question_gen/synthesizer.py +++ b/app/question_gen/synthesizer.py @@ -1,4 +1,4 @@ -"""赛题合成核心逻辑 — 节点采样、prompt 构造、去重。 +"""赛题合成核心逻辑 — 节点采样、prompt 构造、VLM 响应解析、去重。 纯函数为主，异步编排仅 generate_one。 通过 DI 接收 VLMProvider / EmbeddingProvider，不 import adapters/。 @@ -6,6 +6,9 @@ from __future__ import annotations +import contextlib +import json +import re from dataclasses import dataclass from typing import TYPE_CHECKING @@ -13,6 +16,7 @@ if TYPE_CHECKING: import random from app.tree.index import L2Node, L3Node, TreeIndex + from core.types import GeneratedQuestion @dataclass(frozen=True) @@ -486,3 +490,137 @@ def sample_anchor( return _sample_l1_l2(tree, task_type, spec, used_node_ids, rng) else: raise ValueError(f"未知层级: {spec.level}") + + +# --------------------------------------------------------------------------- +# Prompt 构造与 VLM 响应解析 +# --------------------------------------------------------------------------- + +_VALID_ANSWERS = frozenset({"A", "B", "C", "D"}) + + +def build_generation_prompt( + task_type: str, + anchor: AnchorContext, + exemplars: list[GeneratedQuestion], +) -> tuple[list[dict[str, str]], list[str]]: + """组装 VLM 出题 prompt。 + + 构造 OpenAI 格式的 messages 列表和帧图片路径列表， + 供 VLMProvider.chat_with_images 直接消费。 + + 参数: + task_type: 题型名称（如 "Object Recognition"）。 + anchor: 锚节点上下文（card_text, subtitle, distractor_texts, frame_paths）。 + exemplars: 少样本示例列表（可为空）。 + + 返回: + (messages, image_paths) — messages 为 OpenAI 格式消息列表， + image_paths 为帧图片路径列表，直接喂给 VLMProvider.chat_with_images。 + """ + # Phase 1: 构造 system message + system_parts: list[str] = [ + "你是一个视频理解题目生成器。", + f"题型: {task_type}", + "约束:", + "- 题目必须基于提供的节点内容", + "- 干扰选项应来自其他节点的信息", + "- 生成风格应与示例保持一致", + '- 以 JSON 格式返回: {"question": "...", "options": ["A. ...", "B. ...", "C. ...", "D. ..."], "answer": "A/B/C/D"}', + ] + + # Phase 2: 加入 few-shot 示例 + if exemplars: + system_parts.append("\n示例:") + for i, ex in enumerate(exemplars, 1): + system_parts.append(f" 示例 {i}:") + system_parts.append(f" question: {ex.question}") + system_parts.append(f" options: {list(ex.options)}") + system_parts.append(f" answer: {ex.answer}") + + system_content = "\n".join(system_parts) + + # Phase 3: 构造 user message + user_parts: list[str] = [f"节点内容:\n{anchor.card_text}"] + + if anchor.subtitle: + user_parts.append(f"\n字幕:\n{anchor.subtitle}") + + if anchor.distractor_texts: + user_parts.append("\n干扰项来源节点摘要:") + for dt in anchor.distractor_texts: + user_parts.append(f"- {dt}") + + user_content = "\n".join(user_parts) + + messages = [ + {"role": "system", "content": system_content}, + {"role": "user", "content": user_content}, + ] + + return messages, list(anchor.frame_paths) + + +def parse_vlm_response( + raw: str, + video_id: str, + task_type: str, + seq: int, +) -> dict: + """解析 VLM 返回的 JSON → 部分字段字典。 + + 尝试直接解析 JSON；若失败，从 markdown 代码块中提取后重试。 + 校验必需字段、选项数量和答案合法性。 + + 参数: + raw: VLM 原始返回文本。 + video_id: 所属视频标识。 + task_type: 题型名称（用于错误消息）。 + seq: 序列号，用于生成 question_id。 + + 返回: + {"question_id": "gen-{video_id}-{seq:03d}", "question": ..., "options": [...], "answer": ...} + 调用方（generate_one）补齐 source_nodes/difficulty 后构造 GeneratedQuestion。 + + 异常: + ValueError: JSON 解析失败、缺必需字段、options 非 4 项、answer 不在 A-D。 + """ + # Phase 1: 尝试直接解析 JSON + data = None + with contextlib.suppress(json.JSONDecodeError): + data = json.loads(raw) + + # Phase 2: 从 markdown 代码块提取 JSON + if data is None: + match = re.search(r"```(?:json)?\s*\n?(.*?)\n?\s*```", raw, re.DOTALL) + if match: + with contextlib.suppress(json.JSONDecodeError): + data = json.loads(match.group(1)) + + if data is None: + raise ValueError(f"VLM 返回无法解析为 JSON: {raw[:200]}") + + # Phase 3: 校验必需字段 + required = ("question", "options", "answer") + missing = [f for f in required if f not in data] + if missing: + raise ValueError(f"VLM 返回缺少必需字段 {missing}: {raw[:200]}") + + # Phase 4: options 必须恰好 4 项 + options = data["options"] + if not isinstance(options, list) or len(options) != 4: + raise ValueError( + f"options 必须恰好 4 项，实际 {len(options) if isinstance(options, list) else type(options).__name__}: {raw[:200]}" + ) + + # Phase 5: answer 必须是 A-D + answer = data["answer"] + if answer not in _VALID_ANSWERS: + raise ValueError(f"answer 必须是 A/B/C/D 之一，实际 '{answer}': {raw[:200]}") + + return { + "question_id": f"gen-{video_id}-{seq:03d}", + "question": data["question"], + "options": list(options), + "answer": answer, + } diff --git a/tests/unit/test_synthesizer.py b/tests/unit/test_synthesizer.py index eaf1d3b..691a7c4 100644 --- a/tests/unit/test_synthesizer.py +++ b/tests/unit/test_synthesizer.py @@ -1,4 +1,4 @@ -"""synthesizer 模块单元测试 — AnchorContext + 题型映射常量 + sample_anchor。""" +"""synthesizer 模块单元测试 — AnchorContext + 题型映射常量 + sample_anchor + prompt 构造 + VLM 解析。""" from __future__ import annotations @@ -12,9 +12,12 @@ from app.question_gen.synthesizer import ( TASK_TYPE_LEVEL_MAP, AnchorContext, TaskTypeSpec, + build_generation_prompt, + parse_vlm_response, sample_anchor, ) from app.tree.index import TreeIndex +from core.types import GeneratedQuestion ALL_12_TYPES = [ "Object Recognition", @@ -229,3 +232,150 @@ class TestSampleAnchor: # context_fields 包含 spatial_layout，card_text 必须出现该字段名 assert "spatial_layout" in ctx.card_text assert len(ctx.card_text) > 20 + + +# --------------------------------------------------------------------------- +# build_generation_prompt 测试 +# --------------------------------------------------------------------------- + + +class TestBuildGenerationPrompt: + """build_generation_prompt 消息结构与内容测试。""" + + def test_messages_structure(self) -> None: + """带 exemplars 时，system 包含题型和示例，image_paths 来自 anchor。""" + anchor = AnchorContext( + node_id="L3_001", + card_text="A person typing on a laptop", + frame_paths=["store/videos/test/frames/L1_000_L2_000_L3_000.jpg"], + subtitle="Hello world", + distractor_texts=["Another person walking in park"], + ) + exemplars = [ + GeneratedQuestion( + question_id="ex-1", + video_id="v1", + task_type="Object Recognition", + question="What object?", + options=("A. Cat", "B. Dog", "C. Bird", "D. Fish"), + answer="A", + source_nodes=(), + difficulty="medium", + ), + ] + messages, image_paths = build_generation_prompt( + "Object Recognition", + anchor, + exemplars, + ) + assert messages[0]["role"] == "system" + assert "Object Recognition" in messages[0]["content"] + assert any("What object?" in str(m) for m in messages) + assert image_paths == anchor.frame_paths + + def test_distractor_in_user_message(self) -> None: + """干扰项文本应出现在 user message 中。""" + anchor = AnchorContext( + node_id="L2_003", + card_text="Event card text", + frame_paths=["a.jpg", "b.jpg"], + subtitle="", + distractor_texts=["Distractor node summary"], + ) + messages, _ = build_generation_prompt("Action Reasoning", anchor, []) + user_msg = [m for m in messages if m["role"] == "user"][0] + assert "Distractor node summary" in user_msg["content"] + + def test_no_exemplars_no_crash(self) -> None: + """exemplars 为空时不应报错，system 消息中无示例段落。""" + anchor = AnchorContext( + node_id="L3_010", + card_text="Some card text", + frame_paths=["frame.jpg"], + subtitle="", + distractor_texts=[], + ) + messages, image_paths = build_generation_prompt("OCR Problems", anchor, []) + assert len(messages) >= 2 + assert image_paths == ["frame.jpg"] + + def test_subtitle_included_when_non_empty(self) -> None: + """非空 subtitle 应出现在 user message 中。""" + anchor = AnchorContext( + node_id="L3_002", + card_text="Card text here", + frame_paths=["f.jpg"], + subtitle="This is a subtitle line", + distractor_texts=[], + ) + messages, _ = build_generation_prompt("Attribute Perception", anchor, []) + user_msg = [m for m in messages if m["role"] == "user"][0] + assert "This is a subtitle line" in user_msg["content"] + + def test_empty_subtitle_not_in_user_message(self) -> None: + """空 subtitle 不应在 user message 中产生 subtitle 段落。""" + anchor = AnchorContext( + node_id="L3_003", + card_text="Card", + frame_paths=["f.jpg"], + subtitle="", + distractor_texts=[], + ) + messages, _ = build_generation_prompt("OCR Problems", anchor, []) + user_msg = [m for m in messages if m["role"] == "user"][0] + # 不应出现空的 subtitle 标记 + assert "字幕" not in user_msg["content"] + + +# --------------------------------------------------------------------------- +# parse_vlm_response 测试 +# --------------------------------------------------------------------------- + + +class TestParseVlmResponse: + """parse_vlm_response 解析与校验测试。""" + + def test_valid_json(self) -> None: + """合法 JSON 正常解析，question_id 格式正确。""" + raw = '{"question": "What?", "options": ["A. X", "B. Y", "C. Z", "D. W"], "answer": "A"}' + result = parse_vlm_response(raw, "vid1", "Object Recognition", 1) + assert result["question"] == "What?" + assert result["answer"] == "A" + assert len(result["options"]) == 4 + assert result["question_id"] == "gen-vid1-001" + + def test_json_in_code_block(self) -> None: + """从 markdown 代码块中提取 JSON。""" + raw = '```json\n{"question": "Q?", "options": ["A. 1", "B. 2", "C. 3", "D. 4"], "answer": "B"}\n```' + result = parse_vlm_response(raw, "vid1", "Object Recognition", 2) + assert result["question"] == "Q?" + assert result["question_id"] == "gen-vid1-002" + + def test_invalid_json_raises(self) -> None: + """非 JSON 文本应抛出 ValueError。""" + with pytest.raises(ValueError, match="VLM 返回"): + parse_vlm_response("not json", "vid1", "Object Recognition", 1) + + def test_missing_fields_raises(self) -> None: + """缺少必需字段应抛出 ValueError。""" + raw = '{"question": "What?"}' + with pytest.raises(ValueError): + parse_vlm_response(raw, "vid1", "Object Recognition", 1) + + def test_options_must_be_four(self) -> None: + """options 非 4 项应抛出 ValueError。""" + raw = '{"question": "Q?", "options": ["A. X", "B. Y"], "answer": "A"}' + with pytest.raises(ValueError, match="4"): + parse_vlm_response(raw, "vid1", "Object Recognition", 1) + + def test_answer_must_be_abcd(self) -> None: + """answer 不在 A-D 范围应抛出 ValueError。""" + raw = '{"question": "Q?", "options": ["A. X", "B. Y", "C. Z", "D. W"], "answer": "E"}' + with pytest.raises(ValueError, match="A.*D"): + parse_vlm_response(raw, "vid1", "Object Recognition", 1) + + def test_seq_zero_padded(self) -> None: + """seq 应按 3 位零填充格式化到 question_id 中。""" + raw = '{"question": "Q?", "options": ["A. 1", "B. 2", "C. 3", "D. 4"], "answer": "C"}' + result = parse_vlm_response(raw, "video_abc", "Action Reasoning", 42) + assert result["question_id"] == "gen-video_abc-042"