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

Merge branch 'fix/python314-upgrade'

Browse Source 
# Conflicts:
#	src/api/handlers/base/base_handler.py
#	src/api/handlers/base/request_builder.py
#	src/models/endpoint_models.py
#	src/services/orchestration/candidate_resolver.py
#	src/services/orchestration/fallback_orchestrator.py
This commit is contained in:
fawney19
2026-01-30 12:59:52 +08:00
parent c33cd07ea9 f8ca3773be
commit cd1b103223
257 changed files with 4115 additions and 5236 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/services/orchestration/candidate_resolver.py
View File

@@ -6,7 +6,7 @@
import uuid
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple
from typing import Any
from sqlalchemy.orm import Session
@@ -48,12 +48,12 @@ class CandidateResolver:
api_format: APIFormat,
model_name: str,
affinity_key: str,
user_api_key: Optional[ApiKey] = None,
request_id: Optional[str] = None,
user_api_key: ApiKey | None = None,
request_id: str | None = None,
is_stream: bool = False,
capability_requirements: Optional[Dict[str, bool]] = None,
preferred_key_ids: Optional[list[str]] = None,
) -> Tuple[List[ProviderCandidate], str]:
capability_requirements: dict[str, bool] | None = None,
preferred_key_ids: list[str] | None = None,
) -> tuple[list[ProviderCandidate], str]:
"""
获取所有可用候选
@@ -73,10 +73,10 @@ class CandidateResolver:
Raises:
ProviderNotAvailableException: 没有找到任何可用候选时
"""
all_candidates: List[ProviderCandidate] = []
all_candidates: list[ProviderCandidate] = []
provider_offset = 0
provider_batch_size = 20
global_model_id: Optional[str] = None
global_model_id: str | None = None
while True:
candidates, resolved_global_model_id = await self.cache_scheduler.list_all_candidates(
@@ -136,12 +136,12 @@ class CandidateResolver:
def create_candidate_records(
self,
all_candidates: List[ProviderCandidate],
request_id: Optional[str],
all_candidates: list[ProviderCandidate],
request_id: str | None,
user_id: str,
user_api_key: ApiKey,
required_capabilities: Optional[Dict[str, bool]] = None,
) -> Dict[Tuple[int, int], str]:
required_capabilities: dict[str, bool] | None = None,
) -> dict[tuple[int, int], str]:
"""
为所有候选预先创建 available 状态记录（批量插入优化）
@@ -157,8 +157,8 @@ class CandidateResolver:
"""
from src.models.database import RequestCandidate
candidate_records_to_insert: List[Dict[str, Any]] = []
candidate_record_map: Dict[Tuple[int, int], str] = {}
candidate_records_to_insert: list[dict[str, Any]] = []
candidate_record_map: dict[tuple[int, int], str] = {}
# 只保存启用的能力（值为 True 的）
active_capabilities = None
@@ -232,8 +232,8 @@ class CandidateResolver:
def get_active_candidates(
self,
all_candidates: List[ProviderCandidate],
) -> List[Tuple[int, ProviderCandidate]]:
all_candidates: list[ProviderCandidate],
) -> list[tuple[int, ProviderCandidate]]:
"""
获取所有非跳过的候选（带索引）
@@ -247,7 +247,7 @@ class CandidateResolver:
def count_total_attempts(
self,
all_candidates: List[ProviderCandidate],
all_candidates: list[ProviderCandidate],
) -> int:
"""
计算总尝试次数

56
src/services/orchestration/error_classifier.py
View File

