open-webui-automation/tools
2
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

refactor: add VeniceChat namespace for helper functions

Browse Source 
- 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
This commit is contained in:
xcaliber
2026-01-15 11:37:14 +00:00
parent 0464abaa32
commit 455fc3556f
1 changed files with 72 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
venice/chat.py
View File

@@ -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}"