Update app.py

app.py

@@ -1,332 +1,160 @@
-import os
-import re
-import cohere
 import gradio as gr
-from typing import Iterator, List, Tuple
+from threading import Thread
+import requests
+from typing import List, Tuple
+import json
+import os
+from queue import Queue
 
-# Configuration Constants
-DEFAULT_SYSTEM_PROMPT = """ 
-أنت مترجم ثنائي اللغة متخصص في الترجمة بين العربية والإنجليزية. هدفك هو تقديم ترجمات دقيقة، ملائمة للسياق، ومتسقة من الناحية الأسلوبية، مع الالتزام بالإرشادات التالية:
-أسلوب الكتابة:
-1. الدقة النحوية: احرص دائمًا على أن تكون الترجمة صحيحة نحويًا.
-2. الملاءمة السياقية: قم بتكييف الترجمة مع السياق والجمهور المستهدف:
-استخدم جملًا موجزة وواضحة في المحتوى الطبي والقانوني.
-اعتمد لغة جذابة وخيالية في المواد التسويقية.
-حافظ على الرسمية والبلاغية في المحتوى القانوني.
-3. البنية السليمة: احترم تراكيب الجمل العربية وتجنب النسخ المباشر لقواعد لغة المصدر. فضل استخدام "الجملة الفعلية" إلا إذا كانت "الجملة الاسمية" أكثر ملاءمة (مثلًا في العناوين أو التنويهات).
-الخيارات الأسلوبية:
-استخدم التشكيل فقط عند الضرورة لتوضيح المعنى.
-تعامل مع الأسماء والأحرف المختصرة حسب السياق:
-قم بتعريب الأسماء وأسماء الأدوية إلا إذا وُجدت لها مرادفات عربية.
-ترجم أسماء البرامج والإدارات والهيئات عندما يكون ذلك مفيدًا.
-استخدم الأرقام العربية وتأكد من التعامل الصحيح مع الوحدات والعناوين والمراجع.
-علامات الترقيم:
-طبق قواعد الترقيم العربية لضمان وضوح النص.
-استخدم الفاصلة العربية (،) والفاصلة المنقوطة (؛) حسب الأعراف.
-تجنب الإفراط في استخدام علامات الاقتباس واحرص على وضع النقطتين (:) بشكل منطقي.
-الأخطاء الشائعة التي يجب تجنبها:
-تجنب ترجمة "is" بـ "يُعتبر" إلا إذا كان ذلك مناسبًا للسياق.
-استخدم حروف الجر والعطف بشكل صحيح لضمان تدفق الجمل بسلاسة.
-قلل من التكرار واستخدم الضمائر حيثما أمكن.
-تجنب الإفراط في استخدام تراكيب مثل "(قام + الفعل)" و"الخاص بـ."
-المصطلحات المحددة:
-في الترجمات القانونية، حافظ على النبرة الرسمية وتأكد من دقة المصطلحات.
-في الترجمات الطبية، بسّط المصطلحات التقنية للجمهور العام مع الاحتفاظ بالتعقيد عند مخاطبة المتخصصين.
-في الترجمات التسويقية، ركّز على الإبداع بدلًا من الترجمة الحرفية، مع الحفاظ على الرسالة الأساسية.
-إرشادات التنسيق:
-التزم بمعايير الطباعة العربية بشكل متسق.
-حافظ على تنسيق البيانات المهمة (مثل التواريخ، والقياسات، والاستشهادات القانونية).
-عند الشك:
-قدم الأولوية للوضوح، والتناسق، والملاءمة مع احتياجات الجمهور المستهدف. قم دائمًا بموازنة التعليمات الخاصة بالمشروع مع هذه الإرشادات، مع إعطاء الأولوية لمتطلبات العميل عند وجود أي تعارض.
-"""
+# Replace with your Hugging Face API token
+HF_API_TOKEN = os.getenv("HF_API_TOKEN")
+API_URL = f"https://api-inference.huggingface.co/models/FuseAI/FuseO1-DeepSeekR1-QwQ-SkyT1-32B-Preview"
+
+DEFAULT_SYSTEM_PROMPT = """
+You are an Advanced AI Coding Assistant, designed to solve complex challenges and deliver efficient, dependable solutions. Follow this structured workflow for every task:
+
+1. Understand: Analyze the problem thoroughly. Identify core objectives, resolve ambiguities, and ask clarifying questions if needed to ensure a complete understanding.
+
+2. Plan: Outline a clear, step-by-step approach, detailing the tools, frameworks, and algorithms required to achieve the solution effectively.
 
-TITLE = "<h1><center>Arabic localizer Translation Assistant </center></h1>"
-PLACEHOLDER = "اسأل عن أي شيء! سأفكر فيه خطوة بخطوة."
+3. Implement: Execute the plan with well-structured, efficient, and well-commented code. Provide a clear explanation of your thought process and the rationale behind key decisions as you proceed.
+
+4. Validate: Test the solution rigorously to ensure accuracy, efficiency, and alignment with best practices. Debug and optimize where necessary.
+
+5. Conclude: Summarize the solution with a clear conclusion, highlighting its effectiveness. Suggest improvements, optimizations, or alternative approaches if applicable.
+
+Guiding Principles:
+Use code as a tool for reasoning, with clear and educational explanations.
+Prioritize code readability, scalability, and maintainability.
+Adapt explanations to the user's skill level to maximize learning value.
+Refine solutions iteratively, incorporating feedback or evolving requirements.
+"""
 
 CSS = """
-.duplicate-button {
-    margin: auto !important;
-    color: white !important;
-    background: black !important;
-    border-radius: 100vh !important;
-}
-h3 {
-    text-align: center;
-}
-.message-wrap {
-    overflow-x: auto;
-}
-.message-wrap p {
-    margin-bottom: 1em;
-}
-.message-wrap pre {
-    background-color: #f6f8fa;
-    border-radius: 3px;
-    padding: 16px;
-    overflow-x: auto;
-}
-.message-wrap code {
-    background-color: rgba(175,184,193,0.2);
-    border-radius: 3px;
-    padding: 0.2em 0.4em;
-    font-family: monospace;
-}
-.custom-tag {
-    color: #0066cc;
-    font-weight: bold;
-}
-.chat-area {
-    height: 500px !important;
-    overflow-y: auto !important;
-}
-.arabic-input {
-    direction: rtl;
-    text-align: right;
-    font-family: 'Arabic', Arial, sans-serif;
-}
+.gr-chatbot { min-height: 500px; border-radius: 15px; }
+.special-tag { color: #2ecc71; font-weight: 600; }
+footer { display: none !important; }
 """
 
