به‌روزرسانی: دریافت زندهٔ کاتالوگ مدل‌ها و افورت‌های استدلال پشتیبانی‌شده بر اساس اکانت انتخابی

apps/api/app/infrastructure/adapters/agy.py

@@ -37,6 +37,7 @@ import uuid
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
+from typing import Any
 from urllib.parse import parse_qs, urlencode, urlparse
 
 import httpx
@@ -209,6 +210,18 @@ class AgyLoginResult:
     refresh_token: str
 
 
+@dataclass(slots=True, kw_only=True)
+class AgyAccountModelInfo:
+    """Model catalogue item as reported by an agy account."""
+
+    id: str
+    display_name: str
+    supports_thinking: bool = False
+    supported_efforts: list[str] = field(default_factory=list)
+    default_effort: str | None = None
+    supports_images: bool = False
+
+
 class AgyOAuthError(ValidationError):
     """The pasted URL or the token exchange was rejected by Google."""
 
@@ -630,15 +643,18 @@ class AgyModelAdapter:
         accounts = [a for a in (_account_of(c) for c in await self._load()) if a.refresh_token]
         if not accounts:
             return []
-        return await self._fetch_models(accounts[0])
+        items = await self._fetch_models(accounts[0])
+        return [m.id for m in items]
 
-    async def available_models_for(self, credential: ModelCredential) -> list[str]:
+    async def available_models_for(
+        self, credential: ModelCredential
+    ) -> list[AgyAccountModelInfo]:
         """Live model catalog of exactly one account.
 
         Called per account selection in the admin console, so the list is
         fetched fresh from Google every time (the access token is cached, the
         ``fetchAvailableModels`` round-trip is not) and reflects what *this*
-        subscription is entitled to.
+        subscription is entitled to along with each model's supported efforts.
         """
         account = _account_of(credential)
         if not account.refresh_token:
@@ -647,7 +663,7 @@ class AgyModelAdapter:
             )
         return await self._fetch_models(account)
 
-    async def _fetch_models(self, account: AgyAccount) -> list[str]:
+    async def _fetch_models(self, account: AgyAccount) -> list[AgyAccountModelInfo]:
         token = await self._access_token(account)
         for base in BASE_URLS:
             try:
@@ -663,11 +679,87 @@ class AgyModelAdapter:
                     )
                 if response.is_success:
                     models = response.json().get("models") or {}
-                    return sorted(models)
+                    return self._parse_model_catalog(models)
             except httpx.HTTPError:
                 continue
         return []
 
+    @classmethod
+    def _parse_model_catalog(
+        cls, raw_models: dict[str, Any]
+    ) -> list[AgyAccountModelInfo]:
+        items: list[AgyAccountModelInfo] = []
+        for model_id, details in sorted(raw_models.items()):
+            if not isinstance(details, dict):
+                details = {}
+            # Omit internal non-generative / tab-completion preview models
+            if details.get("isInternal") and not details.get("displayName"):
+                continue
+            if model_id.startswith("tab_") or model_id.startswith("chat_"):
+                continue
+
+            display_name = str(details.get("displayName") or model_id)
+            supports_images = bool(details.get("supportsImages"))
+            supports_thinking = bool(details.get("supportsThinking"))
+            if not supports_thinking and _supports_thinking(model_id):
+                supports_thinking = True
+
+            supported_efforts = cls._supported_efforts_for(model_id, details)
+            default_effort = cls._default_effort_for(supported_efforts)
+
+            items.append(
+                AgyAccountModelInfo(
+                    id=model_id,
+                    display_name=display_name,
+                    supports_thinking=supports_thinking,
+                    supported_efforts=supported_efforts,
+                    default_effort=default_effort,
+                    supports_images=supports_images,
+                )
+            )
+        return items
+
+    @classmethod
+    def _supported_efforts_for(
+        cls, model_id: str, details: dict[str, Any]
+    ) -> list[str]:
+        lowered = model_id.lower()
+        if "image" in lowered:
+            return []
+        if "claude" in lowered:
+            return []
+
+        supports_thinking = details.get("supportsThinking")
+        if supports_thinking is False:
+            return []
+        if supports_thinking is None and not _supports_thinking(model_id):
+            return []
+
+        if "gpt" in lowered:
+            return ["medium"]
+        if "pro" in lowered and "3.1" in lowered:
+            return ["low", "high"]
+
+        budget = details.get("thinkingBudget")
+        if isinstance(budget, int) and budget > 0:
+            if budget <= 4096:
+                return ["low"]
+            if budget <= 8192:
+                return ["medium"]
+            return ["high"]
+
+        return ["low", "medium", "high"]
+
+    @staticmethod
+    def _default_effort_for(supported: list[str]) -> str | None:
+        if not supported:
+            return None
+        if "high" in supported:
+            return "high"
+        if "medium" in supported:
+            return "medium"
+        return supported[0]
+
     # -- internals ----------------------------------------------------------
 
     async def _load(self) -> list[ModelCredential]: