File size: 7,472 Bytes
18a28ff
9f0a178
 
f7cc9ad
8b05c02
7bcabe0
6cc39d6
d120873
e46dbee
 
75255bc
 
 
 
d1936ad
 
 
 
 
 
 
d120873
c1dac64
4e5cd58
 
eaec419
d3a5771
4e5cd58
 
 
eaec419
4e5cd58
 
53869cd
1d71a28
a696db8
 
c3a14f0
 
d3a5771
75255bc
4e5cd58
 
 
 
 
 
 
1d71a28
75255bc
d120873
42c14b6
75255bc
698fb3f
75255bc
d120873
42c14b6
75255bc
7db8a97
c3a14f0
7db8a97
c3a14f0
75255bc
4e95015
d43d990
 
 
dc27630
678e936
0a92611
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
eec15ee
f873413
eb10e5e
 
678e936
f873413
d43d990
 
0734240
d43d990
 
43e522d
d43d990
 
 
 
 
 
 
 
 
 
44b38c0
0a92611
d5c671b
 
0519180
7ebabfa
0a92611
eec15ee
aded46b
0a92611
4c7bc2b
40d6045
d1936ad
98d0958
9797698
d1936ad
 
9797698
6cc39d6
4e95015
8b05c02
 
1d71a28
8b05c02
27ced65
c832cd2
53869cd
f7f2bbc
8b05c02
a696db8
4e95015
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
from fastapi import FastAPI, Request, Query
import src.Paraphrase as Paraphrase
import src.Translate as 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

app = FastAPI()
# app = FastAPI(docs_url="/docs")
MODELS = {'enro': 'BlackKakapo/opus-mt-en-ro',
          'roen': 'BlackKakapo/opus-mt-ro-en',
          'gemma': 'Gargaz/gemma-2b-romanian-better',
          '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"

@app.get("/")
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}

@app.get("/paraphrase", operation_id="get_paraphrase", description="Paraphrase text", tags=["paraphrase"], summary="Paraphrase text")
def paraphrase(text: str, model: str  = MODELS['paraphrase']):
    resultValue, exception = Paraphrase.paraphraseParaphraseMethod(text, model)
    return {"input": text, "result": resultValue, "exception": exception}

@app.get("/translate", operation_id="get_translate", description="Translate text", tags=["translate"], summary="Translate text")
def translate(text: str, model: Optional[str] = MODELS['enro']):
    if 'BlackKakapo' in model:
        translation = Translate.paraphraseTranslateMethod(text, model)
    else:
        translation: str = Translate.gemma_direct(text, model)
    return {"input": text, "result": translation, "model": model}

# Keep track of installed (src, tgt) pairs
installed_pairs = set()

# 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
@app.get("/bergamot", operation_id="get_bergamot", description="Translate text with Bergamot", tags=["bergamot"], summary="Translate text with Bergamot")
def bergamot(input_text: list[str] = Query(description="Input list of strings"), sl: str = 'de', tl: str = 'en', model_name: Optional[str] = 'deen'):
    """
    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
        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 to the selected target language
            message_text(str):  A descriptive message summarizing the translation process. Example: "Translated from English to German with ende."
    
    Example:
        >>> bergamot("Hello world", "en", "de", "ende")
        {"input_text": "Hello world", "translated_text": "Hallo Welt", "message_text": "Translated from English to German with ende."}
    """
    try:
        import bergamot
        # print(type(input_text))
        # input_text = [input_text] if isinstance(input_text, str) else input_text           
        config = bergamot.ServiceConfig(numWorkers=4)
        service = bergamot.Service(config)
        repo_id="TiberiuCristianLeon/Bergamot"
        branches = ['base', 'base-memory', 'tiny']
        subfolder = f"{sl}{tl}"
            # List all files in the repo
        all_files = list_repo_files(repo_id, repo_type='model')
        print(all_files, len(all_files))
        for branch in branches:
            branch_files = [f for f in all_files if f.startswith(branch)] 
            model_files = [f for f in branch_files if f.startswith(model_name)]
            print(model_files)
            for file_path in model_files:
                if subfolder not in file_path:
                    local_path = hf_hub_download(repo_id=repo_id, filename=file_path)
                    print(f"Downloaded to: {local_path}")
        installed_pairs.add(subfolder)
        model = service.modelFromConfigPath(f"./{model_name}/config.yml")
        options = bergamot.ResponseOptions(alignment=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 error:
        response, message_text = error, error
    return {"input": input_text, "translated_text": response, "message_text": message_text}

@app.get("/embed", operation_id="get_embeddings", description="Embed text", tags=["embed"], summary="Embed 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", "bergamot"]
)
# Mount the MCP server directly to the FASTAPI app
mcp.mount()