feat: support aggregated image model protocols

services/api/app/runtime_settings.py

@@ -61,6 +61,8 @@ class RuntimeSettingsTestResult(BaseModel):
 
 ModelCategory = Literal["language", "multimodal"]
 ModelCapability = Literal["orchestration", "spatial_understanding", "image_generation", "image_editing"]
+ImageProtocol = Literal["openai_images", "aigc_media"]
+ImageParameterProfile = Literal["generic", "gpt_image_2", "nano_banana_pro", "seedream_5_pro"]
 
 
 class ModelPoolItem(BaseModel):
@@ -76,6 +78,9 @@ class ModelPoolItem(BaseModel):
     chat_path: str = "/chat/completions"
     image_generation_path: str = "/images/generations"
     image_edit_path: str = "/images/edits"
+    image_status_path: str = "/v1/media/status"
+    image_protocol: ImageProtocol = "openai_images"
+    image_parameter_profile: ImageParameterProfile = "generic"
     enabled: bool = True
 
 
@@ -233,6 +238,37 @@ class EncryptedSettingsStore:
         document["tests"].pop("ai_models", None)
         return True
 
+    @staticmethod
+    def _migrate_model_protocol_fields(document: dict[str, Any]) -> bool:
+        changed = False
+        known_profiles = {
+            "gpt-image-2": "gpt_image_2",
+            "gemini-3-pro-image-preview": "nano_banana_pro",
+            "doubao-seedream-5-0-pro-260628": "seedream_5_pro",
+        }
+        for item in document.get("values", {}).get("model_pool", []):
+            if not isinstance(item, dict):
+                continue
+            defaults = {
+                "image_status_path": "/v1/media/status",
+                "image_protocol": "openai_images",
+                "image_parameter_profile": "generic",
+            }
+            host = urlparse(str(item.get("base_url", ""))).hostname or ""
+            profile = known_profiles.get(str(item.get("model_id", "")))
+            if host == "api.lk888.ai" and profile and "image_generation" in item.get("capabilities", []):
+                defaults.update(
+                    image_protocol="aigc_media",
+                    image_parameter_profile=profile,
+                    image_generation_path="/v1/media/generate",
+                    image_status_path="/v1/media/status",
+                )
+            for key, value in defaults.items():
+                if key not in item:
+                    item[key] = value
+                    changed = True
+        return changed
+
     def load_document(self) -> dict[str, Any]:
         if not self.data_path.exists():
             return {"values": self.defaults(), "tests": {}}
@@ -243,7 +279,9 @@ class EncryptedSettingsStore:
             raise RuntimeError("运行期配置无法解密，请检查主密钥是否发生变化。") from exc
         document.setdefault("values", {})
         document.setdefault("tests", {})
-        if self._migrate_legacy_models(document):
+        migrated = self._migrate_legacy_models(document)
+        migrated = self._migrate_model_protocol_fields(document) or migrated
+        if migrated:
             self.save_document(document)
         return document
 
@@ -718,14 +756,116 @@ async def test_model_pool_item(values: dict[str, Any], model_id: str) -> Runtime
             message=f"模型“{name}”测试失败：账号模型列表中未找到 ID“{item['model_id']}”。",
             details={"model_id": item["model_id"], "model_count": len(listed_ids)},
         )
+    protocol_error = _validate_image_protocol(item)
+    if protocol_error:
+        return RuntimeSettingsTestResult(
+            target=target,
+            ok=False,
+            message=f"模型“{name}”的 ID 可用，但生图协议配置不正确：{protocol_error}",
+            details={"model_id": item["model_id"], "model_count": len(listed_ids)},
+        )
     return RuntimeSettingsTestResult(
         target=target,
         ok=True,
-        message=f"模型“{name}”连接正常，模型 ID“{item['model_id']}”可用。",
+        message=f"模型“{name}”连接正常，模型 ID“{item['model_id']}”可用。此项不实际生图，不消耗额度。",
         details={"model_id": item["model_id"], "model_count": len(listed_ids)},
     )
 
 