-def is_arabic_text(text: str) -> bool:
-    """Detect if the text contains Arabic characters"""
-    return any('\u0600' <= char <= '\u06FF' for char in text)
+def format_response(text: str) -> str:
+    """Format the response by adding HTML styling to special tags."""
+    return text.replace("[Understand]", '\n<strong class="special-tag">[Understand]</strong>\n') \
+              .replace("[Plan]", '\n<strong class="special-tag">[Plan]</strong>\n') \
+              .replace("[Conclude]", '\n<strong class="special-tag">[Conclude]</strong>\n') \
+              .replace("[Reason]", '\n<strong class="special-tag">[Reason]</strong>\n') \
+              .replace("[Verify]", '\n<strong class="special-tag">[Verify]</strong>\n')
 
-def format_text(text: str) -> str:
-    """Format text with proper spacing and tag highlighting"""
-    tag_patterns = [
-        (r'<Thinking>', '\n<Thinking>\n'),
-        (r'</Thinking>', '\n</Thinking>\n'),
-        (r'<Critique>', '\n<Critique>\n'),
-        (r'</Critique>', '\n</Critique>\n'),
-        (r'<Revising>', '\n<Revising>\n'),
-        (r'</Revising>', '\n</Revising>\n'),
-        (r'<Final>', '\n<Final>\n'),
-        (r'</Final>', '\n</Final>\n')
-    ]
-    
-    formatted = text
-    for pattern, replacement in tag_patterns:
-        formatted = re.sub(pattern, replacement, formatted)
+def query_huggingface(
+    payload: dict,
+    api_url: str = API_URL,
+    headers: dict = {"Authorization": f"Bearer {HF_API_TOKEN}"}
+) -> requests.Response:
+    """Send a request to the Hugging Face API endpoint."""
+    return requests.post(api_url, headers=headers, json=payload)
+
+def stream_response(response_queue: Queue, chat_history: List[Tuple[str, str]], message: str):
+    """Stream the response from the API and update the chat history."""
+    partial_message = ""
+    new_history = chat_history + [(message, "")]
     
-    formatted = '\n'.join(line for line in formatted.split('\n') if line.strip())
+    while True:
+        try:
+            token = response_queue.get()
+            if token is None:  # End of stream
+                break
+            
+            partial_message += token
+            formatted = format_response(partial_message)
+            new_history[-1] = (message, formatted + "▌")
+            yield new_history
+            
+        except Exception as e:
+            print(f"Error in streaming: {e}")
+            break
     
-    return formatted
+    # Final update without cursor
+    new_history[-1] = (message, format_response(partial_message))
+    yield new_history
 
