diff --git a/app/question_gen/pipeline_v2.py b/app/question_gen/pipeline_v2.py new file mode 100644 index 0000000..12afcc1 --- /dev/null +++ b/app/question_gen/pipeline_v2.py @@ -0,0 +1,735 @@ +"""v2 出题管线编排 — 重出循环 + 重量抽检 + 并发控制。 + +完整编排流程: +1. 基于视频列表和任务类型分配 slot +2. 每个 slot 执行 generate → postprocess → gates → 重出循环 +3. 去重检测（embedding cosine similarity） +4. 接受题目的随机子集做重量抽检（盲 Agent 试答） +5. 记录全部结果到 QuestionGenStore + +典型调用:: + + result = await run_pipeline_v2( + video_ids=["v1", "v2"], + trees={"v1": tree1, "v2": tree2}, + vlm=vlm_client, + llm=llm_client, + embed_fn=embed_fn, + store=store, + config=config, + task_types=TASK_TYPES, + ) +""" + +from __future__ import annotations + +import asyncio +import random +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING + +import numpy as np +import yaml +from loguru import logger + +from app.question_gen.families import QuestionFamilySpec, get_family_for_slot +from app.question_gen.gates import GateReport, GateResult, GateVerdict, run_gates +from app.question_gen.generator_v2 import CandidateQuestion, generate_one_v2 +from app.question_gen.postprocess import run_postprocess +from app.question_gen.sampler_v2 import sample_material_v2 +from core.types import GeneratedQuestion + +if TYPE_CHECKING: + from collections.abc import Callable + + from app.question_gen.run_store import QuestionGenStore + from app.tree.index import TreeIndex + from core.protocols import LLMProvider, VLMProvider + + +# --------------------------------------------------------------------------- +# 数据类型 +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class SlotAssignment: + """出题 slot 分配 — 一个 slot 对应一道待生成的题目。 + + 属性: + slot_id: slot 唯一标识。 + video_id: 分配到的视频 ID。 + task_type: 任务类型。 + family: 分配的问题家族规格。 + seq: slot 序号（同 task_type 内从 1 开始）。 + """ + + slot_id: str + video_id: str + task_type: str + family: QuestionFamilySpec + seq: int + + +@dataclass +class PipelineResult: + """管线运行结果。 + + 属性: + accepted: 最终通过的题目列表。 + rejected_count: 被拒绝（含重出耗尽）的题目数。 + heavy_sampled: 重量抽检结果列表 — (question_id, difficulty_steps)。 + """ + + accepted: list[GeneratedQuestion] + rejected_count: int + heavy_sampled: list[tuple[str, int]] + + +@dataclass(frozen=True) +class PipelineConfig: + """管线配置。 + + 属性: + family_ratios: 家族名到权重的映射。 + per_type: 每种 task_type 生成的题目数。 + retry_limit: 单 slot 最大重出次数。 + heavy_sample_rate: 重量抽检采样比例 [0.0, 1.0]。 + dedup_threshold: 去重余弦相似度阈值。 + concurrency: 并发 slot 数上限。 + seed: 随机种子。 + output_dir: 输出目录。 + gate_models: 门控模型配置字典。 + heavy_agent_model: 重量抽检使用的模型名。 + """ + + family_ratios: dict[str, float] + per_type: int + retry_limit: int + heavy_sample_rate: float + dedup_threshold: float + concurrency: int + seed: int + output_dir: Path + gate_models: dict[str, str] + heavy_agent_model: str + + +# --------------------------------------------------------------------------- +# 配置加载 +# --------------------------------------------------------------------------- + + +def load_pipeline_config(yaml_path: Path) -> PipelineConfig: + """从 YAML 文件加载管线配置。 + + 读取 question_gen_v2 区段，映射为 PipelineConfig dataclass。 + + 参数: + yaml_path: YAML 配置文件路径。 + + 返回: + PipelineConfig 实例。 + + 异常: + FileNotFoundError: 文件不存在。 + KeyError: 缺少 question_gen_v2 区段。 + """ + if not yaml_path.exists(): + msg = f"配置文件不存在: {yaml_path}" + raise FileNotFoundError(msg) + + with yaml_path.open(encoding="utf-8") as f: + raw = yaml.safe_load(f) + + if "question_gen_v2" not in raw: + msg = f"配置文件缺少 'question_gen_v2' 区段: {yaml_path}" + raise KeyError(msg) + + section = raw["question_gen_v2"] + + # 家族名统一为大写 + raw_ratios = section["family_ratios"] + family_ratios = {k.upper(): float(v) for k, v in raw_ratios.items()} + + return PipelineConfig( + family_ratios=family_ratios, + per_type=int(section["per_type"]), + retry_limit=int(section["retry_limit"]), + heavy_sample_rate=float(section["heavy_sample_rate"]), + dedup_threshold=float(section["dedup_threshold"]), + concurrency=int(section["concurrency"]), + seed=int(section["seed"]), + output_dir=Path(section["output_dir"]), + gate_models=section["gate"], + heavy_agent_model=str(section["heavy_agent_model"]), + ) + + +# --------------------------------------------------------------------------- +# Slot 分配 +# --------------------------------------------------------------------------- + + +def _assign_slots( + video_ids: list[str], + task_types: list[str], + per_type: int, + family_ratios: dict[str, float], + rng: random.Random, +) -> list[SlotAssignment]: + """将出题目标分配为具体 slot 列表。 + + 总 slot 数 = len(task_types) * per_type。 + 在视频间 round-robin 分配，每个 slot 通过 get_family_for_slot 决定家族。 + + 参数: + video_ids: 视频 ID 列表。 + task_types: 任务类型列表。 + per_type: 每种 task_type 的目标题数。 + family_ratios: 家族权重映射。 + rng: 可控随机数生成器。 + + 返回: + SlotAssignment 列表。 + """ + slots: list[SlotAssignment] = [] + global_seq = 0 + + for task_type in task_types: + for i in range(per_type): + video_id = video_ids[i % len(video_ids)] + family = get_family_for_slot(task_type, family_ratios, rng) + global_seq += 1 + slot_id = f"{task_type}_{global_seq:04d}" + slots.append( + SlotAssignment( + slot_id=slot_id, + video_id=video_id, + task_type=task_type, + family=family, + seq=global_seq, + ) + ) + + return slots + + +# --------------------------------------------------------------------------- +# 去重检测 +# --------------------------------------------------------------------------- + + +def _is_duplicate( + question_text: str, + embed_pool: list[np.ndarray], + embed_fn: Callable[[str], np.ndarray], + threshold: float, +) -> bool: + """检测题目文本是否与已接受题库重复。 + + 计算题目 embedding 与 embed_pool 中所有向量的余弦相似度， + 若最大相似度超过阈值则视为重复。 + + 参数: + question_text: 待检测的题目文本。 + embed_pool: 已接受题目的 embedding 向量列表。 + embed_fn: 文本到向量的映射函数。 + threshold: 余弦相似度阈值。 + + 返回: + True 表示重复。 + """ + if not embed_pool: + return False + + query_vec = embed_fn(question_text) + query_norm = np.linalg.norm(query_vec) + if query_norm == 0: + return False + + for pool_vec in embed_pool: + pool_norm = np.linalg.norm(pool_vec) + if pool_norm == 0: + continue + similarity = float(np.dot(query_vec, pool_vec) / (query_norm * pool_norm)) + if similarity > threshold: + return True + + return False + + +# --------------------------------------------------------------------------- +# CandidateQuestion → GeneratedQuestion 转换 +# --------------------------------------------------------------------------- + + +def _to_generated_question(candidate: CandidateQuestion) -> GeneratedQuestion: + """将 CandidateQuestion 转换为 GeneratedQuestion。 + + 参数: + candidate: 门控通过的候选题目。 + + 返回: + GeneratedQuestion 实例（difficulty_steps 初始为 None）。 + """ + return GeneratedQuestion( + question_id=candidate.question_id, + video_id=candidate.video_id, + task_type=candidate.task_type, + question=candidate.question, + options=candidate.options, + answer=candidate.answer, + source_nodes=candidate.source_nodes, + difficulty=candidate.difficulty, + skill_target=candidate.skill_target, + difficulty_steps=None, + ) + + +# --------------------------------------------------------------------------- +# 单 Slot 处理（重出循环） +# --------------------------------------------------------------------------- + + +async def _process_one_slot( + slot: SlotAssignment, + tree: TreeIndex, + vlm: VLMProvider, + llm: LLMProvider, + embed_fn: Callable[[str], np.ndarray], + embed_pool: list[np.ndarray], + store: QuestionGenStore, + config: PipelineConfig, + used_node_ids: set[str], + rng: random.Random, + sem: asyncio.Semaphore, + *, + session_id: str, + run_id: str, +) -> GeneratedQuestion | None: + """处理单个 slot 的完整重出循环。 + + 流程: + 1. 采样素材 + 2. 调用 VLM 生成候选题 + 3. 记录到 store + 4. 后处理（洗牌、检测） + 5. verbatim 检查 + 6. 四门质量检查 + 7. 去重检测 + 8. 通过 → 接受；失败 → 重出（最多 retry_limit 次） + + 参数: + slot: slot 分配信息。 + tree: 视频树索引。 + vlm: VLM 调用端口。 + llm: LLM 调用端口。 + embed_fn: embedding 函数。 + embed_pool: 已接受题目的 embedding 池。 + store: 日志记录器。 + config: 管线配置。 + used_node_ids: 已用节点 ID 集合。 + rng: 随机数生成器。 + sem: 并发信号量。 + session_id: 会话 ID。 + run_id: 批次 ID。 + + 返回: + GeneratedQuestion（通过全部检查）或 None（重出耗尽）。 + """ + async with sem: + prev_reason: str | None = None + + for attempt in range(1, config.retry_limit + 1): + # Phase 1: 采样素材 + try: + material = sample_material_v2( + tree=tree, + family_spec=slot.family, + task_type=slot.task_type, + used_node_ids=used_node_ids, + rng=rng, + ) + except (RuntimeError, KeyError) as e: + logger.warning( + "slot {} 采样失败 (attempt {}/{}): {}", + slot.slot_id, + attempt, + config.retry_limit, + e, + ) + continue + + # Phase 2: 生成候选题 + try: + candidate = await generate_one_v2( + vlm=vlm, + tree=tree, + material=material, + family_spec=slot.family, + task_type=slot.task_type, + seq=slot.seq, + video_id=slot.video_id, + reject_reason=prev_reason, + session_id=session_id, + ) + except (ValueError, FileNotFoundError) as e: + logger.warning( + "slot {} 生成失败 (attempt {}/{}): {}", + slot.slot_id, + attempt, + config.retry_limit, + e, + ) + continue + + # Phase 3: 记录到 store + item_id = f"{slot.slot_id}_att{attempt}_{uuid.uuid4().hex[:8]}" + store.record_item( + item_id=item_id, + run_id=run_id, + slot_id=slot.slot_id, + video_id=slot.video_id, + family=slot.family.name, + task_type=slot.task_type, + skill_target=slot.family.skill_target, + attempt=attempt, + question_text=candidate.question, + ) + + # Phase 4: 后处理 + source_texts = list(material.subtitle_sentences) + pp = run_postprocess( + question_text=candidate.question, + options=candidate.options, + answer=candidate.answer, + source_texts=source_texts, + rng=rng, + ) + + # Phase 5: verbatim 短路检查 + if pp.verbatim_ratio > 0.5: + logger.info( + "slot {} verbatim 过高 ({:.3f}), attempt {}/{}", + slot.slot_id, + pp.verbatim_ratio, + attempt, + config.retry_limit, + ) + prev_reason = f"verbatim_ratio={pp.verbatim_ratio:.3f} exceeds 0.5" + # 仍需记录门控结果 + skip_result = GateResult( + verdict=GateVerdict.SKIP, + reason="skipped due to verbatim", + raw_response="", + ) + verbatim_report = GateReport( + key_verify=GateResult( + verdict=GateVerdict.FAIL, + reason=prev_reason, + raw_response="", + ), + blind_answer=skip_result, + multi_true=skip_result, + leak_test=skip_result, + ) + store.update_gates(item_id, verbatim_report) + continue + + # Phase 6: 四门质量检查 + report = await run_gates( + candidate=candidate, + tree=tree, + llm=llm, + family_spec=slot.family, + postprocess=pp, + session_id=session_id, + ) + store.update_gates(item_id, report) + + if not report.passed: + prev_reason = report.reject_reason + logger.info( + "slot {} 门控失败 (attempt {}/{}): {}", + slot.slot_id, + attempt, + config.retry_limit, + prev_reason, + ) + continue + + # Phase 7: 去重检测 + if _is_duplicate(candidate.question, embed_pool, embed_fn, config.dedup_threshold): + prev_reason = "duplicate detected by embedding similarity" + logger.info( + "slot {} 重复题被拒绝 (attempt {}/{})", + slot.slot_id, + attempt, + config.retry_limit, + ) + continue + + # Phase 8: 通过全部检查 → 接受 + result = _to_generated_question(candidate) + # 将题目 embedding 加入池 + embed_pool.append(embed_fn(candidate.question)) + # 标记使用的节点 + used_node_ids.update(candidate.source_nodes) + + logger.info( + "slot {} 接受: question_id={}, attempt={}/{}", + slot.slot_id, + result.question_id, + attempt, + config.retry_limit, + ) + return result + + # 重出耗尽 + logger.warning("slot {} 重出耗尽 ({} 次)", slot.slot_id, config.retry_limit) + return None + + +# --------------------------------------------------------------------------- +# 重量抽检 +# --------------------------------------------------------------------------- + + +async def _heavy_check_one( + question: GeneratedQuestion, + tree: TreeIndex, + llm: LLMProvider, + *, + session_id: str, +) -> int: + """盲 Agent 试答 — 计算推理步数作为难度指标。 + + 让 LLM 在不访问树的情况下尝试回答问题， + 统计其推理步骤数作为 difficulty_steps。 + + 参数: + question: 待检测的题目。 + tree: 视频树索引（本函数不使用，保留接口一致性）。 + llm: LLM 调用端口。 + session_id: 会话 ID。 + + 返回: + 推理步数（int）。解析失败时返回默认值 1。 + """ + prompt = ( + "You are a blind test agent. Answer this multiple-choice question " + "WITHOUT any video context. Think step by step.\n\n" + f"Question: {question.question}\n" + f"Options:\n" + "\n".join(question.options) + "\n\n" + "Respond with ONLY a JSON object:\n" + '{"steps": [{"thought": "..."}], "answer": "A|B|C|D"}' + ) + + response = await llm.chat( + [{"role": "user", "content": prompt}], + session_id=session_id, + ) + + # 解析步骤数 + try: + import json + + content = response.content.strip() + if "```" in content: + parts = content.split("```") + for part in parts: + stripped = part.strip() + if stripped.startswith("json"): + stripped = stripped[4:].strip() + if stripped.startswith("{"): + content = stripped + break + + data = json.loads(content) + steps = data.get("steps", []) + return max(len(steps), 1) + except (json.JSONDecodeError, TypeError, AttributeError): + logger.debug("heavy_check 响应解析失败，返回默认步数 1") + return 1 + + +# --------------------------------------------------------------------------- +# 管线主入口 +# --------------------------------------------------------------------------- + + +async def run_pipeline_v2( + video_ids: list[str], + trees: dict[str, TreeIndex], + vlm: VLMProvider, + llm: LLMProvider, + embed_fn: Callable[[str], np.ndarray], + store: QuestionGenStore, + config: PipelineConfig, + *, + task_types: list[str] | None = None, + progress: dict[str, str] | None = None, +) -> PipelineResult: + """v2 出题管线主入口 — 编排全部 slot 的生成、检查与抽检。 + + 流程: + 1. 分配 slot + 2. 跳过 progress 中已完成的 slot + 3. 信号量限流并发处理每个 slot + 4. 对接受题目按 heavy_sample_rate 随机抽检 + 5. 更新 store 统计 + 6. 返回 PipelineResult + + 参数: + video_ids: 视频 ID 列表。 + trees: video_id → TreeIndex 映射。 + vlm: VLM 调用端口。 + llm: LLM 调用端口。 + embed_fn: embedding 函数。 + store: 日志记录器。 + config: 管线配置。 + task_types: 任务类型列表（默认使用 12 类标准集）。 + progress: 已完成 slot 映射 {slot_id → "accepted"|"rejected"}。 + + 返回: + PipelineResult 实例。 + """ + # 默认 12 类标准任务类型 + if task_types is None: + task_types = [ + "Action Recognition", + "Action Reasoning", + "Action Prediction", + "Action Sequence", + "Object Recognition", + "Object Reasoning", + "Object Interaction", + "Scene Understanding", + "Event Reasoning", + "Causal Reasoning", + "Temporal Reasoning", + "Spatial Reasoning", + ] + + progress = progress or {} + rng = random.Random(config.seed) + + # Phase 1: 分配 slot + slots = _assign_slots(video_ids, task_types, config.per_type, config.family_ratios, rng) + logger.info( + "管线启动: {} slots, {} 视频, {} 任务类型", len(slots), len(video_ids), len(task_types) + ) + + # Phase 2: 创建 run 记录 + run_id = uuid.uuid4().hex + import subprocess + + try: + git_sha = subprocess.check_output( + ["git", "rev-parse", "--short", "HEAD"], + text=True, + timeout=5, + ).strip() + except (subprocess.SubprocessError, FileNotFoundError): + git_sha = "unknown" + + store.record_run_start(run_id, git_sha, str(config)) + + # Phase 3: 过滤已完成 slot + pending_slots = [s for s in slots if s.slot_id not in progress] + logger.info( + "待处理 slots: {} / {} (已跳过 {})", + len(pending_slots), + len(slots), + len(slots) - len(pending_slots), + ) + + # Phase 4: 并发处理 + sem = asyncio.Semaphore(config.concurrency) + embed_pool: list[np.ndarray] = [] + used_node_ids: set[str] = set() + session_id = f"pipeline_v2_{run_id[:8]}" + + async def _process_wrapper(slot: SlotAssignment) -> GeneratedQuestion | None: + tree = trees.get(slot.video_id) + if tree is None: + logger.warning("slot {} 对应视频 {} 的树不存在，跳过", slot.slot_id, slot.video_id) + return None + return await _process_one_slot( + slot=slot, + tree=tree, + vlm=vlm, + llm=llm, + embed_fn=embed_fn, + embed_pool=embed_pool, + store=store, + config=config, + used_node_ids=used_node_ids, + rng=rng, + sem=sem, + session_id=session_id, + run_id=run_id, + ) + + results = await asyncio.gather(*[_process_wrapper(s) for s in pending_slots]) + + # Phase 5: 统计结果 + accepted: list[GeneratedQuestion] = [r for r in results if r is not None] + rejected_count = len(pending_slots) - len(accepted) + + logger.info( + "管线生成完成: accepted={}, rejected={}", + len(accepted), + rejected_count, + ) + + # Phase 6: 重量抽检 + heavy_sampled: list[tuple[str, int]] = [] + if config.heavy_sample_rate > 0 and accepted: + sample_count = max(1, int(len(accepted) * config.heavy_sample_rate)) + sample_count = min(sample_count, len(accepted)) + heavy_rng = random.Random(config.seed + 1) + sampled_questions = heavy_rng.sample(accepted, sample_count) + + logger.info("重量抽检: {} / {} 题", len(sampled_questions), len(accepted)) + + heavy_tasks = [] + for q in sampled_questions: + tree = trees.get(q.video_id) + if tree is None: + continue + heavy_tasks.append(_heavy_check_one(q, tree, llm, session_id=session_id)) + + heavy_results = await asyncio.gather(*heavy_tasks) + + for q, steps in zip(sampled_questions, heavy_results, strict=True): + heavy_sampled.append((q.question_id, steps)) + # 更新 store + # 找到对应的 item_id（最后一次 attempt 的记录） + cursor = store._conn.execute( + "SELECT item_id FROM question_gen_items WHERE slot_id LIKE ? AND final_status='accepted' LIMIT 1", + (f"%{q.question_id.split('_')[-1]}%",), + ) + row = cursor.fetchone() + if row: + store.update_difficulty(row[0], steps) + + # Phase 7: 更新 run 统计 + from app.question_gen.run_store import RunStats + + stats = RunStats( + total_slots=len(slots), + accepted=len(accepted), + rejected=rejected_count, + heavy_sampled=len(heavy_sampled), + ) + store.record_run_end(run_id, "completed", stats) + + return PipelineResult( + accepted=accepted, + rejected_count=rejected_count, + heavy_sampled=heavy_sampled, + ) diff --git a/tests/integration/test_pipeline_v2.py b/tests/integration/test_pipeline_v2.py new file mode 100644 index 0000000..670087e --- /dev/null +++ b/tests/integration/test_pipeline_v2.py @@ -0,0 +1,650 @@ +"""v2 出题管线集成测试 — 覆盖 slot 分配、重出循环、重量抽检与完整编排。 + +测试策略：使用受控 mock VLM/LLM 返回，验证管线逻辑正确性。 +""" + +from __future__ import annotations + +import asyncio +import json +import random +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from pathlib import Path + +import numpy as np +import pytest + +from app.question_gen.families import ALL_FAMILIES +from app.question_gen.pipeline_v2 import ( + PipelineConfig, + PipelineResult, + SlotAssignment, + _assign_slots, + _process_one_slot, + run_pipeline_v2, +) +from app.tree.index import ( + IndexMeta, + L1Card, + L1Node, + L2Card, + L2Node, + L3Card, + L3Node, + TreeIndex, +) +from core.types import LLMResponse + +# --------------------------------------------------------------------------- +# 测试用 fixtures +# --------------------------------------------------------------------------- + + +def _make_llm_response(content: str) -> LLMResponse: + """构造标准 LLMResponse。""" + return LLMResponse( + content=content, + thinking="", + model="test-model", + provider="test", + prompt_tokens=10, + completion_tokens=20, + latency_ms=100, + ttft_ms=50.0, + max_inter_token_ms=10.0, + cache_hit=False, + call_id="call-001", + ) + + +def _make_gate_pass_response() -> str: + """门控全部通过的 JSON 响应。""" + return json.dumps({"verdict": "pass", "reason": "looks good"}) + + +def _make_gate_fail_response(reason: str = "quality issue") -> str: + """门控失败的 JSON 响应。""" + return json.dumps({"verdict": "fail", "reason": reason}) + + +def _make_candidate_json( + question: str = "What happened next?", + answer: str = "A", +) -> str: + """构造 VLM 返回的候选题 JSON。""" + return json.dumps( + { + "question": question, + "options": [ + "A. The cat jumped", + "B. The dog ran", + "C. Nothing happened", + "D. It rained", + ], + "answer": answer, + "difficulty": "medium", + } + ) + + +def _make_tree() -> TreeIndex: + """构造最小合法三层树。""" + l3_nodes = [ + L3Node( + id=f"vid_L1_000_L2_000_L3_{i:03d}", + card=L3Card( + frame_summary=f"Frame {i} shows activity", + visible_entities=["person", "object"], + ongoing_actions=["walking"], + visible_text=[], + spatial_layout="center", + visual_attributes={}, + subtitle=f"Subtitle sentence {i} with some unique content here", + ), + timestamp=float(i * 2), + frame_path=f"frames/L1_000_L2_000_L3_{i:03d}.jpg", + ) + for i in range(5) + ] + l2 = L2Node( + id="vid_L1_000_L2_000", + card=L2Card( + event_description="A person walks through the park", + entities=["person", "park"], + actions=["walking"], + action_subjects=["person"], + visible_text=[], + spatial_relations="person in center of park", + state_changes=None, + subtitle="Person walking in park doing activities", + ), + children=l3_nodes, + ) + l2_b = L2Node( + id="vid_L1_000_L2_001", + card=L2Card( + event_description="A dog runs across the field", + entities=["dog", "field"], + actions=["running"], + action_subjects=["dog"], + visible_text=[], + spatial_relations="dog in the field", + state_changes=None, + subtitle="Dog running across the field", + ), + children=[ + L3Node( + id=f"vid_L1_000_L2_001_L3_{i:03d}", + card=L3Card( + frame_summary=f"Dog frame {i}", + visible_entities=["dog"], + ongoing_actions=["running"], + visible_text=[], + spatial_layout="wide", + visual_attributes={}, + subtitle=f"Dog subtitle {i}", + ), + timestamp=float(10 + i * 2), + frame_path=f"frames/L1_000_L2_001_L3_{i:03d}.jpg", + ) + for i in range(4) + ], + ) + l1 = L1Node( + id="vid_L1_000", + card=L1Card( + scene_summary="Outdoor activities in a park", + main_setting="outdoor park", + key_entities=["person", "dog"], + main_actions=["walking", "running"], + topic_keywords=["outdoor", "activity"], + visible_text=[], + temporal_flow="sequential activities", + ), + children=[l2, l2_b], + ) + meta = IndexMeta(source_path="test_video.mp4", modality="video") + return TreeIndex(metadata=meta, roots=[l1]) + + +class MockVLM: + """受控 VLM mock — 每次调用返回候选题 JSON。""" + + def __init__(self, responses: list[str] | None = None) -> None: + self._responses = responses or [_make_candidate_json()] + self._call_count = 0 + + async def chat_with_images( + self, + messages: list[dict[str, Any]], + images: list[str | Path], + *, + session_id: str | None = None, + parent_call_id: str | None = None, + ) -> LLMResponse: + idx = min(self._call_count, len(self._responses) - 1) + self._call_count += 1 + return _make_llm_response(self._responses[idx]) + + +class MockLLM: + """受控 LLM mock — 支持配置门控 pass/fail 序列。""" + + def __init__(self, responses: list[str] | None = None) -> None: + self._responses = responses or [_make_gate_pass_response()] + self._call_count = 0 + + async def chat( + self, + messages: list[dict[str, Any]], + *, + session_id: str | None = None, + parent_call_id: str | None = None, + ) -> LLMResponse: + idx = min(self._call_count, len(self._responses) - 1) + self._call_count += 1 + return _make_llm_response(self._responses[idx]) + + +def _mock_embed_fn(text: str) -> np.ndarray: + """确定性 embedding：基于文本 hash 生成向量。""" + rng = np.random.default_rng(hash(text) % (2**32)) + vec = rng.standard_normal(64).astype(np.float32) + return vec / np.linalg.norm(vec) + + +@pytest.fixture +def tree() -> TreeIndex: + return _make_tree() + + +@pytest.fixture +def default_config(tmp_path: Path) -> PipelineConfig: + return PipelineConfig( + family_ratios={ + "RETRIEVAL": 0.30, + "REASONING": 0.25, + "ENUMERATION": 0.20, + "VISUAL": 0.15, + "SPATIAL": 0.10, + }, + per_type=2, + retry_limit=3, + heavy_sample_rate=0.15, + dedup_threshold=0.85, + concurrency=2, + seed=42, + output_dir=tmp_path / "output", + gate_models={ + "blind_answer_model": "gpt-4.1-mini", + "leak_test_model": "gpt-4.1-mini", + "key_verify_model": "gpt-4.1-mini", + "multi_true_model": "gpt-4.1-mini", + }, + heavy_agent_model="gpt-4.1-mini", + ) + + +@pytest.fixture +def store(tmp_path: Path): + from app.question_gen.run_store import QuestionGenStore + + db_path = tmp_path / "test_qgen.db" + s = QuestionGenStore(db_path=db_path) + yield s + s.close() + + +# --------------------------------------------------------------------------- +# TestSlotAssignment +# --------------------------------------------------------------------------- + + +class TestSlotAssignment: + """_assign_slots 单元测试。""" + + def test_per_type_count(self): + """验证生成的 slot 总数 = len(task_types) * per_type。""" + video_ids = ["vid_001", "vid_002"] + task_types = ["Action Recognition", "Object Recognition", "Causal Reasoning"] + per_type = 4 + family_ratios = {"RETRIEVAL": 0.5, "VISUAL": 0.5} + rng = random.Random(42) + + slots = _assign_slots(video_ids, task_types, per_type, family_ratios, rng) + + assert len(slots) == len(task_types) * per_type + + def test_family_distribution(self): + """验证家族分配来自 get_family_for_slot（合法 family 与 task_type 匹配）。""" + video_ids = ["vid_001", "vid_002", "vid_003"] + task_types = ["Action Recognition"] + per_type = 20 + family_ratios = {"RETRIEVAL": 0.5, "VISUAL": 0.5} + rng = random.Random(42) + + slots = _assign_slots(video_ids, task_types, per_type, family_ratios, rng) + + for slot in slots: + assert slot.task_type == "Action Recognition" + # Family 必须是 task_type 合法的家族之一 + family = slot.family + assert slot.task_type in family.legal_task_types + + def test_round_robin_across_videos(self): + """验证 slot 在视频间轮转分配。""" + video_ids = ["vid_A", "vid_B"] + task_types = ["Action Recognition"] + per_type = 4 + family_ratios = {"RETRIEVAL": 1.0} + rng = random.Random(42) + + slots = _assign_slots(video_ids, task_types, per_type, family_ratios, rng) + + video_assignments = [s.video_id for s in slots] + # 应该轮转分配 + assert video_assignments.count("vid_A") == 2 + assert video_assignments.count("vid_B") == 2 + + def test_deterministic_with_seed(self): + """相同 seed 产出相同 slot 序列。""" + video_ids = ["vid_001", "vid_002"] + task_types = ["Action Recognition", "Causal Reasoning"] + per_type = 3 + family_ratios = {"RETRIEVAL": 0.5, "REASONING": 0.5} + + rng1 = random.Random(99) + slots1 = _assign_slots(video_ids, task_types, per_type, family_ratios, rng1) + + rng2 = random.Random(99) + slots2 = _assign_slots(video_ids, task_types, per_type, family_ratios, rng2) + + for s1, s2 in zip(slots1, slots2, strict=True): + assert s1.slot_id == s2.slot_id + assert s1.video_id == s2.video_id + assert s1.family.name == s2.family.name + + +# --------------------------------------------------------------------------- +# TestProcessOneSlot +# --------------------------------------------------------------------------- + + +class TestProcessOneSlot: + """_process_one_slot 集成测试。""" + + @pytest.mark.asyncio + async def test_happy_path(self, tree, default_config, store, tmp_path): + """首次生成即通过门控 → 返回 GeneratedQuestion。""" + vlm = MockVLM() + llm = MockLLM([_make_gate_pass_response()] * 4) + sem = asyncio.Semaphore(2) + used_node_ids: set[str] = set() + rng = random.Random(42) + + family = ALL_FAMILIES[0] # RETRIEVAL + slot = SlotAssignment( + slot_id="slot_001", + video_id="vid_L1_000", + task_type="Action Recognition", + family=family, + seq=1, + ) + + run_id = "test-run-001" + store.record_run_start(run_id, "abc123", "{}") + + result = await _process_one_slot( + slot=slot, + tree=tree, + vlm=vlm, + llm=llm, + embed_fn=_mock_embed_fn, + embed_pool=[], + store=store, + config=default_config, + used_node_ids=used_node_ids, + rng=rng, + sem=sem, + session_id="sess-001", + run_id=run_id, + ) + + assert result is not None + assert result.question_id + assert result.skill_target == family.skill_target + + @pytest.mark.asyncio + async def test_retry_on_fail(self, tree, default_config, store, tmp_path): + """第一次门控失败，第二次通过 → 重出成功。""" + vlm = MockVLM( + [ + _make_candidate_json("First question?"), + _make_candidate_json("Second better question?"), + ] + ) + # 第一轮 4 个门有一个 fail，第二轮 4 个门全 pass + llm_responses = [ + _make_gate_fail_response("answer not grounded"), + _make_gate_pass_response(), + _make_gate_pass_response(), + _make_gate_pass_response(), + # 第二轮 + _make_gate_pass_response(), + _make_gate_pass_response(), + _make_gate_pass_response(), + _make_gate_pass_response(), + ] + llm = MockLLM(llm_responses) + sem = asyncio.Semaphore(2) + used_node_ids: set[str] = set() + rng = random.Random(42) + + family = ALL_FAMILIES[0] # RETRIEVAL + slot = SlotAssignment( + slot_id="slot_002", + video_id="vid_L1_000", + task_type="Action Recognition", + family=family, + seq=2, + ) + + run_id = "test-run-002" + store.record_run_start(run_id, "abc123", "{}") + + result = await _process_one_slot( + slot=slot, + tree=tree, + vlm=vlm, + llm=llm, + embed_fn=_mock_embed_fn, + embed_pool=[], + store=store, + config=default_config, + used_node_ids=used_node_ids, + rng=rng, + sem=sem, + session_id="sess-002", + run_id=run_id, + ) + + assert result is not None + # 确认第二个问题被接受 + assert "Second" in result.question or result.question_id is not None + + @pytest.mark.asyncio + async def test_max_retries_none(self, tree, default_config, store, tmp_path): + """所有重试均失败 → 返回 None。""" + vlm = MockVLM([_make_candidate_json()] * 5) + # 所有门控均失败 + llm = MockLLM([_make_gate_fail_response("always fails")] * 20) + sem = asyncio.Semaphore(2) + used_node_ids: set[str] = set() + rng = random.Random(42) + + family = ALL_FAMILIES[0] # RETRIEVAL + slot = SlotAssignment( + slot_id="slot_003", + video_id="vid_L1_000", + task_type="Action Recognition", + family=family, + seq=3, + ) + + run_id = "test-run-003" + store.record_run_start(run_id, "abc123", "{}") + + result = await _process_one_slot( + slot=slot, + tree=tree, + vlm=vlm, + llm=llm, + embed_fn=_mock_embed_fn, + embed_pool=[], + store=store, + config=default_config, + used_node_ids=used_node_ids, + rng=rng, + sem=sem, + session_id="sess-003", + run_id=run_id, + ) + + assert result is None + + +# --------------------------------------------------------------------------- +# TestPipelineV2 +# --------------------------------------------------------------------------- + + +class TestPipelineV2: + """run_pipeline_v2 完整流程测试。""" + + @pytest.mark.asyncio + async def test_full_flow(self, tree, default_config, store, tmp_path): + """完整管线运行，产出 PipelineResult。""" + vlm = MockVLM([_make_candidate_json(f"Question {i}?") for i in range(50)]) + llm = MockLLM([_make_gate_pass_response()] * 200) + + config = PipelineConfig( + family_ratios=default_config.family_ratios, + per_type=2, + retry_limit=2, + heavy_sample_rate=0.5, # 高比例便于测试 + dedup_threshold=0.85, + concurrency=2, + seed=42, + output_dir=tmp_path / "out", + gate_models=default_config.gate_models, + heavy_agent_model="gpt-4.1-mini", + ) + + # 只用一个 task_type 确保树能满足采样 + result = await run_pipeline_v2( + video_ids=["vid_L1_000"], + trees={"vid_L1_000": tree}, + vlm=vlm, + llm=llm, + embed_fn=_mock_embed_fn, + store=store, + config=config, + task_types=["Action Recognition"], + ) + + assert isinstance(result, PipelineResult) + assert result.rejected_count >= 0 + # 至少一题被接受（VLM 和 LLM 全部正常返回） + assert len(result.accepted) > 0 + + @pytest.mark.asyncio + async def test_progress_resume(self, tree, default_config, store, tmp_path): + """传入 progress dict → 已处理 slot 被跳过。""" + llm = MockLLM([_make_gate_pass_response()] * 100) + + config = PipelineConfig( + family_ratios=default_config.family_ratios, + per_type=2, + retry_limit=2, + heavy_sample_rate=0.0, # 不做 heavy check + dedup_threshold=0.85, + concurrency=2, + seed=42, + output_dir=tmp_path / "out", + gate_models=default_config.gate_models, + heavy_agent_model="gpt-4.1-mini", + ) + + # 用相同 seed 计算 slot_ids（_assign_slots 是确定性的） + rng_preview = random.Random(config.seed) + preview_slots = _assign_slots( + ["vid_L1_000"], + ["Action Recognition"], + config.per_type, + config.family_ratios, + rng_preview, + ) + # 构造 progress 标记所有 slot 已完成 + progress = {s.slot_id: "accepted" for s in preview_slots} + + # 用 progress 跑管线 → 所有 slot 被跳过 + vlm2 = MockVLM([_make_candidate_json()] * 20) + + result2 = await run_pipeline_v2( + video_ids=["vid_L1_000"], + trees={"vid_L1_000": tree}, + vlm=vlm2, + llm=llm, + embed_fn=_mock_embed_fn, + store=store, + config=config, + task_types=["Action Recognition"], + progress=progress, + ) + + # 所有 slot 在 progress 中 → VLM 零调用 + assert vlm2._call_count == 0 + assert len(result2.accepted) == 0 + + @pytest.mark.asyncio + async def test_heavy_check_samples(self, tree, default_config, store, tmp_path): + """heavy_sample_rate > 0 时有题被抽检。""" + # LLM 响应：前面是 gate pass，后面增加 heavy check 的步骤响应 + heavy_response = json.dumps( + { + "steps": [ + {"thought": "step 1"}, + {"thought": "step 2"}, + {"thought": "step 3"}, + ], + "answer": "A", + } + ) + llm = MockLLM([_make_gate_pass_response()] * 100 + [heavy_response] * 20) + vlm = MockVLM([_make_candidate_json(f"Q{i}?") for i in range(20)]) + + config = PipelineConfig( + family_ratios=default_config.family_ratios, + per_type=2, + retry_limit=2, + heavy_sample_rate=1.0, # 100% 抽检 + dedup_threshold=0.85, + concurrency=2, + seed=42, + output_dir=tmp_path / "out", + gate_models=default_config.gate_models, + heavy_agent_model="gpt-4.1-mini", + ) + + result = await run_pipeline_v2( + video_ids=["vid_L1_000"], + trees={"vid_L1_000": tree}, + vlm=vlm, + llm=llm, + embed_fn=_mock_embed_fn, + store=store, + config=config, + task_types=["Action Recognition"], + ) + + # 100% 抽检 → heavy_sampled 等于 accepted 数 + if result.accepted: + assert len(result.heavy_sampled) == len(result.accepted) + + @pytest.mark.asyncio + async def test_store_records_all(self, tree, default_config, store, tmp_path): + """验证 store 中记录了每道题的生成与门控结果。""" + vlm = MockVLM([_make_candidate_json()] * 10) + llm = MockLLM([_make_gate_pass_response()] * 50) + + config = PipelineConfig( + family_ratios=default_config.family_ratios, + per_type=2, + retry_limit=2, + heavy_sample_rate=0.0, + dedup_threshold=0.85, + concurrency=1, + seed=42, + output_dir=tmp_path / "out", + gate_models=default_config.gate_models, + heavy_agent_model="gpt-4.1-mini", + ) + + result = await run_pipeline_v2( + video_ids=["vid_L1_000"], + trees={"vid_L1_000": tree}, + vlm=vlm, + llm=llm, + embed_fn=_mock_embed_fn, + store=store, + config=config, + task_types=["Action Recognition"], + ) + + # 查询 store 中的 items + cursor = store._conn.execute("SELECT COUNT(*) FROM question_gen_items") + item_count = cursor.fetchone()[0] + # 至少有 accepted 数量的记录 + assert item_count >= len(result.accepted)