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
This commit is contained in:
2026-07-11 11:14:45 -04:00
parent 25f5537974
commit e9073bfdc2
2 changed files with 61 additions and 8 deletions
+19 -8
View File
@@ -13,8 +13,8 @@
并发模型(异步版):: 并发模型(异步版)::
build() → asyncio.run(_build_async()) build() → asyncio.run(build_async())
_build_async(): build_async():
asyncio.Semaphore(concurrency) 控制最大 VLM/LLM 并发数 asyncio.Semaphore(concurrency) 控制最大 VLM/LLM 并发数
各 L1 段并发构建,段内 L2 clip 各启动 _chain 协程: 各 L1 段并发构建,段内 L2 clip 各启动 _chain 协程:
提取全部 L3 帧 → 采样 L2 代表帧 → L2 VLM → L3 VLM 提取全部 L3 帧 → 采样 L2 代表帧 → L2 VLM → L3 VLM
@@ -156,8 +156,8 @@ class VideoTreeBuilder:
转化为三层 TreeIndex。 转化为三层 TreeIndex。
并发架构: 并发架构:
build() 为同步壳,内部调用 asyncio.run(_build_async())。 build() 为同步壳,内部调用 asyncio.run(build_async())。
_build_async() 使用 asyncio.Semaphore(concurrency) 控制并发 VLM/LLM 数量。 build_async() 使用 asyncio.Semaphore(concurrency) 控制并发 VLM/LLM 数量。
所有 VLM 调用通过 VLMProvider 的异步接口发起,零线程阻塞。 所有 VLM 调用通过 VLMProvider 的异步接口发起,零线程阻塞。
所有 LLM 调用通过 LLMProvider 的异步接口发起(L1 摘要)。 所有 LLM 调用通过 LLMProvider 的异步接口发起(L1 摘要)。
ffmpeg 提帧在独立 ThreadPoolExecutor 中并行,不阻塞事件循环。 ffmpeg 提帧在独立 ThreadPoolExecutor 中并行,不阻塞事件循环。
@@ -166,6 +166,7 @@ class VideoTreeBuilder:
_vlm: VLM 图文调用端口。 _vlm: VLM 图文调用端口。
_llm: LLM 文本调用端口(L1 摘要)。 _llm: LLM 文本调用端口(L1 摘要)。
_config: 树构建配置。 _config: 树构建配置。
_api_semaphore: 外部注入的全局 API 并发信号量(None 时 build_async 自建)。
_ffmpeg_pool: ffmpeg 专用线程池(max_workers=_FFMPEG_MAX_WORKERS)。 _ffmpeg_pool: ffmpeg 专用线程池(max_workers=_FFMPEG_MAX_WORKERS)。
""" """
@@ -174,6 +175,8 @@ class VideoTreeBuilder:
vlm: VLMProvider, vlm: VLMProvider,
llm: LLMProvider, llm: LLMProvider,
config: TreeConfig, config: TreeConfig,
*,
api_semaphore: asyncio.Semaphore | None = None,
) -> None: ) -> None:
"""初始化视频树构建器。 """初始化视频树构建器。
@@ -183,10 +186,14 @@ class VideoTreeBuilder:
config: 树构建配置(TreeConfig),关键字段: config: 树构建配置(TreeConfig),关键字段:
l1_segment_duration, l2_clip_duration, l3_fps, l1_segment_duration, l2_clip_duration, l3_fps,
l2_representative_frames, cache_dir, concurrency。 l2_representative_frames, cache_dir, concurrency。
api_semaphore: 外部注入的全局 VLM/LLM 并发信号量(批量建树时跨视频共享);
None 时 build_async 内部按 config.concurrency 自建,
单视频行为零变化。
""" """
self._vlm = vlm self._vlm = vlm
self._llm = llm self._llm = llm
self._config = config self._config = config
self._api_semaphore = api_semaphore
self._ffmpeg_pool = ThreadPoolExecutor(max_workers=_FFMPEG_MAX_WORKERS) self._ffmpeg_pool = ThreadPoolExecutor(max_workers=_FFMPEG_MAX_WORKERS)
self._cache_root = Path(self._config.cache_dir) self._cache_root = Path(self._config.cache_dir)
self._session_id: str = "" self._session_id: str = ""
@@ -296,13 +303,13 @@ class VideoTreeBuilder:
返回: 返回:
三层 TreeIndex 对象。 三层 TreeIndex 对象。
""" """
return asyncio.run(self._build_async(video_path, srt_entries)) return asyncio.run(self.build_async(video_path, srt_entries))
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# 核心异步构建逻辑(保真算法 #1:L2→L3 链式触发) # 核心异步构建逻辑(保真算法 #1:L2→L3 链式触发)
# ------------------------------------------------------------------ # ------------------------------------------------------------------
async def _build_async( async def build_async(
self, self,
video_path: str, video_path: str,
srt_entries: list[SRTEntry] | None = None, srt_entries: list[SRTEntry] | None = None,
@@ -353,8 +360,12 @@ class VideoTreeBuilder:
# Phase 1.1: 读取已有进度(保真算法 #3:断点续跑) # Phase 1.1: 读取已有进度(保真算法 #3:断点续跑)
finished_l1_ids = self._load_resume_state(source_id, total_l1) finished_l1_ids = self._load_resume_state(source_id, total_l1)
# 创建 VLM/LLM 并发控制信号量 # 创建 VLM/LLM 并发控制信号量(外部注入时跨视频全局共享,Spec-2)
vlm_sem = asyncio.Semaphore(self._config.concurrency) 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) # Phase 2-5: 按 L1 段并发,段内 L2→L3 链式触发(保真算法 #1)
async def _build_segment( async def _build_segment(
+42
View File
@@ -14,6 +14,7 @@
from __future__ import annotations from __future__ import annotations
import asyncio
import json import json
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -997,3 +998,44 @@ class TestHelpers:
long_name = "a" * 100 + ".mp4" long_name = "a" * 100 + ".mp4"
stem = VideoTreeBuilder._source_stem(f"/path/{long_name}") stem = VideoTreeBuilder._source_stem(f"/path/{long_name}")
assert len(stem) == 64 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)