refactor(compression): 请求压缩策略改为跟随客户端行为，响应支持gzip压缩

- 移除全局 ENABLE_REQUEST_COMPRESSION 配置，改为根据客户端 Content-Encoding
  决定是否对上游请求体进行 gzip 压缩
- 非流式响应根据客户端 Accept-Encoding 返回 gzip 压缩的 JSON
- ApiRequestContext 记录客户端编码偏好并透传至 handler 链路
- 新增 http_compression 模块统一处理压缩相关判断逻辑
- 上游请求头丢弃列表新增 content-encoding 防止客户端值泄露
- ensure_json_body 支持解压 gzip 编码的请求体

src/api/base/context.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import gzip
 import json
 import time
 import uuid
@@ -9,6 +10,8 @@ from typing import Any
 from fastapi import HTTPException, Request
 from sqlalchemy.orm import Session
 
+from src.core.api_format.headers import get_header_value
+from src.core.http_compression import is_gzip_content_encoding, normalize_content_encoding
 from src.core.logger import logger
 from src.models.database import ApiKey, ManagementToken, User
 from src.utils.perf import PerfRecorder
@@ -47,6 +50,8 @@ class ApiRequestContext:
 
     # 高频轮询端点日志抑制标志
     quiet_logging: bool = False
+    client_content_encoding: str | None = None
+    client_accept_encoding: str | None = None
 
     def ensure_json_body(self) -> dict[str, Any]:
         """确保请求体已解析为JSON并返回。"""
@@ -67,8 +72,25 @@ class ApiRequestContext:
                 return
             perf_metrics.setdefault("pipeline", {})["json_parse_ms"] = int(duration * 1000)
 
+        body_to_parse = self.raw_body
+        content_encoding = self.client_content_encoding or normalize_content_encoding(
+            get_header_value(self.original_headers, "content-encoding")
+        )
+        if is_gzip_content_encoding(content_encoding):
             try:
-            self.json_body = json.loads(self.raw_body.decode("utf-8"))
+                body_to_parse = gzip.decompress(body_to_parse)
+            except OSError as exc:
+                parse_duration = PerfRecorder.stop(
+                    parse_start,
+                    "pipeline_json_parse",
+                    labels={"mode": self.mode},
+                )
+                _record_parse_duration(parse_duration)
+                logger.warning("gzip 请求体解压失败: {}", exc)
+                raise HTTPException(status_code=400, detail="gzip 请求体解压失败") from exc
+
+        try:
+            self.json_body = json.loads(body_to_parse.decode("utf-8"))
             parse_duration = PerfRecorder.stop(
                 parse_start,
                 "pipeline_json_parse",
@@ -118,6 +140,12 @@ class ApiRequestContext:
         start_time = time.time()
         client_ip = get_client_ip(request)
         user_agent = request.headers.get("user-agent", "unknown")
+        client_content_encoding = normalize_content_encoding(
+            request.headers.get("content-encoding")
+        )
+        client_accept_encoding = request.headers.get("accept-encoding")
+        if isinstance(client_accept_encoding, str):
+            client_accept_encoding = client_accept_encoding.strip() or None
 
         context = cls(
             request=request,
@@ -134,6 +162,8 @@ class ApiRequestContext:
             mode=mode,
             api_format_hint=api_format_hint,
             path_params=path_params or {},
+            client_content_encoding=client_content_encoding,
+            client_accept_encoding=client_accept_encoding,
         )
 
         perf_metrics = getattr(request.state, "perf_metrics", None)

src/api/handlers/base/chat_adapter_base.py

@@ -129,6 +129,7 @@ class ChatAdapterBase(HandlerAdapterBase):
                     original_headers=original_headers,
                     original_request_body=original_request_body,
                     query_params=query_params,
+                    client_content_encoding=context.client_content_encoding,
                 )
             return await handler.process_sync(
                 request=request_obj,
@@ -136,6 +137,8 @@ class ChatAdapterBase(HandlerAdapterBase):
                 original_headers=original_headers,
                 original_request_body=original_request_body,
                 query_params=query_params,
+                client_content_encoding=context.client_content_encoding,
+                client_accept_encoding=context.client_accept_encoding,
             )
 
         except HTTPException:

src/api/handlers/base/chat_handler_base.py

@@ -55,6 +55,7 @@ from src.api.handlers.base.utils import (
     build_sse_headers,
     filter_proxy_response_headers,
     get_format_converter_registry,
+    resolve_client_content_encoding,
 )
 from src.config.settings import config
 from src.core.api_format.conversion.stream_bridge import (
@@ -464,9 +465,14 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
         original_headers: dict[str, Any],
         original_request_body: dict[str, Any],
         query_params: dict[str, str] | None = None,
+        client_content_encoding: str | None = None,
     ) -> StreamingResponse | JSONResponse:
         """处理流式响应"""
         logger.debug(f"开始流式响应处理 ({self.FORMAT_ID})")
