From dbc9d38cd7fb76689a2e466708050a44546556df Mon Sep 17 00:00:00 2001 From: iomgaa Date: Thu, 9 Jul 2026 00:08:23 -0400 Subject: [PATCH] =?UTF-8?q?plan(tree/repair):=20=E4=B8=89=E9=A1=B9?= =?UTF-8?q?=E6=94=B9=E9=80=A0=E5=AE=9E=E7=8E=B0=E8=AE=A1=E5=88=92(?= =?UTF-8?q?=E9=81=A5=E6=B5=8B=E5=8A=A0=E5=9B=BA+=E6=96=AD=E7=82=B9?= =?UTF-8?q?=E7=BB=AD=E8=B7=91+=E5=B9=B6=E5=8F=91)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 6 个 Task: telemetry 防御加固 → call_id 根因修复 → detector L2/L1 扩展 → progress 管理 → 并发编排+CLI → lint+全量测试 --- .../designs/2026-07-07-question-gen-design.md | 180 ++++ .../designs/tree-repair-resilience.md | 9 + research-wiki/graph/edges.json | 17 + research-wiki/index.md | 10 +- research-wiki/log.md | 4 + .../plans/2026-07-07-question-gen.md | 894 ++++++++++++++++++ .../2026-07-09-tree-repair-resilience.md | 786 +++++++++++++++ research-wiki/plans/question-gen.md | 9 + research-wiki/plans/tree-repair-resilience.md | 9 + 9 files changed, 1915 insertions(+), 3 deletions(-) create mode 100644 research-wiki/designs/2026-07-07-question-gen-design.md create mode 100644 research-wiki/designs/tree-repair-resilience.md create mode 100644 research-wiki/plans/2026-07-07-question-gen.md create mode 100644 research-wiki/plans/2026-07-09-tree-repair-resilience.md create mode 100644 research-wiki/plans/question-gen.md create mode 100644 research-wiki/plans/tree-repair-resilience.md diff --git a/research-wiki/designs/2026-07-07-question-gen-design.md b/research-wiki/designs/2026-07-07-question-gen-design.md new file mode 100644 index 0000000..16070e7 --- /dev/null +++ b/research-wiki/designs/2026-07-07-question-gen-design.md @@ -0,0 +1,180 @@ +--- +id: question-gen +title: 出题模块迁移设计（question_gen） +type: design +created: 2026-07-07 +status: approved +--- + +# 出题模块迁移设计 + +## 1. 目标 + +从 TRM4 `core/harness/question_gen.py` 迁移出题数据结构与采样逻辑到 TRM5 Clean Architecture，同时预留 LLM 驱动出题的 Protocol 接口。 + +| 维度 | 说明 | +|------|------| +| 迁移范围 | benchmark 加载 + 分层采样（纯函数，180 行） | +| 预留接口 | `QuestionGenerator` Protocol（不实现，后续参考 TRM4 `research-wiki/designs/2026-07-06-question-gen-synth-design.md`） | +| 不做 | LLM 出题实现、校准脚本、去重机制 | + +## 2. Clean Architecture 分层决策 + +### 2.1 类型放置 + +`GeneratedQuestion` 被 `core/evolution/`（diagnose、validate）和 `app/harness/`（runner、batching、pools、inference）跨层使用。按依赖方向（core 不可依赖 app），必须放 `core/types.py`，与 `LLMResponse` 同级。 + +```mermaid +flowchart LR + CT["core/types.py\nGeneratedQuestion"] --> CE["core/evolution/\ndiagnose · validate"] + CT --> AH["app/harness/\nrunner · batching · pools"] + CT --> AQ["app/question_gen/\nloader"] +``` + +### 2.2 模块结构 + +``` +core/types.py ← 追加 GeneratedQuestion +app/ports.py ← 追加 QuestionGenerator Protocol +app/question_gen/ +├── __init__.py ← 公开 API re-export +└── loader.py ← load_benchmark() + stratified_sample() +``` + +**否决方案**： + +| 方案 | 否决理由 | +|------|---------| +| `GeneratedQuestion` 放 `app/question_gen/types.py` | `core/evolution/` 无法 import `app/` 层，违反依赖方向 | +| loader / sampler 拆两文件 | sampler 仅 ~100 行，不值得独立文件 | +| Protocol 放 `app/question_gen/protocols.py` | 与 `EmbeddingProvider` 在 `app/ports.py` 的既有模式不一致 | + +## 3. 类型定义 + +### 3.1 GeneratedQuestion（`core/types.py` 追加） + +```python +@dataclass(frozen=True) +class GeneratedQuestion: + """单条生成/加载的题目。跨层共享类型。""" + question_id: str + video_id: str + task_type: str + question: str + options: tuple[str, ...] + answer: str + source_nodes: tuple[str, ...] + difficulty: str +``` + +**与 TRM4 的有意变更**： + +| 变更 | 理由 | +|------|------| +| `options: list → tuple` | 配合 `frozen=True` 不可变语义 | +| `source_nodes: list → tuple` | 同上 | +| `difficulty` 移除默认值 `"medium"` | 显式传入（§4.1 P4: 显式优于隐式） | + +**移除 `QuestionGenResult`**：TRM5 无消费者，YAGNI。 + +### 3.2 QuestionGenerator Protocol（`app/ports.py` 追加） + +```python +@runtime_checkable +class QuestionGenerator(Protocol): + """LLM 驱动的题目生成端口（预留接口）。""" + async def generate( + self, + video_id: str, + task_type: str, + tree: TreeIndex, + *, + exemplars: list[GeneratedQuestion], + ) -> GeneratedQuestion: ... +``` + +接口设计参考 TRM4 仓库 `research-wiki/designs/2026-07-06-question-gen-synth-design.md`（位于 `/home/iomgaa/Projects/Video-Tree-TRM4/`，不复制到 TRM5）中的"题型-层级映射 + few-shot exemplar"模式。`tree` 参数提供锚节点上下文，`exemplars` 提供风格示例。具体实现在后续 `tools/generate_questions.py`（一次性脚本）中完成，通过 `adapters/` 层的 Protocol 实现注入。 + +## 4. 函数接口 + +### 4.1 load_benchmark + +``` +load_benchmark(questions_dir: Path) -> list[GeneratedQuestion] +``` + +从指定目录 glob `*.json`，每个文件以 `stem` 为 `video_id`，解析为 `GeneratedQuestion` 列表。JSON 格式与 `store/questions/benchmarks/Video-MME/*.json` 完全一致。 + +**与 TRM4 对比**：算法 100% 保真。`options` 和 `source_nodes` 转为 `tuple`。 + +**`difficulty` 字段处理规则**：现有 benchmark JSON（`store/questions/benchmarks/Video-MME/`）不含 `difficulty` 字段，这是 legacy schema 特征。加载时按如下规则显式转换（非默认值掩盖）： + +| JSON 情况 | 处理 | +|-----------|------| +| 有 `difficulty` 字段 | 取 JSON 值 | +| 无 `difficulty` 字段 | 赋 `_LEGACY_DEFAULT_DIFFICULTY = "medium"` 常量 | + +常量集中定义在 `loader.py` 顶部，测试用例覆盖两种情况。 + +### 4.2 stratified_sample + +``` +stratified_sample( + questions: list[GeneratedQuestion], + correctness: dict[str, bool], + size: int, + correct_ratio: float | None, + task_types: list[str] | None, + seed: int, + min_per_class: int | None, +) -> list[GeneratedQuestion] +``` + +所有参数显式传入，无默认值（§4.1 P4）。 + +**算法保真清单**（逐一比对 TRM4）： + +| 逻辑点 | TRM4 行为 | TRM5 保持 | +|--------|----------|----------| +| `task_types` 过滤 | `task_types` 非 None 时，先过滤 pool 只保留指定题型 | 保持 | +| `correct_ratio=None` | 自然分布分支，随机抽样 `size` 道 | 保持 | +| `correct_ratio` 有值 | 按对错比例分层，对题 `round(size * ratio)` | 保持 | +| `correctness.get(id, False)` | 未知 correctness 的题统一当错题处理 | 保持 | +| 分层返回顺序 | 对题在前、错题在后 | 保持 | +| 池不足 | `ValueError` 报错，不静默降级 | 保持 | +| `min_per_class` 补足 | 遍历 pool 全部题型（非仅 sampled 命中的），按首次出现顺序确定性枚举 | 保持 | +| 补足不足时 | 全取，不报错 | 保持 | +| 随机种子 | `random.Random(seed)` 局部实例 | 保持 | + +内部辅助函数 `_ratio_stratified_sample` 和 `_backfill_per_class` 完整保留。 + +## 5. 职责边界 + +| 组件 | 职责 | 位置 | 谁 import 谁 | +|------|------|------|-------------| +| `GeneratedQuestion` | 题目数据结构 | `core/types.py` | 被所有层 import | +| `load_benchmark` / `stratified_sample` | 加载 + 采样 | `app/question_gen/loader.py` | 被 `app/harness/` import | +| `QuestionGenerator` Protocol | LLM 出题接口定义 | `app/ports.py` | 被未来 `adapters/` 实现 | +| `tools/generate_questions.py`（未来） | LLM 出题一次性脚本 | `tools/` | 独立工具，不被其他模块 import | + +`tools/generate_questions.py` 未来可实例化 `QuestionGenerator` 的 adapter 实现，但 `tools/` 本身不被 `app/` import（§5 硬性规则）。 + +## 6. 文档同步 + +以下章节需要更新： + +| 文档 | 章节 | 变更 | +|------|------|------| +| `ARCHITECTURE.md` §1 表格 | DataLoader 行 `app/question_gen/generator.py` | → `app/question_gen/loader.py` | +| `ARCHITECTURE.md` §2.2 Mermaid | `QGEN` 节点 `generator.py` | → `loader.py` | +| `CLAUDE.md` §1.5 表格 | DataLoader 行 `app/question_gen/generator.py` | → `app/question_gen/loader.py` | + +**不变更**：`ARCHITECTURE.md §6` 核心算法保真清单 — `stratified_sample` 是采样工具函数，不属于 13 项核心算法（那些是建树 + 训练的关键算法）。 + +## 7. 测试策略 + +| 测试 | 路径 | 覆盖点 | +|------|------|--------| +| `GeneratedQuestion` 冻结性 | `tests/unit/test_core_types.py`（追加） | frozen 不可变、字段完整性 | +| `load_benchmark` | `tests/unit/test_question_loader.py` | 正常加载、空目录、JSON 格式异常 | +| `stratified_sample` | `tests/unit/test_question_loader.py` | 自然分布、分层采样、题型过滤、未知 correctness 当错题、对题在前返回顺序、题型保底、池不足报错、种子可复现 | diff --git a/research-wiki/designs/tree-repair-resilience.md b/research-wiki/designs/tree-repair-resilience.md new file mode 100644 index 0000000..e3298c9 --- /dev/null +++ b/research-wiki/designs/tree-repair-resilience.md @@ -0,0 +1,9 @@ +--- +type: design +node_id: design:tree-repair-resilience +title: "建树修复管线：熔断根因修复 + 断点续跑 + 并发改造" +date: 2026-07-09 +--- + +# 建树修复管线：熔断根因修复 + 断点续跑 + 并发改造 + diff --git a/research-wiki/graph/edges.json b/research-wiki/graph/edges.json index 1a3c7d2..4e23f0b 100644 --- a/research-wiki/graph/edges.json +++ b/research-wiki/graph/edges.json @@ -45,6 +45,16 @@ "id": "plan:app-harness", "label": "app/harness/ 训练循环编排层实现计划", "type": "plan" + }, + { + "id": "design:tree-repair-resilience", + "label": "建树修复管线：熔断根因修复 + 断点续跑 + 并发改造", + "type": "design" + }, + { + "id": "plan:tree-repair-resilience", + "label": "建树修复管线三项改造实现计划", + "type": "plan" } ], "links": [ @@ -82,6 +92,13 @@ "relation": "implements", "evidence": "实现 2026-07-07-app-harness-design.md 的 14 文件训练循环编排层", "added": "2026-07-07T15:45:08.729979+00:00" + }, + { + "source": "plan:tree-repair-resilience", + "target": "design:tree-repair-resilience", + "relation": "implements", + "evidence": "实现设计文档的三项改造：遥测加固+断点续跑+并发", + "added": "2026-07-09T04:08:15.312470+00:00" } ] } \ No newline at end of file diff --git a/research-wiki/index.md b/research-wiki/index.md index 08eba39..b25d16f 100644 --- a/research-wiki/index.md +++ b/research-wiki/index.md @@ -1,25 +1,29 @@ # Research Wiki 索引 -> 自动生成，更新时间：2026-07-07 15:45 UTC +> 自动生成，更新时间：2026-07-09 04:08 UTC -## design (7) +## design (9) - [2026-07-06-core-agent-adapters-llm-design](designs/2026-07-06-core-agent-adapters-llm-design.md) `design:2026-07-06-core-agent-adapters-llm-design` - [2026-07-07-app-harness-design](designs/2026-07-07-app-harness-design.md) `design:2026-07-07-app-harness-design` - [2026-07-07-core-evolution-design](designs/2026-07-07-core-evolution-design.md) `design:2026-07-07-core-evolution-design` +- [2026-07-08-tree-repair-resilience-design](designs/2026-07-08-tree-repair-resilience-design.md) `design:2026-07-08-tree-repair-resilience-design` - [出题模块迁移设计（question_gen）](designs/2026-07-07-question-gen-design.md) `design:2026-07-07-question-gen-design` +- [建树修复管线：熔断根因修复 + 断点续跑 + 并发改造](designs/tree-repair-resilience.md) `design:tree-repair-resilience` - [建树模块竖切设计：数据结构 + 建树 + 修复 + 迁移](designs/2026-07-07-tree-module-design.md) `design:2026-07-07-tree-module-design` - [建树模块竖切设计：数据结构 + 建树 + 修复 + 迁移](designs/tree-module-vertical-slice.md) `design:tree-module-vertical-slice` - [搜索 Agent 装配层设计（app/search/）](designs/2026-07-07-search-module-design.md) `design:2026-07-07-search-module-design` -## plan (11) +## plan (13) - [2026-07-06-core-agent-adapters-llm](plans/2026-07-06-core-agent-adapters-llm.md) `plan:2026-07-06-core-agent-adapters-llm` - [2026-07-07-app-harness](plans/2026-07-07-app-harness.md) `plan:2026-07-07-app-harness` - [2026-07-07-core-evolution](plans/2026-07-07-core-evolution.md) `plan:2026-07-07-core-evolution` - [2026-07-07-question-gen](plans/2026-07-07-question-gen.md) `plan:2026-07-07-question-gen` - [2026-07-07-tree-module-vertical-slice](plans/2026-07-07-tree-module-vertical-slice.md) `plan:2026-07-07-tree-module-vertical-slice` +- [2026-07-09-tree-repair-resilience](plans/2026-07-09-tree-repair-resilience.md) `plan:2026-07-09-tree-repair-resilience` - [app/harness/ 训练循环编排层实现计划](plans/app-harness.md) `plan:app-harness` - [app/search/ 搜索 Agent 装配层实现计划](plans/2026-07-07-search-module.md) `plan:2026-07-07-search-module` - [core/agent/ + adapters/llm 基础设施实现计划](plans/core-agent-adapters-llm.md) `plan:core-agent-adapters-llm` - [question_gen 模块实现计划](plans/question-gen.md) `plan:question-gen` +- [建树修复管线三项改造实现计划](plans/tree-repair-resilience.md) `plan:tree-repair-resilience` - [建树模块竖切实现计划](plans/tree-module-vertical-slice.md) `plan:tree-module-vertical-slice` - [项目基础设施初始化计划](plans/infrastructure-setup.md) `plan:infrastructure-setup` diff --git a/research-wiki/log.md b/research-wiki/log.md index 13959b3..b8ee52f 100644 --- a/research-wiki/log.md +++ b/research-wiki/log.md @@ -23,3 +23,7 @@ - [2026-07-07 15:45 UTC] 新增 plan: app/harness/ 训练循环编排层实现计划 (plan:app-harness) - [2026-07-07 15:45 UTC] 新增边: plan:app-harness --implements--> design:app-harness-design - [2026-07-07 15:45 UTC] 重建索引: 18 篇页面 +- [2026-07-09 03:29 UTC] 新增 design: 建树修复管线：熔断根因修复 + 断点续跑 + 并发改造 (design:tree-repair-resilience) +- [2026-07-09 04:08 UTC] 新增 plan: 建树修复管线三项改造实现计划 (plan:tree-repair-resilience) +- [2026-07-09 04:08 UTC] 新增边: plan:tree-repair-resilience --implements--> design:tree-repair-resilience +- [2026-07-09 04:08 UTC] 重建索引: 22 篇页面 diff --git a/research-wiki/plans/2026-07-07-question-gen.md b/research-wiki/plans/2026-07-07-question-gen.md new file mode 100644 index 0000000..a6d8413 --- /dev/null +++ b/research-wiki/plans/2026-07-07-question-gen.md @@ -0,0 +1,894 @@ +# question_gen 模块实现计划 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 从 TRM4 迁移出题数据结构与采样逻辑到 TRM5 Clean Architecture，预留 LLM 出题 Protocol。 + +**Architecture:** `GeneratedQuestion` 放 `core/types.py`（跨层共享），加载和采样逻辑放 `app/question_gen/loader.py`，`QuestionGenerator` Protocol 追加到 `app/ports.py`。 + +**Tech Stack:** Python 3.11, dataclasses, pytest, loguru + +**设计文档:** `research-wiki/designs/2026-07-07-question-gen-design.md` + +**核心算法保真:** 本计划不涉及 ARCHITECTURE.md §6 中 13 项核心算法的迁移。`stratified_sample` 是采样工具函数，不在保真清单内，但仍逐行比对 TRM4 实现保证行为一致。 + +--- + +### Task 1: GeneratedQuestion 数据类型 + +**Files:** +- Modify: `core/types.py` +- Modify: `tests/unit/test_core_types.py` + +- [ ] **Step 1: 在 test_core_types.py 追加 GeneratedQuestion 测试** + +```python +from core.types import GeneratedQuestion + + +class TestGeneratedQuestion: + @pytest.fixture() + def sample_question(self) -> GeneratedQuestion: + return GeneratedQuestion( + question_id="719-1", + video_id="B7Hh0PY1kks", + task_type="Action Reasoning", + question="What are the differing motivations?", + options=("A. Option 1", "B. Option 2", "C. Option 3", "D. Option 4"), + answer="B", + source_nodes=(), + difficulty="medium", + ) + + def test_frozen_prevents_mutation(self, sample_question: GeneratedQuestion) -> None: + with pytest.raises(AttributeError): + sample_question.question = "篡改" + + def test_all_fields_accessible(self, sample_question: GeneratedQuestion) -> None: + assert sample_question.question_id == "719-1" + assert sample_question.video_id == "B7Hh0PY1kks" + assert sample_question.task_type == "Action Reasoning" + assert sample_question.question == "What are the differing motivations?" + assert sample_question.options == ("A. Option 1", "B. Option 2", "C. Option 3", "D. Option 4") + assert sample_question.answer == "B" + assert sample_question.source_nodes == () + assert sample_question.difficulty == "medium" + + def test_options_is_tuple(self, sample_question: GeneratedQuestion) -> None: + assert isinstance(sample_question.options, tuple) + + def test_source_nodes_is_tuple(self, sample_question: GeneratedQuestion) -> None: + assert isinstance(sample_question.source_nodes, tuple) +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_core_types.py::TestGeneratedQuestion -v` +Expected: FAIL — `ImportError: cannot import name 'GeneratedQuestion'` + +- [ ] **Step 3: 在 core/types.py 追加 GeneratedQuestion** + +在 `LLMResponse` 类之后追加： + +```python +@dataclass(frozen=True) +class GeneratedQuestion: + """单条生成/加载的题目。 + + 跨层共享类型，被 core/evolution/ 和 app/harness/、app/question_gen/ 使用。 + frozen=True 确保题目不可变。 + + 属性: + question_id: 题目唯一标识。 + video_id: 所属视频标识。 + task_type: 题型（如 "Action Reasoning"）。 + question: 题目文本。 + options: 选项元组（如 ("A. ...", "B. ...", "C. ...", "D. ...")）。 + answer: 正确答案字母（如 "B"）。 + source_nodes: 来源节点 ID 元组。 + difficulty: 难度等级。 + """ + + question_id: str + video_id: str + task_type: str + question: str + options: tuple[str, ...] + answer: str + source_nodes: tuple[str, ...] + difficulty: str +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_core_types.py -v` +Expected: 全部 PASS（含原有 LLMResponse 测试 + 新增 GeneratedQuestion 测试） + +- [ ] **Step 5: 提交** + +``` +feat(core): 追加 GeneratedQuestion frozen dataclass +``` + +--- + +### Task 2: load_benchmark 加载函数 + +**Files:** +- Create: `app/question_gen/loader.py` +- Create: `tests/unit/test_question_loader.py` + +- [ ] **Step 1: 编写 load_benchmark 测试** + +在 `tests/unit/test_question_loader.py` 中创建： + +```python +"""app/question_gen/loader.py 单元测试。""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from app.question_gen.loader import load_benchmark +from core.types import GeneratedQuestion + + +@pytest.fixture() +def benchmark_dir(tmp_path: Path) -> Path: + """创建包含 benchmark JSON 的临时目录。""" + data = [ + { + "question_id": "1-1", + "task_type": "Action Reasoning", + "question": "What happened?", + "options": ["A. X", "B. Y", "C. Z", "D. W"], + "answer": "A", + }, + { + "question_id": "1-2", + "task_type": "OCR Problems", + "question": "What text is shown?", + "options": ["A. Hello", "B. World", "C. Foo", "D. Bar"], + "answer": "B", + }, + ] + (tmp_path / "video_abc.json").write_text(json.dumps(data), encoding="utf-8") + return tmp_path + + +class TestLoadBenchmark: + def test_loads_questions_from_json(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + assert len(questions) == 2 + + def test_video_id_from_filename(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + assert all(q.video_id == "video_abc" for q in questions) + + def test_fields_mapped_correctly(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + q = questions[0] + assert q.question_id == "1-1" + assert q.task_type == "Action Reasoning" + assert q.question == "What happened?" + assert q.options == ("A. X", "B. Y", "C. Z", "D. W") + assert q.answer == "A" + + def test_options_is_tuple(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + assert isinstance(questions[0].options, tuple) + + def test_source_nodes_is_empty_tuple(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + assert questions[0].source_nodes == () + + def test_difficulty_defaults_to_medium_for_legacy(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + assert questions[0].difficulty == "medium" + + def test_difficulty_from_json_when_present(self, tmp_path: Path) -> None: + data = [ + { + "question_id": "2-1", + "task_type": "OCR Problems", + "question": "Q?", + "options": ["A. 1", "B. 2", "C. 3", "D. 4"], + "answer": "C", + "difficulty": "hard", + } + ] + (tmp_path / "vid.json").write_text(json.dumps(data), encoding="utf-8") + questions = load_benchmark(tmp_path) + assert questions[0].difficulty == "hard" + + def test_empty_directory_returns_empty_list(self, tmp_path: Path) -> None: + questions = load_benchmark(tmp_path) + assert questions == [] + + def test_sorted_by_filename(self, tmp_path: Path) -> None: + for name in ["z_video.json", "a_video.json"]: + data = [{"question_id": f"{name}-1", "task_type": "T", "question": "Q?", "options": ["A", "B", "C", "D"], "answer": "A"}] + (tmp_path / name).write_text(json.dumps(data), encoding="utf-8") + questions = load_benchmark(tmp_path) + assert questions[0].video_id == "a_video" + assert questions[1].video_id == "z_video" + + def test_returns_generated_question_instances(self, benchmark_dir: Path) -> None: + questions = load_benchmark(benchmark_dir) + assert all(isinstance(q, GeneratedQuestion) for q in questions) + + def test_loads_real_benchmark(self) -> None: + """使用真实 benchmark 数据验证加载正确性。""" + real_dir = Path("store/questions/benchmarks/Video-MME") + if not real_dir.exists(): + pytest.skip("真实 benchmark 数据不存在") + questions = load_benchmark(real_dir) + assert len(questions) > 0 + for q in questions: + assert isinstance(q, GeneratedQuestion) + assert len(q.options) == 4 + assert q.answer in ("A", "B", "C", "D") + + def test_malformed_json_raises(self, tmp_path: Path) -> None: + """非法 JSON 文件应抛出 json.JSONDecodeError。""" + (tmp_path / "bad.json").write_text("not valid json{{{", encoding="utf-8") + with pytest.raises(json.JSONDecodeError): + load_benchmark(tmp_path) + + def test_missing_required_field_raises(self, tmp_path: Path) -> None: + """缺少必需字段（如 question_id）应抛出 KeyError。""" + data = [{"task_type": "T", "question": "Q?", "options": ["A"], "answer": "A"}] + (tmp_path / "vid.json").write_text(json.dumps(data), encoding="utf-8") + with pytest.raises(KeyError): + load_benchmark(tmp_path) +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_question_loader.py::TestLoadBenchmark -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'app.question_gen.loader'` + +- [ ] **Step 3: 实现 loader.py 的 load_benchmark** + +创建 `app/question_gen/loader.py`： + +```python +"""题目加载与分层采样。 + +从 benchmark JSON 目录加载题目，提供按对错比例的分层采样。 +对应训练循环中的 DataLoader 角色。 +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from core.types import GeneratedQuestion + +_LEGACY_DEFAULT_DIFFICULTY = "medium" + + +def load_benchmark(questions_dir: Path) -> list[GeneratedQuestion]: + """从 benchmark JSON 目录加载题目列表。 + + 每个 JSON 文件以文件名（不含扩展名）作为 video_id， + 文件内容为题目数组。 + + 参数: + questions_dir: 包含 *.json 文件的目录路径。 + + 返回: + 按文件名排序加载的题目列表。 + """ + results: list[GeneratedQuestion] = [] + for path in sorted(questions_dir.glob("*.json")): + video_id = path.stem + with open(path, encoding="utf-8") as f: + qa_list: list[dict] = json.load(f) + for qa in qa_list: + results.append( + GeneratedQuestion( + question_id=qa["question_id"], + video_id=video_id, + task_type=qa["task_type"], + question=qa["question"], + options=tuple(qa["options"]), + answer=qa["answer"], + source_nodes=tuple(qa.get("source_nodes", ())), + difficulty=qa.get("difficulty", _LEGACY_DEFAULT_DIFFICULTY), + ) + ) + return results +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_question_loader.py::TestLoadBenchmark -v` +Expected: 全部 PASS + +- [ ] **Step 5: 提交** + +``` +feat(question_gen): load_benchmark — benchmark JSON 加载 +``` + +--- + +### Task 3: stratified_sample 分层采样 + +**Files:** +- Modify: `app/question_gen/loader.py` +- Modify: `tests/unit/test_question_loader.py` + +- [ ] **Step 1: 编写 stratified_sample 测试** + +在 `tests/unit/test_question_loader.py` 追加： + +```python +from app.question_gen.loader import stratified_sample + + +def _make_questions(n: int, task_type: str = "T") -> list[GeneratedQuestion]: + """辅助函数：批量构造题目。""" + return [ + GeneratedQuestion( + question_id=f"{task_type}-{i}", + video_id="v1", + task_type=task_type, + question=f"Q{i}?", + options=("A", "B", "C", "D"), + answer="A", + source_nodes=(), + difficulty="medium", + ) + for i in range(n) + ] + + +class TestStratifiedSample: + def test_natural_distribution(self) -> None: + """correct_ratio=None 时走自然分布随机抽样。""" + questions = _make_questions(20) + result = stratified_sample( + questions=questions, + correctness={}, + size=10, + correct_ratio=None, + task_types=None, + seed=42, + min_per_class=None, + ) + assert len(result) == 10 + + def test_natural_distribution_pool_insufficient(self) -> None: + """自然分布时池不足应 ValueError。""" + questions = _make_questions(5) + with pytest.raises(ValueError, match="自然分布采样不足"): + stratified_sample( + questions=questions, + correctness={}, + size=10, + correct_ratio=None, + task_types=None, + seed=42, + min_per_class=None, + ) + + def test_ratio_stratified(self) -> None: + """按对错比例分层采样。""" + questions = _make_questions(20) + correctness = {f"T-{i}": i < 10 for i in range(20)} + result = stratified_sample( + questions=questions, + correctness=correctness, + size=10, + correct_ratio=0.6, + task_types=None, + seed=42, + min_per_class=None, + ) + assert len(result) == 10 + correct_count = sum(1 for q in result if correctness.get(q.question_id, False)) + assert correct_count == 6 + + def test_ratio_stratified_correct_first(self) -> None: + """分层采样返回顺序：对题在前、错题在后。""" + questions = _make_questions(20) + correctness = {f"T-{i}": i < 10 for i in range(20)} + result = stratified_sample( + questions=questions, + correctness=correctness, + size=10, + correct_ratio=0.5, + task_types=None, + seed=42, + min_per_class=None, + ) + n_correct = round(10 * 0.5) + for q in result[:n_correct]: + assert correctness.get(q.question_id, False) is True + for q in result[n_correct:]: + assert correctness.get(q.question_id, False) is False + + def test_ratio_stratified_pool_insufficient(self) -> None: + """分层时对题或错题不足应 ValueError。""" + questions = _make_questions(10) + correctness = {f"T-{i}": True for i in range(10)} + with pytest.raises(ValueError, match="分层不足"): + stratified_sample( + questions=questions, + correctness=correctness, + size=10, + correct_ratio=0.5, + task_types=None, + seed=42, + min_per_class=None, + ) + + def test_task_types_filter(self) -> None: + """task_types 过滤只保留指定题型。""" + q_a = _make_questions(10, task_type="TypeA") + q_b = _make_questions(10, task_type="TypeB") + result = stratified_sample( + questions=q_a + q_b, + correctness={}, + size=5, + correct_ratio=None, + task_types=["TypeA"], + seed=42, + min_per_class=None, + ) + assert all(q.task_type == "TypeA" for q in result) + + def test_unknown_correctness_treated_as_wrong(self) -> None: + """correctness 中不存在的 question_id 被当作错题。""" + questions = _make_questions(20) + correctness = {f"T-{i}": True for i in range(10)} + result = stratified_sample( + questions=questions, + correctness=correctness, + size=10, + correct_ratio=0.5, + task_types=None, + seed=42, + min_per_class=None, + ) + n_correct = round(10 * 0.5) + for q in result[:n_correct]: + assert q.question_id in correctness + + def test_seed_reproducibility(self) -> None: + """相同种子产生相同结果。""" + questions = _make_questions(20) + r1 = stratified_sample(questions=questions, correctness={}, size=10, correct_ratio=None, task_types=None, seed=123, min_per_class=None) + r2 = stratified_sample(questions=questions, correctness={}, size=10, correct_ratio=None, task_types=None, seed=123, min_per_class=None) + assert [q.question_id for q in r1] == [q.question_id for q in r2] + + def test_different_seeds_differ(self) -> None: + """不同种子产生不同结果（概率性，但 20 选 10 几乎必然不同）。""" + questions = _make_questions(20) + r1 = stratified_sample(questions=questions, correctness={}, size=10, correct_ratio=None, task_types=None, seed=1, min_per_class=None) + r2 = stratified_sample(questions=questions, correctness={}, size=10, correct_ratio=None, task_types=None, seed=2, min_per_class=None) + assert [q.question_id for q in r1] != [q.question_id for q in r2] + + def test_min_per_class_backfill(self) -> None: + """min_per_class 补足稀疏题型。""" + q_a = _make_questions(10, task_type="TypeA") + q_b = _make_questions(10, task_type="TypeB") + all_q = q_a + q_b + correctness = {q.question_id: True for q in q_a[:5]} + result = stratified_sample( + questions=all_q, + correctness=correctness, + size=3, + correct_ratio=None, + task_types=None, + seed=42, + min_per_class=2, + ) + type_counts: dict[str, int] = {} + for q in result: + type_counts[q.task_type] = type_counts.get(q.task_type, 0) + 1 + assert type_counts.get("TypeA", 0) >= 2 + assert type_counts.get("TypeB", 0) >= 2 + + def test_min_per_class_partial_backfill(self) -> None: + """题型可补题数不足缺口时全取，不报错。""" + q_sparse = _make_questions(1, task_type="Sparse") + q_main = _make_questions(10, task_type="Main") + result = stratified_sample( + questions=q_sparse + q_main, + correctness={}, + size=5, + correct_ratio=None, + task_types=None, + seed=42, + min_per_class=3, + ) + sparse_in_result = [q for q in result if q.task_type == "Sparse"] + assert len(sparse_in_result) == 1 + + def test_min_per_class_no_duplicates(self) -> None: + """补足后不产生重复 question_id。""" + q_a = _make_questions(5, task_type="TypeA") + q_b = _make_questions(5, task_type="TypeB") + result = stratified_sample( + questions=q_a + q_b, + correctness={}, + size=3, + correct_ratio=None, + task_types=None, + seed=42, + min_per_class=2, + ) + ids = [q.question_id for q in result] + assert len(ids) == len(set(ids)) + + def test_backfill_enumerates_all_pool_types(self) -> None: + """补足遍历 pool 全部题型，包括主采样未命中的。""" + q_main = _make_questions(10, task_type="Main") + q_rare = _make_questions(3, task_type="Rare") + result = stratified_sample( + questions=q_main + q_rare, + correctness={}, + size=2, + correct_ratio=None, + task_types=None, + seed=0, + min_per_class=1, + ) + types_in_result = {q.task_type for q in result} + assert "Rare" in types_in_result +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_question_loader.py::TestStratifiedSample -v` +Expected: FAIL — `ImportError: cannot import name 'stratified_sample'` + +- [ ] **Step 3: 实现 stratified_sample 及内部辅助函数** + +在 `app/question_gen/loader.py` 的导入区追加 `import random`（标准库，放在 `import json` 之后），然后在文件末尾追加以下函数： + +```python +def stratified_sample( + questions: list[GeneratedQuestion], + correctness: dict[str, bool], + size: int, + correct_ratio: float | None, + task_types: list[str] | None, + seed: int, + min_per_class: int | None, +) -> list[GeneratedQuestion]: + """按题型过滤后采样 size 道题，可选按对错比例分层并按题型保底。 + + 参数: + questions: 候选题目全集。 + correctness: question_id → 基线是否答对。 + size: 采样总量。 + correct_ratio: 采样中"基线答对"题的占比；None 表示自然分布。 + task_types: 限定题型；None 表示不限。 + seed: 随机种子，保证可复现。 + min_per_class: 每个题型补足到的下限；None 表示不补足。 + + 返回: + 采样后的题目列表。 + + 异常: + ValueError: 自然分布时池不足 size，或分层时某层题目不足。 + """ + rng = random.Random(seed) + pool = [q for q in questions if task_types is None or q.task_type in task_types] + + if correct_ratio is None: + if len(pool) < size: + raise ValueError(f"自然分布采样不足: 需 {size} 道, 实有 {len(pool)} 道") + sampled = rng.sample(pool, size) + else: + sampled = _ratio_stratified_sample(pool, correctness, size, correct_ratio, rng) + + if min_per_class is not None: + sampled = _backfill_per_class(sampled, pool, min_per_class, rng) + return sampled + + +def _ratio_stratified_sample( + pool: list[GeneratedQuestion], + correctness: dict[str, bool], + size: int, + correct_ratio: float, + rng: random.Random, +) -> list[GeneratedQuestion]: + """按对错比例分层采样：对题占 correct_ratio，其余为错题。 + + 参数: + pool: 题型过滤后的候选题。 + correctness: question_id → 基线是否答对。 + size: 采样总量。 + correct_ratio: 对题占比。 + rng: 随机数发生器。 + + 返回: + 采样后的题目列表（对题在前、错题在后）。 + + 异常: + ValueError: 对题或错题层不足。 + """ + correct = [q for q in pool if correctness.get(q.question_id, False)] + wrong = [q for q in pool if not correctness.get(q.question_id, False)] + n_correct = round(size * correct_ratio) + n_wrong = size - n_correct + if len(correct) < n_correct or len(wrong) < n_wrong: + raise ValueError( + f"分层不足: 需对{n_correct}/错{n_wrong}, " + f"实有对{len(correct)}/错{len(wrong)}" + ) + return rng.sample(correct, n_correct) + rng.sample(wrong, n_wrong) + + +def _backfill_per_class( + sampled: list[GeneratedQuestion], + pool: list[GeneratedQuestion], + min_per_class: int, + rng: random.Random, +) -> list[GeneratedQuestion]: + """对候选池中出现的每个题型，将采样结果补足到 min_per_class 道。 + + 遍历对象是候选池 pool 里出现的全部题型（非仅 sampled 命中的）， + 保证任意稀疏题型都能拿到足额样本。 + + 参数: + sampled: 主采样结果（不修改，返回新列表）。 + pool: 候选题全集（补足来源 + 题型枚举来源）。 + min_per_class: 每个题型的下限。 + rng: 随机数发生器。 + + 返回: + 补足后的题目列表。 + """ + selected_ids = {q.question_id for q in sampled} + result = list(sampled) + counts: dict[str, int] = {} + for q in sampled: + counts[q.task_type] = counts.get(q.task_type, 0) + 1 + ordered_task_types: dict[str, None] = {} + for q in pool: + ordered_task_types.setdefault(q.task_type, None) + for task_type in ordered_task_types: + deficit = min_per_class - counts.get(task_type, 0) + if deficit <= 0: + continue + candidates = [ + q + for q in pool + if q.task_type == task_type and q.question_id not in selected_ids + ] + take = rng.sample(candidates, min(deficit, len(candidates))) + for q in take: + selected_ids.add(q.question_id) + result.append(q) + return result +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_question_loader.py -v` +Expected: 全部 PASS（TestLoadBenchmark + TestStratifiedSample） + +- [ ] **Step 5: 提交** + +``` +feat(question_gen): stratified_sample — 分层采样 + 题型保底 +``` + +--- + +### Task 4: QuestionGenerator Protocol 与模块公开 API + +**Files:** +- Modify: `app/ports.py` +- Modify: `app/question_gen/__init__.py` +- Create: `tests/unit/test_question_gen_api.py` + +- [ ] **Step 1: 编写 Protocol 可导入性和 __init__ 公开 API 测试** + +创建 `tests/unit/test_question_gen_api.py`： + +```python +"""app/ports.py QuestionGenerator Protocol 与 app/question_gen 公开 API 测试。""" +from __future__ import annotations + +import importlib +from typing import runtime_checkable + +from app.ports import QuestionGenerator +from core.types import GeneratedQuestion + + +class TestQuestionGeneratorProtocol: + def test_importable(self) -> None: + """QuestionGenerator 可从 app.ports 导入。""" + assert QuestionGenerator is not None + + def test_is_runtime_checkable(self) -> None: + """QuestionGenerator 是 runtime_checkable Protocol。""" + assert hasattr(QuestionGenerator, "__protocol_attrs__") or hasattr( + QuestionGenerator, "__abstractmethods__" + ) + + def test_generate_method_exists(self) -> None: + """Protocol 定义了 generate 方法。""" + assert hasattr(QuestionGenerator, "generate") + + +class TestQuestionGenPublicAPI: + def test_load_benchmark_importable_from_package(self) -> None: + """load_benchmark 可从 app.question_gen 直接导入。""" + mod = importlib.import_module("app.question_gen") + assert hasattr(mod, "load_benchmark") + + def test_stratified_sample_importable_from_package(self) -> None: + """stratified_sample 可从 app.question_gen 直接导入。""" + mod = importlib.import_module("app.question_gen") + assert hasattr(mod, "stratified_sample") + + def test_all_exports(self) -> None: + """__all__ 包含预期的公开 API。""" + mod = importlib.import_module("app.question_gen") + assert set(mod.__all__) == {"load_benchmark", "stratified_sample"} +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_question_gen_api.py -v` +Expected: FAIL — `ImportError: cannot import name 'QuestionGenerator' from 'app.ports'` + +- [ ] **Step 3: 在 app/ports.py 追加 QuestionGenerator Protocol** + +将已有的 `if TYPE_CHECKING:` 块扩展，追加 `TreeIndex` 和 `GeneratedQuestion` 导入，然后在 `EmbeddingProvider` 之后追加： + +```python +@runtime_checkable +class QuestionGenerator(Protocol): + """LLM 驱动的题目生成端口（预留接口）。 + + 参数: + video_id: 视频标识。 + task_type: 题型。 + tree: 视频树索引，提供锚节点上下文。 + exemplars: 风格示例题目列表。 + + 返回: + 生成的单条题目。 + """ + + async def generate( + self, + video_id: str, + task_type: str, + tree: TreeIndex, + *, + exemplars: list[GeneratedQuestion], + ) -> GeneratedQuestion: ... +``` + +合并后的 `TYPE_CHECKING` 块： + +```python +if TYPE_CHECKING: + import numpy as np + + from app.tree.index import TreeIndex + from core.types import GeneratedQuestion +``` + +- [ ] **Step 4: 更新 app/question_gen/__init__.py 公开 API** + +```python +"""出题模块 — benchmark 加载与分层采样。""" + +from app.question_gen.loader import load_benchmark, stratified_sample + +__all__ = ["load_benchmark", "stratified_sample"] +``` + +- [ ] **Step 5: 运行测试确认通过** + +Run: `conda activate Video-Tree-TRM & pytest tests/unit/test_question_gen_api.py -v` +Expected: 全部 PASS + +- [ ] **Step 6: 运行全量测试确认无回归** + +Run: `conda activate Video-Tree-TRM & pytest tests/ -v` +Expected: 全部 PASS + +- [ ] **Step 7: 提交** + +``` +feat(question_gen): QuestionGenerator Protocol + 模块公开 API +``` + +--- + +### Task 5: 文档同步与 lint + +**Files:** +- Modify: `research-wiki/ARCHITECTURE.md` +- Modify: `CLAUDE.md` + +- [ ] **Step 1: 更新 ARCHITECTURE.md** + +需要修改 4 处： + +1. **§1 表格**（第 17 行附近）： + +``` +| DataLoader | 出题 question_gen | `app/question_gen/generator.py` | +``` +→ +``` +| DataLoader | 出题 question_gen | `app/question_gen/loader.py` | +``` + +2. **§2.2 Mermaid**（第 83 行附近）： + +``` +CLI --> QGEN["app/question_gen/generator.py\n新题构建"] +``` +→ +``` +CLI --> QGEN["app/question_gen/loader.py\n新题构建"] +``` + +3. **§2.3 目录树**（第 132 行附近）： + +``` +│ │ ├── question_gen.py # 数据加载、三池切分 +``` + +此行描述 `harness/` 内部的数据加载，但在 TRM5 中数据加载已移至 `question_gen/loader.py`。删除此行（`harness/` 的三池切分模块在未来开发 harness 时再规划）。 + +4. **§2.3 目录树**（第 138-141 行）： + +``` +│ ├── question_gen/ # 模块3：新题构建 +│ │ ├── generator.py # 题目生成 +│ │ ├── calibrator.py # 基线校准 +│ │ └── dedup.py # 去重 +``` +→ +``` +│ ├── question_gen/ # 模块3：出题（加载 + 采样 + 未来 LLM 生成） +│ │ └── loader.py # benchmark 加载、分层采样 +``` + +- [ ] **Step 2: 更新 CLAUDE.md** + +1. **§1.5 表格**（第 22 行附近）： + +``` +| `DataLoader` | 出题 question_gen | `app/question_gen/generator.py` | +``` +→ +``` +| `DataLoader` | 出题 question_gen | `app/question_gen/loader.py` | +``` + +- [ ] **Step 3: 运行 lint** + +Run: `conda activate Video-Tree-TRM & ruff check app/ core/ --fix && ruff format app/ core/` +Expected: 无错误或仅自动修复 + +- [ ] **Step 4: 运行全量测试** + +Run: `conda activate Video-Tree-TRM & pytest tests/ -v` +Expected: 全部 PASS + +- [ ] **Step 5: 提交** + +``` +docs: 同步 question_gen 模块路径到 ARCHITECTURE.md 和 CLAUDE.md +``` diff --git a/research-wiki/plans/2026-07-09-tree-repair-resilience.md b/research-wiki/plans/2026-07-09-tree-repair-resilience.md new file mode 100644 index 0000000..3b490cf --- /dev/null +++ b/research-wiki/plans/2026-07-09-tree-repair-resilience.md @@ -0,0 +1,786 @@ +# 建树修复管线三项改造 实现计划 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 根治遥测主键冲突导致的熔断误触发，为修复管线增加视频级断点续跑和 16 路并发。 + +**Architecture:** 三层改动互相独立——(1) adapters 层遥测防御加固 + call_id 根因修复，对所有 GovernedLLMClient 使用方生效；(2) detector 扩展 L2/L1 空字段检测，零 LLM 成本；(3) tools/repair_trees.py 编排层并发 + 断点续跑。 + +**Tech Stack:** Python 3.11, asyncio, sqlite3, loguru + +**本计划不涉及核心算法迁移，保真校验不适用。** + +--- + +### Task 1: 遥测防御性加固 + +**Files:** +- Modify: `adapters/telemetry.py:47-55`（`_INSERT_SQL`）、`adapters/telemetry.py:69-114`（`_write`） +- Test: `tests/unit/test_telemetry.py` + +- [ ] **Step 1: 写失败测试 — 重复 call_id 写入不抛异常** + +在 `tests/unit/test_telemetry.py` 末尾追加： + +```python +@pytest.mark.asyncio +async def test_duplicate_call_id_does_not_raise(recorder, db_path): + """重复 call_id 写入应静默忽略（INSERT OR IGNORE），不抛异常。""" + kwargs = _make_call_kwargs() + await recorder.record_llm_call(**kwargs) + # 第二次用相同 call_id 写入不应抛异常 + await recorder.record_llm_call(**kwargs) + + conn = sqlite3.connect(str(db_path)) + rows = conn.execute("SELECT COUNT(*) FROM llm_calls").fetchone() + conn.close() + assert rows[0] == 1 # 只有一条记录 +``` + +- [ ] **Step 2: 运行测试验证失败** + +运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_telemetry.py::test_duplicate_call_id_does_not_raise -v` +预期: FAIL — `sqlite3.IntegrityError: UNIQUE constraint failed` + +- [ ] **Step 3: 写失败测试 — DB 错误不冒泡** + +```python +@pytest.mark.asyncio +async def test_db_error_does_not_propagate(tmp_path): + """SQLite 写入失败时 record_llm_call 应静默降级（logger.warning），不抛异常。""" + # 用目录路径当 db_path —— sqlite3.connect 对目录名会在真正操作时报错 + bad_recorder = SQLiteTelemetryRecorder(db_path=tmp_path / "nonexistent_dir" / "bad.db") + kwargs = _make_call_kwargs() + # 不应抛异常 + await bad_recorder.record_llm_call(**kwargs) +``` + +- [ ] **Step 4: 运行测试验证失败** + +运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_telemetry.py::test_db_error_does_not_propagate -v` +预期: FAIL — `sqlite3.OperationalError: unable to open database file` + +- [ ] **Step 5: 写失败测试 — 并发写不报锁错** + +```python +@pytest.mark.asyncio +async def test_concurrent_writes_no_lock_error(recorder, db_path): + """16 路并发 record_llm_call 应全部成功，无 database is locked 错误。""" + import asyncio + + tasks = [] + for _ in range(16): + kwargs = _make_call_kwargs() # 每次生成不同 call_id + tasks.append(recorder.record_llm_call(**kwargs)) + await asyncio.gather(*tasks) + + conn = sqlite3.connect(str(db_path)) + count = conn.execute("SELECT COUNT(*) FROM llm_calls").fetchone()[0] + conn.close() + assert count == 16 +``` + +- [ ] **Step 6: 运行测试验证失败** + +运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_telemetry.py::test_concurrent_writes_no_lock_error -v` +预期: 可能 PASS（取决于并发时序）或 FAIL — `database is locked` + +- [ ] **Step 7: 实现遥测三层加固** + +修改 `adapters/telemetry.py`： + +1. `_INSERT_SQL`: `INSERT INTO` → `INSERT OR IGNORE INTO` +2. `_write` 方法: `sqlite3.connect()` 后追加 WAL + busy_timeout pragma，整个方法体包 `try/except sqlite3.Error` + +```python +_INSERT_SQL = """ + INSERT OR IGNORE INTO llm_calls ( + call_id, parent_call_id, session_id, + model_name, provider, messages, response, thinking, + prompt_tokens, completion_tokens, latency_ms, + ttft_ms, max_inter_token_ms, + cache_hit, error + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?); + """ + +def _write(self, *, call_id, parent_call_id, session_id, model_name, + provider, messages, response, thinking, prompt_tokens, + completion_tokens, latency_ms, ttft_ms, max_inter_token_ms, + cache_hit, error): + """同步写入一条 LLM 调用记录到 SQLite。""" + try: + conn = sqlite3.connect(str(self._db_path), timeout=10.0) + try: + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=5000") + self._ensure_table(conn) + conn.execute( + self._INSERT_SQL, + (call_id, parent_call_id, session_id, model_name, + provider, messages, response, thinking, prompt_tokens, + completion_tokens, latency_ms, ttft_ms, + max_inter_token_ms, int(cache_hit), error), + ) + conn.commit() + finally: + conn.close() + except sqlite3.Error as exc: + logger.warning("遥测写入失败（已降级），call_id={}: {}", call_id, exc) +``` + +在文件顶部添加 `from loguru import logger`。 + +- [ ] **Step 8: 运行全部遥测测试验证通过** + +运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_telemetry.py -v` +预期: 全部 PASS（含三个新测试） + +- [ ] **Step 9: 提交** + +```bash +git add adapters/telemetry.py tests/unit/test_telemetry.py +git commit -m "fix(telemetry): INSERT OR IGNORE + WAL + try/except 三层防御加固 + +根治遥测写入主键冲突(UNIQUE constraint)和并发写锁(database is locked) +导致的异常冒泡，遥测侧信道错误不再污染 LLM 重试链。" +``` + +--- + +### Task 2: GovernedLLMClient call_id 移入重试循环 + +**Files:** +- Modify: `adapters/llm.py:298-299`（`call_id` 生成位置）、`adapters/llm.py:358-369`（成功路径 `response.call_id`） +- Test: `tests/unit/test_governed_llm.py` + +- [ ] **Step 1: 更新现有测试断言 — call_id 应每次重试独立** + +`test_governed_llm.py:200-202` 当前断言 `assert len(call_ids) == 1`（所有重试记录共用一个 call_id），需要反转为 `assert len(call_ids) == 3`（每次 attempt 独立）。 + +找到 `test_transient_error_retries_and_records_telemetry` 中： + +```python + # 所有遥测记录应使用同一个 call_id（Important 2 修复验证） + call_ids = {c["call_id"] for c in telemetry.calls} + assert len(call_ids) == 1 +``` + +替换为： + +```python + # 每次 attempt 应使用独立的 call_id（根因修复：防遥测主键冲突） + call_ids = {c["call_id"] for c in telemetry.calls} + assert len(call_ids) == 3 # 2 次失败 + 1 次成功 = 3 个独立 call_id +``` + +- [ ] **Step 2: 运行测试验证失败** + +运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_governed_llm.py::test_transient_error_retries_and_records_telemetry -v` +预期: FAIL — `assert 1 == 3`（当前还是共用一个 call_id） + +- [ ] **Step 3: 将 call_id 生成移入重试循环** + +修改 `adapters/llm.py`。将第 298-299 行的 `call_id = str(uuid4())` **从重试循环外移入循环内**： + +找到（约 line 298-338）： + +```python + # ② call_id 生成 + call_id = str(uuid4()) + + # ③ 缓存查询（cache 为 None 时跳过） + cached = await self._cache.get(self._model, messages) if self._cache is not None else None + if cached is not None: + ... + return response + + # ④ 重试循环 + 流式消费 + last_exc: Exception | None = None + for attempt in range(self._max_retries): +``` + +改为： + +```python + # ② 缓存查询（cache 为 None 时跳过）— call_id 在缓存路径独立生成 + cached = await self._cache.get(self._model, messages) if self._cache is not None else None + if cached is not None: + cache_call_id = str(uuid4()) + response = LLMResponse( + ... + call_id=cache_call_id, + ) + await self._telemetry.record_llm_call( + call_id=cache_call_id, + ... + ) + return response + + # ③ 重试循环 + 流式消费（每次 attempt 独立 call_id） + last_exc: Exception | None = None + for attempt in range(self._max_retries): + call_id = str(uuid4()) + attempt_start = time.monotonic() +``` + +注意：缓存命中路径的 `call_id` 改为独立的 `cache_call_id`（保持在循环外生成，因为不走重试）。 + +- [ ] **Step 4: 运行全部 GovernedLLM 测试验证通过** + +运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_governed_llm.py -v` +预期: 全部 PASS + +- [ ] **Step 5: 提交** + +```bash +git add adapters/llm.py tests/unit/test_governed_llm.py +git commit -m "fix(llm): call_id 移入重试循环，每 attempt 独立 + +消除重试时遥测主键冲突的根因。每次 attempt 独立记录， +parent_call_id 不受影响（循环外固定），更利于事后诊断重试轨迹。" +``` + +--- + +### Task 3: 检测器扩展 L2/L1 空字段 + +**Files:** +- Modify: `app/tree/repair/detector.py:57-68`（L1 循环）、`app/tree/repair/detector.py:73-84`（L2 循环） +- Test: `tests/unit/test_repair_detector.py` + +- [ ] **Step 1: 写失败测试 — L2 event_description 为空触发 empty_field** + +在 `tests/unit/test_repair_detector.py` 追加： + +```python +def test_detects_l2_empty_event_description(): + """L2 event_description 为空应报 empty_field。""" + l3 = L3Node( + id="l1_0_l2_0_l3_0", + card=L3Card("正常帧", ["实体"], ["动作"], [], "居中", {}), + timestamp=1.0, + ) + l2 = L2Node( + id="l1_0_l2_0", + card=L2Card("", [], [], [], [], "", None), # event_description 为空 + time_range=(0.0, 10.0), + children=[l3], + ) + l1 = L1Node( + id="l1_0", + card=L1Card("正常场景", "", [], [], [], [], ""), + time_range=(0.0, 10.0), + children=[l2], + ) + index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1]) + issues = detect_issues(index) + l2_empties = [i for i in issues if i.level == 2 and i.issue_type == "empty_field"] + assert len(l2_empties) == 1 + assert "event_description" in l2_empties[0].details +``` + +- [ ] **Step 2: 写失败测试 — L1 scene_summary 为空触发 empty_field** + +```python +def test_detects_l1_empty_scene_summary(): + """L1 scene_summary 为空应报 empty_field。""" + l3 = L3Node( + id="l1_0_l2_0_l3_0", + card=L3Card("正常帧", ["实体"], ["动作"], [], "居中", {}), + timestamp=1.0, + ) + l2 = L2Node( + id="l1_0_l2_0", + card=L2Card("正常事件", [], [], [], [], "", None), + time_range=(0.0, 10.0), + children=[l3], + ) + l1 = L1Node( + id="l1_0", + card=L1Card("", "", [], [], [], [], ""), # scene_summary 为空 + time_range=(0.0, 10.0), + children=[l2], + ) + index = TreeIndex(metadata=IndexMeta("/t.mp4", "video"), roots=[l1]) + issues = detect_issues(index) + l1_empties = [i for i in issues if i.level == 1 and i.issue_type == "empty_field"] + assert len(l1_empties) == 1 + assert "scene_summary" in l1_empties[0].details +``` + +- [ ] **Step 3: 运行测试验证失败** + +运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_repair_detector.py::test_detects_l2_empty_event_description tests/unit/test_repair_detector.py::test_detects_l1_empty_scene_summary -v` +预期: 两个都 FAIL + +- [ ] **Step 4: 实现 L2/L1 空字段检测** + +修改 `app/tree/repair/detector.py` 的 `detect_issues` 函数： + +在 L2 的 `if not l2.children:` 检查之后（`continue` 之前），插入 L2 空字段检测： + +```python + for l2 in l1.children: + # L2: event_description 不为空 + if not l2.card.event_description: + issues.append( + NodeIssue( + node_id=l2.id, + level=2, + issue_type="empty_field", + details="L2 节点字段为空: event_description", + ) + ) + + # L2: children 不为空 + if not l2.children: + ... +``` + +在 L1 的 `if not l1.children:` 检查之前，插入 L1 空字段检测： + +```python + for l1 in index.roots: + # L1: scene_summary 不为空 + if not l1.card.scene_summary: + issues.append( + NodeIssue( + node_id=l1.id, + level=1, + issue_type="empty_field", + details="L1 节点字段为空: scene_summary", + ) + ) + + # L1: children 不为空 + if not l1.children: + ... +``` + +同步更新 docstring 的"检查项"列表。 + +- [ ] **Step 5: 运行全部 detector 测试验证通过** + +运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_repair_detector.py -v` +预期: 全部 PASS + +- [ ] **Step 6: 提交** + +```bash +git add app/tree/repair/detector.py tests/unit/test_repair_detector.py +git commit -m "feat(detector): 扩展空字段检测到 L2 event_description / L1 scene_summary + +断点续跑判据需要 L2/L1 层的 empty_field 检测。零 LLM 成本。" +``` + +--- + +### Task 4: 断点续跑 — progress 文件管理 + +**Files:** +- Modify: `tools/repair_trees.py`（新增 progress 读写函数 + 跳过逻辑） +- Test: `tests/unit/test_repair_progress.py`（新建） + +- [ ] **Step 1: 写失败测试 — progress 读写与幂等性** + +新建 `tests/unit/test_repair_progress.py`： + +```python +"""修复管线断点续跑 progress 管理测试。""" +from __future__ import annotations + +import asyncio +import json +from pathlib import Path + +import pytest + + +def test_load_progress_missing_file(tmp_path): + """progress 文件不存在时返回空集合。""" + from tools.repair_trees import load_progress + + result = load_progress(tmp_path / "nonexistent.json") + assert result == set() + + +def test_load_progress_valid_file(tmp_path): + """正常读取已有 progress 文件。""" + from tools.repair_trees import load_progress + + path = tmp_path / "progress.json" + path.write_text(json.dumps({"finished_video_ids": ["vid_a", "vid_b"]})) + result = load_progress(path) + assert result == {"vid_a", "vid_b"} + + +def test_load_progress_corrupted_file(tmp_path): + """损坏的 JSON 文件返回空集合（不抛异常）。""" + from tools.repair_trees import load_progress + + path = tmp_path / "progress.json" + path.write_text("{invalid json") + result = load_progress(path) + assert result == set() + + +@pytest.mark.asyncio +async def test_save_progress_atomic(tmp_path): + """save_progress 原子写入，并发调用不丢失更新。""" + from tools.repair_trees import save_progress + + path = tmp_path / "progress.json" + lock = asyncio.Lock() + + await save_progress(path, lock, "vid_a") + await save_progress(path, lock, "vid_b") + + data = json.loads(path.read_text()) + assert set(data["finished_video_ids"]) == {"vid_a", "vid_b"} + + +@pytest.mark.asyncio +async def test_save_progress_concurrent(tmp_path): + """16 路并发 save_progress 不丢失更新。""" + from tools.repair_trees import save_progress + + path = tmp_path / "progress.json" + lock = asyncio.Lock() + + tasks = [save_progress(path, lock, f"vid_{i}") for i in range(16)] + await asyncio.gather(*tasks) + + data = json.loads(path.read_text()) + assert len(data["finished_video_ids"]) == 16 + + +def test_should_skip_finished(tmp_path): + """已在 finished 集合中的视频应跳过。""" + from tools.repair_trees import should_skip_video + + finished = {"vid_a", "vid_b"} + assert should_skip_video("vid_a", finished, reaggregate_all=False) is True + assert should_skip_video("vid_c", finished, reaggregate_all=False) is False + + +def test_should_skip_reaggregate_all_forces_rerun(tmp_path): + """--reaggregate-all 标志强制不跳过。""" + from tools.repair_trees import should_skip_video + + finished = {"vid_a"} + assert should_skip_video("vid_a", finished, reaggregate_all=True) is False +``` + +- [ ] **Step 2: 运行测试验证失败** + +运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_repair_progress.py -v` +预期: FAIL — `ImportError: cannot import name 'load_progress'` + +- [ ] **Step 3: 在 repair_trees.py 实现 progress 管理函数** + +在 `tools/repair_trees.py` 的 `_build_clients()` 函数之前插入： + +```python +# --------------------------------------------------------------------------- +# 断点续跑 — progress 文件管理 +# --------------------------------------------------------------------------- + +PROGRESS_FILE = "repair_progress.json" + + +def load_progress(path: Path) -> set[str]: + """读取 progress 文件，返回已完成视频 ID 集合。 + + 参数: + path: progress JSON 文件路径。 + + 返回: + 已完成视频 ID 集合。文件不存在或损坏时返回空集。 + """ + if not path.exists(): + return set() + try: + data = json.loads(path.read_text(encoding="utf-8")) + return set(data.get("finished_video_ids", [])) + except (json.JSONDecodeError, KeyError, TypeError): + logger.warning("progress 文件损坏，忽略: {}", path) + return set() + + +async def save_progress(path: Path, lock: asyncio.Lock, vid: str) -> None: + """原子追加一个视频 ID 到 progress 文件。 + + 参数: + path: progress JSON 文件路径。 + lock: asyncio.Lock，防并发读改写丢更新。 + vid: 要追加的视频 ID。 + """ + async with lock: + finished = load_progress(path) + finished.add(vid) + tmp = path.with_suffix(".tmp") + tmp.write_text( + json.dumps({"finished_video_ids": sorted(finished)}, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + os.replace(str(tmp), str(path)) + + +def should_skip_video(vid: str, finished: set[str], *, reaggregate_all: bool) -> bool: + """判断是否跳过该视频。 + + 参数: + vid: 视频 ID。 + finished: progress 中已完成的视频 ID 集合。 + reaggregate_all: --reaggregate-all 标志。 + + 返回: + True 表示跳过。 + """ + if reaggregate_all: + return False + return vid in finished +``` + +- [ ] **Step 4: 运行测试验证通过** + +运行: `conda activate Video-Tree-TRM && pytest tests/unit/test_repair_progress.py -v` +预期: 全部 PASS + +- [ ] **Step 5: 提交** + +```bash +git add tools/repair_trees.py tests/unit/test_repair_progress.py +git commit -m "feat(repair): 断点续跑 progress 文件管理 + +load_progress / save_progress(asyncio.Lock + os.replace 原子写入) +/ should_skip_video。支持并发安全的读改写和 --reaggregate-all 兜底。" +``` + +--- + +### Task 5: 并发编排 + CLI 参数 + 熔断阈值适配 + +**Files:** +- Modify: `tools/repair_trees.py`（`main_async` 并发化 + `parse_args` 新参数 + `_build_clients` 阈值适配） +- Modify: `.env.example:44`（阈值注释更新） +- Test: 手动集成测试（`--dry-run` 模式验证并发 + 跳过逻辑） + +- [ ] **Step 1: 更新 parse_args 增加 --concurrency 和 --reaggregate-all** + +修改 `tools/repair_trees.py` 的 `parse_args` 函数，在 `--dry-run` 之后追加： + +```python + parser.add_argument( + "--concurrency", + type=int, + default=16, + help="并发修复视频数（默认: 16）", + ) + parser.add_argument( + "--reaggregate-all", + action="store_true", + help="强制全量重聚合，忽略 progress 文件", + ) +``` + +- [ ] **Step 2: 更新 _build_clients 接受 concurrency 参数适配熔断阈值** + +修改 `_build_clients` 签名为 `_build_clients(concurrency: int = 16)`。 + +在读取 `breaker_threshold` 后追加： + +```python + breaker_threshold = int(os.getenv("LLM_CIRCUIT_BREAKER_THRESHOLD", "5")) + breaker_threshold = max(breaker_threshold, concurrency * 2) +``` + +传入两个 `CircuitBreaker` 实例时使用适配后的 `breaker_threshold`。 + +- [ ] **Step 3: 重写 main_async 实现并发 + 断点续跑** + +将 `main_async` 的串行 for 循环替换为 Semaphore 并发编排： + +```python +async def main_async(args: argparse.Namespace) -> None: + """异步主流程：并发修复视频。""" + videos_dir = Path(args.videos_dir) + srt_dir = Path(args.srt_dir) + questions_dir = Path(args.questions_dir) + concurrency = args.concurrency + reaggregate_all = args.reaggregate_all + + # 扫描所有视频 + vid_dirs = sorted( + d for d in videos_dir.iterdir() + if d.is_dir() and (d / "tree.json").exists() + ) + logger.info("发现 {} 个视频", len(vid_dirs)) + + # 加载 progress + progress_path = PROJECT_ROOT / "logs" / PROGRESS_FILE + finished = load_progress(progress_path) + if finished: + logger.info("已完成 {} 个视频（从 progress 文件加载）", len(finished)) + + if args.dry_run: + logger.info("=== DRY RUN 模式：仅检测不修复 ===") + + # 构建客户端（dry_run 模式不需要） + llm, vlm = (None, None) if args.dry_run else _build_clients(concurrency) + + # 过滤跳过的视频 + pending = [] + skipped_count = 0 + for vid_dir in vid_dirs: + vid = vid_dir.name + if should_skip_video(vid, finished, reaggregate_all=reaggregate_all): + skipped_count += 1 + continue + pending.append(vid_dir) + + if skipped_count: + logger.info("跳过 {} 个已完成视频，待处理 {} 个", skipped_count, len(pending)) + + # 并发编排 + sem = asyncio.Semaphore(concurrency) + progress_lock = asyncio.Lock() + all_stats: list[dict] = [] + stats_lock = asyncio.Lock() + start_time = time.time() + completed = 0 + + async def _process(vid_dir: Path) -> None: + nonlocal completed + async with sem: + vid = vid_dir.name + tree_path = vid_dir / "tree.json" + frames_dir = vid_dir + + logger.info("开始修复 {}", vid) + + stats = await _repair_one_video( + vid, tree_path, frames_dir, srt_dir, questions_dir, + llm, vlm, dry_run=args.dry_run, + ) + + async with stats_lock: + all_stats.append(stats) + completed += 1 + + # 无 error 且非 dry_run 才记 finished + if stats["error"] is None and not args.dry_run: + await save_progress(progress_path, progress_lock, vid) + + # 进度日志 + if completed % 10 == 0: + elapsed = time.time() - start_time + rate = completed / elapsed * 60 + logger.info( + "进度: {}/{}, 已用 {:.0f}s, 速率 {:.1f} 视频/分钟", + completed, len(pending), elapsed, rate, + ) + + tasks = [asyncio.create_task(_process(vd)) for vd in pending] + await asyncio.gather(*tasks) + + # 最终汇总 + elapsed = time.time() - start_time + total_issues = sum(s["issues_found"] for s in all_stats) + total_repaired = sum(s["l3_repaired"] for s in all_stats) + total_injected = sum(s["facts_injected"] for s in all_stats) + total_errors = sum(1 for s in all_stats if s["error"]) + + logger.info("=" * 60) + logger.info("修复完成") + logger.info(" 视频总数: {}", len(all_stats)) + logger.info(" 跳过数: {}", skipped_count) + logger.info(" 问题总数: {}", total_issues) + logger.info(" L3 修复数: {}", total_repaired) + logger.info(" 事实注入数: {}", total_injected) + logger.info(" 失败数: {}", total_errors) + logger.info(" 总耗时: {:.0f}s", elapsed) + logger.info(" 并发数: {}", concurrency) + logger.info("=" * 60) + + if total_errors > 0: + logger.warning("以下视频修复失败:") + for s in all_stats: + if s["error"]: + logger.warning(" {}: {}", s["vid"], s["error"]) +``` + +- [ ] **Step 4: 更新 .env.example 熔断阈值注释** + +将 `.env.example` 中： + +``` +LLM_CIRCUIT_BREAKER_THRESHOLD=5 +``` + +改为： + +``` +LLM_CIRCUIT_BREAKER_THRESHOLD=5 # 实际阈值 = max(此值, concurrency*2) +``` + +- [ ] **Step 5: dry-run 模式冒烟测试** + +运行: `conda activate Video-Tree-TRM && python tools/repair_trees.py --dry-run --concurrency 4 2>&1 | head -30` +预期: 看到"发现 N 个视频"、"跳过 M 个已完成视频"、并发日志，无报错 + +- [ ] **Step 6: 提交** + +```bash +git add tools/repair_trees.py .env.example +git commit -m "feat(repair): asyncio.Semaphore 并发 + 断点续跑 + CLI 参数 + +--concurrency 默认 16，--reaggregate-all 强制全量重聚合。 +Semaphore 限视频并发数，视频内四步串行。progress 文件 +asyncio.Lock + os.replace 原子写入。熔断阈值 max(.env, concurrency*2)。" +``` + +--- + +### Task 6: Lint + 全量测试 + 最终验证 + +**Files:** +- 无新文件 + +- [ ] **Step 1: ruff 格式化与检查** + +运行: +```bash +conda activate Video-Tree-TRM && ruff format adapters/telemetry.py adapters/llm.py app/tree/repair/detector.py tools/repair_trees.py +conda activate Video-Tree-TRM && ruff check adapters/telemetry.py adapters/llm.py app/tree/repair/detector.py tools/repair_trees.py --fix +``` +预期: 无错误 + +- [ ] **Step 2: 全量测试** + +运行: `conda activate Video-Tree-TRM && pytest tests/unit/ -v --tb=short` +预期: 全部 PASS + +- [ ] **Step 3: 提交 lint 修正（如有）** + +```bash +git add -u +git commit -m "style: ruff format adapters + detector + repair_trees" +``` + +- [ ] **Step 4: 确认改动文件范围** + +运行: `git diff --stat main` +预期改动文件： + +| 文件 | 性质 | +|------|------| +| `adapters/telemetry.py` | 修改 | +| `adapters/llm.py` | 修改 | +| `app/tree/repair/detector.py` | 修改 | +| `tools/repair_trees.py` | 修改 | +| `.env.example` | 修改 | +| `tests/unit/test_telemetry.py` | 修改 | +| `tests/unit/test_governed_llm.py` | 修改 | +| `tests/unit/test_repair_detector.py` | 修改 | +| `tests/unit/test_repair_progress.py` | 新建 | diff --git a/research-wiki/plans/question-gen.md b/research-wiki/plans/question-gen.md new file mode 100644 index 0000000..1880f3f --- /dev/null +++ b/research-wiki/plans/question-gen.md @@ -0,0 +1,9 @@ +--- +type: plan +node_id: plan:question-gen +title: "question_gen 模块实现计划" +date: 2026-07-07 +--- + +# question_gen 模块实现计划 + diff --git a/research-wiki/plans/tree-repair-resilience.md b/research-wiki/plans/tree-repair-resilience.md new file mode 100644 index 0000000..40ceee3 --- /dev/null +++ b/research-wiki/plans/tree-repair-resilience.md @@ -0,0 +1,9 @@ +--- +type: plan +node_id: plan:tree-repair-resilience +title: 建树修复管线三项改造实现计划 +date: 2026-07-09 +--- + +# 建树修复管线三项改造实现计划 +