Create main.py

main.py

@@ -0,0 +1,426 @@
+import os
+import json
+import logging
+import gradio as gr
+from openai import OpenAI
+from pydoc import html
+from typing import List, Generator, Optional
+import requests
+from bs4 import BeautifulSoup
+import re
+
+# تعريف LATEX_DELIMS
+LATEX_DELIMS = [
+    {"left": "$$",  "right": "$$",  "display": True},
+    {"left": "$",   "right": "$",   "display": False},
+    {"left": "\\[", "right": "\\]", "display": True},
+    {"left": "\\(", "right": "\\)", "display": False},
+]
+
+# إعداد التسجيل
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# تحقق من الملفات في /app/ (للتصحيح)
+logger.info("Files in /app/: %s", os.listdir("/app"))
+
+# إعداد العميل لـ Hugging Face Inference API
+HF_TOKEN = os.getenv("HF_TOKEN")
+API_ENDPOINT = os.getenv("API_ENDPOINT", "https://api-inference.huggingface.co/v1")
+MODEL_NAME = os.getenv("MODEL_NAME", "openai/gpt-oss-120b:cerebras")
+SECONDARY_MODEL_NAME = os.getenv("SECONDARY_MODEL_NAME", "MGZON/mgzon-flan-t5-base")
+if not HF_TOKEN:
+    logger.error("HF_TOKEN is not set in environment variables.")
+    raise ValueError("HF_TOKEN is required for Inference API.")
+client = OpenAI(api_key=HF_TOKEN, base_url=API_ENDPOINT)
+
+# إعدادات الـ queue
+QUEUE_SIZE = int(os.getenv("QUEUE_SIZE", 80))
+CONCURRENCY_LIMIT = int(os.getenv("CONCURRENCY_LIMIT", 20))
+
+# كلمات مفتاحية لتحديد إذا كان السؤال متعلق بـ MGZon
+MGZON_KEYWORDS = [
+    "mgzon", "mgzon products", "mgzon services", "mgzon data", "mgzon platform",
+    "mgzon features", "mgzon mission", "mgzon technology", "mgzon solutions"
+]
+
+# دالة لاختيار النموذج تلقائيًا
+def select_model(query: str) -> str:
+    """Selects the appropriate model based on the query content."""
+    query_lower = query.lower()
+    for keyword in MGZON_KEYWORDS:
+        if keyword in query_lower:
+            logger.info(f"Selected {SECONDARY_MODEL_NAME} for MGZon-related query: {query}")
+            return SECONDARY_MODEL_NAME
+    logger.info(f"Selected {MODEL_NAME} for general query: {query}")
+    return MODEL_NAME
+
+# دالة بحث ويب محسنة
+def web_search(query: str) -> str:
+    try:
+        google_api_key = os.getenv("GOOGLE_API_KEY")
+        google_cse_id = os.getenv("GOOGLE_CSE_ID")
+        if not google_api_key or not google_cse_id:
+            return "Web search requires GOOGLE_API_KEY and GOOGLE_CSE_ID to be set."
+
+        url = f"https://www.googleapis.com/customsearch/v1?key={google_api_key}&cx={google_cse_id}&q={query}"
+        response = requests.get(url)
+        response.raise_for_status()
+        results = response.json().get("items", [])
+        if not results:
+            return "No web results found."
+
+        # جمع النتايج
+        search_results = []
+        for i, item in enumerate(results[:3]):  # نأخذ أول 3 نتايج
+            title = item.get("title", "")
+            snippet = item.get("snippet", "")
+            link = item.get("link", "")
+            # محاولة استخراج محتوى الصفحة
+            try:
+                page_response = requests.get(link, timeout=5)
+                page_response.raise_for_status()
+                soup = BeautifulSoup(page_response.text, "html.parser")
+                # استخراج النصوص من الصفحة (فقط الفقرات)
+                paragraphs = soup.find_all("p")
+                page_content = " ".join([p.get_text() for p in paragraphs][:500])  # نأخذ أول 500 حرف
+            except Exception as e:
+                logger.warning(f"Failed to fetch page content for {link}: {e}")
+                page_content = snippet
+            search_results.append(f"Result {i+1}:\nTitle: {title}\nLink: {link}\nContent: {page_content}\n")
+        
+        return "\n".join(search_results)
+    except Exception as e:
+        logger.exception("Web search failed")
+        return f"Web search error: {e}"
+
+# دالة request_generation (محدثة لدعم المهام المتعددة)
+def request_generation(
+    api_key: str,
+    api_base: str,
+    message: str,
+    system_prompt: str,
+    model_name: str,
+    chat_history: Optional[List[dict]] = None,
+    temperature: float = 0.9,
+    max_new_tokens: int = 2048,
+    reasoning_effort: str = "off",
+    tools: Optional[List[dict]] = None,
+    tool_choice: Optional[str] = None,
+    deep_search: bool = False,
+) -> Generator[str, None, None]:
+    """Streams Responses API events. Emits:
+      - "analysis" sentinel once, then raw reasoning deltas
+      - "assistantfinal" sentinel once, then visible output deltas
+    If no visible deltas, emits a tool-call fallback message."""
+    client = OpenAI(api_key=api_key, base_url=api_base)
+
+    # تحديد نوع المهمة بناءً على السؤال
+    task_type = "general"
+    if "code" in message.lower() or "programming" in message.lower() or any(ext in message.lower() for ext in ["python", "javascript", "react", "django", "flask"]):
+        task_type = "code"
+        enhanced_system_prompt = f"{system_prompt}\nYou are an expert programmer. Provide accurate, well-commented code with examples and explanations. Support frameworks like React, Django, Flask, and others as needed."
+    elif any(keyword in message.lower() for keyword in ["analyze", "analysis", "تحليل"]):
+        task_type = "analysis"
+        enhanced_system_prompt = f"{system_prompt}\nProvide detailed analysis with step-by-step reasoning, examples, and data-driven insights."
+    elif any(keyword in message.lower() for keyword in ["review", "مراجعة"]):
+        task_type = "review"
+        enhanced_system_prompt = f"{system_prompt}\nReview the provided content thoroughly, identify issues, and suggest improvements with detailed explanations."
+    elif any(keyword in message.lower() for keyword in ["publish", "نشر"]):
+        task_type = "publish"
+        enhanced_system_prompt = f"{system_prompt}\nPrepare content for publishing, ensuring clarity, professionalism, and adherence to best practices."
+    else:
+        enhanced_system_prompt = f"{system_prompt}\nPlease provide detailed and comprehensive responses, including explanations, examples, and relevant details where applicable."
+
+    logger.info(f"Task type detected: {task_type}")
+
+    # تنظيف الـ messages من metadata
+    input_messages: List[dict] = [{"role": "system", "content": enhanced_system_prompt}]
+    if chat_history:
+        for msg in chat_history:
+            clean_msg = {"role": msg.get("role"), "content": msg.get("content")}
+            if clean_msg["content"]:
+                input_messages.append(clean_msg)
+    
+    # إذا كان DeepSearch مفعّل أو السؤال عام، أضف نتائج البحث
+    if deep_search or model_name == MODEL_NAME:
+        search_result = web_search(message)
+        input_messages.append({"role": "user", "content": f"User query: {message}\nWeb search context: {search_result}"})
+    else:
+        input_messages.append({"role": "user", "content": message})
+
+    # إعداد tools و tool_choice (فقط لـ GPT-based models)
+    tools = tools if tools and "gpt-oss" in model_name else []
+    tool_choice = tool_choice if tool_choice in ["auto", "none", "any", "required"] and "gpt-oss" in model_name else "none"
+
+    try:
+        stream = client.chat.completions.create(
+            model=model_name,
+            messages=input_messages,
+            temperature=temperature,
+            max_tokens=max_new_tokens,
+            stream=True,
+            tools=tools,
+            tool_choice=tool_choice,
+        )
+
+        reasoning_started = False
+        reasoning_closed = False
+        saw_visible_output = False
+        last_tool_name = None
+        last_tool_args = None
+        buffer = ""
+
+        for chunk in stream:
+            if chunk.choices[0].delta.content:
+                content = chunk.choices[0].delta.content
+                if content == "<|channel|>analysis<|message|>":
+                    if not reasoning_started:
+                        yield "analysis"
+                        reasoning_started = True
+                    continue
+                if content == "<|channel|>final<|message|>":
+                    if reasoning_started and not reasoning_closed:
+                        yield "assistantfinal"
+                        reasoning_closed = True
+                    continue
+
+                saw_visible_output = True
+                buffer += content
+
+                if "\n" in buffer or len(buffer) > 150:
+                    yield buffer
+                    buffer = ""
+                continue
+
+            if chunk.choices[0].delta.tool_calls and "gpt-oss" in model_name:
+                tool_call = chunk.choices[0].delta.tool_calls[0]
+                name = getattr(tool_call, "function", {}).get("name", None)
+                args = getattr(tool_call, "function", {}).get("arguments", None)
+                if name:
+                    last_tool_name = name
+                if args:
+                    last_tool_args = args
+                continue
+
+            if chunk.choices[0].finish_reason in ("stop", "tool_calls", "error"):
+                if buffer:
+                    yield buffer
+                    buffer = ""
+
+                if reasoning_started and not reasoning_closed:
+                    yield "assistantfinal"
+                    reasoning_closed = True
+
+                if not saw_visible_output:
+                    msg = "I attempted to call a tool, but tools aren't executed in this environment, so no final answer was produced."
+                    if last_tool_name:
+                        try:
+                            args_text = json.dumps(last_tool_args, ensure_ascii=False, default=str)
+                        except Exception:
+                            args_text = str(last_tool_args)
+                        msg += f"\n\n• Tool requested: **{last_tool_name}**\n• Arguments: `{args_text}`"
+                    yield msg
+
+                if chunk.choices[0].finish_reason == "error":
+                    yield f"Error: Unknown error"
+                break
+
+        if buffer:
+            yield buffer
+
+    except Exception as e:
+        logger.exception("[Gateway] Streaming failed")
+        yield f"Error: {e}"
+
+# وظيفة التنسيق النهائي
+def format_final(analysis_text: str, visible_text: str) -> str:
+    """Render final message with collapsible analysis + normal Markdown answer."""
+    reasoning_safe = html.escape((analysis_text or "").strip())
+    response = (visible_text or "").strip()
+    return (
+        "<details><summary><strong>🤔 Analysis</strong></summary>\n"
+        "<pre style='white-space:pre-wrap;'>"
+        f"{reasoning_safe}"
+        "</pre>\n</details>\n\n"
+        "**💬 Response:**\n\n"
+        f"{response}"
+    )
+
+# وظيفة التوليد مع محاكاة streaming
+def generate(message, history, system_prompt, temperature, reasoning_effort, enable_browsing, max_new_tokens):
+    if not message.strip():
+        yield "Please enter a prompt."
+        return
+
+    # اختيار النموذج تلقائيًا
+    model_name = select_model(message)
+
+    # Flatten gradio history وتنظيف metadata
+    chat_history = []
+    for h in history:
+        if isinstance(h, dict):
+            clean_msg = {"role": h.get("role"), "content": h.get("content")}
+            if clean_msg["content"]:
+                chat_history.append(clean_msg)
+        elif isinstance(h, (list, tuple)) and len(h) == 2:
+            u, a = h
+            if u: chat_history.append({"role": "user", "content": u})
+            if a: chat_history.append({"role": "assistant", "content": a})
+
+    # إعداد الأدوات
+    tools = [
+        {
+            "type": "function",
+            "function": {
+                "name": "web_search_preview",
+                "description": "Perform a web search to gather additional context",
+                "parameters": {
+                    "type": "object",
+                    "properties": {"query": {"type": "string", "description": "Search query"}},
+                    "required": ["query"],
+                },
+            },
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "code_generation",
+                "description": "Generate or modify code for various frameworks (React, Django, Flask, etc.)",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "code": {"type": "string", "description": "Existing code to modify or empty for new code"},
+                        "framework": {"type": "string", "description": "Framework (e.g., React, Django, Flask)"},
+                        "task": {"type": "string", "description": "Task description (e.g., create a component, fix a bug)"},
+                    },
+                    "required": ["task"],
+                },
+            },
+        }
+    ] if "gpt-oss" in model_name else []
+    tool_choice = "auto" if "gpt-oss" in model_name else "none"
+
+    in_analysis = False
+    in_visible = False
+    raw_analysis = ""
+    raw_visible = ""
+    raw_started = False
+    last_flush_len = 0
+
+    def make_raw_preview() -> str:
+        return (
+            "```text\n"
+            "Analysis (live):\n"
+            f"{raw_analysis}\n\n"
+            "Response (draft):\n"
+            f"{raw_visible}\n"
+            "```"
+        )
+
+    try:
+        # استدعاء request_generation
+        stream = request_generation(
+            api_key=HF_TOKEN,
+            api_base=API_ENDPOINT,
+            message=message,
+            system_prompt=system_prompt,
+            model_name=model_name,
+            chat_history=chat_history,
+            temperature=temperature,
+            max_new_tokens=max_new_tokens,
+            tools=tools,
+            tool_choice=tool_choice,
+            deep_search=enable_browsing or model_name == MODEL_NAME,
+        )
+
+        for chunk in stream:
+            if chunk == "analysis":
+                in_analysis, in_visible = True, False
+                if not raw_started:
+                    raw_started = True
+                    yield make_raw_preview()
+                continue
+            if chunk == "assistantfinal":
+                in_analysis, in_visible = False, True
+                if not raw_started:
+                    raw_started = True
+                    yield make_raw_preview()
+                continue
+
+            if in_analysis:
+                raw_analysis += chunk
+            elif in_visible:
+                raw_visible += chunk
+            else:
+                raw_visible += chunk
+
+            total_len = len(raw_analysis) + len(raw_visible)
+            if total_len - last_flush_len >= 120 or "\n" in chunk:
+                last_flush_len = total_len
+                yield make_raw_preview()
+
+        final_markdown = format_final(raw_analysis, raw_visible)
+        if final_markdown.count("$") % 2:
+            final_markdown += "$"
+        yield final_markdown
+
+    except Exception as e:
+        logger.exception("Stream failed")
+        yield f"❌ Error: {e}"
+
+# إعداد CSS
+css = """
+.gradio-container { max-width: 800px; margin: auto; }
+.chatbot { border: 1px solid #ccc; border-radius: 10px; }
+.input-textbox { font-size: 16px; }
+"""
+
+# إعداد واجهة Gradio
+chatbot_ui = gr.ChatInterface(
+    fn=generate,
+    type="messages",
+    chatbot=gr.Chatbot(
+        label="MGZon Chatbot",
+        type="messages",
+        height=600,
+        latex_delimiters=LATEX_DELIMS,
+    ),
+    additional_inputs_accordion=gr.Accordion("⚙️ Settings", open=True),
+    additional_inputs=[
+        gr.Textbox(label="System prompt", value="You are a helpful assistant capable of code generation, analysis, review, and more.", lines=2),
+        gr.Slider(label="Temperature", minimum=0.0, maximum=1.0, step=0.1, value=0.9),
+        gr.Radio(label="Reasoning Effort", choices=["low", "medium", "high"], value="medium"),
+        gr.Checkbox(label="Enable DeepSearch (web browsing)", value=True),
+        gr.Slider(label="Max New Tokens", minimum=50, maximum=2048, step=50, value=2048),
+    ],
+    stop_btn="Stop",
+    examples=[
+        ["Explain the difference between supervised and unsupervised learning."],
+        ["Generate a React component for a login form."],
+        ["Review this Python code: print('Hello World')"],
+        ["Analyze the performance of a Django REST API."],
+        ["Tell me about MGZon products and services."],
+        ["Create a Flask route for user authentication."],
+        ["What are the latest trends in AI?"],
+        ["Provide guidelines for publishing a technical blog post."],
+    ],
+    title="MGZon Chatbot",
+    description="A versatile chatbot powered by GPT-OSS-120B and MGZon-Flan-T5-Base (auto-selected based on query). Supports code generation, analysis, review, web search, and MGZon-specific queries. Licensed under Apache 2.0. ***DISCLAIMER:*** Analysis may contain internal thoughts not suitable for final response.",
+    theme="gradio/soft",
+    css=css,
+)
+
+# دمج FastAPI مع Gradio
+from fastapi import FastAPI
+from gradio import mount_gradio_app
+
+app = FastAPI(title="MGZon Chatbot API")
+app = mount_gradio_app(app, chatbot_ui, path="/")
+
+# تشغيل الخادم
+if __name__ == "__main__":
+    import uvicorn
+    chatbot_ui.queue(max_size=QUEUE_SIZE, concurrency_count=CONCURRENCY_LIMIT).launch(
+        server_name="0.0.0.0", server_port=int(os.getenv("PORT", 7860)), share=False
+    )