"""" title: Informative Style Enforcer author: Sam Koscelny version: 1.0 required_open_webui_version: 0.3.9 """ # informative_chatbot.py from pydantic import BaseModel, Field from typing import Optional, List, Callable, Awaitable, Dict, Any import aiohttp import asyncio import time class Action: class Valves(BaseModel): models: List[str] = Field( default=["mistral-7b"], description="List of models to use for generating responses." ) openai_api_base: str = Field( default="http://host.docker.internal:11434/v1", description="Base URL for Ollama API." ) max_tokens: int = Field( default=150, description="Maximum number of tokens in the response." ) enable_status_indicator: bool = Field( default=True, description="Enable or disable status updates." ) def __init__(self): self.valves = self.Valves() self.last_emit_time = 0 async def generate_response( self, body: Dict[str, Any], __event_emitter__: Callable[[Dict[str, Any]], Awaitable[None]] = None, ) -> Optional[Dict[str, Any]]: user_query = body.get("query", "").strip() if not user_query: error_msg = "No query provided in the request body." await self.emit_status(__event_emitter__, "error", error_msg, True) return {"error": error_msg} styled_response = await self.get_informative_response(user_query, __event_emitter__) if styled_response.startswith("Error:"): await self.emit_status(__event_emitter__, "error", styled_response, True) return {"error": styled_response} body["response"] = styled_response await self.emit_status(__event_emitter__, "info", "Informative chatbot response generated.", True) return body async def get_informative_response( self, prompt: str, __event_emitter__: Callable[[Dict[str, Any]], Awaitable[None]] = None ) -> str: adjusted_prompt = "You are a professional healthcare advisor. Provide detailed and informative responses.\n" + prompt try: response = await self.query_ollama( self.valves.models[0], adjusted_prompt, __event_emitter__, temperature=0.3, top_p=0.8 ) return self.add_formal_structure(response) except Exception as e: error_msg = f"Error generating informative response: {str(e)}" await self.emit_status(__event_emitter__, "error", error_msg, True) return f"Error: {error_msg}" def add_formal_structure(self, text: str) -> str: formatted_response = "### Detailed Information\n" for line in text.split("\n"): formatted_response += f"- {line}\n" if line else "\n" return formatted_response async def query_ollama( self, model: str, prompt: str, __event_emitter__: Callable[[Dict[str, Any]], Awaitable[None]] = None, temperature: float = 0.3, top_p: float = 0.8 ) -> str: url = f"{self.valves.openai_api_base}/chat/completions" headers = {"Content-Type": "application/json"} data = { "model": model, "messages": [{"role": "user", "content": prompt}], "temperature": temperature, "top_p": top_p } async with aiohttp.ClientSession() as session: async with session.post(url, headers=headers, json=data) as response: response.raise_for_status() result = await response.json() return result["choices"][0]["message"]["content"] async def emit_status( self, __event_emitter__: Callable[[Dict[str, Any]], Awaitable[None]], level: str, message: str, done: bool, ): status_message = { "type": "status", "data": { "status": "complete" if done else "in_progress", "level": level, "description": message, "done": done, }, } await __event_emitter__(status_message)