"""Shared model-identity helper for third-party sources (leaderboards, aggregators, provider price lists). Those sources name models with *API ids* (`anthropic/claude-3-7-sonnet-20250219`, `gemini/gemini-2.5-pro-preview-05-06`, `openrouter/x-ai/grok-4`, `gpt-4o-2024-08-06`, `Qwen/Qwen2.5-Coder-32B-Instruct`) or free text. This module turns such an id into: * the developer organisation (registry `org_ref`) when the id carries a first-party prefix or family word — never a guess beyond the deterministic tables below; * resolver-friendly aliases (the raw id, the id without provider prefix, the id without evaluator effort suffixes); * identifiers, **only when the caller marks the id as trusted** (= the string really is the vendor's own API id, e.g. the `--model` argument aider passed to the API). Free-text tags never become identifiers: the resolver refuses to merge an alias match when the entity already carries a *different* value for the same scheme, so a wrong `anthropic_model_id` would split an entity instead of linking it. * the evaluation-effort configuration encoded in the id or its label ("(xhigh)", "-thinking-64k", "(Non-reasoning)") — effort variants are result *configurations*, never model entities. """ from __future__ import annotations import re from dataclasses import dataclass, field from typing import Any from aiatlas.ontology.models import OFFICIAL_THINKING_RELEASES, analyze_model_name, family_release_hint from aiatlas.registry import org_by_hf, org_ref, organizations from aiatlas.sdk.facts import EntityRef, Facts # provider / gateway prefixes used by litellm, aider, OpenRouter, LiveBench… → (identifier scheme for the remainder, developer org key) # scheme None = the prefix is a gateway or an OpenAI-compatible endpoint, the remainder decides. PROVIDER_PREFIXES: dict[str, tuple[str | None, str | None]] = { "anthropic": ("anthropic_model_id", "anthropic"), "gemini": ("gemini_model_id", "google"), "google": ("gemini_model_id", "google"), "vertex_ai": (None, "google"), "deepseek": ("deepseek_model_id", "deepseek"), "mistral": ("mistral_model_id", "mistral"), "mistralai": ("mistral_model_id", "mistral"), "xai": ("xai_model_id", "xai"), "x-ai": ("xai_model_id", "xai"), "cohere": ("cohere_model_id", "cohere"), "groq": ("groq_model_id", None), "together_ai": ("together_ai_model_slug", None), "fireworks_ai": ("fireworks_model_id", None), "openai": (None, None), # litellm/aider route *any* OpenAI-compatible endpoint through `openai/…` "azure": (None, None), "bedrock": (None, None), "nvidia_nim": (None, None), "openrouter": (None, None), } # OpenRouter vendor slug → registry organisation (shared with the OpenRouter connector) VENDOR_ORG: dict[str, str] = { "meta": "meta-ai", "meta-llama": "meta-ai", "liquid": "liquid-ai", "x-ai": "xai", "mistralai": "mistral", "z-ai": "zhipu", "thudm": "zhipu", "moonshotai": "moonshot", "bytedance": "bytedance", "bytedance-seed": "bytedance", "ibm-granite": "ibm", "google": "google", "amazon": "amazon", "qwen": "qwen", "deepseek": "deepseek", "anthropic": "anthropic", "openai": "openai", "cohere": "cohere", "nvidia": "nvidia", "microsoft": "microsoft", "perplexity": "perplexity", "minimax": "minimax", "tencent": "tencent", "baidu": "baidu", "ai21": "ai21", "nousresearch": "nous-research", "openrouter": "openrouter", "all-hands": "all-hands-ai", "stepfun": "stepfun", "stepfun-ai": "stepfun", "upstage": "upstage", "kwaipilot": "kwaipilot", "thinking-machines": "thinking-machines", "thinkingmachines": "thinking-machines", "allenai": "allenai", "eleutherai": "eleutherai", "internlm": "internlm", } # first-party naming patterns of bare API ids → (identifier scheme or None, developer org key). Ordered: specific families first. _BARE: list[tuple[re.Pattern[str], str | None, str | None]] = [ (re.compile(r"^claude[-_ ]"), "anthropic_model_id", "anthropic"), (re.compile(r"^(chatgpt-|gpt-(?!oss)|o[1-9](-|$)|codex|davinci|text-embedding-|dall-e|sora|tts-1|computer-use-preview)"), "openai_model_id", "openai"), (re.compile(r"^gpt-oss"), None, "openai"), (re.compile(r"^(gemini-|imagen-|veo-|lyria-|gemini-embedding)"), "gemini_model_id", "google"), (re.compile(r"^(gemma-|paligemma|medgemma|shieldgemma)"), None, "google"), (re.compile(r"^grok-"), "xai_model_id", "xai"), (re.compile(r"^deepseek-"), "deepseek_model_id", "deepseek"), (re.compile(r"^(mistral-|mixtral-|ministral-|codestral|magistral|devstral|pixtral|voxtral|open-mistral|open-mixtral|mistral-embed)"), "mistral_model_id", "mistral"), (re.compile(r"^(command-|command$|c4ai-|embed-(english|multilingual|v)|rerank-)"), "cohere_model_id", "cohere"), (re.compile(r"^aya-"), None, "cohere"), (re.compile(r"^(hermes)"), None, "nous-research"), (re.compile(r"^(llama[- ]?\d[\d.]*[- ].*nemotron|nemotron|nvidia-)"), None, "nvidia"), (re.compile(r"^(meta-)?llama-\d"), None, "meta-ai"), (re.compile(r"^(qwen|qwq|qvq)"), None, "qwen"), (re.compile(r"^kimi-"), None, "moonshot"), (re.compile(r"^(glm-|chatglm)"), None, "zhipu"), (re.compile(r"^minimax-"), None, "minimax"), (re.compile(r"^nova-"), None, "amazon"), (re.compile(r"^phi-"), None, "microsoft"), (re.compile(r"^sonar"), None, "perplexity"), (re.compile(r"^jamba"), None, "ai21"), (re.compile(r"^granite-"), None, "ibm"), (re.compile(r"^hunyuan"), None, "tencent"), (re.compile(r"^ernie"), None, "baidu"), (re.compile(r"^(doubao|seed-)"), None, "bytedance"), (re.compile(r"^lfm"), None, "liquid-ai"), (re.compile(r"^step-?\d"), None, "stepfun"), (re.compile(r"^solar-"), None, "upstage"), (re.compile(r"^(kat-|kwaipilot)"), None, "kwaipilot"), (re.compile(r"^dbrx"), None, "databricks"), (re.compile(r"^(olmo|molmo|tulu)"), None, "allenai"), (re.compile(r"^(smollm|smolvlm)"), None, "huggingface"), ] _FIREWORKS_PATH = re.compile(r"^accounts/fireworks/models/") _DATE_IN_ID = re.compile(r"(20\d{2}-?(0[1-9]|1[0-2])-?(0[1-9]|[12]\d|3[01]))|(?x-?high|extra-high|high|medium|low|minimal|max)(?:\s+effort)?(?:\s+with\s+fallback)?$", re.IGNORECASE) _LABEL_BUDGET = re.compile(r"^(?P\d+)k\s*(think(ing)?|reasoning)?(\s*tokens)?$", re.IGNORECASE) _LABEL_OFF = re.compile(r"^(non-?reasoning|no-?\s?think(ing)?|thinking\s*off|non-?thinking|without\s+thinking)$", re.IGNORECASE) _LABEL_ON = re.compile(r"^(thinking|reasoning|think|thinking\s*on|reasoner)$", re.IGNORECASE) _LABEL_DEFAULT = re.compile(r"^(default(\s+think(ing)?)?)$", re.IGNORECASE) _LABEL_ADAPTIVE = re.compile(r"^(adaptive(\s+(thinking|reasoning))?|thinking\s+auto|auto)$", re.IGNORECASE) _PAREN = re.compile(r"\s*\(([^()]*)\)\s*$") @dataclass class ModelIdentity: raw: str base_id: str # id without provider prefix, variant suffix and effort suffixes scheme: str | None = None # identifier scheme the id belongs to (when recognised) identifiers: dict[str, str] = field(default_factory=dict) aliases: list[str] = field(default_factory=list) org_key: str | None = None # registry organisation key of the developer effort: dict[str, str] = field(default_factory=dict) provider_prefix: str | None = None pinned: bool = False # the id names a dated snapshot (stable), not a rolling alias @property def is_effort_variant(self) -> bool: return bool(self.effort) # ---------------------------------------------------------------------------------------------- effort labels / suffixes def effort_from_label(label: str | None) -> dict[str, str] | None: """'(xhigh)' → {reasoning_effort: xhigh} · '(Non-reasoning)' → {reasoning: off} · '(32k thinking tokens)' → {reasoning: on, thinking_budget: 32k} · '(default think)' → {thinking_budget: default}. None when the label is not an evaluation setting.""" if not label: return None s = label.strip().strip("()").strip().replace("_", "-") if not s: return None m = _LABEL_EFFORT.match(s) if m: eff = m.group("eff").lower().replace("x-high", "xhigh").replace("extra-high", "xhigh") out = {"reasoning_effort": eff} if "fallback" in s.lower(): out["effort_fallback"] = "on" return out m = _LABEL_BUDGET.match(s) if m: return {"reasoning": "on", "thinking_budget": f"{m.group('k')}k"} if _LABEL_OFF.match(s): return {"reasoning": "off"} if _LABEL_ON.match(s): return {"reasoning": "on"} if _LABEL_ADAPTIVE.match(s): return {"reasoning": "adaptive"} if _LABEL_DEFAULT.match(s): return {"thinking_budget": "default"} return None def split_effort_label(name: str | None) -> tuple[str, dict[str, str]]: """'Claude Opus 5 (xhigh)' → ('Claude Opus 5', {reasoning_effort: xhigh}); 'GPT-5.6 Sol xHigh Effort' → ('GPT-5.6 Sol', {reasoning_effort: xhigh}); 'Mistral Small 4 Non-reasoning' → reasoning off. Labels that are not settings ("(0324)", "(Jul)", "Thinking" alone) stay in the name.""" if not name: return "", {} n = name.strip() m = _PAREN.search(n) if m: eff = effort_from_label(m.group(1)) if eff is not None: return n[: m.start()].strip(), eff m = _TRAILING_EFFORT.search(n) if m: eff = m.group("eff").lower().replace("x-high", "xhigh").replace("extra-high", "xhigh") return n[: m.start()].strip(), {"reasoning_effort": eff} m = _TRAILING_OFF.search(n) if m: return n[: m.start()].strip(), {"reasoning": "off"} return n, {} # Suffixes that also name size tiers or official products ("mistral-medium", "sonar-reasoning"): they count as an evaluator setting only # when the remaining stem still carries a version/size digit ("o3-mini-high", "deepseek-v3-1-reasoning"). ("max", "fast", "instant" are # not effort suffixes at all in the ontology — Qwen3-Max, Grok 4.1 Fast and Claude Instant are model tiers.) _AMBIGUOUS_SUFFIXES = {"medium", "low", "high", "reasoning", "think"} _TRAILING_EFFORT = re.compile(r"\s+(?Px-?high|extra-high|high|medium|low|minimal|max)\s+effort$", re.IGNORECASE) _TRAILING_OFF = re.compile(r"\s+(non-?reasoning|no-?think(ing)?|non-?thinking)$", re.IGNORECASE) def strip_effort(model_id: str) -> tuple[str, dict[str, str]]: """Remove every trailing evaluator effort suffix of an id ('…-thinking-64k-high-effort' → '…', {reasoning: on, thinking_budget: 64k, reasoning_effort: high}). Official '-thinking' releases (ontology list) are left alone.""" low = re.sub(r"[\s_]+", "-", model_id.strip().lower()) effort: dict[str, str] = {} for _ in range(4): if low in OFFICIAL_THINKING_RELEASES: break a = analyze_model_name(low) if a.is_effort_variant and a.effort_suffix: stem = low[: -(len(a.effort_suffix) + 1)] if a.effort_suffix in _AMBIGUOUS_SUFFIXES and not re.search(r"\d", stem): break effort = {**a.effort, **effort} low = stem continue hit = next((s for s in sorted(_EXTRA_SUFFIXES, key=len, reverse=True) if low.endswith("-" + s)), None) if hit: effort = {**_EXTRA_SUFFIXES[hit], **effort} low = low[: -(len(hit) + 1)] continue break # keep the original casing of the surviving prefix base = model_id.strip()[: len(low)] if re.sub(r"[\s_]+", "-", model_id.strip().lower()).startswith(low) else low return base.rstrip("-_ "), effort def strip_effort_words(display: str, *, only_if: bool = True) -> str: """Trailing effort words of a display name ('Claude 4.6 Opus Thinking High Effort' → 'Claude 4.6 Opus'). Applied only when the caller knows (from the id) that the row is an effort variant — 'Kimi K2 Thinking' and 'Sonar Reasoning' are real releases.""" if not only_if: return display.strip() words = display.strip().split() while len(words) > 1 and _EFFORT_WORDS.match(words[-1]): words.pop() return " ".join(words) def is_pinned(model_id: str) -> bool: return bool(_DATE_IN_ID.search(model_id)) # ---------------------------------------------------------------------------------------------- identity def org_key_for_vendor(vendor: str | None) -> str | None: if not vendor: return None v = vendor.strip().lower() key = VENDOR_ORG.get(v) if key and key in organizations(): return key known = org_by_hf(v) if known: return known["key"] if v in organizations(): return v return None def org_key_for_bare_id(model_id: str) -> tuple[str | None, str | None]: low = model_id.strip().lower() for pat, scheme, org in _BARE: if pat.search(low): return scheme, (org if org in organizations() else None) return None, None def model_identity(api_id: str | None, *, trusted: bool = False) -> ModelIdentity | None: """Analyse an API id. `trusted=True` when the string is the vendor's own API id (adds the vendor identifier when recognised).""" if not api_id or not isinstance(api_id, str): return None raw = api_id.strip() if not raw or " " in raw: return None ident = ModelIdentity(raw=raw, base_id=raw) rest = raw prefix, _, tail = raw.partition("/") plow = prefix.lower() scheme: str | None = None org_key: str | None = None if tail and plow in PROVIDER_PREFIXES: ident.provider_prefix = plow scheme, org_key = PROVIDER_PREFIXES[plow] rest = tail if plow == "openrouter": vendor, _, slug = rest.partition("/") if slug: base_slug = slug.split(":", 1)[0] ident.aliases += [f"{vendor}/{base_slug}", base_slug] if trusted: ident.identifiers["openrouter"] = f"{vendor}/{base_slug}" org_key = org_key_for_vendor(vendor) rest = base_slug elif plow == "fireworks_ai": rest = _FIREWORKS_PATH.sub("", rest) if trusted and "/" not in rest: ident.identifiers["fireworks_model_id"] = f"fireworks/{rest}" scheme = None elif plow in ("nvidia_nim",): vendor, _, slug = rest.partition("/") if slug: org_key = org_key_for_vendor(vendor) rest = slug elif plow == "openai": # `openai/Qwen/Qwen2.5-Coder-32B-Instruct` (OpenAI-compatible endpoint): the remainder decides bare_scheme, bare_org = org_key_for_bare_id(rest.split("/")[-1]) if "/" in rest: hf_org = rest.split("/")[0] known = org_by_hf(hf_org) if known: org_key = known["key"] if trusted: ident.identifiers["hf_repo"] = rest ident.aliases.append(rest) rest = rest.split("/")[-1] else: scheme, org_key = bare_scheme, bare_org elif tail and "/" in raw and org_by_hf(prefix): # `Qwen/Qwen2.5-Coder-32B-Instruct` — a Hugging Face repository id org_key = org_by_hf(prefix)["key"] # type: ignore[index] if trusted: ident.identifiers["hf_repo"] = raw ident.aliases.append(raw) rest = tail if org_key is None: bare_scheme, bare_org = org_key_for_bare_id(rest) scheme = scheme or bare_scheme org_key = bare_org base, effort = strip_effort(rest) ident.base_id = base or rest ident.effort = effort ident.scheme = scheme ident.org_key = org_key if org_key in organizations() else None ident.pinned = is_pinned(ident.base_id) if scheme and trusted and "/" not in ident.base_id and not ident.identifiers: ident.identifiers[scheme] = ident.base_id for a in (raw, rest, ident.base_id): if a and a not in ident.aliases: ident.aliases.append(a) return ident def family_ref(name: str | None, org: EntityRef | None) -> EntityRef | None: """`model_family` hint for a model name ("Claude Opus 5" → Claude; "Qwen3.6-35B-A3B" → Qwen3.6). None when no family is detected.""" if not name: return None # size tokens first ("Qwen3-0.6B" would otherwise read as version 3.0): 0.6B / 35B-A3B / 17B-128E bare = re.sub(r"[-_ ]?\d+(?:\.\d+)?\s?[bmt](?:[-_]?a\d+(?:\.\d+)?[bmt])?(?:[-_]?\d+e)?(?![a-z0-9])", "", name.split("/")[-1], flags=re.IGNORECASE) label = family_release_hint(bare or name) if not label: return None return EntityRef(entity_type="model_family", name=label, organization=org, identity_confidence="medium", identifiers={"family_key": re.sub(r"[^a-z0-9.]+", "-", label.lower()).strip("-") + (f"@{org.slug_hint}" if org and org.slug_hint else "")}) def org_ref_in(facts: Facts, key: str | None) -> EntityRef | None: """Registry organisation appended once to `facts.entities`.""" if not key or key not in organizations(): return None for e in facts.entities: if e.identifiers.get("registry_org") == key: return e ref = org_ref(key) facts.entities.append(ref) return ref def model_ref_from_api_id(facts: Facts, api_id: str | None, *, name: str | None = None, trusted: bool = False, identity_confidence: str = "medium", organization: EntityRef | None = None, extra_aliases: list[str] | None = None, family: bool = True, attributes: dict[str, Any] | None = None) -> tuple[EntityRef, dict[str, str]]: """Model EntityRef for a third-party API id (deduplicated inside `facts`), plus the effort configuration stripped from the id/label. `name` is the source's display label (effort parentheticals are folded into the configuration); without it the base id is the name.""" ident = model_identity(api_id, trusted=trusted) label, label_effort = split_effort_label(name) if name else ("", {}) effort = {**(ident.effort if ident else {}), **label_effort} display = label or (ident.base_id if ident else (api_id or "").strip()) if ident and ident.is_effort_variant and label and not label_effort: display = strip_effort_words(label, only_if=True) # the first-party pattern org (what the lab connectors use) wins over the board's own creator label, so that the resolver # disambiguates aliases with the same organisation the official source attached org = (org_ref_in(facts, ident.org_key) if ident else None) or organization identifiers = dict(ident.identifiers) if ident else {} aliases = [a for a in dict.fromkeys([*(ident.aliases if ident else []), *(extra_aliases or [])]) if a and a != display] # one ref per identity inside the document: same identifier, or same display name (a board's rows for one model share its label) for e in facts.entities: if e.entity_type != "model": continue if (identifiers and any(e.identifiers.get(k) == v for k, v in identifiers.items())) or e.name.lower() == display.lower(): for k, v in identifiers.items(): e.identifiers.setdefault(k, v) for a in aliases: if a not in e.aliases and a != e.name: e.aliases.append(a) if e.organization is None and org is not None: e.organization = org if e.family is None and family: e.family = family_ref(display, org) return e, effort ref = facts.entity("model", display[:200], identifiers=identifiers, aliases=aliases, organization=org, identity_confidence=identity_confidence, family=family_ref(display, org) if family else None, attributes=dict(attributes or {})) return ref, effort __all__ = ["PROVIDER_PREFIXES", "VENDOR_ORG", "ModelIdentity", "effort_from_label", "family_ref", "is_pinned", "model_identity", "model_ref_from_api_id", "org_key_for_bare_id", "org_key_for_vendor", "org_ref_in", "split_effort_label", "strip_effort", "strip_effort_words"]