feat: 模型列表 API 支持格式转换兼容性过滤

- 新增 get_compatible_provider_formats 函数，基于端点 format_acceptance_config 过滤兼容的 Provider
- 模型列表查询根据客户端格式和全局转换开关返回可用模型
- 缓存 key 增加 client_format 维度，避免不同格式的缓存混用
- GlobalModel 解析支持 provider_model_mappings 和 model_mappings 匹配
- 修复格式兼容性检查中 config 类型校验缺失问题

src/api/base/models_service.py

@@ -17,28 +17,28 @@ from sqlalchemy.orm import Session
 
 from src.config.constants import CacheTTL
 from src.core.cache_service import CacheService
+from src.core.api_format.conversion.compatibility import is_format_compatible
 from src.core.logger import logger
 from src.services.model.availability import ModelAvailabilityQuery
-from src.models.database import (
+from src.models.database import ApiKey, Model, Provider, ProviderEndpoint, User
-    ApiKey,
-    Model,
-    User,
-)
 
 # 缓存 key 前缀
 _CACHE_KEY_PREFIX = "models:list"
 _CACHE_TTL = CacheTTL.MODEL  # 300 秒
 
 
-def _get_cache_key(api_formats: list[str]) -> str:
+def _get_cache_key(api_formats: list[str], client_format: Optional[str] = None) -> str:
     """生成缓存 key"""
     formats_str = ",".join(sorted(api_formats))
-    return f"{_CACHE_KEY_PREFIX}:{formats_str}"
+    format_key = (client_format or "any").lower()
+    return f"{_CACHE_KEY_PREFIX}:{format_key}:{formats_str}"
 
 
-async def _get_cached_models(api_formats: list[str]) -> Optional[list["ModelInfo"]]:
+async def _get_cached_models(
+    api_formats: list[str], client_format: Optional[str] = None
+) -> Optional[list["ModelInfo"]]:
     """从缓存获取模型列表"""
-    cache_key = _get_cache_key(api_formats)
+    cache_key = _get_cache_key(api_formats, client_format)
     try:
         cached = await CacheService.get(cache_key)
         if cached:
