From 5c7394d0cd2fde7cff0038f36541998b25393b81 Mon Sep 17 00:00:00 2001 From: xcaliber Date: Thu, 15 Jan 2026 11:35:40 +0000 Subject: [PATCH] refactor: add VeniceChat namespace for helper functions - Added VeniceChat class with get_api_key, truncate, format_error static methods - Prevents Open WebUI framework method name collision issues - All tool methods now use VeniceChat.get_api_key() pattern - Version bump to 1.4.0 --- venice/chat.py | 239 +++++++++++++++++++++++++++++++------------------ 1 file changed, 151 insertions(+), 88 deletions(-) diff --git a/venice/chat.py b/venice/chat.py index 583d104..1e43269 100644 --- a/venice/chat.py +++ b/venice/chat.py @@ -1,7 +1,7 @@ """ title: Venice.ai Chat author: Jeff Smith -version: 1.2.0 +version: 1.4.0 license: MIT required_open_webui_version: 0.6.0 requirements: httpx, pydantic @@ -17,15 +17,74 @@ description: | - Explicit ID: Validates model exists before calling Use venice_info/list_models("text") to discover available models. +changelog: + 1.4.0: + - Added VeniceChat namespace class for helper functions to avoid method collisions + - Moved _get_api_key, _truncate, _format_error to VeniceChat namespace + - Prevents Open WebUI framework introspection method name collisions + 1.3.0: + - Fixed UserValves access pattern for per-user API keys + - Added __request__ parameter handling for zero-config API calls + - Enhanced __init__ for framework-driven configuration injection + - Added _format_error() helper for consistent error messages + - Set self.citation = True for tool usage visibility + - Improved response formatting consistency """ -from typing import Optional, Callable, Any +from typing import Optional, Callable, Any, Dict from pydantic import BaseModel, Field import httpx import json import time +class VeniceChat: + """ + Namespaced helpers for Venice chat 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.""" + # Check __user__ parameter first (for direct method calls) + 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"] + + # Fall back to UserValves instance + return user_valves.VENICE_API_KEY or valves.VENICE_API_KEY + + @staticmethod + def truncate(text: str, max_size: int) -> str: + """Truncate response to max size.""" + if max_size and len(text) > max_size: + return text[:max_size] + f"\n\n[...{len(text) - max_size} chars omitted]" + return text + + @staticmethod + def format_error(e, context: str = "") -> str: + """Format HTTP error with detailed context for LLM understanding.""" + try: + if hasattr(e, "response") and e.response is not None: + error_msg = e.response.text[:200] + try: + error_json = e.response.json() + error_msg = error_json.get("message", error_msg) + except Exception: + pass + else: + error_msg = str(e)[:200] + except Exception: + error_msg = str(e)[:200] + + context_str = f" ({context})" if context else "" + return f"Error{context_str}: {error_msg}" + + class Tools: """ Venice.ai chat completions tool. @@ -76,40 +135,33 @@ class Tools: ) def __init__(self): + """Initialize with optional valve configuration from framework""" + # Handle valves configuration from framework self.valves = self.Valves() + + # Enable tool usage visibility for debugging + self.citation = True + + # Handle user valves configuration self.user_valves = self.UserValves() - self.citation = False + # Simple in-memory cache self._cache: dict = {} self._cache_times: dict = {} - 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 - - def _truncate(self, text: str) -> str: - """Truncate response to max size.""" - max_size = self.valves.MAX_RESPONSE_SIZE - if max_size and len(text) > max_size: - return ( - text[:max_size] - + f"\n\n[...truncated, {len(text) - max_size} chars omitted]" - ) - return text - def _is_cache_valid(self, key: str) -> bool: """Check if cached data is still valid.""" if key not in self._cache_times: return False return (time.time() - self._cache_times[key]) < self.valves.MODEL_CACHE_TTL - async def _get_traits(self) -> dict: + async def _get_traits(self, __user__: dict = None) -> dict: """Fetch model traits from Venice (cached).""" cache_key = "traits" if self._is_cache_valid(cache_key): return self._cache.get(cache_key, {}) - api_key = self._get_api_key() + api_key = VeniceChat.get_api_key(self.valves, self.user_valves, __user__) if not api_key: return {} @@ -128,13 +180,15 @@ class Tools: pass return {} - async def _get_available_models(self, model_type: str = "text") -> list[dict]: + async def _get_available_models( + self, model_type: str = "text", __user__: dict = None + ) -> list[dict]: """Fetch available models (cached).""" cache_key = f"models_{model_type}" if self._is_cache_valid(cache_key): return self._cache.get(cache_key, []) - api_key = self._get_api_key() + api_key = VeniceChat.get_api_key(self.valves, self.user_valves, __user__) if not api_key: return [] @@ -159,6 +213,7 @@ class Tools: model_type: str = "text", require_reasoning: bool = False, __model__: dict = None, + __user__: dict = None, ) -> tuple[str, Optional[str]]: """ Resolve model specification to actual model ID with validation. @@ -190,7 +245,7 @@ class Tools: # If still no model, try traits API if not model: - traits = await self._get_traits() + traits = await self._get_traits(__user__) if require_reasoning: # Try reasoning-specific traits @@ -208,7 +263,7 @@ class Tools: # If still no model, pick first available with required capability if not model: - models = await self._get_available_models(model_type) + models = await self._get_available_models(model_type, __user__) for m in models: spec = m.get("model_spec", {}) if spec.get("offline"): @@ -232,7 +287,7 @@ class Tools: ) # Validate model exists and is online - models = await self._get_available_models(model_type) + models = await self._get_available_models(model_type, __user__) model_map = {m.get("id"): m for m in models} if model not in model_map: @@ -317,19 +372,19 @@ class Tools: :param web_search: Enable web search for current information :return: Model response """ - api_key = self._get_api_key() + api_key = VeniceChat.get_api_key(self.valves, self.user_valves, __user__) if not api_key: - return "Venice Chat\nStatus: 0\nError: API key not configured. Set in UserValves or ask admin." + return "Error: Venice API key not configured. Set VENICE_API_KEY in UserValves or ask admin." if not message or not message.strip(): - return "Venice Chat\nStatus: 0\nError: Message required" + return "Error: Message is required" # Resolve and validate model resolved_model, error = await self._resolve_model( - model, "text", False, __model__ + model, "text", False, __model__, __user__ ) if error: - return f"Venice Chat\nStatus: 0\nError: {error}" + return f"Error: {error}" enable_web_search = ( web_search if web_search is not None else self.valves.ENABLE_WEB_SEARCH @@ -384,56 +439,56 @@ class Tools: choices = result.get("choices", []) if not choices: - return f"Venice Chat ({resolved_model})\nStatus: 200\nError: No response from model" + return f"Error: No response from model {resolved_model}" assistant_message = choices[0].get("message", {}) content = assistant_message.get("content", "") reasoning = assistant_message.get("reasoning_content") # Build response - lines = [f"Venice Chat ({resolved_model})", "Status: 200", ""] + lines = [f"**Venice Chat** ({resolved_model})", ""] if reasoning: - lines.append(f"Reasoning:\n{reasoning}") + lines.append(f"**Reasoning:**\n{reasoning}") lines.append("") - lines.append(f"Response:\n{content}") + lines.append(f"**Response:**\n{content}") # Include web citations if present venice_params = result.get("venice_parameters", {}) citations = venice_params.get("web_search_citations", []) if citations: lines.append("") - lines.append("Sources:") + lines.append("**Sources:**") for cite in citations[:5]: title = cite.get("title", "Link") url = cite.get("url", "") - lines.append(f" - {title}: {url}") + lines.append(f"- {title}: {url}") # Usage stats usage = result.get("usage", {}) if usage: lines.append("") lines.append( - f"Tokens: {usage.get('prompt_tokens', 0)} in / {usage.get('completion_tokens', 0)} out" + f"_Tokens: {usage.get('prompt_tokens', 0)} in / {usage.get('completion_tokens', 0)} out_" ) - return self._truncate("\n".join(lines)) + return VeniceChat.truncate("\n".join(lines), self.valves.MAX_RESPONSE_SIZE) except httpx.HTTPStatusError as e: + error_msg = VeniceChat.format_error(e, f"chat request to {resolved_model}") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) - return f"Venice Chat ({resolved_model})\nStatus: {e.response.status_code}\nError: {e.response.text[:200]}" + return f"Error: {error_msg}" except httpx.TimeoutException: if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) - return ( - f"Venice Chat ({resolved_model})\nStatus: 408\nError: Request timed out" - ) + return f"Error: Request timed out for {resolved_model}" except Exception as e: + error_msg = VeniceChat.format_error(e, "chat request") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) - return f"Venice Chat ({resolved_model})\nStatus: 0\nError: {type(e).__name__}: {e}" + return f"Error: {error_msg}" async def chat_conversation( self, @@ -456,26 +511,26 @@ class Tools: :param max_tokens: Maximum response tokens (default 2048) :return: Model response """ - api_key = self._get_api_key() + api_key = VeniceChat.get_api_key(self.valves, self.user_valves, __user__) if not api_key: - return "Venice Chat Conversation\nStatus: 0\nError: API key not configured." + return "Error: Venice API key not configured." if not messages_json: - return "Venice Chat Conversation\nStatus: 0\nError: messages_json required" + return "Error: messages_json is required" try: conversation = json.loads(messages_json) if not isinstance(conversation, list): - return "Venice Chat Conversation\nStatus: 0\nError: messages_json must be a JSON array" + return "Error: messages_json must be a JSON array" except json.JSONDecodeError as e: - return f"Venice Chat Conversation\nStatus: 0\nError: Invalid JSON - {e}" + return f"Error: Invalid JSON - {e}" # Resolve and validate model resolved_model, error = await self._resolve_model( - model, "text", False, __model__ + model, "text", False, __model__, __user__ ) if error: - return f"Venice Chat Conversation\nStatus: 0\nError: {error}" + return f"Error: {error}" if __event_emitter__: await __event_emitter__( @@ -526,30 +581,32 @@ class Tools: choices = result.get("choices", []) if not choices: - return f"Venice Chat Conversation ({resolved_model})\nStatus: 200\nError: No response" + return f"Error: No response from model {resolved_model}" assistant_message = choices[0].get("message", {}) content = assistant_message.get("content", "") reasoning = assistant_message.get("reasoning_content") - lines = [f"Venice Chat Conversation ({resolved_model})", "Status: 200", ""] + lines = [f"**Venice Chat Conversation** ({resolved_model})", ""] if reasoning: - lines.append(f"Reasoning:\n{reasoning}") + lines.append(f"**Reasoning:**\n{reasoning}") lines.append("") - lines.append(f"Response:\n{content}") + lines.append(f"**Response:**\n{content}") - return self._truncate("\n".join(lines)) + return VeniceChat.truncate("\n".join(lines), self.valves.MAX_RESPONSE_SIZE) except httpx.HTTPStatusError as e: + error_msg = VeniceChat.format_error(e, f"conversation with {resolved_model}") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) - return f"Venice Chat Conversation ({resolved_model})\nStatus: {e.response.status_code}\nError: {e.response.text[:200]}" + return f"Error: {error_msg}" except Exception as e: + error_msg = VeniceChat.format_error(e, "conversation request") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) - return f"Venice Chat Conversation ({resolved_model})\nStatus: 0\nError: {type(e).__name__}: {e}" + return f"Error: {error_msg}" async def ask_reasoning_model( self, @@ -568,22 +625,26 @@ class Tools: :param model: Model with reasoning capability, "self", or empty for auto-select. Use venice_info/list_models("text") to find models with [reasoning]. :return: Response with reasoning process """ - api_key = self._get_api_key() + api_key = VeniceChat.get_api_key(self.valves, self.user_valves, __user__) if not api_key: - return "Venice Reasoning\nStatus: 0\nError: API key not configured." + return "Error: Venice API key not configured." if not question or not question.strip(): - return "Venice Reasoning\nStatus: 0\nError: Question required" + return "Error: Question is required" if reasoning_effort not in ["low", "medium", "high"]: - return "Venice Reasoning\nStatus: 0\nError: reasoning_effort must be low, medium, or high" + return "Error: reasoning_effort must be low, medium, or high" # Resolve and validate model (require reasoning capability) resolved_model, error = await self._resolve_model( - model, "text", require_reasoning=True, __model__=__model__ + model, + "text", + require_reasoning=True, + __model__=__model__, + __user__=__user__, ) if error: - return f"Venice Reasoning\nStatus: 0\nError: {error}" + return f"Error: {error}" if __event_emitter__: await __event_emitter__( @@ -627,24 +688,23 @@ class Tools: choices = result.get("choices", []) if not choices: - return f"Venice Reasoning ({resolved_model})\nStatus: 200\nError: No response" + return f"Error: No response from model {resolved_model}" assistant_message = choices[0].get("message", {}) content = assistant_message.get("content", "") reasoning = assistant_message.get("reasoning_content", "") lines = [ - f"Venice Reasoning ({resolved_model})", - "Status: 200", - f"Effort: {reasoning_effort}", + f"**Venice Reasoning** ({resolved_model})", + f"**Effort:** {reasoning_effort}", "", ] if reasoning: - lines.append(f"Reasoning Process:\n{reasoning}") + lines.append(f"**Reasoning Process:**\n{reasoning}") lines.append("") - lines.append(f"Answer:\n{content}") + lines.append(f"**Answer:**\n{content}") # Usage stats usage = result.get("usage", {}) @@ -653,23 +713,25 @@ class Tools: total = usage.get("total_tokens", 0) reasoning_tokens = usage.get("reasoning_tokens", 0) lines.append( - f"Tokens: {total:,} total ({reasoning_tokens:,} reasoning)" + f"_Tokens: {total:,} total ({reasoning_tokens:,} reasoning)_" ) - return self._truncate("\n".join(lines)) + return VeniceChat.truncate("\n".join(lines), self.valves.MAX_RESPONSE_SIZE) except httpx.HTTPStatusError as e: + error_msg = VeniceChat.format_error(e, f"reasoning with {resolved_model}") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) - return f"Venice Reasoning ({resolved_model})\nStatus: {e.response.status_code}\nError: {e.response.text[:200]}" + return f"Error: {error_msg}" except httpx.TimeoutException: if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) - return f"Venice Reasoning ({resolved_model})\nStatus: 408\nError: Request timed out (reasoning can take a while)" + return f"Error: Request timed out for {resolved_model} (reasoning can take a while)" except Exception as e: + error_msg = VeniceChat.format_error(e, "reasoning request") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) - return f"Venice Reasoning ({resolved_model})\nStatus: 0\nError: {type(e).__name__}: {e}" + return f"Error: {error_msg}" async def web_search_query( self, @@ -686,19 +748,19 @@ class Tools: :param model: Model to use, "self", or empty for auto-select :return: Response with web sources """ - api_key = self._get_api_key() + api_key = VeniceChat.get_api_key(self.valves, self.user_valves, __user__) if not api_key: - return "Venice Web Search\nStatus: 0\nError: API key not configured." + return "Error: Venice API key not configured." if not query or not query.strip(): - return "Venice Web Search\nStatus: 0\nError: Query required" + return "Error: Query is required" # Resolve model - prefer models with web search capability resolved_model, error = await self._resolve_model( - model, "text", False, __model__ + model, "text", False, __model__, __user__ ) if error: - return f"Venice Web Search\nStatus: 0\nError: {error}" + return f"Error: {error}" if __event_emitter__: await __event_emitter__( @@ -743,16 +805,15 @@ class Tools: choices = result.get("choices", []) if not choices: - return f"Venice Web Search ({resolved_model})\nStatus: 200\nError: No response" + return f"Error: No response from model {resolved_model}" assistant_message = choices[0].get("message", {}) content = assistant_message.get("content", "") lines = [ - f"Venice Web Search ({resolved_model})", - "Status: 200", + f"**Venice Web Search** ({resolved_model})", "", - f"Response:\n{content}", + f"**Response:**\n{content}", ] # Include citations @@ -760,20 +821,22 @@ class Tools: citations = venice_params.get("web_search_citations", []) if citations: lines.append("") - lines.append(f"Sources ({len(citations)}):") + lines.append(f"**Sources** ({len(citations)}):") for cite in citations[:10]: title = cite.get("title", "Link") url = cite.get("url", "") - lines.append(f" - {title}") - lines.append(f" {url}") + lines.append(f"- {title}") + lines.append(f" {url}") - return self._truncate("\n".join(lines)) + return VeniceChat.truncate("\n".join(lines), self.valves.MAX_RESPONSE_SIZE) except httpx.HTTPStatusError as e: + error_msg = VeniceChat.format_error(e, f"web search with {resolved_model}") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) - return f"Venice Web Search ({resolved_model})\nStatus: {e.response.status_code}\nError: {e.response.text[:200]}" + return f"Error: {error_msg}" except Exception as e: + error_msg = VeniceChat.format_error(e, "web search request") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) - return f"Venice Web Search ({resolved_model})\nStatus: 0\nError: {type(e).__name__}: {e}" + return f"Error: {error_msg}"