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

refactor: 拆分职责、引入 dataclass 封装并增强缓存健壮性

Browse Source 
- ErrorClassifier 副作用操作分离为 ErrorHandlerService（缓存失效、健康记录、RPM 调整）
- chat_handler_base 提取 ProviderRequestResult dataclass 和 _prepare_provider_request 方法
- failover 提取 AttemptErrorOutcome dataclass 和辅助方法
- formula_engine 拆分 _resolve_mapping 为子方法，增加求值异常日志
- usage recording 引入 UsageCostInfo dataclass 封装成本参数
- 前端 types.ts 拆分为 types/ 子模块
- cache backend 工厂函数加锁防止并发重复创建，LocalCache 容量检查修正
- CacheSync 监听增加断线重连机制，publish 增加重试
- guide 页面修正 useSiteInfo() 调用顺序
This commit is contained in:
fawney19
2026-02-14 16:34:52 +08:00
parent 26ede849e2
commit 6ea33c6bb8
24 changed files with 2005 additions and 1767 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -25,6 +25,7 @@ import asyncio
import json
from abc import ABC, abstractmethod
from collections.abc import AsyncGenerator, Awaitable, Callable
from dataclasses import dataclass, field
from typing import Any
import httpx
@@ -225,6 +226,22 @@ def _build_error_json_payload(
return _build_client_error_response_best_effort(message, client_format)
@dataclass
class ProviderRequestResult:
"""_prepare_provider_request() 的返回结果，封装请求构建阶段的所有产出。"""
request_body: dict[str, Any]
url_model: str
mapped_model: str | None
envelope: Any # ProviderEnvelope | None
extra_headers: dict[str, str] = field(default_factory=dict)
upstream_is_stream: bool = True
needs_conversion: bool = False
provider_api_format: str = ""
client_api_format: str = ""
auth_info: Any = None
class ChatHandlerBase(BaseMessageHandler, ABC):
"""
Chat Handler 基类
@@ -755,62 +772,43 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
await self._record_stream_failure(ctx, e, original_headers, original_request_body)
raise
async def _execute_stream_request(
async def _prepare_provider_request(
self,
ctx: StreamContext,
stream_processor: StreamProcessor,
*,
model: str,
provider: Provider,
endpoint: ProviderEndpoint,
key: ProviderAPIKey,
original_request_body: dict[str, Any],
original_headers: dict[str, str],
query_params: dict[str, str] | None = None,
candidate: ProviderCandidate | None = None,
is_disconnected: Callable[[], Awaitable[bool]] | None = None,
) -> AsyncGenerator[bytes]:
"""执行流式请求并返回流生成器"""
# 重置上下文状态（重试时清除之前的数据）
ctx.reset_for_retry()
# 更新 Provider 信息
ctx.update_provider_info(
provider_name=str(provider.name),
provider_id=str(provider.id),
endpoint_id=str(endpoint.id),
key_id=str(key.id),
provider_api_format=str(endpoint.api_format) if endpoint.api_format else None,
)
ctx.provider_type = str(getattr(provider, "provider_type", "") or "")
# 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
client_api_format: str,
provider_api_format: str,
candidate: ProviderCandidate | None,
client_is_stream: bool,
) -> ProviderRequestResult:
"""
构建 Provider 请求：模型映射、格式转换、envelope 包装。
流式和非流式请求共享此逻辑，唯一差异是 client_is_stream 参数。
"""
# 提前获取认证信息（Vertex AI 格式判断需要使用 auth_config）
auth_info = await get_provider_auth(endpoint, key)
# 解析 Vertex AI 动态格式并计算 needs_conversion
provider_api_format, needs_conversion = _resolve_vertex_ai_format(
key, auth_info, ctx.model, provider_api_format, client_api_format, candidate
key, auth_info, model, provider_api_format, client_api_format, candidate
)
ctx.provider_api_format = provider_api_format
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,
source_model=model,
provider_id=str(provider.id),
api_format=provider_api_format,
)
# 应用模型映射到请求体
if mapped_model:
ctx.mapped_model = mapped_model
request_body = self.apply_mapped_model(original_request_body, mapped_model)
else:
request_body = dict(original_request_body)
@@ -824,14 +822,14 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
same_format_variant = behavior.same_format_variant
cross_format_variant = behavior.cross_format_variant
# Upstream streaming policy (per-endpoint): may force upstream to sync/stream mode.
# Upstream streaming policy (per-endpoint).
upstream_policy = get_upstream_stream_policy(
endpoint,
provider_type=provider_type,
endpoint_sig=str(provider_api_format),
)
upstream_is_stream = resolve_upstream_is_stream(
client_is_stream=True,
client_is_stream=client_is_stream,
policy=upstream_policy,
)
@@ -849,7 +847,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
request_body,
str(provider_api_format),
mapped_model,
ctx.model,
model,
)
# 格式转换后，为需要 stream 字段的格式设置流式标志
self._set_stream_after_conversion(
@@ -886,13 +884,13 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
)
# 获取 URL 模型名
url_model = self.get_model_for_url(request_body, mapped_model) or ctx.model
url_model = self.get_model_for_url(request_body, mapped_model) or model
# Provider envelope: wrap request after auth is available and before RequestBuilder.build().
# Provider envelope: wrap request.
if envelope:
request_body, url_model = envelope.wrap_request(
request_body,
model=url_model or ctx.model or "",
model=url_model or model or "",
url_model=url_model,
decrypted_auth_config=auth_info.decrypted_auth_config if auth_info else None,
)
@@ -902,6 +900,78 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
if envelope:
extra_headers.update(envelope.extra_headers() or {})
return ProviderRequestResult(
request_body=request_body,
url_model=url_model,
mapped_model=mapped_model,
envelope=envelope,
extra_headers=extra_headers,
upstream_is_stream=upstream_is_stream,
needs_conversion=needs_conversion,
provider_api_format=provider_api_format,
client_api_format=client_api_format,
auth_info=auth_info,
)
async def _execute_stream_request(
self,
ctx: StreamContext,
stream_processor: StreamProcessor,
provider: Provider,
endpoint: ProviderEndpoint,
key: ProviderAPIKey,
original_request_body: dict[str, Any],
original_headers: dict[str, str],
query_params: dict[str, str] | None = None,
candidate: ProviderCandidate | None = None,
is_disconnected: Callable[[], Awaitable[bool]] | None = None,
) -> AsyncGenerator[bytes]:
"""执行流式请求并返回流生成器"""
# 重置上下文状态（重试时清除之前的数据）
ctx.reset_for_retry()
# 更新 Provider 信息
ctx.update_provider_info(
provider_name=str(provider.name),
provider_id=str(provider.id),
endpoint_id=str(endpoint.id),
key_id=str(key.id),
provider_api_format=str(endpoint.api_format) if endpoint.api_format else None,
)
ctx.provider_type = str(getattr(provider, "provider_type", "") or "")
# 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
# 构建 Provider 请求（模型映射、格式转换、envelope 包装）
prep = await self._prepare_provider_request(
model=ctx.model,
provider=provider,
endpoint=endpoint,
key=key,
original_request_body=original_request_body,
client_api_format=client_api_format,
provider_api_format=provider_api_format,
candidate=candidate,
client_is_stream=True,
)
provider_api_format = prep.provider_api_format
needs_conversion = prep.needs_conversion
ctx.provider_api_format = provider_api_format
ctx.needs_conversion = needs_conversion
mapped_model = prep.mapped_model
if mapped_model:
ctx.mapped_model = mapped_model
request_body = prep.request_body
url_model = prep.url_model
envelope = prep.envelope
upstream_is_stream = prep.upstream_is_stream
auth_info = prep.auth_info
# 构建请求（上游始终使用 header 认证，不跟随客户端的 query 方式）
provider_payload, provider_headers = self._request_builder.build(
request_body,
@@ -909,7 +979,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
endpoint,
key,
is_stream=upstream_is_stream,
extra_headers=extra_headers if extra_headers else None,
extra_headers=prep.extra_headers if prep.extra_headers else None,
pre_computed_auth=auth_info.as_tuple() if auth_info else None,
)
if upstream_is_stream:
@@ -1070,6 +1140,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
)
# Convert sync JSON -> InternalResponse, then InternalResponse -> client stream events.
registry = get_format_converter_registry()
src_norm = (
registry.get_normalizer(str(provider_api_format)) if provider_api_format else None
)
@@ -1425,125 +1496,35 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
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)
)
# 提前获取认证信息（Vertex AI 格式判断需要使用 auth_config）
auth_info = await get_provider_auth(endpoint, key)
# 解析 Vertex AI 动态格式并计算 needs_conversion
provider_api_format, needs_conversion = _resolve_vertex_ai_format(
key, auth_info, model, provider_api_format, client_api_format, candidate
# 构建 Provider 请求（模型映射、格式转换、envelope 包装）
prep = await self._prepare_provider_request(
model=model,
provider=provider,
endpoint=endpoint,
key=key,
original_request_body=request_body_ref["body"],
client_api_format=client_api_format,
provider_api_format=provider_api_format,
candidate=candidate,
client_is_stream=False,
)
provider_api_format = prep.provider_api_format
needs_conversion = prep.needs_conversion
provider_api_format_for_error = provider_api_format
client_api_format_for_error = client_api_format
needs_conversion_for_error = 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=model,
provider_id=str(provider.id),
api_format=provider_api_format,
)
# 应用模型映射
mapped_model = prep.mapped_model
if mapped_model:
mapped_model_result = mapped_model # 保存映射后的模型名，用于 Usage 记录
request_body = self.apply_mapped_model(request_body_ref["body"], mapped_model)
else:
request_body = dict(request_body_ref["body"])
provider_type = str(getattr(provider, "provider_type", "") or "").lower()
behavior = get_provider_behavior(
provider_type=provider_type,
endpoint_sig=provider_api_format,
)
envelope = behavior.envelope
same_format_variant = behavior.same_format_variant
cross_format_variant = behavior.cross_format_variant
# Upstream streaming policy (per-endpoint).
upstream_policy = get_upstream_stream_policy(
endpoint,
provider_type=provider_type,
endpoint_sig=str(provider_api_format),
)
upstream_is_stream = resolve_upstream_is_stream(
client_is_stream=False,
policy=upstream_policy,
)
# 跨格式：先做请求体转换（失败触发 failover）
registry = get_format_converter_registry()
if needs_conversion:
request_body = registry.convert_request(
request_body,
client_api_format,
provider_api_format,
target_variant=cross_format_variant,
)
# 格式转换后，为需要 model 字段的格式设置模型名
self._set_model_after_conversion(
request_body,
provider_api_format,
mapped_model,
model,
)
# 格式转换后，为需要 stream 字段的格式设置流式标志
self._set_stream_after_conversion(
request_body,
client_api_format,
provider_api_format,
is_stream=upstream_is_stream,
)
else:
# 同格式：按原逻辑做轻量清理（子类可覆盖以移除不需要的字段）
request_body = self.prepare_provider_request_body(request_body)
# 同格式时也需要应用 target_variant 转换（如 Codex）
if same_format_variant:
request_body = registry.convert_request(
request_body,
provider_api_format,
provider_api_format,
target_variant=same_format_variant,
)
# 模型感知的请求后处理（如图像生成模型移除不兼容字段）
request_body = self.finalize_provider_request(
request_body,
mapped_model=mapped_model,
provider_api_format=str(provider_api_format) if provider_api_format else None,
)
# Force upstream stream/sync mode in request body (best-effort).
if provider_api_format:
enforce_stream_mode_for_upstream(
request_body,
provider_api_format=str(provider_api_format),
upstream_is_stream=upstream_is_stream,
)
# 获取 URL 模型名（兜底使用外层的 model，确保 Gemini 等格式能正确构建 URL）
url_model = self.get_model_for_url(request_body, mapped_model) or model
# Provider envelope: wrap request after auth is available and before RequestBuilder.build().
if envelope:
request_body, url_model = envelope.wrap_request(
request_body,
model=url_model or model or "",
url_model=url_model,
decrypted_auth_config=auth_info.decrypted_auth_config if auth_info else None,
)
# Provider envelope: extra upstream headers (e.g. dedicated User-Agent).
extra_headers: dict[str, str] = {}
if envelope:
extra_headers.update(envelope.extra_headers() or {})
mapped_model_result = mapped_model
request_body = prep.request_body
url_model = prep.url_model
envelope = prep.envelope
upstream_is_stream = prep.upstream_is_stream
auth_info = prep.auth_info
# 构建请求（上游始终使用 header 认证，不跟随客户端的 query 方式）
provider_payload, provider_hdrs = self._request_builder.build(
@@ -1552,7 +1533,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
endpoint,
key,
is_stream=upstream_is_stream,
extra_headers=extra_headers if extra_headers else None,
extra_headers=prep.extra_headers if prep.extra_headers else None,
pre_computed_auth=auth_info.as_tuple() if auth_info else None,
)
if upstream_is_stream:
@@ -1584,6 +1565,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
_effective_proxy = resolve_effective_proxy(provider.proxy, getattr(key, "proxy", None))
sync_proxy_info = resolve_proxy_info(_effective_proxy)
_proxy_label = get_proxy_label(sync_proxy_info)
provider_type = str(getattr(provider, "provider_type", "") or "").lower()
logger.info(
f" [{self.request_id}] 发送{'上游流式(聚合)' if upstream_is_stream else '非流式'}请求: "
@@ -1678,6 +1660,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
provider_parser=provider_parser,
)
registry = get_format_converter_registry()
tgt_norm = (
registry.get_normalizer(client_api_format)
if client_api_format

15
src/plugins/cache/memory.py vendored
View File

@@ -6,7 +6,6 @@
from __future__ import annotations
import asyncio
import threading
import time
from collections import OrderedDict
from typing import Any
@@ -24,7 +23,7 @@ class MemoryCachePlugin(CachePlugin):
super().__init__(name, config)
self._cache: OrderedDict = OrderedDict()
self._expiry: dict[str, float] = {}
self._lock = threading.RLock()
self._lock = asyncio.Lock()
self._hits = 0
self._misses = 0
self._evictions = 0
@@ -60,7 +59,7 @@ class MemoryCachePlugin(CachePlugin):
now = time.time()
expired_keys = []
with self._lock:
async with self._lock:
for key, expiry in self._expiry.items():
if expiry < now:
expired_keys.append(key)
@@ -81,7 +80,7 @@ class MemoryCachePlugin(CachePlugin):
async def get(self, key: str) -> Any | None:
"""获取缓存值"""
with self._lock:
async with self._lock:
# 检查是否过期
if key in self._expiry:
if self._expiry[key] < time.time():
@@ -103,7 +102,7 @@ class MemoryCachePlugin(CachePlugin):
async def set(self, key: str, value: Any, ttl: int | None = None) -> bool:
"""设置缓存值"""
with self._lock:
async with self._lock:
# 检查大小限制
if key not in self._cache:
self._check_size()
@@ -126,7 +125,7 @@ class MemoryCachePlugin(CachePlugin):
async def delete(self, key: str) -> bool:
"""删除缓存项"""
with self._lock:
async with self._lock:
if key in self._cache:
self._cache.pop(key)
self._expiry.pop(key, None)
@@ -135,7 +134,7 @@ class MemoryCachePlugin(CachePlugin):
async def exists(self, key: str) -> bool:
"""检查缓存项是否存在"""
with self._lock:
async with self._lock:
# 检查是否过期
if key in self._expiry:
if self._expiry[key] < time.time():
@@ -147,7 +146,7 @@ class MemoryCachePlugin(CachePlugin):
async def clear(self) -> bool:
"""清空所有缓存"""
with self._lock:
async with self._lock:
self._cache.clear()
self._expiry.clear()
return True

289
src/services/billing/formula_engine.py
View File

@@ -17,6 +17,7 @@ from decimal import Decimal
from functools import lru_cache
from typing import Any, Iterable, Literal
from src.core.logger import logger
from src.services.billing.precision import DECIMAL_CONTEXT_PRECISION, to_decimal
@@ -378,6 +379,11 @@ class FormulaEngine:
if status == "missing_required":
missing_required.append(var_name)
continue
if status == "error":
# 求值异常但非 required，使用 default 值继续
resolved[var_name] = value
progressed = True
continue
resolved[var_name] = value
progressed = True
if not progressed:
@@ -387,6 +393,11 @@ class FormulaEngine:
for var_name, mapping in unresolved.items():
required = bool(mapping.get("required", False))
default = mapping.get("default", 0)
logger.warning(
"[FormulaEngine] computed 维度 '{}' 在迭代后仍未解析, required={}",
var_name,
required,
)
if required:
missing_required.append(var_name)
else:
@@ -505,9 +516,14 @@ class FormulaEngine:
except NameError:
# dependency not ready yet
return (None, "pending") if required else (default, "pending")
except Exception:
# treat as config error: fallback to default unless required
return (None, "missing_required") if required else (default, "ok")
except Exception as exc:
logger.warning(
"[FormulaEngine] computed 维度 '{}' 求值异常: {}, expression={!r}",
var_name,
exc,
expr,
)
return (None, "missing_required") if required else (default, "error")
def _resolve_mapping(
self,
@@ -517,16 +533,41 @@ class FormulaEngine:
) -> tuple[Any, bool, dict[str, Any] | None]:
"""
Returns:
(value, is_missing_required)
(value, is_missing_required, tier_meta)
说明：
- is_missing_required 仅在 required=true 且缺失时为 True
- required=false 的缺失会使用 default 或 0 兜底，并返回 is_missing_required=False
"""
source = (mapping.get("source") or "constant").lower()
if source == "constant":
return self._resolve_constant(mapping)
if source == "dimension":
return self._resolve_dimension(var_name, mapping, dims)
if source == "matrix":
return self._resolve_matrix(var_name, mapping, dims)
if source == "tiered":
return self._resolve_tiered(var_name, mapping, dims)
# 未知 source：视为配置错误，但不直接中断计费（返回 default）
return mapping.get("default", 0), False, None
@staticmethod
def _resolve_constant(
mapping: dict[str, Any],
) -> tuple[Any, bool, dict[str, Any] | None]:
"""constant 默认行为：由 variables 提供；dimension_mappings 显式 constant 时仅做兜底"""
return mapping.get("default", 0), False, None
@staticmethod
def _resolve_dimension(
var_name: str,
mapping: dict[str, Any],
dims: dict[str, Any],
) -> tuple[Any, bool, dict[str, Any] | None]:
"""解析 dimension source：从 dims 中取值并尝试转换为 Decimal"""
required = bool(mapping.get("required", False))
allow_zero = bool(mapping.get("allow_zero", False))
default = mapping.get("default", 0)
def _missing() -> tuple[Any, bool]:
@@ -534,36 +575,15 @@ class FormulaEngine:
return None, True
return default, False
if source == "constant":
# constant 默认行为：由 variables 提供；dimension_mappings 显式 constant 时仅做兜底
return default, False, None
if source == "dimension":
key = mapping.get("key") or var_name
raw = dims.get(key)
if raw is None:
key = mapping.get("key") or var_name
raw = dims.get(key)
if raw is None:
v, m = _missing()
return v, m, None
if isinstance(raw, str):
if raw == "":
v, m = _missing()
return v, m, None
if isinstance(raw, str):
if raw == "":
v, m = _missing()
return v, m, None
# 尝试将字符串解析为数字，否则按字符串返回（供上层自行决定）
try:
num = to_decimal(raw)
if num == 0 and not allow_zero:
v, m = _missing()
return v, m, None
return num, False, None
except Exception:
return raw, False, None
if isinstance(raw, (int, float, Decimal)):
num = to_decimal(raw)
if num == 0 and not allow_zero:
v, m = _missing()
return v, m, None
return num, False, None
# 其他类型：尽量转为 float，否则视为缺失
try:
num = to_decimal(raw)
if num == 0 and not allow_zero:
@@ -571,61 +591,118 @@ class FormulaEngine:
return v, m, None
return num, False, None
except Exception:
return raw, False, None
if isinstance(raw, (int, float, Decimal)):
num = to_decimal(raw)
if num == 0 and not allow_zero:
v, m = _missing()
return v, m, None
return num, False, None
try:
num = to_decimal(raw)
if num == 0 and not allow_zero:
v, m = _missing()
return v, m, None
return num, False, None
except Exception:
v, m = _missing()
return v, m, None
if source == "matrix":
key = mapping.get("key") or var_name
raw = dims.get(key)
if raw is None or raw == "":
v, m = _missing()
return v, m, None
raw_key = str(raw)
matrix = mapping.get("map") or {}
if raw_key in matrix:
try:
return to_decimal(matrix[raw_key]), False, None
except Exception:
return matrix[raw_key], False, None
# matrix 未命中：若 required=true 则仍视为缺失；否则使用 default
@staticmethod
def _resolve_matrix(
var_name: str,
mapping: dict[str, Any],
dims: dict[str, Any],
) -> tuple[Any, bool, dict[str, Any] | None]:
"""解析 matrix source：从 map 中按 key 查找值"""
required = bool(mapping.get("required", False))
default = mapping.get("default", 0)
def _missing() -> tuple[Any, bool]:
if required:
return None, True, None
return default, False, None
return None, True
return default, False
if source == "tiered":
tier_key = mapping.get("tier_key")
if not tier_key:
v, m = _missing()
return v, m, None
raw_tier_value = dims.get(tier_key)
if raw_tier_value is None:
v, m = _missing()
return v, m, None
key = mapping.get("key") or var_name
raw = dims.get(key)
if raw is None or raw == "":
v, m = _missing()
return v, m, None
raw_key = str(raw)
matrix = mapping.get("map") or {}
if raw_key in matrix:
try:
tier_value = to_decimal(raw_tier_value)
return to_decimal(matrix[raw_key]), False, None
except Exception:
v, m = _missing()
return v, m, None
return matrix[raw_key], False, None
if required:
return None, True, None
return default, False, None
if tier_value == 0 and not allow_zero:
v, m = _missing()
return v, m, None
def _resolve_tiered(
self,
var_name: str,
mapping: dict[str, Any],
dims: dict[str, Any],
) -> tuple[Any, bool, dict[str, Any] | None]:
"""解析 tiered source：按阶梯匹配值"""
required = bool(mapping.get("required", False))
allow_zero = bool(mapping.get("allow_zero", False))
default = mapping.get("default", 0)
# Optional TTL override (legacy: Claude cache pricing)
ttl_key = mapping.get("ttl_key")
ttl_value_key = mapping.get("ttl_value_key")
ttl_minutes: Decimal | None = None
if ttl_key and ttl_value_key and dims.get(ttl_key) is not None:
try:
ttl_minutes = to_decimal(dims.get(ttl_key))
except Exception:
ttl_minutes = None
def _missing() -> tuple[Any, bool]:
if required:
return None, True
return default, False
tiers = mapping.get("tiers") or []
# tiers: [{up_to: 128000, value: 2.5}, {up_to: null, value: 1.25}]
for idx, tier in enumerate(tiers):
up_to = tier.get("up_to")
if up_to is None:
tier_key = mapping.get("tier_key")
if not tier_key:
v, m = _missing()
return v, m, None
raw_tier_value = dims.get(tier_key)
if raw_tier_value is None:
v, m = _missing()
return v, m, None
try:
tier_value = to_decimal(raw_tier_value)
except Exception:
v, m = _missing()
return v, m, None
if tier_value == 0 and not allow_zero:
v, m = _missing()
return v, m, None
# Optional TTL override (legacy: Claude cache pricing)
ttl_key = mapping.get("ttl_key")
ttl_value_key = mapping.get("ttl_value_key")
ttl_minutes: Decimal | None = None
if ttl_key and ttl_value_key and dims.get(ttl_key) is not None:
try:
ttl_minutes = to_decimal(dims.get(ttl_key))
except Exception:
ttl_minutes = None
tiers = mapping.get("tiers") or []
# tiers: [{up_to: 128000, value: 2.5}, {up_to: null, value: 1.25}]
for idx, tier in enumerate(tiers):
up_to = tier.get("up_to")
if up_to is None:
value = to_decimal(tier.get("value", default))
if (
ttl_minutes is not None
and ttl_value_key
and isinstance(tier.get("cache_ttl_pricing"), list)
):
value = self._resolve_ttl_pricing(
tier.get("cache_ttl_pricing") or [],
ttl_minutes,
str(ttl_value_key),
fallback=value,
)
return value, False, {"tier_index": idx, "tier_info": dict(tier)}
try:
if tier_value <= to_decimal(up_to):
value = to_decimal(tier.get("value", default))
if (
ttl_minutes is not None
@@ -639,43 +716,25 @@ class FormulaEngine:
fallback=value,
)
return value, False, {"tier_index": idx, "tier_info": dict(tier)}
try:
if tier_value <= to_decimal(up_to):
value = to_decimal(tier.get("value", default))
if (
ttl_minutes is not None
and ttl_value_key
and isinstance(tier.get("cache_ttl_pricing"), list)
):
value = self._resolve_ttl_pricing(
tier.get("cache_ttl_pricing") or [],
ttl_minutes,
str(ttl_value_key),
fallback=value,
)
return value, False, {"tier_index": idx, "tier_info": dict(tier)}
except Exception:
# up_to 配置异常：忽略并继续
continue
# 无匹配：使用最后一个或 default
if tiers:
last = tiers[-1]
value = to_decimal(last.get("value", default))
if (
ttl_minutes is not None
and ttl_value_key
and isinstance(last.get("cache_ttl_pricing"), list)
):
value = self._resolve_ttl_pricing(
last.get("cache_ttl_pricing") or [],
ttl_minutes,
str(ttl_value_key),
fallback=value,
)
return value, False, {"tier_index": len(tiers) - 1, "tier_info": dict(last)}
return default, False, None
# 未知 source：视为配置错误，但不直接中断计费（返回 default）
except Exception:
# up_to 配置异常：忽略并继续
continue
# 无匹配：使用最后一个或 default
if tiers:
last = tiers[-1]
value = to_decimal(last.get("value", default))
if (
ttl_minutes is not None
and ttl_value_key
and isinstance(last.get("cache_ttl_pricing"), list)
):
value = self._resolve_ttl_pricing(
last.get("cache_ttl_pricing") or [],
ttl_minutes,
str(ttl_value_key),
fallback=value,
)
return value, False, {"tier_index": len(tiers) - 1, "tier_info": dict(last)}
return default, False, None
def _resolve_ttl_pricing(

4
src/services/cache/affinity_manager.py vendored
View File

@@ -93,6 +93,10 @@ class CacheAffinityManager:
self._memory_lock: asyncio.Lock | None = None
# L1 缓存（即使使用 Redis 也启用，减少网络往返）
# 注意：L1 是本地进程内缓存，多实例部署时存在短暂不一致窗口（TTL 秒级）。
# 当前 TTL 默认 3 秒，对于亲和性路由来说可接受：最坏情况是短暂路由到
# 旧 provider，下次请求即可自动修正。如果需要严格一致性，将 TTL 设为 0
# 以禁用 L1 缓存，或通过 CacheSyncService 接收 pub/sub 主动失效。
self._l1_cache_ttl = int(os.getenv("CACHE_AFFINITY_L1_TTL", str(CacheTTL.L1_LOCAL)))
self._l1_cache: dict[str, tuple[float, dict[str, Any]]] = {}
self._l1_lock = asyncio.Lock()

44
src/services/cache/backend.py vendored
View File

@@ -97,17 +97,16 @@ class LocalCache(BaseCacheBackend):
# 如果键已存在，更新访问顺序
if key in self._cache:
self._cache.move_to_end(key)
self._cache[key] = value
self._expiry[key] = time.time() + ttl
# 检查容量限制，淘汰最旧项
if len(self._cache) > self._max_size:
elif len(self._cache) >= self._max_size:
# 插入新键前淘汰最旧项，确保容量不超过 max_size
oldest_key = next(iter(self._cache))
del self._cache[oldest_key]
if oldest_key in self._expiry:
del self._expiry[oldest_key]
self._cache[key] = value
self._expiry[key] = time.time() + ttl
async def delete(self, key: str) -> None:
"""删除缓存值（线程安全）"""
async with self._lock:
@@ -276,6 +275,7 @@ class RedisCache(BaseCacheBackend):
# 缓存后端工厂
_cache_backends: dict[str, BaseCacheBackend] = {}
_cache_backend_lock = asyncio.Lock()
async def get_cache_backend(
@@ -295,36 +295,44 @@ async def get_cache_backend(
"""
cache_key = f"{name}:{backend_type}"
# 无锁快路径
if cache_key in _cache_backends:
return _cache_backends[cache_key]
# 根据类型创建缓存后端
async with _cache_backend_lock:
# Double-check: 锁内再检查一次，避免重复创建
if cache_key in _cache_backends:
return _cache_backends[cache_key]
backend = _create_cache_backend(name, backend_type, max_size, ttl)
_cache_backends[cache_key] = backend
return backend
def _create_cache_backend(
name: str, backend_type: str, max_size: int, ttl: int
) -> BaseCacheBackend:
"""根据类型创建缓存后端实例"""
if backend_type == "redis":
# 尝试使用 Redis
redis_client = get_redis_client_sync()
if redis_client is None:
logger.warning(f"[CacheBackend] Redis 未初始化，{name} 降级为本地缓存")
backend = LocalCache(max_size=max_size, default_ttl=ttl)
return LocalCache(max_size=max_size, default_ttl=ttl)
else:
backend = RedisCache(redis_client=redis_client, key_prefix=name, default_ttl=ttl)
logger.info(f"[CacheBackend] {name} 使用 Redis 缓存")
return RedisCache(redis_client=redis_client, key_prefix=name, default_ttl=ttl)
elif backend_type == "local":
# 强制使用本地缓存
backend = LocalCache(max_size=max_size, default_ttl=ttl)
logger.info(f"[CacheBackend] {name} 使用本地缓存")
return LocalCache(max_size=max_size, default_ttl=ttl)
else: # auto
# 自动选择：优先 Redis，降级到 Local
redis_client = get_redis_client_sync()
if redis_client is not None:
backend = RedisCache(redis_client=redis_client, key_prefix=name, default_ttl=ttl)
logger.debug(f"[CacheBackend] {name} 自动选择 Redis 缓存")
return RedisCache(redis_client=redis_client, key_prefix=name, default_ttl=ttl)
else:
backend = LocalCache(max_size=max_size, default_ttl=ttl)
logger.debug(f"[CacheBackend] {name} 自动选择本地缓存（Redis 不可用）")
_cache_backends[cache_key] = backend
return backend
return LocalCache(max_size=max_size, default_ttl=ttl)

79
src/services/cache/sync.py vendored
View File

@@ -110,34 +110,45 @@ class CacheSyncService:
logger.debug(f"[CacheSync] 注册处理器: {channel}")
async def _listen(self) -> None:
"""监听 Redis pub/sub 消息"""
"""监听 Redis pub/sub 消息（含断线重连）"""
logger.info("[CacheSync] 开始监听缓存失效消息")
consecutive_failures = 0
max_consecutive_failures = 10
reconnect_interval = 5.0
try:
async for message in self._pubsub.listen():
if message["type"] == "message":
channel = message["channel"]
data = message["data"]
while self._running:
try:
async for message in self._pubsub.listen():
consecutive_failures = 0 # 收到消息即重置
if message["type"] == "message":
channel = message["channel"]
data = message["data"]
# 解析消息
try:
payload = json.loads(data)
logger.debug(f"[CacheSync] 收到消息: {channel} -> {payload}")
try:
payload = json.loads(data)
logger.debug(f"[CacheSync] 收到消息: {channel} -> {payload}")
# 调用注册的处理器
if channel in self._handlers:
handler = self._handlers[channel]
await handler(payload)
else:
logger.warning(f"[CacheSync] 未找到处理器: {channel}")
except json.JSONDecodeError as e:
logger.error(f"[CacheSync] 消息解析失败: {data}, 错误: {e}")
except Exception as e:
logger.error(f"[CacheSync] 处理消息失败: {channel}, 错误: {e}")
except asyncio.CancelledError:
logger.info("[CacheSync] 监听任务已取消")
except Exception as e:
logger.error(f"[CacheSync] 监听失败: {e}")
if channel in self._handlers:
handler = self._handlers[channel]
await handler(payload)
else:
logger.warning(f"[CacheSync] 未找到处理器: {channel}")
except json.JSONDecodeError as e:
logger.error(f"[CacheSync] 消息解析失败: {data}, 错误: {e}")
except Exception as e:
logger.error(f"[CacheSync] 处理消息失败: {channel}, 错误: {e}")
except asyncio.CancelledError:
logger.info("[CacheSync] 监听任务已取消")
return
except Exception as e:
consecutive_failures += 1
logger.error(
f"[CacheSync] 监听失败 ({consecutive_failures}/{max_consecutive_failures}): {e}"
)
if consecutive_failures >= max_consecutive_failures:
logger.error("[CacheSync] 连续失败次数过多，停止重连")
return
await asyncio.sleep(reconnect_interval)
async def publish_global_model_changed(self, model_name: str) -> Any:
"""发布 GlobalModel 变更通知"""
@@ -154,13 +165,19 @@ class CacheSyncService:
await self._publish(self.CHANNEL_CLEAR_ALL, {})
async def _publish(self, channel: str, data: dict) -> None:
"""发布消息到 Redis 频道"""
try:
message = json.dumps(data)
await self._redis.publish(channel, message)
logger.debug(f"[CacheSync] 发布消息: {channel} -> {data}")
except Exception as e:
logger.error(f"[CacheSync] 发布消息失败: {channel}, 错误: {e}")
"""发布消息到 Redis 频道（含简单重试）"""
message = json.dumps(data)
last_error: Exception | None = None
for attempt in range(2):
try:
await self._redis.publish(channel, message)
logger.debug(f"[CacheSync] 发布消息: {channel} -> {data}")
return
except Exception as e:
last_error = e
if attempt == 0:
await asyncio.sleep(0.5)
logger.error(f"[CacheSync] 发布消息失败（已重试）: {channel}, 错误: {last_error}")
# 全局单例

266
src/services/candidate/failover.py
View File

@@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio
import re
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, AsyncIterator
@@ -29,6 +30,16 @@ _SENSITIVE_PATTERN = re.compile(
)
@dataclass
class AttemptErrorOutcome:
"""_handle_attempt_error 的返回结果"""
action: FailoverAction
last_status_code: int | None
max_retries: int
stop_result: ExecutionResult | None = None
class FailoverEngine:
"""
FailoverEngine executes candidate attempts under policies.
@@ -186,23 +197,7 @@ class FailoverEngine:
record_id=record_id,
)
# Mark success-like status
if record_id:
if attempt_result.kind == AttemptKind.STREAM:
# For streaming, mark "streaming" (final status is recorded elsewhere).
self._update_record(
record_id,
status="streaming",
status_code=attempt_result.http_status,
)
else:
self._update_record(
record_id,
status="success",
status_code=attempt_result.http_status,
finished_at=datetime.now(timezone.utc),
)
self.db.commit()
self._record_attempt_success(record_id, attempt_result)
# PRE_EXPAND: mark unused slots after request ends (success)
if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
@@ -234,93 +229,32 @@ class FailoverEngine:
)
except StreamProbeError as exc:
# Probe failed (before first chunk) => eligible for failover
last_status_code = exc.http_status
if record_id:
self._update_record(
record_id,
status="failed",
status_code=exc.http_status,
error_type=type(exc).__name__,
error_message=self._sanitize(str(exc)),
finished_at=datetime.now(timezone.utc),
)
self.db.commit()
self._record_attempt_failure(record_id, exc, exc.http_status)
action = FailoverAction.CONTINUE
except Exception as exc:
has_retry_left = retry_index + 1 < max_retries
# If caller provides an execution_error_handler, prefer it for RequestExecutor's ExecutionError.
handler_used = False
if execution_error_handler is not None:
try:
from src.services.request.executor import (
ExecutionError as _ExecutionError,
)
if isinstance(exc, _ExecutionError):
handler_used = True
action, new_max_retries = await execution_error_handler(
exec_err=exc,
candidate=candidate,
candidate_index=candidate_index,
retry_index=retry_index,
max_retries_for_candidate=max_retries,
record_id=record_id,
attempt_count=attempt_count,
max_attempts=max_attempts,
)
if new_max_retries is not None:
max_retries = max(max_retries, int(new_max_retries))
except Exception:
# Fall back to internal handler below.
handler_used = False
if not handler_used:
action = await self._handle_error(
exc,
candidate=candidate,
has_retry_left=has_retry_left,
)
last_status_code = int(getattr(exc, "status_code", 0) or 0) or int(
getattr(exc, "http_status", 0) or 0
)
if record_id:
self._update_record(
record_id,
status="failed",
status_code=last_status_code or None,
error_type=type(exc).__name__,
error_message=self._sanitize(str(exc)),
finished_at=datetime.now(timezone.utc),
)
self.db.commit()
if action == FailoverAction.STOP:
# PRE_EXPAND: STOP ends the request => mark remaining slots unused.
if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
self._mark_remaining_slots_unused(
candidate_record_map=candidate_record_map,
candidates=candidates,
success_candidate_idx=candidate_index,
success_retry_idx=retry_index,
retry_policy=retry_policy,
)
return ExecutionResult(
success=False,
error_type=type(exc).__name__,
error_message=self._sanitize(str(exc)),
last_status_code=last_status_code or None,
candidate_keys=self._get_candidate_keys(
request_id=request_id,
fallback=candidate_keys_fallback,
candidates=candidates,
),
attempt_count=attempt_count,
)
outcome = await self._handle_attempt_error(
exc,
candidate=candidate,
candidate_index=candidate_index,
retry_index=retry_index,
max_retries=max_retries,
record_id=record_id,
attempt_count=attempt_count,
max_attempts=max_attempts,
execution_error_handler=execution_error_handler,
retry_policy=retry_policy,
candidate_record_map=candidate_record_map,
candidates=candidates,
request_id=request_id,
candidate_keys_fallback=candidate_keys_fallback,
)
action = outcome.action
last_status_code = outcome.last_status_code
max_retries = outcome.max_retries
if outcome.stop_result is not None:
return outcome.stop_result
# action switch: continue/ retry
if action == FailoverAction.CONTINUE:
@@ -357,6 +291,138 @@ class FailoverEngine:
attempt_count=attempt_count,
)
def _record_attempt_success(self, record_id: str | None, attempt_result: AttemptResult) -> None:
"""Mark attempt record as success/streaming."""
if not record_id:
return
if attempt_result.kind == AttemptKind.STREAM:
self._update_record(
record_id,
status="streaming",
status_code=attempt_result.http_status,
)
else:
self._update_record(
record_id,
status="success",
status_code=attempt_result.http_status,
finished_at=datetime.now(timezone.utc),
)
self.db.commit()
def _record_attempt_failure(
self, record_id: str | None, exc: Exception, status_code: int | None = None
) -> None:
"""Mark attempt record as failed."""
if not record_id:
return
self._update_record(
record_id,
status="failed",
status_code=status_code,
error_type=type(exc).__name__,
error_message=self._sanitize(str(exc)),
finished_at=datetime.now(timezone.utc),
)
self.db.commit()
async def _handle_attempt_error(
self,
exc: Exception,
*,
candidate: ProviderCandidate,
candidate_index: int,
retry_index: int,
max_retries: int,
record_id: str | None,
attempt_count: int,
max_attempts: int | None,
execution_error_handler: Any,
retry_policy: RetryPolicy,
candidate_record_map: dict[tuple[int, int], str] | None,
candidates: list[ProviderCandidate],
request_id: str | None,
candidate_keys_fallback: list[CandidateKey],
) -> AttemptErrorOutcome:
"""
Handle attempt exception: delegate to external/internal handler, update records.
Returns:
AttemptErrorOutcome; stop_result is non-None only when action==STOP.
"""
has_retry_left = retry_index + 1 < max_retries
# If caller provides an execution_error_handler, prefer it for ExecutionError.
handler_used = False
action = FailoverAction.CONTINUE
if execution_error_handler is not None:
try:
from src.services.request.executor import ExecutionError as _ExecutionError
if isinstance(exc, _ExecutionError):
handler_used = True
action, new_max_retries = await execution_error_handler(
exec_err=exc,
candidate=candidate,
candidate_index=candidate_index,
retry_index=retry_index,
max_retries_for_candidate=max_retries,
record_id=record_id,
attempt_count=attempt_count,
max_attempts=max_attempts,
)
if new_max_retries is not None:
max_retries = max(max_retries, int(new_max_retries))
except Exception:
handler_used = False
last_status_code: int | None = None
if not handler_used:
action = await self._handle_error(
exc,
candidate=candidate,
has_retry_left=has_retry_left,
)
last_status_code = int(getattr(exc, "status_code", 0) or 0) or int(
getattr(exc, "http_status", 0) or 0
)
self._record_attempt_failure(record_id, exc, last_status_code or None)
if action == FailoverAction.STOP:
if retry_policy.mode == RetryMode.PRE_EXPAND and candidate_record_map:
self._mark_remaining_slots_unused(
candidate_record_map=candidate_record_map,
candidates=candidates,
success_candidate_idx=candidate_index,
success_retry_idx=retry_index,
retry_policy=retry_policy,
)
return AttemptErrorOutcome(
action=action,
last_status_code=last_status_code,
max_retries=max_retries,
stop_result=ExecutionResult(
success=False,
error_type=type(exc).__name__,
error_message=self._sanitize(str(exc)),
last_status_code=last_status_code or None,
candidate_keys=self._get_candidate_keys(
request_id=request_id,
fallback=candidate_keys_fallback,
candidates=candidates,
),
attempt_count=attempt_count,
),
)
return AttemptErrorOutcome(
action=action,
last_status_code=last_status_code,
max_retries=max_retries,
)
def _sanitize(self, message: str, max_length: int = 200) -> str:
if not message:
return "request_failed"

5
src/services/orchestration/__init__.py
View File

@@ -4,16 +4,19 @@ Orchestration 模块
提供请求编排相关的组件：
- CandidateResolver: 候选解析器，负责获取和排序可用的 Provider 组合
- RequestDispatcher: 请求分发器，负责执行单个候选请求
- ErrorClassifier: 错误分类器，负责错误分类和处理策略
- ErrorClassifier: 错误分类器，负责错误分类（纯逻辑，无副作用）
- ErrorHandlerService: 错误处理服务，负责错误后的副作用（缓存失效、健康记录等）
"""
from .candidate_resolver import CandidateResolver
from .error_classifier import ErrorAction, ErrorClassifier
from .error_handler import ErrorHandlerService
from .request_dispatcher import RequestDispatcher
__all__ = [
"CandidateResolver",
"RequestDispatcher",
"ErrorClassifier",
"ErrorHandlerService",
"ErrorAction",
]

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

@@ -13,8 +13,6 @@ from typing import Any
import httpx
from sqlalchemy.orm import Session
from src.core.api_format.signature import make_signature_key
from src.core.crypto import CryptoService
from src.core.exceptions import (
ConcurrencyLimitError,
ProviderAuthException,
@@ -28,10 +26,8 @@ from src.core.exceptions import (
from src.core.logger import logger
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_endpoint_signature
from src.services.orchestration.error_handler import ErrorHandlerService
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):
@@ -117,37 +113,11 @@ class ErrorClassifier:
self.db = db
self.adaptive_manager = adaptive_manager or get_adaptive_rpm_manager()
self.cache_scheduler = cache_scheduler
def _extract_oauth_email(self, key: ProviderAPIKey | None) -> str | None:
if not key or str(getattr(key, "auth_type", "") or "").lower() != "oauth":
return None
encrypted_auth_config = getattr(key, "auth_config", None)
if not encrypted_auth_config:
return None
try:
decrypted = CryptoService().decrypt(encrypted_auth_config, silent=True)
auth_config = json.loads(decrypted) if decrypted else {}
except Exception:
return None
email = auth_config.get("email")
if isinstance(email, str):
email = email.strip()
if email:
return email
return None
def _format_key_display(self, key: ProviderAPIKey | None) -> str:
if not key:
return "key=unknown"
key_id = str(getattr(key, "id", "") or "")[:8] or "unknown"
name = str(getattr(key, "name", "") or "").strip()
email = self._extract_oauth_email(key)
parts = [f"key={key_id}"]
if email:
parts.append(f"email={email}")
if name and name != email:
parts.append(f"name={name}")
return " ".join(parts)
self._error_handler = ErrorHandlerService(
db=db,
adaptive_manager=self.adaptive_manager,
cache_scheduler=cache_scheduler,
)
# 表示客户端错误的 error type（不区分大小写）
# 这些 type 表明是请求本身的问题，不应重试
@@ -376,38 +346,6 @@ class ErrorClassifier:
search_text = error_text.lower()
return any(p.lower() in search_text for p in self.THINKING_ERROR_PATTERNS)
def _is_account_validation_required(self, error_text: str | None) -> bool:
"""
检测 403 错误是否为 Google 账号验证要求 (VALIDATION_REQUIRED)
Google 会在某些情况下要求账号所有者手动完成人机验证，
此时所有 API 请求都会返回 403 + VALIDATION_REQUIRED。
这是账号级别的永久性错误，重试无法修复，需要人工干预。
匹配条件（满足任一即可）：
- error.details 中包含 reason=VALIDATION_REQUIRED
- error.status 为 PERMISSION_DENIED 且 message 包含 "verify your account"
- error.message 包含 "verify your account"
Args:
error_text: 错误响应文本
Returns:
是否为账号验证要求错误
"""
if not error_text:
return False
search_text = error_text.lower()
# 快速路径：关键词匹配
if "validation_required" in search_text:
return True
if "verify your account" in search_text and "permission_denied" in search_text:
return True
return False
def _extract_error_message(self, error_text: str | None) -> str | None:
"""
从错误响应中提取错误消息
@@ -480,66 +418,14 @@ class ErrorClassifier:
exception: ProviderRateLimitException,
request_id: str | None = None,
) -> str:
"""
处理 429 速率限制错误的自适应调整
Args:
key: API Key 对象
provider_name: 提供商名称
current_rpm: 当前分钟内的请求数
exception: 速率限制异常
request_id: 请求 ID（用于日志）
Returns:
限制类型: "concurrent" 或 "rpm" 或 "unknown"
"""
try:
# 提取响应头（如果有）
response_headers = {}
if hasattr(exception, "response_headers"):
response_headers = exception.response_headers or {}
# 检测速率限制类型
rate_limit_info = detect_rate_limit_type(
headers=response_headers,
provider_name=provider_name,
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"当前RPM={current_rpm}"
)
# 调用自适应管理器处理
new_limit = self.adaptive_manager.handle_429_error(
db=self.db,
key=key,
rate_limit_info=rate_limit_info,
current_rpm=current_rpm,
)
if rate_limit_info.limit_type == RateLimitType.CONCURRENT:
logger.warning(f" [{request_id}] 并发限制触发（不调整RPM）")
return "concurrent"
elif rate_limit_info.limit_type == RateLimitType.RPM:
if new_limit is not None:
logger.warning(
f" [{request_id}] 自适应调整: Key {key.id[:8]}... RPM限制 -> {new_limit}"
)
else:
logger.info(
f" [{request_id}] 学习中: Key {key.id[:8]}... 观察已记录，暂不设限"
)
return "rpm"
else:
return "unknown"
except Exception as e:
logger.exception(f" [{request_id}] 处理429错误时异常: {e}")
return "unknown"
"""委托给 ErrorHandlerService"""
return await self._error_handler.handle_rate_limit(
key=key,
provider_name=provider_name,
current_rpm=current_rpm,
exception=exception,
request_id=request_id,
)
def convert_http_error(
self,
@@ -650,32 +536,10 @@ class ErrorClassifier:
attempt: int,
max_attempts: int,
) -> dict[str, Any]:
"""
处理 HTTP 错误，返回 extra_data
Args:
http_error: HTTP 状态错误
provider: Provider 对象
endpoint: Endpoint 对象
key: API Key 对象
affinity_key: 亲和性标识符（通常为 API Key ID）
api_format: API 格式
global_model_id: GlobalModel ID（规范化的模型标识）
request_id: 请求 ID
captured_key_concurrent: 捕获的并发数
elapsed_ms: 耗时（毫秒）
attempt: 当前尝试次数
max_attempts: 最大尝试次数
Returns:
Dict[str, Any]: 额外数据，包含：
- error_response: 错误响应文本（如有）
- converted_error: 转换后的异常对象（用于判断是否应该重试）
"""
"""处理 HTTP 错误，返回 extra_data（分类 + 委托副作用给 ErrorHandlerService）"""
provider_name = str(provider.name)
# 尝试读取错误响应内容
# 优先使用 handler 附加的 upstream_response 属性（流式请求中 response.text 可能为空）
error_response_text = getattr(http_error, "upstream_response", None)
if not error_response_text:
try:
@@ -689,111 +553,35 @@ class ErrorClassifier:
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)
# 构建 extra_data，包含转换后的异常
extra_data: dict[str, Any] = {
"converted_error": converted_error,
}
if error_response_text:
extra_data["error_response"] = error_response_text
# client_format：用于缓存亲和性/缓存失效（用户视角）
client_format_str = normalize_endpoint_signature(api_format)
# provider_format：用于健康度/熔断 bucket（Provider 真实端点格式）
fam = str(getattr(endpoint, "api_family", "")).strip().lower()
kind = str(getattr(endpoint, "endpoint_kind", "")).strip().lower()
provider_format_str = make_signature_key(fam, kind) if fam and kind else client_format_str
# 处理客户端请求错误（不应重试，不失效缓存，不记录健康失败）
if isinstance(converted_error, UpstreamClientException):
logger.warning(
f" [{request_id}] 客户端请求错误，不进行重试: {converted_error.message}"
)
return extra_data
# 处理认证错误
if isinstance(converted_error, ProviderAuthException):
if endpoint and key and self.cache_scheduler is not None:
await self.cache_scheduler.invalidate_cache(
affinity_key=affinity_key,
api_format=client_format_str,
global_model_id=global_model_id,
endpoint_id=str(endpoint.id),
key_id=str(key.id),
)
if key:
health_monitor.record_failure(
db=self.db,
key_id=str(key.id),
api_format=provider_format_str,
error_type="ProviderAuthException",
)
# 403 VALIDATION_REQUIRED → 标记 OAuth key 为账号级别封禁
# 这与 test-model 端点的行为对齐（provider_query.py 第 669-690 行）
status_code = http_error.response.status_code if http_error.response else None
if (
status_code == 403
and key
and str(getattr(key, "auth_type", "") or "").lower() == "oauth"
and self._is_account_validation_required(error_response_text)
):
try:
from datetime import datetime, timezone
from src.services.provider.oauth_token import (
OAUTH_ACCOUNT_BLOCK_PREFIX,
)
key.oauth_invalid_at = datetime.now(timezone.utc)
key.oauth_invalid_reason = f"{OAUTH_ACCOUNT_BLOCK_PREFIX}Google 要求验证账号"
key.is_active = False
self.db.commit()
logger.warning(
" [{}] {} 因 403 VALIDATION_REQUIRED 已标记为账号异常并自动停用",
request_id,
self._format_key_display(key),
)
except Exception as mark_exc:
logger.debug(" [{}] 标记 oauth_invalid 失败: {}", request_id, mark_exc)
return extra_data
# 处理限流错误
if isinstance(converted_error, ProviderRateLimitException) and key:
await self.handle_rate_limit(
key=key,
provider_name=provider_name,
current_rpm=captured_key_concurrent,
exception=converted_error,
request_id=request_id,
)
if endpoint and self.cache_scheduler is not None:
await self.cache_scheduler.invalidate_cache(
affinity_key=affinity_key,
api_format=client_format_str,
global_model_id=global_model_id,
endpoint_id=str(endpoint.id),
key_id=str(key.id),
)
else:
# 其他错误也失效缓存
if endpoint and key and self.cache_scheduler is not None:
await self.cache_scheduler.invalidate_cache(
affinity_key=affinity_key,
api_format=client_format_str,
global_model_id=global_model_id,
endpoint_id=str(endpoint.id),
key_id=str(key.id),
)
# 记录健康失败
if key:
health_monitor.record_failure(
db=self.db,
key_id=str(key.id),
api_format=provider_format_str,
error_type=type(converted_error).__name__,
)
# 副作用（委托给 ErrorHandlerService）
await self._error_handler.handle_http_error(
http_error,
converted_error,
error_response_text,
provider=provider,
endpoint=endpoint,
key=key,
affinity_key=affinity_key,
api_format=api_format,
global_model_id=global_model_id,
request_id=request_id,
captured_key_concurrent=captured_key_concurrent,
)
return extra_data
@@ -813,69 +601,20 @@ class ErrorClassifier:
attempt: int,
max_attempts: int,
) -> None:
"""
处理可重试错误
Args:
error: 异常对象
provider: Provider 对象
endpoint: Endpoint 对象
key: API Key 对象
affinity_key: 亲和性标识符（通常为 API Key ID）
api_format: API 格式
global_model_id: GlobalModel ID（规范化的模型标识，用于缓存亲和性）
captured_key_concurrent: 捕获的并发数
elapsed_ms: 耗时（毫秒）
request_id: 请求 ID
attempt: 当前尝试次数
max_attempts: 最大尝试次数
"""
provider_name = str(provider.name)
"""委托给 ErrorHandlerService"""
logger.warning(
f" [{request_id}] 请求失败 (attempt={attempt}/{max_attempts}): "
f"{type(error).__name__}: {str(error)}"
)
# client_format：用于缓存亲和性/缓存失效（用户视角）
client_format_str = normalize_endpoint_signature(api_format)
# provider_format：用于健康度/熔断 bucket（Provider 真实端点格式）
fam = str(getattr(endpoint, "api_family", "")).strip().lower()
kind = str(getattr(endpoint, "endpoint_kind", "")).strip().lower()
provider_format_str = make_signature_key(fam, kind) if fam and kind else client_format_str
# 处理限流错误
if isinstance(error, ProviderRateLimitException) and key:
await self.handle_rate_limit(
key=key,
provider_name=provider_name,
current_rpm=captured_key_concurrent,
exception=error,
request_id=request_id,
)
if endpoint and self.cache_scheduler is not None:
await self.cache_scheduler.invalidate_cache(
affinity_key=affinity_key,
api_format=client_format_str,
global_model_id=global_model_id,
endpoint_id=str(endpoint.id),
key_id=str(key.id),
)
elif endpoint and key and self.cache_scheduler is not None:
# 其他错误也失效缓存
await self.cache_scheduler.invalidate_cache(
affinity_key=affinity_key,
api_format=client_format_str,
global_model_id=global_model_id,
endpoint_id=str(endpoint.id),
key_id=str(key.id),
)
# 记录健康失败
if key:
health_monitor.record_failure(
db=self.db,
key_id=str(key.id),
api_format=provider_format_str,
error_type=type(error).__name__,
)
await self._error_handler.handle_retriable_error(
error,
provider=provider,
endpoint=endpoint,
key=key,
affinity_key=affinity_key,
api_format=api_format,
global_model_id=global_model_id,
captured_key_concurrent=captured_key_concurrent,
request_id=request_id,
)

330
src/services/orchestration/error_handler.py Normal file
View File

@@ -0,0 +1,330 @@
"""
错误处理服务
负责错误发生后的副作用操作（缓存失效、健康记录、RPM 调整、OAuth Key 标记等）。
与 ErrorClassifier（纯分类，无副作用）分离，遵循单一职责原则。
"""
from __future__ import annotations
import json
from typing import Any
import httpx
from sqlalchemy.orm import Session
from src.core.api_format.signature import make_signature_key
from src.core.crypto import CryptoService
from src.core.exceptions import (
ProviderAuthException,
ProviderRateLimitException,
UpstreamClientException,
)
from src.core.logger import logger
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_endpoint_signature
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 ErrorHandlerService:
"""
错误处理服务 - 负责错误发生后的副作用操作
职责：
1. 缓存亲和性失效
2. 健康监控记录
3. 429 自适应 RPM 调整
4. OAuth Key 状态标记
"""
def __init__(
self,
db: Session,
adaptive_manager: Any | None = None,
cache_scheduler: CacheAwareScheduler | None = None,
) -> None:
self.db = db
self.adaptive_manager = adaptive_manager or get_adaptive_rpm_manager()
self.cache_scheduler = cache_scheduler
async def handle_rate_limit(
self,
key: ProviderAPIKey,
provider_name: str,
current_rpm: int | None,
exception: ProviderRateLimitException,
request_id: str | None = None,
) -> str:
"""
处理 429 速率限制错误的自适应调整
Returns:
限制类型: "concurrent" 或 "rpm" 或 "unknown"
"""
try:
response_headers = {}
if hasattr(exception, "response_headers"):
response_headers = exception.response_headers or {}
rate_limit_info = detect_rate_limit_type(
headers=response_headers,
provider_name=provider_name,
current_usage=current_rpm,
)
logger.info(
" [{}] 429错误分析: 类型={}, retry_after={}s, 当前RPM={}",
request_id,
rate_limit_info.limit_type,
rate_limit_info.retry_after,
current_rpm,
)
new_limit = self.adaptive_manager.handle_429_error(
db=self.db,
key=key,
rate_limit_info=rate_limit_info,
current_rpm=current_rpm,
)
if rate_limit_info.limit_type == RateLimitType.CONCURRENT:
logger.warning(" [{}] 并发限制触发（不调整RPM）", request_id)
return "concurrent"
elif rate_limit_info.limit_type == RateLimitType.RPM:
if new_limit is not None:
logger.warning(
" [{}] 自适应调整: Key {}... RPM限制 -> {}",
request_id,
str(key.id)[:8],
new_limit,
)
else:
logger.info(
" [{}] 学习中: Key {}... 观察已记录，暂不设限",
request_id,
str(key.id)[:8],
)
return "rpm"
else:
return "unknown"
except Exception as e:
logger.exception(" [{}] 处理429错误时异常: {}", request_id, e)
return "unknown"
async def handle_http_error(
self,
http_error: httpx.HTTPStatusError,
converted_error: Exception,
error_response_text: str | None,
*,
provider: Provider,
endpoint: ProviderEndpoint,
key: ProviderAPIKey,
affinity_key: str,
api_format: str,
global_model_id: str,
request_id: str | None,
captured_key_concurrent: int | None,
) -> None:
"""
处理 HTTP 错误的副作用（缓存失效、健康记录、OAuth 标记）。
纯副作用方法：不返回分类结果，不做错误转换。
"""
client_format_str = normalize_endpoint_signature(api_format)
fam = str(getattr(endpoint, "api_family", "")).strip().lower()
kind = str(getattr(endpoint, "endpoint_kind", "")).strip().lower()
provider_format_str = make_signature_key(fam, kind) if fam and kind else client_format_str
# 客户端请求错误：不失效缓存，不记录健康失败
if isinstance(converted_error, UpstreamClientException):
return
can_invalidate = bool(endpoint and key and self.cache_scheduler is not None)
# 认证错误
if isinstance(converted_error, ProviderAuthException):
if can_invalidate:
await self._invalidate_cache(
affinity_key, client_format_str, global_model_id, endpoint, key
)
if key:
health_monitor.record_failure(
db=self.db,
key_id=str(key.id),
api_format=provider_format_str,
error_type="ProviderAuthException",
)
# 403 VALIDATION_REQUIRED -> 标记 OAuth key 为账号级别封禁
status_code = http_error.response.status_code if http_error.response else None
if (
status_code == 403
and key
and str(getattr(key, "auth_type", "") or "").lower() == "oauth"
and self._is_account_validation_required(error_response_text)
):
self._mark_oauth_key_blocked(key, request_id)
return
# 限流错误
if isinstance(converted_error, ProviderRateLimitException) and key:
await self.handle_rate_limit(
key=key,
provider_name=str(provider.name),
current_rpm=captured_key_concurrent,
exception=converted_error,
request_id=request_id,
)
# 所有非客户端错误均失效缓存
if can_invalidate:
await self._invalidate_cache(
affinity_key, client_format_str, global_model_id, endpoint, key
)
# 记录健康失败
if key:
health_monitor.record_failure(
db=self.db,
key_id=str(key.id),
api_format=provider_format_str,
error_type=type(converted_error).__name__,
)
async def handle_retriable_error(
self,
error: Exception,
*,
provider: Provider,
endpoint: ProviderEndpoint,
key: ProviderAPIKey,
affinity_key: str,
api_format: str,
global_model_id: str,
captured_key_concurrent: int | None,
request_id: str | None,
) -> None:
"""处理可重试错误的副作用（缓存失效、健康记录）"""
client_format_str = normalize_endpoint_signature(api_format)
fam = str(getattr(endpoint, "api_family", "")).strip().lower()
kind = str(getattr(endpoint, "endpoint_kind", "")).strip().lower()
provider_format_str = make_signature_key(fam, kind) if fam and kind else client_format_str
# 限流错误
if isinstance(error, ProviderRateLimitException) and key:
await self.handle_rate_limit(
key=key,
provider_name=str(provider.name),
current_rpm=captured_key_concurrent,
exception=error,
request_id=request_id,
)
# 失效缓存
if endpoint and key and self.cache_scheduler is not None:
await self._invalidate_cache(
affinity_key, client_format_str, global_model_id, endpoint, key
)
# 记录健康失败
if key:
health_monitor.record_failure(
db=self.db,
key_id=str(key.id),
api_format=provider_format_str,
error_type=type(error).__name__,
)
async def _invalidate_cache(
self,
affinity_key: str,
api_format: str,
global_model_id: str,
endpoint: ProviderEndpoint,
key: ProviderAPIKey,
) -> None:
"""失效缓存亲和性（调用方需确保 cache_scheduler 可用）"""
assert self.cache_scheduler is not None # noqa: S101
await self.cache_scheduler.invalidate_cache(
affinity_key=affinity_key,
api_format=api_format,
global_model_id=global_model_id,
endpoint_id=str(endpoint.id),
key_id=str(key.id),
)
@staticmethod
def _extract_oauth_email(key: ProviderAPIKey | None) -> str | None:
"""从 OAuth Key 的加密 auth_config 中提取邮箱"""
if not key or str(getattr(key, "auth_type", "") or "").lower() != "oauth":
return None
encrypted_auth_config = getattr(key, "auth_config", None)
if not encrypted_auth_config:
return None
try:
decrypted = CryptoService().decrypt(encrypted_auth_config, silent=True)
auth_config = json.loads(decrypted) if decrypted else {}
except Exception:
return None
email = auth_config.get("email")
if isinstance(email, str):
email = email.strip()
if email:
return email
return None
@classmethod
def _format_key_display(cls, key: ProviderAPIKey | None) -> str:
"""格式化 Key 显示信息（用于日志）"""
if not key:
return "key=unknown"
key_id = str(getattr(key, "id", "") or "")[:8] or "unknown"
name = str(getattr(key, "name", "") or "").strip()
email = cls._extract_oauth_email(key)
parts = [f"key={key_id}"]
if email:
parts.append(f"email={email}")
if name and name != email:
parts.append(f"name={name}")
return " ".join(parts)
@staticmethod
def _is_account_validation_required(error_text: str | None) -> bool:
"""
检测 403 错误是否为 Google 账号验证要求 (VALIDATION_REQUIRED)
匹配条件（满足任一即可）：
- error.details 中包含 reason=VALIDATION_REQUIRED
- error.status 为 PERMISSION_DENIED 且 message 包含 "verify your account"
"""
if not error_text:
return False
search_text = error_text.lower()
if "validation_required" in search_text:
return True
if "verify your account" in search_text and "permission_denied" in search_text:
return True
return False
def _mark_oauth_key_blocked(self, key: ProviderAPIKey, request_id: str | None) -> None:
"""标记 OAuth key 为账号级别封禁"""
try:
from datetime import datetime, timezone
from src.services.provider.oauth_token import OAUTH_ACCOUNT_BLOCK_PREFIX
key.oauth_invalid_at = datetime.now(timezone.utc)
key.oauth_invalid_reason = f"{OAUTH_ACCOUNT_BLOCK_PREFIX}Google 要求验证账号"
key.is_active = False
self.db.commit()
logger.warning(
" [{}] {} 因 403 VALIDATION_REQUIRED 已标记为账号异常并自动停用",
request_id,
self._format_key_display(key),
)
except Exception as mark_exc:
logger.debug(" [{}] 标记 oauth_invalid 失败: {}", request_id, mark_exc)

25
src/services/usage/_types.py
View File

@@ -1,6 +1,6 @@
from __future__ import annotations
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import Any
from sqlalchemy.orm import Session
@@ -79,3 +79,26 @@ class UsageRecordParams:
valid_statuses = {"pending", "streaming", "completed", "failed", "cancelled"}
if self.status not in valid_statuses:
raise ValueError(f"无效的状态值: {self.status}，有效值: {valid_statuses}")
@dataclass
class UsageCostInfo:
"""成本与价格信息，用于 _build_usage_params 参数封装"""
# 成本计算结果
input_cost: float = 0.0
output_cost: float = 0.0
cache_creation_cost: float = 0.0
cache_read_cost: float = 0.0
cache_cost: float = 0.0
request_cost: float = 0.0
total_cost: float = 0.0
# 价格信息
input_price: float | None = None
output_price: float | None = None
cache_creation_price: float | None = None
cache_read_price: float | None = None
request_price: float | None = None
# 倍率
actual_rate_multiplier: float = 1.0
is_free_tier: bool = False

92
src/services/usage/recording.py
View File

@@ -12,7 +12,7 @@ from src.core.logger import logger
from src.models.database import ApiKey, Provider, Usage, User
from src.services.billing.token_normalization import normalize_input_tokens_for_billing
from src.services.system.config import SystemConfigService
from src.services.usage._types import UsageRecordParams
from src.services.usage._types import UsageCostInfo, UsageRecordParams
from src.services.usage.error_classifier import classify_error
@@ -74,26 +74,26 @@ class UsageRecordingMixin:
provider_api_key_id: str | None,
status: str,
target_model: str | None,
# 成本计算结果
input_cost: float,
output_cost: float,
cache_creation_cost: float,
cache_read_cost: float,
cache_cost: float,
request_cost: float,
total_cost: float,
# 价格信息
input_price: float | None,
output_price: float | None,
cache_creation_price: float | None,
cache_read_price: float | None,
request_price: float | None,
# 倍率
actual_rate_multiplier: float,
is_free_tier: bool,
cost: UsageCostInfo,
) -> dict[str, Any]:
"""构建 Usage 记录的参数字典（内部方法，避免代码重复）"""
# 展开成本信息
input_cost = cost.input_cost
output_cost = cost.output_cost
cache_creation_cost = cost.cache_creation_cost
cache_read_cost = cost.cache_read_cost
cache_cost = cost.cache_cost
request_cost = cost.request_cost
total_cost = cost.total_cost
input_price = cost.input_price
output_price = cost.output_price
cache_creation_price = cost.cache_creation_price
cache_read_price = cost.cache_read_price
request_price = cost.request_price
actual_rate_multiplier = cost.actual_rate_multiplier
is_free_tier = cost.is_free_tier
# 根据配置决定是否记录请求详情
should_log_headers = SystemConfigService.should_log_headers(db)
should_log_body = SystemConfigService.should_log_body(db)
@@ -463,20 +463,22 @@ class UsageRecordingMixin:
provider_api_key_id=params.provider_api_key_id,
status=params.status,
target_model=params.target_model,
input_cost=input_cost,
output_cost=output_cost,
cache_creation_cost=cache_creation_cost,
cache_read_cost=cache_read_cost,
cache_cost=cache_cost,
request_cost=request_cost,
total_cost=total_cost,
input_price=input_price,
output_price=output_price,
cache_creation_price=cache_creation_price,
cache_read_price=cache_read_price,
request_price=request_price,
actual_rate_multiplier=actual_rate_multiplier,
is_free_tier=is_free_tier,
cost=UsageCostInfo(
input_cost=input_cost,
output_cost=output_cost,
cache_creation_cost=cache_creation_cost,
cache_read_cost=cache_read_cost,
cache_cost=cache_cost,
request_cost=request_cost,
total_cost=total_cost,
input_price=input_price,
output_price=output_price,
cache_creation_price=cache_creation_price,
cache_read_price=cache_read_price,
request_price=request_price,
actual_rate_multiplier=actual_rate_multiplier,
is_free_tier=is_free_tier,
),
)
return usage_params, total_cost
@@ -921,21 +923,17 @@ class UsageRecordingMixin:
provider_api_key_id=provider_api_key_id,
status=status,
target_model=target_model,
input_cost=input_cost,
output_cost=output_cost,
cache_creation_cost=cache_creation_cost,
cache_read_cost=cache_read_cost,
cache_cost=cache_cost,
request_cost=request_cost,
total_cost=total_cost,
# token 价格对异步任务不适用，保持 None
input_price=None,
output_price=None,
cache_creation_price=None,
cache_read_price=None,
request_price=None,
actual_rate_multiplier=actual_rate_multiplier,
is_free_tier=is_free_tier,
cost=UsageCostInfo(
input_cost=input_cost,
output_cost=output_cost,
cache_creation_cost=cache_creation_cost,
cache_read_cost=cache_read_cost,
cache_cost=cache_cost,
request_cost=request_cost,
total_cost=total_cost,
actual_rate_multiplier=actual_rate_multiplier,
is_free_tier=is_free_tier,
),
)
# Upsert（并发幂等：优先用 billing_status 作为结算闸门）