from transformers import T5Tokenizer, T5ForConditionalGeneration, AutoTokenizer, AutoModelForSeq2SeqLM, pipeline, logging, AutoModelForCausalLM
import httpx
import os
import torch
import polars as pl
import languagecodes
# Language options and mappings
df = pl.read_parquet("isolanguages.parquet")
non_empty_isos = df.slice(1).filter(pl.col("ISO639-1") != "").rows()
all_langs = {iso[0]: (iso[1], iso[2], iso[3]) for iso in non_empty_isos} # {'Romanian': ('ro', 'rum', 'ron')}
iso1toall = {iso[1]: (iso[0], iso[2], iso[3]) for iso in non_empty_isos} # {'ro': ('Romanian', 'rum', 'ron')}
class Translators:
MODELS = ["Helsinki-NLP", "QUICKMT", "Argos", "HPLT", "HPLT-OPUS", "Google",
"google/translategemma-4b-it", "nvidia/Riva-Translate-4B-Instruct-v1.1",
"xiaomi-research/GemmaX2-28-2B-v0.2", "xiaomi-research/MiLMMT-46-1B-v0.1",
"yanolja/YanoljaNEXT-Rosetta-4B-2511", "yanolja/YanoljaNEXT-Rosetta-4B-2510", "yanolja/YanoljaNEXT-Rosetta-4B",
"Helsinki-NLP/opus-mt-tc-bible-big-mul-mul", "Helsinki-NLP/opus-mt-tc-bible-big-mul-deu_eng_nld",
"Helsinki-NLP/opus-mt-tc-bible-big-mul-deu_eng_fra_por_spa", "Helsinki-NLP/opus-mt-tc-bible-big-deu_eng_fra_por_spa-mul",
"Helsinki-NLP/opus-mt-tc-bible-big-roa-deu_eng_fra_por_spa", "Helsinki-NLP/opus-mt-tc-bible-big-deu_eng_fra_por_spa-roa", "Helsinki-NLP/opus-mt-tc-bible-big-roa-en",
"facebook/nllb-200-distilled-600M", "facebook/nllb-200-distilled-1.3B", "facebook/nllb-200-1.3B", "facebook/nllb-200-3.3B",
"facebook/mbart-large-50-many-to-many-mmt", "facebook/mbart-large-50-one-to-many-mmt", "facebook/mbart-large-50-many-to-one-mmt",
"facebook/hf-seamless-m4t-medium", "facebook/seamless-m4t-large", "facebook/seamless-m4t-v2-large",
"facebook/m2m100_418M", "facebook/m2m100_1.2B",
"alirezamsh/small100", "naist-nlp/mitre_466m", "naist-nlp/mitre_913m",
"bigscience/mt0-small", "bigscience/mt0-base", "bigscience/mt0-large", "bigscience/mt0-xl",
"bigscience/bloomz-560m", "bigscience/bloomz-1b1", "bigscience/bloomz-1b7", "bigscience/bloomz-3b",
"google/madlad400-3b-mt", "jbochi/madlad400-3b-mt",
"NiuTrans/LMT-60-0.6B", "NiuTrans/LMT-60-1.7B", "NiuTrans/LMT-60-4B",
"Lego-MT/Lego-MT", "BSC-LT/salamandraTA-2b-instruct",
"winninghealth/WiNGPT-Babel", "winninghealth/WiNGPT-Babel-2", "winninghealth/WiNGPT-Babel-2.1",
"Unbabel/Tower-Plus-2B",
"utter-project/EuroLLM-1.7B", "utter-project/EuroLLM-1.7B-Instruct", "utter-project/EuroMoE-2.6B-A0.6B-Instruct-2512",
"google-t5/t5-small", "google-t5/t5-base", "google-t5/t5-large",
"google/flan-t5-small", "google/flan-t5-base", "google/flan-t5-large", "google/flan-t5-xl"]
def __init__(self, model_name: str, sl: str, tl: str, input_text: str):
self.model_name = model_name
self.sl, self.tl = sl, tl
self.input_text = input_text
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
self.max_new_tokens = 512
def google(self):
self.input_text = " ".join(self.input_text.split())
url = os.environ['GCLIENT'] + f'sl={self.sl}&tl={self.tl}&q={self.input_text}'
response = httpx.get(url)
return response.json()[0][0][0]
def riva4B11(self):
language_pairs = {
'en-zh-cn': {'source': 'English', 'target': 'Simplified Chinese'},
'en-zh': {'source': 'English', 'target': 'Simplified Chinese'},
'en-zh-tw': {'source': 'English', 'target': 'Traditional Chinese'},
'en-ar': {'source': 'English', 'target': 'Arabic'},
'en-de': {'source': 'English', 'target': 'German'},
'en-es': {'source': 'English', 'target': 'European Spanish'},
'en-es-es': {'source': 'English', 'target': 'European Spanish'},
'en-es-us': {'source': 'English', 'target': 'Latin American Spanish'},
'en-fr': {'source': 'English', 'target': 'French'},
'en-ja': {'source': 'English', 'target': 'Japanese'},
'en-ko': {'source': 'English', 'target': 'Korean'},
'en-ru': {'source': 'English', 'target': 'Russian'},
'en-pt': {'source': 'English', 'target': 'Brazilian Portuguese'},
'en-pt-br': {'source': 'English', 'target': 'Brazilian Portuguese'},
'zh-en': {'source': 'Simplified Chinese', 'target': 'English'},
'zh-cn-en': {'source': 'Simplified Chinese', 'target': 'English'},
'zh-tw-en': {'source': 'Traditional Chinese', 'target': 'English'},
'ar-en': {'source': 'Arabic', 'target': 'English'},
'de-en': {'source': 'German', 'target': 'English'},
'es-en': {'source': 'European Spanish', 'target': 'English'},
'es-es-en': {'source': 'European Spanish', 'target': 'English'},
'es-us-en': {'source': 'Latin American Spanish', 'target': 'English'},
'fr-en': {'source': 'French', 'target': 'English'},
'ja-en': {'source': 'Japanese', 'target': 'English'},
'ko-en': {'source': 'Korean', 'target': 'English'},
'ru-en': {'source': 'Russian', 'target': 'English'},
'pt-en': {'source': 'Brazilian Portuguese', 'target': 'English'},
'pt-br-en': {'source': 'Brazilian Portuguese', 'target': 'English'},
}
tokenizer = AutoTokenizer.from_pretrained(self.model_name)
model = AutoModelForCausalLM.from_pretrained(self.model_name,
torch_dtype="auto", device_map="auto")
# Use the prompt template (along with chat template)
messages = [{
"role": "system",
"content": f"{self.sl}-{self.tl}",
},
{"role": "user", "content": self.input_text}]
tokenized_chat = tokenizer.apply_chat_template(messages, tokenize=True, add_generation_prompt=True, return_tensors="pt").to(model.device)
outputs = model.generate(tokenized_chat, max_new_tokens=self.max_new_tokens, pad_token_id=tokenizer.eos_token_id)
return tokenizer.decode(outputs[0]).split('Assistant')[1].strip().removesuffix('')
def xiaomi(self):
model = AutoModelForCausalLM.from_pretrained(self.model_name)
tokenizer = AutoTokenizer.from_pretrained(self.model_name)
text = f"Translate this from {self.sl} to {self.tl}:\n{self.sl}: {self.input_text}\n{self.tl}:"
inputs = tokenizer(text, add_special_tokens=False, return_tensors="pt")
outputs = model.generate(**inputs, max_new_tokens=self.max_new_tokens)
return tokenizer.decode(outputs[0], skip_special_tokens=True).rsplit(f'{self.tl}:')[-1].strip()
def translategemma(self):
from huggingface_hub import login
hftoken=os.environ.get("HF_TOKEN")
login(token=hftoken)
pipe = pipeline(
"image-text-to-text",
model = self.model_name,
device = self.device,
dtype = torch.bfloat16,
token=hftoken)
# ---- Text Translation ----
messages = [
{
"role": "user",
"content": [
{
"type": "text",
"source_lang_code": self.sl,
"target_lang_code": self.tl,
"text": self.input_text,
}
],
}
]
output = pipe(text=messages, max_new_tokens=self.max_new_tokens)
return output[0]["generated_text"][-1]["content"]
def simplepipe(self):
try:
pipe = pipeline("translation", model=self.model_name, device=self.device)
translation = pipe(self.input_text)
message = f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with {self.model_name}.'
return translation[0]['translation_text'], message
except Exception as error:
return f"Error translating with model: {self.model_name}! Try other available language combination or model.", error
def mitre(self):
from transformers import AutoModel, AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained(self.model_name, trust_remote_code=True, use_fast=False)
model = AutoModel.from_pretrained(self.model_name, trust_remote_code=True).to(self.device)
model.eval()
# Translating from one or several sentences to a sole language
src_tokens = tokenizer.encode_source_tokens_to_input_ids(self.input_text, target_language=self.tl)
with torch.inference_mode(): # no_grad inference_mode
generated_tokens = model.generate(src_tokens)
result = tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]
return result
def rosetta(self):
model = AutoModelForCausalLM.from_pretrained(
self.model_name,
dtype=torch.bfloat16, # float32 slow
low_cpu_mem_usage=False, # True
device_map="auto")
tokenizer = AutoTokenizer.from_pretrained(self.model_name)
system = f"Translate the user's text to {self.tl}. Provide the final translation in a formal tone immediately immediately without any other text."
messages = [
{"role": "system", "content": system},
{"role": "user", "content": self.input_text},
]
prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = tokenizer(prompt, return_tensors="pt").to(self.device)
input_length = inputs["input_ids"].shape[1]
model.eval()
with torch.inference_mode():
outputs = model.generate(
**inputs,
max_new_tokens=self.max_new_tokens,
)
generated_tokens = outputs[0][input_length:]
translation = tokenizer.decode(generated_tokens, skip_special_tokens=True)
return translation
def niutrans(self):
tokenizer = AutoTokenizer.from_pretrained(self.model_name, padding_side='left')
model = AutoModelForCausalLM.from_pretrained(self.model_name)
prompt = f"Translate the following text from {self.sl} into {self.tl}.\n{self.sl}: {self.input_text}.\n{self.tl}: "
messages = [{"role": "user", "content": prompt}]
text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True,
)
model_inputs = tokenizer([text], return_tensors="pt").to(model.device)
generated_ids = model.generate(**model_inputs, max_new_tokens=512, num_beams=5, do_sample=False)
output_ids = generated_ids[0][len(model_inputs.input_ids[0]):].tolist()
outputs = tokenizer.batch_decode(output_ids, skip_special_tokens=True)
outputs = ''.join(outputs) if isinstance(outputs, list) else outputs
return outputs
def salamandratapipe(self):
pipe = pipeline("text-generation", model=self.model_name)
messages = [{"role": "user", "content": f"Translate the following text from {self.sl} into {self.tl}.\n{self.sl}: {self.input_text} \n{self.tl}:"}]
return pipe(messages, max_new_tokens=self.max_new_tokens, early_stopping=True, num_beams=5)[0]["generated_text"][1]["content"]
def hplt(self, opus = False):
# langs = ['ar', 'bs', 'ca', 'en', 'et', 'eu', 'fi', 'ga', 'gl', 'hi', 'hr', 'is', 'mt', 'nn', 'sq', 'sw', 'zh_hant']
hplt_models = ['ar-en', 'bs-en', 'ca-en', 'en-ar', 'en-bs', 'en-ca', 'en-et', 'en-eu', 'en-fi',
'en-ga', 'en-gl', 'en-hi', 'en-hr', 'en-is', 'en-mt', 'en-nn', 'en-sq', 'en-sw',
'en-zh_hant', 'et-en', 'eu-en', 'fi-en', 'ga-en', 'gl-en', 'hi-en', 'hr-en',
'is-en', 'mt-en', 'nn-en', 'sq-en', 'sw-en', 'zh_hant-en']
lang_map = {"zh": "zh_hant"}
self.sl = lang_map.get(self.sl, self.sl)
self.tl = lang_map.get(self.tl, self.tl)
if opus:
hplt_model = f'HPLT/translate-{self.sl}-{self.tl}-v1.0-hplt_opus' # HPLT/translate-en-hr-v1.0-hplt_opus
else:
hplt_model = f'HPLT/translate-{self.sl}-{self.tl}-v1.0-hplt' # HPLT/translate-en-hr-v1.0-hplt
if f'{self.sl}-{self.tl}' in hplt_models:
pipe = pipeline("translation", model=hplt_model, device=self.device)
translation = pipe(self.input_text)
translated_text = translation[0]['translation_text']
message_text = f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with {hplt_model}.'
else:
translated_text = f'HPLT model from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} not available!'
message_text = f"Available models: {', '.join(hplt_models)}"
return translated_text, message_text
@staticmethod
def download_argos_model(available_packages, from_code, to_code):
import argostranslate.package
print('Downloading model for', from_code, to_code)
# Download and install Argos Translate package from path
package_to_install = next(
filter(lambda x: x.from_code == from_code and x.to_code == to_code, available_packages)
)
argostranslate.package.install_from_path(package_to_install.download())
def argos(self):
import argostranslate.translate, argostranslate.package
argostranslate.package.update_package_index()
available_packages = argostranslate.package.get_available_packages()
available_slanguages = [lang.from_code for lang in available_packages]
available_tlanguages = [lang.to_code for lang in available_packages]
available_languages = sorted(list(set(available_slanguages + available_tlanguages)))
combos: tuple[str|str] = sorted(list(zip(available_slanguages, available_tlanguages)))
packages_info = ', '.join(f"{pkg.from_name} ({pkg.from_code}) -> {pkg.to_name} ({pkg.to_code})" for pkg in available_packages)
# print(available_languages, combos, packages_info)
if self.sl not in available_languages and self.tl not in available_languages:
translated_text = f'''No supported Argos model available from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]}!
Try other model or languages combination from the available Argos models: {', '.join(available_languages)}.'''
else:
try:
if (self.sl, self.tl) in combos:
self.__class__.download_argos_model(available_packages, self.sl, self.tl) # Download model
translated_text = argostranslate.translate.translate(self.input_text, self.sl, self.tl) # Direct translation
elif (self.sl, 'en') in combos and ('en', self.tl) in combos:
self.__class__.download_argos_model(available_packages, self.sl, 'en') # Download model
translated_pivottext = argostranslate.translate.translate(self.input_text, self.sl, 'en') # Translate to pivot language English
self.__class__.download_argos_model(available_packages, 'en', self.tl) # Download model
translated_text = argostranslate.translate.translate(translated_pivottext, 'en', self.tl) # Translate from pivot language English
message = f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with Argos using pivot language English.'
else:
translated_text = f"No Argos model for {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]}. Try other model or languages combination from the available Argos models: {packages_info}."
except StopIteration as IterationError:
# packages_info = ', '.join(f"{pkg.get_description()}->{str(pkg.links)} {str(pkg.source_languages)}" for pkg in available_packages)
translated_text = f"No Argos model for {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]}. Error: {IterationError}. Try other model or languages combination from the available Argos models: {packages_info}."
except Exception as generalerror:
translated_text = f"General error: {generalerror}"
return translated_text
@staticmethod
def quickmttranslate(model_path, input_text):
# 'auto' auto-detects GPU, set to "cpu" to force CPU inference
# device = 'gpu' if torch.cuda.is_available() else 'cpu'
from quickmt import Translator
from huggingface_hub import snapshot_download
# Download Model (if not downloaded already) and return path to local model
# Device is either 'auto', 'cpu' or 'cuda'
translator = Translator(
snapshot_download(str(model_path), ignore_patterns="eole-model/*"),
device="auto", compute_type="auto"
)
# Translate - set beam size to 5 for higher quality (but slower speed)
# set beam size to 1 for faster speed (but lower quality) device="auto/cpu/gpu"
translation = translator(input_text, beam_size=5, max_input_length = 512, max_decoding_length = 512)
# translation = Translator(f"./quickmt-{self.sl}-{self.tl}/", device="auto/cpu", intra_threads=2, inter_threads=2, compute_type="int8")
# ctranslate2._ext.Translator(model_path: str, device: str = 'cpu', *, device_index: Union[int, List[int]] = 0, compute_type: Union[str, Dict[str, str]] = 'default',
# inter_threads: int = 1, intra_threads: int = 0, max_queued_batches: int = 0, flash_attention: bool = False, tensor_parallel: bool = False, files: object = None)
# Options for compute_type: default, auto, int8, int8_float32, int8_float16, int8_bfloat16, int16, float16, bfloat16, float32
# "int8" will work well for inference on CPU and give "int8_float16" or "int8_bfloat16" a try for GPU inference.
# (self: ctranslate2._ext.Translator, source: List[List[str]], target_prefix: Optional[List[Optional[List[str]]]] = None, *, max_batch_size: int = 0,
# batch_type: str = 'examples', asynchronous: bool = False, beam_size: int = 2, patience: float = 1, num_hypotheses: int = 1, length_penalty: float = 1,
# coverage_penalty: float = 0, repetition_penalty: float = 1, no_repeat_ngram_size: int = 0, disable_unk: bool = False,
# suppress_sequences: Optional[List[List[str]]] = None, end_token: Optional[Union[str, List[str], List[int]]] = None, return_end_token: bool = False,
# prefix_bias_beta: float = 0, max_input_length: int = 1024, max_decoding_length: int = 256, min_decoding_length: int = 1, use_vmap: bool = False,
# return_scores: bool = False, return_logits_vocab: bool = False, return_attention: bool = False, return_alternatives: bool = False,
# min_alternative_expansion_prob: float = 0, sampling_topk: int = 1, sampling_topp: float = 1, sampling_temperature: float = 1, replace_unknowns: bool = False,
# callback: Callable[[ctranslate2._ext.GenerationStepResult], bool] = None) -> Union[List[ctranslate2._ext.TranslationResult], List[ctranslate2._ext.AsyncTranslationResult]]
# print(model_path, input_text, translation)
return translation
@staticmethod
def list_usermodels(user):
from huggingface_hub import list_models
listmodels = list_models(author=user)
collection_ids = [id.id for id in listmodels]
return collection_ids
def quickmt(self):
model_name = f"quickmt-{self.sl}-{self.tl}"
model_path = f"quickmt/quickmt-{self.sl}-{self.tl}"
# quickmt_models = list_usermodels('quickmt')
# quickmt_models = [_.split('quickmt/quickmt-')[1] for _ in quickmt_models]
# quickmt_models.sort()
quickmt_models = ['zh-en', 'en-zh', 'fr-en', 'en-fr', 'de-en', 'en-de', 'ko-en', 'en-ko',
'en-ar', 'ar-en', 'ja-en', 'en-ja', 'it-en', 'en-it', 'hi-en', 'en-hi',
'en-es', 'es-en', 'ru-en', 'en-ru', 'pt-en', 'en-pt', 'bn-en', 'en-bn',
'id-en', 'en-id', 'fa-en', 'en-fa', 'vi-en', 'en-vi', 'tr-en', 'en-tr',
'ur-en', 'en-ur', 'en-hu', 'hu-en', 'th-en', 'en-th', 'el-en', 'en-el',
'pl-en', 'en-pl', 'en-lv', 'lv-en', 'cs-en', 'ro-en', 'en-ro', 'en-cs',
'he-en', 'en-he', 'da-en', 'en-da', 'en-sv', 'en-is', 'sv-en', 'is-en']
# available_languages = list(set([lang for model in quickmt_models for lang in model.split('-')]))
# available_languages.sort()
available_languages = ['ar', 'bn', 'cs', 'da', 'de', 'el', 'en', 'es', 'fa', 'fr', 'he', 'hi','hu',
'id', 'is', 'it', 'ja', 'ko', 'lv', 'pl', 'pt', 'ro', 'ru', 'sv', 'th', 'tr', 'ur', 'vi', 'zh']
# print(quickmt_models, available_languages)
# Direct translation model
if f"{self.sl}-{self.tl}" in quickmt_models:
translated_text = Translators.quickmttranslate(model_path, self.input_text)
message = f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with {model_name}.'
# Pivot language English
elif self.sl in available_languages and self.tl in available_languages:
model_name = f"quickmt-{self.sl}-en"
model_path = f"quickmt/quickmt-{self.sl}-en"
entranslation = Translators.quickmttranslate(model_path, self.input_text)
model_name = f"quickmt-en-{self.tl}"
model_path = f"quickmt/quickmt-en-{self.tl}"
translated_text = Translators.quickmttranslate(model_path, entranslation)
message = f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with Quickmt using pivot language English.'
else:
translated_text = f'No Quickmt model available for translation from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]}!'
message = f"Available models: {', '.join(quickmt_models)}"
return translated_text, message
def HelsinkiNLP_mulroa(self):
try:
pipe = pipeline("translation", model=self.model_name, device=self.device)
tgt_lang = iso1toall.get(self.tl)[2] # 'deu', 'ron', 'eng', 'fra'
translation = pipe(f'>>{tgt_lang}<< {self.input_text}')
return translation[0]['translation_text'], f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with {self.model_name}.'
except Exception as error:
return f"Error translating with model: {self.model_name}! Try other available language combination.", error
def HelsinkiNLP(self):
try: # Standard bilingual model
model_name = f"Helsinki-NLP/opus-mt-{self.sl}-{self.tl}"
pipe = pipeline("translation", model=model_name, device=self.device)
translation = pipe(self.input_text)
return translation[0]['translation_text'], f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with {model_name}.'
except EnvironmentError:
try: # Tatoeba models
model_name = f"Helsinki-NLP/opus-tatoeba-{self.sl}-{self.tl}"
pipe = pipeline("translation", model=model_name, device=self.device)
translation = pipe(self.input_text)
return translation[0]['translation_text'], f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with {model_name}.'
except EnvironmentError as error:
self.model_name = "Helsinki-NLP/opus-mt-tc-bible-big-mul-mul" # Last resort: try multi to multi
return self.HelsinkiNLP_mulroa()
except KeyError as error:
return f"Error: Translation direction {self.sl} to {self.tl} is not supported by Helsinki Translation Models", error
def madlad(self):
model = T5ForConditionalGeneration.from_pretrained(self.model_name, device_map="auto")
tokenizer = T5Tokenizer.from_pretrained(self.model_name)
text = f"<2{self.tl}> {self.input_text}"
# input_ids = tokenizer(text, return_tensors="pt").input_ids.to(model.device)
# outputs = model.generate(input_ids=input_ids, max_new_tokens=512)
# return tokenizer.decode(outputs[0], skip_special_tokens=True)
# return tokenizer.batch_decode(outputs, skip_special_tokens=True)
# Use a pipeline as a high-level helper
translator = pipeline('translation', model=model, tokenizer=tokenizer, src_lang=self.sl, tgt_lang=self.tl)
translated_text = translator(text, max_length=512)
return translated_text[0]['translation_text']
def flan(self):
tokenizer = T5Tokenizer.from_pretrained(self.model_name, legacy=False)
model = T5ForConditionalGeneration.from_pretrained(self.model_name)
prompt = f"translate {self.sl} to {self.tl}: {self.input_text}"
input_ids = tokenizer(prompt, return_tensors="pt").input_ids
outputs = model.generate(input_ids)
return tokenizer.decode(outputs[0], skip_special_tokens=True).strip()
def tfive(self):
tokenizer = T5Tokenizer.from_pretrained(self.model_name)
model = T5ForConditionalGeneration.from_pretrained(self.model_name, device_map="auto")
prompt = f"translate {self.sl} to {self.tl}: {self.input_text}"
input_ids = tokenizer.encode(prompt, return_tensors="pt")
output_ids = model.generate(input_ids, max_length=512)
translated_text = tokenizer.decode(output_ids[0], skip_special_tokens=True).strip()
return translated_text
def mbart_many_to_many(self):
from transformers import MBartForConditionalGeneration, MBart50TokenizerFast
model = MBartForConditionalGeneration.from_pretrained(self.model_name)
tokenizer = MBart50TokenizerFast.from_pretrained(self.model_name)
# translate source to target
tokenizer.src_lang = languagecodes.mbart_large_languages[self.sl]
encoded = tokenizer(self.input_text, return_tensors="pt")
generated_tokens = model.generate(
**encoded,
forced_bos_token_id=tokenizer.lang_code_to_id[languagecodes.mbart_large_languages[self.tl]]
)
return tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]
def mbart_one_to_many(self):
# translate from English
from transformers import MBartForConditionalGeneration, MBart50TokenizerFast
model = MBartForConditionalGeneration.from_pretrained(self.model_name, device_map="auto", torch_dtype="auto").to(self.device)
tokenizer = MBart50TokenizerFast.from_pretrained(self.model_name, src_lang="en_XX")
model_inputs = tokenizer(self.input_text, return_tensors="pt")
langid = languagecodes.mbart_large_languages[self.tl]
generated_tokens = model.generate(
**model_inputs,
forced_bos_token_id=tokenizer.lang_code_to_id[langid]
)
return tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]
def mbart_many_to_one(self):
# translate to English
from transformers import MBartForConditionalGeneration, MBart50TokenizerFast
model = MBartForConditionalGeneration.from_pretrained(self.model_name)
tokenizer = MBart50TokenizerFast.from_pretrained(self.model_name)
tokenizer.src_lang = languagecodes.mbart_large_languages[self.sl]
encoded = tokenizer(self.input_text, return_tensors="pt")
generated_tokens = model.generate(**encoded)
return tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]
def mtom(self):
from transformers import M2M100ForConditionalGeneration, M2M100Tokenizer
model = M2M100ForConditionalGeneration.from_pretrained(self.model_name)
tokenizer = M2M100Tokenizer.from_pretrained(self.model_name)
tokenizer.src_lang = self.sl
encoded = tokenizer(self.input_text, return_tensors="pt")
generated_tokens = model.generate(**encoded, forced_bos_token_id=tokenizer.get_lang_id(self.tl))
return tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]
def smallonehundred(self):
from transformers import M2M100ForConditionalGeneration
from tokenization_small100 import SMALL100Tokenizer
model = M2M100ForConditionalGeneration.from_pretrained(self.model_name)
tokenizer = SMALL100Tokenizer.from_pretrained(self.model_name)
tokenizer.tgt_lang = self.tl
encoded_sl = tokenizer(self.input_text, return_tensors="pt")
generated_tokens = model.generate(**encoded_sl, max_length=256, num_beams=5)
return tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]
def LegoMT(self):
from transformers import M2M100ForConditionalGeneration, M2M100Tokenizer
model = M2M100ForConditionalGeneration.from_pretrained(self.model_name) # "Lego-MT/Lego-MT"
tokenizer = M2M100Tokenizer.from_pretrained(self.model_name)
tokenizer.src_lang = self.sl
encoded = tokenizer(self.input_text, return_tensors="pt")
generated_tokens = model.generate(**encoded, forced_bos_token_id=tokenizer.get_lang_id(self.tl))
return tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]
def bigscience(self):
tokenizer = AutoTokenizer.from_pretrained(self.model_name)
model = AutoModelForSeq2SeqLM.from_pretrained(self.model_name)
self.input_text = self.input_text if self.input_text.endswith('.') else f'{self.input_text}.'
inputs = tokenizer.encode(f"Translate to {self.tl}: {self.input_text}", return_tensors="pt")
outputs = model.generate(inputs)
translation = tokenizer.decode(outputs[0])
translation = translation.replace(' ', '').replace('', '')
return translation
def bloomz(self):
tokenizer = AutoTokenizer.from_pretrained(self.model_name)
model = AutoModelForCausalLM.from_pretrained(self.model_name)
self.input_text = self.input_text if self.input_text.endswith('.') else f'{self.input_text}.'
# inputs = tokenizer.encode(f"Translate from {self.sl} to {self.tl}: {self.input_text} Translation:", return_tensors="pt")
inputs = tokenizer.encode(f"Translate to {self.tl}: {self.input_text}", return_tensors="pt")
outputs = model.generate(inputs)
translation = tokenizer.decode(outputs[0])
translation = translation.replace(' ', '').replace('', '')
translation = translation.split('Translation:')[-1].strip() if 'Translation:' in translation else translation.strip()
return translation
def nllb(self):
tokenizer = AutoTokenizer.from_pretrained(self.model_name, src_lang=self.sl)
# model = AutoModelForSeq2SeqLM.from_pretrained(self.model_name, device_map="auto", torch_dtype=torch.bfloat16)
model = AutoModelForSeq2SeqLM.from_pretrained(self.model_name)
translator = pipeline('translation', model=model, tokenizer=tokenizer, src_lang=self.sl, tgt_lang=self.tl)
translated_text = translator(self.input_text, max_length=512)
return translated_text[0]['translation_text']
def seamlessm4t1(self):
from transformers import AutoProcessor, SeamlessM4TModel
processor = AutoProcessor.from_pretrained(self.model_name)
model = SeamlessM4TModel.from_pretrained(self.model_name)
src_lang = iso1toall.get(self.sl)[2] # 'deu', 'ron', 'eng', 'fra'
tgt_lang = iso1toall.get(self.tl)[2]
text_inputs = processor(text = self.input_text, src_lang=src_lang, return_tensors="pt")
output_tokens = model.generate(**text_inputs, tgt_lang=tgt_lang, generate_speech=False)
return processor.decode(output_tokens[0].tolist()[0], skip_special_tokens=True)
def seamlessm4t2(self):
from transformers import AutoProcessor, SeamlessM4Tv2ForTextToText
processor = AutoProcessor.from_pretrained(self.model_name)
model = SeamlessM4Tv2ForTextToText.from_pretrained(self.model_name)
src_lang = iso1toall.get(self.sl)[2] # 'deu', 'ron', 'eng', 'fra'
tgt_lang = iso1toall.get(self.tl)[2]
text_inputs = processor(text=self.input_text, src_lang=src_lang, return_tensors="pt")
decoder_input_ids = model.generate(**text_inputs, tgt_lang=tgt_lang)[0].tolist()
return processor.decode(decoder_input_ids, skip_special_tokens=True)
def wingpt(self):
model = AutoModelForCausalLM.from_pretrained(
self.model_name,
torch_dtype="auto",
device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained(self.model_name)
# input_json = '{"input_text": self.input_text}'
messages = [
{"role": "system", "content": f"Translate this to {self.tl} language"},
{"role": "user", "content": self.input_text}
]
text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)
model_inputs = tokenizer([text], return_tensors="pt").to(model.device)
generated_ids = model.generate(
**model_inputs,
max_new_tokens=512,
temperature=0.1
)
generated_ids = [
output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
]
output = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
result = output.split('\n')[-1].strip() if '\n' in output else output.strip()
return result
def eurollm(self):
tokenizer = AutoTokenizer.from_pretrained(self.model_name)
model = AutoModelForCausalLM.from_pretrained(self.model_name)
prompt = f"{self.sl}: {self.input_text} {self.tl}:"
inputs = tokenizer(prompt, return_tensors="pt")
outputs = model.generate(**inputs, max_new_tokens=512)
output = tokenizer.decode(outputs[0], skip_special_tokens=True)
print(output)
# result = output.rsplit(f'{self.tl}:')[-1].strip() if f'{self.tl}:' in output else output.strip()
result = output.rsplit(f'{self.tl}:')[-1].strip() if '\n' in output or f'{self.tl}:' in output else output.strip()
return result
def EuroMoE(self):
tokenizer = AutoTokenizer.from_pretrained(self.model_name)
model = AutoModelForCausalLM.from_pretrained(self.model_name)
messages = [{
"role": "system",
"content": "You are a professional translating assistant generating formal, concise and accurate translations without any comments or additions.",
},
{
"role": "user", "content": f"Translate the following text from {self.sl} to {self.tl}: {self.input_text}"
} ]
inputs = tokenizer.apply_chat_template(messages, tokenize=True, add_generation_prompt=True, return_tensors="pt")
outputs = model.generate(inputs, max_new_tokens=self.max_new_tokens)
return tokenizer.decode(outputs[0], skip_special_tokens=True).rsplit('assistant\n')[-1].strip()
def eurollm_instruct(self):
tokenizer = AutoTokenizer.from_pretrained(self.model_name)
model = AutoModelForCausalLM.from_pretrained(self.model_name)
text = f'<|im_start|>system\n<|im_end|>\n<|im_start|>user\nTranslate the following {self.sl} source text to {self.tl}:\n{self.sl}: {self.input_text} \n{self.tl}: <|im_end|>\n<|im_start|>assistant\n'
inputs = tokenizer(text, return_tensors="pt")
outputs = model.generate(**inputs, max_new_tokens=512)
output = tokenizer.decode(outputs[0], skip_special_tokens=True)
if f'{self.tl}:' in output:
output = output.rsplit(f'{self.tl}:')[-1].strip().replace('assistant\n', '').strip()
return output
def unbabel(self):
pipe = pipeline("text-generation", model=self.model_name, torch_dtype=torch.bfloat16, device_map="auto")
messages = [{"role": "user",
"content": f"Translate the following text from {self.sl} into {self.tl}.\n{self.sl}: {self.input_text}.\n{self.tl}:"}]
prompt = pipe.tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=False)
tokenized_input = pipe.tokenizer(self.input_text, return_tensors="pt")
num_input_tokens = len(tokenized_input["input_ids"][0])
max_new_tokens = round(num_input_tokens + 0.5 * num_input_tokens)
outputs = pipe(prompt, max_new_tokens=max_new_tokens, do_sample=False)
translated_text = outputs[0]["generated_text"]
print(f"Input chars: {len(self.input_text)}", f"Input tokens: {num_input_tokens}", f"max_new_tokens: {max_new_tokens}",
"Chars to tokens ratio:", round(len(self.input_text) / num_input_tokens, 2), f"Raw translation: {translated_text}")
markers = ["", "<|im_end|>", "<|im_start|>assistant"] # , "\n"
for marker in markers:
if marker in translated_text:
translated_text = translated_text.split(marker)[1].strip()
translated_text = translated_text.replace('Answer:', '', 1).strip() if translated_text.startswith('Answer:') else translated_text
translated_text = translated_text.split("Translated text:")[0].strip() if "Translated text:" in translated_text else translated_text
split_translated_text = translated_text.split('\n', translated_text.count('\n'))
translated_text = '\n'.join(split_translated_text[:self.input_text.count('\n')+1])
return translated_text
def bergamot(model_name: str = 'deen', sl: str = 'de', tl: str = 'en', input_text: str = 'Hallo, mein Freund'):
try:
import bergamot
# input_text = [input_text] if isinstance(input_text, str) else input_text
config = bergamot.ServiceConfig(numWorkers=4)
service = bergamot.Service(config)
model = service.modelFromConfigPath(f"./{model_name}/bergamot.config.yml")
options = bergamot.ResponseOptions(alignment=False, qualityScores=False, HTML=False)
rawresponse = service.translate(model, bergamot.VectorString(input_text), options)
translated_text: str = next(iter(rawresponse)).target.text
message_text = f"Translated from {sl} to {tl} with Bergamot {model_name}."
except Exception as error:
response = error
return translated_text, message_text
def timer(func):
from time import time
def wrapper(*args, **kwargs):
start_time = time()
translated_text, message_text = func(*args, **kwargs)
end_time = time()
execution_time = end_time - start_time
# print(f"Function {func.__name__!r} executed in {execution_time:.4f} seconds.")
message_text = f'Executed in {execution_time:.2f} seconds! {message_text}'
return translated_text, message_text
return wrapper
@timer
def translate_text(model_name: str, s_language: str, t_language: str, input_text: str) -> tuple[str, str]:
"""
Translates the input text from the source language to the target language using a specified model.
Parameters:
input_text (str): The source text to be translated
s_language (str): The source language of the input text
t_language (str): The target language in which the input text is translated
model_name (str): The selected translation model name
Returns:
tuple:
translated_text(str): The input text translated to the selected target language
message_text(str): A descriptive message summarizing the translation process. Example: "Translated from English to German with Helsinki-NLP."
Example:
>>> translate_text("Hello world", "English", "German", "Helsinki-NLP")
("Hallo Welt", "Translated from English to German with Helsinki-NLP.")
"""
sl = all_langs[s_language][0]
tl = all_langs[t_language][0]
if not input_text.strip() or input_text.strip() == '':
translated_text = f'No input text entered!'
message_text = 'Please enter a text to translate!'
return translated_text, message_text
if sl == tl:
translated_text = f'Source language {s_language} identical to target language {t_language}!'
message_text = 'Please choose different target and source language!'
return translated_text, message_text
message_text = f'Translated from {s_language} to {t_language} with {model_name}'
translated_text = None
try:
if "-mul" in model_name.lower() or "mul-" in model_name.lower() or "-roa" in model_name.lower():
translated_text, message_text = Translators(model_name, sl, tl, input_text).HelsinkiNLP_mulroa()
elif model_name == "Helsinki-NLP":
translated_text, message_text = Translators(model_name, sl, tl, input_text).HelsinkiNLP()
elif model_name == 'Argos':
translated_text = Translators(model_name, sl, tl, input_text).argos()
elif model_name == "QUICKMT":
translated_text, message_text = Translators(model_name, sl, tl, input_text).quickmt()
elif model_name == 'Google':
translated_text = Translators(model_name, sl, tl, input_text).google()
elif model_name == 'google/translategemma-4b-it':
translated_text = Translators(model_name, sl, tl, input_text).translategemma()
elif model_name == "nvidia/Riva-Translate-4B-Instruct-v1.1":
translated_text = Translators(model_name, sl, tl, input_text).riva4B11()
elif "xiaomi" in model_name:
translated_text = Translators(model_name, s_language, t_language, input_text).xiaomi()
elif model_name == "Helsinki-NLP/opus-mt-tc-bible-big-roa-en":
translated_text, message_text = Translators(model_name, sl, tl, input_text).simplepipe()
elif 'mitre' in model_name.lower():
translated_text = Translators(model_name, sl, tl, input_text).mitre()
elif "m2m" in model_name.lower():
translated_text = Translators(model_name, sl, tl, input_text).mtom()
elif "small100" in model_name.lower():
translated_text = Translators(model_name, sl, tl, input_text).smallonehundred()
elif "rosetta" in model_name.lower():
translated_text = Translators(model_name, s_language, t_language, input_text).rosetta()
elif "lego" in model_name.lower():
translated_text = Translators(model_name, sl, tl, input_text).LegoMT()
elif "niutrans" in model_name.lower():
translated_text = Translators(model_name, sl, tl, input_text).niutrans()
elif "salamandra" in model_name.lower():
translated_text = Translators(model_name, s_language, t_language, input_text).salamandratapipe()
elif model_name.startswith('google-t5'):
translated_text = Translators(model_name, s_language, t_language, input_text).tfive()
elif 'flan' in model_name.lower():
translated_text = Translators(model_name, s_language, t_language, input_text).flan()
elif 'madlad' in model_name.lower():
translated_text = Translators(model_name, sl, tl, input_text).madlad()
elif 'mt0' in model_name.lower():
translated_text = Translators(model_name, s_language, t_language, input_text).bigscience()
elif 'bloomz' in model_name.lower():
translated_text = Translators(model_name, s_language, t_language, input_text).bloomz()
elif 'nllb' in model_name.lower():
nnlbsl, nnlbtl = languagecodes.nllb_language_codes[s_language], languagecodes.nllb_language_codes[t_language]
translated_text = Translators(model_name, nnlbsl, nnlbtl, input_text).nllb()
elif model_name == "facebook/mbart-large-50-many-to-many-mmt":
translated_text = Translators(model_name, s_language, t_language, input_text).mbart_many_to_many()
elif model_name == "facebook/mbart-large-50-one-to-many-mmt":
translated_text = Translators(model_name, s_language, t_language, input_text).mbart_one_to_many()
elif model_name == "facebook/mbart-large-50-many-to-one-mmt":
translated_text = Translators(model_name, s_language, t_language, input_text).mbart_many_to_one()
elif model_name == "facebook/seamless-m4t-v2-large":
translated_text = Translators(model_name, sl, tl, input_text).seamlessm4t2()
elif "m4t-medium" in model_name or "m4t-large" in model_name:
translated_text = Translators(model_name, sl, tl, input_text).seamlessm4t1()
elif model_name == "utter-project/EuroLLM-1.7B-Instruct":
translated_text = Translators(model_name, s_language, t_language, input_text).eurollm_instruct()
elif model_name == "utter-project/EuroLLM-1.7B":
translated_text = Translators(model_name, s_language, t_language, input_text).eurollm()
elif "EuroMoE" in model_name:
translated_text = Translators(model_name, s_language, t_language, input_text).EuroMoE()
elif 'Unbabel' in model_name:
translated_text = Translators(model_name, s_language, t_language, input_text).unbabel()
elif "winninghealth/WiNGPT" in model_name:
translated_text = Translators(model_name, s_language, t_language, input_text).wingpt()
elif "HPLT" in model_name:
if model_name == "HPLT-OPUS":
translated_text, message_text = Translators(model_name, sl, tl, input_text).hplt(opus = True)
else:
translated_text, message_text = Translators(model_name, sl, tl, input_text).hplt()
elif model_name == "Bergamot":
translated_text, message_text = Translators(model_name, s_language, t_language, input_text).bergamot()
except Exception as error:
translated_text = error
finally:
print(input_text, translated_text, message_text)
return translated_text, message_text