""" title: ElevenLabs TTS Action author: justinh-rahb author_url: https://github.com/justinh-rahb funding_url: https://github.com/open-webui version: 0.1.1 license: MIT icon_url: data:image/svg+xml;base64,PHN2ZyB3aWR0aD0iMzIiIGhlaWdodD0iMzIiIHhtbG5zPSJodHRwOi8vd3d3LnczLm9yZy8yMDAwL3N2ZyI+PHJlY3QgeD0iOCIgeT0iNiIgd2lkdGg9IjYiIGhlaWdodD0iMjAiIGZpbGw9IiM0QzRDNEMiLz48cmVjdCB4PSIxOCIgeT0iNiIgd2lkdGg9IjYiIGhlaWdodD0iMjAiIGZpbGw9IiM0QzRDNEMiLz48L3N2Zz4= required_open_webui_version: 0.3.10 """ import requests import uuid import os from pydantic import BaseModel, Field from typing import Callable, Union, Any, Dict, Tuple from config import UPLOAD_DIR from apps.webui.models.files import Files DEBUG = True class Action: class Valves(BaseModel): ELEVENLABS_API_KEY: str = Field( default=None, description="Your ElevenLabs API key." ) ELEVENLABS_MODEL_ID: str = Field( default="eleven_multilingual_v2", description="ID of the ElevenLabs TTS model to use.", ) def __init__(self): self.valves = self.Valves() self.voice_id_cache = {} def status_object( self, description: str = "Unknown State", status: str = "in_progress", done: bool = False, ) -> Dict: return { "type": "status", "data": { "status": status, "description": description, "done": done, }, } async def fetch_available_voices(self) -> Tuple[str, Dict[str, str]]: if DEBUG: print("Debug: Fetching available voices") base_url = "https://api.elevenlabs.io/v1" headers = { "xi-api-key": self.valves.ELEVENLABS_API_KEY, "Content-Type": "application/json", } voices_url = f"{base_url}/voices" try: response = requests.get(voices_url, headers=headers) response.raise_for_status() voices_data = response.json() display_message = "Available voices from ElevenLabs:\n\n" voice_options = {} for voice in voices_data.get("voices", []): voice_name = voice["name"] voice_id = voice["voice_id"] display_message += f"- {voice_name}\n" voice_options[voice_name] = voice_id if DEBUG: print(f"Debug: Found {len(voices_data.get('voices', []))} voices") return display_message, voice_options except requests.RequestException as e: if DEBUG: print(f"Debug: Error fetching voices: {str(e)}") return "Sorry, couldn't fetch available voices at the moment.", {} async def action( self, body: dict, __user__: dict = {}, __event_emitter__: Callable[[dict], Any] = None, __event_call__: Callable[[dict], Any] = None, ) -> None: if DEBUG: print(f"Debug: ElevenLabs TTS action invoked") try: if __event_emitter__: await __event_emitter__( self.status_object("Initializing ElevenLabs Text-to-Speech") ) if not self.valves.ELEVENLABS_API_KEY: raise ValueError("ElevenLabs API key is not set") if "id" not in __user__: raise ValueError("User not authenticated") display_message, self.voice_id_cache = await self.fetch_available_voices() if not self.voice_id_cache: raise ValueError("No available voices to select") response = await __event_call__( { "type": "input", "data": { "title": "Select Voice", "message": display_message, "input_type": "select", "options": list(self.voice_id_cache.keys()), }, } ) if DEBUG: print(f"Debug: Voice selection response: {response}") if isinstance(response, str): selected_voice_name = response elif isinstance(response, dict): selected_voice_name = response.get("message") else: raise ValueError(f"Unexpected response type: {type(response)}") selected_voice_id = self.voice_id_cache.get(selected_voice_name) if DEBUG: print( f"Debug: Selected voice: {selected_voice_name} ({selected_voice_id})" ) if not selected_voice_id: raise ValueError(f"Invalid voice selection: {selected_voice_name}") messages = body.get("messages", []) assistant_message = next( ( message.get("content") for message in reversed(messages) if message.get("role") == "assistant" ), None, ) if not assistant_message: raise ValueError("No assistant message to convert") if __event_emitter__: await __event_emitter__(self.status_object("Generating speech")) base_url = "https://api.elevenlabs.io/v1" headers = { "xi-api-key": self.valves.ELEVENLABS_API_KEY, "Content-Type": "application/json", } tts_url = f"{base_url}/text-to-speech/{selected_voice_id}" payload = { "text": assistant_message, "model_id": self.valves.ELEVENLABS_MODEL_ID, "voice_settings": {"stability": 0.5, "similarity_boost": 0.5}, } response = requests.post(tts_url, json=payload, headers=headers) response.raise_for_status() if response.status_code == 200: audio_data = response.content file_name = f"tts_{uuid.uuid4()}.mp3" file_id = self._create_file( file_name, "Generated Audio", audio_data, "audio/mpeg", __user__ ) if file_id: file_url = self._get_file_url(file_id) if __event_emitter__: await __event_emitter__( self.status_object( "Generated successfully", status="complete", done=True ) ) if file_url: await __event_emitter__( { "type": "message", "data": { "content": f"\n\n---\n- [Download ElevenLabs Audio]({file_url})\n" }, } ) else: raise ValueError("Error saving audio file") else: raise ValueError(f"Unexpected API response: {response.text}") except Exception as e: if DEBUG: print(f"Debug: Error in action method: {str(e)}") if __event_emitter__: await __event_emitter__( self.status_object(f"Error: {str(e)}", status="error", done=True) ) def _create_file( self, file_name: str, title: str, content: Union[str, bytes], content_type: str, __user__: dict = {}, ) -> str: if DEBUG: print(f"Debug: Creating file: {file_name}") if "id" not in __user__: if DEBUG: print("Debug: User ID is not available") return None base_path = os.path.join(UPLOAD_DIR) file_id = str(uuid.uuid4()) file_path = os.path.join(base_path, f"{file_id}_{file_name}") mode = "w" if isinstance(content, str) else "wb" try: os.makedirs(base_path, exist_ok=True) with open(file_path, mode) as f: f.write(content) meta = { "source": file_path, "title": title, "content_type": content_type, "size": os.path.getsize(file_path), "path": file_path, } class FileForm(BaseModel): id: str filename: str meta: dict = {} formData = FileForm(id=file_id, filename=file_name, meta=meta) file = Files.insert_new_file(__user__["id"], formData) if DEBUG: print(f"Debug: File saved. Path: {file_path}") return file.id except Exception as e: if DEBUG: print(f"Debug: Error saving file: {e}") return None def _get_file_url(self, file_id: str) -> str: return f"/api/v1/files/{file_id}/content"