""" title: MultiLanguage LibreTranslate Action description: Advanced translation to multiple languages simultaneously using LibreTranslate API with retry mechanisms and language detection author: @iamg30 author_url: https://openwebui.com/u/iamg30/ forked from https://openwebui.com/f/jthesse/libretranslate_action funding_url: https://github.com/open-webui version: 0.2.0 license: MIT icon_url: data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHdpZHRoPSIxZW0iIGhlaWdodD0iMWVtIiB2aWV3Qm94PSIwIDAgMjQgMjQiPjxwYXRoIGZpbGw9ImN1cnJlbnRDb2xvciIgZD0ibTEyLjg3IDE1LjA3bC0yLjU0LTIuNTFsLjAzLS4wM0ExNy41IDE3LjUgMCAwIDAgMTQuMDcgNkgxN1Y0aC03VjJIOHYySDF2MmgxMS4xN0MxMS41IDcuOTIgMTAuNDQgOS43NSA5IDExLjM1QzguMDcgMTAuMzIgNy4zIDkuMTkgNi42OSA4aC0yYy43MyAxLjYzIDEuNzMgMy4xNyAyLjk4IDQuNTZsLTUuMDkgNS4wMkw0IDE5bDUtNWwzLjExIDMuMTF6TTE4LjUgMTBoLTJMMTIgMjJoMmwxLjEyLTNoNC43NUwyMSAyMmgyem0tMi42MiA3bDEuNjItNC4zM0wxOS4xMiAxN3oiLz48L3N2Zz4= """ from pydantic import BaseModel, Field from typing import Optional, List, Dict, Any, Tuple, Callable, Coroutine from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type import requests import asyncio import time import logging from fastapi import Request from open_webui.utils.misc import get_last_assistant_message # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger("MultiLanguageLibreTranslateAction") # Pre-defined list of supported languages SUPPORTED_LANGUAGES = { "sq": "Albanian", "ar": "Arabic", "az": "Azerbaijani", "bn": "Bengali", "bg": "Bulgarian", "ca": "Catalan, Valencian", "zh": "Chinese", "cs": "Czech", "da": "Danish", "nl": "Dutch, Flemish", "en": "English", "eo": "Esperanto", "et": "Estonian", "fi": "Finnish", "fr": "French", "de": "German", "el": "Greek", "he": "Hebrew", "hi": "Hindi", "hu": "Hungarian", "id": "Indonesian", "ga": "Irish", "it": "Italian", "ja": "Japanese", "ko": "Korean", "lv": "Latvian", "lt": "Lithuanian", "ms": "Malay", "fa": "Persian", "pl": "Polish", "pt": "Portuguese", "ro": "Romanian, Moldavian, Moldovan", "ru": "Russian", "sk": "Slovak", "sl": "Slovenian", "es": "Spanish, Castilian", "sv": "Swedish", "tl": "Tagalog", "th": "Thai", "tr": "Turkish", "uk": "Ukrainian", "ur": "Urdu", } class ValidationResult: """ Class to represent the result of a validation operation. Attributes: is_valid (bool): Whether the validation passed or failed error_message (str): The error message if validation failed, empty string otherwise result (Any): The validated result if validation passed, None otherwise """ def __init__(self, is_valid: bool, error_message: str = "", result: Any = None): self.is_valid = is_valid self.error_message = error_message self.result = result class Action: class Valves(BaseModel): LIBRE_TRANSLATE_URL: str = Field( default="http://localhost:5000", description="The URL of the LibreTranslate instance.", ) API_KEY: str = Field( default="", description="API key for LibreTranslate (if required).", ) ENABLE_LANGUAGE_DETECTION: bool = Field( default=True, description="Automatically detect source language when set to 'auto'.", ) MAX_RETRY_ATTEMPTS: int = Field( default=3, description="Maximum number of retry attempts for API calls.", ) RETRY_DELAY_SECONDS: int = Field( default=2, description="Delay between retry attempts in seconds.", ) PARALLEL_TRANSLATIONS: bool = Field( default=False, description="Process translations in parallel (faster but may overload server).", ) LANGUAGE_DETECTION_WAIT_SECONDS: float = Field( default=0.0, description="Wait time after language detection (for display purposes).", ) class UserValves(BaseModel): SOURCE_LANGUAGE: str = Field( default="auto", description="User-specific source language for assistant messages", ) TARGET_LANGUAGES: list[str] = Field( default=["de", "fr", "es"], description="User-specific target languages for assistant messages", ) def __init__(self): """Initialize the Action with default values.""" self.valves = self.Valves() self.api_languages = [] async def send_status( self, emitter: Callable[[Dict[str, Any]], Coroutine[Any, Any, None]], status_type: str = "status", description: str = "", done: bool = False, ) -> None: """ Send a status update through the event emitter. Args: emitter: Callable function that emits events status_type: Type of status ("status", "error", "warning", etc.) description: Human-readable description of the status done: Whether this status update represents a completed operation """ if not emitter: logger.warning(f"Status update not sent (no emitter): {description}") return try: await emitter( { "type": "status", "data": { "status": status_type, "description": description, "done": done, }, } ) except Exception as e: logger.error(f"Failed to send status update: {str(e)}") async def send_error_status( self, emitter: Callable[[Dict[str, Any]], Coroutine[Any, Any, None]], message: str, ) -> None: """ Send an error status through the event emitter. Args: emitter: Callable function that emits events message: Error message to display """ await self.send_status( emitter, status_type="error", description=message, done=True ) async def send_warning_status( self, emitter: Callable[[Dict[str, Any]], Coroutine[Any, Any, None]], message: str, ) -> None: """ Send a warning status through the event emitter. Args: emitter: Callable function that emits events message: Warning message to display """ await self.send_status( emitter, status_type="warning", description=message, done=False ) async def send_progress_status( self, emitter: Callable[[Dict[str, Any]], Coroutine[Any, Any, None]], message: str, ) -> None: """ Send a progress status through the event emitter. Args: emitter: Callable function that emits events message: Progress message to display """ await self.send_status(emitter, description=message, done=False) async def send_completion_status( self, emitter: Callable[[Dict[str, Any]], Coroutine[Any, Any, None]], message: str, ) -> None: """ Send a completion status through the event emitter. Args: emitter: Callable function that emits events message: Completion message to display """ await self.send_status(emitter, description=message, done=True) async def send_message( self, emitter: Callable[[Dict[str, Any]], Coroutine[Any, Any, None]], content: str, ) -> None: """ Send a message through the event emitter. Args: emitter: Callable function that emits events content: Message content to send """ if not emitter: logger.warning(f"Message not sent (no emitter): {content[:50]}...") return try: await emitter({"type": "message", "data": {"content": content}}) except Exception as e: logger.error(f"Failed to send message: {str(e)}") async def get_supported_languages(self) -> List[Dict[str, Any]]: """ Fetch supported languages from the LibreTranslate API. Returns: A list of supported languages with their codes and names. Raises: ConnectionError: If unable to connect to the LibreTranslate service """ try: url = f"{self.valves.LIBRE_TRANSLATE_URL}/languages" response = requests.get(url, timeout=10) if response.status_code == 200: self.api_languages = response.json() return self.api_languages else: logger.error( f"Failed to get supported languages: {response.status_code} - {response.text}" ) raise ConnectionError( f"API returned status code {response.status_code}" ) except requests.exceptions.RequestException as e: logger.error(f"Error fetching supported languages: {str(e)}") raise ConnectionError(f"Failed to connect to LibreTranslate: {str(e)}") except Exception as e: logger.error(f"Unexpected error fetching supported languages: {str(e)}") raise async def detect_language(self, text: str) -> str: """ Detect the language of the provided text. Args: text: The text to detect the language of Returns: The detected language code or "auto" if detection fails or is disabled Raises: ConnectionError: If unable to connect to the LibreTranslate service """ if not text or not self.valves.ENABLE_LANGUAGE_DETECTION: return "auto" # Use only the first 1000 characters for detection (for efficiency) sample_text = text[:1000].strip() if not sample_text: return "auto" payload = {"q": sample_text} if self.valves.API_KEY: payload["api_key"] = self.valves.API_KEY try: url = f"{self.valves.LIBRE_TRANSLATE_URL}/detect" response = requests.post(url, json=payload, timeout=10) if response.status_code == 200: detections = response.json() if detections and len(detections) > 0: return detections[0]["language"] else: logger.warning( f"Language detection failed: {response.status_code} - {response.text}" ) return "auto" except requests.exceptions.RequestException as e: logger.error(f"Language detection request error: {str(e)}") return "auto" except Exception as e: logger.error(f"Unexpected language detection error: {str(e)}") return "auto" async def translate_text( self, text: str, source: str, target: str ) -> Dict[str, Any]: """ Translate text with retry logic. Args: text: The text to translate source: The source language code target: The target language code Returns: Dictionary containing the translation result Raises: ValueError: If empty text is provided ConnectionError: If translation fails after max retries """ if not text: raise ValueError("Empty text provided for translation") # Define retry function with fixed parameters @retry( stop=stop_after_attempt(self.valves.MAX_RETRY_ATTEMPTS), wait=wait_fixed(self.valves.RETRY_DELAY_SECONDS), retry=retry_if_exception_type( (requests.exceptions.RequestException, ConnectionError) ), reraise=True, ) def do_translate(): payload = { "q": text, "source": source, "target": target, } if self.valves.API_KEY: payload["api_key"] = self.valves.API_KEY response = requests.post( f"{self.valves.LIBRE_TRANSLATE_URL}/translate", json=payload, timeout=30, # Longer timeout for translation ) if response.status_code != 200: error_msg = ( f"Translation error: {response.status_code} - {response.text}" ) logger.error(error_msg) raise ConnectionError(error_msg) return response.json() try: return do_translate() except Exception as e: logger.error( f"Translation failed after {self.valves.MAX_RETRY_ATTEMPTS} attempts: {str(e)}" ) raise def format_language_name(self, language_code: str) -> str: """ Get the human-readable name for a language code. Args: language_code: The language code to get the name for Returns: A formatted string with the language name and code """ if not language_code: return "Unknown language" if language_code == "auto": return "Auto-detect" if language_code in SUPPORTED_LANGUAGES: return f"{SUPPORTED_LANGUAGES[language_code]} ({language_code})" # Check if we have data from the API for lang in self.api_languages: if lang.get("code") == language_code: return f"{lang.get('name')} ({language_code})" return f"Unknown language ({language_code})" def is_language_supported(self, language_code: str) -> bool: """ Check if a language is supported. Args: language_code: The language code to check Returns: True if the language is supported, False otherwise """ if not language_code: return False if language_code == "auto": return True # Check predefined list if language_code in SUPPORTED_LANGUAGES: return True # Check API data if available for lang in self.api_languages: if lang.get("code") == language_code: return True return False def validate_service_config(self) -> ValidationResult: """ Validate the LibreTranslate service configuration. Returns: ValidationResult: Result of the validation """ if not self.valves.LIBRE_TRANSLATE_URL: return ValidationResult(False, "LibreTranslate URL is not configured") return ValidationResult(True) def validate_languages( self, source_lang: str, target_langs: List[str] ) -> Tuple[ValidationResult, List[str], List[str]]: """ Validate source and target languages. Args: source_lang: Source language code target_langs: List of target language codes Returns: Tuple containing: - ValidationResult: Overall validation result - List[str]: List of valid target languages - List[str]: List of invalid target languages """ # Validate source language if source_lang != "auto" and not self.is_language_supported(source_lang): return ( ValidationResult( False, f"Source language '{source_lang}' is not supported" ), [], [], ) # Ensure target languages are provided if not target_langs or len(target_langs) == 0: return (ValidationResult(False, "No target languages specified"), [], []) # Filter out unsupported target languages valid_target_languages = [] invalid_languages = [] for lang in target_langs: if self.is_language_supported(lang): valid_target_languages.append(lang) else: invalid_languages.append(lang) if not valid_target_languages: return ( ValidationResult(False, "No valid target languages specified"), [], invalid_languages, ) return ( ValidationResult(True, "", valid_target_languages), valid_target_languages, invalid_languages, ) def get_assistant_message(self, messages: List[Dict[str, Any]]) -> ValidationResult: """ Extract the last assistant message from the message history. Args: messages: List of conversation messages Returns: ValidationResult: Contains the extracted message if successful """ if not messages: return ValidationResult(False, "No messages provided") try: assistant_message = get_last_assistant_message(messages) if not assistant_message: return ValidationResult( False, "No assistant message found to translate" ) if not isinstance(assistant_message, str) or not assistant_message.strip(): return ValidationResult(False, "Assistant message is empty or invalid") return ValidationResult(True, "", assistant_message) except Exception as e: logger.error(f"Error retrieving assistant message: {str(e)}") return ValidationResult( False, f"Error retrieving assistant message: {str(e)}" ) async def translate_to_language( self, text: str, source: str, target: str, emitter: Callable[[Dict[str, Any]], Coroutine[Any, Any, None]] = None, ) -> Tuple[bool, Optional[str]]: """ Translate text to a specific language and emit results. Args: text: The text to translate source: The source language code target: The target language code emitter: Callable function that emits events Returns: Tuple containing: - bool: Whether the translation was successful - Optional[str]: Error message if translation failed, None otherwise """ target_name = self.format_language_name(target) try: # Update status if emitter: await self.send_progress_status( emitter, f"Translating to {target_name}..." ) # Perform translation data = await self.translate_text(text, source, target) # Format the translated text with language header content = f"\n--- {target_name} -------------------------\n{data['translatedText']}" # Send the translated message if emitter: await self.send_message(emitter, content) await self.send_completion_status( emitter, f"Translation to {target_name} complete" ) logger.info(f"Successfully translated to {target}") return True, None except Exception as e: error_msg = f"Translation to {target} failed: {str(e)}" logger.error(error_msg) if emitter: await self.send_error_status( emitter, f"Error translating to {target_name}: {str(e)}" ) return False, error_msg async def action( self, body: dict, __user__=None, __request__: Request = None, __event_emitter__=None, __event_call__=None, ) -> Optional[dict]: """ Main action function that orchestrates the translation process. Args: body: Request body containing messages __user__: User information containing valves settings __request__: FastAPI request object __event_emitter__: Callable function for emitting events __event_call__: Event call information Returns: Optional dictionary with action results or None """ logger.info(f"Starting MultiLanguage LibreTranslate action") # Validate event emitter if not __event_emitter__: logger.error("Event emitter not provided") return # Validate service configuration service_validation = self.validate_service_config() if not service_validation.is_valid: await self.send_error_status( __event_emitter__, service_validation.error_message ) return # Validate user settings if not __user__: await self.send_error_status( __event_emitter__, "User settings not available" ) return # Get user-specific settings try: # The user settings are accessed as a dictionary, not as an object with attributes source = __user__["valves"].SOURCE_LANGUAGE target_languages = __user__["valves"].TARGET_LANGUAGES except (KeyError, AttributeError, TypeError) as e: logger.error(f"User settings structure error: {str(e)}") await self.send_error_status( __event_emitter__, f"Failed to retrieve user language settings: {str(e)}", ) return # Try to get languages from API for more accurate validation try: await self.get_supported_languages() except Exception as e: logger.warning(f"Could not retrieve supported languages from API: {str(e)}") # Continue with predefined languages only # Validate languages lang_validation, valid_target_languages, invalid_languages = ( self.validate_languages(source, target_languages) ) if not lang_validation.is_valid: await self.send_error_status( __event_emitter__, lang_validation.error_message ) return # Warn about invalid languages if invalid_languages: await self.send_warning_status( __event_emitter__, f"Skipping unsupported languages: {', '.join(invalid_languages)}", ) # Get the last assistant message message_validation = self.get_assistant_message(body.get("messages", [])) if not message_validation.is_valid: await self.send_error_status( __event_emitter__, message_validation.error_message ) return assistant_message = message_validation.result logger.info(f"Retrieved assistant message of length: {len(assistant_message)}") # Detect language if source is set to auto and detection is enabled actual_source = source if source == "auto" and self.valves.ENABLE_LANGUAGE_DETECTION: await self.send_progress_status( __event_emitter__, "Detecting source language..." ) try: detected_lang = await self.detect_language(assistant_message) if detected_lang != "auto": actual_source = detected_lang source_name = self.format_language_name(actual_source) await self.send_progress_status( __event_emitter__, f"Detected source language: {source_name}" ) logger.info(f"Detected source language: {actual_source}") # Optional wait time after detection for UI display purposes if self.valves.LANGUAGE_DETECTION_WAIT_SECONDS > 0: await asyncio.sleep(self.valves.LANGUAGE_DETECTION_WAIT_SECONDS) except Exception as e: logger.warning(f"Language detection failed: {str(e)}") # Continue with original source setting # Translate to all valid target languages if self.valves.PARALLEL_TRANSLATIONS: # Process translations in parallel await self.send_progress_status( __event_emitter__, f"Translating to {len(valid_target_languages)} languages in parallel...", ) tasks = [ self.translate_to_language( assistant_message, actual_source, target, __event_emitter__ ) for target in valid_target_languages ] # Gather results with error handling results = await asyncio.gather(*tasks) # Process results and collect errors successful = 0 errors = [] for i, (success, error_msg) in enumerate(results): if success: successful += 1 elif error_msg: errors.append(f"{valid_target_languages[i]}: {error_msg}") # Report final status status_message = ( f"Completed {successful}/{len(valid_target_languages)} translations" ) if errors: status_message += f" with {len(errors)} errors" logger.error(f"Translation errors: {'; '.join(errors)}") await self.send_completion_status(__event_emitter__, status_message) else: # Process translations sequentially successful = 0 errors = [] for target in valid_target_languages: success, error_msg = await self.translate_to_language( assistant_message, actual_source, target, __event_emitter__ ) if success: successful += 1 elif error_msg: errors.append(f"{target}: {error_msg}") # Report final status status_message = ( f"Completed {successful}/{len(valid_target_languages)} translations" ) if errors: status_message += f" with {len(errors)} errors" logger.error(f"Translation errors: {'; '.join(errors)}") await self.send_completion_status(__event_emitter__, status_message)