+        effective_client_content_encoding = resolve_client_content_encoding(
+            original_headers,
+            client_content_encoding,
+        )
 
         # 转换请求格式
         converted_request = await self._convert_request(request)
@@ -534,6 +540,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
                 query_params,
                 candidate,
                 is_disconnected=http_request.is_disconnected,
+                client_content_encoding=effective_client_content_encoding,
             )
 
         try:
@@ -824,6 +831,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
         query_params: dict[str, str] | None = None,
         candidate: ProviderCandidate | None = None,
         is_disconnected: Callable[[], Awaitable[bool]] | None = None,
+        client_content_encoding: str | None = None,
     ) -> AsyncGenerator[bytes]:
         """执行流式请求并返回流生成器"""
         # 重置上下文状态（重试时清除之前的数据）
@@ -939,6 +947,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
                     headers=provider_headers,
                     payload=provider_payload,
                     timeout=request_timeout_sync,
+                    client_content_encoding=client_content_encoding,
                 )
                 resp = await http_client.post(**_pkw)
             except (httpx.ConnectError, httpx.ConnectTimeout, httpx.TimeoutException) as e:
@@ -978,6 +987,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
                         headers=provider_headers,
                         payload=provider_payload,
                         timeout=request_timeout_sync,
+                        client_content_encoding=client_content_encoding,
                         refresh_auth=True,
                     )
                     resp = await http_client.post(**_pkw)
@@ -1143,6 +1153,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
                 url=url,
                 headers=provider_headers,
                 payload=provider_payload,
+                client_content_encoding=client_content_encoding,
                 # 流式请求不应使用 provider.request_timeout 作为“整条流总时长”超时，
                 # 否则会在长响应中途被硬切断（常见于 request_timeout=15s 的配置）。
                 # 首字节超时由外层 wait_for(stream_first_byte_timeout) 控制；
@@ -1304,11 +1315,19 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
         original_headers: dict[str, Any],
         original_request_body: dict[str, Any],
         query_params: dict[str, str] | None = None,
+        client_content_encoding: str | None = None,
+        client_accept_encoding: str | None = None,
     ) -> JSONResponse:
         """处理非流式响应"""
         from src.api.handlers.base.chat_sync_executor import ChatSyncExecutor
 
         executor = ChatSyncExecutor(self)
         return await executor.execute(
-            request, http_request, original_headers, original_request_body, query_params
+            request,
+            http_request,
+            original_headers,
+            original_request_body,
+            query_params,
+            client_content_encoding=client_content_encoding,
+            client_accept_encoding=client_accept_encoding,
         )

src/api/handlers/base/chat_sync_executor.py

@@ -29,8 +29,11 @@ from src.api.handlers.base.stream_context import (
     is_format_converted,
 )
 from src.api.handlers.base.utils import (
+    build_json_response_for_client,
     filter_proxy_response_headers,
     get_format_converter_registry,
+    resolve_client_accept_encoding,
+    resolve_client_content_encoding,
 )
 from src.core.error_utils import extract_client_error_message
 from src.core.exceptions import (
@@ -87,10 +90,20 @@ class ChatSyncExecutor:
         original_headers: dict[str, Any],
         original_request_body: dict[str, Any],
         query_params: dict[str, str] | None = None,
+        client_content_encoding: str | None = None,
+        client_accept_encoding: str | None = None,
     ) -> JSONResponse:
         """处理非流式响应（原 process_sync 的完整逻辑）"""
         handler = self._handler
         logger.debug(f"开始非流式响应处理 ({handler.FORMAT_ID})")
+        effective_client_content_encoding = resolve_client_content_encoding(
+            original_headers,
+            client_content_encoding,
+        )
+        effective_client_accept_encoding = resolve_client_accept_encoding(
+            original_headers,
+            client_accept_encoding,
+        )
 
         # 转换请求格式
         converted_request = await handler._convert_request(request)
@@ -130,6 +143,7 @@ class ChatSyncExecutor:
                 original_headers=original_headers,
                 request_body_ref=request_body_ref,
                 query_params=query_params,
+                client_content_encoding=effective_client_content_encoding,
             )
 
         try:
@@ -191,6 +205,13 @@ class ChatSyncExecutor:
             # JSONResponse 会自动设置 content-type，但我们记录实际返回的完整头
             client_response_headers = filter_proxy_response_headers(ctx.response_headers)
             client_response_headers["content-type"] = "application/json"
+            client_response = build_json_response_for_client(
+                status_code=ctx.status_code,
+                content=ctx.response_json,
+                headers=client_response_headers,
+                client_accept_encoding=effective_client_accept_encoding,
+            )
+            actual_client_response_headers = dict(client_response.headers)
 
             request_metadata = handler._build_request_metadata() or {}
             if ctx.sync_proxy_info:
@@ -207,7 +228,7 @@ class ChatSyncExecutor:
                 request_headers=original_headers,
                 request_body=original_request_body,
                 response_headers=ctx.response_headers,
