refactor: enhance dynamic model fetching with improved concurrency and error handling

apps/models_provider/base_model_provider.py

@@ -10,7 +10,7 @@
 from functools import reduce
 from typing import Dict, Iterator, List, Type
 
-from common.exception.app_exception import AppApiException
+from common.exception.app_exception import AppApiException, BadRequest400, NotFound404
 from common.utils.common import encryption
 from django.utils.translation import gettext_lazy as _
 from pydantic import BaseModel
@@ -28,6 +28,15 @@ class ValidCode(Enum):
     model_not_fount = 404
 
 
+# 凭据/连通性校验阶段抛出的 AppApiException，按业务码(ValidCode)路由到一致的
+# HTTP 异常，使业务码与状态码对齐；未命中任一 ValidCode 的异常统一按客户端
+# 参数错误(BadRequest400)处理。
+_VALID_CODE_HTTP_EXCEPTION = {
+    ValidCode.valid_error.value: BadRequest400,
+    ValidCode.model_not_fount.value: NotFound404,
+}
+
+
 class DownModelChunk:
     def __init__(self, status: DownModelChunkStatus, digest: str, progress: int, details: str, index: int):
         self.details = details
@@ -104,13 +113,15 @@ def get_model_provide_info(self):
     def get_model_type_list(self):
         return self.get_model_info_manage().get_model_type_list()
 
-    def get_model_list(self, model_type, api_key=None, api_base=None, credential=None):
+    def get_model_list(self, model_type, api_key=None, api_base=None, credential=None, allow_builtin_fallback=True):
         if model_type is None:
             raise AppApiException(500, _("Model type cannot be empty"))
         if api_key or credential:
             manage = self._get_dynamic_manage_cached(api_key, api_base, credential)
             if manage is not None:
                 return manage.get_model_list_by_model_type(model_type)
+            if not allow_builtin_fallback:
+                return []
         return self.get_model_info_manage().get_model_list_by_model_type(model_type)
 
     # ---------- 动态模型列表框架 ----------
@@ -123,6 +134,18 @@ def get_dynamic_model_info_list(self, api_key, api_base, credential=None):
         items = self.fetch_dynamic_model_items(api_key, api_base, credential)
         if not items:
             return None
+        return self._build_dynamic_model_info_list(items, self._classify_dynamic_item_or_id)
+
+    def _classify_dynamic_item_or_id(self, item, model_id):
+        """默认判型：优先用 /models 结构化元数据，否则按模型 id 关键词。"""
+        return self.classify_dynamic_item(item) if isinstance(item, dict) else self.classify_dynamic_model(model_id)
+
+    def _build_dynamic_model_info_list(self, items, classify, id_extractor=None):
+        """共享的动态条目 -> ModelInfo 构建：内置名沿用原实现，其余按 classify 判型找实现。
+
+        classify(item, model_id) 返回 ModelTypeConst，返回 None 表示跳过该条；
+        id_extractor(item) 返回最终模型 id，缺省取 dict 的 id 字段或 item 本身。
+        """
         builtin_manage = self.get_model_info_manage()
         builtin_by_name = {}
         impl_map = {}
@@ -131,7 +154,7 @@ def get_dynamic_model_info_list(self, api_key, api_base, credential=None):
             impl_map.setdefault(info.model_type, (info.model_credential, info.model_class))
         info_list = []
         for item in items:
-            model_id = item.get("id") if isinstance(item, dict) else item
+            model_id = id_extractor(item) if id_extractor else (item.get("id") if isinstance(item, dict) else item)
             if not model_id:
                 continue
             # 内置列表中已有的模型：类型与实现直接沿用原有逻辑，不依赖关键词猜型。
@@ -141,10 +164,9 @@ def get_dynamic_model_info_list(self, api_key, api_base, credential=None):
                     ModelInfo(model_id, builtin.desc, builtin.model_type, builtin.model_credential, builtin.model_class)
                 )
                 continue
-            if isinstance(item, dict):
-                model_type = self.classify_dynamic_item(item)
-            else:
-                model_type = self.classify_dynamic_model(model_id)
+            model_type = classify(item, model_id)
+            if model_type is None:
+                continue
             impl = impl_map.get(_model_type_name(model_type))
             if impl is None:
                 # 视觉输入的对话模型（IMAGE 类型）在无视觉实现时按 LLM 兜底；其余未支持类型直接丢弃。
@@ -174,6 +196,7 @@ def classify_dynamic_item(self, item):
 
     def fetch_dynamic_model_items(self, api_key, api_base, credential=None):
         """OpenAI 兼容协议实时获取模型条目（保留官方元数据，供判型使用）。"""
