app.py

# -*- coding: utf-8 -*-

import os
import torch
import transformers
from transformers import pipeline
import gradio as gr
import spaces

# Load system prompt from file
def load_system_prompt():
    try:
        with open('system_prompt.txt', 'r', encoding='utf-8') as f:
            return f.read().strip()
    except FileNotFoundError:
        return "أنت مساعد ذكي مفيد."

DEFAULT_SYSTEM_PROMPT = load_system_prompt()

model_path = "unsloth/gemma-3-4b-it-unsloth-bnb-4bit"

# إذا كان فيه HF_TOKEN في البيئة
hf_token = os.getenv("HF_TOKEN")

# Initialize pipeline for chat
pipeline_model = pipeline(
    "text-generation",
    model=model_path,
    device_map="auto",
    token=hf_token,
    trust_remote_code=True
)

def generate_with_pipeline(messages, max_new_tokens=256, temperature=0.7, top_p=0.9, top_k=50, repetition_penalty=1.0):
    """Generate response using the pipeline with messages format"""
    # Apply chat template
    prompt = pipeline_model.tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True
    )

    outputs = pipeline_model(
        prompt,
        max_new_tokens=max_new_tokens,
        temperature=temperature,
        top_p=top_p,
        top_k=top_k,
        repetition_penalty=repetition_penalty,
        do_sample=True,
        return_full_text=False
    )
    return outputs[0]["generated_text"]


@spaces.GPU()
def generate_response(message, history, max_new_tokens, temperature, top_p, top_k, repetition_penalty):
    """
    Generate response with full conversation history
    
    Args:
        message: Current user message (dict with 'text' key when type="messages")
        history: List of previous messages (already in correct format for type="messages")
        max_new_tokens, temperature, top_p, top_k, repetition_penalty: Generation parameters
    """
    try:
        # Build messages list starting with system prompt
        messages = [{"role": "system", "content": DEFAULT_SYSTEM_PROMPT}]
        
        # Add conversation history
        # When type="messages", history is a list of message dicts with 'role' and 'content'
        if history:
            for msg in history:
                if isinstance(msg, dict) and 'role' in msg and 'content' in msg:
                    messages.append({"role": msg['role'], "content": msg['content']})
        
        # Add current user message
        if isinstance(message, dict):
            current_message = message.get("text", "") or message.get("content", "")
        else:
            current_message = str(message)
        
        messages.append({"role": "user", "content": current_message})
        
        # Debug: print messages structure
        print(f"Messages sent to model: {len(messages)} messages")
        
        # Generate response
        response = generate_with_pipeline(
            messages,
            max_new_tokens=max_new_tokens,
            temperature=temperature,
            top_p=top_p,
            top_k=top_k,
            repetition_penalty=repetition_penalty
        )

        if not response or response.strip() == "":
            response = "أهلاً! أنا أليكس مساعد خدمة العملاء. كيف أقدر أساعدك اليوم؟"

        return response

    except Exception as e:
        print(f"Error in generate_response: {e}")
        import traceback
        print(traceback.format_exc())
        return "عذراً، حدث خطأ. يرجى المحاولة مرة أخرى."


# Create Gradio interface
demo = gr.ChatInterface(
    fn=generate_response,
    additional_inputs=[
        gr.Slider(label="الحد الأقصى للكلمات الجديدة", minimum=64, maximum=4096, step=1, value=2048),
        gr.Slider(label="درجة الحرارة", minimum=0.1, maximum=2.0, step=0.1, value=0.7),
        gr.Slider(label="Top-p", minimum=0.05, maximum=1.0, step=0.05, value=0.9),
        gr.Slider(label="Top-k", minimum=1, maximum=100, step=1, value=50),
        gr.Slider(label="عقوبة التكرار", minimum=1.0, maximum=2.0, step=0.05, value=1.0)
    ],
    examples=[
        ["النت عندي معطل من الصبح، تقدر تساعدني؟"],
        ["عندي مشكلة بالاتصال بالواي فاي"],
        ["شنو الباقات المتوفرة عندكم؟"],
        ["كيف أعيد ضبط الجهاز؟"],
        ["My device is not working properly"],
    ],
    cache_examples=False,
    type="messages",
    title="دعم عملاء TechSolutions - مساعد أليكس (العراقي)",
    description="""🤖 مساعد خدمة عملاء ذكي لـ TechSolutions

✨ المميزات:
- 🌐 دعم ثنائي اللغة (عربي وإنجليزي)
- 💬 لهجة محادثة طبيعية
- 🔧 دعم فني واستكشاف الأخطاء
- 📋 معلومات الخدمات والإرشاد
- 🧠 **يتذكر المحادثة السابقة** - يمكنك الرجوع للمواضيع السابقة
- 🎯 مدعوم بـ موديل Unsloth Meta-Llama-3.1-8B-Instruct-bnb-4bit

احجي مع أليكس لحل مشاكلك التقنية، استفسر عن الخدمات، أو احصل على معلومات المنتجات.""",
    fill_height=True,
    textbox=gr.Textbox(
        label="اكتب رسالتك هنا",
        placeholder="مثال: عندي مشكلة بالجهاز..."
    ),
    stop_btn="إيقاف التوليد",
    multimodal=False,
    theme=gr.themes.Soft()
)

if __name__ == "__main__":
    demo.launch()