-                client_response_headers=client_response_headers,
+                client_response_headers=actual_client_response_headers,
                 response_body=ctx.provider_response_json or ctx.response_json,
                 client_response_body=ctx.response_json if ctx.provider_response_json else None,
                 provider_request_body=ctx.provider_request_body,
@@ -243,11 +264,7 @@ class ChatSyncExecutor:
             )
 
             # 透传提供商的响应头
-            return JSONResponse(
-                status_code=ctx.status_code,
-                content=ctx.response_json,
-                headers=client_response_headers,
-            )
+            return client_response
 
         except ThinkingSignatureException as e:
             # Thinking 签名错误：TaskService 层已处理整流重试但仍失败
@@ -279,9 +296,11 @@ class ChatSyncExecutor:
                 provider_format,
                 needs_conversion=ctx.needs_conversion_for_error,
             )
-            return JSONResponse(
+            return build_json_response_for_client(
                 status_code=_get_error_status_code(e),
                 content=payload,
+                headers={"content-type": "application/json"},
+                client_accept_encoding=effective_client_accept_encoding,
             )
 
         except UpstreamClientException as e:
@@ -289,6 +308,20 @@ class ChatSyncExecutor:
             request_metadata = handler._build_request_metadata() or {}
             if ctx.sync_proxy_info:
                 request_metadata["proxy"] = ctx.sync_proxy_info
+            client_format = (ctx.client_api_format_for_error or "").upper()
+            provider_format = (ctx.provider_api_format_for_error or client_format).upper()
+            payload = _build_error_json_payload(
+                e,
+                client_format,
+                provider_format,
+                needs_conversion=ctx.needs_conversion_for_error,
+            )
+            error_response = build_json_response_for_client(
+                status_code=_get_error_status_code(e),
+                content=payload,
+                headers={"content-type": "application/json"},
+                client_accept_encoding=effective_client_accept_encoding,
+            )
             await handler.telemetry.record_failure(
                 provider=ctx.provider_name or "unknown",
                 model=model,
@@ -304,7 +337,7 @@ class ChatSyncExecutor:
                 endpoint_kind=handler.endpoint_kind,
                 provider_request_headers=ctx.provider_request_headers,
                 response_headers=ctx.response_headers,
-                client_response_headers={"content-type": "application/json"},
+                client_response_headers=dict(error_response.headers),
                 provider_id=ctx.provider_id,
                 provider_endpoint_id=ctx.endpoint_id,
                 provider_api_key_id=ctx.key_id,
@@ -316,18 +349,7 @@ class ChatSyncExecutor:
                 target_model=ctx.mapped_model_result,
                 request_metadata=request_metadata,
             )
-            client_format = (ctx.client_api_format_for_error or "").upper()
-            provider_format = (ctx.provider_api_format_for_error or client_format).upper()
-            payload = _build_error_json_payload(
-                e,
-                client_format,
-                provider_format,
-                needs_conversion=ctx.needs_conversion_for_error,
-            )
-            return JSONResponse(
-                status_code=_get_error_status_code(e),
-                content=payload,
-            )
+            return error_response
 
         except Exception as e:
             response_time_ms = handler.elapsed_ms()
@@ -394,6 +416,7 @@ class ChatSyncExecutor:
         original_headers: dict[str, Any],
         request_body_ref: dict[str, Any],
         query_params: dict[str, str] | None = None,
+        client_content_encoding: str | None = None,
     ) -> dict[str, Any]:
         """单次同步请求（原 sync_request_func 内嵌函数）"""
         handler = self._handler
@@ -520,6 +543,7 @@ class ChatSyncExecutor:
                     headers=provider_hdrs,
                     payload=provider_payload,
                     timeout=request_timeout,
+                    client_content_encoding=client_content_encoding,
                 )
                 resp = await http_client.post(**_pkw)
             except (httpx.ConnectError, httpx.ConnectTimeout, httpx.TimeoutException) as e:
@@ -544,6 +568,7 @@ class ChatSyncExecutor:
                     headers=provider_hdrs,
                     payload=provider_payload,
                     timeout=request_timeout,
+                    client_content_encoding=client_content_encoding,
                 )
                 async with http_client.stream(**_stream_args) as stream_resp:
                     resp = stream_resp

src/api/handlers/base/cli_adapter_base.py

@@ -147,12 +147,15 @@ class CliAdapterBase(HandlerAdapterBase):
                     query_params=query_params,
                     path_params=context.path_params,
                     http_request=http_request,
+                    client_content_encoding=context.client_content_encoding,
                 )
             return await handler.process_sync(
                 original_request_body=original_request_body,
                 original_headers=original_headers,
                 query_params=query_params,
                 path_params=context.path_params,
+                client_content_encoding=context.client_content_encoding,
+                client_accept_encoding=context.client_accept_encoding,
             )
 
         except HTTPException:

