diff --git a/app/question_gen/synthesizer.py b/app/question_gen/synthesizer.py index c070723..fbbb83e 100644 --- a/app/question_gen/synthesizer.py +++ b/app/question_gen/synthesizer.py @@ -12,10 +12,15 @@ import re from dataclasses import dataclass from typing import TYPE_CHECKING +import numpy as np +from loguru import logger + if TYPE_CHECKING: import random + from collections.abc import Callable from app.tree.index import L2Node, L3Node, TreeIndex + from core.protocols import VLMProvider from core.types import GeneratedQuestion @@ -624,3 +629,132 @@ def parse_vlm_response( "options": list(options), "answer": answer, } + + +# --------------------------------------------------------------------------- +# Embedding 去重 +# --------------------------------------------------------------------------- + + +def is_duplicate( + question_text: str, + pool_embeddings: np.ndarray, + embed_fn: Callable[[str | list[str]], np.ndarray], + threshold: float, +) -> bool: + """embedding 去重判定。 + + 参数: + question_text: 待检查的题目文本。 + pool_embeddings: 已有题目的 embedding 矩阵 [N, D]（L2 归一化）。 + embed_fn: 文本嵌入函数，返回 [N, D] ndarray（L2 归一化）。 + threshold: 余弦相似度阈值。 + + 返回: + True 表示与池中某题重复。空池永远返回 False。 + """ + if pool_embeddings.shape[0] == 0: + return False + + query = embed_fn(question_text) # [1, D] + query = query.squeeze(0) # [D] + similarities = pool_embeddings @ query # [N] + return bool(np.max(similarities) >= threshold) + + +# --------------------------------------------------------------------------- +# 单题生成 +# --------------------------------------------------------------------------- + + +async def generate_one( + vlm: VLMProvider, + embed_fn: Callable[[str | list[str]], np.ndarray], + tree: TreeIndex, + video_id: str, + task_type: str, + seq: int, + *, + exemplars: list[GeneratedQuestion], + used_node_ids: set[str], + max_retries: int, + similarity_threshold: float, + rng: random.Random, + session_id: str, +) -> GeneratedQuestion | None: + """生成单道候选题（不含去重——去重在调用方汇总点原子执行）。 + + 循环最多 max_retries 次尝试生成。每次尝试： + 1. 采样锚节点 + 2. 构造 prompt + 3. 调用 VLM + 4. 解析响应 + 5. 构造 GeneratedQuestion + + 返回 None 表示耗尽重试。 + + 参数: + vlm: VLM 调用端口。 + embed_fn: 文本嵌入函数（本函数内未使用，由调用方统一去重）。 + tree: 三层树索引。 + video_id: 所属视频标识。 + task_type: 12 种 Video-MME 题型之一。 + seq: 序列号，用于生成 question_id。 + exemplars: 少样本示例列表。 + used_node_ids: 已用节点 ID 集合。 + max_retries: 最大重试次数。 + similarity_threshold: 余弦相似度阈值（本函数内未使用）。 + rng: 可控随机数生成器。 + session_id: 会话 ID（传递给 VLM 遥测）。 + + 返回: + GeneratedQuestion 实例，或 None（耗尽重试）。 + """ + from core.types import GeneratedQuestion as _GeneratedQuestion + + for attempt in range(max_retries): + try: + # Phase 1: 采样锚节点 + anchor = sample_anchor(tree, task_type, used_node_ids, rng) + + # Phase 2: 构造 prompt + messages, images = build_generation_prompt(task_type, anchor, exemplars) + + # Phase 3: 调用 VLM + response = await vlm.chat_with_images( + messages, + images, + session_id=session_id, + ) + + # Phase 4: 解析响应 + parsed = parse_vlm_response(response.content, video_id, task_type, seq) + + # Phase 5: 构造 GeneratedQuestion + return _GeneratedQuestion( + question_id=parsed["question_id"], + video_id=video_id, + task_type=task_type, + question=parsed["question"], + options=tuple(parsed["options"]), + answer=parsed["answer"], + source_nodes=(anchor.node_id,), + difficulty="medium", + ) + except (ValueError, KeyError) as exc: + logger.warning( + "generate_one 尝试 {}/{} 失败 ({}): {}", + attempt + 1, + max_retries, + task_type, + exc, + ) + continue + + logger.warning( + "generate_one 耗尽 {} 次重试 (video={}, task_type={})", + max_retries, + video_id, + task_type, + ) + return None diff --git a/tests/unit/test_synthesizer.py b/tests/unit/test_synthesizer.py index 691a7c4..f564ae4 100644 --- a/tests/unit/test_synthesizer.py +++ b/tests/unit/test_synthesizer.py @@ -1,11 +1,13 @@ -"""synthesizer 模块单元测试 — AnchorContext + 题型映射常量 + sample_anchor + prompt 构造 + VLM 解析。""" +"""synthesizer 模块单元测试 — AnchorContext + 题型映射常量 + sample_anchor + prompt 构造 + VLM 解析 + 去重 + 单题生成。""" from __future__ import annotations import dataclasses import random from pathlib import Path +from unittest.mock import AsyncMock, MagicMock +import numpy as np import pytest from app.question_gen.synthesizer import ( @@ -13,6 +15,8 @@ from app.question_gen.synthesizer import ( AnchorContext, TaskTypeSpec, build_generation_prompt, + generate_one, + is_duplicate, parse_vlm_response, sample_anchor, ) @@ -379,3 +383,119 @@ class TestParseVlmResponse: 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" + + +# --------------------------------------------------------------------------- +# is_duplicate 测试 +# --------------------------------------------------------------------------- + + +class TestIsDuplicate: + """is_duplicate embedding 去重判定测试。""" + + @staticmethod + def _fake_embed(texts: str | list[str]) -> np.ndarray: + """确定性 + L2 归一化的 fake embedding。""" + if isinstance(texts, str): + texts = [texts] + vecs = [] + for t in texts: + rs = np.random.RandomState(hash(t) % 2**31) + v = rs.randn(4).astype(np.float32) + v /= np.linalg.norm(v) + vecs.append(v) + return np.array(vecs, dtype=np.float32) + + def test_empty_pool_never_duplicate(self) -> None: + """空池始终返回 False。""" + pool = np.zeros((0, 4), dtype=np.float32) + assert is_duplicate("anything", pool, self._fake_embed, 0.85) is False + + def test_identical_text_is_duplicate(self) -> None: + """相同文本的 embedding 与自身余弦相似度为 1，必定判重。""" + text = "What is happening in the video?" + emb = self._fake_embed(text) + pool = emb.copy() + assert is_duplicate(text, pool, self._fake_embed, 0.85) is True + + def test_different_text_not_duplicate(self) -> None: + """极高阈值下，不同文本不判重。""" + pool_texts = ["aaa", "bbb", "ccc", "ddd", "eee"] + pool = self._fake_embed(pool_texts) + assert is_duplicate("completely unique text xyz", pool, self._fake_embed, 0.99) is False + + +# --------------------------------------------------------------------------- +# generate_one 测试 +# --------------------------------------------------------------------------- + + +class TestGenerateOne: + """generate_one 单题异步生成测试。""" + + @staticmethod + def _load_test_tree() -> tuple[TreeIndex, str]: + """加载真实测试树。""" + videos_dir = Path("store/videos") + first_vid = sorted(videos_dir.iterdir())[0] + return TreeIndex.load_json(str(first_vid / "tree.json")), first_vid.name + + @pytest.mark.asyncio + async def test_success_path(self) -> None: + """mock VLM 返回合法 JSON，应成功生成 GeneratedQuestion。""" + vlm = AsyncMock() + vlm.chat_with_images.return_value = MagicMock( + content='{"question":"Q?","options":["A. 1","B. 2","C. 3","D. 4"],"answer":"A"}', + ) + + def embed_fn(t: str | list[str]) -> np.ndarray: + shape = (1, 4) if isinstance(t, str) else (len(t), 4) + return np.zeros(shape, dtype=np.float32) + + tree, vid = self._load_test_tree() + result = await generate_one( + vlm=vlm, + embed_fn=embed_fn, + tree=tree, + video_id=vid, + task_type="Object Recognition", + seq=1, + exemplars=[], + used_node_ids=set(), + max_retries=3, + similarity_threshold=0.85, + rng=random.Random(42), + session_id="test", + ) + assert result is not None + assert result.question_id == f"gen-{vid}-001" + assert result.task_type == "Object Recognition" + assert result.source_nodes # non-empty + assert result.difficulty == "medium" + + @pytest.mark.asyncio + async def test_all_retries_exhausted_returns_none(self) -> None: + """VLM 始终返回无效 JSON，耗尽重试后返回 None。""" + vlm = AsyncMock() + vlm.chat_with_images.return_value = MagicMock(content="invalid") + + def embed_fn(t: str | list[str]) -> np.ndarray: + return np.zeros((1, 4), dtype=np.float32) + + tree, vid = self._load_test_tree() + result = await generate_one( + vlm=vlm, + embed_fn=embed_fn, + tree=tree, + video_id=vid, + task_type="Object Recognition", + seq=1, + exemplars=[], + used_node_ids=set(), + max_retries=2, + similarity_threshold=0.85, + rng=random.Random(42), + session_id="test", + ) + assert result is None + assert vlm.chat_with_images.call_count == 2