添加重试机制

app/service/task/impl/google_veo3_task.py

@@ -8,6 +8,7 @@ from app.service.task.task_param import TaskParam
 from app.service.task.task_result import TaskResult
 from app.util.file_util import generate_unique_filename, write_bytes
 from app.util.http_util import logged_client
+from app.util.retry_util import retry_on_failure
 
 logger = logging.getLogger(__name__)
 
@@ -25,7 +26,10 @@ class GoogleVeo3Task(BaseTask):
     def execute(self, param: TaskParam) -> TaskResult:
         t0 = time.time()
         try:
-            op_name = self._create_request(param.inputs)
+            op_name = retry_on_failure(
+                lambda: self._create_request(param.inputs),
+                task_name="google_veo3",
+            )
             logger.info("[google_veo3] Operation created: %s", op_name)
 
             video_uri = self._poll_until_done(op_name)

app/service/task/impl/keling_image2video_task.py

@@ -8,6 +8,7 @@ from app.service.task.base_task import BaseTask
 from app.service.task.task_param import TaskParam
 from app.service.task.task_result import TaskResult
 from app.util.http_util import logged_client
+from app.util.retry_util import retry_on_failure
 
 logger = logging.getLogger(__name__)
 
@@ -39,34 +40,34 @@ class KelingImage2VideoTask(BaseTask):
             "Authorization": f"Bearer {self._generate_jwt()}",
         }
 
-    def _retry_execute(self, param: TaskParam, remaining: int = 30) -> dict | None:
-        logger.info("[%s] Submitting task: POST %s  (remaining_retries=%d)",
-                     self.get_type(), self.execute_url, remaining)
-        try:
-            with logged_client(timeout=60) as client:
-                resp = client.post(self.execute_url, headers=self._headers(), json=param.inputs)
-                logger.info("[%s] Submit response: status=%d", self.get_type(), resp.status_code)
-                bean = resp.json()
-        except Exception as e:
-            raise RuntimeError(f"Request failed: {e}")
-
+    def _submit(self, inputs: dict) -> dict:
+        with logged_client(timeout=60) as client:
+            resp = client.post(self.execute_url, headers=self._headers(), json=inputs)
+            logger.info("[%s] Submit response: status=%d", self.get_type(), resp.status_code)
+            resp.raise_for_status()
+            bean = resp.json()
         code = bean.get("code", 0)
-        if code in (1302, 1303) and remaining > 0:
-            logger.warning("[%s] Rate limited (code=%d), retrying in 5s ... remaining=%d",
-                            self.get_type(), code, remaining - 1)
-            time.sleep(5)
-            return self._retry_execute(param, remaining - 1)
+        if code in (1302, 1303):
+            raise RuntimeError(f"Rate limited (code={code})")
+        if not bean.get("data"):
+            msg = bean.get("message", "unknown error")
+            raise RuntimeError(f"Submit failed: {msg}")
         return bean
 
     def execute(self, param: TaskParam) -> TaskResult:
         t0 = time.time()
-        bean = self._retry_execute(param)
-        if not bean:
-            return TaskResult.failure("submit task fail !")
-        if not bean.get("data"):
-            msg = bean.get("message", "unknown error")
-            logger.error("[%s] Submit failed: %s", self.get_type(), msg)
-            return TaskResult.failure(msg)
+        logger.info("[%s] Submitting task: POST %s", self.get_type(), self.execute_url)
+        try:
+            bean = retry_on_failure(
+                lambda: self._submit(param.inputs),
+                max_retries=5,
+                backoff_base=3.0,
+                task_name=self.get_type(),
+            )
+        except Exception as e:
+            elapsed = time.time() - t0
+            logger.error("[%s] Submit failed in %.2fs: %s", self.get_type(), elapsed, e)
+            return TaskResult.failure(f"submit task fail: {e}")
 
         task_id = bean["data"].get("task_id")
         logger.info("[%s] Task submitted, remote_id=%s", self.get_type(), task_id)

app/service/task/impl/kontext_pro_task.py

@@ -6,6 +6,7 @@ from app.service.task.base_task import BaseTask
 from app.service.task.task_param import TaskParam
 from app.service.task.task_result import TaskResult
 from app.util.http_util import logged_client
+from app.util.retry_util import retry_on_failure
 
 logger = logging.getLogger(__name__)
 
@@ -24,25 +25,29 @@ class KontextProTask(BaseTask):
             "x-key": settings.kontext_api_key,
         }
 