src/api/handlers/base/cli_stream_mixin.py

@@ -27,6 +27,7 @@ from src.api.handlers.base.utils import (
     build_sse_headers,
     filter_proxy_response_headers,
     get_format_converter_registry,
+    resolve_client_content_encoding,
 )
 from src.config.settings import config
 from src.core.api_format.conversion.stream_bridge import (
@@ -67,6 +68,7 @@ class CliStreamMixin:
         query_params: dict[str, str] | None = None,
         path_params: dict[str, Any] | None = None,
         http_request: Request | None = None,
+        client_content_encoding: str | None = None,
     ) -> StreamingResponse:
         """
         处理流式请求
@@ -85,6 +87,10 @@ class CliStreamMixin:
             http_request: FastAPI Request 对象，用于检测客户端断连
         """
         logger.debug("开始流式响应处理 ({})", self.FORMAT_ID)
+        effective_client_content_encoding = resolve_client_content_encoding(
+            original_headers,
+            client_content_encoding,
+        )
 
         # 可变请求体容器：允许 TaskService 在遇到 Thinking 签名错误时整流请求体后重试
         # 结构: {"body": 实际请求体, "_rectified": 是否已整流, "_rectified_this_turn": 本轮是否整流}
@@ -138,6 +144,7 @@ class CliStreamMixin:
                 query_params,
                 candidate,
                 http_request,  # 传递 http_request 用于断连检测
+                effective_client_content_encoding,
             )
 
         try:
@@ -243,6 +250,7 @@ class CliStreamMixin:
         query_params: dict[str, str] | None = None,
         candidate: ProviderCandidate | None = None,
         http_request: Request | None = None,
+        client_content_encoding: str | None = None,
     ) -> AsyncGenerator[bytes]:
         """执行流式请求并返回流生成器"""
         from src.models.database import Provider, ProviderAPIKey, ProviderEndpoint
@@ -456,6 +464,7 @@ class CliStreamMixin:
                     headers=provider_headers,
                     payload=provider_payload,
                     timeout=request_timeout_sync,
+                    client_content_encoding=client_content_encoding,
                 )
                 _connect_start = time.monotonic()
                 resp = await http_client.post(**_pkw)
@@ -497,6 +506,7 @@ class CliStreamMixin:
                         headers=provider_headers,
                         payload=provider_payload,
                         timeout=request_timeout_sync,
+                        client_content_encoding=client_content_encoding,
                         refresh_auth=True,
                     )
                     _connect_start = time.monotonic()
@@ -658,6 +668,7 @@ class CliStreamMixin:
                 url=url,
                 headers=provider_headers,
                 payload=provider_payload,
+                client_content_encoding=client_content_encoding,
                 # 流式请求不应使用 provider.request_timeout 作为“整条流总时长”超时，
                 # 否则会在长响应中途被硬切断（常见于 request_timeout=15s 的配置）。
                 # 首字节超时由外层 wait_for(stream_first_byte_timeout) 控制；

src/api/handlers/base/cli_sync_mixin.py

@@ -16,8 +16,11 @@ from src.api.handlers.base.upstream_stream_bridge import (
     aggregate_upstream_stream_to_internal_response,
 )
 from src.api.handlers.base.utils import (
+    build_json_response_for_client,
     filter_proxy_response_headers,
     get_format_converter_registry,
+    resolve_client_accept_encoding,
+    resolve_client_content_encoding,
 )
 from src.config.settings import config
 from src.core.error_utils import extract_client_error_message
@@ -52,6 +55,8 @@ class CliSyncMixin:
         original_headers: dict[str, str],
         query_params: dict[str, str] | None = None,
         path_params: dict[str, Any] | None = None,
+        client_content_encoding: str | None = None,
+        client_accept_encoding: str | None = None,
     ) -> JSONResponse:
         """
         处理非流式请求
@@ -62,6 +67,14 @@ class CliSyncMixin:
         3. 解析响应并记录统计
         """
         logger.debug("开始非流式响应处理 ({})", self.FORMAT_ID)
+        effective_client_content_encoding = resolve_client_content_encoding(
+            original_headers,
+            client_content_encoding,
+        )
+        effective_client_accept_encoding = resolve_client_accept_encoding(
+            original_headers,
+            client_accept_encoding,
+        )
 
         # 使用子类实现的方法提取 model（不同 API 格式的 model 位置不同）
         model = self.extract_model_from_request(original_request_body, path_params)
@@ -303,6 +316,7 @@ class CliSyncMixin:
                         headers=provider_headers,
                         payload=provider_payload,
                         timeout=request_timeout,
+                        client_content_encoding=effective_client_content_encoding,
                     )
                     resp = await http_client.post(**_pkw)
                 except (httpx.ConnectError, httpx.ConnectTimeout, httpx.TimeoutException) as e:
