From 112f0eb2771eee1e4773a2c53e9a0ff1d48daabf Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sat, 19 Sep 2026 22:43:57 +0800 Subject: [PATCH] =?UTF-8?q?feat(gpu):=20#1978=20AI=E6=95=B0=E5=AD=97?= =?UTF-8?q?=E4=BA=BA=E5=8F=A3=E5=9E=8B=E5=90=8C=E6=AD=A5=E6=8E=A5=E5=85=A5?= =?UTF-8?q?MuseTalk=20GPU=20Worker=EF=BC=88=E4=B8=9A=E5=8A=A1=E4=BE=A7?= =?UTF-8?q?=E9=9B=86=E6=88=90=EF=BC=89=20(#1991)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 9 ++ apps/api/app/services/gpu_lipsync_service.py | 55 +++++++ apps/api/app/services/lipsync_service.py | 94 ++++++++++- deploy/configs/.env.production | 4 + deploy/configs/.env.staging | 4 + packages/config/base.py | 9 ++ tests/unit/test_lipsync_gpu_integration.py | 155 +++++++++++++++++++ 7 files changed, 329 insertions(+), 1 deletion(-) create mode 100644 tests/unit/test_lipsync_gpu_integration.py diff --git a/.env.example b/.env.example index 525794557..6c585351b 100755 --- a/.env.example +++ b/.env.example @@ -220,4 +220,13 @@ GPU_WORKER_TOKEN= # 单任务超时（秒），processing 超过此时长无任务心跳才回退 pending 或标记 failed # #1970：RTX2060 6G 推理 720p 长视频需 5 分钟以上，默认 900 GPU_TASK_TIMEOUT_SECONDS=900 +# 是否启用 GPU 口型同步（开关）。开启后需同时有 Worker 在心跳窗口内（5分钟）才会走 GPU 路径； +# 开关关闭 / 无可用 Worker / GPU 任务失败或超时 → 自动回退现有 MediaKit 云端 lipsync +USE_GPU_LIPSYNC=false +# 业务侧轮询 GPU 任务结果的间隔（秒） +GPU_LIPSYNC_POLL_INTERVAL=5 +# 业务侧等待 GPU 任务总超时（秒）；超时回退 MediaKit +GPU_LIPSYNC_WAIT_TIMEOUT=1200 +# Worker 心跳新鲜度窗口（秒），last_heartbeat_at 在此窗口内视为在线 +GPU_WORKER_STALE_SECONDS=300 diff --git a/apps/api/app/services/gpu_lipsync_service.py b/apps/api/app/services/gpu_lipsync_service.py index f9491006f..485285dbd 100644 --- a/apps/api/app/services/gpu_lipsync_service.py +++ b/apps/api/app/services/gpu_lipsync_service.py @@ -325,3 +325,58 @@ class GpuLipsyncService: t.updated_at = now if stuck_tasks: self.db.flush() + + # ── 业务侧辅助 ────────────────────────────────────────────────── + + def has_available_worker(self) -> bool: + """判断是否有 Worker 在心跳新鲜窗口内可用.""" + stale_cutoff = datetime.now(UTC) - timedelta(seconds=self.settings.gpu_worker_stale_seconds) + return ( + self.db.query(GpuWorkerModel).filter(GpuWorkerModel.last_heartbeat_at >= stale_cutoff).first() is not None + ) + + def wait_for_result( + self, + task_id: str, + timeout_seconds: Optional[int] = None, + poll_interval: Optional[float] = None, + ) -> Optional[GpuLipsyncTaskModel]: + """同步轮询等待 GPU 任务完成。 + + Args: + task_id: 任务 ID（由 create_task 返回） + timeout_seconds: 总超时，默认取 settings.gpu_lipsync_wait_timeout + poll_interval: 轮询间隔秒，默认取 settings.gpu_lipsync_poll_interval + + Returns: + 终态 task（status=done/failed）；超时返回 None（此时调用方应回退 MediaKit）。 + 等待期间会自动调用 _recover_timed_out_tasks 做超时回收。 + """ + import time + + timeout = timeout_seconds if timeout_seconds is not None else self.settings.gpu_lipsync_wait_timeout + interval = poll_interval if poll_interval is not None else self.settings.gpu_lipsync_poll_interval + deadline = time.monotonic() + timeout + + while True: + now = datetime.now(UTC) + # 顺手回收超时任务 + try: + self._recover_timed_out_tasks(now) + self.db.commit() + except Exception as exc: # noqa: BLE001 - 回收失败不阻塞主流程 + logger.warning("wait_for_result 回收超时任务异常: %s", exc) + self.db.rollback() + + task = self.db.get(GpuLipsyncTaskModel, task_id) + if task is None: + return None + if task.status == "done": + return task + if task.status == "failed": + return task + # pending/processing 继续等 + if time.monotonic() >= deadline: + logger.warning("GPU 任务 %s 等待超时（%ds），回退 MediaKit", task_id, timeout) + return None + time.sleep(interval) diff --git a/apps/api/app/services/lipsync_service.py b/apps/api/app/services/lipsync_service.py index 10110d27d..f0080e972 100644 --- a/apps/api/app/services/lipsync_service.py +++ b/apps/api/app/services/lipsync_service.py @@ -36,6 +36,7 @@ from sqlalchemy.orm import Session from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel from packages.application.cosyvoice_service import CosyVoiceError +from packages.config import get_api_settings from packages.domain.sentence_timings import ( compute_sentence_timings, probe_audio_duration, @@ -63,6 +64,7 @@ class LipsyncService: self.client = client or get_mediakit_client() self._cosyvoice = cosyvoice_service self._voice_clone_repo = voice_clone_repo + self.settings = get_api_settings() def _get_cosyvoice(self): """延迟获取 CosyVoiceService（与 tts 路由一致，含 OSS 预签名配置）.""" @@ -215,7 +217,52 @@ class LipsyncService: if timings: job.sentence_timings = timings - # 4. 签名 URL 并提交 MediaKit + # 4. 检查是否走 GPU 路径：开关打开 + 有可用 Worker + use_gpu = False + if self.settings.use_gpu_lipsync: + try: + from app.services.gpu_lipsync_service import GpuLipsyncService + + gpu_svc = GpuLipsyncService(self.db) + if gpu_svc.has_available_worker(): + use_gpu = True + logger.info("[lipsync] 检测到可用 GPU Worker，优先走 MuseTalk 本地推理: job_id=%s", job.id) + else: + logger.info("[lipsync] GPU 开关已开但无可用 Worker（心跳过期），回退 MediaKit: job_id=%s", job.id) + except Exception as exc: + logger.warning("[lipsync] GPU 服务初始化失败，回退 MediaKit: job_id=%s err=%s", job.id, exc) + + if use_gpu: + try: + gpu_task = self._submit_to_gpu(job=job, gpu_svc=gpu_svc) + if gpu_task is not None: + # GPU 任务完成：直接把结果写入 job，标为 completed + job.mediakit_task_id = "" # GPU 路径不走 MediaKit + job.status = STATUS_COMPLETED + job.output_video_url = gpu_task.result_url + job.output_duration = gpu_task.result_duration or 0.0 + job.completed_at = datetime.now(UTC) + job.updated_at = datetime.now(UTC) + self.db.commit() + logger.info( + "[lipsync] GPU MuseTalk 推理完成: job_id=%s gpu_task=%s duration=%.2f", + job.id, + gpu_task.id, + job.output_duration, + ) + # 转存到持久 OSS 路径（GPU 结果已在 gpu-lipsync/results/ 下，直接签短链） + return + # wait_for_result 返回 None 表示超时/最终失败 → 继续走 MediaKit 兜底 + logger.warning("[lipsync] GPU 任务等待超时或失败，回退 MediaKit: job_id=%s", job.id) + self.db.rollback() # 回滚可能的中间状态 + except Exception as exc: + logger.exception("[lipsync] GPU 路径异常，回退 MediaKit: job_id=%s err=%s", job.id, exc) + try: + self.db.rollback() + except Exception: + pass + + # 5. 签名 URL 并提交 MediaKit（兜底路径） video_url = self._sign_media_url(job.video_url) signed_audio_url = self._sign_media_url(job.audio_url) job.audio_url = signed_audio_url @@ -244,6 +291,51 @@ class LipsyncService: self.db.commit() raise + # ── GPU MuseTalk 路径 ──────────────────────────────────────────────── + + def _submit_to_gpu(self, *, job, gpu_svc) -> Optional[object]: + """创建 GPU 任务并同步等待结果。 + + 成功返回终态 task 对象（status=done）；超时或 GPU 最终失败返回 None， + 调用方回退 MediaKit。 + + 注意：job.video_url / job.audio_url 可能是： + - 自家 OSS 存储 key（storage.is_own_url 判断，gpu_svc.create_task 内部 + get_download_url 会自动签预签名 URL 给 Worker） + - 外部公网 URL（CosyVoice 临时链接等）：poll 返回时原样透传给 Worker， + Worker 可直接 GET 下载。 + """ + # 创建 GPU 任务 + gpu_task = gpu_svc.create_task( + video_url=job.video_url, + audio_url=job.audio_url, + lipsync_job_id=job.id, + user_id=job.user_id, + project_id=job.project_id, + ) + logger.info( + "[lipsync] 已创建 GPU 任务: job_id=%s gpu_task=%s", + job.id, + gpu_task.id, + ) + # 同步等待 Worker 处理完成（轮询 DB） + final_task = gpu_svc.wait_for_result(gpu_task.id) + if final_task is None: + logger.warning("[lipsync] GPU 任务等待超时，回退 MediaKit: gpu_task=%s", gpu_task.id) + return None + if final_task.status != "done": + logger.warning( + "[lipsync] GPU 任务失败: gpu_task=%s status=%s err=%s", + gpu_task.id, + final_task.status, + final_task.error_msg, + ) + return None + # result_url 是 OSS 存储 key；签一个长有效期 URL 写回 job.output_video_url + result_signed = self._sign_media_url(final_task.result_url) + final_task.result_url = result_signed or final_task.result_url + return final_task + # ── 创建任务 ────────────────────────────────────────────────────────── def create_job( diff --git a/deploy/configs/.env.production b/deploy/configs/.env.production index 5f3cdaad5..3fe65a938 100644 --- a/deploy/configs/.env.production +++ b/deploy/configs/.env.production @@ -259,3 +259,7 @@ APIZERO_API_KEY=${APIZERO_API_KEY} # ==================== GPU MuseTalk Worker（反向轮询） ==================== GPU_WORKER_TOKEN=${GPU_WORKER_TOKEN} GPU_TASK_TIMEOUT_SECONDS=900 +USE_GPU_LIPSYNC=false +GPU_LIPSYNC_POLL_INTERVAL=5 +GPU_LIPSYNC_WAIT_TIMEOUT=1200 +GPU_WORKER_STALE_SECONDS=300 diff --git a/deploy/configs/.env.staging b/deploy/configs/.env.staging index 374f4d91e..63ee299da 100644 --- a/deploy/configs/.env.staging +++ b/deploy/configs/.env.staging @@ -276,3 +276,7 @@ APIZERO_API_KEY=${APIZERO_API_KEY} # ==================== GPU MuseTalk Worker（反向轮询） ==================== GPU_WORKER_TOKEN=${GPU_WORKER_TOKEN} GPU_TASK_TIMEOUT_SECONDS=900 +USE_GPU_LIPSYNC=true +GPU_LIPSYNC_POLL_INTERVAL=5 +GPU_LIPSYNC_WAIT_TIMEOUT=1200 +GPU_WORKER_STALE_SECONDS=300 diff --git a/packages/config/base.py b/packages/config/base.py index b2c56914e..c6e50622e 100755 --- a/packages/config/base.py +++ b/packages/config/base.py @@ -93,6 +93,15 @@ class SharedSettings(BaseSettings): gpu_result_url_expires: int = 3600 # 输入预签名 URL 有效期（秒，需留出 Worker 下载时间） gpu_input_url_expires: int = 3600 + # 业务侧是否启用 GPU 口型同步（开关）；关或无可用 Worker 时回退 MediaKit 云端 + use_gpu_lipsync: bool = False + # 业务侧轮询 GPU 任务结果的间隔（秒） + gpu_lipsync_poll_interval: float = 5.0 + # 业务侧等待 GPU 任务结果的总超时（秒）；超时后回退 MediaKit。 + # 应小于等于 gpu_task_timeout_seconds（默认900s）+ 冗余，留足 Worker 下载/上传时间。 + gpu_lipsync_wait_timeout: int = 1200 + # 判断 Worker 可用的心跳新鲜度窗口（秒）—— last_heartbeat_at 在窗口内视为在线 + gpu_worker_stale_seconds: int = 300 @property def effective_database_url(self) -> str: diff --git a/tests/unit/test_lipsync_gpu_integration.py b/tests/unit/test_lipsync_gpu_integration.py new file mode 100644 index 000000000..98c978faa --- /dev/null +++ b/tests/unit/test_lipsync_gpu_integration.py @@ -0,0 +1,155 @@ +"""LipsyncService GPU 路径集成测试.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.fixture() +def fake_db(): + db = MagicMock() + return db + + +@pytest.fixture() +def fake_mediakit(): + client = MagicMock() + client.submit_lipsync.return_value = {"task_id": "mk-task-1"} + return client + + +def _make_job(video_url="oss://video.mp4", audio_url="oss://audio.wav"): + job = MagicMock() + job.id = "job-1" + job.user_id = "u1" + job.project_id = "p1" + job.video_url = video_url + job.audio_url = audio_url + job.enable_video_loop = True + job.script_text = "" + job.sentence_timings = None + return job + + +def _make_svc(db, mediakit, use_gpu=False): + from app.services.lipsync_service import LipsyncService + + svc = LipsyncService(db=db, client=mediakit) + svc.settings.use_gpu_lipsync = use_gpu + svc._sign_media_url = lambda u: (u or "") + "?signed" + return svc + + +class TestGpuFallback: + def test_switch_off_uses_mediakit(self, fake_db, fake_mediakit): + """开关关闭时直接走 MediaKit，不调用 _submit_to_gpu.""" + svc = _make_svc(fake_db, fake_mediakit, use_gpu=False) + job = _make_job() + with patch.object(svc, "_submit_to_gpu") as m_sub: + svc._submit_audio_direct(job=job) + m_sub.assert_not_called() + fake_mediakit.submit_lipsync.assert_called_once() + assert job.status == "submitted" + + def test_switch_on_no_worker_falls_back(self, fake_db, fake_mediakit): + """开关打开但 has_available_worker=False → 回退 MediaKit.""" + svc = _make_svc(fake_db, fake_mediakit, use_gpu=True) + fake_gpu_svc = MagicMock() + fake_gpu_svc.has_available_worker.return_value = False + with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc): + job = _make_job() + svc._submit_audio_direct(job=job) + fake_gpu_svc.create_task.assert_not_called() + fake_mediakit.submit_lipsync.assert_called_once() + assert job.status == "submitted" + + def test_gpu_success_marks_completed(self, fake_db, fake_mediakit): + """GPU 路径成功：job 直接 completed，不调 MediaKit.""" + svc = _make_svc(fake_db, fake_mediakit, use_gpu=True) + gpu_done = MagicMock( + id="gpu-task-1", + status="done", + result_url="oss://gpu-results/r.mp4", + result_duration=12.5, + ) + fake_gpu_svc = MagicMock() + fake_gpu_svc.has_available_worker.return_value = True + fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-1") + fake_gpu_svc.wait_for_result.return_value = gpu_done + with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc): + job = _make_job() + svc._submit_audio_direct(job=job) + fake_gpu_svc.create_task.assert_called_once() + fake_mediakit.submit_lipsync.assert_not_called() + assert job.status == "completed" + assert job.output_duration == 12.5 + assert "?signed" in job.output_video_url + fake_db.commit.assert_called() + + def test_gpu_timeout_falls_back(self, fake_db, fake_mediakit): + """wait_for_result 返回 None（超时）→ 回退 MediaKit.""" + svc = _make_svc(fake_db, fake_mediakit, use_gpu=True) + fake_gpu_svc = MagicMock() + fake_gpu_svc.has_available_worker.return_value = True + fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-t") + fake_gpu_svc.wait_for_result.return_value = None + with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc): + job = _make_job() + svc._submit_audio_direct(job=job) + fake_mediakit.submit_lipsync.assert_called_once() + assert job.status == "submitted" + + def test_gpu_failed_status_falls_back(self, fake_db, fake_mediakit): + """GPU 终态 failed → 回退 MediaKit.""" + svc = _make_svc(fake_db, fake_mediakit, use_gpu=True) + fake_gpu_svc = MagicMock() + fake_gpu_svc.has_available_worker.return_value = True + fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-t") + fake_gpu_svc.wait_for_result.return_value = MagicMock(status="failed", error_msg="musetalk crash") + with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc): + job = _make_job() + svc._submit_audio_direct(job=job) + fake_mediakit.submit_lipsync.assert_called_once() + assert job.status == "submitted" + + def test_gpu_exception_falls_back(self, fake_db, fake_mediakit): + """GPU 路径抛异常 → 回退 MediaKit.""" + svc = _make_svc(fake_db, fake_mediakit, use_gpu=True) + fake_gpu_svc = MagicMock() + fake_gpu_svc.has_available_worker.return_value = True + fake_gpu_svc.create_task.side_effect = RuntimeError("DB down") + with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc): + job = _make_job() + svc._submit_audio_direct(job=job) + fake_mediakit.submit_lipsync.assert_called_once() + assert job.status == "submitted" + + +class TestGpuServiceHelpers: + """GpuLipsyncService.has_available_worker 测试.""" + + def test_no_workers(self, fake_db): + from app.services.gpu_lipsync_service import GpuLipsyncService + + svc = GpuLipsyncService(db=fake_db) + fake_db.query.return_value.filter.return_value.first.return_value = None + assert svc.has_available_worker() is False + + def test_fresh_worker_available(self, fake_db): + from app.services.gpu_lipsync_service import GpuLipsyncService + + svc = GpuLipsyncService(db=fake_db) + svc.settings.gpu_worker_stale_seconds = 300 + # 模拟SQL filter条件成立 → first() 返回非None + fake_db.query.return_value.filter.return_value.first.return_value = MagicMock() + assert svc.has_available_worker() is True + + def test_stale_worker_unavailable(self, fake_db): + from app.services.gpu_lipsync_service import GpuLipsyncService + + svc = GpuLipsyncService(db=fake_db) + # filter条件不成立（stale）→ first() 返回None + fake_db.query.return_value.filter.return_value.first.return_value = None + assert svc.has_available_worker() is False