mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-08-09 02:54:37 +00:00
301 lines
11 KiB
Python
301 lines
11 KiB
Python
"""Model prioritization algorithm for selecting cheapest upstream providers."""
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from .core.logging import get_logger
|
|
|
|
if TYPE_CHECKING:
|
|
from .payment.models import Model
|
|
from .upstream import BaseUpstreamProvider
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
def calculate_model_cost_score(model: "Model") -> float:
|
|
"""Calculate a representative cost score for a model.
|
|
|
|
This score is used to compare models when multiple providers offer the same model.
|
|
Lower scores indicate cheaper models.
|
|
|
|
The score is calculated as a weighted average of:
|
|
- Input token cost (weighted by typical input usage)
|
|
- Output token cost (weighted by typical output usage)
|
|
- Fixed request cost
|
|
|
|
Args:
|
|
model: Model instance with pricing information
|
|
|
|
Returns:
|
|
Float representing the cost score. Lower is better.
|
|
"""
|
|
pricing = model.pricing
|
|
|
|
# Weight costs by typical usage patterns
|
|
# Assume average request: 1000 input tokens, 500 output tokens
|
|
TYPICAL_INPUT_TOKENS = 1000.0
|
|
TYPICAL_OUTPUT_TOKENS = 500.0
|
|
|
|
# Calculate weighted cost in USD
|
|
input_cost = pricing.prompt * (TYPICAL_INPUT_TOKENS / 1000.0)
|
|
output_cost = pricing.completion * (TYPICAL_OUTPUT_TOKENS / 1000.0)
|
|
request_cost = pricing.request
|
|
|
|
# Include additional costs if present
|
|
image_cost = (
|
|
getattr(pricing, "image", 0.0) * 0.1
|
|
) # Weight lower as not every request uses images
|
|
web_search_cost = getattr(pricing, "web_search", 0.0) * 0.1
|
|
reasoning_cost = getattr(pricing, "internal_reasoning", 0.0) * 0.2
|
|
|
|
total_cost = (
|
|
input_cost
|
|
+ output_cost
|
|
+ request_cost
|
|
+ image_cost
|
|
+ web_search_cost
|
|
+ reasoning_cost
|
|
)
|
|
|
|
return total_cost
|
|
|
|
|
|
def get_provider_penalty(provider: "BaseUpstreamProvider") -> float:
|
|
"""Calculate a penalty multiplier for certain providers.
|
|
|
|
This allows applying policy-based adjustments beyond pure cost.
|
|
For example, preferring certain providers for reliability or features.
|
|
|
|
Args:
|
|
provider: UpstreamProvider instance
|
|
|
|
Returns:
|
|
Float multiplier to apply to cost (1.0 = no penalty, >1.0 = penalize)
|
|
"""
|
|
# Default: no penalty
|
|
penalty = 1.0
|
|
|
|
# Check if this is OpenRouter (can be identified by base URL)
|
|
base_url = getattr(provider, "base_url", "")
|
|
if "openrouter.ai" in base_url.lower():
|
|
# Small penalty for OpenRouter to prefer other providers when costs are very close
|
|
# This maintains the original behavior of preferring non-OpenRouter providers
|
|
penalty = 1.001 # 0.1% penalty
|
|
|
|
return penalty
|
|
|
|
|
|
def should_prefer_model(
|
|
candidate_model: "Model",
|
|
candidate_provider: "BaseUpstreamProvider",
|
|
current_model: "Model",
|
|
current_provider: "BaseUpstreamProvider",
|
|
alias: str,
|
|
) -> bool:
|
|
"""Determine if candidate model should replace current model for an alias.
|
|
|
|
This is the core decision function for model prioritization. It considers:
|
|
1. Alias matching quality (exact match vs. canonical slug match)
|
|
2. Model cost (lower is better)
|
|
3. Provider penalties (e.g., slight preference against OpenRouter)
|
|
|
|
Args:
|
|
candidate_model: The new model being considered
|
|
candidate_provider: Provider offering the candidate model
|
|
current_model: The currently selected model for this alias
|
|
current_provider: Provider offering the current model
|
|
alias: The model alias being mapped
|
|
|
|
Returns:
|
|
True if candidate should replace current, False otherwise
|
|
"""
|
|
|
|
def get_base_model_id(model_id: str) -> str:
|
|
"""Get base model ID by removing provider prefix."""
|
|
return model_id.split("/", 1)[1] if "/" in model_id else model_id
|
|
|
|
def alias_priority(model: "Model") -> int:
|
|
"""Rank how strong the mapping of alias->model is.
|
|
|
|
Highest priority when alias exactly equals the model ID without provider prefix.
|
|
Next when alias equals canonical slug without prefix. Otherwise lowest.
|
|
"""
|
|
model_base = get_base_model_id(model.id)
|
|
if model_base == alias:
|
|
return 3
|
|
if model.canonical_slug:
|
|
canonical_base = get_base_model_id(model.canonical_slug)
|
|
if canonical_base == alias:
|
|
return 2
|
|
return 1
|
|
|
|
candidate_alias_priority = alias_priority(candidate_model)
|
|
current_alias_priority = alias_priority(current_model)
|
|
|
|
# If candidate has better alias match, prefer it regardless of cost
|
|
if candidate_alias_priority > current_alias_priority:
|
|
return True
|
|
|
|
# If current has better alias match, keep it regardless of cost
|
|
if current_alias_priority > candidate_alias_priority:
|
|
return False
|
|
|
|
# Same alias priority - compare costs
|
|
candidate_cost = calculate_model_cost_score(candidate_model)
|
|
current_cost = calculate_model_cost_score(current_model)
|
|
|
|
# Apply provider penalties
|
|
candidate_adjusted = candidate_cost * get_provider_penalty(candidate_provider)
|
|
current_adjusted = current_cost * get_provider_penalty(current_provider)
|
|
|
|
# Prefer lower adjusted cost
|
|
should_replace = candidate_adjusted < current_adjusted
|
|
|
|
# Log provider changes when candidate wins
|
|
if should_replace:
|
|
candidate_provider_name = getattr(
|
|
candidate_provider, "provider_type", "unknown"
|
|
)
|
|
current_provider_name = getattr(current_provider, "provider_type", "unknown")
|
|
logger.debug(
|
|
f"Model selection for alias '{alias}': choosing {candidate_provider_name} "
|
|
f"(cost: ${candidate_adjusted:.6f}) over {current_provider_name} "
|
|
f"(cost: ${current_adjusted:.6f})"
|
|
)
|
|
|
|
return should_replace
|
|
|
|
|
|
def create_model_mappings(
|
|
upstreams: list["BaseUpstreamProvider"],
|
|
overrides_by_id: dict[str, tuple],
|
|
disabled_model_ids: set[str],
|
|
) -> tuple[dict[str, "Model"], dict[str, "BaseUpstreamProvider"], dict[str, "Model"]]:
|
|
"""Create optimal model mappings based on cost and provider preferences.
|
|
|
|
This is the main entry point for the algorithm. It processes all upstream providers
|
|
and creates three mappings based on cost optimization:
|
|
|
|
1. model_instances: alias -> Model (all model aliases mapped to their Model objects)
|
|
2. provider_map: alias -> UpstreamProvider (which provider to use for each alias)
|
|
3. unique_models: base_id -> Model (unique models without provider prefixes)
|
|
|
|
The algorithm:
|
|
- Processes non-OpenRouter providers first (they're typically cheaper)
|
|
- Then processes OpenRouter models (they can still win if cheaper)
|
|
- For each model alias, uses should_prefer_model() to select the best provider
|
|
|
|
Args:
|
|
upstreams: List of all upstream provider instances
|
|
overrides_by_id: Dict of model overrides from database {model_id: (ModelRow, fee)}
|
|
disabled_model_ids: Set of model IDs that should be excluded
|
|
|
|
Returns:
|
|
Tuple of (model_instances, provider_map, unique_models)
|
|
"""
|
|
from .payment.models import _row_to_model
|
|
from .upstream.helpers import resolve_model_alias
|
|
|
|
model_instances: dict[str, "Model"] = {}
|
|
provider_map: dict[str, "BaseUpstreamProvider"] = {}
|
|
unique_models: dict[str, "Model"] = {}
|
|
|
|
# Separate OpenRouter from other providers
|
|
openrouter: "BaseUpstreamProvider" | None = None
|
|
other_upstreams: list["BaseUpstreamProvider"] = []
|
|
|
|
for upstream in upstreams:
|
|
base_url = getattr(upstream, "base_url", "")
|
|
if base_url == "https://openrouter.ai/api/v1":
|
|
openrouter = upstream
|
|
else:
|
|
other_upstreams.append(upstream)
|
|
|
|
def get_base_model_id(model_id: str) -> str:
|
|
"""Get base model ID by removing provider prefix."""
|
|
return model_id.split("/", 1)[1] if "/" in model_id else model_id
|
|
|
|
def _maybe_set_alias(
|
|
alias: str, model: "Model", provider: "BaseUpstreamProvider"
|
|
) -> None:
|
|
"""Set alias to model/provider if not set or if new model is preferred."""
|
|
existing_model = model_instances.get(alias)
|
|
if not existing_model:
|
|
# No existing mapping, set it
|
|
model_instances[alias] = model
|
|
provider_map[alias] = provider
|
|
else:
|
|
# Check if candidate should replace existing
|
|
existing_provider = provider_map[alias]
|
|
if should_prefer_model(
|
|
model, provider, existing_model, existing_provider, alias
|
|
):
|
|
model_instances[alias] = model
|
|
provider_map[alias] = provider
|
|
|
|
def process_provider_models(
|
|
upstream: "BaseUpstreamProvider", is_openrouter: bool = False
|
|
) -> None:
|
|
"""Process all models from a given provider."""
|
|
upstream_prefix = getattr(upstream, "upstream_name", None)
|
|
|
|
for model in upstream.get_cached_models():
|
|
if not model.enabled or model.id in disabled_model_ids:
|
|
continue
|
|
|
|
# Apply overrides if present
|
|
if model.id in overrides_by_id:
|
|
override_row, provider_fee = overrides_by_id[model.id]
|
|
model_to_use = _row_to_model(
|
|
override_row, apply_provider_fee=True, provider_fee=provider_fee
|
|
)
|
|
else:
|
|
model_to_use = model
|
|
|
|
# Add to unique models
|
|
base_id = get_base_model_id(model_to_use.id)
|
|
if not is_openrouter or base_id not in unique_models:
|
|
unique_model = model_to_use.copy(update={"id": base_id})
|
|
unique_models[base_id] = unique_model
|
|
|
|
# Get all aliases for this model
|
|
aliases = resolve_model_alias(
|
|
model_to_use.id,
|
|
model_to_use.canonical_slug,
|
|
alias_ids=model_to_use.alias_ids,
|
|
)
|
|
|
|
# Add prefixed alias if applicable
|
|
if upstream_prefix and "/" not in model_to_use.id:
|
|
prefixed_id = f"{upstream_prefix}/{model_to_use.id}"
|
|
if prefixed_id not in aliases:
|
|
aliases.append(prefixed_id)
|
|
|
|
# Try to set each alias
|
|
for alias in aliases:
|
|
_maybe_set_alias(alias, model_to_use, upstream)
|
|
|
|
# Process non-OpenRouter providers first (they're typically cheaper)
|
|
for upstream in other_upstreams:
|
|
process_provider_models(upstream, is_openrouter=False)
|
|
|
|
# Process OpenRouter last - models only win if they're cheaper or better matched
|
|
if openrouter:
|
|
process_provider_models(openrouter, is_openrouter=True)
|
|
|
|
# Log provider distribution
|
|
provider_counts: dict[str, int] = {}
|
|
for provider in provider_map.values():
|
|
provider_name = getattr(provider, "upstream_name", "unknown")
|
|
provider_counts[provider_name] = provider_counts.get(provider_name, 0) + 1
|
|
|
|
logger.debug(
|
|
"Created model mappings",
|
|
extra={
|
|
"unique_model_count": len(unique_models),
|
|
"total_alias_count": len(model_instances),
|
|
"provider_distribution": provider_counts,
|
|
},
|
|
)
|
|
|
|
return model_instances, provider_map, unique_models
|