diff --git a/docs/configuration.md b/docs/configuration.md index 1c2d022..e0ed55a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -33,7 +33,12 @@ http://127.0.0.1:8000/docs API 文档 | `WOV_CLEANUP_GRACE_SECONDS` | `3600` | 孤儿清理宽限期（秒） | | `WOV_BATCH_ENABLED` | `1` | 开启文件夹批量处理引擎（处理 source=batch 任务） | | `WOV_BATCH_INTERVAL_SECONDS` | `1.0` | 批量引擎轮询间隔 | -| `WOV_BATCH_STAGE_GROUP_SIZE` | `8` | 批量「分块流水线」分组大小：每组视频按节点顺序跑完全部阶段（全部 extract → 全部 ASR → 全部翻译 → 全部 ASS）再进入下一组，本地模型每组只加载一次；设为 1 等价于每个视频各跑完整链路（产物逐视频落地最及时） | +| `WOV_BATCH_STAGE_GROUP_SIZE` | `8` | 批量流水线分组大小：组内每个视频独立推进阶段（详见 [operations.md](./operations.md#文件夹批量处理)），GPU 阶段组内串行、非 GPU 阶段并行；本机模型每组只加载一次 | +| `WOV_BATCH_PIPELINE_WORKERS` | `4` | 批量组内在途阶段上限：提音/线上翻译/ASS 等不吃 GPU 的阶段可并行，GPU 阶段仍互斥 | +| `WOV_DB_BUSY_TIMEOUT_SECONDS` | `30` | SQLite 写锁等待时长（并发写产物/进度时排队而不是立刻报错）；库自动开启 WAL | +| `WOV_LOCAL_MODEL_HOSTS` | 空 | 视为「本机模型」的额外主机名/IP（逗号分隔）：Ollama 跑在本机局域网地址上时必须声明，否则会被当成不占显存的远端端点 | +| `WOV_LOCAL_LLM_RESERVE_MB` | `22528` | 本机 LLM 端点占用的显存预留（MB）：常驻期间转写阶段会等待显存 | +| `WOV_LOCAL_VLM_RESERVE_MB` | `8192` | 本机 VLM/OCR 端点占用的显存预留（MB） | | `WOV_AUTO_VAD` | `1` | 开启每视频自适应 VAD 调参（详见 [adaptive_vad.md](./adaptive_vad.md)） | | `WHISPER_MODEL_PATH` | 见 [模型权重解析](./node-protocol.md#模型权重解析本地优先) | 显式指定 whisper 模型路径 | | `WHISPER_DEVICE` | `auto` | 转写设备 | diff --git a/docs/decisions.md b/docs/decisions.md index 2887086..1f589fd 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -85,6 +85,23 @@ “COMPLETED 但仍有未结束明细”的脏数据。回归测试见 `test_batch.py` 的 `test_job_hidden_from_engine_until_details_written`。 +## 批量流水线按 GPU 资源调度（而不是按"是否云端"分支） + +- **现象**：LLM 走线上端点时翻译阶段不占显存，但组内阶段仍是"先全部提音、再全部 + 转写、再全部翻译"的串行推进，翻译期间 GPU 全程空闲（实测 19.6W / 272MiB， + 占整轮挂钟约 40%）。 +- **结论**：阶段能否启动只看资源。`wov_app.resources.stage_gpu_need_mb` 给出每个 + 阶段的显存需求（whisper 按权重 ×1.45、本机 LLM 端点按预留、线上端点为 0）， + 进程内 `GpuGate` 负责准入：不需要 GPU 的阶段立刻放行（转写与线上翻译并行）； + 需要 GPU 的阶段互斥，并按"阶段索引最小者优先"派发，组内因此仍是先跑完全部转写 + 再进翻译——本机 Ollama 模型每组只加载一次，行为与旧实现一致。 +- **为什么不是"配置分支"**：代码不判断端点是不是云端，只读"这个阶段要不要占 + GPU、现在够不够"。同一份代码在本机模型下自动退化为串行、在线上模型下自动并行。 +- **降级**：探测不到 `nvidia-smi` 时退化为 GPU 阶段互斥（与串行结果一致）； + 局域网地址上的本机 Ollama 需用 `WOV_LOCAL_MODEL_HOSTS` 声明，否则会被当成远端。 +- **并发写**：流水线多线程写产物/进度，SQLite 开 WAL + busy timeout + （`WOV_DB_BUSY_TIMEOUT_SECONDS`）。 + ## 相关文档 - 当前生效的参数与协议：[node-protocol.md](./node-protocol.md) diff --git a/docs/operations.md b/docs/operations.md index 241235e..663a101 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -104,15 +104,20 @@ run 的 `current_node_id` 在 DAG 拓扑序中的位置推导、节点类型映射中文标签。 粒度限制：`progress` 只有**节点边界**粒度，句级进度（转写分块、翻译批次、OCR 帧） 不落库、只在控制台日志里。 -- **分块流水线执行（本地模型只加载一次）**：批量引擎把待处理视频按 - `WOV_BATCH_STAGE_GROUP_SIZE`（默认 8）分组，**组内按节点顺序跑完全部视频** - （先全部 extract、再全部 ASR、再全部 LLM 翻译、最后 ASS）再进入下一组。每个 - 视频的 run 在阶段边界保持 RUNNING（`execute_run(stop_after=节点)`），下一阶段 - 从产物表跳过已完成节点继续，因此本地模型每组只加载一次、卸载一次，而不是每个 - 视频来回加载卸载；产物仍按组增量落地。LLM 阶段执行时引擎在 run 根目录写 - `keep_model.flag`，节点据此不在每次调用后卸载模型（`nodes/llm.py`），阶段 - 结束由引擎调 `release_local_model()` 统一释放显存，让下一组的 ASR 拿到 GPU - （否则本地模型常驻显存会让 whisper 直接 CUDA OOM）。设为 1 即回到「每个视频 +- **按资源调度的在途流水线（2026-09 起）**：批量引擎把待处理视频按 + `WOV_BATCH_STAGE_GROUP_SIZE`（默认 8）分组，组内**每个视频独立推进自己的 + 阶段**，最多 `WOV_BATCH_PIPELINE_WORKERS`（默认 4）个阶段在途。判定只看资源： + 阶段是否需要 GPU 由 `wov_app.resources.stage_gpu_need_mb` 给出（提音、**线上 + 端点**的翻译、ASS 都不需要），不需要 GPU 的阶段立刻派发，于是转写能与线上翻译 + 并行、提音能与转写并行（此前组内阶段是串行的，翻译时 GPU 全程空转）。需要 GPU + 的阶段由进程内 `GpuGate` 串行准入并按"阶段索引最小者优先"派发，因此组内仍是 + 先跑完全部转写再进翻译——**本机 Ollama 模型每组只加载一次**，不需要按"是否 + 云端"写分支。每个视频的 run 在阶段边界保持 RUNNING（`execute_run(stop_after=节点)`）， + 下一阶段从产物表跳过已完成节点继续。LLM 阶段执行时引擎在 run 根目录写 + `keep_model.flag`，节点据此不在每次调用后卸载模型（`nodes/llm.py`），组末由 + 引擎调 `release_local_model()` 统一释放显存，让下一组的 ASR 拿到 GPU（否则 + 本机模型常驻显存会让 whisper 直接 CUDA OOM）。显存探测不到（无 `nvidia-smi`） + 时退化为"GPU 阶段互斥"，行为与串行一致。设为 1 即回到「每个视频 跑完整链路」的旧行为。 - **失败视频不跨阶段推进**：某阶段失败的视频只在**它失败节点的那个阶段**重试 （下一次引擎循环从断点续跑），不会在后续阶段里重跑前序节点——避免本地 LLM 已 diff --git a/src/wov_app/batch.py b/src/wov_app/batch.py index 2b5ad15..d095396 100644 --- a/src/wov_app/batch.py +++ b/src/wov_app/batch.py @@ -11,15 +11,19 @@ 字幕文件（`.srt/.ass/.ssa/.vtt`），说明该视频已有字幕，直接记为 SKIPPED， 不为它触发任何流水线。运行时（BatchWorker）只消费已定位好的明细列表， **不再重新扫描文件夹**（运行期间新增/删除的视频不会改变本次任务的范围）。 -- **分块流水线执行**：视频按 `WOV_BATCH_STAGE_GROUP_SIZE` 分组，组内按节点 - 顺序跑完全部视频（先全部 extract、再全部 ASR、再全部 LLM 翻译、最后 ASS） - 再进入下一组——本地模型每组只加载一次、卸载一次，产物按组增量落地。 +- **按资源调度的在途流水线**：视频按 `WOV_BATCH_STAGE_GROUP_SIZE` 分组，组内 + 每个视频独立推进自己的阶段（最多 `WOV_BATCH_PIPELINE_WORKERS` 个阶段在途）。 + 阶段是否需要 GPU 由 `wov_app.resources` 判定：提音、线上翻译与 ASS 不需要， + 与其它视频的转写并行（GPU 不再空转）；需要 GPU 的阶段由进程内门控 + `GpuGate` 串行准入，并按"阶段索引最小者优先"派发，于是组内先跑完全部转写 + 再进翻译——本机 Ollama 模型每组只加载一次（线上端点则完全不受该顺序约束）。 阶段边界用 `execute_run(stop_after=节点)` 停在节点（任务保持 RUNNING）， - LLM 阶段靠 `keep_model.flag` 让节点保持模型常驻，阶段结束由引擎统一释放 - 显存（详见 docs/operations.md#文件夹批量处理）。 + LLM 阶段靠 `keep_model.flag` 让节点保持模型常驻，组末由引擎统一释放显存 + （详见 docs/operations.md#文件夹批量处理）。 - **产物放在视频旁**：每个视频处理完成后，把工作流 `final_outputs` 对应的 - 最终产物文件（字幕流水线即中文 `.srt` 与双目 `.ass`）**复制一份到视频的 - 所在目录**，与 .mp4 放在一起；文件名**对齐媒体库既有约定**：中文字幕存为 + 最终产物文件（字幕流水线即日语转写 `.srt`、中文 `.srt` 与双目 `.ass`） + **复制一份到视频的所在目录**，与 .mp4 放在一起；文件名按 `final_outputs` + 别名**对齐媒体库既有约定**：日语转写存为 `<视频名>.JA.srt`、中文字幕存为 `<视频名>.CN.srt`、双目字幕存为 `<视频名>.CN_dual_eye.ass`（文件名稳定且 含视频主名，媒体库可自动匹配，下次批量扫描也会命中"已有字幕"规则跳过）。 - **过程文件清理**：视频收尾完成后删除该视频的整个工作空间与 run 记录， @@ -46,13 +50,20 @@ import shutil import threading import time import uuid +from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait from datetime import datetime, timezone from pathlib import Path from wov_app import registry -from wov_app.config import BATCH_INTERVAL_SECONDS, BATCH_STAGE_GROUP_SIZE, STORAGE_DIR +from wov_app.config import ( + BATCH_INTERVAL_SECONDS, + BATCH_PIPELINE_WORKERS, + BATCH_STAGE_GROUP_SIZE, + STORAGE_DIR, +) from wov_app.db import Database from wov_app.logging import get_logger +from wov_app.resources import GpuGate, stage_gpu_need_mb from wov_app.scheduler import WorkflowScheduler, topological_sort from wov_app.storage import atomic_copy from wov_sdk.models import WorkflowDefinition, WorkflowNode @@ -72,6 +83,14 @@ VIDEO_EXTENSIONS = { # 最终产物（.srt/.ass）也在该集合内，保证下次扫描能命中同一规则直接跳过。 SUBTITLE_EXTENSIONS = {".srt", ".ass", ".ssa", ".vtt"} +# 组内流水线的轮询间隔（秒）：等第一个阶段结束时顺带检查暂停与资源放行。 +PIPELINE_POLL_SECONDS = 0.2 + + +def _make_gate() -> GpuGate: + """创建任务级 GPU 门控（测试通过替换本函数注入假探测结果）。""" + return GpuGate() + # 暂停信号文件名：与节点约定一致，位于 run 根目录（/runs//）。 PAUSE_FLAG = "paused.flag" @@ -420,44 +439,23 @@ class BatchWorker: total = int(job["total"] or 0) group_size = max(1, int(BATCH_STAGE_GROUP_SIZE)) + gate = _make_gate() for start in range(0, len(items), group_size): group = items[start:start + group_size] - for stage_index, node_id in enumerate(order): - node_spec = node_by_id[node_id] - # 末阶段不传 stop_after：让调度器收尾（final_outputs + COMPLETED）。 - is_last_stage = stage_index == len(order) - 1 - executed = False - for item in group: - # 暂停检查：批量任务被暂停后停止处理后续视频，等待用户继续。 - current = self.db.get_batch_job(job_id) - if current is None or current["status"] == "PAUSED": - # 停下前先把已完成/失败项入账，让暂停中的前端看到真实进度。 - self.db.sync_batch_job_progress(job_id) - logger.info("批量任务 %s 已暂停，停止在视频 %s", job_id, item["video_path"]) - return - self.db.update_batch_job(job_id, current_video=str(item["video_path"]), updated_at=_now_iso()) - try: - outcome = self._run_stage( - job, item, version, definition, node_spec, order, is_last_stage, - ) - except Exception as exc: # noqa: BLE001 - # 单视频兜底：不中断整个批量任务，记录错误后继续下一个视频。 - logger.exception( - "批量任务 %s 视频 %s 阶段 %s 处理异常", job_id, item["video_path"], node_id, - ) - self.db.update_batch_video(item["id"], status="FAILED", error=str(exc), updated_at=_now_iso()) - outcome = "FAILED" - executed = executed or outcome is not None - # 每个视频每个阶段后实时同步一次汇总，让进度尽快入账。 - self.db.sync_batch_job_progress(job_id) - # 阶段内被暂停（节点内的 paused.flag）：任务保持 PAUSED 等续跑。 - if outcome == "PAUSED": - self.db.update_batch_job(job_id, status="PAUSED", updated_at=_now_iso()) - return - # 阶段收尾：LLM 阶段结束时统一释放本地模型显存，让下一组的 - # whisper（ASR）拿到 GPU，否则下一个视频转写会 CUDA OOM。 - if executed and node_spec.node_type.startswith(LLM_NODE_PREFIX): - self._release_llm_model(node_spec.params) + outcome = self._run_group( + job=job, + group=group, + order=order, + node_by_id=node_by_id, + definition=definition, + version=version, + gate=gate, + ) + if outcome == "PAUSED": + # 停下前先把已完成/失败项入账，让暂停中的前端看到真实进度。 + self.db.sync_batch_job_progress(job_id) + logger.info("批量任务 %s 已暂停，等待用户继续", job_id) + return # 先按明细实时对齐汇总（done 不计 SKIPPED），再判断能否收尾。 # 仍有未结束视频时不能标 COMPLETED，否则会出现“还有待处理视频却已完成” @@ -496,6 +494,203 @@ class BatchWorker: job_id, total, done, failed, ) + def _is_job_paused(self, job_id: str) -> bool: + """批量任务是否已被暂停（或记录已消失）：暂停后不再派发新阶段。""" + job = self.db.get_batch_job(job_id) + return job is None or job["status"] == "PAUSED" + + def _next_stage_node( + self, + pipeline: dict, + order: list[str], + node_by_id: dict[str, WorkflowNode], + ) -> WorkflowNode | None: + """返回该视频下一个待执行阶段的节点；已跑完或已失败时返回 None。""" + if pipeline["state"] != "ready" or pipeline["stage"] >= len(order): + return None + return node_by_id[order[pipeline["stage"]]] + + def _submit_group_stage( + self, + job: dict, + pipeline: dict, + node_spec: WorkflowNode, + order: list[str], + definition: WorkflowDefinition, + version: dict, + running: dict, + pool: ThreadPoolExecutor, + lease_key: str | None = None, + ) -> None: + """把一个阶段交给线程池执行（异常在池内兜底，不让线程池任务抛出去）。""" + item = pipeline["item"] + is_last_stage = node_spec.id == order[-1] + # 标记在途：同一视频同一阶段只允许有一个执行体，否则会被重复派发。 + pipeline["state"] = "running" + if lease_key: + pipeline["lease_key"] = lease_key + if node_spec.node_type.startswith(LLM_NODE_PREFIX): + pipeline["executed_llm"] = True + pipeline["llm_params"] = node_spec.params + self.db.update_batch_job(job["id"], current_video=str(item["video_path"]), updated_at=_now_iso()) + logger.info( + "批量任务 %s 视频 %s 进入阶段 %s（在途 %d）", + job["id"], Path(item["video_path"]).name, node_spec.id, len(running) + 1, + ) + + def _worker() -> str: + """线程内的阶段执行：单视频异常不中断整组。""" + try: + outcome = self._run_stage( + job, item, version, definition, node_spec, order, is_last_stage, + ) + except Exception as exc: # noqa: BLE001 + logger.exception( + "批量任务 %s 视频 %s 阶段 %s 处理异常", + job["id"], item["video_path"], node_spec.id, + ) + self.db.update_batch_video( + item["id"], status="FAILED", error=str(exc), updated_at=_now_iso(), + ) + return "FAILED" + return outcome or "RUNNING" + + running[pool.submit(_worker)] = pipeline + + def _dispatch_group_stages( + self, + job: dict, + pipelines: list[dict], + order: list[str], + node_by_id: dict[str, WorkflowNode], + definition: WorkflowDefinition, + version: dict, + gate: GpuGate, + running: dict, + pool: ThreadPoolExecutor, + workers: int, + ) -> None: + """派发就绪阶段：非 GPU 阶段可并行，GPU 阶段互斥且按上游优先。""" + # 1) 不需要 GPU 的阶段（提音 / 线上翻译 / ASS）：立刻派发，与其它视频的 + # 转写并行，GPU 不再空转。 + for pipeline in pipelines: + if len(running) >= workers: + break + node_spec = self._next_stage_node(pipeline, order, node_by_id) + if node_spec is None: + continue + if stage_gpu_need_mb(node_spec.node_type, node_spec.params) > 0: + continue + self._submit_group_stage(job, pipeline, node_spec, order, definition, version, running, pool) + # 2) 需要 GPU 的阶段：一次只跑一个，且选"阶段索引最小"的视频——组内因此 + # 先把转写跑完再进翻译，本机 Ollama 模型仍每组只加载一次。 + if len(running) >= workers or gate.holder is not None: + return + candidates = [ + (pipeline["stage"], pipeline["index"], pipeline) + for pipeline in pipelines + if self._next_stage_node(pipeline, order, node_by_id) is not None + ] + gpu_candidates = [ + entry for entry in candidates + if stage_gpu_need_mb( + node_by_id[order[entry[2]["stage"]]].node_type, + node_by_id[order[entry[2]["stage"]]].params, + ) > 0 + ] + if not gpu_candidates: + return + _, _, pipeline = min(gpu_candidates, key=lambda entry: (entry[0], entry[1])) + node_spec = node_by_id[order[pipeline["stage"]]] + need = stage_gpu_need_mb(node_spec.node_type, node_spec.params) + lease_key = f"{pipeline['item']['id']}:{node_spec.id}" + # 显存/在途不满足时本轮跳过，等其它阶段释放后再试（不阻塞派发线程）。 + if not gate.try_acquire(lease_key, need): + return + self._submit_group_stage( + job, pipeline, node_spec, order, definition, version, running, pool, lease_key, + ) + + def _finish_group_stage( + self, + job: dict, + pipeline: dict, + future: Future, + order: list[str], + ) -> None: + """收集一个阶段的执行结果并推进该视频的流水线。""" + try: + outcome = future.result() + except Exception: # noqa: BLE001 - 池内已兜底，这里只保证组不被带崩 + logger.exception("批量任务 %s 阶段执行线程异常", job["id"]) + pipeline["state"] = "failed" + return + self.db.sync_batch_job_progress(job["id"]) + if outcome == "PAUSED": + # 节点在边界（分块/批次/帧）停下：任务保持 PAUSED 等用户继续。 + pipeline["state"] = "paused" + self.db.update_batch_job(job["id"], status="PAUSED", updated_at=_now_iso()) + return + if outcome == "FAILED": + pipeline["state"] = "failed" + return + pipeline["stage"] += 1 + pipeline["state"] = "done" if pipeline["stage"] >= len(order) else "ready" + + def _run_group( + self, + job: dict, + group: list[dict], + order: list[str], + node_by_id: dict[str, WorkflowNode], + definition: WorkflowDefinition, + version: dict, + gate: GpuGate, + ) -> str: + """组内在途流水线：每个视频独立推进阶段，GPU 阶段按上游优先串行。 + + 阶段是否需要 GPU 由 resources.stage_gpu_need_mb 判定（提音/线上翻译/ASS + 不需要），于是它们与其它视频的转写并行；需要 GPU 的阶段由 GpuGate 准入， + 并按"阶段索引最小者优先"派发，组内因此先把转写跑完再进翻译，本机 Ollama + 模型仍每组只加载一次。 + + 返回 "PAUSED" 表示组内被暂停（调用方停止任务），其余情况返回 "DONE"。 + """ + job_id = job["id"] + workers = max(1, int(BATCH_PIPELINE_WORKERS)) + pipelines = [ + {"item": item, "stage": 0, "index": index, "state": "ready", "executed_llm": False} + for index, item in enumerate(group) + ] + with ThreadPoolExecutor(max_workers=workers) as pool: + running: dict[Future, dict] = {} + while True: + paused = self._is_job_paused(job_id) + if not paused: + self._dispatch_group_stages( + job, pipelines, order, node_by_id, definition, version, + gate, running, pool, workers, + ) + if not running: + break + done, _ = wait( + list(running), timeout=PIPELINE_POLL_SECONDS, + return_when=FIRST_COMPLETED, + ) + for future in done: + pipeline = running.pop(future) + lease_key = pipeline.pop("lease_key", None) + if lease_key: + gate.release(lease_key) + self._finish_group_stage(job, pipeline, future, order) + if paused and not running: + return "PAUSED" + # 组末统一释放本地模型：否则下一组的 whisper 转写会 CUDA OOM。 + llm_params = next((p["llm_params"] for p in pipelines if p.get("llm_params")), None) + if llm_params is not None: + self._release_llm_model(llm_params) + return "PAUSED" if self._is_job_paused(job_id) else "DONE" + def _run_stage( self, job: dict, diff --git a/src/wov_app/config.py b/src/wov_app/config.py index 8591dce..8d724c7 100644 --- a/src/wov_app/config.py +++ b/src/wov_app/config.py @@ -29,6 +29,10 @@ SCHEDULER_INTERVAL_SECONDS = float(os.getenv("WOV_SCHEDULER_INTERVAL_SECONDS", " BATCH_ENABLED = os.getenv("WOV_BATCH_ENABLED", "1") == "1" BATCH_INTERVAL_SECONDS = float(os.getenv("WOV_BATCH_INTERVAL_SECONDS", "1.0")) +# 批量流水线的在途阶段数：组内每个视频独立推进，非 GPU 阶段（提音、线上翻译、 +# ASS）并行执行；GPU 阶段仍由资源门控串行化（见 docs/operations.md）。 +BATCH_PIPELINE_WORKERS = int(os.getenv("WOV_BATCH_PIPELINE_WORKERS", "4")) + # 批量"分块流水线"分组大小：每组视频按节点顺序跑完全部阶段（全部 extract → 全部 # ASR → 全部翻译 → 全部 ASS）再处理下一组，使本地模型每组只加载一次；产物仍按 # 组增量落地（详见 docs/operations.md#文件夹批量处理）。 diff --git a/tests/app/test_batch/test_batch.py b/tests/app/test_batch/test_batch.py index f1f6b05..4ad5333 100644 --- a/tests/app/test_batch/test_batch.py +++ b/tests/app/test_batch/test_batch.py @@ -14,6 +14,7 @@ from pathlib import Path import pytest from wov_app import registry +from wov_app.resources import GpuGate from wov_app.batch import ( KEEP_MODEL_FLAG, MARKER_NAME, @@ -561,7 +562,7 @@ def test_worker_processes_recovered_zombie_job(tmp_path: Path, monkeypatch) -> N def test_worker_runs_grouped_stage_pipeline(tmp_path: Path, monkeypatch) -> None: - """分块流水线：组内按节点顺序跑完全部视频，而不是每个视频跑完整链路。""" + """分组流水线：组内每个视频独立推进阶段，阶段顺序仍按 DAG，组间分批。""" # 数据：3 个视频 + 三节点链路，分组大小 2（前两个一组、第三个一组）。 folder = tmp_path / "videos" for name in ("a.mp4", "b.mp4", "c.mp4"): @@ -576,13 +577,12 @@ def test_worker_runs_grouped_stage_pipeline(tmp_path: Path, monkeypatch) -> None job_id = create_job(db, str(folder), "wf") BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id)) - # 验证结果：组1 三个阶段各跑 a/b，再轮到组2 的 c。 - assert trace == [ - ("prep", "a.mp4"), ("prep", "b.mp4"), - ("translate", "a.mp4"), ("translate", "b.mp4"), - ("post", "a.mp4"), ("post", "b.mp4"), - ("prep", "c.mp4"), ("translate", "c.mp4"), ("post", "c.mp4"), - ] + # 验证结果：每个视频的三阶段按 DAG 顺序各跑一次；组 1（a/b）全部跑完才轮到组 2（c）。 + for name in ("a.mp4", "b.mp4", "c.mp4"): + assert [tag for tag, video in trace if video == name] == ["prep", "translate", "post"] + first_group = [i for i, entry in enumerate(trace) if entry[1] in ("a.mp4", "b.mp4")] + second_group = [i for i, entry in enumerate(trace) if entry[1] == "c.mp4"] + assert max(first_group) < min(second_group) # 三个视频都完成且产物按约定名落到视频旁。 assert db.get_batch_job(job_id)["status"] == "COMPLETED" assert sorted(p.name for p in folder.glob("*.srt")) == ["a.CN.srt", "b.CN.srt", "c.CN.srt"] @@ -616,6 +616,155 @@ def test_worker_releases_local_llm_once_per_group(tmp_path: Path, monkeypatch) - assert not list((tmp_path / "storage").rglob(KEEP_MODEL_FLAG)) +@contextmanager +def _timed_nodes(durations: dict[str, float]): + """把三类节点换成"记录起止时间"的假节点，用于断言阶段之间的并行/互斥。 + + 节点类型决定显存需求（`faster-whisper` 需 GPU、`llm-translate` 看端点、 + `echo` 不需要），因此可以直接观察资源门控的调度结果。 + """ + import time as _time + + registry.register_all() + events: list[tuple[str, str, str, float]] = [] + + def handler(request: InvokeRequest) -> InvokeResponse: + tag = str(request.params.get("node_tag")) + # 首阶段可能从 audio_uri 拿到视频路径，后续阶段从 file_uri 拿到上游产物。 + source = str(request.inputs.get("file_uri") or request.inputs.get("audio_uri") or "") + if source and Path(source).suffix.lower() in VIDEO_EXTENSIONS: + video_name = Path(source).name + else: + video_name = Path(source).read_text(encoding="utf-8").strip() if source else "" + events.append((tag, video_name, "start", _time.monotonic())) + # 时长可按 "阶段:视频" 细化（用于制造"A 已转写、B 还在提音"的时序）。 + _time.sleep(durations.get(f"{tag}:{video_name}", durations.get(tag, 0.05))) + output_dir = Path(request.output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + output = output_dir / "payload.srt" + output.write_text(video_name, encoding="utf-8") + events.append((tag, video_name, "end", _time.monotonic())) + return InvokeResponse(status="completed", outputs={"file_uri": str(output)}) + + for node_type in ("echo", "llm-translate", "faster-whisper"): + registry.register(registry.get_node(node_type), handler) + yield events + + +def _overlaps(events, first: tuple[str, str], second: tuple[str, str]) -> bool: + """两个 (阶段, 视频) 执行区间是否重叠（用于断言并行/串行）。""" + def window(key): + start = next(e[3] for e in events if (e[0], e[1], e[2]) == (*key, "start")) + end = next(e[3] for e in events if (e[0], e[1], e[2]) == (*key, "end")) + return start, end + + a_start, a_end = window(first) + b_start, b_end = window(second) + return a_start < b_end and b_start < a_end + + +def _asr_translate_db(tmp_path: Path) -> Database: + """两阶段链路 asr(faster-whisper) → translate(llm) 的临时库。""" + db = Database(tmp_path / "wov.db") + db.upsert_workflow({ + "id": "wf", "name": "转写翻译", "description": "", "published": 1, "latest_version": 1, + }) + definition = { + "name": "转写翻译", "version": 1, + "nodes": [ + {"id": "asr", "node_type": "faster-whisper", "params": {"node_tag": "asr"}, + "inputs": {"audio_uri": "input.video_uri"}}, + {"id": "translate", "node_type": "llm-translate", "params": {"node_tag": "translate"}, + "inputs": {"file_uri": "asr.file_uri"}}, + ], + "edges": [{"from": "asr", "to": "translate"}], + "entry_inputs": {"video_uri": "file"}, + "final_outputs": {"cn_srt": "translate.file_uri"}, + } + db.create_workflow_version("wf", 1, WorkflowDefinition.from_dict(definition).to_dict()) + return db + + +def test_remote_translate_overlaps_other_video_transcription(tmp_path: Path, monkeypatch) -> None: + """线上翻译不占 GPU：A 在翻译时，B 的转写可以同时跑（GPU 不空转）。""" + # 数据：2 个视频 + asr→translate 链路；翻译端点在远端，显存充裕。 + folder = tmp_path / "videos" + for name in ("a.mp4", "b.mp4"): + _make_video(folder / name) + db = _asr_translate_db(tmp_path) + monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch") + monkeypatch.setenv("LLM_API_BASE", "https://api.siliconflow.cn/v1/chat/completions") + monkeypatch.setattr("wov_app.batch._make_gate", lambda: GpuGate(probe=lambda: 24000)) + + # 测试过程 + with _timed_nodes({"asr": 0.4, "translate": 0.6}) as events: + job_id = create_job(db, str(folder), "wf") + BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id)) + + # 验证结果：A 的翻译与 B 的转写时间区间重叠。 + assert _overlaps(events, ("translate", "a.mp4"), ("asr", "b.mp4")) + + +def test_local_llm_endpoint_serializes_and_runs_all_asr_first(tmp_path: Path, monkeypatch) -> None: + """本机 LLM 端点占显存：GPU 阶段互斥，且组内先跑完全部转写再翻译（模型只载一次）。""" + # 数据：2 个视频 + asr→translate 链路；翻译端点在本机（占整卡显存）。 + folder = tmp_path / "videos" + for name in ("a.mp4", "b.mp4"): + _make_video(folder / name) + db = _asr_translate_db(tmp_path) + monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch") + monkeypatch.setenv("LLM_API_BASE", "http://localhost:11434/v1/chat/completions") + monkeypatch.setattr("wov_app.batch._make_gate", lambda: GpuGate(probe=lambda: 24000)) + # 本用例只验调度顺序，不必真去卸载本机模型。 + monkeypatch.setattr("wov_app.batch.release_local_model", lambda model=None: None) + + # 测试过程 + with _timed_nodes({"asr": 0.3, "translate": 0.3}) as events: + job_id = create_job(db, str(folder), "wf") + BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id)) + + # 验证结果：全部转写先于任何翻译；翻译之间互斥不重叠。 + last_asr = max(e[3] for e in events if e[0] == "asr" and e[2] == "end") + first_translate = min(e[3] for e in events if e[0] == "translate" and e[2] == "start") + assert last_asr < first_translate + assert _overlaps(events, ("translate", "a.mp4"), ("translate", "b.mp4")) is False + + +def test_extract_overlaps_transcription_of_other_video(tmp_path: Path, monkeypatch) -> None: + """提音（不需要 GPU）与其它视频的转写并行——组内不再"先全部提音再转写"。""" + # 数据：2 个视频 + extract(echo) → asr(faster-whisper) 链路。 + folder = tmp_path / "videos" + for name in ("a.mp4", "b.mp4"): + _make_video(folder / name) + db = Database(tmp_path / "wov.db") + db.upsert_workflow({ + "id": "wf", "name": "提音转写", "description": "", "published": 1, "latest_version": 1, + }) + definition = { + "name": "提音转写", "version": 1, + "nodes": [ + {"id": "extract", "node_type": "echo", "params": {"node_tag": "extract"}, + "inputs": {"file_uri": "input.video_uri"}}, + {"id": "asr", "node_type": "faster-whisper", "params": {"node_tag": "asr"}, + "inputs": {"audio_uri": "extract.file_uri"}}, + ], + "edges": [{"from": "extract", "to": "asr"}], + "entry_inputs": {"video_uri": "file"}, + "final_outputs": {"cn_srt": "asr.file_uri"}, + } + db.create_workflow_version("wf", 1, WorkflowDefinition.from_dict(definition).to_dict()) + monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch") + monkeypatch.setattr("wov_app.batch._make_gate", lambda: GpuGate(probe=lambda: 24000)) + + # 测试过程 + with _timed_nodes({"extract:a.mp4": 0.05, "extract:b.mp4": 0.8, "asr": 0.4}) as events: + job_id = create_job(db, str(folder), "wf") + BatchWorker(db, interval_seconds=999)._process_job(db.get_batch_job(job_id)) + + # 验证结果：A 的转写与 B 的提音重叠。 + assert _overlaps(events, ("asr", "a.mp4"), ("extract", "b.mp4")) + + def test_worker_defers_video_failed_in_earlier_stage(tmp_path: Path, monkeypatch) -> None: """上一阶段失败的视频不在后续阶段重跑（避免 LLM 已常驻时重跑 ASR 抢显存）。""" # 数据：2 个视频（同一组）+ 三节点链路，prep 阶段让 a 失败。