Create app.py

app.py

@@ -0,0 +1,199 @@
+import gradio as gr
+import time
+import re
+from tqdm import tqdm
+from openai import OpenAI
+
+instructs = {'eng': 'English', 
+            'kaz': 'Kazakh',
+            'rus': 'Russian', 
+            'tur': 'Turkish',
+            'uzn': 'Uzbek',
+            'zho_simpl': 'Chinese (Simplified)'}
+
+openai_api_key = "EMPTY"
+openai_api_base = "http://localhost:7050/v1"
+model_path = "gemma_translator"
+client = OpenAI(
+    api_key=openai_api_key,
+    base_url=openai_api_base,
+)
+
+
+def build_prompt_alpaca(instruction: str, input_text: str = "") -> str:
+    return (
+        "<bos>Below is an instruction that describes a task, paired with an input that provides further context. "
+        "Write a response that appropriately completes the request.\n\n"
+        f"### Instruction:\n{instruction}\n\n### Input:\n{input_text}\n\n### Response:\n"
+    )
+
+
+def get_prediction(instruction, input_text, temperature=0.0, top_p=0.9, max_tokens=1024, 
+                  presence_penalty=0.0, frequency_penalty=0.3, repetition_penalty=1.0,
+                  top_k=64, min_p=0.0, retry_count=3, sleep_time=1):
+    for attempt in range(retry_count):
+        try:
+            prompt = build_prompt_alpaca(instruction, input_text)
+            response = client.completions.create(
+                model=model_path,
+                prompt=prompt,
+                temperature=temperature,
+                top_p=top_p,
+                max_tokens=max_tokens,
+                frequency_penalty=frequency_penalty,
+                stream=True
+            )
+            return response
+            
+        except Exception as e:
+            print(f"Error on attempt {attempt+1}: {str(e)}")
+            if attempt < retry_count - 1:
+                print(f"Retrying in {sleep_time} seconds...")
+                time.sleep(sleep_time)
+                sleep_time *= 2  # Exponential backoff
+            else:
+                print("Max retries reached. Returning empty string.")
+                return ""
+
+
+def process_streaming_response(response):
+    """Process streaming response and return complete text"""
+    if not response:
+        return "Error: No response received"
+    
+    buffer = ""
+    is_first_chunk = True
+    complete_text = ""
+    
+    try:
+        for chunk in response:
+            if hasattr(chunk, 'choices') and chunk.choices and chunk.choices[0].text:
+                text_content = chunk.choices[0].text
+                
+                if is_first_chunk:
+                    text_content = text_content.lstrip()
+                    if text_content:
+                        text_content = " " + text_content
+                    is_first_chunk = False
+                
+                buffer += text_content
+                complete_text += text_content
+        
+        return complete_text.strip()
+    except Exception as e:
+        return f"Error processing response: {str(e)}"
+
+
+def generate_translation(text, target_lang, temperature, top_p, max_tokens,
+                        presence_penalty, frequency_penalty, repetition_penalty,
+                        top_k, min_p, use_v0_prompt=False):
+    """Updated function that accepts all parameters from Gradio"""
+    if not text.strip():
+        return "Please enter some text to translate."
+    
+    text = text.strip()
+
+    # Build instruction based on target language
+    if use_v0_prompt:
+        instruction = f"Translate the following text into {instructs[target_lang]}."
+    else:
+        instruction = f"Translate to {instructs[target_lang]}"
+    
+    try:
+        # Use the get_prediction function for inference
+        response = get_prediction(
+            instruction=instruction, 
+            input_text=text,
+            temperature=temperature,
+            top_p=top_p,
+            max_tokens=int(max_tokens),
+            presence_penalty=presence_penalty,
+            frequency_penalty=frequency_penalty,
+            repetition_penalty=repetition_penalty,
+            top_k=int(top_k),
+            min_p=min_p,
+            retry_count=3,
+            sleep_time=1
+        )
+        
+        # Process the streaming response
+        return process_streaming_response(response)
+        
+    except Exception as e:
+        return f"Error: {str(e)}"
+
+
+def set_example_text(example_text):
+    """Helper function to set example text"""
+    return example_text
+
+
+# Gradio UI
+with gr.Blocks() as demo:
+    gr.Markdown("## 🌐 Multilingual Translation App")
+
+    with gr.Row():
+        input_text = gr.Textbox(
+            label="Enter your text", 
+            placeholder="Type here and press Enter or click Translate", 
+            lines=3
+        )
+
+    with gr.Row():
+        lang_dropdown = gr.Dropdown(
+            choices=list(instructs.keys()), 
+            value="kaz", 
+            label="Translate to"
+        )
+
+    with gr.Accordion("Advanced Parameters (Optional)", open=False):
+        temperature_slider = gr.Slider(0.0, 1.0, value=0.0, step=0.01, label="Temperature")
+        top_p_slider = gr.Slider(0.0, 1.0, value=0.9, step=0.01, label="Top-p")
+        max_tokens_slider = gr.Slider(64, 2048, value=1024, step=1, label="Max Tokens")
+        presence_penalty_slider = gr.Slider(-2.0, 2.0, value=0.0, step=0.01, label="Presence Penalty", info="Usually left at 0.0")
+        frequency_penalty_slider = gr.Slider(-2.0, 2.0, value=0.3, step=0.01, label="Frequency Penalty")
+        repetition_penalty_slider = gr.Slider(0.5, 2.0, value=1.0, step=0.01, label="Repetition Penalty", info="Default is 1.0")
+        top_k_slider = gr.Slider(1, 100, value=64, step=1, label="Top-k", info="Default 64")
+        min_p_slider = gr.Slider(0.0, 1.0, value=0.0, step=0.01, label="Min-p", info="Usually 0.0")
+        use_v0_prompt = gr.Checkbox(label="Use v0 Prompt Format", value=False)
+
+    with gr.Row():
+        submit_btn = gr.Button("Translate", variant="primary")
+
+    output_text = gr.Textbox(label="Translation Result", lines=4)
+
+    # Define inputs in the correct order to match the function parameters
+    inputs = [
+        input_text, lang_dropdown,
+        temperature_slider, top_p_slider, max_tokens_slider,
+        presence_penalty_slider, frequency_penalty_slider, repetition_penalty_slider,
+        top_k_slider, min_p_slider, use_v0_prompt
+    ]
+    
+    # Connect the function to the button and text input
+    submit_btn.click(fn=generate_translation, inputs=inputs, outputs=output_text)
+    input_text.submit(fn=generate_translation, inputs=inputs, outputs=output_text)
+
+    # Example inputs
+    gr.Markdown("### 🔍 Examples:")
+    with gr.Row():
+        examples = [
+            "Hello! How can I help you?",
+            "Hello! how can I help you?",
+            "2 + 2 is?",
+            "Your appointment is on 5th July at 3 PM.",
+            "The total cost is 1250 KZT.",
+            "She was born in 1995."
+        ]
+        
+        for example_text in examples:
+            example_btn = gr.Button(example_text, size="sm")
+            example_btn.click(
+                fn=lambda x=example_text: x,
+                inputs=[],
+                outputs=[input_text]
+            )
+
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=5482, share=True)