-def format_chat_history(history: List[List[str]]) -> str:
-    """Format chat history for display"""
-    formatted = []
-    for user_msg, assistant_msg in history:
-        formatted.append(f"User: {user_msg}")
-        if assistant_msg:
-            formatted.append(f"Assistant: {assistant_msg}")
-    return "\n\n".join(formatted)
+def generate_response(
+    message: str,
+    chat_history: List[Tuple[str, str]],
+    system_prompt: str,
+    temperature: float,
+    max_tokens: int
+) -> List[Tuple[str, str]]:
+    """Generate a response using the Hugging Face API."""
+    # Create conversation history
+    conversation = [{"role": "system", "content": system_prompt}]
+    for user_msg, bot_msg in chat_history:
+        conversation.extend([
+            {"role": "user", "content": user_msg},
+            {"role": "assistant", "content": bot_msg}
+        ])
+    conversation.append({"role": "user", "content": message})
 
-def create_examples() -> List[str]:
-    """Create example queries for the UI"""
-    return [
-        "اشرح مفهوم الذكاء الاصطناعي",
-        "كيف تعمل عملية التمثيل الضوئي؟",
-        "ما هي الأسباب الرئيسية لتغير المناخ؟",
-        "صف عملية تخليق البروتين",
-        "ما هي الميزات الرئيسية للحكومة الديمقراطية؟",
-    ]
+    # Prepare the payload
+    payload = {
+        "inputs": conversation,
+        "parameters": {
+            "max_new_tokens": max_tokens,
+            "temperature": temperature,
+            "stream": True
+        }
+    }
 
-def convert_history_to_cohere_format(history: List[List[str]]) -> List[dict]:
-    """Convert chat history to Cohere's format"""
-    cohere_history = []
-    for user_msg, assistant_msg in history:
-        if user_msg:
-            cohere_history.append({"role": "User", "message": user_msg})
-        if assistant_msg:
-            cohere_history.append({"role": "Chatbot", "message": assistant_msg})
-    return cohere_history
+    # Create a queue for streaming tokens
+    response_queue = Queue()
+    
+    def process_stream():
+        try:
+            response = query_huggingface(payload)
+            
+            # Parse streaming response
+            for line in response.iter_lines():
+                if line:
+                    json_response = json.loads(line)
+                    if "token" in json_response:
+                        response_queue.put(json_response["token"]["text"])
+            
+            response_queue.put(None)  # Signal end of stream
+        except Exception as e:
+            print(f"Error in API call: {e}")
+            response_queue.put(None)
 
-def chat_response(
-    message: str,
-    history: List[List[str]],
-    chat_display: str,
-    system_prompt: str,
-    temperature: float = 0,
-    max_new_tokens: int = 8192,
-    top_p: float = 0.8,
-    top_k: int = 40,
-    penalty: float = 1.2,
-    api_key: str = os.getenv("COHERE_API_KEY")
-) -> Iterator[Tuple[List[List[str]], str]]:
-    """Generate chat responses using Cohere API"""
-    co = cohere.Client(api_key=api_key)
+    # Start processing thread
+    Thread(target=process_stream).start()
     
-    # Convert history to Cohere format
-    chat_history = convert_history_to_cohere_format(history)
+    # Return generator for streaming
+    return stream_response(response_queue, chat_history, message)
+
+# Create Gradio interface
+with gr.Blocks(css=CSS, theme=gr.themes.Soft()) as demo:
+    gr.Markdown("""
+    <h1 align="center">🧠 AI Reasoning Assistant</h1>
+    <p align="center">Ask me Hard questions</p>
+    """)
     
-    # Initialize buffer for streaming
-    buffer = ""
-    history = history + [[message, ""]]
+    chatbot = gr.Chatbot(label="Conversation", elem_id="chatbot")
+    msg = gr.Textbox(label="Your Question", placeholder="Type your question...")
     
-    # Process stream
-    try:
-        # Initialize stream
-        stream = co.chat_stream(
-            model='c4ai-aya-expanse-32b',
-            message=message,
-            temperature=temperature,
-            chat_history=chat_history,
-            prompt_truncation='AUTO',
-            preamble=system_prompt
-        )
-        
-        for event in stream:
-            if event.event_type == "text-generation":
-                buffer += event.text
-                formatted_buffer = format_text(buffer)
-                history[-1][1] = formatted_buffer
-                chat_display = format_chat_history(history)
-                yield history, chat_display
-                
-    except Exception as e:
-        error_message = f"Error: {str(e)}"
-        history[-1][1] = error_message
-        chat_display = format_chat_history(history)
-        yield history, chat_display
-
-def process_example(example: str) -> tuple:
-    """Process example query and return empty history and updated display"""
-    return [], f"User: {example}\n\n"
+    with gr.Accordion("⚙️ Settings", open=False):
+        system_prompt = gr.TextArea(value=DEFAULT_SYSTEM_PROMPT, label="System Instructions")
+        temperature = gr.Slider(0, 1, value=0.5, label="Creativity")
+        max_tokens = gr.Slider(128, 4096, value=2048, label="Max Response Length")
 
