feat: 实现跨 API 格式自动转换功能

- 新增端点级 format_acceptance_config 配置，控制是否接受跨格式请求
- 重构 EndpointFormDialog 为卡片式布局，支持内联编辑和格式转换开关
- StreamProcessor 实现流式响应的跨格式转换，支持 OpenAI/Claude/Gemini 互转
- CacheAwareScheduler 按端点格式筛选候选，同格式优先于跨格式
- 健康度/熔断按 Provider 端点格式分桶，而非客户端请求格式
- 新增 format_conversion_total 和 format_conversion_duration_seconds 指标
- 新增全局配置 format_conversion_enabled 控制总开关
- Input 组件新增 size="sm" 尺寸选项

alembic/versions/20260121_1845_4b4c7b0df1a2_add_format_acceptance_config_to_provider_endpoints.py

@@ -0,0 +1,51 @@
+"""add_format_acceptance_config_to_provider_endpoints
+
+Revision ID: 4b4c7b0df1a2
+Revises: c868729753ad
+Create Date: 2026-01-21 18:45:00+00:00
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy import inspect
+
+
+# revision identifiers, used by Alembic.
+revision = "4b4c7b0df1a2"
+down_revision = "c868729753ad"
+branch_labels = None
+depends_on = None
+
+
+def table_exists(table_name: str) -> bool:
+    bind = op.get_bind()
+    inspector = inspect(bind)
+    return table_name in inspector.get_table_names()
+
+
+def column_exists(table_name: str, column_name: str) -> bool:
+    bind = op.get_bind()
+    inspector = inspect(bind)
+    columns = [col["name"] for col in inspector.get_columns(table_name)]
+    return column_name in columns
+
+
+def upgrade() -> None:
+    if not table_exists("provider_endpoints"):
+        return
+    if column_exists("provider_endpoints", "format_acceptance_config"):
+        return
+    op.add_column(
+        "provider_endpoints",
+        sa.Column("format_acceptance_config", sa.JSON(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    if not table_exists("provider_endpoints"):
+        return
+    if not column_exists("provider_endpoints", "format_acceptance_config"):
+        return
+    op.drop_column("provider_endpoints", "format_acceptance_config")
+

frontend/src/api/endpoints/endpoints.ts

@@ -1,5 +1,5 @@
 import client from '../client'
-import type { ProviderEndpoint, ProxyConfig, HeaderRule } from './types'
+import type { ProviderEndpoint, ProxyConfig, HeaderRule, FormatAcceptanceConfig } from './types'
 
 /**
  * 获取指定 Provider 的所有 Endpoints
@@ -32,6 +32,7 @@ export async function createEndpoint(
     is_active?: boolean
     config?: Record<string, any>
     proxy?: ProxyConfig | null
+    format_acceptance_config?: FormatAcceptanceConfig | null
   }
 ): Promise<ProviderEndpoint> {
   const response = await client.post(`/api/admin/endpoints/providers/${providerId}/endpoints`, data)
@@ -51,6 +52,7 @@ export async function updateEndpoint(
     is_active: boolean
     config: Record<string, any>
     proxy: ProxyConfig | null
+    format_acceptance_config: FormatAcceptanceConfig | null
   }>
 ): Promise<ProviderEndpoint> {
   const response = await client.put(`/api/admin/endpoints/${endpointId}`, data)

frontend/src/api/endpoints/types.ts

@@ -87,6 +87,16 @@ export interface HeaderRuleRename {
 
 export type HeaderRule = HeaderRuleSet | HeaderRuleDrop | HeaderRuleRename
 
+/**
+ * 格式接受策略配置
+ * 用于控制端点是否接受来自不同 API 格式的请求，并自动进行格式转换
+ */
+export interface FormatAcceptanceConfig {
+  enabled: boolean                // 是否启用格式转换
+  accept_formats?: string[]       // 白名单：接受哪些格式的请求
+  reject_formats?: string[]       // 黑名单：拒绝哪些格式（优先级高于白名单）
+}
+
 export interface ProviderEndpoint {
   id: string
   provider_id: string
@@ -100,6 +110,8 @@ export interface ProviderEndpoint {
   is_active: boolean
   config?: Record<string, any>
   proxy?: ProxyConfig | null
+  // 格式转换配置
+  format_acceptance_config?: FormatAcceptanceConfig | null
   total_keys: number
   active_keys: number
   created_at: string

frontend/src/components/ui/collapsible-content.vue

@@ -8,7 +8,7 @@ const props = defineProps<CollapsibleContentProps & { class?: string }>()
 <template>
   <CollapsibleContent
     v-bind="props"
-    :class="cn('overflow-hidden data-[state=closed]:animate-collapsible-up data-[state=open]:animate-collapsible-down', props.class)"
+    :class="cn('data-[state=closed]:overflow-hidden data-[state=closed]:animate-collapsible-up data-[state=open]:animate-collapsible-down', props.class)"
   >
     <slot />
   </CollapsibleContent>

frontend/src/components/ui/input.vue

@@ -75,6 +75,12 @@ interface Props {
   modelValue?: string | number
   class?: string
   autocomplete?: string
+  /**
+   * 输入框尺寸
+   * - 'default': 默认尺寸 (h-11, py-2)
+   * - 'sm': 小尺寸 (h-8, py-1)
+   */
+  size?: 'default' | 'sm'
   /**
    * 遮蔽显示内容（用于 API Key 等敏感信息）
    * 使用 CSS -webkit-text-security 实现，不会触发浏览器密码管理器
@@ -150,9 +156,16 @@ const autocompleteAttr = computed(() => {
   return props.autocomplete ?? 'off'
 })
 
+// 尺寸相关的样式
+const sizeClasses = {
+  default: 'h-11 py-2 px-4',
+  sm: 'h-8 py-1 px-3'
+}
+
 const inputClass = computed(() =>
   cn(
-    'flex h-11 w-full rounded-xl border border-border/60 bg-muted/50 px-4 py-2 text-sm ring-offset-background placeholder:text-muted-foreground focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-primary/40 focus-visible:border-primary/60 text-foreground transition-all',
+    'flex w-full rounded-xl border border-border/60 bg-muted/50 text-sm ring-offset-background placeholder:text-muted-foreground focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-primary/40 focus-visible:border-primary/60 text-foreground transition-all',
+    sizeClasses[props.size || 'default'],
     props.masked && 'pr-10',
     props.class
   )

src/api/admin/endpoints/routes.py

@@ -308,6 +308,7 @@ class AdminCreateProviderEndpointAdapter(AdminApiAdapter):
             is_active=True,
             config=self.endpoint_data.config,
             proxy=self.endpoint_data.proxy.model_dump() if self.endpoint_data.proxy else None,
+            format_acceptance_config=self.endpoint_data.format_acceptance_config,
             created_at=now,
             updated_at=now,
         )

src/api/handlers/base/chat_handler_base.py

@@ -248,6 +248,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
         self,
         source_model: str,
         provider_id: str,
+        api_format: Optional[str] = None,
     ) -> Optional[str]:
         """
         获取模型映射后的实际模型名
@@ -255,6 +256,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
         Args:
             source_model: 用户请求的模型名
             provider_id: Provider ID
+            api_format: Provider 侧 API 格式（用于过滤映射作用域，默认使用 handler FORMAT_ID）
 
         Returns:
             映射后的 provider_model_name，没有映射则返回 None
@@ -269,8 +271,9 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
             # 传入 api_key.id 作为 affinity_key，实现相同用户稳定选择同一映射
             # 传入 api_format 用于过滤适用的映射作用域
             affinity_key = self.api_key.id if self.api_key else None
+            effective_format = api_format or self.FORMAT_ID
             mapped_name = mapping.model.select_provider_model_name(
-                affinity_key, api_format=self.FORMAT_ID
+                affinity_key, api_format=effective_format
             )
             logger.debug(f"[Chat] 模型映射: {source_model} -> {mapped_name}")
             return mapped_name
@@ -297,6 +300,8 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
 
         # 创建类型安全的流式上下文
         ctx = StreamContext(model=model, api_format=api_format)
+        ctx.request_id = self.request_id
+        ctx.client_api_format = api_format.value if hasattr(api_format, "value") else str(api_format)
 
         # 创建更新状态的回调闭包（可以访问 ctx）
         def update_streaming_status() -> None:
@@ -430,12 +435,24 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
             provider_api_format=str(endpoint.api_format) if endpoint.api_format else None,
         )
 
+        # ctx.api_format 是枚举，需要取 value 作为字符串
+        _api_format_str = (
+            ctx.api_format.value if hasattr(ctx.api_format, "value") else str(ctx.api_format)
+        )
+        provider_api_format = ctx.provider_api_format or _api_format_str
+        client_api_format = ctx.client_api_format or _api_format_str
+        needs_conversion = (
+            bool(getattr(candidate, "needs_conversion", False)) if candidate else False
+        )
+        ctx.needs_conversion = needs_conversion
+
         # 获取模型映射（优先使用映射匹配到的模型，其次是 Provider 级别的映射）
         mapped_model = candidate.mapping_matched_model if candidate else None
         if not mapped_model:
             mapped_model = await self._get_mapped_model(
                 source_model=ctx.model,
                 provider_id=str(provider.id),
+                api_format=provider_api_format,
             )
 
         # 应用模型映射到请求体
@@ -445,8 +462,18 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
         else:
             request_body = dict(original_request_body)
 
