From e9073bfdc20d05607842dca7d02b5ab1b3958194 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Sat, 11 Jul 2026 11:14:45 -0400 Subject: [PATCH] feat(tree): expose build_async and accept injected API semaphore Core algorithms #1/#2/#3 unchanged: only entry wrapping and semaphore source switch (injected vs self-built); build logic untouched. - rename _build_async to public build_async (body unchanged) - __init__ accepts keyword-only api_semaphore for cross-video sharing - default path (no injection) is verbatim-equivalent to previous code --- app/tree/video_builder.py | 27 ++++++++++++++------ tests/unit/test_video_builder.py | 42 ++++++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 8 deletions(-) diff --git a/app/tree/video_builder.py b/app/tree/video_builder.py index ca3e51a..fc2166f 100644 --- a/app/tree/video_builder.py +++ b/app/tree/video_builder.py @@ -13,8 +13,8 @@ 并发模型(异步版):: - build() → asyncio.run(_build_async()) - _build_async(): + build() → asyncio.run(build_async()) + build_async(): asyncio.Semaphore(concurrency) 控制最大 VLM/LLM 并发数 各 L1 段并发构建,段内 L2 clip 各启动 _chain 协程: 提取全部 L3 帧 → 采样 L2 代表帧 → L2 VLM → L3 VLM @@ -156,8 +156,8 @@ class VideoTreeBuilder: 转化为三层 TreeIndex。 并发架构: - build() 为同步壳,内部调用 asyncio.run(_build_async())。 - _build_async() 使用 asyncio.Semaphore(concurrency) 控制并发 VLM/LLM 数量。 + build() 为同步壳,内部调用 asyncio.run(build_async())。 + build_async() 使用 asyncio.Semaphore(concurrency) 控制并发 VLM/LLM 数量。 所有 VLM 调用通过 VLMProvider 的异步接口发起,零线程阻塞。 所有 LLM 调用通过 LLMProvider 的异步接口发起(L1 摘要)。 ffmpeg 提帧在独立 ThreadPoolExecutor 中并行,不阻塞事件循环。 @@ -166,6 +166,7 @@ class VideoTreeBuilder: _vlm: VLM 图文调用端口。 _llm: LLM 文本调用端口(L1 摘要)。 _config: 树构建配置。 + _api_semaphore: 外部注入的全局 API 并发信号量(None 时 build_async 自建)。 _ffmpeg_pool: ffmpeg 专用线程池(max_workers=_FFMPEG_MAX_WORKERS)。 """ @@ -174,6 +175,8 @@ class VideoTreeBuilder: vlm: VLMProvider, llm: LLMProvider, config: TreeConfig, + *, + api_semaphore: asyncio.Semaphore | None = None, ) -> None: """初始化视频树构建器。 @@ -183,10 +186,14 @@ class VideoTreeBuilder: config: 树构建配置(TreeConfig),关键字段: l1_segment_duration, l2_clip_duration, l3_fps, l2_representative_frames, cache_dir, concurrency。 + api_semaphore: 外部注入的全局 VLM/LLM 并发信号量(批量建树时跨视频共享); + None 时 build_async 内部按 config.concurrency 自建, + 单视频行为零变化。 """ self._vlm = vlm self._llm = llm self._config = config + self._api_semaphore = api_semaphore self._ffmpeg_pool = ThreadPoolExecutor(max_workers=_FFMPEG_MAX_WORKERS) self._cache_root = Path(self._config.cache_dir) self._session_id: str = "" @@ -296,13 +303,13 @@ class VideoTreeBuilder: 返回: 三层 TreeIndex 对象。 """ - return asyncio.run(self._build_async(video_path, srt_entries)) + return asyncio.run(self.build_async(video_path, srt_entries)) # ------------------------------------------------------------------ # 核心异步构建逻辑(保真算法 #1:L2→L3 链式触发) # ------------------------------------------------------------------ - async def _build_async( + async def build_async( self, video_path: str, srt_entries: list[SRTEntry] | None = None, @@ -353,8 +360,12 @@ class VideoTreeBuilder: # Phase 1.1: 读取已有进度(保真算法 #3:断点续跑) finished_l1_ids = self._load_resume_state(source_id, total_l1) - # 创建 VLM/LLM 并发控制信号量 - vlm_sem = asyncio.Semaphore(self._config.concurrency) + # 创建 VLM/LLM 并发控制信号量(外部注入时跨视频全局共享,Spec-2) + vlm_sem = ( + self._api_semaphore + if self._api_semaphore is not None + else asyncio.Semaphore(self._config.concurrency) + ) # Phase 2-5: 按 L1 段并发,段内 L2→L3 链式触发(保真算法 #1) async def _build_segment( diff --git a/tests/unit/test_video_builder.py b/tests/unit/test_video_builder.py index 5a9addf..91f2e83 100644 --- a/tests/unit/test_video_builder.py +++ b/tests/unit/test_video_builder.py @@ -14,6 +14,7 @@ from __future__ import annotations +import asyncio import json from pathlib import Path from typing import Any @@ -997,3 +998,44 @@ class TestHelpers: long_name = "a" * 100 + ".mp4" stem = VideoTreeBuilder._source_stem(f"/path/{long_name}") assert len(stem) == 64 + + +# ── 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)