chickenflyshigh/onscreen-translator
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Updated README + reduced translation mismatch rates + LICENSE

Browse Source
This commit is contained in:
chickenflyshigh
2024-11-07 23:43:04 +11:00
parent 56d8c18871
commit 7e80713191
16 changed files with 742 additions and 268 deletions
Download Patch File Download Diff File Expand all files Collapse all files
helpers/batching.py
+46 -57
View File
@@ -168,27 +168,15 @@ class ApiModel():
return False
# async def request_func(request):
# @wraps(request)
# async def wrapper(self, text, *args, **kwargs):
# if await self._are_rates_good():
# try:
# self.session_calls += 1
# response = await request(self, text, *args, **kwargs)
# return response
# except Exception as e:
# logger.error(f"Error with model {self.model} from {self.site}. Error: {e}")
# else:
# logger.error(f"Rate limit reached for this model.")
# raise TooManyRequests('Rate limit reached for this model.')
# return wrapper
# @request_func
async def translate(self, texts_to_translate, store = False):
async def translate(self, texts_to_translate, store = False) -> tuple[int, # exit code: 0 for success, 1 for incorrect response type, 2 for incorrect translation count
list[str],
int # number of translations that do not match the number of texts to translate
]:
"""Main Translation Function. All API models will need to define a new class and also define a _request function as shown below in the Gemini and Groq class models."""
if isinstance(texts_to_translate, str):
texts_to_translate = [texts_to_translate]
if len(texts_to_translate) == 0:
return []
return (0, [], 0)
#prompt = f"Without any additional remarks, and without any code, translate the following items of the Python list from {self.from_lang} into {self.target_lang} and output as a Python list ensuring proper escaping of characters and ensuring the length of the list given is exactly equal to the length of the list you provide. Do not output in any other language other than the specified target language: {texts_to_translate}"
prompt = f"""INSTRUCTIONS:
- Provide ONE and ONLY ONE translation to each text provided in the JSON array given.
@@ -208,19 +196,27 @@ Expected format:
Translation:"""
response = await self._request(prompt)
response_list = ast.literal_eval(response.strip())
try:
response_list = ast.literal_eval(response.strip())
except Exception as e:
logger.error(f"Failed to evaluate response from {self.model} from {self.site}. Error: {e}. Response: {response}")
return (1, [], 99999)
logger.debug(repr(self))
logger.info(f'{self.model} translated texts from: {texts_to_translate} to {response_list}.')
if not isinstance(response_list, list):
raise TypeError(f"Incorrect response type. Expected list, got {type(response_list)}")
# raise TypeError(f"Incorrect response type. Expected list, got {type(response_list)}")
logger.error(f"Incorrect response type. Expected list, got {type(response_list)}")
return (1, [], 99999)
if len(response_list) != len(texts_to_translate) and len(texts_to_translate) <= MAX_TRANSLATE:
logger.error(f"{self.model} model failed to translate all the texts. Number of translations to make: {len(texts_to_translate)}; Number of translated texts: {len(response_list)}.")
logger.error(f"Number of translations does not match number of texts to translate. Sent: {len(texts_to_translate)}. Received: {len(response_list)}.")
if store:
self._db_add_translation(texts_to_translate, response_list, mismatch=True)
# raise ValueError(f"Number of translations does not match number of texts to translate. Sent: {len(texts_to_translate)}. Received: {len(response_list)}.")
return (2, response_list, abs(len(texts_to_translate) - len(response_list)))
else:
if store:
self._db_add_translation(texts_to_translate, response_list)
return response_list
return (0, response_list, 0)
class Groq(ApiModel):
def __init__(self, # model name as defined by the API
@@ -248,21 +244,7 @@ class Groq(ApiModel):
response_json = await response.json()
return response_json["choices"][0]["message"]["content"]
# https://console.groq.com/settings/limits for limits
# def request(self, content):
# chat_completion = self.client.chat.completions.create(
# messages=[
# {
# "role": "user",
# "content": content,
# }
# ],
# model=self.model
# )
# return chat_completion.choices[0].message.content
# async def translate(self, texts_to_translate):
# return super().translate(self.request, texts_to_translate)
class Gemini(ApiModel):
def __init__(self, # model name as defined by the API
model,
@@ -273,20 +255,6 @@ class Gemini(ApiModel):
site = 'Google',
**kwargs)
# def request(self, content):
# genai.configure(api_key=self.api_key)
# safety_settings = {
# "HARM_CATEGORY_HARASSMENT": "BLOCK_NONE",
# "HARM_CATEGORY_HATE_SPEECH": "BLOCK_NONE",
# "HARM_CATEGORY_SEXUALLY_EXPLICIT": "BLOCK_NONE",
# "HARM_CATEGORY_DANGEROUS_CONTENT": "BLOCK_NONE"}
# try:
# response = genai.GenerativeModel(self.model).generate_content(content, safety_settings=safety_settings)
# except ResourceExhausted as e:
# logger.error(f"Rate limited with {self.model}. Error: {e}")
# raise ResourceExhausted("Rate limited.")
# return response.text.strip()
async def _request(self, content):
async with aiohttp.ClientSession() as session:
async with session.post(
@@ -307,14 +275,35 @@ class Gemini(ApiModel):
) as response:
response_json = await response.json()
return response_json['candidates'][0]['content']['parts'][0]['text']
# async def translate(self, texts_to_translate):
# return super().translate(self.request, texts_to_translate)
"""
DEFINE YOUR OWN API MODELS BELOW WITH THE SAME TEMPLATE AS BELOW. All fields required are indicated by <required field>.
class <NameOfWebsite>(ApiModel):
def __init__(self, # model name as defined by the API
model,
api_key = <API_KEY>, # api key for the model wrt the site
**kwargs):
super().__init__(model,
api_key = api_key,
site = <name_of_website>,
**kwargs)
async def _request(self, content):
async with aiohttp.ClientSession() as session:
async with session.post(
<API ENDPOINT e.g. https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash:generateContent?key={self.api_key}>,
headers={
"Content-Type": "application/json"
<ANY OTHER HEADERS REQUIRED BY THE API separated by commas>
},
json={
"contents": [{"parts": [{"text": content}]}]
<ANY OTHER JSON PAIRS REQUIRED separated by commas>
}
) as response:
response_json = await response.json()
return <Anything needed to extract the message response from `response_json`>
"""
###################################################################################################
helpers/translation.py
+34 -31
View File
@@ -29,6 +29,7 @@ def translate(translation_func):
###############################
def init_API_LLM(from_lang, target_lang):
"""Initialise the API models. The models are stored in a json file. The models are instantiated, added to database/database api rates are updated and the languages are set."""
from_lang = standardize_lang(from_lang)['translation_model_lang']
target_lang = standardize_lang(target_lang)['translation_model_lang']
with open(API_MODELS_FILEPATH, 'r') as f:
@@ -45,19 +46,19 @@ def init_API_LLM(from_lang, target_lang):
async def translate_API_LLM(texts_to_translate: List[str],
models: List[ApiModel],
call_size: int = 2,
stagger_delay: int = 2) -> List[str]:
call_size: int = 2) -> List[str]:
"""Translate the texts using the models three at a time. If the models fail to translate the text, it will try the next model in the list."""
async def try_translate(model: ApiModel) -> Optional[List[str]]:
try:
result = await model.translate(texts_to_translate, store=True)
logger.debug(f'Try_translate result: {result}')
return result
except Exception as e:
logger.error(f"Translation failed for {model.model} from {model.site}: {e}")
return None
result = await model.translate(texts_to_translate, store=True)
logger.debug(f'Try_translate result: {result}')
return result
random.shuffle(models)
groups = [models[i:i+call_size] for i in range(0, len(models), call_size)]
no_of_models = len(models)
translation_attempts = 0
best_translation = None # (model, translation_errors)
for group in groups:
tasks = set(asyncio.create_task(try_translate(model)) for model in group)
while tasks:
@@ -70,28 +71,30 @@ async def translate_API_LLM(texts_to_translate: List[str],
result = await task
logger.debug(f'Result: {result}')
if result is not None:
# Cancel remaining tasks
for t in pending:
t.cancel()
return result
logger.error("All models have failed to translate the text.")
raise TypeError("Models have likely all outputted garbage translations or rate limited.")
# def translate_API_LLM(text, models):
# random.shuffle(models)
# logger.debug(f"All Models Available: {models}")
# for model in models:
# logger.info(f"Attempting translation with model {model}.")
# try:
# translation = model.translate(text)
# logger.debug(f"Translation obtained: {translation}")
# if translation or translation == []:
# return translation
# except Exception as e:
# logger.error(f"Error with model {repr(model)}. Error: {e}")
# continue
# logger.error("All models have failed to translate the text.")
# raise TypeError("Models have likely all outputted garbage translations or rate limited.")
translation_attempts += 1
status_code, translations, translation_mismatches = result
if status_code == 0:
# Cancel remaining tasks
for t in pending:
t.cancel()
return translations
else:
logger.error(f"Model has failed to translate the text.")
if translation_attempts == no_of_models:
if best_translation is not None:
return translations
else:
logger.error("All models have failed to translate the text.")
raise TypeError("Models have likely all outputted garbage translations or rate limited.")
elif status_code == 1:
if best_translation is None:
best_translation = (translations, translation_mismatches)
else:
best_translation = (translations, translation_mismatches) if len(result[2]) < len(best_translation[1]) else best_translation
else:
continue
###############################
# Best model by far. Aya-23-8B. Gemma is relatively good. If I get the time to quantize either gemma or aya those will be good to use. llama3.2 is really good as well.
def init_AYA():
helpers/utils.py
+11 -1
View File
@@ -176,8 +176,18 @@ def similar_tfidf(list1,list2,threshold) -> float:
return float(cosine_similarity(vec1, vec2)[0, 0]) > threshold
def similar_jacard(list1, list2) -> float:
if not list1 or not list2:
return 0.0
return len(set(list1).intersection(set(list2))) / len(set(list1).union(set(list2)))
def check_similarity(list1, list2, threshold, method = 'tfidf'):
if method == 'tfidf':
return True if similar_tfidf(list1, list2) >= threshold else False
elif method == 'jacard':
return True if similar_jacard(list1, list2) >= threshold else False
else:
raise ValueError("Invalid method. Please use one of 'tfidf' or 'jacard'.")
if __name__ == "__main__":
# Example usage