"""
title: LibreTranslate Filter
description: Translates messages between users and assistants in a chat system using the LibreTranslate API.
author: @iamg30
author_url: https://openwebui.com/u/iamg30/
funding_url: https://github.com/open-webui
version: 0.2.1
license: MIT
"""
from typing import List, Optional, Callable, Awaitable
from pydantic import BaseModel, Field
import requests
import os
from utils.misc import get_last_user_message, get_last_assistant_message
class Filter:
class Valves(BaseModel):
priority: int = Field(
default=0, description="Priority level for the filter operations."
)
libretranslate_url: str = Field(
default="http://localhost:5000",
description="URL for the LibreTranslate API",
)
source_user: str = Field(
default="auto", description="Default source language for user messages"
)
target_user: str = Field(
default="en", description="Default target language for user messages"
)
source_assistant: str = Field(
default="en", description="Default source language for assistant messages"
)
target_assistant: str = Field(
default="es", description="Default target language for assistant messages"
)
DISPLAY_EVENT_EMITTERS: bool = Field(
default=False,
description="Whether to display event emitters during translation.",
)
class UserValves(BaseModel):
source_user: str = Field(
default="auto",
description="User-specific source language for user messages",
)
target_user: str = Field(
default="en", description="User-specific target language for user messages"
)
source_assistant: str = Field(
default="en",
description="User-specific source language for assistant messages",
)
target_assistant: str = Field(
default="es",
description="User-specific target language for assistant messages",
)
DISPLAY_EVENT_EMITTERS: bool = Field(
default=False,
description="Whether to display event emitters during translation (user-specific setting).",
)
def __init__(self):
self.file_handler = False
self.valves = self.Valves()
async def translate(
self,
text: str,
source: str,
target: str,
__event_emitter__: Callable[[dict], Awaitable[None]],
display_event_emitters: bool,
) -> str:
payload = {
"q": text,
"source": source,
"target": target,
}
if display_event_emitters:
await __event_emitter__(
{
"type": "status",
"data": {
"status": "in_progress",
"description": f"Translating text from {source} to {target}",
"done": False,
},
}
)
try:
r = requests.post(
f"{self.valves.libretranslate_url}/translate", json=payload
)
r.raise_for_status()
data = r.json()
translated_text = data["translatedText"]
if display_event_emitters:
await __event_emitter__(
{
"type": "status",
"data": {
"status": "complete",
"description": f"Translation completed",
"done": True,
},
}
)
return translated_text
except Exception as e:
print(f"Error translating text: {e}")
if display_event_emitters:
await __event_emitter__(
{
"type": "status",
"data": {
"status": "error",
"description": f"Error translating text: {e}",
"done": True,
},
}
)
return text
async def inlet(
self,
body: dict,
__event_emitter__: Callable[[dict], Awaitable[None]],
__user__: dict = {},
) -> dict:
print(f"inlet:{__name__}")
# Get user_valves directly from __user__ dict
user_valves = __user__.get("valves")
if not user_valves:
user_valves = self.UserValves() # Use default UserValves if not provided
# Determine whether to display event emitters based on both global and user-specific settings
display_event_emitters = (
self.valves.DISPLAY_EVENT_EMITTERS and user_valves.DISPLAY_EVENT_EMITTERS
)
messages = body["messages"]
user_message = get_last_user_message(messages)
print(f"User message: {user_message}")
source_user = user_valves.source_user or self.valves.source_user
target_user = user_valves.target_user or self.valves.target_user
translated_user_message = await self.translate(
user_message,
source_user,
target_user,
__event_emitter__,
display_event_emitters,
)
print(f"Translated user message: {translated_user_message}")
for message in reversed(messages):
if message["role"] == "user":
message["content"] = translated_user_message
break
body = {**body, "messages": messages}
return body
async def outlet(
self,
body: dict,
__event_emitter__: Callable[[dict], Awaitable[None]],
__user__: dict = {},
) -> dict:
print(f"outlet:{__name__}")
# Get user_valves directly from __user__ dict
user_valves = __user__.get("valves")
if not user_valves:
user_valves = self.UserValves() # Use default UserValves if not provided
# Determine whether to display event emitters based on both global and user-specific settings
display_event_emitters = (
self.valves.DISPLAY_EVENT_EMITTERS and user_valves.DISPLAY_EVENT_EMITTERS
)
messages = body["messages"]
assistant_message = get_last_assistant_message(messages)
print(f"Assistant message: {assistant_message}")
source_assistant = user_valves.source_assistant or self.valves.source_assistant
target_assistant = user_valves.target_assistant or self.valves.target_assistant
translated_assistant_message = await self.translate(
assistant_message,
source_assistant,
target_assistant,
__event_emitter__,
display_event_emitters,
)
print(f"Translated assistant message: {translated_assistant_message}")
for message in reversed(messages):
if message["role"] == "assistant":
message["content"] = translated_assistant_message
break
body = {**body, "messages": messages}
return body