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: 重构限流系统和健康监控，支持按 API 格式区分

Browse Source 
- 将 adaptive_concurrency 重命名为 adaptive_rpm，从并发控制改为 RPM 控制
- 健康监控器支持按 API 格式独立管理健康度和熔断器状态
- 新增 model_permissions 模块，支持按格式配置允许的模型
- 重构前端提供商相关表单组件，新增 Collapsible UI 组件
- 新增数据库迁移脚本支持新的数据结构
This commit is contained in:
fawney19
2026-01-10 18:43:53 +08:00
parent dd2fbf4424
commit 09e0f594ff
97 changed files with 6642 additions and 4169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -171,7 +171,8 @@ class CandidateResolver:
)
candidate_record_map[(candidate_index, 0)] = record_id
else:
max_retries_for_candidate = endpoint.max_retries if candidate.is_cached else 1
# max_retries 已从 Endpoint 迁移到 Provider（Endpoint 仍可能保留旧字段用于兼容）
max_retries_for_candidate = int(provider.max_retries or 2) if candidate.is_cached else 1
for retry_index in range(max_retries_for_candidate):
record_id = str(uuid.uuid4())
@@ -236,7 +237,7 @@ class CandidateResolver:
total = 0
for candidate in all_candidates:
if not candidate.is_skipped:
endpoint = candidate.endpoint
max_retries = int(endpoint.max_retries) if candidate.is_cached else 1
provider = candidate.provider
max_retries = int(provider.max_retries or 2) if candidate.is_cached else 1
total += max_retries
return total

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

