Add venice/info.py

venice/info.py

@@ -0,0 +1,669 @@
+"""
+title: Venice.ai Info
+author: Jeff Smith
+version: 1.1.0
+license: MIT
+required_open_webui_version: 0.6.0
+requirements: httpx, pydantic
+description: |
+    Venice.ai reference utility - account status and model discovery.
+    - Check DIEM balance and rate limits
+    - List available models by type
+    - Look up model capabilities and pricing
+    - List image style presets
+    - List model traits (semantic mappings)
+    - List compatibility mappings (external model aliases)
+
+    This is a read-only info tool. Use venice_image or venice_chat for actions.
+"""
+
+from typing import Callable, Any
+from pydantic import BaseModel, Field
+import httpx
+
+
+class Tools:
+    """
+    Venice.ai information and reference tool.
+    Check account status, discover models, and look up capabilities.
+    All methods are read-only and don't consume DIEM.
+    """
+
+    class Valves(BaseModel):
+        """Admin configuration."""
+
+        VENICE_API_KEY: str = Field(
+            default="", description="Venice.ai API key (admin default)"
+        )
+        DIEM_WARNING_THRESHOLD: float = Field(
+            default=1.0,
+            description="DIEM balance below this triggers low balance warning",
+        )
+        DAILY_DIEM_ALLOCATION: float = Field(
+            default=8.10,
+            description="Expected daily DIEM allocation (for usage calculation)",
+        )
+        TIMEOUT: int = Field(default=30, description="API request timeout in seconds")
+
+    class UserValves(BaseModel):
+        """Per-user configuration."""
+
+        VENICE_API_KEY: str = Field(
+            default="", description="Your Venice.ai API key (overrides admin default)"
+        )
+
+    def __init__(self):
+        self.valves = self.Valves()
+        self.user_valves = self.UserValves()
+        self.citation = False
+
+    def _get_api_key(self) -> str:
+        """Get Venice API key with UserValves priority."""
+        return self.user_valves.VENICE_API_KEY or self.valves.VENICE_API_KEY
+
+    # ==================== Balance Methods ====================
+
+    async def check_balance(
+        self,
+        show_rate_limits: bool = False,
+        __user__: dict = None,
+        __event_emitter__: Callable[[dict], Any] = None,
+    ) -> str:
+        """
+        Check Venice.ai account balance and optionally rate limits.
+        DIEM is Venice's credit system (1 DIEM ≈ $1 USD).
+        Balance resets daily at 00:00 UTC.
+
+        :param show_rate_limits: Include rate limits in output (default False)
+        :return: Account tier, DIEM balance, usage stats
+        """
+        api_key = self._get_api_key()
+        if not api_key:
+            return "Check Balance\nStatus: 0\nError: API key not configured. Set in UserValves or ask admin."
+
+        if __event_emitter__:
+            await __event_emitter__(
+                {
+                    "type": "status",
+                    "data": {"description": "Checking balance...", "done": False},
+                }
+            )
+
+        try:
+            async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client:
+                response = await client.get(
+                    "https://api.venice.ai/api/v1/api_keys/rate_limits",
+                    headers={"Authorization": f"Bearer {api_key}"},
+                )
+                response.raise_for_status()
+                result = response.json()
+
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+
+            data = result.get("data", {})
+            balances = data.get("balances", {})
+            tier = data.get("apiTier", {}).get("id", "unknown")
+            next_epoch = data.get("nextEpochBegins", "unknown")
+
+            diem = balances.get("DIEM", 0)
+            usd = balances.get("USD", 0)
+
+            # Calculate usage
+            daily = self.valves.DAILY_DIEM_ALLOCATION
+            used = max(0, daily - diem)
+            usage_pct = (used / daily * 100) if daily > 0 else 0
+
+            # Status
+            threshold = self.valves.DIEM_WARNING_THRESHOLD
+            if diem < threshold:
+                status = f"⚠️ LOW (below {threshold} DIEM)"
+            elif diem < daily * 0.25:
+                status = "⚡ Getting low"
+            else:
+                status = "✓ OK"
+
+            lines = [
+                "Check Balance",
+                "Status: 200",
+                "",
+                f"Tier: {tier}",
+                f"Balance: {diem:.2f} DIEM (≈ ${diem:.2f} USD) {status}",
+                f"Used today: {used:.2f} DIEM of {daily:.2f} DIEM ({usage_pct:.0f}%)",
+                f"Resets: {next_epoch}",
+                "",
+                "Note: 1 DIEM = $1 USD. Model prices in USD deduct equivalent DIEM.",
+            ]
+
+            if usd < 0:
+                lines.append(f"USD Overage: ${usd:.4f}")
+
+            if show_rate_limits:
+                rate_limits = data.get("rateLimits", [])
+                with_limits = []
+                for limit in rate_limits:
+                    model_id = limit.get("apiModelId", "")
+                    limits = limit.get("rateLimits", [])
+                    if not limits:
+                        continue
+                    parts = []
+                    for rl in limits:
+                        t, a = rl.get("type", ""), rl.get("amount", 0)
+                        if t == "RPM":
+                            parts.append(f"{a} RPM")
+                        elif t == "TPM":
+                            parts.append(f"{a//1000}K TPM")
+                    if parts:
+                        with_limits.append(f"  {model_id}: {', '.join(parts)}")
+
+                if with_limits:
+                    lines.append("")
+                    lines.append(f"Rate Limits ({len(with_limits)} models):")
+                    lines.extend(sorted(with_limits))
+
+            return "\n".join(lines)
+
+        except httpx.HTTPStatusError as e:
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+            return f"Check Balance\nStatus: {e.response.status_code}\nError: {e.response.text[:200]}"
+        except Exception as e:
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+            return f"Check Balance\nStatus: 0\nError: {type(e).__name__}: {e}"
+
+    # ==================== Model Methods ====================
+
+    async def list_models(
+        self,
+        model_type: str = "image",
+        __user__: dict = None,
+        __event_emitter__: Callable[[dict], Any] = None,
+    ) -> str:
+        """
+        List available Venice.ai models by type.
+
+        :param model_type: Type of models: image, text, video, embedding, tts (default: image)
+        :return: Models with pricing and capabilities
+        """
+        valid_types = ["image", "text", "video", "embedding", "tts"]
+        if model_type not in valid_types:
+            return f"List Models\nStatus: 0\nError: Invalid type '{model_type}'. Valid: {', '.join(valid_types)}"
+
+        api_key = self._get_api_key()
+        if not api_key:
+            return "List Models\nStatus: 0\nError: API key not configured."
+
+        if __event_emitter__:
+            await __event_emitter__(
+                {
+                    "type": "status",
+                    "data": {
+                        "description": f"Fetching {model_type} models...",
+                        "done": False,
+                    },
+                }
+            )
+
+        try:
+            async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client:
+                response = await client.get(
+                    f"https://api.venice.ai/api/v1/models?type={model_type}",
+                    headers={"Authorization": f"Bearer {api_key}"},
+                )
+                response.raise_for_status()
+                result = response.json()
+
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+
+            models = result.get("data", [])
+            if not models:
+                return f"List Models ({model_type})\nStatus: 200\nResult: No models available"
+
+            lines = [
+                f"List Models ({model_type})",
+                "Status: 200",
+                "",
+                "Prices shown deduct from DIEM balance (1 DIEM = $1 USD)",
+                "",
+            ]
+
+            for model in models:
+                mid = model.get("id", "unknown")
+                spec = model.get("model_spec", {})
+                name = spec.get("name", mid)
+                offline = spec.get("offline", False)
+                beta = spec.get("betaModel", False)
+
+                # Pricing (shown as DIEM but stored as USD, 1:1)
+                pricing = spec.get("pricing", {})
+                if model_type == "image":
+                    if "generation" in pricing:
+                        p = pricing.get("generation", {}).get("usd", 0)
+                        price = f"{p:.3f} DIEM/img"
+                    elif "resolutions" in pricing:
+                        # Resolution-based pricing (e.g., nano-banana-pro)
+                        res_pricing = pricing.get("resolutions", {})
+                        res_parts = [
+                            f"{res}:{p.get('usd', 0):.2f}"
+                            for res, p in res_pricing.items()
+                        ]
+                        price = f"{' | '.join(res_parts)} DIEM"
+                    else:
+                        price = ""
+                elif model_type == "text":
+                    i = pricing.get("input", {}).get("usd", 0)
+                    o = pricing.get("output", {}).get("usd", 0)
+                    price = f"{i:.4f}/{o:.4f} DIEM/1M"
+                elif model_type == "video":
+                    p = pricing.get("generation", {}).get("usd", 0)
+                    price = f"{p:.2f} DIEM/vid"
+                else:
+                    price = ""
+
+                # Capabilities (text models)
+                caps = spec.get("capabilities", {})
+                cap_list = []
+                if caps.get("supportsVision"):
+                    cap_list.append("vision")
+                if caps.get("supportsFunctionCalling"):
+                    cap_list.append("tools")
+                if caps.get("supportsReasoning"):
+                    cap_list.append("reasoning")
+                if caps.get("supportsWebSearch"):
+                    cap_list.append("web")
+
+                # Web search for image models (e.g., nano-banana-pro)
+                if model_type == "image" and spec.get("supportsWebSearch"):
+                    cap_list.append("web")
+
+                # Build line
+                parts = [f"  {mid}"]
+                if name != mid:
+                    parts.append(f"({name})")
+                if price:
+                    parts.append(price)
+                if cap_list:
+                    parts.append(f"[{', '.join(cap_list)}]")
+                if beta:
+                    parts.append("BETA")
+                if offline:
+                    parts.append("OFFLINE")
+
+                lines.append(" ".join(parts))
+
+            lines.append("")
+            lines.append(f"Total: {len(models)} {model_type} models")
+
+            return "\n".join(lines)
+
+        except httpx.HTTPStatusError as e:
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+            return f"List Models\nStatus: {e.response.status_code}\nError: {e.response.text[:200]}"
+        except Exception as e:
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+            return f"List Models\nStatus: 0\nError: {type(e).__name__}: {e}"
+
+    async def list_styles(
+        self,
+        __user__: dict = None,
+        __event_emitter__: Callable[[dict], Any] = None,
+    ) -> str:
+        """
+        List available image style presets.
+
+        :return: Style names for use with image generation
+        """
+        api_key = self._get_api_key()
+        if not api_key:
+            return "List Styles\nStatus: 0\nError: API key not configured."
+
+        if __event_emitter__:
+            await __event_emitter__(
+                {
+                    "type": "status",
+                    "data": {"description": "Fetching styles...", "done": False},
+                }
+            )
+
+        try:
+            async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client:
+                response = await client.get(
+                    "https://api.venice.ai/api/v1/image/styles",
+                    headers={"Authorization": f"Bearer {api_key}"},
+                )
+                response.raise_for_status()
+                result = response.json()
+
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+
+            styles = result.get("data", [])
+            lines = [
+                "List Styles",
+                "Status: 200",
+                "",
+                "Image Style Presets:",
+            ]
+
+            for style in sorted(styles):
+                lines.append(f"  - {style}")
+
+            lines.append("")
+            lines.append(f"Total: {len(styles)} styles")
+            lines.append("Usage: Pass as 'style_preset' to image generation")
+
+            return "\n".join(lines)
+
+        except httpx.HTTPStatusError as e:
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+            return f"List Styles\nStatus: {e.response.status_code}\nError: {e.response.text[:200]}"
+        except Exception as e:
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+            return f"List Styles\nStatus: 0\nError: {type(e).__name__}: {e}"
+
+    async def list_traits(
+        self,
+        __user__: dict = None,
+        __event_emitter__: Callable[[dict], Any] = None,
+    ) -> str:
+        """
+        List Venice.ai model traits - semantic mappings to recommended models.
+        Traits allow requesting models by capability (e.g., "fastest", "default_code")
+        rather than hardcoding specific model IDs.
+
+        :return: Trait names and their associated models
+        """
+        api_key = self._get_api_key()
+        if not api_key:
+            return "List Traits\nStatus: 0\nError: API key not configured."
+
+        if __event_emitter__:
+            await __event_emitter__(
+                {
+                    "type": "status",
+                    "data": {"description": "Fetching traits...", "done": False},
+                }
+            )
+
+        try:
+            async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client:
+                response = await client.get(
+                    "https://api.venice.ai/api/v1/models/traits",
+                    headers={"Authorization": f"Bearer {api_key}"},
+                )
+                response.raise_for_status()
+                result = response.json()
+
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+
+            traits = result.get("data", {})
+            lines = [
+                "List Traits",
+                "Status: 200",
+                "",
+                "Model Traits (semantic mappings to recommended models):",
+                "",
+            ]
+
+            # Sort traits for consistent output
+            for trait_name in sorted(traits.keys()):
+                model_id = traits[trait_name]
+                lines.append(f"  {trait_name}: {model_id}")
+
+            lines.append("")
+            lines.append(f"Total: {len(traits)} traits")
+            lines.append("")
+            lines.append(
+                "Usage: Request models by trait for automatic best-model selection."
+            )
+
+            return "\n".join(lines)
+
+        except httpx.HTTPStatusError as e:
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+            return f"List Traits\nStatus: {e.response.status_code}\nError: {e.response.text[:200]}"
+        except Exception as e:
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+            return f"List Traits\nStatus: 0\nError: {type(e).__name__}: {e}"
+
+    async def list_compatibility(
+        self,
+        __user__: dict = None,
+        __event_emitter__: Callable[[dict], Any] = None,
+    ) -> str:
+        """
+        List Venice.ai compatibility mappings for external model names.
+        Maps common model names (OpenAI, Anthropic, etc.) to Venice equivalents
+        for drop-in API compatibility.
+
+        :return: External model names grouped by their Venice equivalent
+        """
+        api_key = self._get_api_key()
+        if not api_key:
+            return "List Compatibility\nStatus: 0\nError: API key not configured."
+
+        if __event_emitter__:
+            await __event_emitter__(
+                {
+                    "type": "status",
+                    "data": {
+                        "description": "Fetching compatibility mappings...",
+                        "done": False,
+                    },
+                }
+            )
+
+        try:
+            async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client:
+                response = await client.get(
+                    "https://api.venice.ai/api/v1/models/compatibility_mapping",
+                    headers={"Authorization": f"Bearer {api_key}"},
+                )
+                response.raise_for_status()
+                result = response.json()
+
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+
+            mappings = result.get("data", {})
+            lines = [
+                "List Compatibility Mappings",
+                "Status: 200",
+                "",
+                "External model names mapped to Venice equivalents:",
+                "",
+            ]
+
+            # Group by target model for cleaner output
+            by_target: dict[str, list[str]] = {}
+            for external_name, venice_model in mappings.items():
+                if venice_model not in by_target:
+                    by_target[venice_model] = []
+                by_target[venice_model].append(external_name)
+
+            for venice_model in sorted(by_target.keys()):
+                external_names = sorted(by_target[venice_model])
+                lines.append(f"  {venice_model}:")
+                for ext_name in external_names:
+                    lines.append(f"    <- {ext_name}")
+                lines.append("")
+
+            lines.append(
+                f"Total: {len(mappings)} mappings to {len(by_target)} Venice models"
+            )
+            lines.append("")
+            lines.append(
+                "Usage: Use external model names (gpt-4o, claude-3-5-sonnet, etc.) for compatibility."
+            )
+
+            return "\n".join(lines)
+
+        except httpx.HTTPStatusError as e:
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+            return f"List Compatibility\nStatus: {e.response.status_code}\nError: {e.response.text[:200]}"
+        except Exception as e:
+            if __event_emitter__:
+                await __event_emitter__({"type": "status", "data": {"done": True}})
+            return f"List Compatibility\nStatus: 0\nError: {type(e).__name__}: {e}"
+
+    async def get_model_info(
+        self,
+        model_id: str,
+        __user__: dict = None,
+        __event_emitter__: Callable[[dict], Any] = None,
+    ) -> str:
+        """
+        Get detailed information about a specific model.
+
+        :param model_id: The model ID to look up
+        :return: Capabilities, pricing, constraints, and status
+        """
+        api_key = self._get_api_key()
+        if not api_key:
+            return "Get Model Info\nStatus: 0\nError: API key not configured."
+
+        if not model_id:
+            return "Get Model Info\nStatus: 0\nError: model_id required"
+
+        if __event_emitter__:
+            await __event_emitter__(
+                {
+                    "type": "status",
+                    "data": {"description": f"Looking up {model_id}...", "done": False},
+                }
+            )
+
+        # Search each type
+        for model_type in ["text", "image", "video", "embedding", "tts"]:
+            try:
+                async with httpx.AsyncClient(
+                    timeout=float(self.valves.TIMEOUT)
+                ) as client:
+                    response = await client.get(
+                        f"https://api.venice.ai/api/v1/models?type={model_type}",
+                        headers={"Authorization": f"Bearer {api_key}"},
+                    )
+                    response.raise_for_status()
+                    result = response.json()
+
+                    for model in result.get("data", []):
+                        if model.get("id") == model_id:
+                            if __event_emitter__:
+                                await __event_emitter__(
+                                    {"type": "status", "data": {"done": True}}
+                                )
+
+                            spec = model.get("model_spec", {})
+                            lines = [
+                                f"Get Model Info ({model_id})",
+                                "Status: 200",
+                                "",
+                                f"Name: {spec.get('name', model_id)}",
+                                f"Type: {model_type}",
+                                f"Privacy: {spec.get('privacy', 'unknown')}",
+                                f"Offline: {spec.get('offline', False)}",
+                                f"Beta: {spec.get('betaModel', False)}",
+                            ]
+
+                            desc = spec.get("description")
+                            if desc:
+                                lines.append(f"Description: {desc}")
+
+                            ctx = spec.get("availableContextTokens")
+                            if ctx:
+                                lines.append(f"Context: {ctx:,} tokens")
+
+                            # Traits
+                            traits = spec.get("traits", [])
+                            if traits:
+                                lines.append(f"Traits: {', '.join(traits)}")
+
+                            # Constraints (for image models)
+                            constraints = spec.get("constraints", {})
+                            if constraints:
+                                lines.append("")
+                                lines.append("Constraints:")
+                                if "promptCharacterLimit" in constraints:
+                                    lines.append(
+                                        f"  Prompt limit: {constraints['promptCharacterLimit']:,} chars"
+                                    )
+                                if "steps" in constraints:
+                                    steps = constraints["steps"]
+                                    lines.append(
+                                        f"  Steps: default={steps.get('default')}, max={steps.get('max')}"
+                                    )
+                                if "widthHeightDivisor" in constraints:
+                                    lines.append(
+                                        f"  Width/Height divisor: {constraints['widthHeightDivisor']}"
+                                    )
+                                if "resolutions" in constraints:
+                                    lines.append(
+                                        f"  Resolutions: {', '.join(constraints['resolutions'])}"
+                                    )
+
+                            # Pricing
+                            pricing = spec.get("pricing", {})
+                            if pricing:
+                                lines.append("")
+                                lines.append("Pricing (1 DIEM = $1 USD):")
+                                if "input" in pricing:
+                                    p = pricing["input"].get("usd", 0)
+                                    lines.append(f"  Input: {p:.4f} DIEM/1M tokens")
+                                if "output" in pricing:
+                                    p = pricing["output"].get("usd", 0)
+                                    lines.append(f"  Output: {p:.4f} DIEM/1M tokens")
+                                if "generation" in pricing:
+                                    p = pricing["generation"].get("usd", 0)
+                                    lines.append(f"  Generation: {p:.4f} DIEM")
+                                if "resolutions" in pricing:
+                                    lines.append("  Resolution-based:")
+                                    for res, price in pricing["resolutions"].items():
+                                        lines.append(
+                                            f"    {res}: {price.get('usd', 0):.2f} DIEM"
+                                        )
+                                if "upscale" in pricing:
+                                    lines.append("  Upscale:")
+                                    for scale, price in pricing["upscale"].items():
+                                        lines.append(
+                                            f"    {scale}: {price.get('usd', 0):.2f} DIEM"
+                                        )
+
+                            # Capabilities
+                            caps = spec.get("capabilities", {})
+                            active_caps = [k for k, v in caps.items() if v]
+
+                            # Web search for image models
+                            if model_type == "image" and spec.get("supportsWebSearch"):
+                                active_caps.append("supportsWebSearch")
+
+                            if active_caps:
+                                lines.append("")
+                                lines.append("Capabilities:")
+                                for cap in active_caps:
+                                    lines.append(f"  - {cap}")
+
+                            # Model source
+                            source = spec.get("modelSource")
+                            if source:
+                                lines.append("")
+                                lines.append(f"Source: {source}")
+
+                            return "\n".join(lines)
+
+            except Exception:
+                continue
+
+        if __event_emitter__:
+            await __event_emitter__({"type": "status", "data": {"done": True}})
+
+        return f"Get Model Info ({model_id})\nStatus: 404\nError: Model not found"