iomgaa
dbc9d38cd7
6 个 Task: telemetry 防御加固 → call_id 根因修复 → detector L2/L1 扩展 → progress 管理 → 并发编排+CLI → lint+全量测试
787 lines
26 KiB
Markdown
787 lines
26 KiB
Markdown
# 建树修复管线三项改造 实现计划
|
||
|
||
> **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` | 新建 |
|