i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 17:30:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: 统一代理配置优先级链(key>provider>系统默认)并复用 HTTP 连接池

Browse Source 
- 引入 resolve_proxy_param / build_proxy_client_kwargs 工具函数，统一
  httpx 客户端的代理+SSL+超时配置，替换各模块中零散的 get_ssl_context() 调用
- 所有涉及上游请求的模块（provider_query, usage replay, endpoint check,
  model fetch, OAuth, Vertex Auth, Gemini Files/Video 等）改用
  resolve_effective_proxy 按 key > provider > 系统默认优先级解析代理
- 流式请求改用 HTTPClientPool.get_upstream_client 复用连接池，移除各处
  http_client.aclose() 避免关闭共享客户端
- StreamProcessor._cleanup 不再关闭池中客户端，仅清理响应上下文
- 前端 EndpointFormDialog 增加 body_rules 帮助说明 Popover
- Mock handler 补充 OAuth 字段、endpoint extras 及新增 mock 路由
This commit is contained in:
fawney19
2026-02-11 03:16:53 +08:00
parent cd0acf1e6b
commit bda03d187e
22 changed files with 749 additions and 322 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/api/handlers/base/chat_adapter_base.py
View File

@@ -638,6 +638,8 @@ class ChatAdapterBase(ApiAdapter):
auth_type: str | None = None, # noqa: ARG003
provider_type: str | None = None, # noqa: ARG003
decrypted_auth_config: dict[str, Any] | None = None, # noqa: ARG003
# 代理参数（已解析，直接传递给 run_endpoint_check）
proxy_param: Any | None = None,
) -> dict[str, Any]:
"""
测试模型连接性（非流式）
@@ -700,6 +702,7 @@ class ChatAdapterBase(ApiAdapter):
provider_id=provider_id,
api_key_id=api_key_id,
model_name=model_name or request_data.get("model"),
proxy_param=proxy_param,
)

30
src/api/handlers/base/chat_handler_base.py
View File

@@ -1130,27 +1130,18 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
return _streamified()
# 配置 HTTP 超时
# 注意：read timeout 用于检测连接断开，不是整体请求超时
# 整体请求超时由 asyncio.wait_for 控制，使用全局配置
timeout_config = httpx.Timeout(
connect=config.http_connect_timeout,
read=config.http_read_timeout, # 使用全局配置，用于检测连接断开
write=config.http_write_timeout,
pool=config.http_pool_timeout,
)
# 流式请求使用 stream_first_byte_timeout 作为首字节超时
# 优先使用 Provider 配置，否则使用全局配置
request_timeout = provider.stream_first_byte_timeout or config.stream_first_byte_timeout
# 创建 HTTP 客户端（支持代理配置，Key 级别优先于 Provider 级别）
# 获取 HTTP 客户端（支持代理配置，Key 级别优先于 Provider 级别）
# 使用连接池复用客户端，避免每次流式请求都新建 TCP/TLS 连接
from src.clients.http_client import HTTPClientPool
from src.services.proxy_node.resolver import build_stream_kwargs, resolve_delegate_config
delegate_cfg = resolve_delegate_config(effective_proxy)
http_client = HTTPClientPool.create_upstream_stream_client(
delegate_cfg, proxy_config=effective_proxy, timeout=timeout_config
http_client = await HTTPClientPool.get_upstream_client(
delegate_cfg, proxy_config=effective_proxy
)
# 用于存储内部函数的结果（必须在函数定义前声明，供 nonlocal 使用）
@@ -1214,13 +1205,12 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
break
except ClientDisconnectedException:
# 客户端断开连接，清理资源
# 客户端断开连接，清理响应上下文（不关闭池中复用的客户端）
if response_ctx is not None:
try:
await response_ctx.__aexit__(None, None, None)
except Exception:
pass
await http_client.aclose()
logger.warning(f" [{self.request_id}] 客户端在等待首字节时断开连接")
ctx.status_code = 499
ctx.error_message = "client_disconnected_during_prefetch"
@@ -1228,13 +1218,12 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
except TimeoutError:
# 整体请求超时（建立连接 + 获取首字节）
# 清理可能已建立的连接上下文
# 清理可能已建立的连接上下文（不关闭池中复用的客户端）
if response_ctx is not None:
try:
await response_ctx.__aexit__(None, None, None)
except Exception:
pass
await http_client.aclose()
logger.warning(
f" [{self.request_id}] 请求超时: Provider={provider.name}, timeout={request_timeout}s"
)
@@ -1244,13 +1233,12 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
)
except (httpx.ConnectError, httpx.ConnectTimeout, httpx.TimeoutException) as e:
# 连接/读写超时：清理可能已建立的连接上下文
# 连接/读写超时：清理可能已建立的连接上下文（不关闭池中复用的客户端）
if response_ctx is not None:
try:
await response_ctx.__aexit__(None, None, None)
except Exception:
pass
await http_client.aclose()
if envelope:
envelope.on_connection_error(base_url=ctx.selected_base_url, exc=e)
if ctx.selected_base_url:
@@ -1289,7 +1277,6 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
logger.error(
f"Provider 返回错误: {e.response.status_code}\n Response: {error_text}"
)
await http_client.aclose()
# 将上游错误信息附加到异常，以便故障转移时能够返回给客户端
e.upstream_response = error_text # type: ignore[attr-defined]
raise
@@ -1300,11 +1287,9 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
await response_ctx.__aexit__(None, None, None)
except Exception:
pass
await http_client.aclose()
raise
except Exception:
await http_client.aclose()
raise
# 类型断言：成功执行后这些变量不会为 None
@@ -1317,7 +1302,6 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
ctx,
byte_iterator,
response_ctx,
http_client,
prefetched_chunks,
start_time=self.start_time,
)

3
src/api/handlers/base/cli_adapter_base.py
View File

@@ -604,6 +604,8 @@ class CliAdapterBase(ApiAdapter):
auth_type: str | None = None,
provider_type: str | None = None,
decrypted_auth_config: dict[str, Any] | None = None,
# 代理参数（已解析，直接传递给 run_endpoint_check）
proxy_param: Any | None = None,
) -> dict[str, Any]:
"""
测试模型连接性（非流式）
@@ -779,6 +781,7 @@ class CliAdapterBase(ApiAdapter):
provider_id=provider_id,
api_key_id=api_key_id,
model_name=effective_model_name,
proxy_param=proxy_param,
)
# =========================================================================

