Create app.py

app.py

@@ -0,0 +1,620 @@
+import gradio as gr
+import os
+import json
+import torch
+from transformers import (
+    AutoTokenizer, AutoModelForCausalLM, 
+    TrainingArguments, Trainer, 
+    DataCollatorForLanguageModeling,
+    pipeline
+)
+from datasets import Dataset
+from huggingface_hub import HfApi, login
+import spaces
+from typing import Optional, Dict, Any, List, Tuple
+import logging
+import traceback
+from datetime import datetime
+import random
+import re
+from faker import Faker
+import hashlib
+import time
+from collections import defaultdict
+from functools import wraps
+
+# Setup logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+# ==================== RATE LIMITING ====================
+
+class RateLimiter:
+    """Token bucket rate limiter"""
+    
+    def __init__(self):
+        self.requests = defaultdict(list)
+        self.limits = {
+            'synthetic_generation': {'calls': 10, 'period': 3600},
+            'model_training': {'calls': 3, 'period': 3600},
+            'model_inference': {'calls': 50, 'period': 3600},
+        }
+    
+    def _get_user_id(self, request: gr.Request) -> str:
+        if request:
+            identifier = f"{request.client.host}_{request.headers.get('user-agent', '')}"
+            return hashlib.md5(identifier.encode()).hexdigest()
+        return "anonymous"
+    
+    def _clean_old_requests(self, user_id: str, endpoint: str):
+        if user_id not in self.requests:
+            return
+        current_time = time.time()
+        period = self.limits[endpoint]['period']
+        self.requests[user_id] = [
+            req for req in self.requests[user_id]
+            if req['endpoint'] == endpoint and current_time - req['timestamp'] < period
+        ]
+    
+    def check_rate_limit(self, user_id: str, endpoint: str) -> Tuple[bool, str]:
+        self._clean_old_requests(user_id, endpoint)
+        user_requests = [req for req in self.requests[user_id] if req['endpoint'] == endpoint]
+        limit = self.limits[endpoint]['calls']
+        period = self.limits[endpoint]['period']
+        
+        if len(user_requests) >= limit:
+            time_until_reset = period - (time.time() - user_requests[0]['timestamp'])
+            minutes = int(time_until_reset / 60)
+            return False, f"⏱️ Rate limit exceeded! Please wait {minutes} minutes."
+        
+        self.requests[user_id].append({'endpoint': endpoint, 'timestamp': time.time()})
+        remaining = limit - len(user_requests) - 1
+        return True, f"✅ Request accepted ({remaining} remaining this hour)"
+
+rate_limiter = RateLimiter()
+
+def rate_limit(endpoint: str):
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            request = kwargs.get('request', None)
+            if request:
+                user_id = rate_limiter._get_user_id(request)
+                allowed, message = rate_limiter.check_rate_limit(user_id, endpoint)
+                if not allowed:
+                    return f"🚫 {message}"
+            return func(*args, **kwargs)
+        return wrapper
+    return decorator
+
+# ==================== AUTHENTICATION ====================
+
+class AuthManager:
+    def __init__(self):
+        self.authenticated_tokens = {}
+        self.token_expiry = 86400
+    
+    def validate_hf_token(self, token: str) -> Tuple[bool, str, Optional[str]]:
+        try:
+            if not token or not token.strip():
+                return False, "❌ Please provide a HuggingFace token", None
+            
+            token_hash = hashlib.sha256(token.encode()).hexdigest()
+            if token_hash in self.authenticated_tokens:
+                cached = self.authenticated_tokens[token_hash]
+                if time.time() - cached['timestamp'] < self.token_expiry:
+                    return True, f"✅ Welcome back, {cached['username']}!", cached['username']
+            
+            api = HfApi(token=token)
+            user_info = api.whoami()
+            username = user_info.get('name', 'Anonymous Architect')
+            
+            self.authenticated_tokens[token_hash] = {
+                'username': username,
+                'timestamp': time.time()
+            }
+            
+            return True, f"🎉 Welcome, {username}!", username
+            
+        except Exception as e:
+            return False, f"🔐 Token validation failed: {str(e)}", None
+
+auth_manager = AuthManager()
+
+# ==================== ERROR HANDLING ====================
+
+class ArchitechError(Exception):
+    pass
+
+class DataGenerationError(ArchitechError):
+    pass
+
+class ModelTrainingError(ArchitechError):
+    pass
+
+class ModelInferenceError(ArchitechError):
+    pass
+
+def handle_errors(error_type: str = "general"):
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            try:
+                return func(*args, **kwargs)
+            except torch.cuda.OutOfMemoryError:
+                return "🔥 **GPU Memory Overflow!** Try: smaller batch size, smaller model, or less data."
+            except PermissionError:
+                return "🔒 **Permission Denied!** Check your HuggingFace token has WRITE access."
+            except ConnectionError:
+                return "🌐 **Connection Issue!** Can't reach HuggingFace. Check your network."
+            except ValueError as e:
+                return f"⚠️ **Invalid Input!** {str(e)}"
+            except (DataGenerationError, ModelTrainingError, ModelInferenceError) as e:
+                return f"🔧 **Architech Error:** {str(e)}"
+            except Exception as e:
+                logger.error(f"Error in {func.__name__}: {traceback.format_exc()}")
+                return f"💥 **Unexpected Error:** {str(e)}"
+        return wrapper
+    return decorator# ==================== SYNTHETIC DATA GENERATOR ====================
+
+class SyntheticDataGenerator:
+    def __init__(self):
+        self.faker = Faker()
+        self.generation_templates = {
+            "conversational": [
+                "Human: {question}\nAssistant: {answer}",
+                "User: {question}\nBot: {answer}",
+            ],
+            "instruction": [
+                "### Instruction:\n{instruction}\n\n### Response:\n{response}",
+            ],
+        }
+        
+        self.domain_knowledge = {
+            "technology": {
+                "topics": ["AI", "machine learning", "cloud computing"],
+                "concepts": ["algorithms", "APIs", "databases"],
+                "contexts": ["software development", "digital transformation"]
+            },
+            "healthcare": {
+                "topics": ["telemedicine", "diagnostics", "patient care"],
+                "concepts": ["treatments", "procedures"],
+                "contexts": ["clinical practice", "patient education"]
+            },
+            "finance": {
+                "topics": ["fintech", "investment", "risk management"],
+                "concepts": ["portfolios", "compliance"],
+                "contexts": ["financial advisory", "personal finance"]
+            },
+            "general": {
+                "topics": ["communication", "problem-solving"],
+                "concepts": ["strategies", "best practices"],
+                "contexts": ["daily life", "personal growth"]
+            }
+        }
+    
+    def _generate_question(self, topic, concept, context):
+        templates = [
+            f"How does {concept} work in {context}?",
+            f"What are the benefits of {concept} for {topic}?",
+            f"Can you explain {concept}?",
+            f"What's the best approach to {concept}?"
+        ]
+        return random.choice(templates)
+    
+    def _generate_answer(self, question, topic, concept):
+        templates = [
+            f"{concept} in {topic} works through strategic implementation. Key benefits include improved efficiency and better outcomes.",
+            f"Great question! {concept} is fundamental because it addresses core challenges. Best practices include planning and testing.",
+            f"When it comes to {concept}, consider scalability and performance. Success depends on proper implementation."
+        ]
+        return random.choice(templates)
+    
+    def _generate_single_example(self, task_desc, domain_data, templates, complexity):
+        template = random.choice(templates)
+        topic = random.choice(domain_data["topics"])
+        concept = random.choice(domain_data["concepts"])
+        context = random.choice(domain_data["contexts"])
+        
+        question = self._generate_question(topic, concept, context)
+        answer = self._generate_answer(question, topic, concept)
+        
+        text = template.format(question=question, answer=answer)
+        return {"text": text}
+    
+    @handle_errors("data_generation")
+    def generate_synthetic_dataset(
+        self,
+        task_description: str,
+        domain: str,
+        dataset_size: int = 100,
+        format_type: str = "conversational",
+        complexity: str = "medium",
+        progress=gr.Progress()
+    ) -> str:
+        if not task_description or len(task_description.strip()) < 10:
+            raise DataGenerationError("Task description too short! Need at least 10 characters.")
+        
+        if dataset_size < 10 or dataset_size > 1000:
+            raise DataGenerationError("Dataset size must be between 10 and 1000.")
+        
+        progress(0.1, f"🎯 Generating {dataset_size} examples...")
+        
+        domain_data = self.domain_knowledge.get(domain, self.domain_knowledge["general"])
+        templates = self.generation_templates.get(format_type, self.generation_templates["conversational"])
+        
+        synthetic_data = []
+        for i in range(dataset_size):
+            if i % 20 == 0:
+                progress(0.1 + (0.7 * i / dataset_size), f"📝 Creating {i+1}/{dataset_size}...")
+            
+            example = self._generate_single_example(task_description, domain_data, templates, complexity)
+            synthetic_data.append(example)
+        
+        os.makedirs("./synthetic_datasets", exist_ok=True)
+        dataset_filename = f"synthetic_{domain}_{format_type}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
+        dataset_path = os.path.join("./synthetic_datasets", dataset_filename)
+        
+        with open(dataset_path, 'w') as f:
+            json.dump(synthetic_data, f, indent=2)
+        
+        preview = "\n\n---\n\n".join([ex["text"] for ex in synthetic_data[:3]])
+        
+        return f"""🎊 **SYNTHETIC DATASET GENERATED!**
+
+**Dataset Details:**
+- 📊 Size: {len(synthetic_data)} examples
+- 🎯 Domain: {domain.title()}
+- 📝 Format: {format_type.title()}
+- 💾 Saved as: `{dataset_filename}`
+
+**Preview (First 3 Examples):**
+
+{preview}
+
+**Next Steps:** Use this in the 'Train Model' or 'Test Model' tabs!"""# ==================== MODEL INFERENCE ====================
+
+class ModelInference:
+    def __init__(self):
+        self.loaded_models = {}
+    
+    @handle_errors("inference")
+    def load_model(self, model_name: str, hf_token: str, progress=gr.Progress()) -> str:
+        progress(0.1, "🔍 Locating your model...")
+        
+        is_valid, message, username = auth_manager.validate_hf_token(hf_token)
+        if not is_valid:
+            raise ModelInferenceError(message)
+        
+        full_model_name = f"{username}/{model_name}" if "/" not in model_name else model_name
+        
+        progress(0.3, "📥 Downloading model...")
+        
+        try:
+            tokenizer = AutoTokenizer.from_pretrained(full_model_name, token=hf_token)
+            model = AutoModelForCausalLM.from_pretrained(
+                full_model_name,
+                token=hf_token,
+                torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
+                device_map="auto" if torch.cuda.is_available() else None
+            )
+            
+            self.loaded_models[model_name] = {
+                'model': model,
+                'tokenizer': tokenizer,
+                'pipeline': pipeline('text-generation', model=model, tokenizer=tokenizer)
+            }
+            
+            progress(1.0, "✅ Model loaded!")
+            return f"✅ **Model Loaded Successfully!**\n\nModel: `{full_model_name}`\n\nReady for inference!"
+            
+        except Exception as e:
+            raise ModelInferenceError(f"Failed to load model: {str(e)}")
+    
+    @handle_errors("inference")
+    def generate_text(
+        self,
+        model_name: str,
+        prompt: str,
+        max_length: int = 100,
+        temperature: float = 0.7,
+        top_p: float = 0.9
+    ) -> str:
+        if model_name not in self.loaded_models:
+            raise ModelInferenceError("Model not loaded! Please load the model first.")
+        
+        if not prompt or len(prompt.strip()) < 3:
+            raise ModelInferenceError("Prompt too short! Please provide at least 3 characters.")
+        
+        pipe = self.loaded_models[model_name]['pipeline']
+        
+        result = pipe(
+            prompt,
+            max_length=max_length,
+            temperature=temperature,
+            top_p=top_p,
+            do_sample=True,
+            num_return_sequences=1
+        )
+        
+        generated_text = result[0]['generated_text']
+        
+        return f"""**🎯 Generated Response:**
+
+{generated_text}
+
+---
+*Model: {model_name} | Length: {len(generated_text)} chars*"""
+
+model_inference = ModelInference()# ==================== ARCHITECH AGENT ====================
+
+class ArchitechAgent:
+    def __init__(self):
+        self.hf_api = HfApi()
+        self.synthetic_generator = SyntheticDataGenerator()
+        self.personality_responses = [
+            "🎯 Let's cook up some AI magic!",
+            "🚀 Time to turn your vision into reality!",
+            "🧠 Let's architect some brilliance!",
+        ]
+    
+    def get_personality_response(self) -> str:
+        return random.choice(self.personality_responses)
+    
+    @rate_limit('synthetic_generation')
+    @handle_errors("data_generation")
+    def generate_synthetic_dataset_wrapper(self, *args, **kwargs):
+        return self.synthetic_generator.generate_synthetic_dataset(*args, **kwargs)
+    
+    @spaces.GPU
+    @rate_limit('model_training')
+    @handle_errors("training")
+    def train_custom_model(
+        self,
+        task_description: str,
+        training_data: str,
+        model_name: str,
+        hf_token: str,
+        base_model: str = "distilgpt2",
+        use_synthetic_data: bool = True,
+        synthetic_domain: str = "general",
+        synthetic_size: int = 100,
+        learning_rate: float = 2e-4,
+        num_epochs: int = 3,
+        batch_size: int = 2,
+        progress=gr.Progress()
+    ) -> str:
+        
+        is_valid, message, username = auth_manager.validate_hf_token(hf_token)
+        if not is_valid:
+            raise ModelTrainingError(message)
+        
+        progress(0.1, "🧠 Loading base model...")
+        
+        tokenizer = AutoTokenizer.from_pretrained(base_model)
+        if tokenizer.pad_token is None:
+            tokenizer.pad_token = tokenizer.eos_token
+        
+        model = AutoModelForCausalLM.from_pretrained(
+            base_model,
+            torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
+            device_map="auto" if torch.cuda.is_available() else None
+        )
+        
+        if use_synthetic_data:
+            progress(0.2, "🎨 Generating synthetic data...")
+            result = self.synthetic_generator.generate_synthetic_dataset(
+                task_description, synthetic_domain, synthetic_size, "conversational", "medium", progress
+            )
+            
+            dataset_files = [f for f in os.listdir("./synthetic_datasets") if f.endswith('.json')]
+            if not dataset_files:
+                raise ModelTrainingError("No synthetic dataset found!")
+            
+            latest_dataset = max(dataset_files, key=lambda x: os.path.getctime(os.path.join("./synthetic_datasets", x)))
+            with open(os.path.join("./synthetic_datasets", latest_dataset), 'r') as f:
+                synthetic_data = json.load(f)
+            texts = [item["text"] for item in synthetic_data]
+        else:
+            texts = [t.strip() for t in training_data.split("\n\n") if t.strip()]
+        
+        if not texts:
+            raise ModelTrainingError("No training data available!")
+        
+        progress(0.3, f"✨ Tokenizing {len(texts)} examples...")
+        
+        dataset = Dataset.from_dict({"text": texts})
+        
+        def tokenize_function(examples):
+            return tokenizer(examples["text"], truncation=True, padding=True, max_length=256)
+        
+        tokenized_dataset = dataset.map(tokenize_function, batched=True, remove_columns=["text"])
+        
+        progress(0.4, "⚙️ Configuring training...")
+        
+        training_args = TrainingArguments(
+            output_dir=f"./results_{model_name}",
+            num_train_epochs=num_epochs,
+            per_device_train_batch_size=batch_size,
+            gradient_accumulation_steps=4,
+            learning_rate=learning_rate,
+            logging_steps=50,
+            save_steps=500,
+            save_total_limit=2,
+            fp16=torch.cuda.is_available(),
+            report_to="none"
+        )
+        
+        data_collator = DataCollatorForLanguageModeling(tokenizer=tokenizer, mlm=False)
+        
+        trainer = Trainer(
+            model=model,
+            args=training_args,
+            train_dataset=tokenized_dataset,
+            data_collator=data_collator,
+        )
+        
+        progress(0.6, "💪 Training in progress...")
+        trainer.train()
+        
+        progress(0.8, "💾 Saving model...")
+        output_dir = f"./trained_{model_name}"
+        trainer.save_model(output_dir)
+        tokenizer.save_pretrained(output_dir)
+        
+        progress(0.9, "📤 Pushing to HuggingFace...")
+        
+        try:
+            login(token=hf_token)
+            model.push_to_hub(model_name, token=hf_token)
+            tokenizer.push_to_hub(model_name, token=hf_token)
+            hub_url = f"https://huggingface.co/{username}/{model_name}"
+            
+            return f"""🎉 **TRAINING COMPLETE!**
+
+✅ Training successful
+💾 Model saved locally
+📤 Pushed to Hub
+🔗 **Your model:** {hub_url}
+
+**Stats:**
+- Examples: {len(texts)}
+- Epochs: {num_epochs}
+- Learning rate: {learning_rate}
+
+**Test it in the 'Test Model' tab!**"""
+        
+        except Exception as e:
+            return f"✅ Training done but upload failed: {str(e)}\nModel saved at: {output_dir}"# ==================== GRADIO INTERFACE ====================
+
+def create_gradio_interface():
+    agent = ArchitechAgent()
+    
+    with gr.Blocks(title="🏗️ Architech", theme=gr.themes.Soft()) as demo:
+        gr.Markdown("""
+        # 🏗️ **Architech - Your AI Model Architect**
+        
+        *Describe what you want, and I'll build it for you!*
+        """)
+        
+        with gr.Tabs():
+            # Generate Dataset
+            with gr.Tab("📊 Generate Dataset"):
+                with gr.Row():
+                    with gr.Column():
+                        task_desc = gr.Textbox(label="Task Description", lines=3, 
+                            placeholder="E.g., 'Customer support chatbot for tech products'")
+                        domain = gr.Dropdown(
+                            choices=["technology", "healthcare", "finance", "general"],
+                            label="Domain", value="general")
+                        dataset_size = gr.Slider(50, 500, 100, step=50, label="Dataset Size")
+                        format_type = gr.Dropdown(
+                            choices=["conversational", "instruction"],
+                            label="Format", value="conversational")
+                        gen_btn = gr.Button("🎨 Generate Dataset", variant="primary")
+                    with gr.Column():
+                        gen_output = gr.Markdown()
+                
+                gen_btn.click(
+                    fn=agent.generate_synthetic_dataset_wrapper,
+                    inputs=[task_desc, domain, dataset_size, format_type, gr.State("medium")],
+                    outputs=gen_output
+                )
+            
+            # Train Model
+            with gr.Tab("🚀 Train Model"):
+                with gr.Row():
+                    with gr.Column():
+                        task_desc_train = gr.Textbox(label="Task Description", lines=2)
+                        model_name = gr.Textbox(label="Model Name", placeholder="my-awesome-model")
+                        hf_token = gr.Textbox(label="HuggingFace Token", type="password")
+                        use_synthetic = gr.Checkbox(label="Use Synthetic Data", value=True)
+                        
+                        with gr.Accordion("⚙️ Advanced", open=False):
+                            base_model = gr.Dropdown(
+                                choices=["distilgpt2", "gpt2", "microsoft/DialoGPT-small"],
+                                label="Base Model", value="distilgpt2")
+                            learning_rate = gr.Slider(1e-5, 5e-4, 2e-4, label="Learning Rate")
+                            num_epochs = gr.Slider(1, 5, 3, step=1, label="Epochs")
+                            batch_size = gr.Slider(1, 4, 2, step=1, label="Batch Size")
+                        
+                        train_btn = gr.Button("🎯 Train Model", variant="primary")
+                    
+                    with gr.Column():
+                        train_output = gr.Markdown()
+                
+                train_btn.click(
+                    fn=agent.train_custom_model,
+                    inputs=[task_desc_train, gr.State(""), model_name, hf_token,
+                           base_model, use_synthetic, gr.State("general"), 
+                           gr.State(100), learning_rate, num_epochs, batch_size],
+                    outputs=train_output
+                )
+            
+            # Test Model
+            with gr.Tab("🧪 Test Model"):
+                with gr.Row():
+                    with gr.Column():
+                        test_model_name = gr.Textbox(label="Model Name", 
+                            placeholder="username/model-name")
+                        test_token = gr.Textbox(label="HuggingFace Token", type="password")
+                        load_btn = gr.Button("📥 Load Model")
+                        
+                        gr.Markdown("---")
+                        
+                        test_prompt = gr.Textbox(label="Test Prompt", lines=3,
+                            placeholder="Enter your prompt here...")
+                        max_length = gr.Slider(50, 200, 100, label="Max Length")
+                        temperature = gr.Slider(0.1, 1.0, 0.7, label="Temperature")
+                        
+                        test_btn = gr.Button("🎯 Generate", variant="primary")
+                    
+                    with gr.Column():
+                        load_output = gr.Markdown()
+                        test_output = gr.Markdown()
+                
+                load_btn.click(
+                    fn=model_inference.load_model,
+                    inputs=[test_model_name, test_token],
+                    outputs=load_output
+                )
+                
+                test_btn.click(
+                    fn=model_inference.generate_text,
+                    inputs=[test_model_name, test_prompt, max_length, temperature, gr.State(0.9)],
+                    outputs=test_output
+                )
+            
+            # About
+            with gr.Tab("ℹ️ About"):
+                gr.Markdown("""
+                ## 🏗️ Architech - Your AI Model Architect
+                
+                ### Features:
+                - 🎨 **Generate Synthetic Data**: No training data? No problem!
+                - 🚀 **Train Custom Models**: Fine-tune models for your specific needs
+                - 🧪 **Test Your Models**: Load and test your models instantly
+                - ⚡ **Rate Limited**: Fair usage for all users
+                - 🔒 **Secure**: Token-based authentication
+                
+                ### How to Use:
+                1. Generate synthetic training data for your task
+                2. Train a custom model with your data
+                3. Test and deploy your model!
+                
+                ### Rate Limits:
+                - Dataset Generation: 10 per hour
+                - Model Training: 3 per hour
+                - Model Inference: 50 per hour
+                
+                *Built with ❤️ using Gradio, Transformers, and HuggingFace*
+                """)
+    
+    return demo
+
+if __name__ == "__main__":
+    demo = create_gradio_interface()
+    demo.launch()