@@ -0,0 +1,880 @@
# 建树批量并行入口（Spec-2）
> **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 并发 + 全局共享 API 信号量 16），单视频建树行为零变化。
**Architecture: ** `VideoTreeBuilder` 公开 `build_async()` （消除事件循环嵌套）并支持外部注入 API Semaphore； `tools/build_trees.py` 复刻 `repair_trees.py` 的编排惯例（视频级 Semaphore + gather + progress.json 断点续跑 + 熔断阈值缩放）；`scripts/build_trees.sh` 自包含实验入口。设计文档：`research-wiki/designs/2026-07-11-batch-tree-build-design.md` 。
**Tech Stack: ** Python 3.11 asyncio / pytest + pytest-asyncio / loguru / PyYAML。
**验证策略说明 ** ：仓库 data/ 下当前无视频文件（`find data -name "*.mp4"` 为空），设计 §6 的"3-4 短视频小批量集成"改为**桩编排集成测试**（
---
### Task 1: VideoTreeBuilder 异步入口公开 + API Semaphore 注入
**Files: **
- Modify: `app/tree/video_builder.py` （ `__init__` 172-192 行；`build()` 284-299 行；`_build_async` 改名 `build_async` 305 行起；`vlm_sem = asyncio.Semaphore(...)` 357 行）
- Test: `tests/unit/test_video_builder.py` （追加测试类）
- [ ] **Step 1: 写失败测试 **
`tests/unit/test_video_builder.py` 末尾追加（fixture `mock_vlm` /`mock_llm` /`tree_config` 已存在于该文件 159-191 行）：
``` python
# ── Semaphore 注入与异步入口（Spec-2）─────────────────────────
class TestApiSemaphoreInjection :
""" API Semaphore 注入与 build_async 公开入口。 """
def test_injected_semaphore_stored (
self ,
mock_vlm : MockVLMProvider ,
mock_llm : MockLLMProvider ,
tree_config : TreeConfig ,
) - > None :
""" 构造器注入的 Semaphore 应被保存供 build_async 使用。 """
sem = asyncio . Semaphore ( 3 )
builder = VideoTreeBuilder (
vlm = mock_vlm , llm = mock_llm , config = tree_config , api_semaphore = sem
)
assert builder . _api_semaphore is sem
def test_default_no_injection (
self ,
mock_vlm : MockVLMProvider ,
mock_llm : MockLLMProvider ,
tree_config : TreeConfig ,
) - > None :
""" 未注入时属性为 None（build_async 内部自建，单视频行为零变化）。 """
builder = VideoTreeBuilder ( vlm = mock_vlm , llm = mock_llm , config = tree_config )
assert builder . _api_semaphore is None
def test_build_async_is_public (
self ,
mock_vlm : MockVLMProvider ,
mock_llm : MockLLMProvider ,
tree_config : TreeConfig ,
) - > None :
""" build_async 必须是公开协程方法（供批量编排在事件循环内调用）。 """
builder = VideoTreeBuilder ( vlm = mock_vlm , llm = mock_llm , config = tree_config )
assert hasattr ( builder , " build_async " )
assert asyncio . iscoroutinefunction ( builder . build_async )
```
（文件顶部如无 `import asyncio` 则补上。）
- [ ] **Step 2: 运行测试确认失败 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_video_builder.py::TestApiSemaphoreInjection -v`
Expected: FAIL——`api_semaphore` 是未知构造参数（TypeError），`build_async` 不存在。
- [ ] **Step 3: 实现 **
三处修改：
其一，`__init__` 增加 keyword-only 可选参数（docstring 同步补充）：
``` python
def __init__ (
self ,
vlm : VLMProvider ,
llm : LLMProvider ,
config : TreeConfig ,
* ,
api_semaphore : asyncio . Semaphore | None = None ,
) - > None :
```
赋值区追加：
``` python
self . _api_semaphore = api_semaphore
```
docstring 参数段追加一行：`api_semaphore: 外部注入的全局 VLM/LLM 并发信号量（批量建树时跨视频共享）；None 时 build_async 内部按 config.concurrency 自建，单视频行为零变化。`
其二，`_build_async` 改名为公开 `build_async` （方法体不动，docstring 首行相应更新），`build()` 同步壳改为：
``` python
return asyncio . run ( self . build_async ( video_path , srt_entries ) )
```
同时更新模块 docstring（ （ `_build_async` 字样为 `build_async` 。
其三，`build_async` 内 357 行的信号量创建改为：
``` python
# 创建 VLM/LLM 并发控制信号量（外部注入时跨视频全局共享，Spec-2）
vlm_sem = (
self . _api_semaphore
if self . _api_semaphore is not None
else asyncio . Semaphore ( self . _config . concurrency )
)
```
- [ ] **Step 4: 运行测试确认通过 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_video_builder.py -v`
Expected: 全部 PASS（既有测试 + 新增 3 个，确认改名无回归——若有既有测试直接引用 `_build_async` ，同步改为 `build_async` ）。
- [ ] **Step 5: Commit **
``` bash
git add app/tree/video_builder.py tests/unit/test_video_builder.py
git commit -m "feat(tree): expose build_async and accept injected API semaphore"
```
（ #1/ #2/ #3 不变——仅入口封装与信号量来源切换，构建逻辑零改动。）
---
### Task 2: tools/build_trees.py 辅助纯函数
**Files: **
- Create: `tools/build_trees.py` （本任务只写模块骨架 + 纯函数；编排在 Task 3）
- Test: `tests/unit/test_build_trees.py` （新建）
- [ ] **Step 1: 写失败测试 **
新建 `tests/unit/test_build_trees.py` ：
``` python
""" tools/build_trees.py 单元测试。
覆盖纯函数（完整性校验、待建清单发现、SRT 查找）与编排集成（Task 3 追加）。
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
import pytest
PROJECT_ROOT = Path ( __file__ ) . resolve ( ) . parent . parent . parent
if str ( PROJECT_ROOT ) not in sys . path :
sys . path . insert ( 0 , str ( PROJECT_ROOT ) )
from app . tree . index import IndexMeta , L1Card , L1Node , TreeIndex
from tools . build_trees import _discover_pending , _find_srt_entries , _tree_is_complete
def _write_valid_tree ( tree_path : Path ) - > None :
""" 写入一棵最小合法树。 """
l1 = L1Node (
id = " vid_L1_000 " ,
card = L1Card ( " 场景 " , " 室内 " , [ " 实体 " ] , [ " 动作 " ] , [ " 关键词 " ] , [ ] , " 线性 " ) ,
time_range = ( 0.0 , 10.0 ) ,
children = [ ] ,
)
index = TreeIndex ( metadata = IndexMeta ( " /v.mp4 " , " video " ) , roots = [ l1 ] )
tree_path . parent . mkdir ( parents = True , exist_ok = True )
index . save_json ( str ( tree_path ) )
class TestTreeIsComplete :
""" _tree_is_complete 测试。 """
def test_valid_tree ( self , tmp_path : Path ) - > None :
""" 合法 tree.json 判定完整。 """
tree_path = tmp_path / " vid " / " tree.json "
_write_valid_tree ( tree_path )
assert _tree_is_complete ( tree_path ) is True
def test_missing_file ( self , tmp_path : Path ) - > None :
""" 文件不存在判定不完整。 """
assert _tree_is_complete ( tmp_path / " nope " / " tree.json " ) is False
def test_corrupt_json ( self , tmp_path : Path ) - > None :
""" 损坏 JSON 判定不完整（不抛异常）。 """
p = tmp_path / " vid " / " tree.json "
p . parent . mkdir ( parents = True )
p . write_text ( " { broken " , encoding = " utf-8 " )
assert _tree_is_complete ( p ) is False
class TestDiscoverPending :
""" _discover_pending 测试。 """
def _touch_videos ( self , videos_dir : Path , names : list [ str ] ) - > None :
videos_dir . mkdir ( parents = True , exist_ok = True )
for n in names :
( videos_dir / n ) . write_bytes ( b " " )
def test_all_pending_when_fresh ( self , tmp_path : Path ) - > None :
""" 无进度无产物时全部待建，按名排序。 """
videos = tmp_path / " videos "
self . _touch_videos ( videos , [ " b.mp4 " , " a.mkv " , " c.txt " ] )
pending = _discover_pending ( videos , tmp_path / " out " , set ( ) )
assert [ p . name for p in pending ] == [ " a.mkv " , " b.mp4 " ] # 非视频扩展名被忽略
def test_skips_finished_and_complete ( self , tmp_path : Path ) - > None :
""" progress 已记录或 tree.json 完整的视频被跳过。 """
videos = tmp_path / " videos "
out = tmp_path / " out "
self . _touch_videos ( videos , [ " a.mp4 " , " b.mp4 " , " c.mp4 " ] )
_write_valid_tree ( out / " b " / " tree.json " ) # b 已有完整树
pending = _discover_pending ( videos , out , { " a " } ) # a 在 progress 中
assert [ p . name for p in pending ] == [ " c.mp4 " ]
def test_incomplete_tree_not_skipped ( self , tmp_path : Path ) - > None :
""" tree.json 损坏的视频仍待建（重建覆盖）。 """
videos = tmp_path / " videos "
out = tmp_path / " out "
self . _touch_videos ( videos , [ " a.mp4 " ] )
( out / " a " ) . mkdir ( parents = True )
( out / " a " / " tree.json " ) . write_text ( " { broken " , encoding = " utf-8 " )
pending = _discover_pending ( videos , out , set ( ) )
assert [ p . name for p in pending ] == [ " a.mp4 " ]
class TestFindSrtEntries :
""" _find_srt_entries 测试。 """
def test_found ( self , tmp_path : Path ) - > None :
""" 同名 .srt 存在时解析返回条目。 """
srt = tmp_path / " vid.srt "
srt . write_text (
" 1 \n 00:00:01,000 --> 00:00:03,000 \n hello world \n \n " ,
encoding = " utf-8 " ,
)
entries = _find_srt_entries ( tmp_path / " vid.mp4 " , tmp_path )
assert entries is not None
assert len ( entries ) == 1
def test_missing_returns_none ( self , tmp_path : Path ) - > None :
""" 无同名 .srt 返回 None。 """
assert _find_srt_entries ( tmp_path / " vid.mp4 " , tmp_path ) is None
```
- [ ] **Step 2: 运行确认失败 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_build_trees.py -v`
Expected: FAIL（
- [ ] **Step 3: 实现模块骨架 + 纯函数 **
新建 `tools/build_trees.py` ：
``` python
#!/usr/bin/env python3
""" 批量并行建树入口：多视频并发构建三层 TreeIndex。
并发模型（Spec-2）
视频级 Semaphore(video_concurrency) + gather —— 复刻 repair_trees.py 惯例；
全局共享一个 API Semaphore(api_concurrency) 注入所有 VideoTreeBuilder，
端点压力与单视频建树完全一致，吞吐提升来自非 API 阶段跨视频重叠。
用法:
conda activate Video-Tree-TRM
python tools/build_trees.py --videos-dir <dir> [--out-dir store/videos]
[--srt-dir <dir>] [--video-concurrency 16] [--limit 0]
api_concurrency 为工程配置，从 .env 读取 TREE_BUILD_API_CONCURRENCY（默认 16）。
app/core/adapters 不 import 此脚本。
"""
from __future__ import annotations
import argparse
import asyncio
import json
import os
import sys
import time
from pathlib import Path
PROJECT_ROOT = Path ( __file__ ) . resolve ( ) . parent . parent
sys . path . insert ( 0 , str ( PROJECT_ROOT ) )
from dotenv import load_dotenv
from loguru import logger
load_dotenv ( PROJECT_ROOT / " .env " )
import yaml
from app . tree . config import TreeConfig
from app . tree . index import TreeIndex
from app . tree . subtitle import parse_srt
from app . tree . video_builder import VideoTreeBuilder
# ---------------------------------------------------------------------------
# 日志配置：不缓存，立即输出
# ---------------------------------------------------------------------------
logger . remove ( )
logger . add (
sys . stderr ,
format = " { time:HH:mm:ss} | {level:<7} | {message} " ,
level = " DEBUG " ,
colorize = True ,
)
logger . add (
PROJECT_ROOT / " logs " / " build_trees.log " ,
format = " { time:YYYY-MM-DD HH:mm:ss} | {level:<7} | {message} " ,
level = " DEBUG " ,
rotation = " 50 MB " ,
)
# ---------------------------------------------------------------------------
# 断点续跑 — progress 文件管理（复刻 repair_trees.py 惯例）
# ---------------------------------------------------------------------------
PROGRESS_FILE = " build_progress.json "
_VIDEO_SUFFIXES = frozenset ( { " .mp4 " , " .mkv " , " .avi " , " .webm " } )
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 , AttributeError ) :
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 _tree_is_complete ( tree_path : Path ) - > bool :
""" 判断 tree.json 是否存在且可加载为非空树。
参数:
tree_path: tree.json 路径。
返回:
True 表示完整（跳过重建）；文件缺失/损坏/空树返回 False。
"""
if not tree_path . exists ( ) :
return False
try :
index = TreeIndex . load_json ( str ( tree_path ) )
except ( json . JSONDecodeError , KeyError , TypeError , ValueError , AssertionError ) as exc :
logger . warning ( " tree.json 无法加载，视为不完整: {} ( {} ) " , tree_path , exc )
return False
return len ( index . roots ) > 0
def _discover_pending (
videos_dir : Path ,
out_dir : Path ,
finished : set [ str ] ,
) - > list [ Path ] :
""" 扫描视频目录，返回待建视频文件列表（按文件名排序）。
跳过条件：video_id 在 progress 中，或 out_dir/<video_id>/tree.json 完整。
参数:
videos_dir: 视频文件目录。
out_dir: 树输出根目录。
finished: progress 中已完成的视频 ID 集合。
返回:
待建视频文件路径列表。
"""
pending : list [ Path ] = [ ]
for f in sorted ( videos_dir . iterdir ( ) ) :
if not f . is_file ( ) or f . suffix . lower ( ) not in _VIDEO_SUFFIXES :
continue
vid = f . stem
if vid in finished :
continue
if _tree_is_complete ( out_dir / vid / " tree.json " ) :
continue
pending . append ( f )
return pending
def _find_srt_entries ( video_path : Path , srt_dir : Path ) :
""" 按视频同名规则查找并解析 SRT 字幕。
参数:
video_path: 视频文件路径。
srt_dir: SRT 目录。
返回:
SRTEntry 列表；无同名 .srt 时返回 None。
"""
srt_path = srt_dir / f " { video_path . stem } .srt "
if not srt_path . exists ( ) :
return None
return parse_srt ( str ( srt_path ) )
```
（编排函数与 main 在 Task 3 追加；本步模块以纯函数收尾即可运行测试。）
- [ ] **Step 4: 运行确认通过 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_build_trees.py -v`
Expected: 全部 PASS。
- [ ] **Step 5: Commit **
``` bash
git add tools/build_trees.py tests/unit/test_build_trees.py
git commit -m "feat(tools): add build_trees skeleton with discovery helpers"
```
---
### Task 3: 编排 main_async + CLI + 桩编排集成测试
**Files: **
- Modify: `tools/build_trees.py` （追加客户端构建、编排、CLI）
- Test: `tests/unit/test_build_trees.py` （追加编排测试）
- [ ] **Step 1: 写失败测试（桩 builder 编排集成） **
`tests/unit/test_build_trees.py` 追加：
``` python
# ── 编排集成（桩 builder，无真实视频/LLM）──────────────────────
import asyncio
from tools . build_trees import main_async
class _StubBuilder :
""" 记录并发与注入信号量的桩 builder。 """
instances : list [ " _StubBuilder " ] = [ ]
inflight = 0
max_inflight = 0
def __init__ ( self , vlm , llm , config , * , api_semaphore = None ) - > None :
self . api_semaphore = api_semaphore
_StubBuilder . instances . append ( self )
async def build_async ( self , video_path : str , srt_entries = None ) :
_StubBuilder . inflight + = 1
_StubBuilder . max_inflight = max ( _StubBuilder . max_inflight , _StubBuilder . inflight )
await asyncio . sleep ( 0.02 )
_StubBuilder . inflight - = 1
from app . tree . index import IndexMeta , L1Card , L1Node , TreeIndex
l1 = L1Node (
id = " x_L1_000 " ,
card = L1Card ( " s " , " 室内 " , [ ] , [ ] , [ ] , [ ] , " 线性 " ) ,
time_range = ( 0.0 , 1.0 ) ,
children = [ ] ,
)
return TreeIndex ( metadata = IndexMeta ( video_path , " video " ) , roots = [ l1 ] )
@pytest.fixture ( )
def batch_env ( tmp_path : Path , monkeypatch : pytest . MonkeyPatch ) :
""" 5 个假视频 + 桩 builder + 隔离的 progress 路径。 """
import tools . build_trees as bt
_StubBuilder . instances = [ ]
_StubBuilder . inflight = 0
_StubBuilder . max_inflight = 0
monkeypatch . setattr ( bt , " VideoTreeBuilder " , _StubBuilder )
monkeypatch . setattr ( bt , " _build_clients " , lambda api_concurrency : ( None , None ) )
videos = tmp_path / " videos "
videos . mkdir ( )
for i in range ( 5 ) :
( videos / f " v { i } .mp4 " ) . write_bytes ( b " " )
return {
" videos " : videos ,
" out " : tmp_path / " out " ,
" progress " : tmp_path / " build_progress.json " ,
}
def _make_args ( env : dict , video_concurrency : int = 2 , limit : int = 0 ) - > " argparse.Namespace " :
import argparse
return argparse . Namespace (
videos_dir = str ( env [ " videos " ] ) ,
out_dir = str ( env [ " out " ] ) ,
srt_dir = str ( env [ " videos " ] ) ,
video_concurrency = video_concurrency ,
limit = limit ,
progress_path = str ( env [ " progress " ] ) ,
)
class TestOrchestration :
""" main_async 编排行为（桩 builder）。 """
@pytest.mark.asyncio
async def test_video_concurrency_capped ( self , batch_env : dict ) - > None :
""" 同时在建视频数不得超过 video_concurrency。 """
await main_async ( _make_args ( batch_env , video_concurrency = 2 ) )
assert _StubBuilder . max_inflight < = 2
assert len ( _StubBuilder . instances ) == 5
@pytest.mark.asyncio
async def test_shared_api_semaphore ( self , batch_env : dict ) - > None :
""" 全部 builder 实例共享同一个 API Semaphore 对象。 """
await main_async ( _make_args ( batch_env ) )
sems = { id ( b . api_semaphore ) for b in _StubBuilder . instances }
assert len ( sems ) == 1
assert _StubBuilder . instances [ 0 ] . api_semaphore is not None
@pytest.mark.asyncio
async def test_trees_saved_and_progress_recorded ( self , batch_env : dict ) - > None :
""" 每个视频产出 tree.json 且 progress 记录全部完成。 """
await main_async ( _make_args ( batch_env ) )
for i in range ( 5 ) :
assert ( batch_env [ " out " ] / f " v { i } " / " tree.json " ) . exists ( )
from tools . build_trees import load_progress
assert load_progress ( batch_env [ " progress " ] ) == { f " v { i } " for i in range ( 5 ) }
@pytest.mark.asyncio
async def test_resume_skips_finished ( self , batch_env : dict ) - > None :
""" 第二次运行跳过全部已完成视频。 """
await main_async ( _make_args ( batch_env ) )
n_first = len ( _StubBuilder . instances )
await main_async ( _make_args ( batch_env ) )
assert len ( _StubBuilder . instances ) == n_first # 无新建
@pytest.mark.asyncio
async def test_partial_completion_resume ( self , batch_env : dict ) - > None :
""" 部分完成后重跑只建剩余视频（模拟中断后恢复）。 """
batch_env [ " progress " ] . write_text (
json . dumps ( { " finished_video_ids " : [ " v0 " , " v1 " ] } ) ,
encoding = " utf-8 " ,
)
await main_async ( _make_args ( batch_env ) )
assert len ( _StubBuilder . instances ) == 3 # 仅 v2/v3/v4
@pytest.mark.asyncio
async def test_limit ( self , batch_env : dict ) - > None :
""" --limit 2 只建前两个（烟测入口）。 """
await main_async ( _make_args ( batch_env , limit = 2 ) )
assert len ( _StubBuilder . instances ) == 2
```
- [ ] **Step 2: 运行确认失败 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_build_trees.py::TestOrchestration -v`
Expected: FAIL（
- [ ] **Step 3: 实现客户端构建 + 编排 + CLI **
`tools/build_trees.py` 追加：
``` python
# ---------------------------------------------------------------------------
# LLM/VLM 客户端构建（复刻 repair_trees.py 惯例）
# ---------------------------------------------------------------------------
def _build_clients ( api_concurrency : int ) :
""" 构建 GovernedLLMClient（
参数:
api_concurrency: 全局 API 并发上限（熔断阈值取 max(cfg, api_concurrency*2)）。
返回:
(llm_client, vlm_client) 元组。
"""
from adapters . breaker import CircuitBreaker
from adapters . llm import GovernedLLMClient
from adapters . telemetry import SQLiteTelemetryRecorder
from adapters . vlm import GovernedVLMClient
( PROJECT_ROOT / " logs " ) . mkdir ( exist_ok = True )
telemetry = SQLiteTelemetryRecorder ( str ( PROJECT_ROOT / " logs " / " build_trees_telemetry.db " ) )
breaker_threshold = int ( os . getenv ( " LLM_CIRCUIT_BREAKER_THRESHOLD " , " 5 " ) )
breaker_threshold = max ( breaker_threshold , api_concurrency * 2 )
breaker_cooldown = int ( os . getenv ( " LLM_CIRCUIT_BREAKER_COOLDOWN " , " 60 " ) )
timeout_s = float ( os . getenv ( " LLM_TIMEOUT " , " 120 " ) )
max_retries = int ( os . getenv ( " LLM_MAX_RETRIES " , " 3 " ) )
base_delay = float ( os . getenv ( " LLM_RETRY_BASE_DELAY " , " 2.0 " ) )
max_delay = float ( os . getenv ( " LLM_RETRY_MAX_DELAY " , " 30.0 " ) )
ttft = float ( os . getenv ( " LLM_TTFT_TIMEOUT " , " 30 " ) )
inter_token = float ( os . getenv ( " LLM_INTER_TOKEN_TIMEOUT " , " 15 " ) )
llm = GovernedLLMClient (
model = os . environ [ " SEARCH_LLM_MODEL " ] ,
base_url = os . environ [ " SEARCH_LLM_BASE_URL " ] ,
api_key = os . environ [ " SEARCH_LLM_API_KEY " ] ,
provider = " deepseek " ,
thinking = False ,
breaker = CircuitBreaker ( fail_threshold = breaker_threshold , cooldown_s = breaker_cooldown ) ,
cache = None ,
telemetry = telemetry ,
timeout_s = timeout_s ,
ttft_timeout_s = ttft ,
inter_token_timeout_s = inter_token ,
max_retries = max_retries ,
retry_base_delay_s = base_delay ,
retry_max_delay_s = max_delay ,
)
vlm_base = GovernedLLMClient (
model = os . environ [ " VL_LLM_MODEL " ] ,
base_url = os . environ [ " VL_LLM_BASE_URL " ] ,
api_key = os . environ [ " VL_LLM_API_KEY " ] ,
provider = " qwen " ,
thinking = False ,
breaker = CircuitBreaker ( fail_threshold = breaker_threshold , cooldown_s = breaker_cooldown ) ,
cache = None ,
telemetry = telemetry ,
timeout_s = timeout_s ,
ttft_timeout_s = ttft ,
inter_token_timeout_s = inter_token ,
max_retries = max_retries ,
retry_base_delay_s = base_delay ,
retry_max_delay_s = max_delay ,
)
return llm , GovernedVLMClient ( vlm_base )
# ---------------------------------------------------------------------------
# 主编排
# ---------------------------------------------------------------------------
async def main_async ( args : argparse . Namespace ) - > None :
""" 异步主流程：视频级并发建树 + 全局共享 API 信号量。
参数:
args: CLI 参数（videos_dir/out_dir/srt_dir/video_concurrency/limit/progress_path）。
"""
videos_dir = Path ( args . videos_dir )
out_dir = Path ( args . out_dir )
srt_dir = Path ( args . srt_dir )
assert videos_dir . is_dir ( ) , f " 视频目录不存在: { videos_dir } "
api_concurrency = int ( os . getenv ( " TREE_BUILD_API_CONCURRENCY " , " 16 " ) )
# Phase 1: 待建发现（progress + 完整性双重跳过）
progress_path = Path ( args . progress_path )
finished = load_progress ( progress_path )
if finished :
logger . info ( " progress 已记录 {} 个完成视频 " , len ( finished ) )
pending = _discover_pending ( videos_dir , out_dir , finished )
if args . limit > 0 :
pending = pending [ : args . limit ]
logger . info (
" 待建 {} 个视频, video_concurrency= {} , api_concurrency= {} " ,
len ( pending ) , args . video_concurrency , api_concurrency ,
)
if not pending :
return
# Phase 2: 客户端与共享信号量
llm , vlm = _build_clients ( api_concurrency )
with open ( PROJECT_ROOT / " config " / " default.yaml " , encoding = " utf-8 " ) as f :
tree_cfg = TreeConfig . from_dict ( yaml . safe_load ( f ) [ " tree " ] )
api_sem = asyncio . Semaphore ( api_concurrency )
video_sem = asyncio . Semaphore ( args . video_concurrency )
progress_lock = asyncio . Lock ( )
start_time = time . time ( )
completed = 0
failed : list [ str ] = [ ]
# Phase 3: 视频级并发编排（复刻 repair_trees 模式）
async def _build_one ( video_path : Path ) - > None :
nonlocal completed
async with video_sem :
vid = video_path . stem
logger . info ( " 开始建树 {} " , vid )
builder = VideoTreeBuilder ( vlm = vlm , llm = llm , config = tree_cfg , api_semaphore = api_sem )
srt_entries = _find_srt_entries ( video_path , srt_dir )
try :
index = await builder . build_async ( str ( video_path ) , srt_entries )
except Exception as exc :
logger . error ( " 建树失败 {} ( {} ): {} " , vid , type ( exc ) . __name__ , exc )
failed . append ( vid )
return
tree_path = out_dir / vid / " tree.json "
tree_path . parent . mkdir ( parents = True , exist_ok = True )
index . save_json ( str ( tree_path ) )
await save_progress ( progress_path , progress_lock , vid )
completed + = 1
if completed % 5 == 0 :
elapsed = time . time ( ) - start_time
rate = completed / elapsed * 60 if elapsed > 0 else 0
logger . info (
" 进度: {} / {} , 已用 {:.0f} s, 速率 {:.2f} 视频/分钟 " ,
completed , len ( pending ) , elapsed , rate ,
)
await asyncio . gather ( * [ asyncio . create_task ( _build_one ( p ) ) for p in pending ] )
# Phase 4: 汇总
elapsed = time . time ( ) - start_time
logger . info (
" 批量建树完成: 成功 {} , 失败 {} , 总耗时 {:.0f} s {} " ,
completed , len ( failed ) , elapsed ,
f " , 失败清单: { failed } " if failed else " " ,
)
def main ( ) - > None :
""" 同步入口。 """
parser = argparse . ArgumentParser ( description = " 批量并行建树 " )
parser . add_argument ( " --videos-dir " , type = str , required = True , help = " 视频文件目录 " )
parser . add_argument ( " --out-dir " , type = str , default = " store/videos " , help = " 树输出根目录 " )
parser . add_argument ( " --srt-dir " , type = str , default = " " , help = " SRT 目录（默认同 videos-dir） " )
parser . add_argument ( " --video-concurrency " , type = int , default = 16 , help = " 同时在建视频数 " )
parser . add_argument ( " --limit " , type = int , default = 0 , help = " 只建前 N 个（0=全部，烟测用） " )
parser . add_argument (
" --progress-path " ,
type = str ,
default = str ( PROJECT_ROOT / " logs " / PROGRESS_FILE ) ,
dest = " progress_path " ,
help = " progress 文件路径 " ,
)
args = parser . parse_args ( )
if not args . srt_dir :
args . srt_dir = args . videos_dir
asyncio . run ( main_async ( args ) )
if __name__ == " __main__ " :
main ( )
```
注意：`_build_one` 中 `except Exception` 用于**单视频错误隔离**（一个视频失败不拖垮整批，失败清单汇总上报，vid 不进 progress 可重跑）——这是编排层的错误隔离语义而非吞错，日志含异常类型与内容。
- [ ] **Step 4: 运行确认通过 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_build_trees.py -v`
Expected: 全部 PASS（纯函数 8 个 + 编排 6 个，共 14 个）。
- [ ] **Step 5: Commit **
``` bash
git add tools/build_trees.py tests/unit/test_build_trees.py
git commit -m "feat(tools): batch tree build orchestration with shared API semaphore"
```
---
### Task 4: scripts/build_trees.sh + 工程配置 + 回归收尾
**Files: **
- Create: `scripts/build_trees.sh`
- Modify: `.env.example` （追加 TREE_BUILD_API_CONCURRENCY）
- [ ] **Step 1: 写 sh 脚本 **
新建 `scripts/build_trees.sh` ：
``` bash
#!/usr/bin/env bash
# 批量并行建树（Spec-2）
# 职责：对目录下所有视频并发构建三层 TreeIndex，断点续跑。
#
# 用法:
# bash scripts/build_trees.sh --videos-dir <dir> # 全量
# bash scripts/build_trees.sh --videos-dir <dir> --limit 2 # 真实视频烟测
# VIDEO_CONCURRENCY=8 bash scripts/build_trees.sh --videos-dir <dir>
#
# 真实视频烟测流程（首次使用时执行）:
# 1. --limit 2 建两个视频，观察日志速率与 API 并发
# 2. 中途 Ctrl+C 后重跑，确认已完成视频被跳过、未完成视频续跑
# 3. 检查 store/videos/<vid>/tree.json 可被 TreeIndex.load_json 加载
#
# 日志输出:
# stderr → 终端实时显示
# logs/build_trees.log → 完整日志（自动 rotation 50MB）
# logs/build_trees_telemetry.db → LLM/VLM 调用遥测
# logs/build_progress.json → 断点续跑进度
#
# api_concurrency 走 .env 的 TREE_BUILD_API_CONCURRENCY（工程配置，默认 16）
set -euo pipefail
cd " $( dirname " $0 " ) /.. "
VIDEO_CONCURRENCY = " ${ VIDEO_CONCURRENCY :- 16 } "
export PYTHONUNBUFFERED = 1
# shellcheck source=/dev/null
source " $( conda info --base) /etc/profile.d/conda.sh "
conda activate Video-Tree-TRM
python tools/build_trees.py \
--out-dir store/videos \
--video-concurrency " $VIDEO_CONCURRENCY " \
" $@ "
```
``` bash
chmod +x scripts/build_trees.sh
```
- [ ] **Step 2: 补 .env.example **
`.env.example` 追加（若文件不存在则跳过并在 NOTES 说明）：
```
# 建树批量并行：全局 VLM/LLM 在途调用上限（Spec-2 工程配置）
TREE_BUILD_API_CONCURRENCY=16
```
- [ ] **Step 3: 全量回归 + lint **
Run: `conda run -n Video-Tree-TRM ruff format --check tools/build_trees.py tests/unit/test_build_trees.py app/tree/video_builder.py`
Run: `conda run -n Video-Tree-TRM ruff check tools/build_trees.py tests/unit/test_build_trees.py app/tree/video_builder.py`
Run: `conda run -n Video-Tree-TRM radon cc tools/build_trees.py app/tree/video_builder.py -n C -s` （新增函数不得出现 C 级；video_builder 既有 C 级为存量）
Run: `conda run -n Video-Tree-TRM pytest tests/ -q`
Expected: 全绿。
- [ ] **Step 4: Commit **
``` bash
git add scripts/build_trees.sh .env.example
git commit -m "feat(scripts): add batch tree build entry with smoke-test guide"
```
---
## Self-Review 记录
1. **Spec 覆盖 ** ：设计 §3 全部落地——build_async 公开 + Semaphore 注入（Task 1）、两层并发参数与全局共享信号量（Task 3）、熔断阈值缩放（Task 3 `_build_clients` ）、progress 断点续跑与完整性跳过（Task 2/3）、sh 入口与 D7 配置归属（Task 4）；§4 速率日志（Task 3 Phase 3）；§6 验证改为桩编排测试（头部"验证策略说明"已声明理由：仓库无视频文件），自动化覆盖"已完成跳过"（`test_resume_skips_finished` ）与"部分完成后恢复"（`test_partial_completion_resume` ）；**段级断点续跑**由 builder 既有 `_load_resume_state` 机制承担（算法 #3 ，本计划零改动）+ 真实视频烟测流程（sh 头注释步骤 2）在首次实际建树时验证。
2. **占位符扫描 ** ：无 TBD/TODO；所有代码步骤含完整代码。
3. **类型一致性 ** ： `_discover_pending(videos_dir, out_dir, finished)` 、`_find_srt_entries(video_path, srt_dir)` 、`main_async(args)` 、`_build_clients(api_concurrency)` 在测试与实现间签名一致；`_make_args` 构造的 Namespace 字段与 `main_async` 读取字段一一对应（含 progress_path）。
## 核心算法保真校验
本计划涉及**算法 #1 （ #2 （ #3 （断点续跑）**所在文件 `app/tree/video_builder.py` ：改动仅为 (a) `_build_async` 改名公开（方法体零改动）、(b) 信号量来源三元切换（默认路径与现状逐字等价）、(c) 构造器加可选参数。三项算法的核心逻辑（L2→L3 链式触发、批量 VLM+fallback、progress 段级恢复）均未触碰。执行时质量门第 7 项应 diff 确认 `build_async` 方法体与原 `_build_async` 逐行一致。其余算法不涉及。
