错误的时候添加了更多的信息

README.md

@@ -483,47 +483,6 @@ for message in pubsub.listen():
 | base64_encode | Base64 编码 |
 | base64_decode | Base64 解码 |
 
-## 任务提交重试机制
-
-所有调用外部 API Key 的任务在**提交阶段**内置了自动重试，采用指数退避策略。轮询（polling）阶段不受影响。
-
-### 涉及的任务
-
-| 任务 | API Key | 最大重试次数 | 退避基数 |
-|------|---------|:----------:|:-------:|
-| kontext_pro | `kontext_api_key` | 3 | 2s |
-| kontext_max | `kontext_api_key` | 3 | 2s |
-| seedream_image | `seed_api_key` | 3 | 2s |
-| seedance_video | `seed_api_key` | 3 | 2s |
-| nano_banana | `gemini_api_key` | 3 | 2s |
-| google_veo3 | `gemini_api_key` | 3 | 2s |
-| keling_image2video | `keling_app_key/secret` | 5 | 3s |
-| keling_video2effect | `keling_app_key/secret` | 5 | 3s |
-
-> 不涉及外部 API Key 的本地任务（smart_beauty、image_editor、video_editor、file_download、file_upload、base64_encode、base64_decode）不启用提交重试。
-
-### 重试触发条件
-
-- HTTP 请求异常（网络错误、超时等）
-- HTTP 状态码非 2xx
-- API 返回业务级错误（如空响应、缺少任务 ID、错误码等）
-
-### 退避策略
-
-等待时间 = `backoff_base ^ attempt`（秒），例如退避基数为 2s 时：
-
-| 重试次数 | 等待时间 |
-|:--------:|:-------:|
-| 第 1 次 | 2s |
-| 第 2 次 | 4s |
-| 第 3 次 | 8s |
-
-可灵系列任务因存在频率限制（错误码 1302/1303），配置了更高的重试次数（5 次）和更长的退避基数（3s）。
-
-### 实现
-
-重试逻辑由 `app/util/retry_util.py` 中的 `retry_on_failure` 函数提供，各任务将提交逻辑封装在 `_submit()` 方法中，提交不成功时抛出异常以触发重试。
-
 ## 配置说明
 
 所有配置通过环境变量注入，前缀为 `AI_FLOW_`。详见 `.env.example`。

app/service/task/impl/google_veo3_task.py

@@ -8,7 +8,6 @@ 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__)
 
@@ -26,10 +25,7 @@ class GoogleVeo3Task(BaseTask):
     def execute(self, param: TaskParam) -> TaskResult:
         t0 = time.time()
         try:
-            op_name = retry_on_failure(
-                lambda: self._create_request(param.inputs),
-                task_name="google_veo3",
-            )
+            op_name = self._create_request(param.inputs)
             logger.info("[google_veo3] Operation created: %s", op_name)
 
             video_uri = self._poll_until_done(op_name)
@@ -55,8 +51,7 @@ class GoogleVeo3Task(BaseTask):
         with logged_client(timeout=120) as client:
             resp = client.post(endpoint, headers=self._headers(), json=inputs)
             logger.info("[google_veo3] Submit response: status=%d", resp.status_code)
-            if not resp.is_success:
-                raise RuntimeError(f"HTTP {resp.status_code}: {resp.text}")
+            resp.raise_for_status()
             data = resp.json()
         op_name = data.get("name")
         if not op_name:

app/service/task/impl/keling_image2video_task.py

@@ -8,7 +8,6 @@ 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__)
 
@@ -40,35 +39,34 @@ class KelingImage2VideoTask(BaseTask):
             "Authorization": f"Bearer {self._generate_jwt()}",
         }
 
-    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)
-            if not resp.is_success:
-                raise RuntimeError(f"HTTP {resp.status_code}: {resp.text}")
-            bean = resp.json()
+    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}")
+
         code = bean.get("code", 0)
-        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}")
+        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)
         return bean
 
     def execute(self, param: TaskParam) -> TaskResult:
         t0 = time.time()
-        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}")
+        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)
 
         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,7 +6,6 @@ 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__)
 
@@ -25,30 +24,25 @@ 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)
-            if not resp.is_success:
-                raise RuntimeError(f"HTTP {resp.status_code}: {resp.text}")
-            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:
-            bean = retry_on_failure(
-                lambda: self._submit(param.inputs),
-                task_name=self.get_type(),
-            )
+            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()
         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,7 +6,6 @@ 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__)
 
@@ -15,18 +14,6 @@ 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))
-            if not resp.is_success:
-                raise RuntimeError(f"HTTP {resp.status_code}: {resp.text}")
-            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 = {
@@ -37,17 +24,24 @@ class NanoBananaTask(BaseTask):
         logger.info("[nano_banana] Calling Gemini API: POST %s", endpoint)
         t0 = time.time()
         try:
-            data = retry_on_failure(
-                lambda: self._submit(endpoint, headers, param.inputs),
-                task_name="nano_banana",
-            )
+            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()
         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,7 +6,6 @@ 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__)
 
@@ -21,29 +20,21 @@ 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)
-            if not resp.is_success:
-                raise RuntimeError(f"HTTP {resp.status_code}: {resp.text}")
-            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:
-            data = retry_on_failure(
-                lambda: self._submit(execute_url, param.inputs),
-                task_name="seedance_video",
-            )
+            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()
             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,7 +6,6 @@ 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__)
 
@@ -21,30 +20,25 @@ 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)
-            if not resp.is_success:
-                raise RuntimeError(f"HTTP {resp.status_code}: {resp.text}")
-            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:
-            data = retry_on_failure(
-                lambda: self._submit(url, param.inputs),
-                task_name="seedream_image",
-            )
+            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}")
 
             images: list[str] = []
             for item in data.get("data", []):

app/util/retry_util.py

@@ -1,39 +0,0 @@
-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]