iomgaa
043d4aa46f
9 tasks: type extension → postprocess → families → sampler_v2 → generator_v2 → gates → run_store → pipeline_v2 → CLI integration. Includes structured-logging schemas/metrics and Codex review revisions.
572 lines
22 KiB
Markdown
572 lines
22 KiB
Markdown
---
|
||
type: plan
|
||
node_id: plan:2026-07-11-question-gen-v2
|
||
title: "出题管线 v2 实现计划"
|
||
date: 2026-07-12
|
||
---
|
||
|
||
# 出题管线 v2 实现计划
|
||
|
||
> **For agentic workers:** REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||
|
||
**Goal:** 将出题管线从"按 task_type 平铺生成"升级为"按失败机理题族生成 + 逐题轻量四门质量关 + 重出循环 + 重量抽检"，在不影响 harness/推理侧的前提下显著提升题目区分度与泄漏控制。
|
||
|
||
**Architecture:** 新增 6 个模块（postprocess / families / sampler_v2 / generator_v2 / gates / pipeline_v2）+ 1 个日志存储（run_store），均位于 `app/question_gen/` 下；扩展 `core/types.py` 和 `config/default.yaml`；改造 `tools/generate_questions.py` CLI。v1 `generate_one` 保留不删。
|
||
|
||
**Tech Stack:** Python 3.11, asyncio, loguru, SQLite, GovernedLLMClient/GovernedVLMClient (DI), numpy, pytest
|
||
|
||
---
|
||
|
||
## 依赖拓扑
|
||
|
||
```mermaid
|
||
graph TD
|
||
T1[Task 1: 类型扩展+配置] --> T2[Task 2: 后处理层]
|
||
T1 --> T3[Task 3: 题族规格]
|
||
T3 --> T4[Task 4: v2 采样器]
|
||
T4 --> T5[Task 5: v2 生成器]
|
||
T3 --> T6[Task 6: 轻量四门]
|
||
T2 --> T6
|
||
T1 --> T7[Task 7: SQLite 日志]
|
||
T6 --> T7
|
||
T5 --> T8[Task 8: 编排管线]
|
||
T6 --> T8
|
||
T7 --> T8
|
||
T8 --> T9[Task 9: CLI + 回归]
|
||
```
|
||
|
||
可并行: T2 / T3 / T7（仅依赖 T1）。
|
||
|
||
---
|
||
|
||
## Task 1: GeneratedQuestion 类型扩展 + 配置 YAML
|
||
|
||
**Files:** `core/types.py`(改) | `config/default.yaml`(改) | `app/question_gen/loader.py`(改) | `tests/unit/test_types_v2.py`(新)
|
||
|
||
> **Codex 审查补充:** 若 `app/harness/pools.py` 存在 `_q_to_dict`/`_dict_to_q` 转换函数，需同步保留 `skill_target`/`difficulty_steps` 字段。实现者须 grep 确认并修改。
|
||
|
||
- [ ] **Step 1 (RED):** 测试新字段默认 None + 显式赋值 + frozen 不可变
|
||
|
||
```python
|
||
# tests/unit/test_types_v2.py
|
||
class TestGeneratedQuestionV2:
|
||
def test_new_fields_default_none(self):
|
||
q = GeneratedQuestion(question_id="q1", video_id="v1", task_type="TN",
|
||
question="?", options=("A","B","C","D"), answer="A", source_nodes=("n1",), difficulty="medium")
|
||
assert q.skill_target is None and q.difficulty_steps is None
|
||
|
||
def test_new_fields_explicit(self):
|
||
q = GeneratedQuestion(..., skill_target="M1", difficulty_steps=5)
|
||
assert q.skill_target == "M1"
|
||
|
||
class TestLoaderV2Compat:
|
||
def test_load_old_format_json(self, tmp_path): ... # 无新字段 -> None
|
||
def test_load_new_format_json(self, tmp_path): ... # 有新字段 -> 正确加载
|
||
```
|
||
|
||
- [ ] **Step 2 (GREEN):** `core/types.py` 添加 `skill_target: str | None = field(default=None)` 和 `difficulty_steps: int | None = field(default=None)`
|
||
|
||
- [ ] **Step 3 (GREEN):** `loader.py` 构造处加 `skill_target=item.get("skill_target"), difficulty_steps=item.get("difficulty_steps")`
|
||
|
||
- [ ] **Step 4:** `config/default.yaml` 新增 `question_gen_v2:` 节
|
||
|
||
```yaml
|
||
question_gen_v2:
|
||
family_ratios: {retrieval: 0.30, reasoning: 0.25, enumeration: 0.20, visual: 0.15, spatial: 0.10}
|
||
gate: {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"}
|
||
dedup_threshold: 0.85
|
||
retry_limit: 3
|
||
heavy_sample_rate: 0.15
|
||
heavy_agent_model: "gpt-4.1-mini"
|
||
output_dir: "store/questions/generated-v2"
|
||
per_type: 20 # 12 类 × 20 = 240 题（设计 §3 硬约束）
|
||
concurrency: 4
|
||
seed: 42
|
||
```
|
||
|
||
- [ ] **Step 5 (COMMIT):** `feat(types): extend GeneratedQuestion with skill_target & difficulty_steps`
|
||
|
||
---
|
||
|
||
## Task 2: 确定性后处理层 `app/question_gen/postprocess.py`
|
||
|
||
**Files:** `app/question_gen/postprocess.py`(新) | `tests/unit/test_postprocess.py`(新)
|
||
|
||
### 签名
|
||
|
||
```python
|
||
@dataclass(frozen=True)
|
||
class PostprocessResult:
|
||
options: tuple[str, ...]
|
||
answer: str
|
||
referent_violations: list[str]
|
||
verbatim_ratio: float
|
||
has_time_anchor: bool
|
||
|
||
def shuffle_options(options: tuple[str, ...], answer: str, rng: random.Random) -> tuple[tuple[str, ...], str]: ...
|
||
def check_referent_blacklist(question_text: str) -> list[str]: ...
|
||
def check_verbatim(question_text: str, correct_option: str, source_texts: list[str], window: int = 6) -> float: ...
|
||
def has_time_anchor(question_text: str) -> bool: ...
|
||
def check_forbidden_material(source_nodes_text: str, task_type: str) -> list[str]: ...
|
||
"""出题禁区：检测 T1 类素材（瞬时动作/记分牌时序/无对白因果）和 T7 噪声模式（选项重复/计数边界口径含糊）。返回违规描述列表，空列表=通过。"""
|
||
def run_postprocess(question_text: str, options: tuple[str, ...], answer: str, source_texts: list[str], rng: random.Random) -> PostprocessResult: ...
|
||
```
|
||
|
||
- [ ] **Step 1 (RED):** 测试
|
||
|
||
```python
|
||
class TestShuffleOptions:
|
||
def test_deterministic_with_seed(self): ...
|
||
def test_answer_remapped_correctly(self): ... # answer 字母始终指向原正确文本
|
||
|
||
class TestReferentBlacklist:
|
||
def test_clean_passes(self): ...
|
||
def test_this_clip_caught(self): ...
|
||
|
||
class TestVerbatim:
|
||
def test_zero_overlap(self): ...
|
||
def test_full_copy_returns_one(self): ...
|
||
def test_partial(self): ...
|
||
|
||
class TestTimeAnchor:
|
||
def test_timestamp(self): ... # "at 01:30" -> True
|
||
def test_no_anchor(self): ... # -> False
|
||
```
|
||
|
||
- [ ] **Step 2 (GREEN):** 实现。逻辑: shuffle 用 index permutation+remap; blacklist 用预编译 `_BLACKLIST_PATTERNS`; verbatim 用 word-level n-gram set intersection ratio; time_anchor 用正则 `\d{1,2}:\d{2}` + 短语列表。
|
||
|
||
- [ ] **Step 3 (COMMIT):** `feat(question_gen): add deterministic postprocess layer`
|
||
|
||
---
|
||
|
||
## Task 3: 题族规格声明 `app/question_gen/families.py`
|
||
|
||
**Files:** `app/question_gen/families.py`(新) | `tests/unit/test_families.py`(新)
|
||
|
||
### 签名
|
||
|
||
```python
|
||
@dataclass(frozen=True)
|
||
class LeakTestProfile:
|
||
shortcut_type: str # "temporal_proximity"/"option_length"/"frequency"/"visual_salience"/"spatial_default"
|
||
probe_template: str # store/prompts/question_gen/ 下模板名
|
||
pass_threshold: float
|
||
|
||
@dataclass(frozen=True)
|
||
class SamplingConstraint:
|
||
min_subtitles: int; min_l3_nodes: int; require_frames: bool; cross_l2_span: bool
|
||
|
||
@dataclass(frozen=True)
|
||
class QuestionFamilySpec:
|
||
name: str; skill_target: str; sampling: SamplingConstraint
|
||
legal_task_types: frozenset[str]; leak_profile: LeakTestProfile; prompt_template: str
|
||
|
||
RETRIEVAL_FAMILY: QuestionFamilySpec # M1, 30%
|
||
REASONING_FAMILY: QuestionFamilySpec # M2, 25%
|
||
ENUMERATION_FAMILY: QuestionFamilySpec # M3, 20%
|
||
VISUAL_FAMILY: QuestionFamilySpec # M4, 15%
|
||
SPATIAL_FAMILY: QuestionFamilySpec # M5, 10%
|
||
ALL_FAMILIES: tuple[QuestionFamilySpec, ...]
|
||
|
||
def get_family_for_slot(task_type: str, family_ratios: dict[str, float], rng: random.Random) -> QuestionFamilySpec: ...
|
||
```
|
||
|
||
- [ ] **Step 1 (RED):**
|
||
|
||
```python
|
||
class TestFamilySpec:
|
||
def test_all_families_cover_all_task_types(self): ... # 12 类全覆盖
|
||
def test_skill_targets_unique(self): ...
|
||
class TestGetFamilyForSlot:
|
||
def test_respects_legal_task_types(self): ... # 返回族必含给定 type
|
||
def test_deterministic_with_seed(self): ...
|
||
def test_invalid_task_type_raises(self): ...
|
||
def test_distribution_approximates_ratios(self): ... # chi-square p>0.01
|
||
```
|
||
|
||
- [ ] **Step 2 (GREEN):** 实现。`get_family_for_slot`: 过滤 legal -> 归一化权重 -> `rng.choices`。
|
||
|
||
- [ ] **Step 3 (COMMIT):** `feat(question_gen): add 5 question family specs with sampling constraints`
|
||
|
||
---
|
||
|
||
## Task 4: v2 采样器 `app/question_gen/sampler_v2.py`
|
||
|
||
**Files:** `app/question_gen/sampler_v2.py`(新) | `tests/unit/test_sampler_v2.py`(新)
|
||
|
||
### 签名
|
||
|
||
```python
|
||
@dataclass(frozen=True)
|
||
class MaterialContext:
|
||
anchor: AnchorContext
|
||
source_nodes: tuple[str, ...]
|
||
subtitle_sentences: list[str]
|
||
frame_paths: list[str]
|
||
cross_l2_texts: list[str]
|
||
|
||
def _validate_sampling_constraints(tree: TreeIndex, node_id: str, constraint: SamplingConstraint) -> bool: ...
|
||
def _collect_subtitle_sentences(tree: TreeIndex, node_ids: tuple[str, ...]) -> list[str]: ...
|
||
def _collect_cross_l2_context(tree: TreeIndex, anchor_l2_id: str, max_peers: int = 3) -> list[str]: ...
|
||
|
||
def sample_material_v2(
|
||
tree: TreeIndex, family_spec: QuestionFamilySpec, task_type: str,
|
||
used_node_ids: set[str], rng: random.Random, *, max_attempts: int = 10,
|
||
) -> MaterialContext: ...
|
||
# Raises RuntimeError if max_attempts exhausted
|
||
```
|
||
|
||
- [ ] **Step 1 (RED):**
|
||
|
||
```python
|
||
class TestSampleMaterialV2:
|
||
def test_returns_material_context(self, real_tree): ...
|
||
def test_respects_used_nodes(self, real_tree): ...
|
||
def test_constraint_violation_retries(self, real_tree): ... # -> RuntimeError
|
||
def test_cross_l2_populated_for_reasoning(self, real_tree): ...
|
||
def test_subtitle_sentences_from_anchor(self, real_tree): ...
|
||
```
|
||
|
||
- [ ] **Step 2 (GREEN):** 实现。路由 `TASK_TYPE_LEVEL_MAP[task_type].level` -> `_sample_lX`; 验证 constraints; 收集 subtitles + cross_l2。
|
||
|
||
- [ ] **Step 3 (COMMIT):** `feat(question_gen): add v2 material sampler with family constraints`
|
||
|
||
---
|
||
|
||
## Task 5: v2 生成器 `app/question_gen/generator_v2.py`
|
||
|
||
**Files:** `app/question_gen/generator_v2.py`(新) | `store/prompts/question_gen/{retrieval,reasoning,enumeration,visual,spatial}.md`(新) | `tests/unit/test_generator_v2.py`(新)
|
||
|
||
### 签名
|
||
|
||
```python
|
||
@dataclass(frozen=True)
|
||
class CandidateQuestion:
|
||
question_id: str; video_id: str; task_type: str; skill_target: str
|
||
question: str; options: tuple[str, ...]; answer: str
|
||
source_nodes: tuple[str, ...]; difficulty: str
|
||
subtitle_sentences: tuple[str, ...] = field(default_factory=tuple) # 验证材料
|
||
frame_paths: tuple[str, ...] = field(default_factory=tuple)
|
||
|
||
def _load_prompt_template(family_spec: QuestionFamilySpec) -> str: ...
|
||
def _build_v2_prompt(family_spec: QuestionFamilySpec, material: MaterialContext, task_type: str, seq: int, *, reject_reason: str | None = None) -> tuple[list[dict[str, str]], list[str]]: ...
|
||
def _parse_v2_response(raw: str, video_id: str, task_type: str, skill_target: str, seq: int, source_nodes: tuple[str, ...]) -> CandidateQuestion: ...
|
||
|
||
async def generate_one_v2(
|
||
vlm: VLMProvider, tree: TreeIndex, material: MaterialContext,
|
||
family_spec: QuestionFamilySpec, task_type: str, seq: int, *,
|
||
video_id: str, reject_reason: str | None = None, session_id: str,
|
||
) -> CandidateQuestion: ...
|
||
```
|
||
|
||
- [ ] **Step 1 (RED):**
|
||
|
||
```python
|
||
class TestBuildV2Prompt:
|
||
def test_includes_family_template(self): ...
|
||
def test_reject_reason_injected(self): ...
|
||
class TestParseV2Response:
|
||
def test_valid_json(self): ...
|
||
def test_missing_field_raises(self): ...
|
||
def test_invalid_answer_raises(self): ...
|
||
class TestGenerateOneV2:
|
||
async def test_happy_path(self, mock_vlm): ... # -> CandidateQuestion
|
||
```
|
||
|
||
- [ ] **Step 2 (GREEN):** 实现 + 5 个 per-family prompt 模板（JSON output format）。
|
||
|
||
- [ ] **Step 3 (COMMIT):** `feat(question_gen): add v2 generator with per-family prompt templates`
|
||
|
||
---
|
||
|
||
## Task 6: 轻量四门 `app/question_gen/gates.py`
|
||
|
||
**Files:** `app/question_gen/gates.py`(新) | `store/prompts/question_gen/gate_{key_verify,blind_answer,multi_true,leak_*}.md`(新) | `tests/unit/test_gates.py`(新)
|
||
|
||
### 签名
|
||
|
||
```python
|
||
class GateVerdict(Enum): PASS = "pass"; FAIL = "fail"; SKIP = "skip"
|
||
|
||
@dataclass(frozen=True)
|
||
class GateResult:
|
||
verdict: GateVerdict; reason: str; raw_response: str
|
||
|
||
@dataclass(frozen=True)
|
||
class GateReport:
|
||
key_verify: GateResult; blind_answer: GateResult; multi_true: GateResult; leak_test: GateResult
|
||
@property
|
||
def passed(self) -> bool: ... # 全门 PASS|SKIP
|
||
@property
|
||
def reject_reason(self) -> str | None: ... # 首个 FAIL 门 reason
|
||
|
||
async def _gate_key_verify(candidate, tree, llm, *, session_id) -> GateResult: ...
|
||
async def _gate_blind_answer(candidate, llm, *, session_id) -> GateResult: ...
|
||
async def _gate_multi_true(candidate, tree, llm, *, session_id) -> GateResult: ...
|
||
async def _gate_leak_test(candidate, family_spec, llm, *, session_id) -> GateResult: ...
|
||
|
||
async def run_gates(
|
||
candidate: CandidateQuestion, tree: TreeIndex, llm: LLMProvider,
|
||
family_spec: QuestionFamilySpec, postprocess: PostprocessResult, *, session_id: str,
|
||
) -> GateReport: ...
|
||
# 前置: verbatim_ratio > 0.5 直接 FAIL key_verify; 并发 asyncio.gather 四门
|
||
```
|
||
|
||
- [ ] **Step 1 (RED):**
|
||
|
||
```python
|
||
class TestGateKeyVerify:
|
||
async def test_pass_evidence(self, mock_llm): ...
|
||
async def test_fail_no_evidence(self, mock_llm): ...
|
||
class TestGateBlindAnswer:
|
||
async def test_pass_wrong(self, mock_llm): ...
|
||
async def test_fail_correct(self, mock_llm): ...
|
||
class TestGateMultiTrue:
|
||
async def test_pass_single(self, mock_llm): ...
|
||
async def test_fail_multi(self, mock_llm): ...
|
||
class TestGateLeakTest:
|
||
async def test_per_family_template(self, mock_llm): ...
|
||
class TestRunGates:
|
||
async def test_all_pass(self): ...
|
||
async def test_high_verbatim_shortcircuits(self): ...
|
||
```
|
||
|
||
- [ ] **Step 2 (GREEN):** 实现 + 8 个 gate prompt 模板（key_verify / blind_answer / multi_true / leak x 5 族）。
|
||
|
||
- [ ] **Step 3 (COMMIT):** `feat(question_gen): add lightweight 4-gate quality check`
|
||
|
||
---
|
||
|
||
## Task 7: SQLite 日志记录器 `app/question_gen/run_store.py`
|
||
|
||
**Files:** `app/question_gen/run_store.py`(新) | `tests/unit/test_run_store.py`(新)
|
||
|
||
### 签名
|
||
|
||
```python
|
||
@dataclass(frozen=True)
|
||
class RunStats:
|
||
total_slots: int; accepted: int; rejected: int; heavy_sampled: int
|
||
|
||
class QuestionGenStore:
|
||
def __init__(self, db_path: Path) -> None: ...
|
||
def _init_schema(self) -> None: ... # 幂等 DDL（参照 research-wiki/schemas/）
|
||
def record_run_start(self, run_id: str, git_sha: str, config_snapshot: str) -> None: ...
|
||
def record_run_end(self, run_id: str, status: str, stats: RunStats) -> None: ...
|
||
def record_item(self, item_id: str, run_id: str, slot_id: str, video_id: str,
|
||
family: str, task_type: str, skill_target: str, attempt: int, question_text: str) -> None: ...
|
||
def update_gates(self, item_id: str, report: GateReport) -> None: ...
|
||
def update_difficulty(self, item_id: str, difficulty_steps: int) -> None: ...
|
||
def get_run_stats(self, run_id: str) -> RunStats: ...
|
||
def close(self) -> None: ...
|
||
```
|
||
|
||
- [ ] **Step 1 (RED):**
|
||
|
||
```python
|
||
class TestQuestionGenStore:
|
||
def test_schema_idempotent(self, store): ...
|
||
def test_record_run_lifecycle(self, store): ... # start -> end -> get_stats
|
||
def test_record_item_and_gates(self, store): ...
|
||
def test_update_difficulty(self, store): ...
|
||
```
|
||
|
||
- [ ] **Step 2 (GREEN):** 实现。同步 sqlite3（写入频率低无需 aiosqlite）。
|
||
|
||
- [ ] **Step 3 (COMMIT):** `feat(question_gen): add SQLite run store for generation telemetry`
|
||
|
||
---
|
||
|
||
## Task 8: 重出循环 + 重量抽检 + v2 编排 `app/question_gen/pipeline_v2.py`
|
||
|
||
**Files:** `app/question_gen/pipeline_v2.py`(新) | `tests/integration/test_pipeline_v2.py`(新)
|
||
|
||
### 签名
|
||
|
||
```python
|
||
@dataclass(frozen=True)
|
||
class SlotAssignment:
|
||
slot_id: str; video_id: str; task_type: str; family: QuestionFamilySpec; seq: int
|
||
|
||
@dataclass
|
||
class PipelineResult:
|
||
accepted: list[GeneratedQuestion]; rejected_count: int
|
||
heavy_sampled: list[tuple[str, int]]
|
||
|
||
@dataclass(frozen=True)
|
||
class PipelineConfig:
|
||
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: ...
|
||
|
||
def _assign_slots(video_ids: list[str], task_types: list[str], per_type: int,
|
||
family_ratios: dict[str, float], rng: random.Random) -> list[SlotAssignment]: ...
|
||
|
||
async def _process_one_slot(slot, tree, vlm, llm, embed_fn, embed_pool, store, config,
|
||
used_node_ids, rng, sem, *, session_id) -> GeneratedQuestion | None: ...
|
||
# 生成 -> postprocess -> gates -> 重出(<=retry_limit) -> dedup -> accept/reject
|
||
|
||
async def _heavy_check_one(question, tree, llm, *, session_id) -> int: ...
|
||
# 盲 Agent 试答 -> difficulty_steps
|
||
|
||
async def run_pipeline_v2(video_ids, trees, vlm, llm, embed_fn, store, config, *, progress=None) -> PipelineResult: ...
|
||
# slots -> skip done -> sem-bounded gather -> heavy 15% -> stats
|
||
```
|
||
|
||
**_process_one_slot 核心流程:**
|
||
|
||
```
|
||
for attempt in 1..retry_limit:
|
||
material = sample_material_v2(...) # attempt>retry_limit 时重采样
|
||
candidate = generate_one_v2(..., reject_reason=prev_reason)
|
||
store.record_item(...)
|
||
pp = run_postprocess(candidate.question, candidate.options, ...)
|
||
if pp.verbatim_ratio > 0.5: reject("verbatim")
|
||
report = await run_gates(candidate, tree, llm, family, pp)
|
||
store.update_gates(...)
|
||
if report.passed:
|
||
if not is_duplicate(candidate.question, embed_pool, embed_fn, threshold):
|
||
return to_generated_question(candidate)
|
||
prev_reason = report.reject_reason
|
||
return None
|
||
```
|
||
|
||
- [ ] **Step 1 (RED):**
|
||
|
||
```python
|
||
class TestSlotAssignment:
|
||
def test_per_type_count(self): ...
|
||
def test_family_distribution(self): ...
|
||
class TestProcessOneSlot:
|
||
async def test_happy_path(self, deps): ...
|
||
async def test_retry_on_fail(self, deps): ...
|
||
async def test_max_retries_none(self, deps): ...
|
||
class TestPipelineV2:
|
||
async def test_full_flow(self, deps): ...
|
||
async def test_progress_resume(self, deps): ...
|
||
async def test_heavy_check_samples(self, deps): ...
|
||
async def test_store_records_all(self, deps): ...
|
||
```
|
||
|
||
- [ ] **Step 2 (GREEN):** 实现编排。
|
||
|
||
- [ ] **Step 3 (COMMIT):** `feat(question_gen): add v2 pipeline with retry loop and heavy check`
|
||
|
||
---
|
||
|
||
## Task 9: CLI 改造 + sh 脚本 + 全量回归
|
||
|
||
**Files:** `tools/generate_questions.py`(改) | `scripts/generate_questions_v2.sh`(新) | `app/question_gen/__init__.py`(改) | `tests/integration/test_cli_generate_v2.py`(新)
|
||
|
||
### 改造
|
||
|
||
```python
|
||
# tools/generate_questions.py 新增
|
||
def _add_generate_v2_parser(subparsers) -> None:
|
||
p = subparsers.add_parser("generate-v2")
|
||
p.add_argument("--config", type=Path, default=Path("config/default.yaml"))
|
||
p.add_argument("--store-dir", type=Path, required=True)
|
||
p.add_argument("--db-path", type=Path, default=Path("logs/question_gen.db"))
|
||
p.add_argument("--seed", type=int, default=None)
|
||
p.add_argument("--dry-run", action="store_true")
|
||
|
||
async def _run_generate_v2(args) -> None: ...
|
||
# load config -> build clients(DI) -> discover videos -> load trees -> init store -> resume progress -> run_pipeline_v2 -> cleanup
|
||
```
|
||
|
||
```bash
|
||
# scripts/generate_questions_v2.sh
|
||
#!/usr/bin/env bash
|
||
set -euo pipefail
|
||
STORE_DIR="${STORE_DIR:-store}"; CONFIG="${CONFIG:-config/default.yaml}"
|
||
DB_PATH="${DB_PATH:-logs/question_gen.db}"
|
||
source activate Video-Tree-TRM
|
||
[ "${MODE:-}" = "mock" ] && export LLM_MOCK=1 VLM_MOCK=1
|
||
python tools/generate_questions.py generate-v2 --store-dir "$STORE_DIR" --config "$CONFIG" --db-path "$DB_PATH" ${SEED:+--seed $SEED}
|
||
```
|
||
|
||
- [ ] **Step 1 (RED):**
|
||
|
||
```python
|
||
class TestCLIGenerateV2:
|
||
def test_subcommand_help(self): ... # returncode 0, "--config" in stdout
|
||
def test_dry_run(self, tmp_path): ... # 不调 LLM
|
||
```
|
||
|
||
- [ ] **Step 2 (GREEN):** 实现 CLI + sh。
|
||
|
||
- [ ] **Step 3:** 更新 `__init__.py` 导出: `run_pipeline_v2, PipelineConfig, PipelineResult, QuestionFamilySpec, ALL_FAMILIES, CandidateQuestion, generate_one_v2, GateReport, run_gates`
|
||
|
||
- [ ] **Step 4:** 全量回归
|
||
|
||
```bash
|
||
conda activate Video-Tree-TRM && pytest tests/ --cov=app --cov=core --cov-report=term-missing -x
|
||
conda activate Video-Tree-TRM && ruff check app/ core/ adapters/ tools/ --fix && ruff format app/ core/ adapters/ tools/
|
||
conda activate Video-Tree-TRM && radon cc app/question_gen/ -nc
|
||
```
|
||
|
||
- [ ] **Step 5 (COMMIT):** `feat(question_gen): add generate-v2 CLI subcommand and experiment script`
|
||
|
||
---
|
||
|
||
## Self-Review
|
||
|
||
### 设计合规性
|
||
|
||
| 设计决策 | 计划落点 | OK |
|
||
|----------|---------|:--:|
|
||
| 双标签 task_type + skill_target | T1 类型扩展 + T5 CandidateQuestion | Y |
|
||
| 5 题族 | T3 families.py | Y |
|
||
| 确定性后处理 | T2 postprocess.py（零 LLM） | Y |
|
||
| 轻量四门 ~4 LLM/题 | T6 gates.py | Y |
|
||
| 重出循环 max 3 + 拒因回填 | T8 _process_one_slot | Y |
|
||
| 重量抽检 15% | T8 _heavy_check_one | Y |
|
||
| SQLite 日志 | T7 run_store.py（schema 匹配 wiki） | Y |
|
||
| 输出 generated-v2/ | T8 config.output_dir | Y |
|
||
| 科研配置 YAML | T1 question_gen_v2 节 | Y |
|
||
|
||
### 约束合规性
|
||
|
||
| 约束 | 验证 |
|
||
|------|------|
|
||
| 不改 harness/推理侧 | 全部新增在 app/question_gen/ |
|
||
| 保留 v1 generate_one | synthesizer.py 零改动 |
|
||
| GovernedLLMClient DI | 通过 LLMProvider/VLMProvider Protocol 注入 |
|
||
| 中文 docstring | 所有公共函数含中文文档 |
|
||
| TDD 红-绿-重构 | 每 Task 先 RED 再 GREEN |
|
||
| loguru 禁 print | 全部新模块 |
|
||
| 依赖方向 | 新模块在 app/ 层，仅依赖 core/ |
|
||
|
||
### Codex Plan Review 修订记录
|
||
|
||
| 采纳 | 修订 |
|
||
|------|------|
|
||
| per_type 30→20 | 配置已修正为 20（12×20=240 硬约束） |
|
||
| pools.py 遗漏 | T1 补充注释：实现者须 grep _q_to_dict 并同步修改 |
|
||
| 出题禁区未覆盖 | T2 签名追加 `check_forbidden_material` 函数 |
|
||
| T6→T7 依赖缺失 | 拓扑图已添加边 |
|
||
| parent_call_id | 实现者在 T6/T9 中须贯穿 session_id → parent_call_id（GovernedLLMClient 已支持） |
|
||
| T8 过大 | 保留单 task 但实现者可按 _process_one_slot / _heavy_check_one / resume 分步 commit |
|
||
| heavy_check 复用推理管线 | 采纳为实施约束：T8 内 heavy_check 须调用现有 AgentLoop + 树环境，不重新实现 |
|
||
|
||
驳回项：
|
||
- "采样约束覆盖不足"——这些是 prompt 模板内容（store/prompts/question_gen/），非代码逻辑；设计 §2.1 约束通过四门兜底验证
|
||
- "T5 依赖 T2"——生成器输出原始答案位置，shuffle 在编排层（T8 pipeline_v2）后处理，不在生成器内
|
||
- "chi-square 测试不稳定"——测试用固定 10000 次采样 + p>0.001 宽容阈值
|
||
|
||
### 核心算法保真校验
|
||
|
||
本计划**不涉及**算法清单 12 项中任何一项的修改。出题"gate"与训练 CE-Gate(#4) 完全不同；heavy_check 使用树搜索(#11)但不修改其实现。
|
||
|
||
### 风险与缓解
|
||
|
||
| 风险 | 缓解 |
|
||
|------|------|
|
||
| 四门 LLM 成本 | gate model 用 gpt-4.1-mini; asyncio.gather 并发 |
|
||
| 重出循环耗时 | retry_limit=3 硬上限 + 超限放弃 |
|
||
| heavy_check 超时 | GovernedLLMClient 超时 + 步数上限 |
|
||
| 族配比与 task_type 冲突 | legal_task_types 过滤后归一化 |
|