+def _validate_image_protocol(item: dict[str, Any]) -> str:
+    if "image_generation" not in item.get("capabilities", []):
+        return ""
+    host = urlparse(str(item.get("base_url", ""))).hostname or ""
+    known_aigc_models = {
+        "gpt-image-2": "gpt_image_2",
+        "gemini-3-pro-image-preview": "nano_banana_pro",
+        "doubao-seedream-5-0-pro-260628": "seedream_5_pro",
+    }
+    expected_profile = known_aigc_models.get(str(item.get("model_id", "")))
+    if host == "api.lk888.ai" and expected_profile:
+        if item.get("image_protocol", "openai_images") != "aigc_media":
+            return "聚合引擎上的该模型应选择“AIGC 异步媒体协议”"
+        generation_path = urlparse(_join_api_url(str(item["base_url"]), str(item.get("image_generation_path", "")))).path
+        status_path = urlparse(_join_api_url(str(item["base_url"]), str(item.get("image_status_path", "")))).path
+        if generation_path != "/v1/media/generate":
+            return "生图地址应解析为 /v1/media/generate"
+        if status_path != "/v1/media/status":
+            return "任务查询地址应解析为 /v1/media/status"
+        if item.get("image_parameter_profile", "generic") != expected_profile:
+            return "参数预设与模型不匹配，请重新选择对应的模型预设"
+    return ""
+
+
+async def test_image_generation(values: dict[str, Any], model_id: str) -> RuntimeSettingsTestResult:
+    """Run one real, billable image request to prove the configured protocol end to end."""
+    target = f"generation:{model_id}"
+    item = next(
+        (candidate for candidate in values.get("model_pool", []) if candidate.get("id") == model_id),
+        None,
+    )
+    if item is None:
+        return RuntimeSettingsTestResult(target=target, ok=False, message="模型不存在或已被删除。")
+    if "image_generation" not in item.get("capabilities", []):
+        return RuntimeSettingsTestResult(target=target, ok=False, message="该模型没有启用图像生成能力。")
+    protocol_error = _validate_image_protocol(item)
+    if protocol_error:
+        return RuntimeSettingsTestResult(target=target, ok=False, message=f"无法试生成：{protocol_error}")
+
+    profile = item.get("image_parameter_profile", "generic")
+    trial_options: dict[str, Any] = {}
+    if profile == "gpt_image_2":
+        trial_options = {"size": "1024x1024", "quality": "low"}
+    elif profile == "nano_banana_pro":
+        trial_options = {"aspectRatio": "1:1", "imageSize": "1K"}
+    elif profile == "seedream_5_pro":
+        trial_options = {"aspect_ratio": "1:1", "size": "1K"}
+
+    try:
+        from app.integrations.openai_compatible import OpenAICompatibleGateway
+
+        result = await OpenAICompatibleGateway(values).generate_image(
+            "极简室内材质测试图：一个米白色立方体放在浅灰背景中，无文字",
+            model_instance_id=model_id,
+            **trial_options,
+        )
+        output = _find_generated_image(result)
+        if not output:
+            return RuntimeSettingsTestResult(
+                target=target,
+                ok=False,
+                message="请求已结束，但响应中没有找到图片地址或图片数据。",
+            )
+        details = {"model_id": item["model_id"], "output_kind": output[0]}
+        if output[0] == "url":
+            details["result_url"] = output[1]
+        return RuntimeSettingsTestResult(
+            target=target,
+            ok=True,
+            message=f"模型“{item.get('name', item['model_id'])}”已完成一次真实生图，端到端配置可用。",
+            details=details,
+        )
+    except (httpx.HTTPError, RuntimeError, ValueError) as exc:
+        return RuntimeSettingsTestResult(target=target, ok=False, message=f"真实生图失败：{exc}")
+
+
+def _find_generated_image(payload: dict[str, Any]) -> tuple[str, str] | None:
+    if payload.get("result_url"):
+        return "url", str(payload["result_url"])
+    for row in payload.get("data", []) if isinstance(payload.get("data"), list) else []:
+        if not isinstance(row, dict):
+            continue
+        if row.get("url"):
+            return "url", str(row["url"])
+        if row.get("b64_json"):
+            return "inline", ""
+    for candidate in payload.get("candidates", []) if isinstance(payload.get("candidates"), list) else []:
+        parts = candidate.get("content", {}).get("parts", []) if isinstance(candidate, dict) else []
+        for part in parts:
+            if isinstance(part, dict) and (part.get("inlineData", {}).get("data") or part.get("inline_data", {}).get("data")):
+                return "inline", ""
+    return None
+
+
 async def _test_gpu(values: dict[str, Any]) -> RuntimeSettingsTestResult:
     mode = values.get("gpu_mode", "disabled")
     if mode == "disabled":