@@ -327,6 +341,7 @@ class CliSyncMixin:
                         headers=provider_headers,
                         payload=provider_payload,
                         timeout=request_timeout,
+                        client_content_encoding=effective_client_content_encoding,
                     )
                     async with http_client.stream(**_stream_args) as stream_resp:
                         resp = stream_resp
@@ -532,6 +547,13 @@ class CliSyncMixin:
             # 非流式成功时，返回给客户端的是提供商响应头（透传）
             client_response_headers = filter_proxy_response_headers(response_headers)
             client_response_headers["content-type"] = "application/json"
+            client_response = build_json_response_for_client(
+                status_code=status_code,
+                content=response_json,
+                headers=client_response_headers,
+                client_accept_encoding=effective_client_accept_encoding,
+            )
+            actual_client_response_headers = dict(client_response.headers)
 
             request_metadata = self._build_request_metadata() or {}
             if sync_proxy_info:
@@ -548,7 +570,7 @@ class CliSyncMixin:
                 request_headers=original_headers,
                 request_body=original_request_body,
                 response_headers=response_headers,
-                client_response_headers=client_response_headers,
+                client_response_headers=actual_client_response_headers,
                 response_body=provider_response_json or response_json,
                 client_response_body=response_json if provider_response_json else None,
                 provider_request_body=provider_request_body,
@@ -576,11 +598,7 @@ class CliSyncMixin:
             logger.info("{} 非流式响应处理完成", self.FORMAT_ID)
 
             # 透传提供商的响应头
-            return JSONResponse(
-                status_code=status_code,
-                content=response_json,
-                headers=client_response_headers,
-            )
+            return client_response
 
         except ThinkingSignatureException as e:
             # Thinking 签名错误：TaskService 层已处理整流重试但仍失败

src/api/handlers/base/utils.py

@@ -4,11 +4,16 @@ Handler 基础工具函数
 
 from __future__ import annotations
 
+import gzip
 import json
 from typing import TYPE_CHECKING, Any
 
+from fastapi.responses import JSONResponse, Response
+
 from src.core.api_format import filter_response_headers
+from src.core.api_format.headers import get_header_value
 from src.core.exceptions import EmbeddedErrorException, ProviderNotAvailableException
+from src.core.http_compression import accepts_gzip, normalize_content_encoding
 from src.core.logger import logger
 
 if TYPE_CHECKING:
