iomgaa
4fb7a61f8b
1. Apply postprocess shuffle result (pp.options, pp.answer) to final GeneratedQuestion output instead of using original candidate values. 2. Record dedup rejection in store via new mark_item_rejected() method, preventing items from staying as 'accepted' after dedup rejects them. 3. Add .flatten() to embed_fn outputs in _is_duplicate and embed_pool append to handle 2D (1,D) arrays from embedding implementations. 4. Validate exactly 4 options in _validate_parsed_fields (was >= 2), matching the A-D answer constraint. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
795 lines
25 KiB
Python
795 lines
25 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 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).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,
|
||
*,
|
||
options: tuple[str, ...] | None = None,
|
||
answer: str | None = None,
|
||
) -> GeneratedQuestion:
|
||
"""将 CandidateQuestion 转换为 GeneratedQuestion。
|
||
|
||
参数:
|
||
candidate: 门控通过的候选题目。
|
||
options: 洗牌后的选项元组（若为 None 则使用 candidate 原始选项）。
|
||
answer: 重映射后的答案字母（若为 None 则使用 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=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,
|
||
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"
|
||
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, options=pp.options, answer=pp.answer)
|
||
# 将题目 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",
|
||
"Action Prediction",
|
||
"Action Sequence",
|
||
"Object Recognition",
|
||
"Object Reasoning",
|
||
"Object Interaction",
|
||
"Scene Understanding",
|
||
"Event Reasoning",
|
||
"Causal Reasoning",
|
||
"Temporal Reasoning",
|
||
"Spatial 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,
|
||
) -> 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 实例。
|
||
"""
|
||
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, config.family_ratios, rng)
|
||
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
|
||
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 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,
|
||
)
|