feat(gpu): #1978 AI数字人口型同步接入MuseTalk GPU Worker

业务侧（LipsyncService）集成 GpuLipsyncService：
- 新增开关 USE_GPU_LIPSYNC（env，默认false；staging默认true方便联调，生产默认false待稳定后开）
- _submit_audio_direct 在签名提交 MediaKit 前判断：
  * 开关关闭 → 直接走现有 MediaKit 云端 lipsync
  * 开关开但无可用 Worker（last_heartbeat_at 超5分钟窗口） → 回退 MediaKit
  * 开关开+Worker可用 → 创建GPU任务→同步轮询等待结果
- GPU路径成功：直接标记 job.status=completed、output_video_url=签名后的结果URL
- GPU超时（默认1200s）/终态failed/异常 → 全部回退 MediaKit，对用户透明
- 轮询间隔默认5s，可配 GPU_LIPSYNC_POLL_INTERVAL
- Worker心跳新鲜度窗口可配 GPU_WORKER_STALE_SECONDS（默认300s）

GpuLipsyncService新增方法：
- has_available_worker()：判断5分钟内有心跳的Worker存在
- wait_for_result(task_id, timeout, poll_interval)：同步轮询DB等待终态，
  期间顺手 _recover_timed_out_tasks；超时返回None让调用方兜底

env/CI配置：
- .env.example 新增4个配置项注释
- deploy/configs/.env.staging USE_GPU_LIPSYNC=true（联调）
- deploy/configs/.env.production USE_GPU_LIPSYNC=false（暂不开）

单元测试：tests/unit/test_lipsync_gpu_integration.py 9个用例覆盖：
- 开关关不调用GPU、无Worker回退、GPU成功completed、超时回退、
  failed回退、异常回退、has_available_worker三种场景
- 全部 30个GPU相关测试全绿（含此前11+10个）

.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
 

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)

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(

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

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

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:

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