@@ -152,6 +157,71 @@ def filter_proxy_response_headers(headers: dict[str, str] | None) -> dict[str, s
     return filter_response_headers(headers)
 
 
+def resolve_client_content_encoding(
+    original_headers: dict[str, str],
+    hinted_content_encoding: str | None = None,
+) -> str | None:
+    """解析客户端请求体编码（优先使用上层透传值）。"""
+    if hinted_content_encoding is not None:
+        return normalize_content_encoding(hinted_content_encoding)
+    return normalize_content_encoding(get_header_value(original_headers, "content-encoding"))
+
+
+def resolve_client_accept_encoding(
+    original_headers: dict[str, str],
+    hinted_accept_encoding: str | None = None,
+) -> str | None:
+    """解析客户端 Accept-Encoding（优先使用上层透传值）。"""
+    if isinstance(hinted_accept_encoding, str):
+        normalized_hint = hinted_accept_encoding.strip()
+        if normalized_hint:
+            return normalized_hint
+    header_value = get_header_value(original_headers, "accept-encoding")
+    normalized_header = header_value.strip()
+    return normalized_header or None
+
+
+def build_json_response_for_client(
+    *,
+    status_code: int,
+    content: Any,
+    headers: dict[str, str] | None,
+    client_accept_encoding: str | None,
+) -> Response:
+    """根据客户端 Accept-Encoding 返回普通或 gzip 压缩 JSON 响应。"""
+    response_headers = dict(headers or {})
+    response_headers.setdefault("content-type", "application/json")
+
+    if not accepts_gzip(client_accept_encoding):
+        return JSONResponse(status_code=status_code, content=content, headers=response_headers)
+
+    json_bytes = json.dumps(content, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+    compressed_bytes = gzip.compress(json_bytes, compresslevel=6)
+
+    cleaned_headers = {
+        key: value
+        for key, value in response_headers.items()
+        if key.lower() not in {"content-length", "content-encoding", "vary"}
+    }
+    cleaned_headers["Content-Encoding"] = "gzip"
+
+    existing_vary = next(
+        (v for k, v in response_headers.items() if k.lower() == "vary"), ""
+    )
+    vary_values = [part.strip() for part in str(existing_vary).split(",") if part.strip()]
+    if not any(part.lower() == "accept-encoding" for part in vary_values):
+        vary_values.append("Accept-Encoding")
+    if vary_values:
+        cleaned_headers["Vary"] = ", ".join(vary_values)
+
+    return Response(
+        status_code=status_code,
+        content=compressed_bytes,
+        headers=cleaned_headers,
+        media_type="application/json",
+    )
+
+
 def check_html_response(line: str) -> bool:
     """
     检查行是否为 HTML 响应（base_url 配置错误的常见症状）

src/config/settings.py

@@ -182,16 +182,6 @@ class Config:
         #   - 三家上游（Claude/OpenAI/Gemini）均已确认支持 HTTP/2
         #   - 出现兼容性问题时可通过环境变量快速回退到 HTTP/1.1
         self.enable_http2 = os.getenv("ENABLE_HTTP2", "true").lower() == "true"
-        # ENABLE_REQUEST_COMPRESSION: 是否对上游请求体启用 gzip 压缩
-        #   - 仅对超过 REQUEST_COMPRESSION_MIN_SIZE 的请求体生效
-        #   - 上游 Cloudflare/Google Front End 层会透明解压
-        self.enable_request_compression = (
-            os.getenv("ENABLE_REQUEST_COMPRESSION", "true").lower() == "true"
-        )
-        # REQUEST_COMPRESSION_MIN_SIZE: 触发请求体压缩的最小字节数
-        #   - gzip 有固定头部开销，小请求压缩后可能反而变大
-        #   - 默认 1024 字节（1KB）
-        self.request_compression_min_size = int(os.getenv("REQUEST_COMPRESSION_MIN_SIZE", "1024"))
 
         # 流式处理配置
         # STREAM_PREFETCH_LINES: 预读行数，用于检测嵌套错误

src/core/api_format/headers.py

@@ -76,6 +76,7 @@ UPSTREAM_DROP_HEADERS: frozenset[str] = frozenset(
         "connection",
         # 编码头 - 丢弃客户端值，由 BROWSER_FINGERPRINT_HEADERS 统一设置
         "accept-encoding",
+        "content-encoding",
         # 反向代理 / 网关注入的头部 - 属于本站基础设施，不应泄露给上游
         "x-real-ip",
         "x-real-proto",

src/core/http_compression.py

@@ -0,0 +1,46 @@
+"""HTTP 压缩相关辅助函数。"""
+
+from __future__ import annotations
+
+
+def normalize_content_encoding(value: str | None) -> str | None:
+    """标准化 Content-Encoding 值（仅做清洗，不做兼容扩展）。"""
+    if not isinstance(value, str):
+        return None
+    normalized = value.strip().lower()
+    return normalized or None
+
+
+def is_gzip_content_encoding(value: str | None) -> bool:
+    """判断 Content-Encoding 是否为 gzip。"""
+    return normalize_content_encoding(value) == "gzip"
+
+
+def accepts_gzip(accept_encoding: str | None) -> bool:
+    """判断 Accept-Encoding 是否可接受 gzip。"""
+    if not isinstance(accept_encoding, str):
+        return False
+
+    for item in accept_encoding.split(","):
+        token_and_params = [part.strip() for part in item.split(";") if part.strip()]
+        if not token_and_params:
+            continue
+
+        encoding = token_and_params[0].lower()
+        if encoding not in {"gzip", "*"}:
+            continue
+
+        quality = 1.0
+        for param in token_and_params[1:]:
+            if not param.lower().startswith("q="):
+                continue
+            try:
+                quality = float(param[2:])
+            except ValueError:
+                quality = 0.0
+            break
+
+        if quality > 0:
+            return True
+
+    return False

src/services/proxy_node/resolver.py

@@ -16,8 +16,8 @@ from urllib.parse import quote, urlparse
 
 import httpx
 
-from src.config import config
 from src.core.exceptions import ProxyNodeUnavailableError
+from src.core.http_compression import is_gzip_content_encoding
 from src.core.logger import logger
 
 # ---------------------------------------------------------------------------
@@ -620,9 +620,10 @@ def resolve_delegate_config(proxy_config: dict[str, Any] | None) -> dict[str, An
 def _maybe_compress_payload(
     payload: Any,
     headers: dict[str, str],
+    client_content_encoding: str | None = None,
 ) -> tuple[bytes, dict[str, str]]:
     """
-    将 payload 序列化为 JSON bytes，按配置决定是否 gzip 压缩。
+    将 payload 序列化为 JSON bytes，并按客户端请求行为决定是否 gzip 压缩。
 
     NOTE: 使用紧凑分隔符 ``(",", ":")`` 序列化（无空格），相比 httpx ``json=``
     参数的默认 ``json.dumps``（带空格分隔符）体积更小，所有上游 API 均兼容。
@@ -631,14 +632,14 @@ def _maybe_compress_payload(
         (body_bytes, updated_headers)
     """
     json_bytes = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+    normalized_headers = {k: v for k, v in headers.items() if k.lower() != "content-encoding"}
 
-    if config.enable_request_compression and len(json_bytes) >= config.request_compression_min_size:
+    if is_gzip_content_encoding(client_content_encoding):
         compressed = gzip.compress(json_bytes, compresslevel=6)
-        if len(compressed) < len(json_bytes):
-            headers = {**headers, "Content-Encoding": "gzip"}
-            return compressed, headers
+        normalized_headers = {**normalized_headers, "Content-Encoding": "gzip"}
+        return compressed, normalized_headers
 
-    return json_bytes, headers
+    return json_bytes, normalized_headers
 
 
 def build_post_kwargs(
@@ -648,6 +649,7 @@ def build_post_kwargs(
     headers: dict[str, str],
     payload: Any,
     timeout: float,
+    client_content_encoding: str | None = None,
     refresh_auth: bool = False,
 ) -> dict[str, Any]:
     """
@@ -658,7 +660,11 @@ def build_post_kwargs(
     ``_delegate_cfg`` 和 ``refresh_auth`` 已废弃（tunnel 模式下认证由 transport 层处理），
     保留仅为兼容现有调用方签名。
     """
-    content, final_headers = _maybe_compress_payload(payload, headers)
+    content, final_headers = _maybe_compress_payload(
+        payload,
+        headers,
+        client_content_encoding=client_content_encoding,
+    )
     return {
         "url": url,
         "content": content,
@@ -674,6 +680,7 @@ def build_stream_kwargs(
     headers: dict[str, str],
     payload: Any,
     timeout: float | None = None,
+    client_content_encoding: str | None = None,
 ) -> dict[str, Any]:
     """
     构建上游 stream 请求的 httpx kwargs
@@ -683,7 +690,11 @@ def build_stream_kwargs(
 
     ``_delegate_cfg`` 已废弃，保留仅为兼容现有调用方签名。
     """
-    content, final_headers = _maybe_compress_payload(payload, headers)
+    content, final_headers = _maybe_compress_payload(
+        payload,
+        headers,
+        client_content_encoding=client_content_encoding,
+    )
     kwargs: dict[str, Any] = {
         "method": "POST",
         "url": url,

tests/api/handlers/base/test_utils.py

@@ -1,11 +1,17 @@
 """测试 handler 基础工具函数"""
 
+import gzip
+import json
+
 import pytest
 
 from src.api.handlers.base.utils import (
+    build_json_response_for_client,
     build_sse_headers,
     extract_cache_creation_tokens,
     filter_proxy_response_headers,
+    resolve_client_accept_encoding,
+    resolve_client_content_encoding,
 )
 
 
@@ -135,3 +141,51 @@ class TestFilterProxyResponseHeaders:
 
         assert result["X-Request-Id"] == "abc"
         assert result["Anthropic-RateLimit-Requests-Remaining"] == "10"
+
+
+class TestResolveClientEncoding:
+    def test_content_encoding_prefers_hint(self) -> None:
+        headers = {"content-encoding": "gzip"}
+        result = resolve_client_content_encoding(headers, hinted_content_encoding="br")
+        assert result == "br"
+
+    def test_content_encoding_fallback_to_headers(self) -> None:
+        headers = {"Content-Encoding": "gzip"}
+        result = resolve_client_content_encoding(headers)
+        assert result == "gzip"
+
+    def test_accept_encoding_prefers_hint(self) -> None:
+        headers = {"accept-encoding": "gzip"}
+        result = resolve_client_accept_encoding(headers, hinted_accept_encoding="br")
+        assert result == "br"
+
+    def test_accept_encoding_fallback_to_headers(self) -> None:
+        headers = {"Accept-Encoding": "gzip, deflate"}
+        result = resolve_client_accept_encoding(headers)
+        assert result == "gzip, deflate"
+
+
+class TestBuildJsonResponseForClient:
+    def test_returns_gzip_response_when_client_accepts_gzip(self) -> None:
+        response = build_json_response_for_client(
+            status_code=200,
+            content={"ok": True},
+            headers={"content-type": "application/json"},
+            client_accept_encoding="gzip, deflate",
+        )
+
+        assert response.headers.get("content-encoding") == "gzip"
+        assert "accept-encoding" in response.headers.get("vary", "").lower()
+        decompressed = gzip.decompress(bytes(response.body))
+        assert json.loads(decompressed.decode("utf-8")) == {"ok": True}
+
+    def test_returns_plain_json_when_gzip_not_accepted(self) -> None:
+        response = build_json_response_for_client(
+            status_code=200,
+            content={"ok": True},
+            headers={"content-type": "application/json"},
+            client_accept_encoding="gzip;q=0, deflate",
+        )
+
+        assert response.headers.get("content-encoding") is None
+        assert json.loads(bytes(response.body).decode("utf-8")) == {"ok": True}

tests/unit/test_api_request_context.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import gzip
+import json
+
+import pytest
+from fastapi import HTTPException
+from starlette.requests import Request
+
+from src.api.base.context import ApiRequestContext
+
+
+def _build_request(headers: dict[str, str] | None = None) -> Request:
+    header_items = [
+        (str(key).encode("latin-1"), str(value).encode("latin-1"))
+        for key, value in (headers or {}).items()
+    ]
+    scope = {
+        "type": "http",
+        "http_version": "1.1",
+        "method": "POST",
+        "scheme": "http",
+        "path": "/v1/messages",
+        "raw_path": b"/v1/messages",
+        "query_string": b"",
+        "headers": header_items,
+        "client": ("127.0.0.1", 12345),
+        "server": ("testserver", 80),
+    }
+
+    async def receive() -> dict[str, object]:
+        return {"type": "http.request", "body": b"", "more_body": False}
+
+    request = Request(scope, receive)
+    request.state.perf_metrics = {}
+    return request
+
+
+def _build_context(raw_body: bytes, headers: dict[str, str] | None = None) -> ApiRequestContext:
+    request = _build_request(headers=headers)
+    return ApiRequestContext(
+        request=request,
+        db=None,  # type: ignore[arg-type]
+        user=None,
+        api_key=None,
+        request_id="req_test",
+        start_time=0.0,
+        client_ip="127.0.0.1",
+        user_agent="pytest",
+        original_headers=headers or {},
+        query_params={},
+        raw_body=raw_body,
+    )
+
+
+class TestApiRequestContextEnsureJsonBody:
+    def test_decompresses_gzip_body(self) -> None:
+        payload = {"message": "hello", "count": 2}
+        raw_body = gzip.compress(json.dumps(payload).encode("utf-8"))
+        context = _build_context(raw_body, headers={"content-encoding": "gzip"})
+
+        result = context.ensure_json_body()
+
+        assert result == payload
+
+    def test_rejects_invalid_gzip_body(self) -> None:
+        context = _build_context(b"not-gzip-body", headers={"content-encoding": "gzip"})
+
+        with pytest.raises(HTTPException) as exc_info:
+            context.ensure_json_body()
+
+        assert exc_info.value.status_code == 400
+        assert exc_info.value.detail == "gzip 请求体解压失败"
+
+    def test_build_records_client_encoding_preferences(self) -> None:
+        request = _build_request(
+            headers={
+                "content-encoding": "gzip",
+                "accept-encoding": "gzip, deflate",
+            }
+        )
+        context = ApiRequestContext.build(
+            request=request,
+            db=None,  # type: ignore[arg-type]
+            user=None,
+            api_key=None,
+            raw_body=b"{}",
+        )
+
+        assert context.client_content_encoding == "gzip"
+        assert context.client_accept_encoding == "gzip, deflate"

tests/unit/test_headers.py

@@ -101,6 +101,7 @@ class TestBuildUpstreamHeaders:
                 "X-Api-Key": "client",
                 "User-Agent": "ua",
                 "Content-Type": "text/plain",
+                "Content-Encoding": "gzip",
             },
             "openai:chat",
             "provider",
@@ -112,6 +113,7 @@ class TestBuildUpstreamHeaders:
         assert result["Authorization"] == "Bearer provider"
         assert result["User-Agent"] == "extra"
         assert result["Content-Type"] == "text/plain"
+        assert "Content-Encoding" not in result
         assert result["X-Endpoint"] == "1"
         assert result["X-Extra"] == "1"
 

tests/unit/test_proxy_resolver_compression.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import gzip
+import json
+
+from src.services.proxy_node.resolver import build_post_kwargs, build_stream_kwargs
+
+
+class TestProxyResolverCompression:
+    def test_build_post_kwargs_compresses_when_client_sent_gzip(self) -> None:
+        payload = {"message": "hello", "tokens": [1, 2, 3]}
+        kwargs = build_post_kwargs(
+            None,
+            url="https://example.com/v1/messages",
+            headers={"Content-Type": "application/json"},
+            payload=payload,
+            timeout=10.0,
+            client_content_encoding="gzip",
+        )
+
+        assert kwargs["headers"]["Content-Encoding"] == "gzip"
+        assert json.loads(gzip.decompress(kwargs["content"]).decode("utf-8")) == payload
+
+    def test_build_post_kwargs_keeps_plain_body_without_client_gzip(self) -> None:
+        payload = {"message": "plain"}
+        kwargs = build_post_kwargs(
+            None,
+            url="https://example.com/v1/messages",
+            headers={"Content-Type": "application/json"},
+            payload=payload,
+            timeout=10.0,
+            client_content_encoding=None,
+        )
+
+        assert "Content-Encoding" not in kwargs["headers"]
+        assert json.loads(kwargs["content"].decode("utf-8")) == payload
+
+    def test_build_stream_kwargs_drops_stale_content_encoding_header(self) -> None:
+        kwargs = build_stream_kwargs(
+            None,
+            url="https://example.com/v1/messages",
+            headers={"content-encoding": "gzip", "Content-Type": "application/json"},
+            payload={"message": "no-gzip"},
+            timeout=10.0,
+            client_content_encoding=None,
+        )
+
+        assert all(key.lower() != "content-encoding" for key in kwargs["headers"])