+        api_key = api_key or self._credential_api_key(credential)
         api_base = (
             (api_base or self._credential_api_base(credential) or self.get_dynamic_default_api_base() or "")
             .strip()
@@ -252,6 +275,17 @@ def _credential_api_base(credential):
                 return value
         return None
 
+    @staticmethod
+    def _credential_api_key(credential):
+        """从完整凭证里找 API Key 字段（各供应商字段名不同：api_key/dashscope_api_key 等）。"""
+        if not isinstance(credential, dict):
+            return None
+        for key in ("api_key", "dashscope_api_key", "apikey", "access_key", "access_key_id", "token"):
+            value = str(credential.get(key) or "").strip()
+            if value:
+                return value
+        return None
+
     @staticmethod
     def _credential_cache_fingerprint(credential):
         """完整凭证的缓存指纹：仅取字符串值参与排序，避免大对象。"""
@@ -316,9 +350,23 @@ def is_valid_credential(
         raise_exception=False,
     ):
         model_info = self.get_model_info_manage().get_model_info(model_type, model_name)
-        return model_info.model_credential.is_valid(
-            model_type, model_name, model_credential, model_params, self, raise_exception=raise_exception
-        )
+        validate_params = dict(model_params)
+        # 兼容 OpenAI 协议的模型在连通性校验时注入短超时并关闭重试，
+        # 防止不可达地址触发 OpenAI 客户端默认 600s read 超时导致接口长期挂起。
+        validation_timeout = getattr(model_info.model_class, "credential_validation_timeout", None)
+        if validation_timeout is not None:
+            validate_params["request_timeout"] = validation_timeout
+            validate_params["max_retries"] = 0
+        try:
+            return model_info.model_credential.is_valid(
+                model_type, model_name, model_credential, validate_params, self, raise_exception=raise_exception
+            )
+        except AppApiException as e:
+            if not raise_exception:
+                raise
+            # 拿到模型侧返回的 ValidCode 就按它路由；否则统一按 400 处理。
+            http_exception = _VALID_CODE_HTTP_EXCEPTION.get(e.code, BadRequest400)
+            raise http_exception(http_exception.status_code, e.message) from e
 
     def get_model(self, model_type, model_name, model_credential: Dict[str, object], **model_kwargs) -> BaseModel:
         model_info = self.get_model_info_manage().get_model_info(model_type, model_name)

apps/models_provider/impl/aliyun_bai_lian_model_provider/aliyun_bai_lian_model_provider.py

@@ -297,6 +297,7 @@ def get_model_info_manage(self):
 
     def get_dynamic_model_info_list(self, api_key, api_base, credential=None):
         """DashScope 专有实现：按 capabilities/inference_metadata 精确映射模型类型。"""
+        api_key = api_key or self._credential_api_key(credential)
         # 模型列表始终走官方 DashScope 接口；compatible-mode 地址归一化回原生 /api/v1。
         items = []
         for base in dict.fromkeys([self._query_base_for(api_base), "https://dashscope.aliyuncs.com/api/v1"]):
@@ -353,46 +354,75 @@ def get_base_model_list(api_key: str, api_base: str = None):
         """调用阿里云百炼（DashScope）官方 /models 列表接口实时获取可用模型。
 
         兼容两种返回结构：
-        - DashScope：{"output": {"models": [...]}}
+        - DashScope：{"output": {"total": N, "models": [...]}}
         - 网关/OpenAI 风格：{"data": [...]}（如 ai.fit2cloud.cn 网关）
         传入 api_base 时会作为 SDK 的 base_address，否则使用 SDK 默认的
         https://dashscope.aliyuncs.com/api/v1。
+
+        首屏返回 total 后并发拉取剩余页，避免 4 次串行请求造成 2s+ 延迟；
+        网关类接口不返回 total 时退化为串行直到短页或没有新增。
         """
         import dashscope
+        from concurrent.futures import ThreadPoolExecutor
 
-        models = []
-        seen = set()
-        page_no = 1
         page_size = 100
-        while True:
-            kwargs = {"page": page_no, "page_size": page_size, "api_key": api_key}
-            if api_base:
-                kwargs["base_address"] = api_base
-            response = dashscope.Models.list(**kwargs)
+
+        def page_models(response):
             if getattr(response, "status_code", None) != 200:
-                if models:
-                    return models
                 return []
             output = response.output or {}
             if isinstance(output, dict):
-                page_models = output.get("models") or output.get("data") or []
-            elif isinstance(output, list):
-                page_models = output
-            else:
-                page_models = []
-            if not page_models:
-                break
-            new_count = 0
-            for item in page_models:
+                return output.get("models") or output.get("data") or []
+            return output or []
+
+        def total_of(response):
+            output = response.output or {}
+            total = output.get("total") if isinstance(output, dict) else None
+            return int(total) if isinstance(total, int) or (isinstance(total, str) and total.isdigit()) else None
+
+        def fetch(page_no):
+            kwargs = {"page": page_no, "page_size": page_size, "api_key": api_key}
+            if api_base:
+                kwargs["base_address"] = api_base
+            return dashscope.Models.list(**kwargs)
+
+        responses = {}
+        first = fetch(1)
+        if getattr(first, "status_code", None) != 200:
+            return []
+        responses[1] = page_models(first)
+
+        total = total_of(first)
+        if total is not None:
+            total_pages = max(1, (total + page_size - 1) // page_size)
+            if total_pages > 1:
+                with ThreadPoolExecutor(max_workers=min(total_pages, 8)) as executor:
+                    future_map = {executor.submit(fetch, page_no): page_no for page_no in range(2, total_pages + 1)}
+                    for future in future_map:
+                        try:
+                            page_no = future_map[future]
+                            responses[page_no] = page_models(future.result())
+                        except Exception:
+                            responses[page_no] = []
+        else:
+            page_no = 2
+            while True:
+                response = fetch(page_no)
+                items = page_models(response)
+                responses[page_no] = items
+                if not items or len(items) < page_size:
+                    break
+                page_no += 1
+
+        models = []
+        seen = set()
+        for page_no in sorted(responses):
+            for item in responses[page_no]:
                 model_id = item.get("model") or item.get("id") or item.get("model_id") or item.get("name")
                 if not model_id or model_id in seen:
                     continue
                 seen.add(model_id)
                 models.append(item)
-                new_count += 1
-            if len(page_models) < page_size or new_count == 0:
-                break
-            page_no += 1
         return models
 
     @staticmethod

apps/models_provider/impl/anthropic_model_provider/anthropic_model_provider.py

@@ -57,6 +57,7 @@ class AnthropicModelProvider(IModelProvider):
 
     def fetch_dynamic_model_items(self, api_key, api_base, credential=None):
         """Anthropic 专有协议：GET {base}/v1/models，鉴权用 x-api-key 头。"""
+        api_key = api_key or self._credential_api_key(credential)
         api_base = (api_base or self.get_dynamic_default_api_base() or "").strip().rstrip("/")
         if not api_base:
             return []

apps/models_provider/impl/aws_bedrock_model_provider/aws_bedrock_model_provider.py

@@ -4,7 +4,6 @@
 import os
 from common.utils.common import get_file_content
 from models_provider.base_model_provider import (
-    _model_type_name,
     IModelProvider,
     ModelProvideInfo,
     ModelInfo,
@@ -239,28 +238,11 @@ def get_dynamic_model_info_list(self, api_key, api_base, credential=None):
             summaries = client.list_foundation_models().get("modelSummaries") or []
         except Exception:
             return None
-        builtin_manage = self.get_model_info_manage()
-        impl_map = {}
-        for info in builtin_manage.model_list:
-            impl_map.setdefault(info.model_type, (info.model_credential, info.model_class))
-        builtin_by_name = {info.name: info for info in builtin_manage.model_list}
-        info_list = []
-        for item in summaries:
-            model_id = item.get("modelId") or item.get("modelArn")
-            if not model_id:
-                continue
-            builtin = builtin_by_name.get(model_id)
-            if builtin is not None:
-                info_list.append(
-                    ModelInfo(model_id, builtin.desc, builtin.model_type, builtin.model_credential, builtin.model_class)
-                )
-                continue
-            model_type = self.classify_dynamic_model(model_id)
-            impl = impl_map.get(_model_type_name(model_type))
-            if impl is None:
-                continue
-            info_list.append(ModelInfo(model_id, _(""), model_type, impl[0], impl[1]))
-        return info_list or None
+        return self._build_dynamic_model_info_list(
+            summaries,
+            lambda item, model_id: self.classify_dynamic_model(model_id),
+            id_extractor=lambda item: item.get("modelId") or item.get("modelArn"),
+        )
 
     def __init__(self):
         self._model_info_manage = _initialize_model_info()

apps/models_provider/impl/base_chat_open_ai.py

@@ -1,7 +1,7 @@
 # coding=utf-8
 import base64
 from concurrent.futures import ThreadPoolExecutor
-from typing import Dict, Optional, Any, Iterator, cast, Union, Sequence, Callable, Mapping, AsyncIterator
+from typing import Dict, Optional, Any, Iterator, cast, Union, Sequence, Callable, Mapping, AsyncIterator, ClassVar
 
 from langchain_core.language_models import LanguageModelInput
 from langchain_core.messages import (
@@ -94,6 +94,9 @@ def _convert_delta_to_message_chunk(
 class BaseChatOpenAI(ChatOpenAI):
     usage_metadata: dict = {}
     custom_get_token_ids = custom_get_token_ids
+    # 连通性校验使用短超时，避免不可达地址导致创建/编辑模型接口长时间挂起。
+    # 仅作用于 is_valid_credential 期间构造的临时校验实例，不影响真实推理。
+    credential_validation_timeout: ClassVar[float] = 5.0
 
     def get_last_generation_info(self) -> Optional[Dict[str, Any]]:
         return self.usage_metadata

apps/models_provider/impl/gemini_model_provider/gemini_model_provider.py

@@ -13,7 +13,6 @@
 
 from common.utils.common import get_file_content
 from models_provider.base_model_provider import (
-    _model_type_name,
     IModelProvider,
     ModelProvideInfo,
     ModelInfo,
@@ -145,6 +144,7 @@
 class GeminiModelProvider(IModelProvider):
     def get_dynamic_model_info_list(self, api_key, api_base, credential=None):
         """Gemini 原生 /v1beta/models 返回 supportedGenerationMethods，可精确判型；失败时回退框架默认。"""
+        api_key = api_key or self._credential_api_key(credential)
         base = (api_base or self.get_dynamic_default_api_base() or "").strip().rstrip("/")
         if not base:
             return None
@@ -159,45 +159,30 @@ def get_dynamic_model_info_list(self, api_key, api_base, credential=None):
             items = response.json().get("models") or []
             if not items:
                 return None
-            builtin_manage = self.get_model_info_manage()
-            builtin_by_name = {}
-            impl_map = {}
-            for info in builtin_manage.model_list:
-                builtin_by_name.setdefault(info.name, info)
-                impl_map.setdefault(info.model_type, (info.model_credential, info.model_class))
-            info_list = []
-            for item in items:
-                name = str(item.get("name") or "").removeprefix("models/")
-                if not name:
-                    continue
-                builtin = builtin_by_name.get(name)
-                if builtin is not None:
-                    info_list.append(
-                        ModelInfo(name, builtin.desc, builtin.model_type, builtin.model_credential, builtin.model_class)
-                    )
-                    continue
-                methods = {str(m).lower() for m in (item.get("supportedGenerationMethods") or [])}
-                if not methods:
-                    # 官方接口未给出可用能力时跳过，避免误判成 LLM。
-                    continue
-                if "embedcontent" in methods:
-                    model_type = ModelTypeConst.EMBEDDING
-                elif any(m in methods for m in ("generatecontent", "counttokens", "generateanswer")):
-                    model_type = ModelTypeConst.LLM
-                elif "generatevideos" in methods:
-                    model_type = ModelTypeConst.TTV
-                else:
-                    model_type = self.classify_dynamic_model(name)
-                impl = impl_map.get(_model_type_name(model_type))
-                if impl is None:
-                    continue
-                info_list.append(ModelInfo(name, _(""), model_type, impl[0], impl[1]))
-            return info_list or None
         except Exception:
             return None
+        return self._build_dynamic_model_info_list(
+            items,
+            self._classify_gemini_item,
+            id_extractor=lambda item: str(item.get("name") or "").removeprefix("models/"),
+        )
+
+    def _classify_gemini_item(self, item, model_id):
+        """按官方 supportedGenerationMethods 判型；无能力字段时跳过，避免误判成 LLM。"""
+        methods = {str(m).lower() for m in (item.get("supportedGenerationMethods") or [])}
+        if not methods:
+            return None
+        if "embedcontent" in methods:
+            return ModelTypeConst.EMBEDDING
+        if any(m in methods for m in ("generatecontent", "counttokens", "generateanswer")):
+            return ModelTypeConst.LLM
+        if "generatevideos" in methods:
+            return ModelTypeConst.TTV
+        return self.classify_dynamic_model(model_id)
 
     def fetch_dynamic_model_ids(self, api_key, api_base, credential=None):
         """Gemini OpenAI 兼容端点：GET {base}/v1beta/openai/models，Bearer 鉴权。"""
+        api_key = api_key or self._credential_api_key(credential)
         api_base = (api_base or self.get_dynamic_default_api_base() or "").strip().rstrip("/")
         if not api_base:
             return []