+    def _submit(self, inputs: dict) -> dict:
+        with logged_client(timeout=60) as client:
+            resp = client.post(self.execute_url, headers=self._headers(), json=inputs)
+            logger.info("[%s] Submit response: status=%d", self.get_type(), resp.status_code)
+            resp.raise_for_status()
+            bean = resp.json()
+        if not bean:
+            raise RuntimeError("Empty response from API")
+        return bean
+
     def execute(self, param: TaskParam) -> TaskResult:
         logger.info("[%s] Submitting task: POST %s", self.get_type(), self.execute_url)
         t0 = time.time()
         try:
-            with logged_client(timeout=60) as client:
-                resp = client.post(self.execute_url, headers=self._headers(), json=param.inputs)
-                elapsed = time.time() - t0
-                logger.info("[%s] Submit response: status=%d, elapsed=%.2fs",
-                            self.get_type(), resp.status_code, elapsed)
-                resp.raise_for_status()
-                bean = resp.json()
+            bean = retry_on_failure(
+                lambda: self._submit(param.inputs),
+                task_name=self.get_type(),
+            )
         except Exception as e:
             elapsed = time.time() - t0
             logger.error("[%s] Submit failed in %.2fs: %s", self.get_type(), elapsed, e)
             return TaskResult.failure(f"submit task fail: {e}")
 
-        if not bean:
-            return TaskResult.failure("submit task fail !")
-
         task_id = bean.get("id")
         polling_url = bean.get("polling_url")
         logger.info("[%s] Task submitted, remote_id=%s, polling_url=%s",

app/service/task/impl/nano_banana_task.py

@@ -6,6 +6,7 @@ from app.service.task.base_task import BaseTask
 from app.service.task.task_param import TaskParam
 from app.service.task.task_result import TaskResult
 from app.util.http_util import logged_client
+from app.util.retry_util import retry_on_failure
 
 logger = logging.getLogger(__name__)
 
@@ -14,6 +15,17 @@ class NanoBananaTask(BaseTask):
     def get_type(self) -> str:
         return "nano_banana"
 
+    def _submit(self, endpoint: str, headers: dict, inputs: dict) -> dict:
+        with logged_client(timeout=120) as client:
+            resp = client.post(endpoint, headers=headers, json=inputs)
+            logger.info("[nano_banana] Submit response: status=%d, body_size=%d",
+                        resp.status_code, len(resp.content))
+            resp.raise_for_status()
+            data = resp.json()
+        if not data.get("candidates"):
+            raise RuntimeError("No candidates in response")
+        return data
+
     def execute(self, param: TaskParam) -> TaskResult:
         endpoint = f"{settings.gemini_base_url}/models/{settings.gemini_nano_banana_model}:generateContent"
         headers = {
@@ -24,24 +36,17 @@ class NanoBananaTask(BaseTask):
         logger.info("[nano_banana] Calling Gemini API: POST %s", endpoint)
         t0 = time.time()
         try:
-            with logged_client(timeout=120) as client:
-                resp = client.post(endpoint, headers=headers, json=param.inputs)
-                elapsed = time.time() - t0
-                logger.info("[nano_banana] Response: status=%d, elapsed=%.2fs, body_size=%d",
-                            resp.status_code, elapsed, len(resp.content))
-                resp.raise_for_status()
-                data = resp.json()
+            data = retry_on_failure(
+                lambda: self._submit(endpoint, headers, param.inputs),
+                task_name="nano_banana",
+            )
         except Exception as e:
             elapsed = time.time() - t0
             logger.error("[nano_banana] Request failed in %.2fs: %s", elapsed, e)
             return TaskResult.failure(f"submit execute fail: {e}")
 
-        candidates = data.get("candidates")
-        if not candidates:
-            logger.warning("[nano_banana] No candidates in response")
-            return TaskResult.failure("submit execute fail !")
-
         encoded_files: list[str] = []
+        candidates = data.get("candidates", [])
         parts = candidates[0].get("content", {}).get("parts", [])
         for part in parts:
             text = part.get("text")

app/service/task/impl/seedance_video_task.py

@@ -6,6 +6,7 @@ from app.service.task.base_task import BaseTask
 from app.service.task.task_param import TaskParam
 from app.service.task.task_result import TaskResult
 from app.util.http_util import logged_client
+from app.util.retry_util import retry_on_failure
 
 logger = logging.getLogger(__name__)
 
@@ -20,21 +21,28 @@ class SeedanceVideoTask(BaseTask):
             "Authorization": f"Bearer {settings.seed_api_key}",
         }
 
+    def _submit(self, url: str, inputs: dict) -> dict:
+        with logged_client(timeout=60) as client:
+            resp = client.post(url, headers=self._headers(), json=inputs)
+            logger.info("[seedance_video] Submit response: status=%d", resp.status_code)
+            resp.raise_for_status()
+            data = resp.json()
+        task_id = data.get("id")
+        if not task_id:
+            raise RuntimeError(f"No task ID in response: {data}")
+        return data
+
     def execute(self, param: TaskParam) -> TaskResult:
         execute_url = settings.seedance_video_url
         t0 = time.time()
 
         logger.info("[seedance_video] Submitting task: POST %s", execute_url)
         try:
-            with logged_client(timeout=60) as client:
-                resp = client.post(execute_url, headers=self._headers(), json=param.inputs)
-                logger.info("[seedance_video] Submit response: status=%d, elapsed=%.2fs",
-                            resp.status_code, time.time() - t0)
-                data = resp.json()
+            data = retry_on_failure(
+                lambda: self._submit(execute_url, param.inputs),
+                task_name="seedance_video",
+            )
             task_id = data.get("id")
-            if not task_id:
-                logger.error("[seedance_video] No task ID in response: %s", data)
-                return TaskResult.failure("Failed to submit task: no task ID received")
             logger.info("[seedance_video] Task submitted, remote_id=%s", task_id)
         except Exception as e:
             logger.error("[seedance_video] Submit failed: %s", e)

app/service/task/impl/seedream_image_task.py

@@ -6,6 +6,7 @@ from app.service.task.base_task import BaseTask
 from app.service.task.task_param import TaskParam
 from app.service.task.task_result import TaskResult
 from app.util.http_util import logged_client
+from app.util.retry_util import retry_on_failure
 
 logger = logging.getLogger(__name__)
 
@@ -20,25 +21,29 @@ class SeedreamImageTask(BaseTask):
             "Authorization": f"Bearer {settings.seed_api_key}",
         }
 