@@ -6,7 +6,7 @@
import json
from enum import Enum
from typing import Any, Dict, Optional, Tuple, Union
from typing import Any
import httpx
from sqlalchemy.orm import Session
@@ -51,7 +51,7 @@ class ErrorClassifier:
"""
# 需要触发故障转移的错误类型
RETRIABLE_ERRORS: Tuple[type, ...] = (
RETRIABLE_ERRORS: tuple[type, ...] = (
ProviderException, # 包含所有 Provider 异常子类
ConnectionError, # Python 标准连接错误
TimeoutError, # Python 标准超时错误
@@ -59,7 +59,7 @@ class ErrorClassifier:
)
# 不可重试的错误类型（直接抛出）
NON_RETRIABLE_ERRORS: Tuple[type, ...] = (
NON_RETRIABLE_ERRORS: tuple[type, ...] = (
ValueError, # 参数错误
TypeError, # 类型错误
KeyError, # 键错误
@@ -73,7 +73,7 @@ class ErrorClassifier:
#
# 重要：不要在此列表中包含 Provider Key 配置问题（如 invalid_api_key）
# 这类错误应该触发故障转移，而不是直接返回给用户
CLIENT_ERROR_PATTERNS: Tuple[str, ...] = (
CLIENT_ERROR_PATTERNS: tuple[str, ...] = (
"could not process image", # 图片处理失败
"image too large", # 图片过大
"invalid image", # 无效图片
@@ -101,7 +101,7 @@ class ErrorClassifier:
self,
db: Session,
adaptive_manager: Any = None,
cache_scheduler: Optional[CacheAwareScheduler] = None,
cache_scheduler: CacheAwareScheduler | None = None,
) -> None:
"""
初始化错误分类器
@@ -117,7 +117,7 @@ class ErrorClassifier:
# 表示客户端错误的 error type（不区分大小写）
# 这些 type 表明是请求本身的问题，不应重试
CLIENT_ERROR_TYPES: Tuple[str, ...] = (
CLIENT_ERROR_TYPES: tuple[str, ...] = (
# Claude/OpenAI 标准
"invalid_request_error",
# Gemini
@@ -131,7 +131,7 @@ class ErrorClassifier:
)
# 表示客户端错误的 reason/code 字段值
CLIENT_ERROR_REASONS: Tuple[str, ...] = (
CLIENT_ERROR_REASONS: tuple[str, ...] = (
"CONTENT_LENGTH_EXCEEDS_THRESHOLD",
"CONTEXT_LENGTH_EXCEEDED",
"MAX_TOKENS_EXCEEDED",
@@ -141,7 +141,7 @@ class ErrorClassifier:
# Provider 兼容性错误模式 - 这类错误应该触发故障转移
# 因为换一个 Provider 可能就能成功
COMPATIBILITY_ERROR_PATTERNS: Tuple[str, ...] = (
COMPATIBILITY_ERROR_PATTERNS: tuple[str, ...] = (
"unsupported parameter", # 不支持的参数
"unsupported model", # 不支持的模型
"unsupported feature", # 不支持的功能
@@ -154,7 +154,7 @@ class ErrorClassifier:
# Thinking 块相关错误模式 - 这类错误需要清洗 thinking 块或调整请求
# 场景：多供应商环境下，Provider A 生成的 thinking 块被发送到 Provider B 时签名验证失败
THINKING_ERROR_PATTERNS: Tuple[str, ...] = (
THINKING_ERROR_PATTERNS: tuple[str, ...] = (
# 签名错误：跨 Provider 发送 thinking 块时，签名无法被目标 Provider 验证
# 例: "invalid `signature` in `thinking` block: signature is for a different request"
"invalid `signature` in `thinking` block",
@@ -176,7 +176,7 @@ class ErrorClassifier:
"expected `redacted_thinking`, found",
)
def _parse_error_response(self, error_text: Optional[str]) -> Dict[str, Any]:
def _parse_error_response(self, error_text: str | None) -> dict[str, Any]:
"""
解析错误响应为结构化数据
@@ -265,7 +265,7 @@ class ErrorClassifier:
return result
def is_client_error(self, error_text: Optional[str]) -> bool:
def is_client_error(self, error_text: str | None) -> bool:
"""
检测错误响应是否为客户端错误（不应重试）
@@ -301,7 +301,7 @@ class ErrorClassifier:
search_text = f"{parsed['message']} {parsed['raw']}".lower()
return any(pattern.lower() in search_text for pattern in self.CLIENT_ERROR_PATTERNS)
def _is_compatibility_error(self, error_text: Optional[str]) -> bool:
def _is_compatibility_error(self, error_text: str | None) -> bool:
"""
检测错误响应是否为 Provider 兼容性错误（应触发故障转移）
@@ -320,7 +320,7 @@ class ErrorClassifier:
search_text = error_text.lower()
return any(pattern.lower() in search_text for pattern in self.COMPATIBILITY_ERROR_PATTERNS)
def _is_thinking_error(self, error_text: Optional[str]) -> bool:
def _is_thinking_error(self, error_text: str | None) -> bool:
"""
检测错误响应是否为 Thinking 块相关错误（签名错误或结构错误）
@@ -339,7 +339,7 @@ class ErrorClassifier:
search_text = error_text.lower()
return any(p.lower() in search_text for p in self.THINKING_ERROR_PATTERNS)
def _extract_error_message(self, error_text: Optional[str]) -> Optional[str]:
def _extract_error_message(self, error_text: str | None) -> str | None:
"""
从错误响应中提取错误消息
@@ -404,9 +404,9 @@ class ErrorClassifier:
self,
key: ProviderAPIKey,
provider_name: str,
current_rpm: Optional[int],
current_rpm: int | None,
exception: ProviderRateLimitException,
request_id: Optional[str] = None,
request_id: str | None = None,
) -> str:
"""
处理 429 速率限制错误的自适应调整
@@ -468,8 +468,8 @@ class ErrorClassifier:
self,
error: httpx.HTTPStatusError,
provider_name: str,
error_response_text: Optional[str] = None,
) -> Union[ProviderException, UpstreamClientException]:
error_response_text: str | None = None,
) -> ProviderException | UpstreamClientException:
"""
转换 HTTP 错误为 Provider 异常
@@ -559,14 +559,14 @@ class ErrorClassifier:
endpoint: ProviderEndpoint,
key: ProviderAPIKey,
affinity_key: str,
api_format: Union[str, APIFormat],
api_format: str | APIFormat,
global_model_id: str,
request_id: Optional[str],
captured_key_concurrent: Optional[int],
elapsed_ms: Optional[int],
request_id: str | None,
captured_key_concurrent: int | None,
elapsed_ms: int | None,
attempt: int,
max_attempts: int,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""
处理 HTTP 错误，返回 extra_data
@@ -609,7 +609,7 @@ class ErrorClassifier:
converted_error = self.convert_http_error(http_error, provider_name, error_response_text)
# 构建 extra_data，包含转换后的异常
extra_data: Dict[str, Any] = {
extra_data: dict[str, Any] = {
"converted_error": converted_error,
}
if error_response_text:
@@ -702,11 +702,11 @@ class ErrorClassifier:
endpoint: ProviderEndpoint,
key: ProviderAPIKey,
affinity_key: str,
api_format: Union[str, APIFormat],
api_format: str | APIFormat,
global_model_id: str,
captured_key_concurrent: Optional[int],
elapsed_ms: Optional[int],
request_id: Optional[str],
captured_key_concurrent: int | None,
elapsed_ms: int | None,
request_id: str | None,
attempt: int,
max_attempts: int,
) -> None:

112
src/services/orchestration/fallback_orchestrator.py
View File

@@ -21,9 +21,10 @@
- 本类作为协调者，组合使用上述组件
"""
from __future__ import annotations
from typing import Any, Callable, Dict, List, NoReturn, Optional, Tuple, Union
from typing import Any, NoReturn
from collections.abc import Callable
import httpx
from redis import Redis
@@ -80,7 +81,7 @@ class FallbackOrchestrator:
- 优势：可预测、高效、公平、资源友好
"""
def __init__(self, db: Session, redis_client: Optional[Redis] = None) -> None:
def __init__(self, db: Session, redis_client: Redis | None = None) -> None:
"""
初始化编排器
@@ -90,15 +91,15 @@ class FallbackOrchestrator:
"""
self.db = db
self.redis = redis_client
self.cache_scheduler: Optional[CacheAwareScheduler] = None
self.cache_scheduler: CacheAwareScheduler | None = None
self.concurrency_manager: Any = None
self.adaptive_manager = get_adaptive_rpm_manager() # 自适应 RPM 管理器
self.request_executor: Optional[RequestExecutor] = None
self.request_executor: RequestExecutor | None = None
# 拆分后的组件（延迟初始化）
self._candidate_resolver: Optional[CandidateResolver] = None
self._request_dispatcher: Optional[RequestDispatcher] = None
self._error_classifier: Optional[ErrorClassifier] = None
self._candidate_resolver: CandidateResolver | None = None
self._request_dispatcher: RequestDispatcher | None = None
self._error_classifier: ErrorClassifier | None = None
async def _ensure_initialized(self) -> None:
"""确保异步组件已初始化"""
@@ -172,12 +173,12 @@ class FallbackOrchestrator:
api_format: APIFormat,
model_name: str,
affinity_key: str,
user_api_key: Optional[ApiKey] = None,
request_id: Optional[str] = None,
user_api_key: ApiKey | None = None,
request_id: str | None = None,
is_stream: bool = False,
capability_requirements: Optional[Dict[str, bool]] = None,
preferred_key_ids: Optional[list[str]] = None,
) -> Tuple[List[ProviderCandidate], str]:
capability_requirements: dict[str, bool] | None = None,
preferred_key_ids: list[str] | None = None,
) -> tuple[list[ProviderCandidate], str]:
"""
收集所有可用的 Provider/Endpoint/Key 候选组合
@@ -192,6 +193,7 @@ class FallbackOrchestrator:
is_stream: 是否是流式请求，如果为 True 则过滤不支持流式的 Provider
capability_requirements: 能力需求（用于过滤不满足能力要求的 Key）
preferred_key_ids: 优先使用的 Provider Key ID 列表（匹配则置顶）
preferred_key_ids: 优先使用的 Provider Key ID 列表（匹配则置顶）
Returns:
(所有候选组合的列表, global_model_id)
@@ -213,12 +215,12 @@ class FallbackOrchestrator:
def _create_candidate_records(
self,
all_candidates: List[ProviderCandidate],
request_id: Optional[str],
all_candidates: list[ProviderCandidate],
request_id: str | None,
user_id: str,
user_api_key: ApiKey,
required_capabilities: Optional[Dict[str, bool]] = None,
) -> Dict[Tuple[int, int], str]:
required_capabilities: dict[str, bool] | None = None,
) -> dict[tuple[int, int], str]:
"""
为所有候选预先创建 available 状态记录（批量插入优化）
@@ -251,7 +253,7 @@ class FallbackOrchestrator:
candidate_record_id: str,
user_api_key: ApiKey,
request_func: Callable[..., Any],
request_id: Optional[str],
request_id: str | None,
api_format: APIFormat,
model_name: str,
affinity_key: str,
@@ -259,7 +261,7 @@ class FallbackOrchestrator:
attempt_counter: int,
max_attempts: int,
is_stream: bool = False,
) -> Tuple[Any, str, str, str, str, str]:
) -> tuple[Any, str, str, str, str, str]:
"""
尝试单个候选执行请求
@@ -308,12 +310,12 @@ class FallbackOrchestrator:
def _handle_thinking_signature_error(
self,
converted_error: ThinkingSignatureException,
request_id: Optional[str],
request_id: str | None,
candidate_record_id: str,
elapsed_ms: int,
captured_key_concurrent: Optional[int],
serializable_extra_data: Dict[str, Any],
request_body_ref: Optional[Dict[str, Any]],
captured_key_concurrent: int | None,
serializable_extra_data: dict[str, Any],
request_body_ref: dict[str, Any] | None,
) -> str:
"""
处理 ThinkingSignatureException 错误
@@ -398,8 +400,8 @@ class FallbackOrchestrator:
candidate_record_id: str,
error: ThinkingSignatureException,
elapsed_ms: int,
captured_key_concurrent: Optional[int],
extra_data: Dict[str, Any],
captured_key_concurrent: int | None,
extra_data: dict[str, Any],
) -> None:
"""标记 Thinking 签名错误导致的候选失败"""
RequestCandidateService.mark_candidate_failed(
@@ -423,10 +425,10 @@ class FallbackOrchestrator:
affinity_key: str,
api_format: APIFormat,
global_model_id: str,
request_id: Optional[str],
request_id: str | None,
attempt: int,
max_attempts: int,
request_body_ref: Optional[Dict[str, Any]] = None,
request_body_ref: dict[str, Any] | None = None,
) -> str:
"""
处理候选执行错误
@@ -654,7 +656,7 @@ class FallbackOrchestrator:
def _create_pending_usage_record(
self,
request_id: Optional[str],
request_id: str | None,
user_api_key: ApiKey,
model_name: str,
is_stream: bool,
@@ -685,23 +687,23 @@ class FallbackOrchestrator:
async def _execute_candidates_loop(
self,
all_candidates: List[ProviderCandidate],
candidate_record_map: Dict[Tuple[int, int], str],
all_candidates: list[ProviderCandidate],
candidate_record_map: dict[tuple[int, int], str],
user_api_key: ApiKey,
request_func: Callable[..., Any],
request_id: Optional[str],
request_id: str | None,
api_format_enum: APIFormat,
model_name: str,
affinity_key: str,
global_model_id: str,
is_stream: bool = False,
request_body_ref: Optional[Dict[str, Any]] = None,
) -> Tuple[Any, str, Optional[str], Optional[str], Optional[str], Optional[str]]:
request_body_ref: dict[str, Any] | None = None,
) -> tuple[Any, str, str | None, str | None, str | None, str | None]:
"""遍历所有候选执行请求，返回第一个成功的结果或抛出异常"""
attempt_counter = 0
max_attempts = 0
last_error: Optional[Exception] = None
last_candidate: Optional[ProviderCandidate] = None
last_error: Exception | None = None
last_candidate: ProviderCandidate | None = None
for candidate_index, candidate in enumerate(all_candidates):
last_candidate = candidate
@@ -731,8 +733,8 @@ class FallbackOrchestrator:
)
if result["success"]:
response: Tuple[
Any, str, Optional[str], Optional[str], Optional[str], Optional[str]
response: tuple[
Any, str, str | None, str | None, str | None, str | None
] = result["response"]
return response
@@ -756,10 +758,10 @@ class FallbackOrchestrator:
self,
candidate: ProviderCandidate,
candidate_index: int,
candidate_record_map: Dict[Tuple[int, int], str],
candidate_record_map: dict[tuple[int, int], str],
user_api_key: ApiKey,
request_func: Callable[..., Any],
request_id: Optional[str],
request_id: str | None,
api_format_enum: APIFormat,
model_name: str,
affinity_key: str,
@@ -767,14 +769,14 @@ class FallbackOrchestrator:
attempt_counter: int,
max_attempts: int,
is_stream: bool = False,
request_body_ref: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
request_body_ref: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""尝试单个候选（含重试逻辑），返回执行结果"""
provider = candidate.provider
endpoint = candidate.endpoint
# 从 Provider 读取 max_retries（已从 Endpoint 迁移）
max_retries_for_candidate = int(provider.max_retries or 2) if candidate.is_cached else 1
last_error: Optional[Exception] = None
last_error: Exception | None = None
retry_index = 0
while retry_index < max_retries_for_candidate:
@@ -879,8 +881,8 @@ class FallbackOrchestrator:
def _attach_metadata_to_error(
self,
error: Optional[Exception],
candidate: Optional[ProviderCandidate],
error: Exception | None,
candidate: ProviderCandidate | None,
model_name: str,
api_format_enum: APIFormat,
) -> None:
@@ -918,12 +920,12 @@ class FallbackOrchestrator:
def _raise_all_failed_exception(
self,
request_id: Optional[str],
request_id: str | None,
max_attempts: int,
last_candidate: Optional[ProviderCandidate],
last_candidate: ProviderCandidate | None,
model_name: str,
api_format_enum: APIFormat,
last_error: Optional[Exception] = None,
last_error: Exception | None = None,
) -> NoReturn:
"""所有组合都失败时抛出异常"""
logger.error(f" [{request_id}] 所有 {max_attempts} 个组合均失败")
@@ -940,8 +942,8 @@ class FallbackOrchestrator:
}
# 提取上游错误响应
upstream_status: Optional[int] = None
upstream_response: Optional[str] = None
upstream_status: int | None = None
upstream_response: str | None = None
if last_error:
# 从 httpx.HTTPStatusError 提取
if isinstance(last_error, httpx.HTTPStatusError):
@@ -984,16 +986,16 @@ class FallbackOrchestrator:
async def execute_with_fallback(
self,
api_format: Union[str, APIFormat],
api_format: str | APIFormat,
model_name: str,
user_api_key: ApiKey,
request_func: Callable[[Provider, ProviderEndpoint, ProviderAPIKey], Any],
request_id: Optional[str] = None,
request_id: str | None = None,
is_stream: bool = False,
capability_requirements: Optional[Dict[str, bool]] = None,
preferred_key_ids: Optional[list[str]] = None,
request_body_ref: Optional[Dict[str, Any]] = None,
) -> Tuple[Any, str, Optional[str], Optional[str], Optional[str], Optional[str]]:
capability_requirements: dict[str, bool] | None = None,
preferred_key_ids: list[str] | None = None,
request_body_ref: dict[str, Any] | None = None,
) -> tuple[Any, str, str | None, str | None, str | None, str | None]:
"""
执行请求，并在失败时自动故障转移（缓存感知）

12
src/services/orchestration/request_dispatcher.py
View File

@@ -4,7 +4,9 @@
负责执行单个候选请求
"""
from typing import Any, Callable, Optional, Tuple
from typing import Any
from collections.abc import Callable
from sqlalchemy.orm import Session
@@ -31,7 +33,7 @@ class RequestDispatcher:
self,
db: Session,
request_executor: RequestExecutor,
cache_scheduler: Optional[CacheAwareScheduler] = None,
cache_scheduler: CacheAwareScheduler | None = None,
) -> None:
"""
初始化请求分发器
@@ -53,7 +55,7 @@ class RequestDispatcher:
candidate_record_id: str,
user_api_key: ApiKey,
request_func: Callable[..., Any],
request_id: Optional[str],
request_id: str | None,
api_format: APIFormat,
model_name: str,
affinity_key: str,
@@ -61,7 +63,7 @@ class RequestDispatcher:
attempt_counter: int,
max_attempts: int,
is_stream: bool = False,
) -> Tuple[Any, str, str, str, str, str]:
) -> tuple[Any, str, str, str, str, str]:
"""
执行请求并返回结果
@@ -98,7 +100,7 @@ class RequestDispatcher:
key_id = str(key.id)
cache_ttl_minutes = int(key.cache_ttl_minutes or 0)
provider_supports_caching = cache_ttl_minutes > 0
provider_cache_ttl_seconds: Optional[int] = (
provider_cache_ttl_seconds: int | None = (
cache_ttl_minutes * 60 if cache_ttl_minutes > 0 else None
)