v1.2.0: Add VeniceInfo namespace for helper functions

This commit is contained in:
2026-01-15 12:19:06 +00:00
parent fe29e792ab
commit 8a988c1ded

View File

@@ -1,7 +1,7 @@
""" """
title: Venice.ai Info title: Venice.ai Info
author: Jeff Smith author: Jeff Smith
version: 1.1.0 version: 1.2.0
license: MIT license: MIT
required_open_webui_version: 0.6.0 required_open_webui_version: 0.6.0
requirements: httpx, pydantic requirements: httpx, pydantic
@@ -15,6 +15,14 @@ description: |
- List compatibility mappings (external model aliases) - List compatibility mappings (external model aliases)
This is a read-only info tool. Use venice_image or venice_chat for actions. This is a read-only info tool. Use venice_image or venice_chat for actions.
v1.2.0: Added VeniceInfo namespace class for helper functions to avoid
method collisions with Open WebUI framework introspection.
changelog:
1.2.0:
- Added VeniceInfo namespace class for helper functions
- Moved get_api_key to VeniceInfo namespace
- Prevents Open WebUI framework introspection method name collisions
""" """
from typing import Callable, Any from typing import Callable, Any
@@ -22,6 +30,24 @@ from pydantic import BaseModel, Field
import httpx import httpx
class VeniceInfo:
"""
Namespaced helpers for Venice info operations.
Using a separate class prevents Open WebUI framework introspection
from colliding with tool methods that have generic names like _get_api_key.
"""
@staticmethod
def get_api_key(valves, user_valves, __user__: dict = None) -> str:
"""Get API key with UserValves priority."""
if __user__ and "valves" in __user__:
user_valves_dict = __user__.get("valves")
if isinstance(user_valves_dict, dict) and user_valves_dict.get("VENICE_API_KEY"):
return user_valves_dict["VENICE_API_KEY"]
return user_valves.VENICE_API_KEY or valves.VENICE_API_KEY
class Tools: class Tools:
""" """
Venice.ai information and reference tool. Venice.ai information and reference tool.
@@ -32,89 +58,43 @@ class Tools:
class Valves(BaseModel): class Valves(BaseModel):
"""Admin configuration.""" """Admin configuration."""
VENICE_API_KEY: str = Field( VENICE_API_KEY: str = Field(default="", description="Venice.ai API key (admin default)")
default="", description="Venice.ai API key (admin default)" DIEM_WARNING_THRESHOLD: float = Field(default=1.0, description="DIEM balance below this triggers warning")
) DAILY_DIEM_ALLOCATION: float = Field(default=8.10, description="Daily DIEM allocation for usage calc")
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") TIMEOUT: int = Field(default=30, description="API request timeout in seconds")
class UserValves(BaseModel): class UserValves(BaseModel):
"""Per-user configuration.""" """Per-user configuration."""
VENICE_API_KEY: str = Field( VENICE_API_KEY: str = Field(default="", description="Your Venice.ai API key")
default="", description="Your Venice.ai API key (overrides admin default)"
)
def __init__(self): def __init__(self):
self.valves = self.Valves() self.valves = self.Valves()
self.user_valves = self.UserValves() self.user_valves = self.UserValves()
self.citation = False self.citation = False
def _get_api_key(self) -> str: async def check_balance(self, show_rate_limits: bool = False, __user__: dict = None, __event_emitter__: Callable[[dict], Any] = None) -> str:
"""Get Venice API key with UserValves priority.""" api_key = VeniceInfo.get_api_key(self.valves, self.user_valves, __user__)
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: if not api_key:
return "Check Balance\nStatus: 0\nError: API key not configured. Set in UserValves or ask admin." return "Check Balance\nStatus: 0\nError: API key not configured."
if __event_emitter__: if __event_emitter__:
await __event_emitter__( await __event_emitter__({"type": "status", "data": {"description": "Checking balance...", "done": False}})
{
"type": "status",
"data": {"description": "Checking balance...", "done": False},
}
)
try: try:
async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client: async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client:
response = await client.get( response = await client.get("https://api.venice.ai/api/v1/api_keys/rate_limits", headers={"Authorization": f"Bearer {api_key}"})
"https://api.venice.ai/api/v1/api_keys/rate_limits",
headers={"Authorization": f"Bearer {api_key}"},
)
response.raise_for_status() response.raise_for_status()
result = response.json() result = response.json()
if __event_emitter__: if __event_emitter__:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
data = result.get("data", {}) data = result.get("data", {})
balances = data.get("balances", {}) balances = data.get("balances", {})
tier = data.get("apiTier", {}).get("id", "unknown") tier = data.get("apiTier", {}).get("id", "unknown")
next_epoch = data.get("nextEpochBegins", "unknown") next_epoch = data.get("nextEpochBegins", "unknown")
diem = balances.get("DIEM", 0) diem = balances.get("DIEM", 0)
usd = balances.get("USD", 0) usd = balances.get("USD", 0)
# Calculate usage
daily = self.valves.DAILY_DIEM_ALLOCATION daily = self.valves.DAILY_DIEM_ALLOCATION
used = max(0, daily - diem) used = max(0, daily - diem)
usage_pct = (used / daily * 100) if daily > 0 else 0 usage_pct = (used / daily * 100) if daily > 0 else 0
# Status
threshold = self.valves.DIEM_WARNING_THRESHOLD threshold = self.valves.DIEM_WARNING_THRESHOLD
if diem < threshold: if diem < threshold:
status = f"⚠️ LOW (below {threshold} DIEM)" status = f"⚠️ LOW (below {threshold} DIEM)"
@@ -122,22 +102,9 @@ class Tools:
status = "⚡ Getting low" status = "⚡ Getting low"
else: else:
status = "✓ OK" 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."]
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: if usd < 0:
lines.append(f"USD Overage: ${usd:.4f}") lines.append(f"USD Overage: ${usd:.4f}")
if show_rate_limits: if show_rate_limits:
rate_limits = data.get("rateLimits", []) rate_limits = data.get("rateLimits", [])
with_limits = [] with_limits = []
@@ -155,14 +122,11 @@ class Tools:
parts.append(f"{a//1000}K TPM") parts.append(f"{a//1000}K TPM")
if parts: if parts:
with_limits.append(f" {model_id}: {', '.join(parts)}") with_limits.append(f" {model_id}: {', '.join(parts)}")
if with_limits: if with_limits:
lines.append("") lines.append("")
lines.append(f"Rate Limits ({len(with_limits)} models):") lines.append(f"Rate Limits ({len(with_limits)} models):")
lines.extend(sorted(with_limits)) lines.extend(sorted(with_limits))
return "\n".join(lines) return "\n".join(lines)
except httpx.HTTPStatusError as e: except httpx.HTTPStatusError as e:
if __event_emitter__: if __event_emitter__:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
@@ -172,83 +136,40 @@ class Tools:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
return f"Check Balance\nStatus: 0\nError: {type(e).__name__}: {e}" 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:
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"] valid_types = ["image", "text", "video", "embedding", "tts"]
if model_type not in valid_types: if model_type not in valid_types:
return f"List Models\nStatus: 0\nError: Invalid type '{model_type}'. Valid: {', '.join(valid_types)}" return f"List Models\nStatus: 0\nError: Invalid type '{model_type}'. Valid: {', '.join(valid_types)}"
api_key = VeniceInfo.get_api_key(self.valves, self.user_valves, __user__)
api_key = self._get_api_key()
if not api_key: if not api_key:
return "List Models\nStatus: 0\nError: API key not configured." return "List Models\nStatus: 0\nError: API key not configured."
if __event_emitter__: if __event_emitter__:
await __event_emitter__( await __event_emitter__({"type": "status", "data": {"description": f"Fetching {model_type} models...", "done": False}})
{
"type": "status",
"data": {
"description": f"Fetching {model_type} models...",
"done": False,
},
}
)
try: try:
async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client: async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client:
response = await client.get( response = await client.get(f"https://api.venice.ai/api/v1/models?type={model_type}", headers={"Authorization": f"Bearer {api_key}"})
f"https://api.venice.ai/api/v1/models?type={model_type}",
headers={"Authorization": f"Bearer {api_key}"},
)
response.raise_for_status() response.raise_for_status()
result = response.json() result = response.json()
if __event_emitter__: if __event_emitter__:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
models = result.get("data", []) models = result.get("data", [])
if not models: if not models:
return f"List Models ({model_type})\nStatus: 200\nResult: No models available" return f"List Models ({model_type})\nStatus: 200\nResult: No models available"
lines = [f"List Models ({model_type})", "Status: 200", "", "Prices in DIEM (1 DIEM = $1 USD)", ""]
lines = [
f"List Models ({model_type})",
"Status: 200",
"",
"Prices shown deduct from DIEM balance (1 DIEM = $1 USD)",
"",
]
for model in models: for model in models:
mid = model.get("id", "unknown") mid = model.get("id", "unknown")
spec = model.get("model_spec", {}) spec = model.get("model_spec", {})
name = spec.get("name", mid) name = spec.get("name", mid)
offline = spec.get("offline", False) offline = spec.get("offline", False)
beta = spec.get("betaModel", False) beta = spec.get("betaModel", False)
# Pricing (shown as DIEM but stored as USD, 1:1)
pricing = spec.get("pricing", {}) pricing = spec.get("pricing", {})
if model_type == "image": if model_type == "image":
if "generation" in pricing: if "generation" in pricing:
p = pricing.get("generation", {}).get("usd", 0) p = pricing.get("generation", {}).get("usd", 0)
price = f"{p:.3f} DIEM/img" price = f"{p:.3f} DIEM/img"
elif "resolutions" in pricing: elif "resolutions" in pricing:
# Resolution-based pricing (e.g., nano-banana-pro)
res_pricing = pricing.get("resolutions", {}) res_pricing = pricing.get("resolutions", {})
res_parts = [ res_parts = [f"{res}:{p.get('usd', 0):.2f}" for res, p in res_pricing.items()]
f"{res}:{p.get('usd', 0):.2f}"
for res, p in res_pricing.items()
]
price = f"{' | '.join(res_parts)} DIEM" price = f"{' | '.join(res_parts)} DIEM"
else: else:
price = "" price = ""
@@ -261,8 +182,6 @@ class Tools:
price = f"{p:.2f} DIEM/vid" price = f"{p:.2f} DIEM/vid"
else: else:
price = "" price = ""
# Capabilities (text models)
caps = spec.get("capabilities", {}) caps = spec.get("capabilities", {})
cap_list = [] cap_list = []
if caps.get("supportsVision"): if caps.get("supportsVision"):
@@ -273,31 +192,23 @@ class Tools:
cap_list.append("reasoning") cap_list.append("reasoning")
if caps.get("supportsWebSearch"): if caps.get("supportsWebSearch"):
cap_list.append("web") cap_list.append("web")
# Web search for image models (e.g., nano-banana-pro)
if model_type == "image" and spec.get("supportsWebSearch"): if model_type == "image" and spec.get("supportsWebSearch"):
cap_list.append("web") cap_list.append("web")
# Build line
parts = [f" {mid}"] parts = [f" {mid}"]
if name != mid: if name != mid:
parts.append(f"({name})") partsname})")
if price: if price:
parts.append(price) .append(f"({ parts.append(price)
if cap_list: if cap_list:
parts.append(f"[{', '.join(cap_list)}]") parts.append(f"[{', '.join(cap_list)}]")
if beta: if beta:
parts.append("BETA") parts.append("BETA")
if offline: if offline:
parts.append("OFFLINE") parts.append("OFFLINE")
lines.append(" ".join(parts)) lines.append(" ".join(parts))
lines.append("") lines.append("")
lines.append(f"Total: {len(models)} {model_type} models") lines.append(f"Total: {len(models)} {model_type} models")
return "\n".join(lines) return "\n".join(lines)
except httpx.HTTPStatusError as e: except httpx.HTTPStatusError as e:
if __event_emitter__: if __event_emitter__:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
@@ -307,57 +218,27 @@ class Tools:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
return f"List Models\nStatus: 0\nError: {type(e).__name__}: {e}" return f"List Models\nStatus: 0\nError: {type(e).__name__}: {e}"
async def list_styles( async def list_styles(self, __user__: dict = None, __event_emitter__: Callable[[dict], Any] = None) -> str:
self, api_key = VeniceInfo.get_api_key(self.valves, self.user_valves, __user__)
__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: if not api_key:
return "List Styles\nStatus: 0\nError: API key not configured." return "List Styles\nStatus: 0\nError: API key not configured."
if __event_emitter__: if __event_emitter__:
await __event_emitter__( await __event_emitter__({"type": "status", "data": {"description": "Fetching styles...", "done": False}})
{
"type": "status",
"data": {"description": "Fetching styles...", "done": False},
}
)
try: try:
async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client: async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client:
response = await client.get( response = await client.get("https://api.venice.ai/api/v1/image/styles", headers={"Authorization": f"Bearer {api_key}"})
"https://api.venice.ai/api/v1/image/styles",
headers={"Authorization": f"Bearer {api_key}"},
)
response.raise_for_status() response.raise_for_status()
result = response.json() result = response.json()
if __event_emitter__: if __event_emitter__:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
styles = result.get("data", []) styles = result.get("data", [])
lines = [ lines = ["List Styles", "Status: 200", "", "Image Style Presets:"]
"List Styles",
"Status: 200",
"",
"Image Style Presets:",
]
for style in sorted(styles): for style in sorted(styles):
lines.append(f" - {style}") lines.append(f" - {style}")
lines.append("") lines.append("")
lines.append(f"Total: {len(styles)} styles") lines.append(f"Total: {len(styles)} styles")
lines.append("Usage: Pass as 'style_preset' to image generation") lines.append("Usage: Pass as 'style_preset' to image generation")
return "\n".join(lines) return "\n".join(lines)
except httpx.HTTPStatusError as e: except httpx.HTTPStatusError as e:
if __event_emitter__: if __event_emitter__:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
@@ -367,65 +248,29 @@ class Tools:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
return f"List Styles\nStatus: 0\nError: {type(e).__name__}: {e}" return f"List Styles\nStatus: 0\nError: {type(e).__name__}: {e}"
async def list_traits( async def list_traits(self, __user__: dict = None, __event_emitter__: Callable[[dict], Any] = None) -> str:
self, api_key = VeniceInfo.get_api_key(self.valves, self.user_valves, __user__)
__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: if not api_key:
return "List Traits\nStatus: 0\nError: API key not configured." return "List Traits\nStatus: 0\nError: API key not configured."
if __event_emitter__: if __event_emitter__:
await __event_emitter__( await __event_emitter__({"type": "status", "data": {"description": "Fetching traits...", "done": False}})
{
"type": "status",
"data": {"description": "Fetching traits...", "done": False},
}
)
try: try:
async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client: async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client:
response = await client.get( response = await client.get("https://api.venice.ai/api/v1/models/traits", headers={"Authorization": f"Bearer {api_key}"})
"https://api.venice.ai/api/v1/models/traits",
headers={"Authorization": f"Bearer {api_key}"},
)
response.raise_for_status() response.raise_for_status()
result = response.json() result = response.json()
if __event_emitter__: if __event_emitter__:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
traits = result.get("data", {}) traits = result.get("data", {})
lines = [ lines = ["List Traits", "Status: 200", "", "Model Traits (semantic mappings):", ""]
"List Traits",
"Status: 200",
"",
"Model Traits (semantic mappings to recommended models):",
"",
]
# Sort traits for consistent output
for trait_name in sorted(traits.keys()): for trait_name in sorted(traits.keys()):
model_id = traits[trait_name] model_id = traits[trait_name]
lines.append(f" {trait_name}: {model_id}") lines.append(f" {trait_name}: {model_id}")
lines.append("") lines.append("")
lines.append(f"Total: {len(traits)} traits") lines.append(f"Total: {len(traits)} traits")
lines.append("") lines.append("")
lines.append( lines.append("Usage: Request models by trait for automatic selection.")
"Usage: Request models by trait for automatic best-model selection."
)
return "\n".join(lines) return "\n".join(lines)
except httpx.HTTPStatusError as e: except httpx.HTTPStatusError as e:
if __event_emitter__: if __event_emitter__:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
@@ -435,78 +280,36 @@ class Tools:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
return f"List Traits\nStatus: 0\nError: {type(e).__name__}: {e}" return f"List Traits\nStatus: 0\nError: {type(e).__name__}: {e}"
async def list_compatibility( async def list_compatibility(self, __user__: dict = None, __event_emitter__: Callable[[dict], Any] = None) -> str:
self, api_key = VeniceInfo.get_api_key(self.valves, self.user_valves, __user__)
__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: if not api_key:
return "List Compatibility\nStatus: 0\nError: API key not configured." return "List Compatibility\nStatus: 0\nError: API key not configured."
if __event_emitter__: if __event_emitter__:
await __event_emitter__( await __event_emitter__({"type": "status", "data": {"description": "Fetching compatibility mappings...", "done": False}})
{
"type": "status",
"data": {
"description": "Fetching compatibility mappings...",
"done": False,
},
}
)
try: try:
async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client: async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client:
response = await client.get( response = await client.get("https://api.venice.ai/api/v1/models/compatibility_mapping", headers={"Authorization": f"Bearer {api_key}"})
"https://api.venice.ai/api/v1/models/compatibility_mapping",
headers={"Authorization": f"Bearer {api_key}"},
)
response.raise_for_status() response.raise_for_status()
result = response.json() result = response.json()
if __event_emitter__: if __event_emitter__:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
mappings = result.get("data", {}) mappings = result.get("data", {})
lines = [ lines = ["List Compatibility Mappings", "Status: 200", "", "External model names mapped to Venice equivalents:", ""]
"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]] = {} by_target: dict[str, list[str]] = {}
for external_name, venice_model in mappings.items(): for external_name, venice_model in mappings.items():
if venice_model not in by_target: if venice_model not in by_target:
by_target[venice_model] = [] by_target[venice_model] = []
by_target[venice_model].append(external_name) by_target[venice_model].append(external_name)
for venice_model in sorted(by_target.keys()): for venice_model in sorted(by_target.keys()):
external_names = sorted(by_target[venice_model]) external_names = sorted(by_target[venice_model])
lines.append(f" {venice_model}:") lines.append(f" {venice_model}:")
for ext_name in external_names: for ext_name in external_names:
lines.append(f" <- {ext_name}") lines.append(f" <- {ext_name}")
lines.append("") lines.append("")
lines.append(f"Total: {len(mappings)} mappings to {len(by_target)} Venice models")
lines.append(
f"Total: {len(mappings)} mappings to {len(by_target)} Venice models"
)
lines.append("") lines.append("")
lines.append( lines.append("Usage: Use external names (gpt-4o, etc.) for compatibility.")
"Usage: Use external model names (gpt-4o, claude-3-5-sonnet, etc.) for compatibility."
)
return "\n".join(lines) return "\n".join(lines)
except httpx.HTTPStatusError as e: except httpx.HTTPStatusError as e:
if __event_emitter__: if __event_emitter__:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
@@ -516,102 +319,46 @@ class Tools:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
return f"List Compatibility\nStatus: 0\nError: {type(e).__name__}: {e}" return f"List Compatibility\nStatus: 0\nError: {type(e).__name__}: {e}"
async def get_model_info( async def get_model_info(self, model_id: str, __user__: dict = None, __event_emitter__: Callable[[dict], Any] = None) -> str:
self, api_key = VeniceInfo.get_api_key(self.valves, self.user_valves, __user__)
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: if not api_key:
return "Get Model Info\nStatus: 0\nError: API key not configured." return "Get Model Info\nStatus: 0\nError: API key not configured."
if not model_id: if not model_id:
return "Get Model Info\nStatus: 0\nError: model_id required" return "Get Model Info\nStatus: 0\nError: model_id required"
if __event_emitter__: if __event_emitter__:
await __event_emitter__( await __event_emitter__({"type": "status", "data": {"description": f"Looking up {model_id}...", "done": False}})
{
"type": "status",
"data": {"description": f"Looking up {model_id}...", "done": False},
}
)
# Search each type
for model_type in ["text", "image", "video", "embedding", "tts"]: for model_type in ["text", "image", "video", "embedding", "tts"]:
try: try:
async with httpx.AsyncClient( async with httpx.AsyncClient(timeout=float(self.valves.TIMEOUT)) as client:
timeout=float(self.valves.TIMEOUT) response = await client.get(f"https://api.venice.ai/api/v1/models?type={model_type}", headers={"Authorization": f"Bearer {api_key}"})
) 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() response.raise_for_status()
result = response.json() result = response.json()
for model in result.get("data", []): for model in result.get("data", []):
if model.get("id") == model_id: if model.get("id") == model_id:
if __event_emitter__: if __event_emitter__:
await __event_emitter__( await __event_emitter__({"type": "status", "data": {"done": True}})
{"type": "status", "data": {"done": True}}
)
spec = model.get("model_spec", {}) spec = model.get("model_spec", {})
lines = [ 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)}"]
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") desc = spec.get("description")
if desc: if desc:
lines.append(f"Description: {desc}") lines.append(f"Description: {desc}")
ctx = spec.get("availableContextTokens") ctx = spec.get("availableContextTokens")
if ctx: if ctx:
lines.append(f"Context: {ctx:,} tokens") lines.append(f"Context: {ctx:,} tokens")
# Traits
traits = spec.get("traits", []) traits = spec.get("traits", [])
if traits: if traits:
lines.append(f"Traits: {', '.join(traits)}") lines.append(f"Traits: {', '.join(traits)}")
# Constraints (for image models)
constraints = spec.get("constraints", {}) constraints = spec.get("constraints", {})
if constraints: if constraints:
lines.append("") lines.append("")
lines.append("Constraints:") lines.append("Constraints:")
if "promptCharacterLimit" in constraints: if "promptCharacterLimit" in constraints:
lines.append( lines.append(f" Prompt limit: {constraints['promptCharacterLimit']:,} chars")
f" Prompt limit: {constraints['promptCharacterLimit']:,} chars"
)
if "steps" in constraints: if "steps" in constraints:
steps = constraints["steps"] steps = constraints["steps"]
lines.append( lines.append(f" Steps: default={steps.get('default')}, max={steps.get('max')}")
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: if "resolutions" in constraints:
lines.append( lines.append(f" Resolutions: {', '.join(constraints['resolutions'])}")
f" Resolutions: {', '.join(constraints['resolutions'])}"
)
# Pricing
pricing = spec.get("pricing", {}) pricing = spec.get("pricing", {})
if pricing: if pricing:
lines.append("") lines.append("")
@@ -628,42 +375,27 @@ class Tools:
if "resolutions" in pricing: if "resolutions" in pricing:
lines.append(" Resolution-based:") lines.append(" Resolution-based:")
for res, price in pricing["resolutions"].items(): for res, price in pricing["resolutions"].items():
lines.append( lines.append(f" {res}: {price.get('usd', 0):.2f} DIEM")
f" {res}: {price.get('usd', 0):.2f} DIEM"
)
if "upscale" in pricing: if "upscale" in pricing:
lines.append(" Upscale:") lines.append(" Upscale:")
for scale, price in pricing["upscale"].items(): for scale, price in pricing["upscale"].items():
lines.append( lines.append(f" {scale}: {price.get('usd', 0):.2f} DIEM")
f" {scale}: {price.get('usd', 0):.2f} DIEM"
)
# Capabilities
caps = spec.get("capabilities", {}) caps = spec.get("capabilities", {})
active_caps = [k for k, v in caps.items() if v] active_caps = [k for k, v in caps.items() if v]
# Web search for image models
if model_type == "image" and spec.get("supportsWebSearch"): if model_type == "image" and spec.get("supportsWebSearch"):
active_caps.append("supportsWebSearch") active_caps.append("supportsWebSearch")
if active_caps: if active_caps:
lines.append("") lines.append("")
lines.append("Capabilities:") lines.append("Capabilities:")
for cap in active_caps: for cap in active_caps:
lines.append(f" - {cap}") lines.append(f" - {cap}")
# Model source
source = spec.get("modelSource") source = spec.get("modelSource")
if source: if source:
lines.append("") lines.append("")
lines.append(f"Source: {source}") lines.append(f"Source: {source}")
return "\n".join(lines) return "\n".join(lines)
except Exception: except Exception:
continue continue
if __event_emitter__: if __event_emitter__:
await __event_emitter__({"type": "status", "data": {"done": True}}) await __event_emitter__({"type": "status", "data": {"done": True}})
return f"Get Model Info ({model_id})\nStatus: 404\nError: Model not found" return f"Get Model Info ({model_id})\nStatus: 404\nError: Model not found"