+    def _submit(self, url: str, inputs: dict) -> dict:
+        with logged_client(timeout=60) as client:
+            resp = client.post(url, headers=self._headers(), json=inputs)
+            logger.info("[seedream_image] Submit response: status=%d", resp.status_code)
+            resp.raise_for_status()
+            data = resp.json()
+        if "error" in data:
+            err = data["error"]
+            code = err.get("code", "")
+            msg = err.get("message", "")
+            raise RuntimeError(f"API error [{code}]: {msg}")
+        return data
+
     def execute(self, param: TaskParam) -> TaskResult:
         url = settings.seedream_image_url
         logger.info("[seedream_image] Calling API: POST %s", url)
         t0 = time.time()
 
         try:
-            with logged_client(timeout=60) as client:
-                resp = client.post(url, headers=self._headers(), json=param.inputs)
-                elapsed = time.time() - t0
-                logger.info("[seedream_image] Response: status=%d, elapsed=%.2fs",
-                            resp.status_code, elapsed)
-                data = resp.json()
-
-            if "error" in data:
-                err = data["error"]
-                code = err.get("code", "")
-                msg = err.get("message", "")
-                logger.error("[seedream_image] API error: code=%s, message=%s", code, msg)
-                return TaskResult.failure(f"API error [{code}]: {msg}")
+            data = retry_on_failure(
+                lambda: self._submit(url, param.inputs),
+                task_name="seedream_image",
+            )
 
             images: list[str] = []
             for item in data.get("data", []):

app/util/retry_util.py

@@ -0,0 +1,39 @@
+import logging
+import time
+from typing import Callable, TypeVar
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+def retry_on_failure(
+    fn: Callable[[], T],
+    *,
+    max_retries: int = 3,
+    backoff_base: float = 2.0,
+    task_name: str = "",
+) -> T:
+    """在失败时使用指数退避进行重试，仅用于任务提交阶段。
+
+    fn 应当在提交不成功时抛出异常，所有重试耗尽后将抛出最后一次的异常。
+    """
+    last_exc: Exception | None = None
+    for attempt in range(1, max_retries + 1):
+        try:
+            return fn()
+        except Exception as e:
+            last_exc = e
+            if attempt < max_retries:
+                wait = backoff_base ** attempt
+                logger.warning(
+                    "[%s] Submit attempt %d/%d failed: %s, retrying in %.1fs …",
+                    task_name, attempt, max_retries, e, wait,
+                )
+                time.sleep(wait)
+            else:
+                logger.error(
+                    "[%s] All %d submit attempts exhausted, last error: %s",
+                    task_name, max_retries, e,
+                )
+    raise last_exc  # type: ignore[misc]