63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
from apps.ai.providers import ResolvedModel
|
|
|
|
from .models import AiModel, ModelAlias
|
|
|
|
|
|
class AliasResolutionError(RuntimeError):
|
|
"""Base error for alias resolution failures."""
|
|
|
|
|
|
class AliasNotFoundError(AliasResolutionError):
|
|
"""Raised when an active alias cannot be found."""
|
|
|
|
|
|
class ModelCapabilityError(AliasResolutionError):
|
|
"""Raised when an alias points to a model without the required capability."""
|
|
|
|
|
|
REQUIRED_CAPABILITIES = {
|
|
ModelAlias.OperationType.TITLE: frozenset({"text"}),
|
|
ModelAlias.OperationType.IMAGE: frozenset({"image"}),
|
|
ModelAlias.OperationType.VISION: frozenset({"text", "vision"}),
|
|
}
|
|
|
|
|
|
def resolve_model_alias(operation_type: str, alias: str | None = None) -> ModelAlias:
|
|
"""Resolve an external capability alias to an active ModelAlias row."""
|
|
required_capabilities = REQUIRED_CAPABILITIES.get(operation_type)
|
|
if required_capabilities is None:
|
|
raise AliasResolutionError(f"unsupported operation_type: {operation_type}")
|
|
|
|
queryset = ModelAlias.objects.select_related("ai_model").filter(
|
|
operation_type=operation_type,
|
|
is_active=True,
|
|
ai_model__is_active=True,
|
|
)
|
|
if alias:
|
|
queryset = queryset.filter(alias=alias)
|
|
else:
|
|
queryset = queryset.filter(is_default=True)
|
|
|
|
model_alias = queryset.order_by("id").first()
|
|
if model_alias is None:
|
|
if alias:
|
|
raise AliasNotFoundError(f"alias not found: {operation_type}:{alias}")
|
|
raise AliasNotFoundError(f"default alias not found: {operation_type}")
|
|
|
|
ai_model: AiModel = model_alias.ai_model
|
|
capabilities = ai_model.capabilities_set()
|
|
missing_capabilities = required_capabilities - capabilities
|
|
if missing_capabilities:
|
|
raise ModelCapabilityError(
|
|
f"alias {model_alias.alias} maps to model {ai_model.name} without "
|
|
f"{', '.join(sorted(missing_capabilities))} capability"
|
|
)
|
|
return model_alias
|
|
|
|
|
|
def resolve_alias(operation_type: str, alias: str | None = None) -> ResolvedModel:
|
|
"""Resolve an external capability alias to a provider-ready model config."""
|
|
return resolve_model_alias(operation_type, alias).ai_model.to_resolved_model()
|