diff --git a/venice/chat.py b/venice/chat.py index 6dfafdc..5d55d2c 100644 --- a/venice/chat.py +++ b/venice/chat.py @@ -1,7 +1,7 @@ """ title: Venice.ai Chat author: Jeff Smith -version: 1.3.0 +version: 1.4.0 license: MIT required_open_webui_version: 0.6.0 requirements: httpx, pydantic @@ -18,6 +18,10 @@ description: | 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 @@ -34,6 +38,53 @@ 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. @@ -98,60 +149,19 @@ class Tools: self._cache: dict = {} self._cache_times: dict = {} - def _get_api_key(self, __user__: dict = None) -> str: - """Get Venice API key with UserValves priority.""" - # Check __user__ parameter first (for direct method calls) - if __user__ and "valves" in __user__: - user_valves = __user__.get("valves") - if isinstance(user_valves, dict) and user_valves.get("VENICE_API_KEY"): - return user_valves["VENICE_API_KEY"] - elif hasattr(user_valves, "VENICE_API_KEY"): - return user_valves.VENICE_API_KEY - - # Fall back to UserValves instance - 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 - def _format_error(self, 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}" - 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(__user__) + api_key = VeniceChat.get_api_key(self.valves, self.user_valves, __user__) if not api_key: return {} @@ -178,7 +188,7 @@ class Tools: if self._is_cache_valid(cache_key): return self._cache.get(cache_key, []) - api_key = self._get_api_key(__user__) + api_key = VeniceChat.get_api_key(self.valves, self.user_valves, __user__) if not api_key: return [] @@ -362,7 +372,7 @@ class Tools: :param web_search: Enable web search for current information :return: Model response """ - api_key = self._get_api_key(__user__) + api_key = VeniceChat.get_api_key(self.valves, self.user_valves, __user__) if not api_key: return "Error: Venice API key not configured. Set VENICE_API_KEY in UserValves or ask admin." @@ -463,10 +473,10 @@ class Tools: 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 = self._format_error(e, f"chat request to {resolved_model}") + 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"Error: {error_msg}" @@ -475,7 +485,7 @@ class Tools: await __event_emitter__({"type": "status", "data": {"done": True}}) return f"Error: Request timed out for {resolved_model}" except Exception as e: - error_msg = self._format_error(e, "chat request") + error_msg = VeniceChat.format_error(e, "chat request") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) return f"Error: {error_msg}" @@ -501,7 +511,7 @@ class Tools: :param max_tokens: Maximum response tokens (default 2048) :return: Model response """ - api_key = self._get_api_key(__user__) + api_key = VeniceChat.get_api_key(self.valves, self.user_valves, __user__) if not api_key: return "Error: Venice API key not configured." @@ -585,15 +595,15 @@ class Tools: 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 = self._format_error(e, f"conversation with {resolved_model}") + error_msg = VeniceChat.format_error(e, f"conversation with {resolved_model}") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) return f"Error: {error_msg}" except Exception as e: - error_msg = self._format_error(e, "conversation request") + error_msg = VeniceChat.format_error(e, "conversation request") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) return f"Error: {error_msg}" @@ -615,7 +625,7 @@ 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(__user__) + api_key = VeniceChat.get_api_key(self.valves, self.user_valves, __user__) if not api_key: return "Error: Venice API key not configured." @@ -706,10 +716,10 @@ class Tools: 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 = self._format_error(e, f"reasoning with {resolved_model}") + error_msg = VeniceChat.format_error(e, f"reasoning with {resolved_model}") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) return f"Error: {error_msg}" @@ -718,7 +728,7 @@ class Tools: await __event_emitter__({"type": "status", "data": {"done": True}}) return f"Error: Request timed out for {resolved_model} (reasoning can take a while)" except Exception as e: - error_msg = self._format_error(e, "reasoning request") + error_msg = VeniceChat.format_error(e, "reasoning request") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) return f"Error: {error_msg}" @@ -738,7 +748,7 @@ class Tools: :param model: Model to use, "self", or empty for auto-select :return: Response with web sources """ - api_key = self._get_api_key(__user__) + api_key = VeniceChat.get_api_key(self.valves, self.user_valves, __user__) if not api_key: return "Error: Venice API key not configured." @@ -765,8 +775,8 @@ class Tools: payload = { "model": resolved_model, - "messages": [{"role": "user", "content": query}], - "temperature": 0.3, # Lower for factual responses + "messages": [{"role": "user",}], + "temperature": 0.3, "content": query # Lower for factual responses "max_tokens": 2048, "stream": False, "venice_parameters": { @@ -818,15 +828,15 @@ class Tools: 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 = self._format_error(e, f"web search with {resolved_model}") + 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"Error: {error_msg}" except Exception as e: - error_msg = self._format_error(e, "web search request") + error_msg = VeniceChat.format_error(e, "web search request") if __event_emitter__: await __event_emitter__({"type": "status", "data": {"done": True}}) return f"Error: {error_msg}"