@@ -26,7 +26,7 @@ from src.models.database import Provider, ProviderAPIKey, ProviderEndpoint
from src.services.cache.aware_scheduler import CacheAwareScheduler
from src.services.health.monitor import health_monitor
from src.services.provider.format import normalize_api_format
from src.services.rate_limit.adaptive_concurrency import get_adaptive_manager
from src.services.rate_limit.adaptive_rpm import get_adaptive_rpm_manager
from src.services.rate_limit.detector import RateLimitType, detect_rate_limit_type
@@ -112,7 +112,7 @@ class ErrorClassifier:
cache_scheduler: 缓存调度器（可选）
"""
self.db = db
self.adaptive_manager = adaptive_manager or get_adaptive_manager()
self.adaptive_manager = adaptive_manager or get_adaptive_rpm_manager()
self.cache_scheduler = cache_scheduler
# 表示客户端错误的 error type（不区分大小写）
@@ -361,7 +361,7 @@ class ErrorClassifier:
self,
key: ProviderAPIKey,
provider_name: str,
current_concurrent: Optional[int],
current_rpm: Optional[int],
exception: ProviderRateLimitException,
request_id: Optional[str] = None,
) -> str:
@@ -371,7 +371,7 @@ class ErrorClassifier:
Args:
key: API Key 对象
provider_name: 提供商名称
current_concurrent: 当前并发数
current_rpm: 当前分钟内的请求数
exception: 速率限制异常
request_id: 请求 ID（用于日志）
@@ -388,27 +388,27 @@ class ErrorClassifier:
rate_limit_info = detect_rate_limit_type(
headers=response_headers,
provider_name=provider_name,
current_concurrent=current_concurrent,
current_usage=current_rpm,
)
logger.info(f" [{request_id}] 429错误分析: "
f"类型={rate_limit_info.limit_type}, "
f"retry_after={rate_limit_info.retry_after}s, "
f"当前并发={current_concurrent}")
f"当前RPM={current_rpm}")
# 调用自适应管理器处理
new_limit = self.adaptive_manager.handle_429_error(
db=self.db,
key=key,
rate_limit_info=rate_limit_info,
current_concurrent=current_concurrent,
current_rpm=current_rpm,
)
if rate_limit_info.limit_type == RateLimitType.CONCURRENT:
logger.warning(f" [{request_id}] 自适应调整: " f"Key {key.id[:8]}... 并发限制 -> {new_limit}")
logger.warning(f" [{request_id}] 并发限制触发（不调整RPM）")
return "concurrent"
elif rate_limit_info.limit_type == RateLimitType.RPM:
logger.info(f" [{request_id}] [RPM] RPM限制，需要切换Provider")
logger.warning(f" [{request_id}] 自适应调整: Key {key.id[:8]}... RPM限制 -> {new_limit}")
return "rpm"
else:
return "unknown"
@@ -439,18 +439,18 @@ class ErrorClassifier:
# 提取可读的错误消息
extracted_message = self._extract_error_message(error_response_text)
# 构建详细错误信息
# 构建详细错误信息（仅用于日志，不暴露给客户端）
if extracted_message:
detailed_message = f"提供商 '{provider_name}' 返回错误 {status}: {extracted_message}"
detailed_message = f"上游服务返回错误 {status}: {extracted_message}"
else:
detailed_message = f"提供商 '{provider_name}' 返回错误: {status}"
detailed_message = f"上游服务返回错误: {status}"
if status == 401:
return ProviderAuthException(provider_name=provider_name)
if status == 429:
return ProviderRateLimitException(
message=error_response_text or f"提供商 '{provider_name}' 速率限制",
message="请求过于频繁，请稍后重试",
provider_name=provider_name,
response_headers=dict(error.response.headers) if error.response else None,
retry_after=(
@@ -583,6 +583,7 @@ class ErrorClassifier:
health_monitor.record_failure(
db=self.db,
key_id=str(key.id),
api_format=api_format_str,
error_type="ProviderAuthException",
)
return extra_data
@@ -592,7 +593,7 @@ class ErrorClassifier:
await self.handle_rate_limit(
key=key,
provider_name=provider_name,
current_concurrent=captured_key_concurrent,
current_rpm=captured_key_concurrent,
exception=converted_error,
request_id=request_id,
)
@@ -620,6 +621,7 @@ class ErrorClassifier:
health_monitor.record_failure(
db=self.db,
key_id=str(key.id),
api_format=api_format_str,
error_type=type(converted_error).__name__,
)
@@ -675,7 +677,7 @@ class ErrorClassifier:
await self.handle_rate_limit(
key=key,
provider_name=provider_name,
current_concurrent=captured_key_concurrent,
current_rpm=captured_key_concurrent,
exception=error,
request_id=request_id,
)
@@ -702,5 +704,6 @@ class ErrorClassifier:
health_monitor.record_failure(
db=self.db,
key_id=str(key.id),
api_format=api_format_str,
error_type=type(error).__name__,
)

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

@@ -44,7 +44,7 @@ from src.services.cache.aware_scheduler import (
get_cache_aware_scheduler,
)
from src.services.provider.format import normalize_api_format
from src.services.rate_limit.adaptive_concurrency import get_adaptive_manager
from src.services.rate_limit.adaptive_rpm import get_adaptive_rpm_manager
from src.services.rate_limit.concurrency_manager import get_concurrency_manager
from src.services.request.candidate import RequestCandidateService
from src.services.request.executor import ExecutionError, RequestExecutor
@@ -87,7 +87,7 @@ class FallbackOrchestrator:
self.redis = redis_client
self.cache_scheduler: Optional[CacheAwareScheduler] = None
self.concurrency_manager: Any = None
self.adaptive_manager = get_adaptive_manager() # 自适应并发管理器
self.adaptive_manager = get_adaptive_rpm_manager() # 自适应 RPM 管理器
self.request_executor: Optional[RequestExecutor] = None
# 拆分后的组件（延迟初始化）
@@ -558,7 +558,8 @@ class FallbackOrchestrator:
"""尝试单个候选（含重试逻辑），返回执行结果"""
provider = candidate.provider
endpoint = candidate.endpoint
max_retries_for_candidate = int(endpoint.max_retries) if candidate.is_cached else 1
# 从 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
for retry_index in range(max_retries_for_candidate):
@@ -710,7 +711,7 @@ class FallbackOrchestrator:
upstream_status = getattr(last_error, "upstream_status", None)
upstream_response = getattr(last_error, "upstream_response", None)
# 如果响应为空或无效，使用异常的字符串表示
# 如果响应为空或无效，使用异常的字符串表示作为 upstream_response
if (
not upstream_response
or not upstream_response.strip()
@@ -718,8 +719,17 @@ class FallbackOrchestrator:
):
upstream_response = str(last_error)
# 构建友好的错误消息（用于返回给客户端，不暴露内部信息）
# 如果 last_error 有 message 属性，优先使用（已经是友好提示）
# 否则使用通用提示
friendly_message = "服务暂时不可用，请稍后重试"
if last_error:
last_error_message = getattr(last_error, "message", None)
if last_error_message and isinstance(last_error_message, str):
friendly_message = last_error_message
raise ProviderNotAvailableException(
f"所有Provider均不可用，已尝试{max_attempts}个组合",
friendly_message,
request_metadata=request_metadata,
upstream_status=upstream_status,
upstream_response=upstream_response,