-def main():
-    """Main function to set up and launch the Gradio interface"""
-    with gr.Blocks(css=CSS, theme="JohnSmith9982/small_and_pretty'") as demo:
-        gr.HTML(TITLE)
-        gr.DuplicateButton(
-            value="",
-            elem_classes="duplicate-button"
-        )
-        
-        with gr.Row():
-            with gr.Column():
-                chat_history = gr.State([])
-                chat_display = gr.TextArea(
-                    value="",
-                    label="Chat History",
-                    interactive=True,
-                    elem_classes=["chat-area"],
-                )
-                
-                message = gr.TextArea(
-                    placeholder=PLACEHOLDER,
-                    label="Your message",
-                    lines=3,
-                    elem_classes=["message-input"]
-                )
-                
-                with gr.Row():
-                    submit = gr.Button("Send")
-                    clear = gr.Button("Clear")
-                
-                with gr.Accordion("⚙️ Advanced Settings", open=False):
-                    system_prompt = gr.TextArea(
-                        value=DEFAULT_SYSTEM_PROMPT,
-                        label="System Prompt",
-                        lines=7,
-                    )
-                    temperature = gr.Slider(
-                        minimum=0,
-                        maximum=1,
-                        step=0.1,
-                        value=0,
-                        label="Temperature",
-                    )
-                    max_tokens = gr.Slider(
-                        minimum=128,
-                        maximum=32000,
-                        step=128,
-                        value=8192,
-                        label="Max Tokens",
-                    )
-                    top_p = gr.Slider(
-                        minimum=0.1,
-                        maximum=1.0,
-                        step=0.1,
-                        value=0.8,
-                        label="Top-p",
-                    )
-                    top_k = gr.Slider(
-                        minimum=1,
-                        maximum=100,
-                        step=1,
-                        value=40,
-                        label="Top-k",
-                    )
-                    penalty = gr.Slider(
-                        minimum=1.0,
-                        maximum=2.0,
-                        step=0.1,
-                        value=1.2,
-                        label="Repetition Penalty",
-                    )
-                
-                examples = gr.Examples(
-                    examples=create_examples(),
-                    inputs=[message],
-                    outputs=[chat_history, chat_display],
-                    fn=process_example,
-                    cache_examples=False,
-                )
-        
-        # JavaScript to handle Enter key submission and RTL direction
-        js_handler = """
-        function handleKeyPress(textArea, submitButton) {
-            textArea.addEventListener('keydown', function(e) {
-                // Check for Arabic characters
-                const containsArabic = /[\u0600-\u06FF]/.test(textArea.value);
-                
-                // Apply RTL class for Arabic text
-                if (containsArabic) {
-                    textArea.classList.add('arabic-input');
-                } else {
-                    textArea.classList.remove('arabic-input');
-                }
-
-                // Handle Enter key for submission (without Shift)
-                if (e.key === 'Enter' && !e.shiftKey) {
-                    e.preventDefault(); // Prevent default new line
-                    submitButton.click(); // Trigger submit button click
-                }
-            });
-        }
-        """
-        
-        # Set up event handlers
-        submit_click = submit.click(
-            chat_response,
-            inputs=[
-                message,
-                chat_history,
-                chat_display,
-                system_prompt,
-                temperature,
-                max_tokens,
-                top_p,
-                top_k,
-                penalty,
-            ],
-            outputs=[chat_history, chat_display],
-            show_progress=True,
-        )
-        
-        # Clear input after submission
-        submit_click.then(lambda: "", outputs=message)
-        
-        # Add JavaScript to handle Enter key and RTL
-        demo.load(None, inputs=[message, submit], js=js_handler)
-        
-        clear.click(
-            lambda: ([], ""),
-            outputs=[chat_history, chat_display],
-            show_progress=True,
-        )
+    clear = gr.Button("Clear History")
     
-    return demo
+    msg.submit(
+        generate_response,
+        [msg, chatbot, system_prompt, temperature, max_tokens],
+        [chatbot],
+        show_progress=True
+    )
+    clear.click(lambda: None, None, chatbot, queue=False)
 
 if __name__ == "__main__":
-    demo = main()
-    demo.launch()
+    if not HF_API_TOKEN:
+        print("Please set your Hugging Face API token as environment variable HF_API_TOKEN")
+    else:
+        demo.queue().launch()