@@ -49,9 +49,13 @@ async def _get_cached_models(api_formats: list[str]) -> Optional[list["ModelInfo
     return None
 
 
-async def _set_cached_models(api_formats: list[str], models: list["ModelInfo"]) -> None:
+async def _set_cached_models(
+    api_formats: list[str],
+    models: list["ModelInfo"],
+    client_format: Optional[str] = None,
+) -> None:
     """将模型列表写入缓存"""
-    cache_key = _get_cache_key(api_formats)
+    cache_key = _get_cache_key(api_formats, client_format)
     try:
         data = [asdict(m) for m in models]
         await CacheService.set(cache_key, data, ttl_seconds=_CACHE_TTL)
@@ -189,7 +193,117 @@ class AccessRestrictions:
         return True
 
 
-def get_available_provider_ids(db: Session, api_formats: list[str]) -> set[str]:
+def _normalize_api_formats(
+    api_formats: Optional[list[str]],
+    provider_to_formats: Optional[dict[str, set[str]]] = None,
+) -> list[str]:
+    """规范化 API 格式列表（大写），必要时从 provider_to_formats 兜底"""
+    if api_formats:
+        return [str(fmt).upper() for fmt in api_formats if fmt is not None]
+    if not provider_to_formats:
+        return []
+    all_formats: set[str] = set()
+    for formats in provider_to_formats.values():
+        all_formats.update(str(fmt).upper() for fmt in formats)
+    return list(all_formats)
+
+
+def _get_provider_model_names_for_formats(
+    model: Model, usable_formats: Optional[set[str]] = None
+) -> set[str]:
+    """
+    获取模型在指定格式下支持的 Provider 模型名称集合
+
+    用于 check_model_allowed_with_mappings 的 candidate_models 参数，
+    确保权限检查时只考虑当前格式支持的模型名。
+    """
+    names: set[str] = {model.provider_model_name}
+    raw_mappings = model.provider_model_mappings
+    if not isinstance(raw_mappings, list):
+        return names
+
+    usable_formats_upper = {f.upper() for f in usable_formats} if usable_formats else None
+
+    for raw in raw_mappings:
+        if not isinstance(raw, dict):
+            continue
+        name = raw.get("name")
+        if not isinstance(name, str) or not name.strip():
+            continue
+
+        mapping_api_formats = raw.get("api_formats")
+        if usable_formats_upper and mapping_api_formats:
+            if isinstance(mapping_api_formats, list):
+                mapping_formats = {
+                    str(fmt).upper()
+                    for fmt in mapping_api_formats
+                    if isinstance(fmt, str)
+                }
+                if not mapping_formats & usable_formats_upper:
+                    continue
+
+        names.add(name.strip())
+
+    return names
+
+
+def get_compatible_provider_formats(
+    db: Session,
+    client_format: str,
+    api_formats: list[str],
+    global_conversion_enabled: bool,
+) -> dict[str, set[str]]:
+    """
+    获取与客户端格式兼容的 Provider -> formats 映射
+
+    兼容性基于端点格式、format_acceptance_config 与全局格式转换开关。
+    """
+    normalized_formats = _normalize_api_formats(api_formats)
+    if not normalized_formats:
+        return {}
+
+    target_formats = set(normalized_formats)
+    client_format_upper = client_format.upper()
+
+    endpoint_rows = (
+        db.query(
+            ProviderEndpoint.provider_id,
+            ProviderEndpoint.api_format,
+            ProviderEndpoint.format_acceptance_config,
+        )
+        .join(Provider, ProviderEndpoint.provider_id == Provider.id)
+        .filter(
+            Provider.is_active.is_(True),
+            ProviderEndpoint.is_active.is_(True),
+            ProviderEndpoint.api_format.in_(list(target_formats)),
+        )
+        .all()
+    )
+
+    provider_to_formats: dict[str, set[str]] = {}
+    for provider_id, endpoint_format, format_acceptance_config in endpoint_rows:
+        if not provider_id or not endpoint_format:
+            continue
+        is_compatible, _needs_conversion, _reason = is_format_compatible(
+            client_format_upper,
+            str(endpoint_format),
+            format_acceptance_config,
+            is_stream=False,
+            global_conversion_enabled=global_conversion_enabled,
+        )
+        if not is_compatible:
+            continue
+        fmt_upper = str(endpoint_format).upper()
+        provider_to_formats.setdefault(provider_id, set()).add(fmt_upper)
+
+    return provider_to_formats
+
+
+def get_available_provider_ids(
+    db: Session,
+    api_formats: list[str],
+    provider_to_formats: Optional[dict[str, set[str]]] = None,
+) -> set[str]:
     """
     返回有可用端点的 Provider IDs
 
@@ -198,19 +312,27 @@ def get_available_provider_ids(db: Session, api_formats: list[str]) -> set[str]:
     - 端点是活跃的
     - Provider 下有活跃的 Key 且支持该 api_format（Key 直属 Provider，通过 api_formats 过滤）
     """
-    provider_to_formats = ModelAvailabilityQuery.get_providers_with_active_endpoints(db, api_formats)
+    normalized_formats = _normalize_api_formats(api_formats, provider_to_formats)
+    if provider_to_formats is None:
+        provider_to_formats = ModelAvailabilityQuery.get_providers_with_active_endpoints(
+            db, normalized_formats
+        )
     if not provider_to_formats:
         return set()
 
     return ModelAvailabilityQuery.get_providers_with_active_keys(
         db,
         set(provider_to_formats.keys()),
-        api_formats,
+        normalized_formats,
         provider_to_formats,
     )
 
 
-def _get_available_model_ids_for_format(db: Session, api_formats: list[str]) -> set[str]:
+def _get_available_model_ids_for_format(
+    db: Session,
+    api_formats: list[str],
+    provider_to_formats: Optional[dict[str, set[str]]] = None,
+) -> set[str]:
     """
     获取指定格式下真正可用的模型 ID 集合
 
@@ -220,14 +342,18 @@ def _get_available_model_ids_for_format(db: Session, api_formats: list[str]) ->
     3. **该端点的 Provider 关联了该模型**
     4. Key 的 allowed_models 允许该模型（null = 允许该 Provider 关联的所有模型）
     """
-    provider_to_formats = ModelAvailabilityQuery.get_providers_with_active_endpoints(db, api_formats)
+    normalized_formats = _normalize_api_formats(api_formats, provider_to_formats)
+    if provider_to_formats is None:
+        provider_to_formats = ModelAvailabilityQuery.get_providers_with_active_endpoints(
+            db, normalized_formats
+        )
     if not provider_to_formats:
         return set()
 
     provider_key_rules = ModelAvailabilityQuery.get_provider_key_rules(
         db,
         provider_ids=set(provider_to_formats.keys()),
-        api_formats=api_formats,
+        api_formats=normalized_formats,
         provider_to_endpoint_formats=provider_to_formats,
     )
 
@@ -260,17 +386,19 @@ def _get_available_model_ids_for_format(db: Session, api_formats: list[str]) ->
         model_mappings = (global_model.config or {}).get("model_mappings")
 
         rules = provider_key_rules.get(model_provider_id, [])
-        for allowed_models, _usable_formats in rules:
+        for allowed_models, usable_formats in rules:
             # None = 不限制
             if allowed_models is None:
                 available_model_ids.add(model_id)
                 break
 
             # 检查是否允许该模型（支持 model_mappings 正则匹配）
+            candidate_models = _get_provider_model_names_for_formats(model, usable_formats)
             is_allowed, _ = check_model_allowed_with_mappings(
                 model_name=model_id,
                 allowed_models=allowed_models,
                 model_mappings=model_mappings,
+                candidate_models=candidate_models,
             )
             if is_allowed:
                 available_model_ids.add(model_id)
@@ -335,6 +463,8 @@ async def list_available_models(
     available_provider_ids: set[str],
     api_formats: Optional[list[str]] = None,
     restrictions: Optional[AccessRestrictions] = None,
+    provider_to_formats: Optional[dict[str, set[str]]] = None,
+    client_format: Optional[str] = None,
 ) -> list[ModelInfo]:
     """
     获取可用模型列表（已去重，带缓存）
@@ -344,6 +474,8 @@ async def list_available_models(
         available_provider_ids: 有可用端点的 Provider ID 集合
         api_formats: API 格式列表，用于检查 Key 的 allowed_models
         restrictions: API Key/User 的访问限制
+        provider_to_formats: Provider -> formats 映射（兼容转换过滤用）
+        client_format: 客户端格式（用于缓存隔离）
 
     Returns:
         去重后的 ModelInfo 列表，按创建时间倒序
@@ -359,16 +491,20 @@ async def list_available_models(
         restrictions.allowed_providers is None and restrictions.allowed_models is None
     )
 
+    normalized_formats = _normalize_api_formats(api_formats, provider_to_formats)
+
     # 尝试从缓存获取
-    if api_formats and use_cache:
+    if normalized_formats and use_cache:
-        cached = await _get_cached_models(api_formats)
+        cached = await _get_cached_models(normalized_formats, client_format)
         if cached is not None:
             return cached
 
     # 如果提供了 api_formats，获取真正可用的模型 ID
     available_model_ids: Optional[set[str]] = None
-    if api_formats:
+    if normalized_formats:
-        available_model_ids = _get_available_model_ids_for_format(db, api_formats)
+        available_model_ids = _get_available_model_ids_for_format(
+            db, normalized_formats, provider_to_formats
+        )
         if not available_model_ids:
             return []
 
@@ -399,12 +535,11 @@ async def list_available_models(
         if info.id in seen_model_ids:
             continue
         seen_model_ids.add(info.id)
-
         result.append(info)
 
     # 只有无限制的情况才写入缓存
-    if api_formats and use_cache:
+    if normalized_formats and use_cache:
-        await _set_cached_models(api_formats, result)
+        await _set_cached_models(normalized_formats, result, client_format)
 
     return result
 
@@ -415,6 +550,7 @@ def find_model_by_id(
     available_provider_ids: set[str],
     api_formats: Optional[list[str]] = None,
     restrictions: Optional[AccessRestrictions] = None,
+    provider_to_formats: Optional[dict[str, set[str]]] = None,
 ) -> Optional[ModelInfo]:
     """
     按 ID 查找模型（仅支持 GlobalModel.name）
@@ -425,6 +561,7 @@ def find_model_by_id(
         available_provider_ids: 有可用端点的 Provider ID 集合
         api_formats: API 格式列表，用于检查 Key 的 allowed_models
         restrictions: API Key/User 的访问限制
+        provider_to_formats: Provider -> formats 映射（兼容转换过滤用）
 
     Returns:
         ModelInfo 或 None
@@ -432,10 +569,14 @@ def find_model_by_id(
     if not available_provider_ids:
         return None
 
+    normalized_formats = _normalize_api_formats(api_formats, provider_to_formats)
+
     # 如果提供了 api_formats，获取真正可用的模型 ID
     available_model_ids: Optional[set[str]] = None
-    if api_formats:
+    if normalized_formats:
-        available_model_ids = _get_available_model_ids_for_format(db, api_formats)
+        available_model_ids = _get_available_model_ids_for_format(
+            db, normalized_formats, provider_to_formats
+        )
         # 快速检查：如果目标模型不在可用列表中，直接返回 None
         if available_model_ids is not None and model_id not in available_model_ids:
             return None

src/api/public/models.py

@@ -17,6 +17,7 @@ from src.api.base.models_service import (
     AccessRestrictions,
     ModelInfo,
     find_model_by_id,
+    get_compatible_provider_formats,
     get_available_provider_ids,
     list_available_models,
 )
@@ -26,10 +27,13 @@ from src.core.api_format import (
     ApiFormatDefinition,
     detect_format_and_key_from_starlette,
 )
+from src.core.api_format.conversion import converter_registry
+from src.core.api_format.utils import is_cli_format
 from src.core.logger import logger
 from src.database import get_db
 from src.models.database import ApiKey, User
 from src.services.auth.service import AuthService
+from src.services.system.config import SystemConfigService
 
 router = APIRouter(tags=["System Catalog"])
 
@@ -39,6 +43,9 @@ _CLAUDE_FORMATS = [APIFormat.CLAUDE.value]
 _OPENAI_FORMATS = [APIFormat.OPENAI.value]
 _GEMINI_FORMATS = [APIFormat.GEMINI.value]
 
+# 所有非 CLI 格式（用于格式转换时的查询）
+_ALL_CHAT_FORMATS = [APIFormat.CLAUDE.value, APIFormat.OPENAI.value, APIFormat.GEMINI.value]
+
 
 def _extract_api_key_from_request(
     request: Request, definition: ApiFormatDefinition
@@ -90,6 +97,52 @@ def _get_formats_for_api(api_format: str) -> list[str]:
         return _OPENAI_FORMATS
 
 
+def _is_format_conversion_enabled(db: Session) -> bool:
+    """检查全局格式转换开关"""
+    return bool(SystemConfigService.get_config(db, "format_conversion_enabled", False))
+
+
+def _get_convertible_formats(client_format: str, global_conversion_enabled: bool) -> list[str]:
+    """
+    获取客户端格式可转换到的所有目标格式列表
+
+    当启用格式转换时，返回所有可以转换的格式；
+    否则只返回客户端格式本身。
+    """
+    if not global_conversion_enabled:
+        return _get_formats_for_api(client_format)
+
+    client_format_upper = client_format.upper()
+
+    # CLI 格式不支持转换
+    if is_cli_format(client_format_upper):
+        return _get_formats_for_api(client_format)
+
+    # 收集所有可转换的格式
+    convertible_formats = []
+    for target_format in _ALL_CHAT_FORMATS:
+        # 相同格式始终可用
+        if target_format == client_format_upper:
+            convertible_formats.append(target_format)
+            continue
+
+        # 检查是否有双向转换器
+        if converter_registry.can_convert_full(client_format_upper, target_format, require_stream=False):
+            convertible_formats.append(target_format)
+
+    return convertible_formats if convertible_formats else _get_formats_for_api(client_format)
+
+
+def _flatten_provider_formats(provider_to_formats: dict[str, set[str]]) -> list[str]:
+    """合并 Provider 格式映射为唯一格式列表"""
+    if not provider_to_formats:
+        return []
+    all_formats: set[str] = set()
+    for formats in provider_to_formats.values():
+        all_formats.update(formats)
+    return sorted(all_formats)
+
+
 def _build_empty_list_response(api_format: str) -> dict:
     """根据 API 格式构建空列表响应"""
     if api_format == "claude":
@@ -454,17 +507,32 @@ async def list_models(
     # 构建访问限制
     restrictions = AccessRestrictions.from_api_key_and_user(key_record, user)
 
-    # 检查 API 格式限制
+    # 获取可用格式（包括可转换的格式）
-    formats = _get_formats_for_api(api_format)
+    global_conversion_enabled = _is_format_conversion_enabled(db)
-    formats, empty_response = _filter_formats_by_restrictions(formats, restrictions, api_format)
+    candidate_formats = _get_convertible_formats(api_format, global_conversion_enabled)
+    candidate_formats, empty_response = _filter_formats_by_restrictions(
+        candidate_formats, restrictions, api_format
+    )
     if empty_response is not None:
         return empty_response
 
-    available_provider_ids = get_available_provider_ids(db, formats)
+    provider_to_formats = get_compatible_provider_formats(
+        db, api_format, candidate_formats, global_conversion_enabled
+    )
+    formats = _flatten_provider_formats(provider_to_formats)
+
+    available_provider_ids = get_available_provider_ids(db, formats, provider_to_formats)
     if not available_provider_ids:
         return _build_empty_list_response(api_format)
 
-    models = await list_available_models(db, available_provider_ids, formats, restrictions)
+    models = await list_available_models(
+        db,
+        available_provider_ids,
+        formats,
+        restrictions,
+        provider_to_formats=provider_to_formats,
+        client_format=api_format,
+    )
     logger.debug(f"[Models] 返回 {len(models)} 个模型")
 
     if api_format == "claude":
@@ -543,14 +611,28 @@ async def retrieve_model(
     # 构建访问限制
     restrictions = AccessRestrictions.from_api_key_and_user(key_record, user)
 
-    # 检查 API 格式限制
+    # 获取可用格式（包括可转换的格式）
-    formats = _get_formats_for_api(api_format)
+    global_conversion_enabled = _is_format_conversion_enabled(db)
-    formats, _ = _filter_formats_by_restrictions(formats, restrictions, api_format)
+    candidate_formats = _get_convertible_formats(api_format, global_conversion_enabled)
+    candidate_formats, _ = _filter_formats_by_restrictions(
+        candidate_formats, restrictions, api_format
+    )
+    provider_to_formats = get_compatible_provider_formats(
+        db, api_format, candidate_formats, global_conversion_enabled
+    )
+    formats = _flatten_provider_formats(provider_to_formats)
     if not formats:
         return _build_404_response(model_id, api_format)
 
-    available_provider_ids = get_available_provider_ids(db, formats)
+    available_provider_ids = get_available_provider_ids(db, formats, provider_to_formats)
-    model_info = find_model_by_id(db, model_id, available_provider_ids, formats, restrictions)
+    model_info = find_model_by_id(
+        db,
+        model_id,
+        available_provider_ids,
+        formats,
+        restrictions,
+        provider_to_formats=provider_to_formats,
+    )
 
     if not model_info:
         return _build_404_response(model_id, api_format)
@@ -614,18 +696,32 @@ async def list_models_gemini(
     # 构建访问限制
     restrictions = AccessRestrictions.from_api_key_and_user(key_record, user)
 
-    # 检查 API 格式限制
+    # 获取可用格式（包括可转换的格式）
-    formats, empty_response = _filter_formats_by_restrictions(
+    global_conversion_enabled = _is_format_conversion_enabled(db)
-        _GEMINI_FORMATS, restrictions, "gemini"
+    candidate_formats = _get_convertible_formats("gemini", global_conversion_enabled)
+    candidate_formats, empty_response = _filter_formats_by_restrictions(
+        candidate_formats, restrictions, "gemini"
     )
     if empty_response is not None:
         return empty_response
 
-    available_provider_ids = get_available_provider_ids(db, formats)
+    provider_to_formats = get_compatible_provider_formats(
+        db, "gemini", candidate_formats, global_conversion_enabled
+    )
+    formats = _flatten_provider_formats(provider_to_formats)
+
+    available_provider_ids = get_available_provider_ids(db, formats, provider_to_formats)
     if not available_provider_ids:
         return {"models": []}
 
-    models = await list_available_models(db, available_provider_ids, formats, restrictions)
+    models = await list_available_models(
+        db,
+        available_provider_ids,
+        formats,
+        restrictions,
+        provider_to_formats=provider_to_formats,
+        client_format="gemini",
+    )
     logger.debug(f"[Models] 返回 {len(models)} 个模型")
     response = _build_gemini_list_response(models, page_size, page_token)
     logger.debug(f"[Models] Gemini 响应: {response}")
@@ -681,14 +777,27 @@ async def get_model_gemini(
     # 构建访问限制
     restrictions = AccessRestrictions.from_api_key_and_user(key_record, user)
 
-    # 检查 API 格式限制
+    # 获取可用格式（包括可转换的格式）
-    formats, _ = _filter_formats_by_restrictions(_GEMINI_FORMATS, restrictions, "gemini")
+    global_conversion_enabled = _is_format_conversion_enabled(db)
+    candidate_formats = _get_convertible_formats("gemini", global_conversion_enabled)
+    candidate_formats, _ = _filter_formats_by_restrictions(
+        candidate_formats, restrictions, "gemini"
+    )
+    provider_to_formats = get_compatible_provider_formats(
+        db, "gemini", candidate_formats, global_conversion_enabled
+    )
+    formats = _flatten_provider_formats(provider_to_formats)
     if not formats:
         return _build_404_response(model_id, "gemini")
 
-    available_provider_ids = get_available_provider_ids(db, formats)
+    available_provider_ids = get_available_provider_ids(db, formats, provider_to_formats)
     model_info = find_model_by_id(
-        db, model_id, available_provider_ids, formats, restrictions
+        db,
+        model_id,
+        available_provider_ids,
+        formats,
+        restrictions,
+        provider_to_formats=provider_to_formats,
     )
 
     if not model_info:

src/core/api_format/conversion/compatibility.py

@@ -70,6 +70,8 @@ def is_format_compatible(
         return False, False, "端点未配置格式转换"
 
     config = endpoint_format_acceptance_config
+    if not isinstance(config, dict):
+        return False, False, "端点格式配置无效"
     if not config.get("enabled", False):
         return False, False, "端点格式转换未启用"
 

src/services/cache/model_cache.py

@@ -247,6 +247,11 @@ class ModelCacheService:
             await CacheService.delete(f"global_model:name:{name}")
             # 同时清除 resolve 缓存，因为 GlobalModel.name 也是一个 resolve key
             await CacheService.delete(f"global_model:resolve:{name}")
+        # 全量清除 resolve 缓存，确保映射规则变更后不命中旧缓存
+        try:
+            await CacheService.delete_pattern("global_model:resolve:*")
+        except Exception as e:
+            logger.error(f"GlobalModel resolve 缓存清除失败，可能导致映射不一致: {e}")
         logger.debug(f"GlobalModel 缓存已清除: {global_model_id}")
 
     @staticmethod
@@ -260,10 +265,11 @@ class ModelCacheService:
         1. 检查缓存
         2. 直接匹配 GlobalModel.name
         3. 通过 provider_model_name 匹配（查询 Model 表）
+        4. 通过 provider_model_mappings 匹配（查询 Model 表）
+        5. 通过 GlobalModel.config.model_mappings 匹配（支持正则）
 
-        注意：此方法不使用 provider_model_mappings 进行全局解析。
+        注意：provider_model_mappings 是 Provider 级别的映射配置，可能存在跨 Provider 冲突；
-        provider_model_mappings 是 Provider 级别的映射配置，只在特定 Provider 上下文中生效，
+        如匹配到多个 GlobalModel，将记录告警并选择第一个匹配结果。
-        由 resolve_provider_model() 处理。
 
         Args:
             db: 数据库会话
@@ -319,10 +325,7 @@ class ModelCacheService:
                 logger.debug(f"GlobalModel 已缓存(映射解析-直接匹配): {normalized_name}")
                 return global_model
 
-            # 3. 通过 provider_model_name 匹配（不考虑 provider_model_mappings）
+            # 3. 通过 provider_model_name 匹配
-            # 重要：provider_model_mappings 是 Provider 级别的映射配置，只在特定 Provider 上下文中生效
-            # 全局解析不应该受到某个 Provider 映射配置的影响
-            # 例如：Provider A 把 "haiku" 映射到 "sonnet"，不应该影响 Provider B 的 "haiku" 解析
             from src.models.database import Provider
 
             models_with_global = (
@@ -375,7 +378,123 @@ class ModelCacheService:
                 )
                 return result_global_model
 
-            # 4. 完全未找到
+            # 4. 通过 provider_model_mappings 匹配
+            models_with_mappings = (
+                db.query(Model, GlobalModel)
+                .join(Provider, Model.provider_id == Provider.id)
+                .join(GlobalModel, Model.global_model_id == GlobalModel.id)
+                .filter(
+                    Provider.is_active == True,
+                    Model.is_active == True,
+                    GlobalModel.is_active == True,
+                    Model.provider_model_mappings.isnot(None),
+                )
+                .all()
+            )
+
+            mapping_matched_global_models: List[GlobalModel] = []
+            mapping_seen_ids: set[str] = set()
+            for model, gm in models_with_mappings:
+                raw_mappings = model.provider_model_mappings
+                if not isinstance(raw_mappings, list):
+                    continue
+                for raw in raw_mappings:
+                    if not isinstance(raw, dict):
+                        continue
+                    name = raw.get("name")
+                    if not isinstance(name, str):
+                        continue
+                    if name.strip() != normalized_name:
+                        continue
+                    if gm.id not in mapping_seen_ids:
+                        mapping_seen_ids.add(gm.id)
+                        mapping_matched_global_models.append(gm)
+                        logger.debug(
+                            f"模型名称 '{normalized_name}' 通过 provider_model_mappings 匹配到 "
+                            f"GlobalModel: {gm.name} (Model: {model.id[:8]}...)"
+                        )
+                    break
+
+            if mapping_matched_global_models:
+                resolution_method = "provider_model_mappings"
+
+                if len(mapping_matched_global_models) > 1:
+                    model_names = [gm.name for gm in mapping_matched_global_models if gm.name]
+                    logger.warning(
+                        f"模型映射冲突: 名称 '{normalized_name}' 匹配到多个不同的 GlobalModel: "
+                        f"{', '.join(model_names)}，使用第一个匹配结果"
+                    )
+                    model_mapping_conflict_total.inc()
+
+                # 按名称排序确保确定性
+                result_global_model = sorted(
+                    mapping_matched_global_models, key=lambda gm: gm.name or ""
+                )[0]
+                global_model_dict = ModelCacheService._global_model_to_dict(result_global_model)
+                await CacheService.set(
+                    cache_key, global_model_dict, ttl_seconds=ModelCacheService.CACHE_TTL
+                )
+                logger.debug(
+                    f"GlobalModel 已缓存(映射解析-{resolution_method}): "
+                    f"{normalized_name} -> {result_global_model.name}"
+                )
+                return result_global_model
+
+            # 5. 通过 GlobalModel.config.model_mappings 匹配（支持正则）
+            from sqlalchemy import func
+
+            from src.core.model_permissions import match_model_with_pattern
+
+            mapping_rows = (
+                db.query(GlobalModel)
+                .filter(
+                    GlobalModel.is_active == True,
+                    GlobalModel.config.isnot(None),
+                    GlobalModel.config["model_mappings"].isnot(None),
+                    func.jsonb_array_length(GlobalModel.config["model_mappings"]) > 0,
+                )
+                .all()
+            )
+
+            mapping_matches: List[GlobalModel] = []
+            for gm in mapping_rows:
+                config = gm.config or {}
+                mappings = config.get("model_mappings")
+                if not isinstance(mappings, list):
+                    continue
+                for pattern in mappings:
+                    if isinstance(pattern, str) and match_model_with_pattern(
+                        pattern, normalized_name
+                    ):
+                        mapping_matches.append(gm)
+                        break
+
+            if mapping_matches:
+                resolution_method = "model_mappings"
+
+                if len(mapping_matches) > 1:
+                    model_names = [gm.name for gm in mapping_matches if gm.name]
+                    logger.warning(
+                        f"模型映射冲突: 名称 '{normalized_name}' 匹配到多个不同的 GlobalModel: "
+                        f"{', '.join(model_names)}，使用第一个匹配结果"
+                    )
+                    model_mapping_conflict_total.inc()
+
+                # 按名称排序确保确定性
+                result_global_model = sorted(
+                    mapping_matches, key=lambda gm: gm.name or ""
+                )[0]
+                global_model_dict = ModelCacheService._global_model_to_dict(result_global_model)
+                await CacheService.set(
+                    cache_key, global_model_dict, ttl_seconds=ModelCacheService.CACHE_TTL
+                )
+                logger.debug(
+                    f"GlobalModel 已缓存(映射解析-{resolution_method}): "
+                    f"{normalized_name} -> {result_global_model.name}"
+                )
+                return result_global_model
+
+            # 6. 完全未找到
             resolution_method = "not_found"
             # 未找到匹配，缓存负结果
             await CacheService.set(