-        # 准备发送给 Provider 的请求体
-        request_body = self.prepare_provider_request_body(request_body)
+        # 跨格式：先做请求体转换（严格模式，失败触发 failover）
+        if needs_conversion:
+            from src.core.api_format import converter_registry
+
+            request_body = converter_registry.convert_request_strict(
+                request_body,
+                str(client_api_format),
+                str(provider_api_format),
+            )
+        else:
+            # 同格式：按原逻辑做轻量清理（子类可覆盖以移除不需要的字段）
+            request_body = self.prepare_provider_request_body(request_body)
 
         # 构建请求
         provider_payload, provider_headers = self._request_builder.build(
@@ -663,6 +690,12 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
             nonlocal provider_request_headers, provider_request_body, mapped_model_result
 
             provider_name = str(provider.name)
+            provider_api_format = str(endpoint.api_format or api_format)
+            # 客户端格式（与流式处理保持一致的命名）
+            client_api_format = (
+                api_format.value if hasattr(api_format, "value") else str(api_format)
+            )
+            needs_conversion = bool(getattr(candidate, "needs_conversion", False))
 
             # 获取模型映射（优先使用映射匹配到的模型，其次是 Provider 级别的映射）
             mapped_model = candidate.mapping_matched_model if candidate else None
@@ -670,6 +703,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
                 mapped_model = await self._get_mapped_model(
                     source_model=model,
                     provider_id=str(provider.id),
+                    api_format=provider_api_format,
                 )
 
             # 应用模型映射
@@ -679,8 +713,18 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
             else:
                 request_body = dict(original_request_body)
 
-            # 准备发送给 Provider 的请求体（子类可覆盖以移除不需要的字段）
-            request_body = self.prepare_provider_request_body(request_body)
+            # 跨格式：先做请求体转换（严格模式，失败触发 failover）
+            if needs_conversion:
+                from src.core.api_format import converter_registry
+
+                request_body = converter_registry.convert_request_strict(
+                    request_body,
+                    client_api_format,
+                    provider_api_format,
+                )
+            else:
+                # 同格式：按原逻辑做轻量清理（子类可覆盖以移除不需要的字段）
+                request_body = self.prepare_provider_request_body(request_body)
 
             # 构建请求
             provider_payload, provider_hdrs = self._request_builder.build(
@@ -789,9 +833,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
                         raw_content = repr(resp.content[:500]) if resp.content else "(empty)"
                     except Exception:
                         raw_content = "(unable to read)"
-                logger.error(
-                    f"[{self.request_id}] 无法解析响应 JSON: {e}, 原始内容: {raw_content}"
-                )
+                logger.error(f"[{self.request_id}] 无法解析响应 JSON: {e}, 原始内容: {raw_content}")
                 # 判断错误类型，生成友好的客户端错误消息（不暴露提供商信息）
                 if raw_content == "(empty)" or not raw_content.strip():
                     client_message = "上游服务返回了空响应"
@@ -808,7 +850,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
 
             # 检查响应体中的嵌套错误（HTTP 200 但响应体包含错误）
             if isinstance(response_json, dict):
-                parser = get_parser_for_format(api_format)
+                parser = get_parser_for_format(provider_api_format)
                 if parser.is_error_response(response_json):
                     parsed = parser.parse_response(response_json, 200)
                     logger.warning(
@@ -825,6 +867,16 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
                         error_status=parsed.error_type,
                     )
 
+            # 跨格式：响应转换回 client_format（严格模式，失败触发 failover）
+            if needs_conversion and isinstance(response_json, dict):
+                from src.core.api_format import converter_registry
+
+                response_json = converter_registry.convert_response_strict(
+                    response_json,
+                    provider_api_format,
+                    str(api_format),
+                )
+
             return response_json if isinstance(response_json, dict) else {}
 
         try:

src/api/handlers/base/stream_context.py

@@ -76,6 +76,7 @@ class StreamContext:
 
     # 格式转换信息（CLI handler 需要）
     client_api_format: str = ""
+    needs_conversion: bool = False  # 是否需要跨格式转换（由 handler 层设置）
 
     # Provider 响应元数据（CLI handler 需要）
     response_metadata: Dict[str, Any] = field(default_factory=dict)
@@ -118,6 +119,7 @@ class StreamContext:
         self.final_usage = None
         self.final_response = None
         self.stream_conversion_state = None
+        self.needs_conversion = False
 
     @property
     def collected_text(self) -> str:

src/api/handlers/base/stream_processor.py

@@ -328,14 +328,12 @@ class StreamProcessor:
             raise
         except (OSError, IOError) as e:
             # 网络 I/O 异常：记录警告，可能需要重试
-            logger.warning(
-                f"  [{self.request_id}] 预读流时发生网络异常: {type(e).__name__}: {e}"
-            )
+            logger.warning(f"  [{self.request_id}] 预读流时发生网络异常: {type(e).__name__}: {e}")
         except Exception as e:
             # 未预期的严重异常：记录错误并重新抛出，避免掩盖问题
             logger.error(
                 f"  [{self.request_id}] 预读流时发生严重异常: {type(e).__name__}: {e}",
-                exc_info=True
+                exc_info=True,
             )
             raise
 
@@ -374,19 +372,269 @@ class StreamProcessor:
             # 使用增量解码器处理跨 chunk 的 UTF-8 字符
             decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
 
-            # 处理预读数据
-            if prefetched_chunks:
-                for chunk in prefetched_chunks:
-                    # 记录首字时间 (TTFB) - 在 yield 之前记录
-                    if start_time is not None:
-                        ctx.record_first_byte_time(start_time)
-                        start_time = None  # 只记录一次
-                    # 首次输出前触发 streaming 回调（确保 TTFB 已写入 ctx）
-                    if not streaming_started and self.on_streaming_start:
-                        self.on_streaming_start()
-                        streaming_started = True
+            # ctx.api_format 可能是 APIFormat 枚举，需要取 value
+            _api_format_str = (
+                ctx.api_format.value
+                if hasattr(ctx.api_format, "value")
+                else str(ctx.api_format or "")
+            )
+            client_format = (ctx.client_api_format or _api_format_str).upper()
+            provider_format = (ctx.provider_api_format or _api_format_str).upper()
+            # 使用 handler 层预计算的 needs_conversion（由 candidate 决定）
+            needs_conversion = ctx.needs_conversion
 
-                    # 把原始数据转发给客户端
+            # 安全检查：needs_conversion 为 True 时，provider_format 必须有值
+            if needs_conversion and not provider_format:
+                logger.warning(
+                    f"[{self.request_id}] needs_conversion=True 但 provider_format 为空，回退到透传模式"
+                )
+                needs_conversion = False
+
+            def _mark_stream_started() -> None:
+                nonlocal start_time, streaming_started
+                # 记录首字时间 (TTFB) - 在 yield 之前记录
+                if start_time is not None:
+                    ctx.record_first_byte_time(start_time)
+                    start_time = None  # 只记录一次
+                # 首次输出前触发 streaming 回调（确保 TTFB 已写入 ctx）
+                if not streaming_started and self.on_streaming_start:
+                    self.on_streaming_start()
+                    streaming_started = True
+
+            def _build_stream_error_payload(message: str) -> dict:
+                if client_format.startswith("OPENAI"):
+                    return {
+                        "error": {
+                            "message": message,
+                            "type": "format_conversion_error",
+                        }
+                    }
+                # Claude 及其他格式使用统一的错误结构
+                return {
+                    "type": "error",
+                    "error": {
+                        "type": "format_conversion_error",
+                        "message": message,
+                    },
+                }
+
+            # 处理预读数据
+            if needs_conversion:
+                # 延迟导入：仅在需要转换时加载转换器模块
+                from src.core.api_format import (
+                    GeminiStreamConversionState,
+                    StreamConversionState,
+                    converter_registry,
+                )
+
+                # 初始化流式转换状态（首次使用时，根据 Provider 格式选择状态类）
+                if ctx.stream_conversion_state is None:
+                    if provider_format == "GEMINI":
+                        ctx.stream_conversion_state = GeminiStreamConversionState(
+                            model=ctx.mapped_model or ctx.model or "",
+                            message_id=ctx.response_id or ctx.request_id or "",
+                        )
+                    else:
+                        ctx.stream_conversion_state = StreamConversionState(
+                            model=ctx.mapped_model or ctx.model or "",
+                            message_id=ctx.response_id or ctx.request_id or "",
+                        )
+
+                skip_next_blank_line = False
+                empty_yield_count = 0  # 空转计数（防护异常情况）
+
+                def _emit_converted_line(normalized_line: str) -> list[bytes]:
+                    nonlocal skip_next_blank_line
+
+                    # 空行：事件分隔符（避免重复输出）
+                    if normalized_line == "":
+                        if skip_next_blank_line:
+                            skip_next_blank_line = False
+                            return []
+                        return [b"\n"]
+
+                    # 丢弃 Provider 的 event 行，避免泄漏/污染目标格式
+                    if normalized_line.startswith("event:"):
+                        return []
+
+                    # OpenAI done 信号
+                    if (
+                        normalized_line.startswith("data:")
+                        and normalized_line[5:].strip() == "[DONE]"
+                    ):
+                        skip_next_blank_line = True
+                        if client_format.startswith("OPENAI"):
+                            return [b"data: [DONE]\n\n"]
+                        return []
+
+                    # 非 data 行：在跨格式场景下统一丢弃（避免泄露 Provider 格式细节）
+                    if not normalized_line.startswith("data:"):
+                        return []
+
+                    data_content = normalized_line[5:].strip()
+                    # Gemini 可能包含 JSON 数组包装符，直接忽略
+                    if data_content in ("", "[", "]", ","):
+                        return []
+
+                    try:
+                        data_obj = json.loads(data_content)
+                    except json.JSONDecodeError:
+                        # 跨格式转换时，JSON 解析失败应跳过而不是透传（避免泄漏 Provider 格式）
+                        logger.warning(
+                            f"[{self.request_id}] JSON 解析失败，跳过该行: {data_content[:100]}"
+                        )
+                        return []
+
+                    if not isinstance(data_obj, dict):
+                        return []
+
+                    try:
+                        converted_events = converter_registry.convert_stream_chunk_strict(
+                            data_obj,
+                            provider_format,
+                            client_format,
+                            state=ctx.stream_conversion_state,
+                        )
+                    except Exception as conv_err:
+                        # 首字节后无法 failover：输出目标格式错误事件并终止流
+                        # 使用 502 表示上游返回了非预期格式（Bad Gateway）
+                        ctx.status_code = 502
+                        ctx.error_message = "format_conversion_failed"
+                        # 日志记录完整错误（内部排查），客户端只返回脱敏消息
+                        logger.warning(f"[{self.request_id}] 流式格式转换失败: {conv_err}")
+                        payload = _build_stream_error_payload("响应格式转换失败，请稍后重试")
+                        error_bytes = f"data: {json.dumps(payload, ensure_ascii=False)}\n\n".encode(
+                            "utf-8"
+                        )
+                        done_bytes = (
+                            b"data: [DONE]\n\n" if client_format.startswith("OPENAI") else b""
+                        )
+                        return [error_bytes, done_bytes]
+
+                    skip_next_blank_line = True
+                    out: list[bytes] = []
+                    for evt in converted_events:
+                        out.append(
+                            f"data: {json.dumps(evt, ensure_ascii=False)}\n\n".encode("utf-8")
+                        )
+                    return out
+
+                # 统一处理 prefetched + iterator
+                if prefetched_chunks:
+                    for chunk in prefetched_chunks:
+                        buffer += chunk
+                        while b"\n" in buffer:
+                            line_bytes, buffer = buffer.split(b"\n", 1)
+                            try:
+                                line = decoder.decode(line_bytes + b"\n", False)
+                            except Exception as e:
+                                logger.warning(
+                                    f"[{self.request_id}] UTF-8 解码失败: {e}, bytes={line_bytes[:50]!r}"
+                                )
+                                line = ""
+
+                            if line:
+                                self._process_line(ctx, sse_parser, line)
+                            normalized_line = line.rstrip("\r\n") if line else ""
+                            out_chunks = _emit_converted_line(normalized_line)
+                            if not out_chunks:
+                                empty_yield_count += 1
+                                if empty_yield_count == StreamDefaults.MAX_EMPTY_YIELDS_WARNING:
+                                    logger.warning(
+                                        f"[{self.request_id}] 流式转换连续 {empty_yield_count} 次空产出"
+                                    )
+                            else:
+                                empty_yield_count = 0
+                            for out in out_chunks:
+                                if not out:
+                                    continue
+                                _mark_stream_started()
+                                yield out
+                                # 转换失败：已输出 error（可能还包含 done），直接终止
+                                if ctx.error_message == "format_conversion_failed":
+                                    return
+
+                async for chunk in byte_iterator:
+                    buffer += chunk
+                    while b"\n" in buffer:
+                        line_bytes, buffer = buffer.split(b"\n", 1)
+                        try:
+                            line = decoder.decode(line_bytes + b"\n", False)
+                        except Exception as e:
+                            logger.warning(
+                                f"[{self.request_id}] UTF-8 解码失败: {e}, bytes={line_bytes[:50]!r}"
+                            )
+                            line = ""
+
+                        if line:
+                            self._process_line(ctx, sse_parser, line)
+                        normalized_line = line.rstrip("\r\n") if line else ""
+                        out_chunks = _emit_converted_line(normalized_line)
+                        if not out_chunks:
+                            empty_yield_count += 1
+                            if empty_yield_count == StreamDefaults.MAX_EMPTY_YIELDS_WARNING:
+                                logger.warning(
+                                    f"[{self.request_id}] 流式转换连续 {empty_yield_count} 次空产出"
+                                )
+                        else:
+                            empty_yield_count = 0
+                        for out in out_chunks:
+                            if not out:
+                                continue
+                            _mark_stream_started()
+                            yield out
+                            if ctx.error_message == "format_conversion_failed":
+                                return
+
+                # 处理剩余缓冲区（needs_conversion 分支内，可复用 _emit_converted_line）
+                if buffer:
+                    try:
+                        line = decoder.decode(buffer, True)
+                    except Exception as e:
+                        logger.warning(
+                            f"[{self.request_id}] 处理剩余缓冲区失败: {e}, bytes={buffer[:50]!r}"
+                        )
+                        line = ""
+                    if line:
+                        self._process_line(ctx, sse_parser, line)
+                        normalized_line = line.rstrip("\r\n")
+                        out_chunks = _emit_converted_line(normalized_line)
+                        for out in out_chunks:
+                            if out:
+                                _mark_stream_started()
+                                yield out
+                                # 转换失败：已输出 error，直接终止
+                                if ctx.error_message == "format_conversion_failed":
+                                    return
+
+            else:
+                if prefetched_chunks:
+                    for chunk in prefetched_chunks:
+                        _mark_stream_started()
+                        yield chunk
+
+                        buffer += chunk
+                        # 处理缓冲区中的完整行
+                        while b"\n" in buffer:
+                            line_bytes, buffer = buffer.split(b"\n", 1)
+                            try:
+                                # 使用增量解码器，可以正确处理跨 chunk 的多字节字符
+                                line = decoder.decode(line_bytes + b"\n", False)
+                                self._process_line(ctx, sse_parser, line)
+                            except Exception as e:
+                                # 解码失败，记录警告但继续处理
+                                logger.warning(
+                                    f"[{self.request_id}] UTF-8 解码失败: {e}, "
+                                    f"bytes={line_bytes[:50]!r}"
+                                )
+                                continue
+
+            # 处理剩余的流数据
+            if not needs_conversion:
+                async for chunk in byte_iterator:
+                    _mark_stream_started()
+
+                    # 原始数据透传
                     yield chunk
 
                     buffer += chunk
@@ -405,46 +653,15 @@ class StreamProcessor:
                             )
                             continue
 
-            # 处理剩余的流数据
-            async for chunk in byte_iterator:
-                # 记录首字时间 (TTFB) - 在 yield 之前记录（如果预读数据为空）
-                if start_time is not None:
-                    ctx.record_first_byte_time(start_time)
-                    start_time = None  # 只记录一次
-                # 首次输出前触发 streaming 回调（确保 TTFB 已写入 ctx）
-                if not streaming_started and self.on_streaming_start:
-                    self.on_streaming_start()
-                    streaming_started = True
-
-                # 原始数据透传
-                yield chunk
-
-                buffer += chunk
-                # 处理缓冲区中的完整行
-                while b"\n" in buffer:
-                    line_bytes, buffer = buffer.split(b"\n", 1)
-                    try:
-                        # 使用增量解码器，可以正确处理跨 chunk 的多字节字符
-                        line = decoder.decode(line_bytes + b"\n", False)
-                        self._process_line(ctx, sse_parser, line)
-                    except Exception as e:
-                        # 解码失败，记录警告但继续处理
-                        logger.warning(
-                            f"[{self.request_id}] UTF-8 解码失败: {e}, "
-                            f"bytes={line_bytes[:50]!r}"
-                        )
-                        continue
-
-            # 处理剩余的缓冲区数据（如果有未完成的行）
-            if buffer:
+            # 处理剩余的缓冲区数据（仅非转换分支，转换分支已在内部处理）
+            if not needs_conversion and buffer:
                 try:
                     # 使用 final=True 处理最后的不完整字符
                     line = decoder.decode(buffer, True)
                     self._process_line(ctx, sse_parser, line)
                 except Exception as e:
                     logger.warning(
-                        f"[{self.request_id}] 处理剩余缓冲区失败: {e}, "
-                        f"bytes={buffer[:50]!r}"
+                        f"[{self.request_id}] 处理剩余缓冲区失败: {e}, bytes={buffer[:50]!r}"
                     )
 
             # 处理剩余事件

src/config/constants.py

@@ -55,6 +55,11 @@ class StreamDefaults:
     # 3. 不会占用过多内存
     MAX_PREFETCH_BYTES = 64 * 1024  # 64KB
 
+    # 流式转换空产出告警阈值
+    # 连续这么多次空行/非 data 行后记录警告日志
+    # 50 次约等于 50 行非 data SSE 数据，足够覆盖正常事件头
+    MAX_EMPTY_YIELDS_WARNING = 50
+
 
 class RPMDefaults:
     """RPM（每分钟请求数）限制默认值

src/core/api_format/conversion/registry.py

@@ -13,9 +13,12 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
+import time
+from contextlib import contextmanager
+from typing import TYPE_CHECKING, Any, Dict, Generator, Optional, Tuple, Union
 
 from src.core.logger import logger
+from src.core.metrics import format_conversion_duration_seconds, format_conversion_total
 
 from .exceptions import FormatConversionError
 
@@ -23,6 +26,35 @@ if TYPE_CHECKING:
     from .state import GeminiStreamConversionState, StreamConversionState
 
 
+@contextmanager
+def _track_conversion_metrics(
+    direction: str, source: str, target: str
+) -> Generator[None, None, None]:
+    """
+    跟踪转换指标的上下文管理器
+
+    Args:
+        direction: 转换方向（request/response/stream）
+        source: 源格式（大写）
+        target: 目标格式（大写）
+
+    Yields:
+        None - 执行转换逻辑
+    """
+    start = time.perf_counter()
+    status = "success"
+    try:
+        yield
+    except Exception:
+        status = "error"
+        raise
+    finally:
+        format_conversion_total.labels(direction, source, target, status).inc()
+        format_conversion_duration_seconds.labels(direction, source, target).observe(
+            time.perf_counter() - start
+        )
+
+
 class FormatConverterRegistry:
     """
     格式转换器注册表
@@ -118,7 +150,9 @@ class FormatConverterRegistry:
             logger.debug(f"[ConverterRegistry] 请求转换成功: {source_format} -> {target_format}")
             return converted
         except Exception as e:
-            logger.error(f"[ConverterRegistry] 请求转换失败: {source_format} -> {target_format}: {e}")
+            logger.error(
+                f"[ConverterRegistry] 请求转换失败: {source_format} -> {target_format}: {e}"
+            )
             return request
 
     def convert_response(
@@ -160,7 +194,9 @@ class FormatConverterRegistry:
             logger.debug(f"[ConverterRegistry] 响应转换成功: {source_format} -> {target_format}")
             return converted
         except Exception as e:
-            logger.error(f"[ConverterRegistry] 响应转换失败: {source_format} -> {target_format}: {e}")
+            logger.error(
+                f"[ConverterRegistry] 响应转换失败: {source_format} -> {target_format}: {e}"
+            )
             return response
 
     def convert_stream_chunk(
@@ -196,14 +232,16 @@ class FormatConverterRegistry:
                 result: list[Dict[str, Any]] = converter.convert_stream_chunk(chunk, state)
                 return result
             except Exception as e:
-                logger.error(f"[ConverterRegistry] 流式块转换失败: {source_format} -> {target_format}: {e}")
+                logger.error(
+                    f"[ConverterRegistry] 流式块转换失败: {source_format} -> {target_format}: {e}"
+                )
                 return [chunk]
 
         # 降级到普通响应转换（作为单个事件返回）
         if hasattr(converter, "convert_response"):
             try:
-                result = converter.convert_response(chunk)
-                return [result]
+                converted: Dict[str, Any] = converter.convert_response(chunk)
+                return [converted]
             except Exception:
                 return [chunk]
 
@@ -284,8 +322,11 @@ class FormatConverterRegistry:
         Raises:
             FormatConversionError: 转换失败时抛出
         """
+        source_upper = source_format.upper()
+        target_upper = target_format.upper()
+
         # 同格式无需转换
-        if source_format.upper() == target_format.upper():
+        if source_upper == target_upper:
             return request
 
         converter = self.get_converter(source_format, target_format)
@@ -293,16 +334,19 @@ class FormatConverterRegistry:
             raise FormatConversionError(source_format, target_format, "未找到转换器")
 
         if not hasattr(converter, "convert_request"):
-            raise FormatConversionError(source_format, target_format, "转换器缺少 convert_request 方法")
+            raise FormatConversionError(
+                source_format, target_format, "转换器缺少 convert_request 方法"
+            )
 
-        try:
-            converted: Dict[str, Any] = converter.convert_request(request)
-            logger.debug(f"[ConverterRegistry] 请求转换成功: {source_format} -> {target_format}")
-            return converted
-        except FormatConversionError:
-            raise
-        except Exception as e:
-            raise FormatConversionError(source_format, target_format, str(e)) from e
+        with _track_conversion_metrics("request", source_upper, target_upper):
+            try:
+                converted: Dict[str, Any] = converter.convert_request(request)
+                logger.debug(f"[ConverterRegistry] 请求转换成功: {source_format} -> {target_format}")
+                return converted
+            except FormatConversionError:
+                raise
+            except Exception as e:
+                raise FormatConversionError(source_format, target_format, str(e)) from e
 
     def convert_response_strict(
         self,
@@ -316,7 +360,10 @@ class FormatConverterRegistry:
         Raises:
             FormatConversionError: 转换失败时抛出
         """
-        if source_format.upper() == target_format.upper():
+        source_upper = source_format.upper()
+        target_upper = target_format.upper()
+
+        if source_upper == target_upper:
             return response
 
         converter = self.get_converter(source_format, target_format)
@@ -324,16 +371,19 @@ class FormatConverterRegistry:
             raise FormatConversionError(source_format, target_format, "未找到转换器")
 
         if not hasattr(converter, "convert_response"):
-            raise FormatConversionError(source_format, target_format, "转换器缺少 convert_response 方法")
+            raise FormatConversionError(
+                source_format, target_format, "转换器缺少 convert_response 方法"
+            )
 
-        try:
-            converted: Dict[str, Any] = converter.convert_response(response)
-            logger.debug(f"[ConverterRegistry] 响应转换成功: {source_format} -> {target_format}")
-            return converted
-        except FormatConversionError:
-            raise
-        except Exception as e:
-            raise FormatConversionError(source_format, target_format, str(e)) from e
+        with _track_conversion_metrics("response", source_upper, target_upper):
+            try:
+                converted: Dict[str, Any] = converter.convert_response(response)
+                logger.debug(f"[ConverterRegistry] 响应转换成功: {source_format} -> {target_format}")
+                return converted
+            except FormatConversionError:
+                raise
+            except Exception as e:
+                raise FormatConversionError(source_format, target_format, str(e)) from e
 
     def convert_stream_chunk_strict(
         self,
@@ -357,7 +407,10 @@ class FormatConverterRegistry:
         Raises:
             FormatConversionError: 转换失败时抛出
         """
-        if source_format.upper() == target_format.upper():
+        source_upper = source_format.upper()
+        target_upper = target_format.upper()
+
+        if source_upper == target_upper:
             return [chunk]
 
         converter = self.get_converter(source_format, target_format)
@@ -369,13 +422,16 @@ class FormatConverterRegistry:
                 source_format, target_format, "转换器缺少 convert_stream_chunk 方法"
             )
 
-        try:
-            result: list[Dict[str, Any]] = converter.convert_stream_chunk(chunk, state)
-            return result
-        except FormatConversionError:
-            raise
-        except Exception as e:
-            raise FormatConversionError(source_format, target_format, f"流式块转换失败: {e}") from e
+        with _track_conversion_metrics("stream", source_upper, target_upper):
+            try:
+                result: list[Dict[str, Any]] = converter.convert_stream_chunk(chunk, state)
+                return result
+            except FormatConversionError:
+                raise
+            except Exception as e:
+                raise FormatConversionError(
+                    source_format, target_format, f"流式块转换失败: {e}"
+                ) from e
 
 
 # 全局单例
@@ -387,4 +443,3 @@ __all__ = [
     "converter_registry",
     "FormatConversionError",
 ]
-

src/core/metrics.py

@@ -65,3 +65,18 @@ model_mapping_conflict_total = Counter(
     "model_mapping_conflict_total",
     "Total number of mapping conflicts detected (same name maps to multiple GlobalModels)",
 )
+
+# ==================== API 格式转换 ====================
+
+format_conversion_total = Counter(
+    "format_conversion_total",
+    "Total number of format conversions",
+    ["direction", "source_format", "target_format", "status"],  # status: success/error
+)
+
+format_conversion_duration_seconds = Histogram(
+    "format_conversion_duration_seconds",
+    "Duration of format conversions in seconds",
+    ["direction", "source_format", "target_format"],
+    buckets=[0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0],
+)

src/models/database.py

@@ -278,7 +278,9 @@ class Usage(Base):
     request_id = Column(String(100), unique=True, index=True, nullable=False)
     provider_name = Column(String(100), nullable=False)  # Provider 名称（非外键）
     model = Column(String(100), nullable=False)
-    target_model = Column(String(100), nullable=True, comment="映射后的目标模型名（若无映射则为空）")
+    target_model = Column(
+        String(100), nullable=True, comment="映射后的目标模型名（若无映射则为空）"
+    )
 
     # Provider 侧追踪信息（记录最终成功的 Provider/Endpoint/Key）
     provider_id = Column(String(36), ForeignKey("providers.id", ondelete="SET NULL"), nullable=True)
@@ -465,7 +467,9 @@ class LDAPConfig(Base):
     user_search_filter = Column(
         String(500), default="(uid={username})", nullable=False
     )  # 用户搜索过滤器
-    username_attr = Column(String(50), default="uid", nullable=False)  # 用户名属性 (uid/sAMAccountName)
+    username_attr = Column(
+        String(50), default="uid", nullable=False
+    )  # 用户名属性 (uid/sAMAccountName)
     email_attr = Column(String(50), default="mail", nullable=False)  # 邮箱属性
     display_name_attr = Column(String(50), default="cn", nullable=False)  # 显示名称属性
     is_enabled = Column(Boolean, default=False, nullable=False)  # 是否启用 LDAP 认证
@@ -706,6 +710,14 @@ class ProviderEndpoint(Base):
     # 额外配置
     config = Column(JSON, nullable=True)  # 端点特定配置（不推荐使用，优先使用专用字段）
 
+    # 格式转换配置
+    format_acceptance_config = Column(
+        JSON,
+        nullable=True,
+        default=None,
+        comment="格式接受策略配置（跨格式转换开关/白黑名单等）",
+    )
+
     # 代理配置
     proxy = Column(JSONB, nullable=True)  # 代理配置: {url, username, password}
 
@@ -1041,8 +1053,7 @@ class Model(Base):
 
         # 获取所有最高优先级的映射
         top_priority_mappings = [
-            mapping for mapping in sorted_mappings
-            if mapping["priority"] == highest_priority
+            mapping for mapping in sorted_mappings if mapping["priority"] == highest_priority
         ]
 
         # 如果有多个相同优先级的映射，通过哈希分散选择
@@ -1119,9 +1130,7 @@ class ProviderAPIKey(Base):
     # 示例: {"cache_1h": true, "context_1m": true}
 
     # 自适应 RPM 调整（仅当 rpm_limit = NULL 时生效）
-    learned_rpm_limit = Column(
-        Integer, nullable=True
-    )  # 学习到的 RPM 限制（自适应模式下的有效值）
+    learned_rpm_limit = Column(Integer, nullable=True)  # 学习到的 RPM 限制（自适应模式下的有效值）
     concurrent_429_count = Column(Integer, default=0, nullable=False)  # 因并发导致的429次数
     rpm_429_count = Column(Integer, default=0, nullable=False)  # 因RPM导致的429次数
     last_429_at = Column(DateTime(timezone=True), nullable=True)  # 最后429时间
@@ -1132,9 +1141,7 @@ class ProviderAPIKey(Base):
     utilization_samples = Column(
         JSON, nullable=True
     )  # 利用率采样窗口 [{"ts": timestamp, "util": 0.8}, ...]
-    last_probe_increase_at = Column(
-        DateTime(timezone=True), nullable=True
-    )  # 上次探测性扩容时间
+    last_probe_increase_at = Column(DateTime(timezone=True), nullable=True)  # 上次探测性扩容时间
 
     # 健康度追踪（按 API 格式存储）
     # 结构: {"CLAUDE": {"health_score": 1.0, "consecutive_failures": 0, "last_failure_at": null, "request_results_window": []}, ...}
@@ -1568,7 +1575,9 @@ class RequestCandidate(Base):
     )
 
     # 状态信息
-    status = Column(String(20), nullable=False)  # 'pending', 'streaming', 'success', 'failed', 'cancelled', 'skipped'
+    status = Column(
+        String(20), nullable=False
+    )  # 'pending', 'streaming', 'success', 'failed', 'cancelled', 'skipped'
     skip_reason = Column(Text, nullable=True)  # 跳过/失败原因
     is_cached = Column(Boolean, default=False)  # 是否为缓存亲和性候选
 

src/models/endpoint_models.py

@@ -45,6 +45,12 @@ class ProviderEndpointCreate(BaseModel):
     # 代理配置
     proxy: Optional[ProxyConfig] = Field(default=None, description="代理配置")
 
+    # 格式转换配置
+    format_acceptance_config: Optional[Dict[str, Any]] = Field(
+        default=None,
+        description="格式接受策略配置（跨格式转换开关/白黑名单等）",
+    )
+
     @field_validator("api_format")
     @classmethod
     def validate_api_format(cls, v: str) -> str:
@@ -85,6 +91,12 @@ class ProviderEndpointUpdate(BaseModel):
     config: Optional[Dict[str, Any]] = Field(default=None, description="额外配置")
     proxy: Optional[ProxyConfig] = Field(default=None, description="代理配置")
 
+    # 格式转换配置
+    format_acceptance_config: Optional[Dict[str, Any]] = Field(
+        default=None,
+        description="格式接受策略配置（跨格式转换开关/白黑名单等）",
+    )
+
     @field_validator("base_url")
     @classmethod
     def validate_base_url(cls, v: Optional[str]) -> Optional[str]:
@@ -126,6 +138,12 @@ class ProviderEndpointResponse(BaseModel):
     # 代理配置（响应中密码已脱敏）
     proxy: Optional[Dict[str, Any]] = Field(default=None, description="代理配置（密码已脱敏）")
 
+    # 格式转换配置
+    format_acceptance_config: Optional[Dict[str, Any]] = Field(
+        default=None,
+        description="格式接受策略配置（跨格式转换开关/白黑名单等）",
+    )
+
     # 统计（从 Keys 聚合）
     total_keys: int = Field(default=0, description="总 Key 数量")
     active_keys: int = Field(default=0, description="活跃 Key 数量")

src/services/cache/aware_scheduler.py

@@ -35,7 +35,7 @@ import random
 import re
 import time
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple, Union
 
 from sqlalchemy.orm import Session, selectinload
 
@@ -65,6 +65,7 @@ from src.services.rate_limit.adaptive_reservation import (
     get_adaptive_reservation_manager,
 )
 from src.services.rate_limit.concurrency_manager import get_concurrency_manager
+from src.services.system.config import SystemConfigService
 
 
 @dataclass
@@ -78,6 +79,8 @@ class ProviderCandidate:
     is_skipped: bool = False  # 是否被跳过
     skip_reason: Optional[str] = None  # 跳过原因
     mapping_matched_model: Optional[str] = None  # 通过映射匹配到的模型名（用于实际请求）
+    needs_conversion: bool = False  # 是否需要格式转换
+    provider_api_format: str = ""  # Provider 端点实际格式（用于健康度/熔断 bucket）
 
 
 @dataclass
@@ -129,7 +132,10 @@ class CacheAwareScheduler:
     }
 
     def __init__(
-        self, redis_client=None, priority_mode: Optional[str] = None, scheduling_mode: Optional[str] = None
+        self,
+        redis_client=None,
+        priority_mode: Optional[str] = None,
+        scheduling_mode: Optional[str] = None,
     ):
         """
         初始化调度器
@@ -149,7 +155,9 @@ class CacheAwareScheduler:
         self.scheduling_mode = self._normalize_scheduling_mode(
             scheduling_mode or self.SCHEDULING_MODE_CACHE_AFFINITY
         )
-        logger.debug(f"[CacheAwareScheduler] 初始化优先级模式: {self.priority_mode}, 调度模式: {self.scheduling_mode}")
+        logger.debug(
+            f"[CacheAwareScheduler] 初始化优先级模式: {self.priority_mode}, 调度模式: {self.scheduling_mode}"
+        )
 
         # 初始化子组件（将在第一次使用时异步初始化）
         self._affinity_manager: Optional[CacheAffinityManager] = None
@@ -429,7 +437,9 @@ class CacheAwareScheduler:
                 import math
 
                 # 与 ConcurrencyManager 的 Lua 脚本保持一致：使用 floor 计算新用户可用槽位
-                available_for_new = max(1, math.floor(effective_key_limit * (1 - reservation_ratio)))
+                available_for_new = max(
+                    1, math.floor(effective_key_limit * (1 - reservation_ratio))
+                )
                 if key_count >= available_for_new:
                     logger.debug(
                         f"Key {key.id[:8]}... 新用户配额已满 "
@@ -531,8 +541,7 @@ class CacheAwareScheduler:
 
         # 合并 allowed_api_formats
         result["allowed_api_formats"] = merge_restrictions(
-            user_api_key.allowed_api_formats,
-            user.allowed_api_formats if user else None
+            user_api_key.allowed_api_formats, user.allowed_api_formats if user else None
         )
 
         return result
@@ -578,7 +587,9 @@ class CacheAwareScheduler:
         target_format = normalize_api_format(api_format)
 
         # 0. 解析 model_name 到 GlobalModel（支持直接匹配和映射名匹配，使用 ModelCacheService）
-        global_model = await ModelCacheService.resolve_global_model_by_name_or_mapping(db, model_name)
+        global_model = await ModelCacheService.resolve_global_model_by_name_or_mapping(
+            db, model_name
+        )
 
         if not global_model:
             logger.warning(f"GlobalModel not found: {model_name}")
@@ -592,7 +603,9 @@ class CacheAwareScheduler:
         # 提取模型映射（用于 Provider Key 的 allowed_models 匹配）
         model_mappings: List[str] = (global_model.config or {}).get("model_mappings", [])
         if model_mappings:
-            logger.debug(f"[Scheduler] GlobalModel={global_model.name} 配置了映射规则: {model_mappings}")
+            logger.debug(
+                f"[Scheduler] GlobalModel={global_model.name} 配置了映射规则: {model_mappings}"
+            )
 
         # 获取合并后的访问限制（ApiKey + User）
         restrictions = self._get_effective_restrictions(user_api_key)
@@ -654,10 +667,13 @@ class CacheAwareScheduler:
             return [], global_model_id
 
         # 2. 构建候选列表（传入 is_stream 和 capability_requirements 用于过滤）
+        global_conversion_enabled = bool(
+            SystemConfigService.get_config(db, "format_conversion_enabled", False)
+        )
         candidates = await self._build_candidates(
             db=db,
             providers=providers,
-            target_format=target_format,
+            client_format=target_format,
             model_name=requested_model_name,
             resolved_model_name=resolved_model_name,
             model_mappings=model_mappings,
@@ -665,6 +681,7 @@ class CacheAwareScheduler:
             max_candidates=max_candidates,
             is_stream=is_stream,
             capability_requirements=capability_requirements,
+            global_conversion_enabled=global_conversion_enabled,
         )
 
         # 3. 应用优先级模式排序
@@ -774,15 +791,25 @@ class CacheAwareScheduler:
             - provider_model_names: Provider 侧可用的模型名称集合（主名称 + 映射名称，按 api_format 过滤）
         """
         # 使用 ModelCacheService 解析模型名称（支持映射名）
-        global_model = await ModelCacheService.resolve_global_model_by_name_or_mapping(db, model_name)
+        global_model = await ModelCacheService.resolve_global_model_by_name_or_mapping(
+            db, model_name
+        )
 
         if not global_model:
             # 完全未找到匹配
             return False, "模型不存在或 Provider 未配置此模型", None, None
 
         # 找到 GlobalModel 后，检查当前 Provider 是否支持
-        is_supported, skip_reason, caps, provider_model_names = await self._check_model_support_for_global_model(
-            db, provider, global_model, model_name, api_format, is_stream, capability_requirements
+        is_supported, skip_reason, caps, provider_model_names = (
+            await self._check_model_support_for_global_model(
+                db,
+                provider,
+                global_model,
+                model_name,
+                api_format,
+                is_stream,
+                capability_requirements,
+            )
         )
         return is_supported, skip_reason, caps, provider_model_names
 
@@ -814,6 +841,7 @@ class CacheAwareScheduler:
         # 注意：从缓存重建的对象是 transient 状态，不能使用 load=False
         # 使用 load=True（默认）允许 SQLAlchemy 正确处理 transient 对象
         from sqlalchemy import inspect
+
         insp = inspect(global_model)
         if insp.transient or insp.detached:
             # transient/detached 对象：使用默认 merge（会查询 DB 检查是否存在）
@@ -940,12 +968,18 @@ class CacheAwareScheduler:
             return False, f"映射规则无效: {str(e)}", None
         except Exception as e:
             # 其他未知异常
-            logger.error(f"映射匹配异常: key_id={key.id}, model={model_name}, error={e}", exc_info=True)
+            logger.error(
+                f"映射匹配异常: key_id={key.id}, model={model_name}, error={e}", exc_info=True
+            )
             # 异常时保守处理：不允许使用该 Key
             return False, "映射匹配失败", None
 
         if not is_allowed:
-            return False, f"模型权限不匹配(允许: {get_allowed_models_preview(key.allowed_models)})", None
+            return (
+                False,
+                f"模型权限不匹配(允许: {get_allowed_models_preview(key.allowed_models)})",
+                None,
+            )
 
         # Key 级别的能力匹配检查
         # 注意：模型级别的能力检查已在 _check_model_support 中完成
@@ -964,7 +998,7 @@ class CacheAwareScheduler:
         self,
         db: Session,
         providers: List[Provider],
-        target_format: APIFormat,
+        client_format: APIFormat,
         model_name: str,
         affinity_key: Optional[str],
         resolved_model_name: Optional[str] = None,
@@ -972,16 +1006,17 @@ class CacheAwareScheduler:
         max_candidates: Optional[int] = None,
         is_stream: bool = False,
         capability_requirements: Optional[Dict[str, bool]] = None,
+        global_conversion_enabled: bool = False,
     ) -> List[ProviderCandidate]:
         """
         构建候选列表
 
-        Key 直属 Provider，通过 api_formats 筛选符合目标格式的 Key。
+        Key 直属 Provider，通过 api_formats 筛选符合端点格式的 Key。
 
         Args:
             db: 数据库会话
             providers: Provider 列表
-            target_format: 目标 API 格式
+            client_format: 客户端请求的 API 格式
             model_name: 模型名称（用户请求的名称，可能是映射名）
             affinity_key: 亲和性标识符（通常为API Key ID）
             resolved_model_name: 解析后的 GlobalModel.name（用于 Key.allowed_models 校验）
@@ -989,89 +1024,122 @@ class CacheAwareScheduler:
             max_candidates: 最大候选数
             is_stream: 是否是流式请求，如果为 True 则过滤不支持流式的 Provider
             capability_requirements: 能力需求（可选）
+            global_conversion_enabled: 全局格式转换开关
 
         Returns:
             候选列表
         """
+        from src.core.api_format.conversion.compatibility import is_format_compatible
+
         candidates: List[ProviderCandidate] = []
-        target_format_str = target_format.value
+        client_format_str = client_format.value
 
         for provider in providers:
-            # 检查模型支持（同时检查流式支持和模型能力需求）
-            supports_model, skip_reason, _model_caps, provider_model_names = await self._check_model_support(
-                db,
-                provider,
-                model_name,
-                api_format=target_format_str,
-                is_stream=is_stream,
-                capability_requirements=capability_requirements,
-            )
-            if not supports_model:
-                logger.debug(f"Provider {provider.name} 不支持模型 {model_name}: {skip_reason}")
-                continue
+            # 按端点格式分别判断兼容性与模型/Key 可用性：
+            # - 同格式端点优先（needs_conversion=False）
+            # - 跨格式端点次之（needs_conversion=True）
+            model_support_cache: Dict[
+                str, Tuple[bool, Optional[str], Optional[List[str]], Optional[Set[str]]]
+            ] = {}
+            exact_candidates: List[ProviderCandidate] = []
+            convertible_candidates: List[ProviderCandidate] = []
 
-            # 查找目标格式对应的 Endpoint（获取请求配置）
-            target_endpoint = None
             for endpoint in provider.endpoints:
+                if not endpoint.is_active:
+                    continue
+
                 endpoint_format_str = (
                     endpoint.api_format
                     if isinstance(endpoint.api_format, str)
                     else endpoint.api_format.value
                 )
-                if endpoint.is_active and endpoint_format_str == target_format_str:
-                    target_endpoint = endpoint
-                    break
 
-            if not target_endpoint:
-                logger.debug(f"Provider {provider.name} 没有活跃的 {target_format_str} 端点")
-                continue
-
-            # Key 直属 Provider，通过 api_formats 筛选
-            active_keys = [
-                key for key in provider.api_keys
-                if key.is_active and target_format_str in (key.api_formats or [])
-            ]
-
-            if not active_keys:
-                logger.debug(f"Provider {provider.name} 没有支持 {target_format_str} 的活跃 Key")
-                continue
-
-            # 检查是否所有 Key 都是 TTL=0（轮换模式）
-            use_random = all(
-                (key.cache_ttl_minutes or 0) == 0 for key in active_keys
-            ) if active_keys else False
-            if use_random and len(active_keys) > 1:
-                logger.debug(
-                    f"  Provider {provider.name} 启用 Key 轮换模式 (TTL=0, {len(active_keys)} keys)"
+                is_compatible, needs_conversion, _compat_reason = is_format_compatible(
+                    client_format_str,
+                    endpoint_format_str,
+                    getattr(endpoint, "format_acceptance_config", None),
+                    is_stream,
+                    global_conversion_enabled,
                 )
-            keys = self._shuffle_keys_by_internal_priority(active_keys, affinity_key, use_random)
+                if not is_compatible:
+                    continue
 
-            for key in keys:
-                # Key 级别的能力检查
-                # 注意：不传入 candidate_models 限制，允许映射匹配到 Key 的 allowed_models 中的任意模型名
-                # 这支持以下场景：Key 只允许使用 gpt-5.2，而 GlobalModel 配置了映射 gpt-5.*2
-                # 映射匹配后，实际请求会使用 gpt-5.2 作为模型名发送给 Provider
-                is_available, skip_reason, mapping_matched_model = self._check_key_availability(
-                    key,
-                    target_format_str,
-                    model_name,
-                    capability_requirements,
-                    resolved_model_name=resolved_model_name,
-                    model_mappings=model_mappings,
+                # 检查模型支持（按端点格式过滤 provider_model_mappings）
+                if endpoint_format_str not in model_support_cache:
+                    model_support_cache[endpoint_format_str] = await self._check_model_support(
+                        db,
+                        provider,
+                        model_name,
+                        api_format=endpoint_format_str,
+                        is_stream=is_stream,
+                        capability_requirements=capability_requirements,
+                    )
+                supports_model, skip_reason, _model_caps, provider_model_names = (
+                    model_support_cache[endpoint_format_str]
+                )
+                if not supports_model:
+                    logger.debug(
+                        f"Provider {provider.name} 端点 {endpoint_format_str} 不支持模型 {model_name}: {skip_reason}"
+                    )
+                    continue
+
+                # Key 直属 Provider，通过 api_formats 按端点格式筛选
+                active_keys = [
+                    key
+                    for key in provider.api_keys
+                    if key.is_active and endpoint_format_str in (key.api_formats or [])
+                ]
+                if not active_keys:
+                    continue
+
+                # 检查是否所有 Key 都是 TTL=0（轮换模式）
+                use_random = all((key.cache_ttl_minutes or 0) == 0 for key in active_keys)
+                if use_random and len(active_keys) > 1:
+                    logger.debug(
+                        f"  Provider {provider.name} 启用 Key 轮换模式 "
+                        f"(endpoint_format={endpoint_format_str}, {len(active_keys)} keys)"
+                    )
+
+                keys = self._shuffle_keys_by_internal_priority(
+                    active_keys, affinity_key, use_random
                 )
 
-                candidate = ProviderCandidate(
-                    provider=provider,
-                    endpoint=target_endpoint,
-                    key=key,
-                    is_skipped=not is_available,
-                    skip_reason=skip_reason,
-                    mapping_matched_model=mapping_matched_model,
-                )
-                candidates.append(candidate)
+                for key in keys:
+                    # Key 级别检查（健康度/熔断按 provider_format bucket）
+                    # 注意：不传入 candidate_models，保持原有映射匹配行为
+                    is_available, key_skip_reason, mapping_matched_model = (
+                        self._check_key_availability(
+                            key,
+                            endpoint_format_str,
+                            model_name,
+                            capability_requirements,
+                            resolved_model_name=resolved_model_name,
+                            model_mappings=model_mappings,
+                        )
+                    )
 
-                if max_candidates and len(candidates) >= max_candidates:
-                    return candidates
+                    candidate = ProviderCandidate(
+                        provider=provider,
+                        endpoint=endpoint,
+                        key=key,
+                        is_skipped=not is_available,
+                        skip_reason=key_skip_reason,
+                        mapping_matched_model=mapping_matched_model,
+                        needs_conversion=needs_conversion,
+                        provider_api_format=str(endpoint_format_str or "").upper(),
+                    )
+
+                    if needs_conversion:
+                        convertible_candidates.append(candidate)
+                    else:
+                        exact_candidates.append(candidate)
+
+            candidates.extend(exact_candidates)
+            candidates.extend(convertible_candidates)
+
+        # max_candidates 截断应在所有候选收集完成后统一处理，确保优先级排序正确
+        if max_candidates and len(candidates) > max_candidates:
+            candidates = candidates[:max_candidates]
 
         return candidates
 
@@ -1173,7 +1241,9 @@ class CacheAwareScheduler:
         normalized = (mode or "").strip().lower()
         if normalized not in self.ALLOWED_SCHEDULING_MODES:
             if normalized:
-                logger.warning(f"[CacheAwareScheduler] 无效的调度模式 '{mode}'，回退为 cache_affinity")
+                logger.warning(
+                    f"[CacheAwareScheduler] 无效的调度模式 '{mode}'，回退为 cache_affinity"
+                )
             return self.SCHEDULING_MODE_CACHE_AFFINITY
         return normalized
 
@@ -1186,8 +1256,10 @@ class CacheAwareScheduler:
         logger.debug(f"[CacheAwareScheduler] 切换调度模式为: {self.scheduling_mode}")
 
     def _apply_priority_mode_sort(
-        self, candidates: List[ProviderCandidate], affinity_key: Optional[str] = None,
-        api_format: Optional[str] = None
+        self,
+        candidates: List[ProviderCandidate],
+        affinity_key: Optional[str] = None,
+        api_format: Optional[str] = None,
     ) -> List[ProviderCandidate]:
         """
         根据优先级模式对候选列表排序（数字越小越优先）
@@ -1209,8 +1281,10 @@ class CacheAwareScheduler:
         return candidates
 
     def _sort_by_global_priority_with_hash(
-        self, candidates: List[ProviderCandidate], affinity_key: Optional[str] = None,
-        api_format: Optional[str] = None
+        self,
+        candidates: List[ProviderCandidate],
+        affinity_key: Optional[str] = None,
+        api_format: Optional[str] = None,
     ) -> List[ProviderCandidate]:
         """
         按 global_priority_by_format 分组排序，同优先级内通过哈希分散实现负载均衡

src/services/orchestration/error_classifier.py

@@ -30,7 +30,6 @@ from src.services.rate_limit.adaptive_rpm import get_adaptive_rpm_manager
 from src.services.rate_limit.detector import RateLimitType, detect_rate_limit_type
 
 
-
 class ErrorAction(Enum):
     """错误处理动作"""
 
@@ -391,10 +390,12 @@ class ErrorClassifier:
                 current_usage=current_rpm,
             )
 
-            logger.info(f"  [{request_id}] 429错误分析: "
+            logger.info(
+                f"  [{request_id}] 429错误分析: "
                 f"类型={rate_limit_info.limit_type}, "
                 f"retry_after={rate_limit_info.retry_after}s, "
-                f"当前RPM={current_rpm}")
+                f"当前RPM={current_rpm}"
+            )
 
             # 调用自适应管理器处理
             new_limit = self.adaptive_manager.handle_429_error(
@@ -408,7 +409,9 @@ class ErrorClassifier:
                 logger.warning(f"  [{request_id}] 并发限制触发（不调整RPM）")
                 return "concurrent"
             elif rate_limit_info.limit_type == RateLimitType.RPM:
-                logger.warning(f"  [{request_id}] 自适应调整: Key {key.id[:8]}... RPM限制 -> {new_limit}")
+                logger.warning(
+                    f"  [{request_id}] 自适应调整: Key {key.id[:8]}... RPM限制 -> {new_limit}"
+                )
                 return "rpm"
             else:
                 return "unknown"
@@ -545,8 +548,10 @@ class ErrorClassifier:
             except Exception:
                 pass
 
-        logger.warning(f"  [{request_id}] HTTP错误 (attempt={attempt}/{max_attempts}): "
-            f"{http_error.response.status_code if http_error.response else 'unknown'}")
+        logger.warning(
+            f"  [{request_id}] HTTP错误 (attempt={attempt}/{max_attempts}): "
+            f"{http_error.response.status_code if http_error.response else 'unknown'}"
+        )
 
         converted_error = self.convert_http_error(http_error, provider_name, error_response_text)
 
@@ -557,16 +562,25 @@ class ErrorClassifier:
         if error_response_text:
             extra_data["error_response"] = error_response_text
 
-        # 转换 api_format 为字符串
-        api_format_str = (
+        # client_format：用于缓存亲和性/缓存失效（用户视角）
+        client_format_str = (
             normalize_api_format(api_format).value
             if isinstance(api_format, (str, APIFormat))
             else str(api_format)
         )
+        # provider_format：用于健康度/熔断 bucket（Provider 真实端点格式）
+        provider_api_format = getattr(endpoint, "api_format", None)
+        provider_format_str = (
+            provider_api_format.value
+            if isinstance(provider_api_format, APIFormat)
+            else str(provider_api_format or client_format_str)
+        ).upper()
 
         # 处理客户端请求错误（不应重试，不失效缓存，不记录健康失败）
         if isinstance(converted_error, UpstreamClientException):
-            logger.warning(f"  [{request_id}] 客户端请求错误，不进行重试: {converted_error.message}")
+            logger.warning(
+                f"  [{request_id}] 客户端请求错误，不进行重试: {converted_error.message}"
+            )
             return extra_data
 
         # 处理认证错误
@@ -574,7 +588,7 @@ class ErrorClassifier:
             if endpoint and key and self.cache_scheduler is not None:
                 await self.cache_scheduler.invalidate_cache(
                     affinity_key=affinity_key,
-                    api_format=api_format_str,
+                    api_format=client_format_str,
                     global_model_id=global_model_id,
                     endpoint_id=str(endpoint.id),
                     key_id=str(key.id),
@@ -583,7 +597,7 @@ class ErrorClassifier:
                 health_monitor.record_failure(
                     db=self.db,
                     key_id=str(key.id),
-                    api_format=api_format_str,
+                    api_format=provider_format_str,
                     error_type="ProviderAuthException",
                 )
             return extra_data
@@ -600,7 +614,7 @@ class ErrorClassifier:
             if endpoint and self.cache_scheduler is not None:
                 await self.cache_scheduler.invalidate_cache(
                     affinity_key=affinity_key,
-                    api_format=api_format_str,
+                    api_format=client_format_str,
                     global_model_id=global_model_id,
                     endpoint_id=str(endpoint.id),
                     key_id=str(key.id),
@@ -610,7 +624,7 @@ class ErrorClassifier:
             if endpoint and key and self.cache_scheduler is not None:
                 await self.cache_scheduler.invalidate_cache(
                     affinity_key=affinity_key,
-                    api_format=api_format_str,
+                    api_format=client_format_str,
                     global_model_id=global_model_id,
                     endpoint_id=str(endpoint.id),
                     key_id=str(key.id),
@@ -621,7 +635,7 @@ class ErrorClassifier:
             health_monitor.record_failure(
                 db=self.db,
                 key_id=str(key.id),
-                api_format=api_format_str,
+                api_format=provider_format_str,
                 error_type=type(converted_error).__name__,
             )
 
@@ -662,15 +676,24 @@ class ErrorClassifier:
         """
         provider_name = str(provider.name)
 
-        logger.warning(f"  [{request_id}] 请求失败 (attempt={attempt}/{max_attempts}): "
-            f"{type(error).__name__}: {str(error)}")
+        logger.warning(
+            f"  [{request_id}] 请求失败 (attempt={attempt}/{max_attempts}): "
+            f"{type(error).__name__}: {str(error)}"
+        )
 
-        # 转换 api_format 为字符串
-        api_format_str = (
+        # client_format：用于缓存亲和性/缓存失效（用户视角）
+        client_format_str = (
             normalize_api_format(api_format).value
             if isinstance(api_format, (str, APIFormat))
             else str(api_format)
         )
+        # provider_format：用于健康度/熔断 bucket（Provider 真实端点格式）
+        provider_api_format = getattr(endpoint, "api_format", None)
+        provider_format_str = (
+            provider_api_format.value
+            if isinstance(provider_api_format, APIFormat)
+            else str(provider_api_format or client_format_str)
+        ).upper()
 
         # 处理限流错误
         if isinstance(error, ProviderRateLimitException) and key:
@@ -684,7 +707,7 @@ class ErrorClassifier:
             if endpoint and self.cache_scheduler is not None:
                 await self.cache_scheduler.invalidate_cache(
                     affinity_key=affinity_key,
-                    api_format=api_format_str,
+                    api_format=client_format_str,
                     global_model_id=global_model_id,
                     endpoint_id=str(endpoint.id),
                     key_id=str(key.id),
@@ -693,7 +716,7 @@ class ErrorClassifier:
             # 其他错误也失效缓存
             await self.cache_scheduler.invalidate_cache(
                 affinity_key=affinity_key,
-                api_format=api_format_str,
+                api_format=client_format_str,
                 global_model_id=global_model_id,
                 endpoint_id=str(endpoint.id),
                 key_id=str(key.id),
@@ -704,6 +727,6 @@ class ErrorClassifier:
             health_monitor.record_failure(
                 db=self.db,
                 key_id=str(key.id),
-                api_format=api_format_str,
+                api_format=provider_format_str,
                 error_type=type(error).__name__,
             )

src/services/orchestration/fallback_orchestrator.py

@@ -29,7 +29,7 @@ import httpx
 from redis import Redis
 from sqlalchemy.orm import Session
 
-from src.core.api_format import APIFormat
+from src.core.api_format import APIFormat, FormatConversionError
 from src.core.error_utils import extract_error_message
 from src.core.exceptions import (
     ConcurrencyLimitError,
@@ -385,7 +385,9 @@ class FallbackOrchestrator:
                     "provider_id": str(provider.id),
                     "provider_endpoint_id": str(endpoint.id),
                     "provider_api_key_id": str(key.id),
-                    "api_format": api_format.value if hasattr(api_format, "value") else str(api_format),
+                    "api_format": (
+                        api_format.value if hasattr(api_format, "value") else str(api_format)
+                    ),
                 }
                 raise client_error
             else:
@@ -425,10 +427,14 @@ class FallbackOrchestrator:
             # 检查是否为客户端请求错误（不应重试）
             converted_error = extra_data.get("converted_error")
             # 从 extra_data 中移除 converted_error，避免序列化问题
-            serializable_extra_data = {k: v for k, v in extra_data.items() if k != "converted_error"}
+            serializable_extra_data = {
+                k: v for k, v in extra_data.items() if k != "converted_error"
+            }
 
             if isinstance(converted_error, UpstreamClientException):
-                logger.warning(f"  [{request_id}] 客户端请求错误，停止重试: {converted_error.message}")
+                logger.warning(
+                    f"  [{request_id}] 客户端请求错误，停止重试: {converted_error.message}"
+                )
                 RequestCandidateService.mark_candidate_failed(
                     db=self.db,
                     candidate_id=candidate_record_id,
@@ -445,7 +451,9 @@ class FallbackOrchestrator:
                     "provider_id": str(provider.id),
                     "provider_endpoint_id": str(endpoint.id),
                     "provider_api_key_id": str(key.id),
-                    "api_format": api_format.value if hasattr(api_format, "value") else str(api_format),
+                    "api_format": (
+                        api_format.value if hasattr(api_format, "value") else str(api_format)
+                    ),
                 }
                 raise converted_error
 
@@ -487,6 +495,19 @@ class FallbackOrchestrator:
             )
             return "continue" if has_retry_left else "break"
 
+        # 格式转换错误：视为候选不可用，直接切换到下一个候选（不记录健康失败）
+        if isinstance(cause, FormatConversionError):
+            logger.warning(f"  [{request_id}] 格式转换失败，切换候选: {cause}")
+            RequestCandidateService.mark_candidate_failed(
+                db=self.db,
+                candidate_id=candidate_record_id,
+                error_type="FormatConversionError",
+                error_message=str(cause),
+                latency_ms=elapsed_ms,
+                concurrent_requests=captured_key_concurrent,
+            )
+            return "break"
+
         # 未知错误：记录失败并抛出
         RequestCandidateService.mark_candidate_failed(
             db=self.db,
@@ -552,8 +573,10 @@ class FallbackOrchestrator:
             last_candidate = candidate
 
             if candidate.is_skipped:
-                logger.debug(f"  [{request_id}] 跳过候选: Provider={candidate.provider.name}, "
-                    f"Reason={candidate.skip_reason}")
+                logger.debug(
+                    f"  [{request_id}] 跳过候选: Provider={candidate.provider.name}, "
+                    f"Reason={candidate.skip_reason}"
+                )
                 continue
 
             result = await self._try_candidate_with_retries(
@@ -573,7 +596,9 @@ class FallbackOrchestrator:
             )
 
             if result["success"]:
-                response: Tuple[Any, str, Optional[str], Optional[str], Optional[str], Optional[str]] = result["response"]
+                response: Tuple[
+                    Any, str, Optional[str], Optional[str], Optional[str], Optional[str]
+                ] = result["response"]
                 return response
 
             # 更新计数器和错误信息
@@ -582,7 +607,9 @@ class FallbackOrchestrator:
             if result.get("error"):
                 last_error = result["error"]
             if result.get("should_raise") and last_error is not None:
-                self._attach_metadata_to_error(last_error, last_candidate, model_name, api_format_enum)
+                self._attach_metadata_to_error(
+                    last_error, last_candidate, model_name, api_format_enum
+                )
                 raise last_error
 
         # 所有组合都已尝试完毕，全部失败
@@ -620,9 +647,13 @@ class FallbackOrchestrator:
             if retry_index == 0:
                 # 首次尝试该候选
                 cache_hint = " (cached)" if candidate.is_cached else ""
-                logger.info(f"  [{request_id[:8] if request_id else 'N/A'}] -> {provider.name}{cache_hint}")
+                logger.info(
+                    f"  [{request_id[:8] if request_id else 'N/A'}] -> {provider.name}{cache_hint}"
+                )
             else:
-                logger.info(f"  [{request_id[:8] if request_id else 'N/A'}] -> {provider.name} (retry {retry_index})")
+                logger.info(
+                    f"  [{request_id[:8] if request_id else 'N/A'}] -> {provider.name} (retry {retry_index})"
+                )
 
             candidate_record_id = candidate_record_map[(candidate_index, retry_index)]
 
@@ -706,14 +737,14 @@ class FallbackOrchestrator:
             ),
             provider=getattr(existing_metadata, "provider", None) or str(candidate.provider.name),
             model=getattr(existing_metadata, "model", None) or model_name,
-            provider_id=getattr(existing_metadata, "provider_id", None) or str(candidate.provider.id),
+            provider_id=getattr(existing_metadata, "provider_id", None)
+            or str(candidate.provider.id),
             provider_endpoint_id=(
                 getattr(existing_metadata, "provider_endpoint_id", None)
                 or str(candidate.endpoint.id)
             ),
             provider_api_key_id=(
-                getattr(existing_metadata, "provider_api_key_id", None)
-                or str(candidate.key.id)
+                getattr(existing_metadata, "provider_api_key_id", None) or str(candidate.key.id)
             ),
             api_format=api_format_enum.value,
         )
@@ -821,12 +852,16 @@ class FallbackOrchestrator:
         user_id = str(user_api_key.user_id)
         api_format_enum = normalize_api_format(api_format)
 
-        logger.debug(f"[FallbackOrchestrator] execute_with_fallback 被调用: "
+        logger.debug(
+            f"[FallbackOrchestrator] execute_with_fallback 被调用: "
             f"api_format={api_format_enum.value}, model_name={model_name}, "
-            f"request_id={request_id}, is_stream={is_stream}")
+            f"request_id={request_id}, is_stream={is_stream}"
+        )
 
         # 创建 pending 状态的使用记录
-        self._create_pending_usage_record(request_id, user_api_key, model_name, is_stream, api_format_enum)
+        self._create_pending_usage_record(
+            request_id, user_api_key, model_name, is_stream, api_format_enum
+        )
 
         # 1. 收集所有候选（同时获取规范化的 global_model_id 用于缓存亲和性）
         all_candidates, global_model_id = await self._fetch_all_candidates(

src/services/request/executor.py

@@ -12,11 +12,11 @@ from src.core.api_format import APIFormat
 from src.core.exceptions import ConcurrencyLimitError
 from src.core.logger import logger
 from src.services.health.monitor import health_monitor
+from src.services.provider.format import normalize_api_format
 from src.services.rate_limit.adaptive_reservation import get_adaptive_reservation_manager
 from src.services.request.candidate import RequestCandidateService
 
 
-
 @dataclass
 class ExecutionContext:
     candidate_id: str
@@ -103,7 +103,9 @@ class RequestExecutor:
             # 获取有效的 RPM 限制（自适应或固定）
             if key.rpm_limit is None:
                 # 自适应模式：使用学习值，未学习时为 None（不限制，等待碰壁学习）
-                effective_key_limit = int(key.learned_rpm_limit) if key.learned_rpm_limit is not None else None
+                effective_key_limit = (
+                    int(key.learned_rpm_limit) if key.learned_rpm_limit is not None else None
+                )
             else:
                 effective_key_limit = int(key.rpm_limit)
 
@@ -114,9 +116,11 @@ class RequestExecutor:
             )
             dynamic_reservation_ratio = reservation_result.ratio
 
-            logger.debug(f"[Executor] 动态预留: key={key.id[:8]}..., "
+            logger.debug(
+                f"[Executor] 动态预留: key={key.id[:8]}..., "
                 f"ratio={dynamic_reservation_ratio:.0%}, phase={reservation_result.phase}, "
-                f"confidence={reservation_result.confidence:.0%}")
+                f"confidence={reservation_result.confidence:.0%}"
+            )
 
             async with self.concurrency_manager.rpm_guard(
                 key_id=key.id,
@@ -140,12 +144,21 @@ class RequestExecutor:
 
                 context.elapsed_ms = int((time.time() - context.start_time) * 1000)
 
+                provider_api_format = getattr(endpoint, "api_format", None)
+                provider_format_str = (
+                    provider_api_format.value
+                    if isinstance(provider_api_format, APIFormat)
+                    else str(provider_api_format or "")
+                )
+                client_format_str = (
+                    api_format.value if isinstance(api_format, APIFormat) else str(api_format)
+                )
+                health_format = normalize_api_format(provider_format_str or client_format_str).value
+
                 health_monitor.record_success(
                     db=self.db,
                     key_id=key.id,
-                    api_format=(
-                        api_format.value if isinstance(api_format, APIFormat) else api_format
-                    ),
+                    api_format=health_format,
                     response_time_ms=context.elapsed_ms,
                 )
 
@@ -180,7 +193,9 @@ class RequestExecutor:
                             "is_cached_user": is_cached_user,
                             "model_name": model_name,
                             "api_format": (
-                                api_format.value if isinstance(api_format, APIFormat) else api_format
+                                api_format.value
+                                if isinstance(api_format, APIFormat)
+                                else api_format
                             ),
                         },
                     )

src/services/system/config.py

@@ -118,6 +118,10 @@ class SystemConfigService:
             "value": "cache_affinity",
             "description": "调度模式：fixed_order(固定顺序模式，严格按优先级顺序) 或 cache_affinity(缓存亲和模式，优先使用已缓存的Provider)",
         },
+        "format_conversion_enabled": {
+            "value": False,
+            "description": "是否启用全局格式自动转换（需要端点配置 format_acceptance_config 才能生效）",
+        },
         "auto_delete_expired_keys": {
             "value": False,
             "description": "是否自动删除过期的API Key（True=物理删除，False=仅禁用），仅管理员可配置",

tests/api/handlers/base/test_stream_processor_format_conversion.py

@@ -0,0 +1,103 @@
+import json
+from typing import Any, Dict, Optional
+from unittest.mock import AsyncMock
+
+import pytest
+
+from src.api.handlers.base.response_parser import ParsedResponse, ResponseParser, StreamStats
+from src.api.handlers.base.stream_context import StreamContext
+from src.api.handlers.base.stream_processor import StreamProcessor
+from src.core.api_format import register_all_converters
+
+
+class DummyParser(ResponseParser):
+    def parse_sse_line(self, line: str, stats: StreamStats) -> Optional[Any]:  # noqa: ANN401
+        return None
+
+    def parse_response(self, response: Dict[str, Any], status_code: int) -> ParsedResponse:
+        return ParsedResponse(raw_response=response, status_code=status_code)
+
+    def extract_usage_from_response(self, response: Dict[str, Any]) -> Dict[str, int]:
+        return {}
+
+    def extract_text_content(self, response: Dict[str, Any]) -> str:
+        return ""
+
+
+async def _empty_async_iter():
+    if False:  # pragma: no cover
+        yield b""
+
+
+@pytest.mark.asyncio
+async def test_create_response_stream_converts_claude_to_openai() -> None:
+    register_all_converters()
+
+    ctx = StreamContext(model="test-model", api_format="OPENAI")
+    ctx.client_api_format = "OPENAI"
+    ctx.provider_api_format = "CLAUDE"
+    ctx.needs_conversion = True
+
+    processor = StreamProcessor(request_id="test-request", default_parser=DummyParser())
+
+    response_ctx = AsyncMock()
+    response_ctx.__aexit__ = AsyncMock(return_value=None)
+
+    http_client = AsyncMock()
+    http_client.aclose = AsyncMock(return_value=None)
+
+    message_start = {
+        "type": "message_start",
+        "message": {
+            "id": "msg_1",
+            "type": "message",
+            "role": "assistant",
+            "model": "claude-test",
+            "content": [],
+            "stop_reason": None,
+            "stop_sequence": None,
+        },
+    }
+    content_delta = {
+        "type": "content_block_delta",
+        "index": 0,
+        "delta": {"type": "text_delta", "text": "Hi"},
+    }
+
+    prefetched_chunks = [
+        b"event: message_start\n",
+        f"data: {json.dumps(message_start)}\n".encode("utf-8"),
+        b"\n",
+        f"data: {json.dumps(content_delta)}\n".encode("utf-8"),
+        b"\n",
+    ]
+
+    out = b"".join(
+        [
+            chunk
+            async for chunk in processor.create_response_stream(
+                ctx,
+                byte_iterator=_empty_async_iter(),
+                response_ctx=response_ctx,
+                http_client=http_client,
+                prefetched_chunks=prefetched_chunks,
+            )
+        ]
+    )
+
+    text = out.decode("utf-8")
+    assert "event:" not in text
+
+    events = []
+    for line in text.splitlines():
+        if line.startswith("data: "):
+            events.append(json.loads(line[6:]))
+
+    assert len(events) >= 2
+    assert any(e.get("object") == "chat.completion.chunk" for e in events)
+    assert any(
+        e.get("choices", [{}])[0].get("delta", {}).get("content") == "Hi"
+        for e in events
+        if isinstance(e, dict)
+    )
+

tests/core/api_format/conversion/test_compatibility.py

@@ -0,0 +1,161 @@
+"""
+is_format_compatible 单元测试
+
+覆盖：
+- 同格式透传
+- CLI 格式禁止转换
+- 全局开关/端点开关/白黑名单
+- 流式转换开关
+- 转换器能力校验
+"""
+
+from unittest.mock import MagicMock
+
+from src.core.api_format.conversion.compatibility import is_format_compatible
+
+
+def test_same_format_is_compatible() -> None:
+    ok, needs_conv, reason = is_format_compatible(
+        "CLAUDE",
+        "CLAUDE",
+        endpoint_format_acceptance_config=None,
+        is_stream=False,
+        global_conversion_enabled=False,
+        registry=MagicMock(),
+    )
+    assert ok is True
+    assert needs_conv is False
+    assert reason is None
+
+
+def test_cli_format_not_convertible() -> None:
+    ok, needs_conv, reason = is_format_compatible(
+        "CLAUDE_CLI",
+        "OPENAI",
+        endpoint_format_acceptance_config={"enabled": True},
+        is_stream=False,
+        global_conversion_enabled=True,
+        registry=MagicMock(),
+    )
+    assert ok is False
+    assert needs_conv is False
+    assert reason and "CLI" in reason
+
+
+def test_global_switch_disabled_blocks_conversion() -> None:
+    ok, needs_conv, reason = is_format_compatible(
+        "CLAUDE",
+        "OPENAI",
+        endpoint_format_acceptance_config={"enabled": True},
+        is_stream=False,
+        global_conversion_enabled=False,
+        registry=MagicMock(),
+    )
+    assert ok is False
+    assert needs_conv is False
+    assert reason and "全局" in reason
+
+
+def test_endpoint_config_none_blocks_conversion() -> None:
+    ok, needs_conv, reason = is_format_compatible(
+        "CLAUDE",
+        "OPENAI",
+        endpoint_format_acceptance_config=None,
+        is_stream=False,
+        global_conversion_enabled=True,
+        registry=MagicMock(),
+    )
+    assert ok is False
+    assert needs_conv is False
+    assert reason and "未配置" in reason
+
+
+def test_endpoint_disabled_blocks_conversion() -> None:
+    ok, needs_conv, reason = is_format_compatible(
+        "CLAUDE",
+        "OPENAI",
+        endpoint_format_acceptance_config={"enabled": False},
+        is_stream=False,
+        global_conversion_enabled=True,
+        registry=MagicMock(),
+    )
+    assert ok is False
+    assert needs_conv is False
+    assert reason and "未启用" in reason
+
+
+def test_accept_formats_allows_only_whitelist() -> None:
+    ok, needs_conv, reason = is_format_compatible(
+        "CLAUDE",
+        "OPENAI",
+        endpoint_format_acceptance_config={"enabled": True, "accept_formats": ["OPENAI"]},
+        is_stream=False,
+        global_conversion_enabled=True,
+        registry=MagicMock(),
+    )
+    assert ok is False
+    assert needs_conv is False
+    assert reason and "不接受" in reason
+
+
+def test_reject_formats_blocks_blacklist() -> None:
+    ok, needs_conv, reason = is_format_compatible(
+        "CLAUDE",
+        "OPENAI",
+        endpoint_format_acceptance_config={"enabled": True, "reject_formats": ["CLAUDE"]},
+        is_stream=False,
+        global_conversion_enabled=True,
+        registry=MagicMock(),
+    )
+    assert ok is False
+    assert needs_conv is False
+    assert reason and "拒绝" in reason
+
+
+def test_stream_conversion_disabled_blocks_stream() -> None:
+    ok, needs_conv, reason = is_format_compatible(
+        "CLAUDE",
+        "OPENAI",
+        endpoint_format_acceptance_config={"enabled": True, "stream_conversion": False},
+        is_stream=True,
+        global_conversion_enabled=True,
+        registry=MagicMock(),
+    )
+    assert ok is False
+    assert needs_conv is False
+    assert reason and "流式" in reason
+
+
+def test_converter_support_required() -> None:
+    registry = MagicMock()
+    registry.can_convert_full.return_value = False
+
+    ok, needs_conv, reason = is_format_compatible(
+        "CLAUDE",
+        "OPENAI",
+        endpoint_format_acceptance_config={"enabled": True},
+        is_stream=False,
+        global_conversion_enabled=True,
+        registry=registry,
+    )
+    assert ok is False
+    assert needs_conv is False
+    assert reason and "转换器" in reason
+
+
+def test_conversion_allowed_when_converter_supports_full() -> None:
+    registry = MagicMock()
+    registry.can_convert_full.return_value = True
+
+    ok, needs_conv, reason = is_format_compatible(
+        "CLAUDE",
+        "OPENAI",
+        endpoint_format_acceptance_config={"enabled": True, "accept_formats": ["CLAUDE"]},
+        is_stream=False,
+        global_conversion_enabled=True,
+        registry=registry,
+    )
+    assert ok is True
+    assert needs_conv is True
+    assert reason is None
+

tests/core/api_format/conversion/test_registry.py

@@ -403,4 +403,4 @@ class TestNonStrictConversion:
         original = {"chunk": "data"}
         result = registry.convert_stream_chunk(original, "A", "B")
 
-        assert result == original
+        assert result == [original]

tests/services/test_format_conversion_candidate_selection.py

@@ -0,0 +1,125 @@
+import pytest
+from unittest.mock import AsyncMock, MagicMock
+
+from src.core.api_format import APIFormat, register_all_converters
+from src.services.cache.aware_scheduler import CacheAwareScheduler
+
+
+def _mock_key(key_id: str, api_formats: list[str]) -> MagicMock:
+    key = MagicMock()
+    key.id = key_id
+    key.is_active = True
+    key.api_formats = api_formats
+    key.cache_ttl_minutes = 1
+    key.internal_priority = 1
+    return key
+
+
+def _mock_endpoint(api_format: str, config: dict | None = None) -> MagicMock:
+    endpoint = MagicMock()
+    endpoint.id = f"ep_{api_format.lower()}"
+    endpoint.is_active = True
+    endpoint.api_format = api_format
+    endpoint.format_acceptance_config = config
+    return endpoint
+
+
+@pytest.mark.asyncio
+async def test_build_candidates_blocks_cross_format_when_global_switch_off() -> None:
+    register_all_converters()
+
+    scheduler = CacheAwareScheduler()
+    scheduler._check_model_support = AsyncMock(return_value=(True, None, None, {"m"}))  # type: ignore[attr-defined]
+    scheduler._check_key_availability = MagicMock(return_value=(True, None, None))  # type: ignore[attr-defined]
+
+    provider = MagicMock()
+    provider.name = "p1"
+    provider.endpoints = [
+        _mock_endpoint(
+            "OPENAI",
+            {"enabled": True, "accept_formats": ["CLAUDE"], "stream_conversion": True},
+        )
+    ]
+    provider.api_keys = [_mock_key("k1", ["OPENAI"])]
+
+    candidates = await scheduler._build_candidates(
+        db=MagicMock(),
+        providers=[provider],
+        client_format=APIFormat.CLAUDE,
+        model_name="dummy-model",
+        affinity_key=None,
+        global_conversion_enabled=False,
+    )
+
+    assert candidates == []
+
+
+@pytest.mark.asyncio
+async def test_build_candidates_includes_cross_format_when_enabled() -> None:
+    register_all_converters()
+
+    scheduler = CacheAwareScheduler()
+    scheduler._check_model_support = AsyncMock(return_value=(True, None, None, {"m"}))  # type: ignore[attr-defined]
+    scheduler._check_key_availability = MagicMock(return_value=(True, None, None))  # type: ignore[attr-defined]
+
+    provider = MagicMock()
+    provider.name = "p1"
+    provider.endpoints = [
+        _mock_endpoint(
+            "OPENAI",
+            {"enabled": True, "accept_formats": ["CLAUDE"], "stream_conversion": True},
+        )
+    ]
+    provider.api_keys = [_mock_key("k1", ["OPENAI"])]
+
+    candidates = await scheduler._build_candidates(
+        db=MagicMock(),
+        providers=[provider],
+        client_format=APIFormat.CLAUDE,
+        model_name="dummy-model",
+        affinity_key=None,
+        global_conversion_enabled=True,
+    )
+
+    assert len(candidates) == 1
+    assert candidates[0].needs_conversion is True
+    assert candidates[0].provider_api_format == "OPENAI"
+
+
+@pytest.mark.asyncio
+async def test_exact_matches_rank_before_convertible() -> None:
+    register_all_converters()
+
+    scheduler = CacheAwareScheduler()
+    scheduler._check_model_support = AsyncMock(return_value=(True, None, None, {"m"}))  # type: ignore[attr-defined]
+    scheduler._check_key_availability = MagicMock(return_value=(True, None, None))  # type: ignore[attr-defined]
+
+    provider = MagicMock()
+    provider.name = "p1"
+    # 故意把 OPENAI 放在 endpoints[0]，验证排序仍然是 CLAUDE（exact）在前
+    provider.endpoints = [
+        _mock_endpoint(
+            "OPENAI",
+            {"enabled": True, "accept_formats": ["CLAUDE"], "stream_conversion": True},
+        ),
+        _mock_endpoint("CLAUDE", None),
+    ]
+    provider.api_keys = [
+        _mock_key("k_openai", ["OPENAI"]),
+        _mock_key("k_claude", ["CLAUDE"]),
+    ]
+
+    candidates = await scheduler._build_candidates(
+        db=MagicMock(),
+        providers=[provider],
+        client_format=APIFormat.CLAUDE,
+        model_name="dummy-model",
+        affinity_key=None,
+        global_conversion_enabled=True,
+    )
+
+    assert len(candidates) == 2
+    assert candidates[0].needs_conversion is False
+    assert candidates[0].provider_api_format == "CLAUDE"
+    assert candidates[1].needs_conversion is True
+    assert candidates[1].provider_api_format == "OPENAI"

tests/services/test_format_conversion_health_buckets.py

@@ -0,0 +1,110 @@
+from contextlib import asynccontextmanager
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from src.core.api_format import APIFormat
+from src.services.orchestration.error_classifier import ErrorClassifier
+from src.services.request.executor import RequestExecutor
+
+
+@asynccontextmanager
+async def _noop_async_cm():
+    yield
+
+
+@pytest.mark.asyncio
+async def test_executor_records_health_by_provider_format() -> None:
+    db = MagicMock()
+
+    concurrency_manager = MagicMock()
+    concurrency_manager.get_current_concurrency = AsyncMock(return_value=(0, 0))
+    concurrency_manager.get_key_rpm_count = AsyncMock(return_value=1)
+    concurrency_manager.rpm_guard = MagicMock(return_value=_noop_async_cm())
+
+    adaptive_manager = MagicMock()
+
+    provider = MagicMock()
+    provider.id = "p1"
+    provider.name = "p1"
+
+    endpoint = MagicMock()
+    endpoint.id = "e1"
+    endpoint.api_format = "OPENAI"
+
+    key = MagicMock()
+    key.id = "k1"
+    key.api_key = "encrypted"
+    key.rpm_limit = 10
+    key.learned_rpm_limit = None
+    key.cache_ttl_minutes = 0
+
+    candidate = MagicMock()
+    candidate.provider = provider
+    candidate.endpoint = endpoint
+    candidate.key = key
+    candidate.is_cached = False
+
+    async def request_func(_provider, _endpoint, _key, _candidate):  # noqa: ANN001
+        return {"ok": True}
+
+    with patch("src.services.request.executor.RequestCandidateService.mark_candidate_started"), patch(
+        "src.services.request.executor.RequestCandidateService.mark_candidate_success"
+    ), patch("src.services.request.executor.get_adaptive_reservation_manager") as mock_res_mgr, patch(
+        "src.services.request.executor.health_monitor.record_success"
+    ) as record_success:
+        mock_res_mgr.return_value.calculate_reservation.return_value = MagicMock(
+            ratio=0.0, phase="stable", confidence=1.0
+        )
+
+        executor = RequestExecutor(db=db, concurrency_manager=concurrency_manager, adaptive_manager=adaptive_manager)
+        await executor.execute(
+            candidate=candidate,
+            candidate_id="c1",
+            candidate_index=0,
+            user_api_key=MagicMock(user_id="u1", id="ak1"),
+            request_func=request_func,
+            request_id="r1",
+            api_format=APIFormat.CLAUDE,  # client_format
+            model_name="m",
+            is_stream=False,
+        )
+
+        record_success.assert_called()
+        assert record_success.call_args.kwargs["api_format"] == "OPENAI"
+
+
+@pytest.mark.asyncio
+async def test_error_classifier_records_failure_by_provider_format() -> None:
+    db = MagicMock()
+    classifier = ErrorClassifier(db=db, cache_scheduler=None, adaptive_manager=MagicMock())
+
+    provider = MagicMock()
+    provider.name = "p1"
+
+    endpoint = MagicMock()
+    endpoint.id = "e1"
+    endpoint.api_format = "OPENAI"
+
+    key = MagicMock()
+    key.id = "k1"
+
+    with patch("src.services.orchestration.error_classifier.health_monitor.record_failure") as record_failure:
+        await classifier.handle_retriable_error(
+            error=RuntimeError("boom"),
+            provider=provider,
+            endpoint=endpoint,
+            key=key,
+            affinity_key="aff",
+            api_format=APIFormat.CLAUDE,  # client_format
+            global_model_id="gm1",
+            captured_key_concurrent=None,
+            elapsed_ms=None,
+            request_id="r1",
+            attempt=1,
+            max_attempts=2,
+        )
+
+        record_failure.assert_called()
+        assert record_failure.call_args.kwargs["api_format"] == "OPENAI"
+