"""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 json 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.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 app.question_gen.strategy import get_strategy from core.types import GeneratedQuestion if TYPE_CHECKING: from collections.abc import Callable from app.question_gen.run_store import QuestionGenStore from app.question_gen.sampler_v2 import MaterialContext from app.question_gen.strategy import TaskTypeStrategy 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: 任务类型。 seq: slot 序号。 """ slot_id: str video_id: str task_type: str 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: """管线配置。 属性: per_type: 每种 task_type 生成的题目数。 retry_limit: 单 slot 最大重出次数。 heavy_sample_rate: 重量抽检采样比例 [0.0, 1.0]。 dedup_threshold: 去重余弦相似度阈值。 concurrency: 并发 slot 数上限。 seed: 随机种子。 output_dir: 输出目录。 candidate_pool_size: grounded selector 首轮候选干扰项数 N。 selector_delta_low: 干扰项视觉分与正解的最小差（区间上界）。 selector_delta_high: 干扰项视觉分与正解的最大差（区间下界）。 """ per_type: int retry_limit: int heavy_sample_rate: float dedup_threshold: float concurrency: int seed: int output_dir: Path candidate_pool_size: int = 24 selector_delta_low: float = 0.05 selector_delta_high: float = 0.35 # --------------------------------------------------------------------------- # 配置加载 # --------------------------------------------------------------------------- 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"] return PipelineConfig( 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"]), candidate_pool_size=int(section.get("candidate_pool_size", 24)), selector_delta_low=float(section.get("selector_delta_low", 0.05)), selector_delta_high=float(section.get("selector_delta_high", 0.35)), ) # --------------------------------------------------------------------------- # Slot 分配 # --------------------------------------------------------------------------- def _assign_slots( video_ids: list[str], task_types: list[str], per_type: int, seq_offset: int = 0, ) -> list[SlotAssignment]: """将出题目标分配为具体 slot 列表。 总 slot 数 = len(task_types) * per_type。 在视频间 round-robin 分配。不再选择 family — strategy 在处理时查找。 参数: video_ids: 视频 ID 列表。 task_types: 任务类型列表。 per_type: 每种 task_type 的目标题数。 seq_offset: 全局序号起始偏移（补生成续编，默认 0）。 返回: SlotAssignment 列表。 """ slots: list[SlotAssignment] = [] global_seq = seq_offset for task_type in task_types: for i in range(per_type): video_id = video_ids[i % len(video_ids)] 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, 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).flatten() 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, *, family: str, options: tuple[str, ...] | None = None, answer: str | None = None, sub_pattern: str | None = None, ) -> GeneratedQuestion: """将 CandidateQuestion 转换为 GeneratedQuestion。 参数: candidate: 门控通过的候选题目。 family: 问题家族名称（如 "RETRIEVAL"）。 options: 洗牌后的选项元组（若为 None 则使用 candidate 原始选项）。 answer: 重映射后的答案字母（若为 None 则使用 candidate 原始答案）。 sub_pattern: 出题子模式标识（AR 特化策略使用，None 表示无）。 返回: 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=options if options is not None else candidate.options, answer=answer if answer is not None else candidate.answer, source_nodes=candidate.source_nodes, difficulty=candidate.difficulty, family=family, skill_target=candidate.skill_target, difficulty_steps=None, sub_pattern=sub_pattern, ) def _extract_correct_text(options: tuple[str, ...], answer: str) -> str: """从四选项中取正解文本（去掉 "X. " 字母前缀）。 参数: options: 选项元组，格式 ("A. ...", "B. ...", ...)。 answer: 正解字母（大小写不敏感）。 返回: 正解选项去前缀后的文本。 异常: ValueError: answer 对应索引超出选项范围。 """ idx = ord(answer.strip().upper()) - ord("A") if not 0 <= idx < len(options): msg = f"answer '{answer}' 超出选项范围 (n={len(options)})" raise ValueError(msg) opt = options[idx] prefix = f"{answer.strip().upper()}. " return opt[len(prefix) :] if opt.startswith(prefix) else opt def _replace_candidate_options( candidate: CandidateQuestion, options: tuple[str, ...], answer: str ) -> CandidateQuestion: """用 selector 重组的选项/答案替换候选（CandidateQuestion frozen）。 参数: candidate: 原候选题目。 options: grounded selector 重组后的四选项。 answer: 重组后的正解字母（恒 "A"）。 返回: 仅替换 options/answer、其余字段照搬的新 CandidateQuestion。 """ return CandidateQuestion( question_id=candidate.question_id, video_id=candidate.video_id, task_type=candidate.task_type, skill_target=candidate.skill_target, question=candidate.question, options=options, answer=answer, source_nodes=candidate.source_nodes, difficulty=candidate.difficulty, subtitle_sentences=candidate.subtitle_sentences, frame_paths=candidate.frame_paths, ) async def _apply_grounded_selector( candidate: CandidateQuestion, strategy: TaskTypeStrategy, material: MaterialContext, vlm: VLMProvider, config: PipelineConfig, store: QuestionGenStore, item_id: str, slot_id: str, attempt: int, *, session_id: str, ) -> tuple[CandidateQuestion | None, str | None]: """对 AR 候选跑 grounded selector，落观测，返回 (candidate, reject_reason)。 完整拥有 Phase 3.5 的分流控制流，使调用方仅需单一失败分支： - 非 AR 策略（`uses_grounded_selector` 为假）：直接放行，返回 (candidate, None)。 - 成功：返回 (重组后的 candidate, None)。 - selector 异常（`build_grounded_options` 内部抛 ValueError/FileNotFoundError）： 仅置 reason，不落 rejected，返回 (None, "selector_error: ...")。 - hard-fail：落 observation + mark_item_rejected，返回 (None, reason)。 observation 无论成败都落库（供 EOB 退化观测与调参）。 参数: candidate: 待重组的候选题目。 strategy: 题型策略（决定是否走 grounded 路径）。 material: 采样素材（提供 frame_paths / subtitles）。 vlm: VLM 调用端口。 config: 管线配置（提供 selector 三参）。 store: 日志记录器（落 selector 观测与拒绝态）。 item_id: 当前 item 的唯一 ID。 slot_id: slot 标识（日志）。 attempt: 当前重出轮次（日志）。 session_id: 遥测会话 ID。 返回: (candidate, None) 放行；(None, reject_reason) 要求调用方重出。 """ if not strategy.uses_grounded_selector: return candidate, None from app.question_gen.distractor_selector import SelectorConfig, build_grounded_options correct_text = _extract_correct_text(candidate.options, candidate.answer) selector_cfg = SelectorConfig( candidate_pool_size=config.candidate_pool_size, delta_low=config.selector_delta_low, delta_high=config.selector_delta_high, ) try: outcome = await build_grounded_options( vlm=vlm, question=candidate.question, correct_text=correct_text, material=material, config=selector_cfg, session_id=session_id, ) except (ValueError, FileNotFoundError, OSError, Exception) as e: # noqa: BLE001 # slot 级隔离：selector 会多次调 VLM，任何异常（含 VLM 网络/超时/熔断 # CircuitOpenError/StreamLivenessTimeout 等非 ValueError 类型）都降级为 # selector_error 重出该 slot，绝不穿透 asyncio.gather 崩掉整批生成。 # 与相邻 Phase 2 生成失败的宽捕获口径一致；异常路径无 observation 可落， # 但仍 mark_item_rejected（该 attempt 的 item 是死记录，重出新建 item_id）。 reason = f"selector_error: {e}" logger.warning("slot {} selector 异常 (attempt {}): {}", slot_id, attempt, e) store.mark_item_rejected(item_id, reason) return None, reason # observation 始终落库（含 hard-fail），供 EOB 退化观测与调参 store.update_selector_scores(item_id, json.dumps(outcome.observation, ensure_ascii=False)) if outcome.hard_fail: reason = "grounded 干扰项不足（selector 硬失败）" store.mark_item_rejected(item_id, reason) logger.info("slot {} selector 硬失败 (attempt {})", slot_id, attempt) return None, reason return _replace_candidate_options(candidate, outcome.options, outcome.answer), 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, all_trees: dict[str, TreeIndex] | None = None, ) -> 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: strategy = get_strategy(slot.task_type) sub_pattern = strategy.select_sub_pattern(rng) prev_reason: str | None = None current_tree = tree current_video_id = slot.video_id for attempt in range(1, config.retry_limit + 1): # 每次重试都换视频（连续失败即切换） if attempt > 1 and all_trees: alt_ids = [v for v in all_trees if v != current_video_id] if alt_ids: current_video_id = rng.choice(alt_ids) current_tree = all_trees[current_video_id] logger.info( "slot {} 连续 {} 次失败，换视频 {} 重试", slot.slot_id, attempt - 1, current_video_id, ) # Phase 1: 采样素材（sub_pattern 可覆盖 level 和 constraint） level = ( sub_pattern.sampling_level_override if sub_pattern and sub_pattern.sampling_level_override is not None else strategy.sampling_level ) constraint = ( sub_pattern.constraint_override if sub_pattern and sub_pattern.constraint_override is not None else strategy.sampling_constraint ) try: material = sample_material_v2( tree=current_tree, task_type=slot.task_type, used_node_ids=used_node_ids, rng=rng, level=level, constraint=constraint, ) 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=current_tree, material=material, task_type=slot.task_type, seq=slot.seq, video_id=current_video_id, prompt_template=strategy.prompt_template, strategy_name=strategy.strategy_name, skill_target=strategy.skill_target, reject_reason=prev_reason, sub_pattern_instruction=sub_pattern.instruction if sub_pattern else None, session_id=session_id, ) except (ValueError, FileNotFoundError, OSError, Exception) 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=current_video_id, family=strategy.strategy_name, task_type=slot.task_type, skill_target=strategy.skill_target, attempt=attempt, question_text=candidate.question, sub_pattern=sub_pattern.name if sub_pattern else None, ) # Phase 3.5: grounded selector（仅 AR 路径；helper 内部完成分流与落库） candidate, selector_reason = await _apply_grounded_selector( candidate, strategy, material, vlm, config, store, item_id, slot.slot_id, attempt, session_id=session_id, ) if selector_reason is not None: prev_reason = selector_reason continue # 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: 四门质量检查（key_verify 使用 VLM 看帧+文本） try: report = await run_gates( candidate=candidate, tree=current_tree, llm=llm, leak_probe_template=strategy.leak_probe_template, postprocess=pp, vlm=vlm, session_id=session_id, ) except Exception as e: logger.warning( "slot {} 门控调用异常 (attempt {}/{}): {}", slot.slot_id, attempt, config.retry_limit, e, ) prev_reason = f"gate_error: {e}" continue 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 # 题型专属额外 gate（仅标准门通过后才执行） extra_results = strategy.extra_gates(candidate) if any(r.verdict == GateVerdict.FAIL for r in extra_results): prev_reason = "; ".join( r.reason for r in extra_results if r.verdict == GateVerdict.FAIL ) logger.info( "slot {} 额外 gate 失败 (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" store.mark_item_rejected(item_id, prev_reason) logger.info( "slot {} 重复题被拒绝 (attempt {}/{})", slot.slot_id, attempt, config.retry_limit, ) continue # Phase 8: 通过全部检查 → 接受（使用洗牌后的选项和答案） result = _to_generated_question( candidate, family=strategy.strategy_name, options=pp.options, answer=pp.answer, sub_pattern=sub_pattern.name if sub_pattern else None, ) # 将题目 embedding 加入池（flatten 确保 1D） embed_pool.append(embed_fn(candidate.question).flatten()) # 标记使用的节点 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 # --------------------------------------------------------------------------- # 管线辅助函数 # --------------------------------------------------------------------------- _DEFAULT_TASK_TYPES: list[str] = [ "Action Recognition", "Action Reasoning", "Attribute Perception", "Counting Problem", "Information Synopsis", "Object Recognition", "Object Reasoning", "OCR Problems", "Spatial Perception", "Spatial Reasoning", "Temporal Perception", "Temporal Reasoning", ] def _get_git_sha() -> str: """获取当前 Git HEAD 短 SHA。 返回: 短 SHA 字符串；获取失败时返回 "unknown"。 """ import subprocess try: return subprocess.check_output( ["git", "rev-parse", "--short", "HEAD"], text=True, timeout=5, ).strip() except (subprocess.SubprocessError, FileNotFoundError): return "unknown" def _filter_pending_slots( slots: list[SlotAssignment], progress: dict[str, str], ) -> list[SlotAssignment]: """过滤出未完成的 slot（跳过 progress 中已记录的）。 参数: slots: 全部 slot 列表。 progress: 已完成 slot 映射 {slot_id → "accepted"|"rejected"}。 返回: 待处理的 slot 列表。 """ pending = [s for s in slots if s.slot_id not in progress] logger.info( "待处理 slots: {} / {} (已跳过 {})", len(pending), len(slots), len(slots) - len(pending), ) return pending async def _run_heavy_sampling( accepted: list[GeneratedQuestion], trees: dict[str, TreeIndex], store: QuestionGenStore, llm: LLMProvider, config: PipelineConfig, *, session_id: str, ) -> list[tuple[str, int]]: """对接受题目按比例随机抽检，计算难度步数。 参数: accepted: 已接受的题目列表。 trees: video_id → TreeIndex 映射。 store: 日志记录器。 llm: LLM 调用端口。 config: 管线配置（含 heavy_sample_rate 和 seed）。 session_id: 会话 ID。 返回: (question_id, difficulty_steps) 元组列表。 """ if config.heavy_sample_rate <= 0 or not accepted: return [] 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) heavy_sampled: list[tuple[str, int]] = [] for q, steps in zip(sampled_questions, heavy_results, strict=True): heavy_sampled.append((q.question_id, steps)) 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) return heavy_sampled # --------------------------------------------------------------------------- # 管线主入口 # --------------------------------------------------------------------------- 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, on_accept: Callable[[GeneratedQuestion], None] | None = None, initial_used_node_ids: set[str] | None = None, initial_embed_pool: list[np.ndarray] | None = None, seq_offset: int = 0, ) -> 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"}。 on_accept: 每接受一题时的回调（用于实时持久化，防崩溃丢数据）。 initial_used_node_ids: 补生成时继承的已用节点集合（避开已用节点，默认空）。 initial_embed_pool: 补生成时继承的已接受题 embedding 池（跨 run 去重，默认空）。 seq_offset: 全局序号起始偏移（补生成续编 seq，防撞 question_id，默认 0）。 返回: PipelineResult 实例。 """ task_types = task_types or _DEFAULT_TASK_TYPES progress = progress or {} rng = random.Random(config.seed) # Phase 1: 分配 slot + 创建 run 记录 slots = _assign_slots(video_ids, task_types, config.per_type, seq_offset=seq_offset) logger.info( "管线启动: {} slots, {} 视频, {} 任务类型", len(slots), len(video_ids), len(task_types) ) run_id = uuid.uuid4().hex store.record_run_start(run_id, _get_git_sha(), str(config)) # Phase 2: 过滤已完成 slot pending_slots = _filter_pending_slots(slots, progress) # Phase 3: 并发处理 sem = asyncio.Semaphore(config.concurrency) # 复制传入容器，避免补生成 run 就地改动调用方对象（不传=空初始化=现状） embed_pool: list[np.ndarray] = list(initial_embed_pool) if initial_embed_pool else [] used_node_ids: set[str] = set(initial_used_node_ids) if initial_used_node_ids else 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 result = 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, all_trees=trees, ) if result is not None and on_accept is not None: on_accept(result) return result results = await asyncio.gather(*[_process_wrapper(s) for s in pending_slots]) # Phase 4: 统计 + 重量抽检 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) heavy_sampled = await _run_heavy_sampling( accepted, trees, store, llm, config, session_id=session_id ) # Phase 5: 更新 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, )