295 lines
10 KiB
Python
295 lines
10 KiB
Python
"""模型路由策略的集中读取与校验。
|
||
|
||
本模块只负责把 Django settings 中的 ``MODEL_ROUTING_POLICY`` 转成不可变配置对象，
|
||
不执行重试、Fallback、Provider 调用或账务操作。业务入口后续只消费这里返回的策略，
|
||
不得各自维护 timeout / sleep / attempts 常量。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Mapping, Sequence
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
|
||
from django.conf import settings
|
||
|
||
|
||
class RoutingPolicyConfigurationError(ValueError):
|
||
"""模型路由策略格式或取值不合法。"""
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class TextRoutingPolicy:
|
||
retry_delays: tuple[float, ...]
|
||
request_timeout: float
|
||
stream_timeout: float
|
||
total_timeout: float
|
||
retry_after_cap: float
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class RequestRoutingPolicy:
|
||
retry_delays: tuple[float, ...]
|
||
request_timeout: float
|
||
total_timeout: float
|
||
retry_after_cap: float
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class VideoRoutingPolicy:
|
||
submit_retry_delays: tuple[float, ...]
|
||
submit_timeout: float
|
||
submit_total_timeout: float
|
||
poll_request_timeout: float
|
||
retry_after_cap: float
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class PostprocessRoutingPolicy:
|
||
retry_delays: tuple[float, ...]
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class ModelRoutingPolicy:
|
||
max_models: int
|
||
max_calls: int
|
||
jitter_ratio: float
|
||
text: TextRoutingPolicy
|
||
image: RequestRoutingPolicy
|
||
audio: RequestRoutingPolicy
|
||
video: VideoRoutingPolicy
|
||
postprocess: PostprocessRoutingPolicy
|
||
|
||
|
||
_LABELS = {
|
||
"max_models": "最多尝试模型数量",
|
||
"max_calls": "最多真实模型调用次数",
|
||
"jitter_ratio": "重试随机抖动比例",
|
||
"retry_delays": "重试等待时间列表",
|
||
"request_timeout": "单次调用超时",
|
||
"stream_timeout": "单次流式调用超时",
|
||
"total_timeout": "逻辑任务总时限",
|
||
"retry_after_cap": "Retry-After 最长等待时间",
|
||
"submit_retry_delays": "视频提交重试等待时间列表",
|
||
"submit_timeout": "视频单次提交超时",
|
||
"submit_total_timeout": "视频提交阶段总时限",
|
||
"poll_request_timeout": "视频单次轮询超时",
|
||
}
|
||
|
||
|
||
def _error(path: str, message: str) -> RoutingPolicyConfigurationError:
|
||
key = path.rsplit(".", 1)[-1].split("[", 1)[0]
|
||
label = _LABELS.get(key, path)
|
||
return RoutingPolicyConfigurationError(f"{path}（{label}）{message}")
|
||
|
||
|
||
def _mapping(value: Any, path: str) -> Mapping[str, Any]:
|
||
if not isinstance(value, Mapping):
|
||
raise _error(path, f"必须是对象，当前值：{value!r}")
|
||
return value
|
||
|
||
|
||
def _required(source: Mapping[str, Any], key: str, path: str) -> Any:
|
||
if key not in source:
|
||
raise _error(f"{path}.{key}", "为必填配置")
|
||
return source[key]
|
||
|
||
|
||
def _integer(value: Any, path: str, *, minimum: int, maximum: int) -> int:
|
||
if isinstance(value, bool) or not isinstance(value, int):
|
||
raise _error(path, f"必须是整数，当前值：{value!r}")
|
||
if not minimum <= value <= maximum:
|
||
raise _error(path, f"必须在 {minimum}～{maximum} 之间，当前值：{value!r}")
|
||
return value
|
||
|
||
|
||
def _number(value: Any, path: str, *, minimum: float, maximum: float) -> float:
|
||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||
raise _error(path, f"必须是数字，当前值：{value!r}")
|
||
result = float(value)
|
||
if not minimum <= result <= maximum:
|
||
raise _error(path, f"必须在 {minimum:g}～{maximum:g} 之间，当前值：{value!r}")
|
||
return result
|
||
|
||
|
||
def _delays(value: Any, path: str) -> tuple[float, ...]:
|
||
if isinstance(value, (str, bytes)) or not isinstance(value, Sequence):
|
||
raise _error(path, f"必须是数字列表，当前值：{value!r}")
|
||
return tuple(
|
||
_number(item, f"{path}[{index}]", minimum=0, maximum=3600)
|
||
for index, item in enumerate(value)
|
||
)
|
||
|
||
|
||
def _ensure_not_greater(*, smaller: float, larger: float, smaller_path: str, larger_path: str) -> None:
|
||
if smaller > larger:
|
||
raise _error(
|
||
smaller_path,
|
||
f"不得大于 {larger_path}（任务总时限），当前为 {smaller:g} > {larger:g}",
|
||
)
|
||
|
||
|
||
def _field_number(
|
||
source: Mapping[str, Any],
|
||
key: str,
|
||
path: str,
|
||
*,
|
||
minimum: float = 0,
|
||
maximum: float = 86400,
|
||
) -> float:
|
||
return _number(_required(source, key, path), f"{path}.{key}", minimum=minimum, maximum=maximum)
|
||
|
||
|
||
def _field_delays(source: Mapping[str, Any], key: str, path: str) -> tuple[float, ...]:
|
||
return _delays(_required(source, key, path), f"{path}.{key}")
|
||
|
||
|
||
def _validate_budget(
|
||
*,
|
||
total: float,
|
||
total_path: str,
|
||
bounded_values: Mapping[str, float],
|
||
delays: tuple[float, ...] = (),
|
||
delays_path: str = "",
|
||
) -> None:
|
||
for value_path, value in bounded_values.items():
|
||
_ensure_not_greater(
|
||
smaller=value,
|
||
larger=total,
|
||
smaller_path=value_path,
|
||
larger_path=total_path,
|
||
)
|
||
for index, delay in enumerate(delays):
|
||
_ensure_not_greater(
|
||
smaller=delay,
|
||
larger=total,
|
||
smaller_path=f"{delays_path}[{index}]",
|
||
larger_path=total_path,
|
||
)
|
||
|
||
|
||
def _request_policy(raw: Any, path: str) -> RequestRoutingPolicy:
|
||
source = _mapping(raw, path)
|
||
retry_delays = _field_delays(source, "retry_delays", path)
|
||
request_timeout = _field_number(source, "request_timeout", path, minimum=1)
|
||
total_timeout = _field_number(source, "total_timeout", path, minimum=1)
|
||
retry_after_cap = _field_number(source, "retry_after_cap", path, maximum=3600)
|
||
_validate_budget(
|
||
total=total_timeout,
|
||
total_path=f"{path}.total_timeout",
|
||
bounded_values={
|
||
f"{path}.request_timeout": request_timeout,
|
||
f"{path}.retry_after_cap": retry_after_cap,
|
||
},
|
||
delays=retry_delays,
|
||
delays_path=f"{path}.retry_delays",
|
||
)
|
||
return RequestRoutingPolicy(
|
||
retry_delays=retry_delays,
|
||
request_timeout=request_timeout,
|
||
total_timeout=total_timeout,
|
||
retry_after_cap=retry_after_cap,
|
||
)
|
||
|
||
|
||
def _text_policy(raw: Any) -> TextRoutingPolicy:
|
||
path = "MODEL_ROUTING_POLICY.text"
|
||
source = _mapping(raw, path)
|
||
retry_delays = _field_delays(source, "retry_delays", path)
|
||
request_timeout = _field_number(source, "request_timeout", path, minimum=1)
|
||
stream_timeout = _field_number(source, "stream_timeout", path, minimum=1)
|
||
total_timeout = _field_number(source, "total_timeout", path, minimum=1)
|
||
retry_after_cap = _field_number(source, "retry_after_cap", path, maximum=3600)
|
||
_validate_budget(
|
||
total=total_timeout,
|
||
total_path=f"{path}.total_timeout",
|
||
bounded_values={
|
||
f"{path}.request_timeout": request_timeout,
|
||
f"{path}.stream_timeout": stream_timeout,
|
||
f"{path}.retry_after_cap": retry_after_cap,
|
||
},
|
||
delays=retry_delays,
|
||
delays_path=f"{path}.retry_delays",
|
||
)
|
||
return TextRoutingPolicy(
|
||
retry_delays=retry_delays,
|
||
request_timeout=request_timeout,
|
||
stream_timeout=stream_timeout,
|
||
total_timeout=total_timeout,
|
||
retry_after_cap=retry_after_cap,
|
||
)
|
||
|
||
|
||
def _video_policy(raw: Any) -> VideoRoutingPolicy:
|
||
path = "MODEL_ROUTING_POLICY.video"
|
||
source = _mapping(raw, path)
|
||
retry_delays = _field_delays(source, "submit_retry_delays", path)
|
||
submit_timeout = _field_number(source, "submit_timeout", path, minimum=1)
|
||
submit_total_timeout = _field_number(source, "submit_total_timeout", path, minimum=1)
|
||
poll_request_timeout = _field_number(source, "poll_request_timeout", path, minimum=1)
|
||
retry_after_cap = _field_number(source, "retry_after_cap", path, maximum=3600)
|
||
_validate_budget(
|
||
total=submit_total_timeout,
|
||
total_path=f"{path}.submit_total_timeout",
|
||
bounded_values={
|
||
f"{path}.submit_timeout": submit_timeout,
|
||
f"{path}.retry_after_cap": retry_after_cap,
|
||
},
|
||
delays=retry_delays,
|
||
delays_path=f"{path}.submit_retry_delays",
|
||
)
|
||
return VideoRoutingPolicy(
|
||
submit_retry_delays=retry_delays,
|
||
submit_timeout=submit_timeout,
|
||
submit_total_timeout=submit_total_timeout,
|
||
poll_request_timeout=poll_request_timeout,
|
||
retry_after_cap=retry_after_cap,
|
||
)
|
||
|
||
|
||
def load_model_routing_policy(raw_policy: Any | None = None) -> ModelRoutingPolicy:
|
||
"""读取并校验模型路由策略，返回不可变对象。
|
||
|
||
``raw_policy`` 主要供测试和启动检查注入；省略时读取 Django settings。
|
||
配置错误统一抛出带中文字段含义、错误值和合法范围的异常。
|
||
"""
|
||
|
||
raw = settings.MODEL_ROUTING_POLICY if raw_policy is None else raw_policy
|
||
source = _mapping(raw, "MODEL_ROUTING_POLICY")
|
||
|
||
path = "MODEL_ROUTING_POLICY"
|
||
max_models = _integer(_required(source, "max_models", path), f"{path}.max_models", minimum=1, maximum=10)
|
||
max_calls = _integer(_required(source, "max_calls", path), f"{path}.max_calls", minimum=1, maximum=20)
|
||
if max_calls < max_models:
|
||
raise _error(
|
||
"MODEL_ROUTING_POLICY.max_calls",
|
||
f"不得小于 max_models（最多尝试模型数量），当前为 {max_calls} < {max_models}",
|
||
)
|
||
jitter_ratio = _field_number(source, "jitter_ratio", path, maximum=1)
|
||
postprocess_path = f"{path}.postprocess"
|
||
postprocess_source = _mapping(_required(source, "postprocess", path), postprocess_path)
|
||
|
||
return ModelRoutingPolicy(
|
||
max_models=max_models,
|
||
max_calls=max_calls,
|
||
jitter_ratio=jitter_ratio,
|
||
text=_text_policy(_required(source, "text", path)),
|
||
image=_request_policy(_required(source, "image", path), f"{path}.image"),
|
||
audio=_request_policy(_required(source, "audio", path), f"{path}.audio"),
|
||
video=_video_policy(_required(source, "video", path)),
|
||
postprocess=PostprocessRoutingPolicy(
|
||
retry_delays=_field_delays(postprocess_source, "retry_delays", postprocess_path)
|
||
),
|
||
)
|
||
|
||
|
||
__all__ = [
|
||
"ModelRoutingPolicy",
|
||
"PostprocessRoutingPolicy",
|
||
"RequestRoutingPolicy",
|
||
"RoutingPolicyConfigurationError",
|
||
"TextRoutingPolicy",
|
||
"VideoRoutingPolicy",
|
||
"load_model_routing_policy",
|
||
]
|