40
src/api/handlers/base/cli_handler_base.py
View File

@@ -1092,16 +1092,6 @@ class CliMessageHandlerBase(BaseMessageHandler):
return _streamified()
# 配置 HTTP 超时
# 注意：read timeout 用于检测连接断开，不是整体请求超时
# 整体请求超时由 _connect_and_prefetch 内部的 asyncio.wait_for 控制
timeout_config = httpx.Timeout(
connect=config.http_connect_timeout,
read=config.http_read_timeout, # 使用全局配置，用于检测连接断开
write=config.http_write_timeout,
pool=config.http_pool_timeout,
)
# 流式请求使用 stream_first_byte_timeout 作为首字节超时
# 优先使用 Provider 配置，否则使用全局配置
request_timeout = provider.stream_first_byte_timeout or config.stream_first_byte_timeout
@@ -1116,13 +1106,14 @@ class CliMessageHandlerBase(BaseMessageHandler):
f"timeout={request_timeout}s, 代理={_proxy_label}"
)
# 创建 HTTP 客户端（支持代理配置，Key 级别优先于 Provider 级别）
# 获取 HTTP 客户端（支持代理配置，Key 级别优先于 Provider 级别）
# 使用连接池复用客户端，避免每次流式请求都新建 TCP/TLS 连接
from src.clients.http_client import HTTPClientPool
from src.services.proxy_node.resolver import build_stream_kwargs, resolve_delegate_config
delegate_cfg = resolve_delegate_config(effective_proxy)
http_client = HTTPClientPool.create_upstream_stream_client(
delegate_cfg, proxy_config=effective_proxy, timeout=timeout_config
http_client = await HTTPClientPool.get_upstream_client(
delegate_cfg, proxy_config=effective_proxy
)
# 用于存储内部函数的结果（必须在函数定义前声明，供 nonlocal 使用）
@@ -1188,7 +1179,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
except TimeoutError as e:
# 整体请求超时（建立连接 + 获取首字节）
# 清理可能已建立的连接上下文
# 清理可能已建立的连接上下文（不关闭池中复用的客户端）
if response_ctx is not None:
try:
await response_ctx.__aexit__(None, None, None)
@@ -1196,7 +1187,6 @@ class CliMessageHandlerBase(BaseMessageHandler):
pass
if envelope:
envelope.on_connection_error(base_url=ctx.selected_base_url, exc=e)
await http_client.aclose()
logger.warning(
f" [{self.request_id}] 请求超时: Provider={provider.name}, timeout={request_timeout}s"
)
@@ -1206,13 +1196,12 @@ class CliMessageHandlerBase(BaseMessageHandler):
)
except ClientDisconnectedException:
# 客户端断开连接，清理资源
# 客户端断开连接，清理响应上下文（不关闭池中复用的客户端）
if response_ctx is not None:
try:
await response_ctx.__aexit__(None, None, None)
except Exception:
pass
await http_client.aclose()
logger.warning(f" [{self.request_id}] 客户端在等待首字节时断开连接")
ctx.status_code = 499
ctx.error_message = "client_disconnected_during_prefetch"
@@ -1225,7 +1214,6 @@ class CliMessageHandlerBase(BaseMessageHandler):
logger.warning(
f"[{envelope.name}] Connection error: {ctx.selected_base_url} ({e})"
)
await http_client.aclose()
raise
except httpx.HTTPStatusError as e:
@@ -1258,23 +1246,20 @@ class CliMessageHandlerBase(BaseMessageHandler):
logger.error(
f"Provider 返回错误状态: {e.response.status_code}\n Response: {error_text}"
)
await http_client.aclose()
# 将上游错误信息附加到异常，以便故障转移时能够返回给客户端
e.upstream_response = error_text # type: ignore[attr-defined]
raise
except EmbeddedErrorException:
# 嵌套错误需要触发重试，关闭连接后重新抛出
# 嵌套错误需要触发重试，关闭连接上下文后重新抛出
try:
if response_ctx is not None:
await response_ctx.__aexit__(None, None, None)
except Exception:
pass
await http_client.aclose()
raise
except Exception:
await http_client.aclose()
raise
# 类型断言：成功执行后这些变量不会为 None
@@ -1287,7 +1272,6 @@ class CliMessageHandlerBase(BaseMessageHandler):
ctx,
byte_iterator,
response_ctx,
http_client,
prefetched_chunks,
)
@@ -1296,7 +1280,6 @@ class CliMessageHandlerBase(BaseMessageHandler):
ctx: StreamContext,
stream_response: httpx.Response,
response_ctx: Any,
http_client: httpx.AsyncClient,
) -> AsyncGenerator[bytes]:
"""创建响应流生成器（使用字节流）"""
try:
@@ -1512,10 +1495,6 @@ class CliMessageHandlerBase(BaseMessageHandler):
await response_ctx.__aexit__(None, None, None)
except Exception:
pass
try:
await http_client.aclose()
except Exception:
pass
def _flush_remaining_sse_data(
self,
@@ -1813,7 +1792,6 @@ class CliMessageHandlerBase(BaseMessageHandler):
ctx: StreamContext,
byte_iterator: Any,
response_ctx: Any,
http_client: httpx.AsyncClient,
prefetched_chunks: list,
) -> AsyncGenerator[bytes]:
"""创建响应流生成器（带预读数据，使用字节流）"""
@@ -2089,10 +2067,6 @@ class CliMessageHandlerBase(BaseMessageHandler):
await response_ctx.__aexit__(None, None, None)
except Exception:
pass
try:
await http_client.aclose()
except Exception:
pass
def _handle_sse_event(
self,

19
src/api/handlers/base/endpoint_checker.py
View File

@@ -73,6 +73,7 @@ async def run_endpoint_check(
provider_id: str | None = None,
db: Any | None = None, # Session对象，需要时才导入
user: Any | None = None, # User对象
proxy_param: Any | None = None, # httpx 可接受的代理参数
) -> dict[str, Any]:
"""
执行端点检查（重构版本，使用新的架构）：
@@ -94,6 +95,7 @@ async def run_endpoint_check(
db=db,
user=user,
request_id=str(uuid.uuid4())[:8],
proxy_param=proxy_param,
)
# 使用协调器执行检查
@@ -565,6 +567,7 @@ class EndpointCheckRequest:
user: Any | None = None
request_id: str | None = None
timeout: float = 30.0
proxy_param: Any | None = None # httpx 可接受的代理参数
@dataclass
@@ -595,7 +598,21 @@ class HttpRequestExecutor:
is_stream = request.json_body.get("stream", False) if request.json_body else False
try:
async with httpx.AsyncClient(timeout=self.timeout, verify=get_ssl_context()) as client:
from src.services.proxy_node.resolver import build_proxy_client_kwargs
if request.proxy_param is not None:
# 调用方已提供解析好的代理参数，直接使用（跳过系统默认回退）
client_kwargs: dict[str, Any] = {
"timeout": self.timeout,
"verify": get_ssl_context(),
}
if request.proxy_param:
client_kwargs["proxy"] = request.proxy_param
else:
# 未提供代理参数，通过 build_proxy_client_kwargs 统一解析（含系统默认回退）
client_kwargs = build_proxy_client_kwargs(timeout=self.timeout)
async with httpx.AsyncClient(**client_kwargs) as client:
if is_stream:
# 流式请求：读取 SSE 事件直到完成
response_data = await self._execute_stream_request(client, request)

17
src/api/handlers/base/stream_processor.py
View File

@@ -575,7 +575,6 @@ class StreamProcessor:
ctx: StreamContext,
byte_iterator: Any,
response_ctx: Any,
http_client: httpx.AsyncClient,
prefetched_chunks: list | None = None,
*,
start_time: float | None = None,
@@ -589,7 +588,6 @@ class StreamProcessor:
ctx: 流式上下文
byte_iterator: 字节流迭代器
response_ctx: HTTP 响应上下文管理器
http_client: HTTP 客户端
prefetched_chunks: 预读的字节块列表（可选）
start_time: 请求开始时间,用于计算 TTFB（可选）
@@ -867,7 +865,11 @@ class StreamProcessor:
self._extract_usage_from_converted_event(ctx, evt, event_type)
# 根据客户端格式生成 SSE 事件
out.append(_format_sse_event(evt) if isinstance(evt, dict) else f"data: {json.dumps(evt, ensure_ascii=False)}\n\n".encode())
out.append(
_format_sse_event(evt)
if isinstance(evt, dict)
else f"data: {json.dumps(evt, ensure_ascii=False)}\n\n".encode()
)
return out
# 统一处理 prefetched + iterator
@@ -1109,7 +1111,7 @@ class StreamProcessor:
ctx.perf_metrics["stream_chunks"] = int(ctx.chunk_count)
if ctx.data_count:
ctx.perf_metrics["stream_data_events"] = int(ctx.data_count)
await self._cleanup(response_ctx, http_client)
await self._cleanup(response_ctx)
def _process_line(
self,
@@ -1363,17 +1365,12 @@ class StreamProcessor:
async def _cleanup(
self,
response_ctx: Any,
http_client: httpx.AsyncClient,
) -> None:
"""清理资源"""
"""清理响应上下文（不关闭池中复用的客户端）"""
try:
await response_ctx.__aexit__(None, None, None)
except Exception:
pass
try:
await http_client.aclose()
except Exception:
pass
async def create_smoothed_stream(

3
src/api/handlers/gemini/adapter.py
View File

@@ -269,6 +269,8 @@ class GeminiChatAdapter(ChatAdapterBase):
auth_type: str | None = None,
provider_type: str | None = None,
decrypted_auth_config: dict[str, Any] | None = None,
# 代理参数（已解析，直接传递给 run_endpoint_check）
proxy_param: Any | None = None,
) -> dict[str, Any]:
"""测试 Gemini API 模型连接性（非流式）"""
from src.api.handlers.base.endpoint_checker import run_endpoint_check
@@ -363,6 +365,7 @@ class GeminiChatAdapter(ChatAdapterBase):
provider_id=provider_id,
api_key_id=api_key_id,
model_name=effective_model_name,
proxy_param=proxy_param,
)

18
src/api/handlers/gemini/video_handler.py
View File

@@ -449,9 +449,25 @@ class GeminiVeoHandler(VideoHandlerBase):
import httpx
try:
# 解析代理配置（key > provider > 系统默认）
from src.services.proxy_node.resolver import (
build_proxy_client_kwargs,
resolve_effective_proxy,
)
provider = getattr(endpoint, "provider", None) if endpoint else None
eff_proxy = resolve_effective_proxy(
getattr(provider, "proxy", None) if provider else None,
getattr(key, "proxy", None),
)
# 使用 follow_redirects=True 跟随重定向
async with httpx.AsyncClient(
follow_redirects=True, timeout=httpx.Timeout(300.0)
**build_proxy_client_kwargs(
eff_proxy,
timeout=httpx.Timeout(300.0),
follow_redirects=True,
)
) as client:
response = await client.get(task.video_url, headers=download_headers)
except Exception as exc: