Spaces:
Running
Running
| from fastapi import FastAPI, Request, Query | |
| import src.Paraphrase as Paraphrase | |
| import src.Translate | |
| from typing import Optional | |
| from fastapi_mcp import FastApiMCP | |
| from huggingface_hub import hf_hub_download, list_repo_files | |
| from sentence_transformers import SentenceTransformer | |
| import bergamot | |
| app = FastAPI() | |
| # app = FastAPI(docs_url="/docs") | |
| MODELS = {'benro': 'BlackKakapo/opus-mt-en-ro', | |
| 'broen': 'BlackKakapo/opus-mt-ro-en', | |
| 'mttcbig': 'Helsinki-NLP/opus-mt-tc-big-en-ro', | |
| 'gemma': 'Gargaz/gemma-2b-romanian-better', | |
| 'mbartenro': 'ancebuc/mbart-translation-en-ro', | |
| 't5enro': 'ancebuc/t5-translation-en-ro', | |
| 'pegasus': 'ancebuc/pegasus-translation-en-ro', | |
| 'mbart': 'facebook/mbart-large-cc25', | |
| 'paraphrase': 'tuner007/pegasus_paraphrase'} | |
| EMBEDDING_MODELS = {"all-MiniLM-L6-v2":384, | |
| "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2":384, | |
| "sentence-transformers/distiluse-base-multilingual-cased-v2":512, | |
| "sentence-transformers/stsb-xlm-r-multilingual":768, | |
| "sentence-transformers/use-cmlm-multilingual":768, | |
| "sentence-transformers/paraphrase-multilingual-mpnet-base-v2":768} | |
| EMBEDDING_MODEL = "sentence-transformers/distiluse-base-multilingual-cased-v2" | |
| def index(request: Request): | |
| from fastapi.responses import HTMLResponse | |
| host_url = "https://" + request.url.netloc | |
| mcp_config = '''{"mcpServers": {"fastapi-mcp": {"url": "https://tiberiucristianleon-fastapimt.hf.space/mcp"}}}''' | |
| html_content = f''' | |
| <html> | |
| <head> | |
| <title>FastAPI with MCP</title> | |
| </head> | |
| <body> | |
| <h2>FastAPI URLS</h2> | |
| <p><a href="{host_url}" target="_blank">Host URL:</a> {host_url}</p> | |
| <p><a href="{host_url}/docs" target="_blank">DOCS</a></p> | |
| <p><a href="{host_url}/redoc" target="_blank">REDOC</a></p> | |
| <p><a href="{host_url}/openapi.json" target="_blank">openapi.json</a></p> | |
| <p><a href="{host_url}/mcp" target="_blank">MCP</a></p> | |
| <p>MCP configuration: {mcp_config}</a></p> | |
| <p>MODELS: {list(MODELS.values())}"</p> | |
| </body> | |
| </html> | |
| ''' | |
| return HTMLResponse(content=html_content) | |
| # @app.get("/") | |
| # async def get_host_url(request: Request): | |
| # host_url = request.url.scheme + "s://" + request.url.netloc | |
| # return {"host_url": host_url, 'endpoints': ['/paraphrase', '/translate', f'{host_url}/docs', f'{host_url}/redoc', f'{host_url}/openapi.json'], 'models': MODELS} | |
| def paraphrase(text: str, model: str = MODELS['paraphrase']): | |
| resultValue, exception = Paraphrase.paraphraseParaphraseMethod(text, model) | |
| return {"input": text, "result": resultValue, "exception": exception} | |
| def listmodels(): | |
| return {"MODELS": MODELS, "EMBEDDING_MODELS": EMBEDDING_MODELS} | |
| # model: Optional[str] = MODELS['benro'] | |
| def translate(input_text: str, model_name: str = MODELS['mttcbig'], sl: str = 'en', tl: str = 'ro'): | |
| message = f'Translated from {sl} to {tl} with {model_name}' | |
| if 'BlackKakapo' in model_name: | |
| translation, model_name = src.Translate.paraphraseTranslateMethod(input_text, model_name) | |
| elif 'Helsinki-NLP' in model_name: | |
| translation, message = src.Translate.Translators(model_name, sl, tl, input_text).HelsinkiNLP_mulroa() | |
| # text2textgenerationpipe, translationpipe | |
| # elif model_name == MODELS['mbartenro']: | |
| # translation, message = src.Translate.Translators(model_name, sl, tl, input_text).text2textgenerationpipe() | |
| elif model_name == MODELS['t5enro'] or model_name == MODELS['pegasus'] or model_name == MODELS['mbartenro']: | |
| translation, message = src.Translate.Translators(model_name, sl, tl, input_text).translationpipe() | |
| elif model_name == MODELS['mbart']: | |
| translation, message = src.Translate.Translators(model_name, sl, tl, input_text).mbartlarge() | |
| else: | |
| translation: str = src.Translate.gemma_direct(input_text, model_name) | |
| return {"input_text": input_text, "translation": translation, "model_name": model_name, "message": message} | |
| class Bergamot: | |
| config = bergamot.ServiceConfig(numWorkers=4) | |
| service = bergamot.Service(config) | |
| def __init__(self, input_text, sl, tl, model_name): | |
| # Keep track of installed (src, tgt) pairs | |
| self.installed_pairs = set() | |
| self.input_text, self.sl, self.tl, self.model_name = input_text, sl, tl, model_name | |
| self.repo_id = "TiberiuCristianLeon/Bergamot" | |
| self.branches = ['base', 'base-memory', 'tiny'] | |
| self.subfolder = f"{sl}{tl}" | |
| self.localfolder = f"{self.subfolder}/{model_name}" | |
| # List all files in the repo | |
| self.all_files = list_repo_files(self.repo_id, repo_type='model') | |
| def downloadbergamotfiles(self): | |
| print('input text type:', type(self.input_text), len(self.all_files), 'installed_pairs', self.installed_pairs, 'defaultlocalfolder', self.localfolder) | |
| try: | |
| for branch in self.branches: | |
| branch_files = [f for f in self.all_files if f.startswith(branch)] | |
| fullmodel_files = [f for f in branch_files if f.startswith(self.model_name)] | |
| print('branch_files', len(branch_files), 'fullmodel_files', fullmodel_files) | |
| model_files = [f.split(f'{self.model_name}/')[1] for f in fullmodel_files] | |
| print('branch_files', len(branch_files), 'model_files', model_files) | |
| for file_path in model_files: | |
| if self.localfolder not in self.installed_pairs: | |
| # local_files_only (bool, optional, defaults to False) — If True, avoid downloading the file and return the path to the local cached file if it exists. | |
| # dry_run (bool, optional, defaults to False) — If True, perform a dry run without actually downloading the file. Returns a DryRunFileInfo object containing information about what would be downloaded. | |
| local_path = hf_hub_download(repo_id=self.repo_id, subfolder=self.model_name, filename=file_path, local_dir=self.subfolder) | |
| print(f"Downloaded to: {local_path}") # Downloaded to: deen/base/deen/config.yml | |
| try: | |
| dry_run = hf_hub_download(repo_id=self.repo_id, subfolder=self.model_name, filename='config.yml', local_dir=self.subfolder) | |
| print('installed_pairs', self.installed_pairs, 'localfolder', self.localfolder, 'dry_run', type(dry_run), dry_run) | |
| if isinstance(dry_run, str): | |
| print('Add to set after dryrun', dry_run) | |
| self.installed_pairs.add(self.localfolder) | |
| except Exception as dryrunerror: | |
| print('installed_pairs', self.installed_pairs, 'localfolder', self.localfolder, 'dry_runerror', dryrunerror) | |
| except Exception as downloaderror: | |
| response, message_text = str(downloaderror), f"Error downloading {self.model_name}: {downloaderror}." | |
| print(downloaderror) | |
| def translate(self): | |
| self.downloadbergamotfiles() | |
| try: | |
| model = self.service.modelFromConfigPath(f"{self.localfolder}/config.yml") | |
| options = bergamot.ResponseOptions(alignment=False, sentenceMappings=False, qualityScores=False, HTML=False) | |
| rawresponse = self.service.translate(model, bergamot.VectorString(self.input_text), options) | |
| response: list|str = [r.target.text for r in rawresponse] if len(rawresponse) > 1 else next(iter(rawresponse)).target.text | |
| print(type(self.input_text), len(self.input_text), len(rawresponse), type(response), response) | |
| message_text = f"Translated from {self.sl} to {self.tl} with {self.model_name}." | |
| except Exception as translateerror: | |
| response, message_text = str(translateerror), f"Error translating from {self.sl} to {self.tl} with {self.model_name}: {translateerror}." | |
| print(translateerror) | |
| return {"input": self.input_text, "translated_text": response, "message_text": message_text} | |
| # https://tiberiucristianleon-fastapimt.hf.space/bergamot?input_text=das%20ist%20keine%20gute%20Frau&input_text=das%20ist%20eine%20gute%20Nachricht&sl=de&tl=en&model=bergamot | |
| def bergamott(input_text: list[str] = Query(description="Input string or list of strings"), sl: str = 'de', tl: str = 'en', model_name: Optional[str] = 'base/deen'): | |
| """ | |
| Translates the input text from the source language to the target language using a specified model. | |
| Parameters: | |
| input_text (str | list[str]): The source text to be translated, can be either a string or a list of strings | |
| sl (str): The source language of the input text | |
| tl (str): The target language into which the input text is translated | |
| model_name (str): The selected translation model name | |
| Returns: | |
| dict: | |
| input_text(str): The input text in the source language | |
| translated_text(str): The input text translated into the selected target language | |
| message_text(str): A descriptive message summarizing the translation process. Example: "Translated from English to German with base/ende." | |
| Example: | |
| >>> bergamot("Hello world", "en", "de", "base/ende") | |
| {"input_text": "Hello world", "translated_text": "Hallo Welt", "message_text": "Translated from English to German with base/ende."} | |
| """ | |
| try: | |
| bergamotinstance = Bergamot(input_text, sl, tl, model_name) | |
| return bergamotinstance.translate() | |
| except Exception as mainerror: | |
| response, message_text = str(mainerror), f"Error translating from {sl} to {tl} with {model_name}: {mainerror}." | |
| print(mainerror) | |
| # https://tiberiucristianleon-fastapimt.hf.space/bergamot?input_text=das%20ist%20keine%20gute%20Frau&input_text=das%20ist%20eine%20gute%20Nachricht&sl=de&tl=en&model=bergamot | |
| def bergamots(input_text: list[str] = Query(description="Input string or list of strings"), sl: str = 'de', tl: str = 'en', model_name: Optional[str] = 'base/deen'): | |
| """ | |
| Translates the input text from the source language to the target language using a specified model. | |
| Parameters: | |
| input_text (str | list[str]): The source text to be translated, can be either a string or a list of strings | |
| sl (str): The source language of the input text | |
| tl (str): The target language into which the input text is translated | |
| model_name (str): The selected translation model name | |
| Returns: | |
| dict: | |
| input_text(str): The input text in the source language | |
| translated_text(str): The input text translated into the selected target language | |
| message_text(str): A descriptive message summarizing the translation process. Example: "Translated from English to German with base/ende." | |
| Example: | |
| >>> bergamot("Hello world", "en", "de", "base/ende") | |
| {"input_text": "Hello world", "translated_text": "Hallo Welt", "message_text": "Translated from English to German with base/ende."} | |
| """ | |
| try: | |
| import bergamot | |
| repo_id="TiberiuCristianLeon/Bergamot" | |
| branches = ['base', 'base-memory', 'tiny'] | |
| # input_text = [input_text] if isinstance(input_text, str) else input_text | |
| config = bergamot.ServiceConfig(numWorkers=4) | |
| service = bergamot.Service(config) | |
| subfolder = f"{sl}{tl}" | |
| localfolder = f"{subfolder}/{model_name}" | |
| # List all files in the repo | |
| all_files = list_repo_files(repo_id, repo_type='model') | |
| print('input text type:', type(input_text), len(all_files), 'installed_pairs', installed_pairs, 'defaultlocalfolder', localfolder) | |
| for branch in branches: | |
| branch_files = [f for f in all_files if f.startswith(branch)] | |
| fullmodel_files = [f for f in branch_files if f.startswith(model_name)] | |
| print('branch_files', len(branch_files), 'fullmodel_files', fullmodel_files) | |
| model_files = [f.split(f'{model_name}/')[1] for f in fullmodel_files] | |
| print('branch_files', len(branch_files), 'model_files', model_files) | |
| for file_path in model_files: | |
| if localfolder not in installed_pairs: | |
| # local_files_only (bool, optional, defaults to False) — If True, avoid downloading the file and return the path to the local cached file if it exists. | |
| # dry_run (bool, optional, defaults to False) — If True, perform a dry run without actually downloading the file. Returns a DryRunFileInfo object containing information about what would be downloaded. | |
| local_path = hf_hub_download(repo_id=repo_id, subfolder=model_name, filename=file_path, local_dir=subfolder) | |
| print(f"Downloaded to: {local_path}") # Downloaded to: deen/base/deen/config.yml | |
| ## Check if model/localfolder in repo files, add to set if exists | |
| modelcheck = [i for i in all_files if model_name in i] | |
| print('Modelcheck', modelcheck) | |
| if modelcheck: | |
| print('Add to set after modelcheck', modelcheck) | |
| installed_pairs.add(localfolder) | |
| try: | |
| dry_run = hf_hub_download(repo_id=repo_id, subfolder=model_name, filename='config.yml', local_dir=subfolder) | |
| print('installed_pairs', installed_pairs, 'localfolder', localfolder, 'dry_run', type(dry_run), dry_run) | |
| if isinstance(dry_run, str): | |
| print('Add to set after dryrun', dry_run) | |
| installed_pairs.add(localfolder) | |
| except Exception as dryrunerror: | |
| print('installed_pairs', installed_pairs, 'localfolder', localfolder, 'dry_runerror', dryrunerror) | |
| model = service.modelFromConfigPath(f"{localfolder}/config.yml") | |
| # model = service.modelFromConfig(localfolder) | |
| options = bergamot.ResponseOptions(alignment=False, sentenceMappings=False, qualityScores=False, HTML=False) | |
| rawresponse = service.translate(model, bergamot.VectorString(input_text), options) | |
| response: list|str = [r.target.text for r in rawresponse] if len(rawresponse) > 1 else next(iter(rawresponse)).target.text | |
| print(type(input_text), len(input_text), len(rawresponse), type(response), response) | |
| # response = [r.target.text for r in model_response][0] if isinstance(response, bergamot._bergamot.VectorResponse) else next(iter(response)).target.text | |
| # response is of type bergamot._bergamot.VectorResponse, an iterable of bergamot._bergamot.Response | |
| message_text = f"Translated from {sl} to {tl} with {model_name}." | |
| except Exception as generalerror: | |
| response, message_text = str(generalerror), f"Error translating from {sl} to {tl} with {model_name}: {generalerror}." | |
| print(generalerror) | |
| return {"input": input_text, "translated_text": response, "message_text": message_text} | |
| def embed(text: str, model: str = EMBEDDING_MODEL): | |
| model = SentenceTransformer(model) | |
| embeddings = model.encode(text) | |
| print(embeddings.shape, len(embeddings)) | |
| # similarities = model.similarity(embeddings, embeddings) | |
| return {"input": text, "embeddings": embeddings.tolist(), "shape": embeddings.shape} | |
| # Create an MCP server based on this app | |
| mcp = FastApiMCP( | |
| app, | |
| name="Translate and paraphrase FASTAPI MCP", | |
| description="MCP server to translate and paraphrase text", | |
| describe_all_responses=True, | |
| describe_full_response_schema=True, | |
| include_operations=["get_translate", "get_paraphrase"], | |
| include_tags=["paraphrase", "translate", "bergamott"] | |
| ) | |
| # Mount the MCP server directly to the FASTAPI app | |
| mcp.mount() |