mirror of
https://github.com/Alishahryar1/free-claude-code.git
synced 2026-06-02 06:13:46 +02:00
Alishahryar1
76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
"""Model routing for Claude-compatible requests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from loguru import logger
|
|
|
|
from config.settings import Settings
|
|
|
|
from .models.anthropic import MessagesRequest, TokenCountRequest
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ResolvedModel:
|
|
original_model: str
|
|
provider_id: str
|
|
provider_model: str
|
|
provider_model_ref: str
|
|
thinking_enabled: bool
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RoutedMessagesRequest:
|
|
request: MessagesRequest
|
|
resolved: ResolvedModel
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RoutedTokenCountRequest:
|
|
request: TokenCountRequest
|
|
resolved: ResolvedModel
|
|
|
|
|
|
class ModelRouter:
|
|
"""Resolve incoming Claude model names to configured provider/model pairs."""
|
|
|
|
def __init__(self, settings: Settings):
|
|
self._settings = settings
|
|
|
|
def resolve(self, claude_model_name: str) -> ResolvedModel:
|
|
provider_model_ref = self._settings.resolve_model(claude_model_name)
|
|
thinking_enabled = self._settings.resolve_thinking(claude_model_name)
|
|
provider_id = Settings.parse_provider_type(provider_model_ref)
|
|
provider_model = Settings.parse_model_name(provider_model_ref)
|
|
if provider_model != claude_model_name:
|
|
logger.debug(
|
|
"MODEL MAPPING: '{}' -> '{}'", claude_model_name, provider_model
|
|
)
|
|
return ResolvedModel(
|
|
original_model=claude_model_name,
|
|
provider_id=provider_id,
|
|
provider_model=provider_model,
|
|
provider_model_ref=provider_model_ref,
|
|
thinking_enabled=thinking_enabled,
|
|
)
|
|
|
|
def resolve_messages_request(
|
|
self, request: MessagesRequest
|
|
) -> RoutedMessagesRequest:
|
|
"""Return an internal routed request context."""
|
|
resolved = self.resolve(request.model)
|
|
routed = request.model_copy(deep=True)
|
|
routed.model = resolved.provider_model
|
|
return RoutedMessagesRequest(request=routed, resolved=resolved)
|
|
|
|
def resolve_token_count_request(
|
|
self, request: TokenCountRequest
|
|
) -> RoutedTokenCountRequest:
|
|
"""Return an internal token-count request context."""
|
|
resolved = self.resolve(request.model)
|
|
routed = request.model_copy(
|
|
update={"model": resolved.provider_model}, deep=True
|
|
)
|
|
return RoutedTokenCountRequest(request=routed, resolved=resolved)
|