Class ModelMapper

uncommon_route/model_map.py:327–713

Discovers upstream models and provides a dynamic pool with pricing, capabilities, and name resolution. After ``discover()`` runs, the mapper knows every model the upstream serves and can provide live pricing and inferred capabilities for each. Static config (``DEFAULT_MODEL_PRICING`

@dataclass
class ModelMapper:
    """Discovers upstream models and provides a dynamic pool with pricing,
    capabilities, and name resolution.

    After ``discover()`` runs, the mapper knows every model the upstream
    serves and can provide live pricing and inferred capabilities for each.
    Static config (``DEFAULT_MODEL_PRICING``, etc.) is only used as a
    fallback when discovery has not yet run.
    """

    upstream_url: str
    provider: str = ""
    is_gateway: bool = False

    _upstream_models: set[str] = field(default_factory=set, repr=False)
    _map: dict[str, str] = field(default_factory=dict, repr=False)
    _discovered: bool = False

    _pool: dict[str, DiscoveredModel] = field(default_factory=dict, repr=False)
    _learned_aliases: dict[str, str] = field(default_factory=dict, repr=False)
    _last_discovery_time: float = 0.0

    def __post_init__(self) -> None:
        self.provider, self.is_gateway = detect_provider(self.upstream_url)
        self._load_learned_aliases()

    # ---- discovery --------------------------------------------------------

    async def discover(self, api_key: str | None = None) -> int:
        """Fetch ``/v1/models`` from upstream and build the model pool.

        Extracts pricing and infers capabilities from upstream data.
        Returns the number of models discovered (0 on failure).
        """
        if not self.upstream_url:
            return 0

        models_url = f"{self.upstream_url.rstrip('/')}/models"
        headers: dict[str, str] = {"user-agent": "uncommon-route/model-discovery"}
        if api_key:
            headers["authorization"] = f"Bearer {api_key}"

        try:
            async with httpx.AsyncClient(
                timeout=httpx.Timeout(10.0, connect=5.0),
            ) as client:
                resp = await client.get(models_url, headers=headers)
                if resp.status_code != 200:
                    logger.warning(
                        "Model discovery: HTTP %d from %s", resp.status_code, models_url,
                    )
                    return 0
                data = resp.json()
                raw = data.get("data", [])

                self._pool.clear()
                self._upstream_models.clear()