iomgaa
96e314c3a0
selector_error 分支（捕获 ValueError/FileNotFoundError）此前只返回 reason， 未 mark_item_rejected，导致 Phase 3 已 record 的 pending attempt 行永远停在 pending；而 hard-fail 分支会标记 rejected。两条失败路径落库风格现统一为 mark_item_rejected（异常路径无 outcome/observation，故不写 selector_scores）。 补单测 test_apply_grounded_selector_marks_rejected_on_error 守卫该路径。
1010 lines
33 KiB
Python
1010 lines
33 KiB
Python
"""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,
|
||
) -> list[SlotAssignment]:
|
||
"""将出题目标分配为具体 slot 列表。
|
||
|
||
总 slot 数 = len(task_types) * per_type。
|
||
在视频间 round-robin 分配。不再选择 family — strategy 在处理时查找。
|
||
|
||
参数:
|
||
video_ids: 视频 ID 列表。
|
||
task_types: 任务类型列表。
|
||
per_type: 每种 task_type 的目标题数。
|
||
|
||
返回:
|
||
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)]
|
||
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) as e:
|
||
# 异常路径无 outcome/observation 可落，但仍须 mark_item_rejected，
|
||
# 与 hard-fail 路径落库风格一致（该 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,
|
||
) -> 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: 每接受一题时的回调（用于实时持久化，防崩溃丢数据）。
|
||
|
||
返回:
|
||
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)
|
||
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)
|
||
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
|
||
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,
|
||
)
|