feat(question_gen): build_generation_prompt + parse_vlm_response
- prompt 组装：system(角色+题型+约束+few-shot) + user(card+字幕+干扰项) - VLM 响应解析：JSON 直接 + markdown code block 回退，四选一 schema 校验 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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,
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user