import requests
from typing import Optional
from datetime import datetime
# Ensure the tool resides within the Tools class structure required by Open WebUI
class Tools:
class CW_PromptGen:
def __init__(self, compliance_level: int = 1):
self.compliance_level = compliance_level
self.compliance_attributes = {}
self.llm_requirements = {}
# Define Valves and UserValves for configuration in Open WebUI
class Valves:
model_name: str = "default-model"
compliance_level: int = 1
# Fetch LLM requirements dynamically based on model type
def fetch_llm_requirements(self, model_name: str):
"""Fetches specific LLM prompt requirements from an external API."""
try:
response = requests.get(f"https://api.example.com/models/{model_name}/requirements")
if response.status_code == 200:
self.llm_requirements = response.json()
return "LLM requirements fetched successfully."
else:
return f"Failed to fetch LLM requirements: {response.status_code}"
except requests.ConnectionError:
return "Internet connection unavailable."
def adaptive_prompt(self, user_input: str) -> str:
"""Generates a prompt customized to the compliance level and LLM requirements."""
base_prompt = "Generate response based on compliance level and user context."
llm_specifics = f" (LLM adjustments: {self.llm_requirements.get('prompt_structure', 'Standard')})"
if self.compliance_level == 1:
return f"{base_prompt} Basic response for input: {user_input}{llm_specifics}"
elif self.compliance_level == 2:
return f"{base_prompt} Intermediate response for input: {user_input}{llm_specifics}"
elif self.compliance_level == 3:
return f"{base_prompt} Advanced response with high detail for input: {user_input}{llm_specifics}"
else:
return f"{base_prompt} Default response for input: {user_input}{llm_specifics}"
# Event Emitters for real-time status updates
async def emit_status(self, message: str, is_done: bool = False):
"""Emit real-time status updates via Open WebUI's event emitters."""
await self.__event_emitter__(
{"type": "status", "data": {"description": message, "done": is_done}}
)
# Main function to handle input, check LLM requirements, and generate the prompt
async def handle_user_input(self, user_input: str, model_name: str) -> dict:
"""Processes input, fetches requirements, and generates a custom prompt."""
await self.emit_status("Fetching LLM requirements...")
llm_fetch_status = self.fetch_llm_requirements(model_name)
await self.emit_status("Generating adaptive prompt...")
prompt = self.adaptive_prompt(user_input)
return {
"fetch_status": llm_fetch_status,
"generated_prompt": prompt,
"compliance_level": self.compliance_level,
"verbosity": self.compliance_attributes.get("verbosity", "standard")
}
