feat: add configurable AI model pool

services/api/app/api/routes.py

@@ -31,7 +31,8 @@ def health(
     runtime_store: EncryptedSettingsStore = Depends(get_runtime_store),
 ) -> dict:
     values = runtime_store.merged_values()
-    router_state = ModelRouter(values)
+    runtime_response = runtime_store.public_response()
+    router_state = ModelRouter(values, runtime_response.tests)
     return {
         "status": "ok",
         "environment": settings.app_env,

services/api/app/api/settings.py

@@ -12,6 +12,7 @@ from app.runtime_settings import (
     SettingsReadiness,
     generate_secret,
     get_runtime_store,
+    test_model_pool_item,
     test_runtime_settings,
 )
 
@@ -24,7 +25,7 @@ def read_settings(
 ) -> RuntimeSettingsResponse:
     try:
         return store.public_response()
-    except RuntimeError as exc:
+    except (RuntimeError, ValueError) as exc:
         raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
 
 
@@ -52,7 +53,7 @@ async def test_settings(
         if not request.values:
             store.record_test(result, values)
         return result
-    except (KeyError, RuntimeError) as exc:
+    except (KeyError, RuntimeError, ValueError) as exc:
         return RuntimeSettingsTestResult(
             target=request.target,
             ok=False,
@@ -60,6 +61,24 @@ async def test_settings(
         )
 
 
+@router.post("/settings/models/{model_id}/test", response_model=RuntimeSettingsTestResult)
+async def test_model(
+    model_id: str,
+    store: EncryptedSettingsStore = Depends(get_runtime_store),
+) -> RuntimeSettingsTestResult:
+    try:
+        values = store.merged_values()
+        result = await test_model_pool_item(values, model_id)
+        store.record_test(result, values)
+        return result
+    except (RuntimeError, ValueError) as exc:
+        return RuntimeSettingsTestResult(
+            target=f"model:{model_id}",
+            ok=False,
+            message=f"模型测试失败：{exc}",
+        )
+
+
 @router.post("/settings/generate-secret", response_model=SecretGenerateResponse)
 def create_secret(request: SecretGenerateRequest) -> SecretGenerateResponse:
     return SecretGenerateResponse(value=generate_secret(request.kind))

services/api/app/integrations/model_router.py

@@ -1,43 +1,122 @@
-from dataclasses import dataclass
 from collections.abc import Mapping
+from dataclasses import dataclass
 from typing import Any
 
 from app.config import Settings
 
 
-def _configured(value: str) -> bool:
-    return bool(value) and not value.startswith("replace_with_")
-
-
 @dataclass(frozen=True)
 class ProviderCapability:
     provider: str
     configured: bool
     strengths: tuple[str, ...]
+    model_id: str = ""
+    name: str = ""
 
 
 class ModelRouter:
-    def __init__(self, settings: Settings | Mapping[str, Any]) -> None:
+    def __init__(
+        self,
+        settings: Settings | Mapping[str, Any],
+        tests: Mapping[str, Mapping[str, Any]] | None = None,
+    ) -> None:
         self.settings = settings
+        self.tests = tests
 
-    def _value(self, key: str, default: str = "") -> str:
+    def _value(self, key: str, default: Any = "") -> Any:
         if isinstance(self.settings, Mapping):
-            return str(self.settings.get(key, default) or "")
-        return str(getattr(self.settings, key, default) or "")
+            return self.settings.get(key, default)
+        return getattr(self.settings, key, default)
 
-    def capabilities(self) -> list[ProviderCapability]:
+    @property
+    def pool(self) -> list[dict[str, Any]]:
+        raw_pool = self._value("model_pool", [])
+        return [item for item in raw_pool if isinstance(item, dict) and item.get("enabled", True)]
+
+    def _tested(self, item: Mapping[str, Any]) -> bool:
+        if self.tests is None:
+            return True
+        return bool(self.tests.get(f"model:{item.get('id')}", {}).get("ok"))
+
+    def candidates(self, capability: str) -> list[dict[str, Any]]:
         return [
-            ProviderCapability(
-                provider=self._value("ai_provider", "custom"),
-                configured=_configured(self._value("ai_api_key"))
-                and bool(self._value("ai_base_url"))
-                and bool(self._value("image_model")),
-                strengths=("统一模型网关", "空间理解", "图像生成与编辑"),
-            ),
+            item
+            for item in self.pool
+            if capability in item.get("capabilities", []) and self._tested(item)
         ]
 
+    def orchestrator(self) -> dict[str, Any]:
+        selected_id = str(self._value("orchestrator_model_id", ""))
+        selected = next(
+            (
+                item
+                for item in self.pool
+                if item.get("id") == selected_id
+                and item.get("category") == "language"
+                and self._tested(item)
+            ),
+            None,
+        )
+        if selected:
+            return selected
+        raise RuntimeError("总调度模型未配置或尚未通过测试。")
+
+    def choose(self, route: str, orchestrator_choice_id: str | None = None) -> dict[str, Any]:
+        if route == "spatial":
+            mode_key, selected_key, capability = (
+                "spatial_routing_mode",
+                "spatial_model_id",
+                "spatial_understanding",
+            )
+        elif route == "image":
+            mode_key, selected_key, capability = (
+                "image_routing_mode",
+                "image_model_id",
+                "image_generation",
+            )
+        else:
+            raise ValueError(f"未知模型路由：{route}")
+
+        candidates = self.candidates(capability)
+        if self._value(mode_key, "auto") == "manual":
+            selected_id = str(self._value(selected_key, ""))
+            selected = next((item for item in candidates if item.get("id") == selected_id), None)
+            if selected:
+                return selected
+            raise RuntimeError(f"手动选择的{route}模型不可用或尚未通过测试。")
+        if orchestrator_choice_id:
+            selected = next((item for item in candidates if item.get("id") == orchestrator_choice_id), None)
+            if selected:
+                return selected
+            raise RuntimeError("总调度返回的模型不在已测试候选列表中。")
+        if len(candidates) == 1:
+            return candidates[0]
+        if len(candidates) > 1:
+            raise RuntimeError("自动路由存在多个候选，需要总调度返回模型实例 ID。")
+        raise RuntimeError(f"没有通过测试且支持{capability}的模型。")
+
+    def capabilities(self) -> list[ProviderCapability]:
+        result: list[ProviderCapability] = []
+        for item in self.pool:
+            strengths: list[str] = []
+            if "orchestration" in item.get("capabilities", []):
+                strengths.append("总调度")
+            if "spatial_understanding" in item.get("capabilities", []):
+                strengths.append("空间理解")
+            if "image_generation" in item.get("capabilities", []):
+                strengths.append("图像生成")
+            if "image_editing" in item.get("capabilities", []):
+                strengths.append("图像编辑")
+            result.append(
+                ProviderCapability(
+                    provider=str(item.get("provider", "custom")),
+                    configured=self._tested(item),
+                    strengths=tuple(strengths),
+                    model_id=str(item.get("model_id", "")),
+                    name=str(item.get("name", "")),
+                )
+            )
+        return result
+
     def choose_image_provider(self, task: str) -> str:
-        configured = [item.provider for item in self.capabilities() if item.configured]
-        if configured:
-            return configured[0]
-        raise RuntimeError("No image provider is configured.")
+        return str(self.choose("image").get("provider", "custom"))

services/api/app/integrations/openai_compatible.py

@@ -4,48 +4,70 @@ from urllib.parse import urljoin
 
 import httpx
 
+from app.integrations.model_router import ModelRouter
+
 
 class OpenAICompatibleGateway:
-    """One adapter for OpenAI, lk666.ai and other OpenAI-compatible gateways."""
+    """Adapter for independently configured OpenAI-compatible model-pool items."""
 
     def __init__(self, values: Mapping[str, Any]) -> None:
         self.values = values
+        self.router = ModelRouter(values)
 
-    def _url(self, path_key: str, fallback: str) -> str:
-        base_url = str(self.values.get("ai_base_url", "")).rstrip("/") + "/"
-        path = str(self.values.get(path_key, fallback)).lstrip("/")
+    @staticmethod
+    def _url(item: Mapping[str, Any], path_key: str, fallback: str) -> str:
+        base_url = str(item.get("base_url", "")).rstrip("/") + "/"
+        path = str(item.get(path_key, fallback)).lstrip("/")
         return urljoin(base_url, path)
 
-    @property
-    def headers(self) -> dict[str, str]:
-        return {
-            "Authorization": f"Bearer {self.values.get('ai_api_key', '')}",
-            "Content-Type": "application/json",
-        }
+    @staticmethod
+    def _headers(item: Mapping[str, Any], *, json_content: bool = True) -> dict[str, str]:
+        headers = {"Authorization": f"Bearer {item.get('api_key', '')}"}
+        if json_content:
+            headers["Content-Type"] = "application/json"
+        return headers
 
-    async def list_models(self) -> list[str]:
+    async def list_models(self, model: Mapping[str, Any] | None = None) -> list[str]:
+        item = model or self.router.orchestrator()
         async with httpx.AsyncClient(timeout=15) as client:
-            response = await client.get(self._url("ai_models_path", "/models"), headers=self.headers)
+            response = await client.get(
+                self._url(item, "models_path", "/models"),
+                headers=self._headers(item),
+            )
             response.raise_for_status()
             payload = response.json()
-        return [item["id"] for item in payload.get("data", []) if isinstance(item, dict) and item.get("id")]
+        return [candidate["id"] for candidate in payload.get("data", []) if isinstance(candidate, dict) and candidate.get("id")]
 
-    async def chat(self, messages: list[dict[str, Any]], *, vision: bool = False) -> dict[str, Any]:
-        model_key = "vision_model" if vision else "orchestrator_model"
+    async def chat(
+        self,
+        messages: list[dict[str, Any]],
+        *,
+        vision: bool = False,
+        model_instance_id: str | None = None,
+    ) -> dict[str, Any]:
+        item = self.router.choose("spatial", model_instance_id) if vision else self.router.orchestrator()
         async with httpx.AsyncClient(timeout=120) as client:
             response = await client.post(
-                self._url("ai_chat_path", "/chat/completions"),
-                headers=self.headers,
-                json={"model": self.values[model_key], "messages": messages},
+                self._url(item, "chat_path", "/chat/completions"),
+                headers=self._headers(item),
+                json={"model": item["model_id"], "messages": messages},
             )
             response.raise_for_status()
             return response.json()
-    async def generate_image(self, prompt: str, **options: Any) -> dict[str, Any]:
-        payload = {"model": self.values["image_model"], "prompt": prompt, **options}
+
+    async def generate_image(
+        self,
+        prompt: str,
+        *,
+        model_instance_id: str | None = None,
+        **options: Any,
+    ) -> dict[str, Any]:
+        item = self.router.choose("image", model_instance_id)
+        payload = {"model": item["model_id"], "prompt": prompt, **options}
         async with httpx.AsyncClient(timeout=180) as client:
             response = await client.post(
-                self._url("ai_image_generation_path", "/images/generations"),
-                headers=self.headers,
+                self._url(item, "image_generation_path", "/images/generations"),
+                headers=self._headers(item),
                 json=payload,
             )
             response.raise_for_status()
@@ -58,19 +80,18 @@ class OpenAICompatibleGateway:
         *,
         filename: str = "image.png",
         mask: bytes | None = None,
+        model_instance_id: str | None = None,
         **options: Any,
     ) -> dict[str, Any]:
-        files: dict[str, tuple[str, bytes, str]] = {
-            "image": (filename, image, "image/png"),
-        }
+        item = self.router.choose("image", model_instance_id)
+        files: dict[str, tuple[str, bytes, str]] = {"image": (filename, image, "image/png")}
         if mask is not None:
             files["mask"] = ("mask.png", mask, "image/png")
-        data = {"model": self.values["image_model"], "prompt": prompt, **options}
-        headers = {"Authorization": self.headers["Authorization"]}
+        data = {"model": item["model_id"], "prompt": prompt, **options}
         async with httpx.AsyncClient(timeout=180) as client:
             response = await client.post(
-                self._url("ai_image_edit_path", "/images/edits"),
-                headers=headers,
+                self._url(item, "image_edit_path", "/images/edits"),
+                headers=self._headers(item, json_content=False),
                 data=data,
                 files=files,
             )

services/api/app/runtime_settings.py

@@ -45,7 +45,6 @@ class RuntimeSettingsTestRequest(BaseModel):
         "infrastructure",
         "storage",
         "baidu_ocr",
-        "ai_models",
         "gpu",
         "langfuse",
         "sentry",
@@ -60,6 +59,26 @@ class RuntimeSettingsTestResult(BaseModel):
     details: dict[str, Any] = Field(default_factory=dict)
 
 
+ModelCategory = Literal["language", "multimodal"]
+ModelCapability = Literal["orchestration", "spatial_understanding", "image_generation", "image_editing"]
+
+
+class ModelPoolItem(BaseModel):
+    id: str = Field(min_length=1, max_length=120)
+    name: str = Field(min_length=1, max_length=120)
+    model_id: str = Field(min_length=1, max_length=200)
+    category: ModelCategory
+    provider: str = Field(default="custom", min_length=1, max_length=80)
+    base_url: str = Field(min_length=1, max_length=500)
+    api_key: str = ""
+    capabilities: list[ModelCapability] = Field(default_factory=list)
+    models_path: str = "/models"
+    chat_path: str = "/chat/completions"
+    image_generation_path: str = "/images/generations"
+    image_edit_path: str = "/images/edits"
+    enabled: bool = True
+
+
 class SecretGenerateRequest(BaseModel):
     kind: Literal["hex24", "hex32", "base64_32"]
 
@@ -79,31 +98,28 @@ REQUIRED_GROUPS: dict[str, list[str]] = {
         "s3_bucket_derived",
         "s3_bucket_renders",
     ],
-    "AI 模型": [
-        "ai_provider",
-        "ai_base_url",
-        "ai_api_key",
-        "orchestrator_model",
-        "vision_model",
-        "image_model",
-    ],
 }
 
 REQUIRED_TESTS = {
     "基础运行": "infrastructure",
     "文件存储": "storage",
-    "AI 模型": "ai_models",
 }
 
+MODEL_SETTINGS_DEFAULTS: dict[str, Any] = {
+    "model_pool": [],
+    "orchestrator_model_id": "",
+    "spatial_routing_mode": "auto",
+    "spatial_model_id": "",
+    "image_routing_mode": "auto",
+    "image_model_id": "",
+}
+MODEL_SETTINGS_KEYS = set(MODEL_SETTINGS_DEFAULTS)
+LEGACY_SECRET_FIELDS = {"ai_api_key"}
+
 TEST_FIELDS: dict[str, set[str]] = {
     "infrastructure": set(),
     "storage": {key for key in FIELDS if key.startswith("s3_")},
     "baidu_ocr": {key for key in FIELDS if key.startswith("baidu_ocr_")},
-    "ai_models": {key for key in FIELDS if key.startswith("ai_")} | {
-        "orchestrator_model",
-        "vision_model",
-        "image_model",
-    },
     "gpu": {key for key in FIELDS if key.startswith("gpu_")},
     "langfuse": {key for key in FIELDS if key.startswith("langfuse_")},
     "sentry": {key for key in FIELDS if key.startswith("sentry_")},
@@ -154,11 +170,68 @@ class EncryptedSettingsStore:
         return Fernet(key)
 
     def defaults(self) -> dict[str, Any]:
-        return {
+        field_defaults = {
             key: field.default
             for key, field in FIELDS.items()
             if key in EDITABLE_FIELDS and field.default is not None
         }
+        return {**field_defaults, **MODEL_SETTINGS_DEFAULTS}
+
+    def _migrate_legacy_models(self, document: dict[str, Any]) -> bool:
+        values = document["values"]
+        if values.get("model_pool") or not values.get("ai_base_url"):
+            return False
+        common = {
+            "provider": values.get("ai_provider", "custom"),
+            "base_url": values.get("ai_base_url", ""),
+            "api_key": values.get("ai_api_key", ""),
+            "models_path": values.get("ai_models_path", "/models"),
+            "chat_path": values.get("ai_chat_path", "/chat/completions"),
+            "image_generation_path": values.get("ai_image_generation_path", "/images/generations"),
+            "image_edit_path": values.get("ai_image_edit_path", "/images/edits"),
+            "enabled": True,
+        }
+        pool: list[dict[str, Any]] = []
+        legacy_roles = [
+            ("legacy-orchestrator", "总调度模型", values.get("orchestrator_model"), "language", ["orchestration"]),
+            ("legacy-spatial", "空间理解模型", values.get("vision_model"), "multimodal", ["spatial_understanding"]),
+            ("legacy-image", "生图模型", values.get("image_model"), "multimodal", ["image_generation", "image_editing"]),
+        ]
+        for item_id, name, model_id, category, capabilities in legacy_roles:
+            if not model_id:
+                continue
+            pool.append({
+                "id": item_id,
+                "name": name,
+                "model_id": model_id,
+                "category": category,
+                "capabilities": capabilities,
+                **common,
+            })
+        if not pool:
+            return False
+        values.update(MODEL_SETTINGS_DEFAULTS)
+        values["model_pool"] = pool
+        values["orchestrator_model_id"] = "legacy-orchestrator" if values.get("orchestrator_model") else ""
+        values["spatial_routing_mode"] = "manual"
+        values["spatial_model_id"] = "legacy-spatial" if values.get("vision_model") else ""
+        values["image_routing_mode"] = "manual"
+        values["image_model_id"] = "legacy-image" if values.get("image_model") else ""
+        for legacy_key in (
+            "ai_provider",
+            "ai_base_url",
+            "ai_api_key",
+            "orchestrator_model",
+            "vision_model",
+            "image_model",
+            "ai_models_path",
+            "ai_chat_path",
+            "ai_image_generation_path",
+            "ai_image_edit_path",
+        ):
+            values.pop(legacy_key, None)
+        document["tests"].pop("ai_models", None)
+        return True
 
     def load_document(self) -> dict[str, Any]:
         if not self.data_path.exists():
@@ -170,6 +243,8 @@ class EncryptedSettingsStore:
             raise RuntimeError("运行期配置无法解密，请检查主密钥是否发生变化。") from exc
         document.setdefault("values", {})
         document.setdefault("tests", {})
+        if self._migrate_legacy_models(document):
+            self.save_document(document)
         return document
 
     def save_document(self, document: dict[str, Any]) -> None:
@@ -184,6 +259,12 @@ class EncryptedSettingsStore:
         document = self.load_document()
         values = {**self.defaults(), **document["values"]}
         for key, value in (pending or {}).items():
+            if key == "model_pool":
+                values[key] = self._coerce_model_pool(value, values.get("model_pool", []))
+                continue
+            if key in MODEL_SETTINGS_KEYS:
+                values[key] = value.strip() if isinstance(value, str) else value
+                continue
             if key not in EDITABLE_FIELDS:
                 continue
             if key in SECRET_FIELDS and not _is_configured(value):
@@ -195,6 +276,18 @@ class EncryptedSettingsStore:
         document = self.load_document()
         changed: set[str] = set()
         for key, raw_value in patch.items():
+            if key == "model_pool":
+                value = self._coerce_model_pool(raw_value, document["values"].get("model_pool", []))
+                if document["values"].get(key) != value:
+                    document["values"][key] = value
+                    changed.add(key)
+                continue
+            if key in MODEL_SETTINGS_KEYS:
+                value = raw_value.strip() if isinstance(raw_value, str) else raw_value
+                if document["values"].get(key) != value:
+                    document["values"][key] = value
+                    changed.add(key)
+                continue
             if key not in EDITABLE_FIELDS:
                 continue
             if key in SECRET_FIELDS and not _is_configured(raw_value):
@@ -206,6 +299,13 @@ class EncryptedSettingsStore:
         for target, fields in TEST_FIELDS.items():
             if changed & fields:
                 document["tests"].pop(target, None)
+        if "model_pool" in changed:
+            current_ids = {item["id"] for item in document["values"].get("model_pool", [])}
+            document["tests"] = {
+                target: result
+                for target, result in document["tests"].items()
+                if not target.startswith("model:") or target.removeprefix("model:") in current_ids
+            }
         self.save_document(document)
 
     def record_test(self, result: RuntimeSettingsTestResult, values: dict[str, Any] | None = None) -> None:
@@ -220,7 +320,18 @@ class EncryptedSettingsStore:
     def public_response(self) -> RuntimeSettingsResponse:
         document = self.load_document()
         values = {**self.defaults(), **document["values"]}
-        public_values = {key: value for key, value in values.items() if key not in SECRET_FIELDS}
+        public_values = {
+            key: value
+            for key, value in values.items()
+            if key not in SECRET_FIELDS and key not in LEGACY_SECRET_FIELDS
+        }
+        public_values["model_pool"] = [
+            {
+                **{key: value for key, value in item.items() if key != "api_key"},
+                "api_key_configured": _is_configured(item.get("api_key")),
+            }
+            for item in values.get("model_pool", [])
+        ]
         configured = {key: _is_configured(values.get(key)) for key in FIELDS}
         configured.update(self.bootstrap_statuses())
         current_tests = self.current_tests(values, document["tests"])
@@ -232,6 +343,38 @@ class EncryptedSettingsStore:
             readiness=self.readiness(values, current_tests),
         )
 
+    @staticmethod
+    def _coerce_model_pool(raw_pool: Any, existing_pool: Any) -> list[dict[str, Any]]:
+        if not isinstance(raw_pool, list):
+            raise ValueError("模型池必须是列表。")
+        if len(raw_pool) > 50:
+            raise ValueError("模型池最多保存 50 个模型。")
+        existing_by_id = {
+            str(item.get("id")): item
+            for item in existing_pool
+            if isinstance(item, dict) and item.get("id")
+        }
+        normalized: list[dict[str, Any]] = []
+        seen: set[str] = set()
+        for raw_item in raw_pool:
+            if not isinstance(raw_item, dict):
+                raise ValueError("模型配置格式不正确。")
+            item_data = dict(raw_item)
+            item_id = str(item_data.get("id", "")).strip()
+            if item_id in seen:
+                raise ValueError(f"模型 ID 重复：{item_id}")
+            if not _is_configured(item_data.get("api_key")):
+                item_data["api_key"] = existing_by_id.get(item_id, {}).get("api_key", "")
+            item_data.pop("api_key_configured", None)
+            item = ModelPoolItem.model_validate(item_data)
+            if item.category == "language":
+                item.capabilities = ["orchestration"]
+            elif not item.capabilities:
+                raise ValueError(f"多模态模型“{item.name}”至少需要选择一种能力。")
+            normalized.append(item.model_dump())
+            seen.add(item.id)
+        return normalized
+
     def bootstrap_statuses(self) -> dict[str, bool]:
         return {
             "database_status": _is_configured(self.bootstrap.database_url),
@@ -272,10 +415,67 @@ class EncryptedSettingsStore:
                 untested.append(f"{group}：尚未通过连接测试")
                 continue
             completed += 1
+
+        pool = [item for item in values.get("model_pool", []) if isinstance(item, dict) and item.get("enabled", True)]
+        models = {str(item.get("id")): item for item in pool if item.get("id")}
+        model_missing: list[str] = []
+        model_untested: list[str] = []
+
+        language_models = [item for item in pool if item.get("category") == "language"]
+        if not language_models:
+            model_missing.append("至少添加一个已启用的大语言模型")
+        orchestrator_id = str(values.get("orchestrator_model_id", ""))
+        orchestrator = models.get(orchestrator_id)
+        if not orchestrator_id:
+            model_missing.append("选择总调度模型")
+        elif not orchestrator or orchestrator.get("category") != "language":
+            model_missing.append("总调度模型已删除、已停用或类型不正确")
+        elif not tests.get(f"model:{orchestrator_id}", {}).get("ok"):
+            model_untested.append(f"总调度模型“{orchestrator.get('name')}”尚未通过测试")
+
+        route_specs = [
+            ("空间理解", "spatial_routing_mode", "spatial_model_id", "spatial_understanding"),
+            ("生图", "image_routing_mode", "image_model_id", "image_generation"),
+        ]
+        for label, mode_key, selected_key, capability in route_specs:
+            eligible = [item for item in pool if capability in item.get("capabilities", [])]
+            mode = values.get(mode_key, "auto")
+            if not eligible:
+                model_missing.append(f"添加支持{label}的多模态模型")
+                continue
+            if mode == "manual":
+                selected_id = str(values.get(selected_key, ""))
+                selected = models.get(selected_id)
+                if not selected_id:
+                    model_missing.append(f"手动选择{label}模型")
+                elif not selected or capability not in selected.get("capabilities", []):
+                    model_missing.append(f"{label}模型已删除、已停用或能力不匹配")
+                elif not tests.get(f"model:{selected_id}", {}).get("ok"):
+                    model_untested.append(f"{label}模型“{selected.get('name')}”尚未通过测试")
+            elif mode == "auto":
+                if not any(tests.get(f"model:{item['id']}", {}).get("ok") for item in eligible):
+                    model_untested.append(f"{label}自动路由没有已测试的候选模型")
+            else:
+                model_missing.append(f"{label}路由方式无效")
+
+        incomplete_models = [
+            item.get("name", item.get("model_id", "未命名模型"))
+            for item in pool
+            if not all(_is_configured(item.get(key)) for key in ("name", "model_id", "base_url", "api_key"))
+        ]
+        if incomplete_models:
+            model_missing.append(f"补全模型配置：{'、'.join(incomplete_models)}")
+
+        if model_missing:
+            missing.append(f"AI 模型：{'；'.join(dict.fromkeys(model_missing))}")
+        elif model_untested:
+            untested.append(f"AI 模型：{'；'.join(dict.fromkeys(model_untested))}")
+        else:
+            completed += 1
         return SettingsReadiness(
             ready=not missing and not untested,
             completed_required=completed,
-            total_required=len(REQUIRED_GROUPS),
+            total_required=len(REQUIRED_GROUPS) + 1,
             missing=missing,
             untested=untested,
         )
@@ -287,6 +487,10 @@ class EncryptedSettingsStore:
     ) -> dict[str, dict[str, Any]]:
         current: dict[str, dict[str, Any]] = {}
         for target, result in tests.items():
+            if target.startswith("model:"):
+                model_id = target.removeprefix("model:")
+                if not any(item.get("id") == model_id for item in values.get("model_pool", [])):
+                    continue
             item = dict(result)
             if item.get("fingerprint") != self.test_fingerprint(target, values):
                 item["ok"] = False
@@ -300,6 +504,12 @@ class EncryptedSettingsStore:
                 "database_url": self.bootstrap.database_url,
                 "redis_url": self.bootstrap.redis_url,
             }
+        elif target.startswith("model:"):
+            model_id = target.removeprefix("model:")
+            relevant = next(
+                (item for item in values.get("model_pool", []) if item.get("id") == model_id),
+                {"id": model_id, "missing": True},
+            )
         else:
             relevant = {key: values.get(key) for key in sorted(TEST_FIELDS.get(target, set()))}
         payload = json.dumps(relevant, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")
@@ -343,8 +553,6 @@ async def test_runtime_settings(
             return await _test_storage(values)
         if target == "baidu_ocr":
             return await _test_baidu_ocr(values)
-        if target == "ai_models":
-            return await _test_ai_models(values)
         if target == "gpu":
             return await _test_gpu(values)
         if target == "langfuse":
@@ -433,24 +641,88 @@ async def _test_baidu_ocr(values: dict[str, Any]) -> RuntimeSettingsTestResult:
     )
 
 
-async def _test_ai_models(values: dict[str, Any]) -> RuntimeSettingsTestResult:
-    url = _join_api_url(values["ai_base_url"], values.get("ai_models_path", "/models"))
-    async with httpx.AsyncClient(timeout=12) as client:
-        response = await client.get(url, headers={"Authorization": f"Bearer {values['ai_api_key']}"})
-        response.raise_for_status()
+async def test_model_pool_item(values: dict[str, Any], model_id: str) -> RuntimeSettingsTestResult:
+    target = f"model:{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="模型不存在或已被删除。")
+    name = item.get("name") or item.get("model_id") or model_id
+    missing_fields = [
+        label
+        for key, label in (("model_id", "模型 ID"), ("base_url", "Base URL"), ("api_key", "API Key"))
+        if not _is_configured(item.get(key))
+    ]
+    if missing_fields:
+        return RuntimeSettingsTestResult(
+            target=target,
+            ok=False,
+            message=f"模型“{name}”配置不完整：缺少{'、'.join(missing_fields)}。",
+        )
+
+    url = _join_api_url(str(item["base_url"]), str(item.get("models_path", "/models")))
+    headers = {"Authorization": f"Bearer {item['api_key']}"}
+    try:
+        async with httpx.AsyncClient(timeout=12) as client:
+            response = await client.get(url, headers=headers)
+            response.raise_for_status()
         payload = response.json()
-    model_ids = {
-        item.get("id")
-        for item in payload.get("data", [])
-        if isinstance(item, dict) and item.get("id")
+    except httpx.HTTPStatusError as exc:
+        status_code = exc.response.status_code
+        if status_code in {401, 403}:
+            reason = "API Key 无效或账号无权访问模型列表"
+        elif status_code == 404:
+            reason = f"模型列表地址不存在，请检查路径 {item.get('models_path', '/models')}"
+        else:
+            reason = f"模型列表接口返回 HTTP {status_code}"
+        return RuntimeSettingsTestResult(
+            target=target,
+            ok=False,
+            message=f"模型“{name}”测试失败：{reason}。",
+            details={"model_id": item["model_id"], "status_code": status_code},
+        )
+    except httpx.RequestError as exc:
+        return RuntimeSettingsTestResult(
+            target=target,
+            ok=False,
+            message=f"模型“{name}”无法连接 API：{exc}。",
+            details={"model_id": item["model_id"]},
+        )
+    except (ValueError, json.JSONDecodeError):
+        return RuntimeSettingsTestResult(
+            target=target,
+            ok=False,
+            message=f"模型“{name}”的模型列表接口没有返回有效 JSON。",
+            details={"model_id": item["model_id"]},
+        )
+
+    rows = payload.get("data", []) if isinstance(payload, dict) else []
+    listed_ids = {
+        candidate.get("id")
+        for candidate in rows
+        if isinstance(candidate, dict) and candidate.get("id")
     }
-    selected = [values.get("orchestrator_model"), values.get("vision_model"), values.get("image_model")]
-    missing = [model for model in selected if model and model_ids and model not in model_ids]
+    if not listed_ids:
+        return RuntimeSettingsTestResult(
+            target=target,
+            ok=False,
+            message=f"模型“{name}”的 API 可以连接，但模型列表为空，暂时无法确认模型 ID。",
+            details={"model_id": item["model_id"], "model_count": 0},
+        )
+    if item["model_id"] not in listed_ids:
+        return RuntimeSettingsTestResult(
+            target=target,
+            ok=False,
+            message=f"模型“{name}”测试失败：账号模型列表中未找到 ID“{item['model_id']}”。",
+            details={"model_id": item["model_id"], "model_count": len(listed_ids)},
+        )
     return RuntimeSettingsTestResult(
-        target="ai_models",
-        ok=not missing,
-        message="API Key 有效，三个模型均可用。" if not missing else "API 可以连接，但部分模型名不在账号模型列表中。",
-        details={"missing_models": missing, "model_count": len(model_ids)},
+        target=target,
+        ok=True,
+        message=f"模型“{name}”连接正常，模型 ID“{item['model_id']}”可用。",
+        details={"model_id": item["model_id"], "model_count": len(listed_ids)},
     )
 
 

services/api/app/settings_schema.py

@@ -179,112 +179,10 @@ CATEGORIES = [
     ),
     SettingCategory(
         id="models",
-        label="AI 模型",
-        description="统一配置总调度、空间理解和生图模型；同一个聚合 API 可以同时承担三类能力。",
-        test_targets=["ai_models"],
+        label="模型池",
+        description="分别添加大语言模型与多模态模型。每个模型独立配置、独立测试，再用于总调度、空间理解或生图路由。",
         required_for_workflow=True,
-        fields=[
-            SettingField(
-                key="ai_provider",
-                label="API 来源",
-                description="选择自然语意预设；自建或其他聚合服务选择兼容接口。",
-                kind="select",
-                required=True,
-                default="lingke",
-                options=[
-                    option("lingke", "聚合引擎 AIGC", "你提供的 lk666.ai 聚合服务，模型名以其控制台为准。"),
-                    option("openai", "OpenAI 官方", "直接使用 OpenAI 官方 API。"),
-                    option("custom", "其他兼容接口", "支持 OpenAI 请求格式的代理、自建或聚合服务。"),
-                ],
-            ),
-            SettingField(
-                key="ai_base_url",
-                label="API Base URL",
-                description="填写到 /v1 层级；聚合引擎的准确地址请从登录后的开发者文档复制。",
-                kind="url",
-                required=True,
-                placeholder="https://example.com/v1",
-            ),
-            SettingField(
-                key="ai_api_key",
-                label="API Key",
-                description="同一聚合账号可供总调度、多模态和生图使用。",
-                kind="password",
-                required=True,
-                secret=True,
-            ),
-            SettingField(
-                key="orchestrator_model",
-                label="总调度模型",
-                description="负责追问、拆解任务、维护 Plan / Scene / Style 状态。",
-                kind="combobox",
-                required=True,
-                placeholder="选择预设或输入平台模型 ID",
-                options=[
-                    option("gpt-5.6-terra", "GPT-5.6 Terra", "速度与规划能力均衡，适合日常工作流。"),
-                    option("gpt-5.6-sol", "GPT-5.6 Sol", "更强推理，适合复杂户型和高价值方案。"),
-                    option("gpt-5", "GPT-5", "通用调度预设，具体可用性取决于账号。"),
-                ],
-            ),
-            SettingField(
-                key="vision_model",
-                label="空间理解模型",
-                description="读取户型、参考图和渲染结果，判断空间关系与审美一致性。",
-                kind="combobox",
-                required=True,
-                placeholder="选择预设或输入多模态模型 ID",
-                options=[
-                    option("gpt-5.6-sol", "GPT-5.6 Sol", "优先空间推理和复杂视觉评审。"),
-                    option("gemini-3-pro", "Gemini 3 Pro", "长上下文与多模态理解预设。"),
-                    option("gpt-5", "GPT-5", "通用多模态预设。"),
-                ],
-            ),
-            SettingField(
-                key="image_model",
-                label="默认生图模型",
-                description="先使用你指定的 gpt-image-2；后续可以增加按任务自动路由。",
-                kind="combobox",
-                required=True,
-                default="gpt-image-2",
-                options=[
-                    option("gpt-image-2", "GPT Image 2", "默认室内方向图与局部编辑模型。"),
-                    option("seedream-5.0", "Seedream 5.0", "适合高质量中文场景生成，模型 ID 以平台为准。"),
-                    option("nano-banana-pro", "Nano Banana Pro", "适合参考图编辑，模型 ID 以平台为准。"),
-                ],
-            ),
-            SettingField(
-                key="ai_models_path",
-                label="模型列表路径",
-                description="测试按钮使用。OpenAI 兼容接口通常为 /models。",
-                kind="text",
-                default="/models",
-                advanced=True,
-            ),
-            SettingField(
-                key="ai_chat_path",
-                label="对话路径",
-                description="OpenAI 兼容接口通常为 /chat/completions。",
-                kind="text",
-                default="/chat/completions",
-                advanced=True,
-            ),
-            SettingField(
-                key="ai_image_generation_path",
-                label="生图路径",
-                description="OpenAI 兼容接口通常为 /images/generations。",
-                kind="text",
-                default="/images/generations",
-                advanced=True,
-            ),
-            SettingField(
-                key="ai_image_edit_path",
-                label="图片编辑路径",
-                description="OpenAI 兼容接口通常为 /images/edits。",
-                kind="text",
-                default="/images/edits",
-                advanced=True,
-            ),
-        ],
+        fields=[],
     ),
     SettingCategory(
         id="gpu",