Initial commit for Hugging Face deployment

.env.example

@@ -0,0 +1,6 @@
+APP_NAME="ML/AI Models API Service"
+DEBUG=true
+HOST="0.0.0.0"
+PORT=8000
+MODEL_CACHE_DIR="model_cache"
+MAX_BATCH_SIZE=32 

.gitignore

@@ -0,0 +1,43 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Virtual Environment
+venv/
+ENV/
+env/
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+
+# Logs
+*.log
+
+# Local development
+.env
+.env.local
+flagged/
+
+# Gradio
+gradio_cached_examples/ 

README-HF.md

@@ -0,0 +1,32 @@
+# Text Classification Model Demo
+
+This is a Gradio interface for text classification using a BERT-based model. The model can classify text into predefined categories.
+
+## Model Details
+
+- Base Model: prajjwal1/bert-tiny
+- Task: Text Classification
+- Interface: Gradio
+
+## Usage
+
+1. Enter your text in the input textbox
+2. Click submit
+3. View the classification results
+
+## Technical Details
+
+- Python 3.9+
+- Key Dependencies:
+  - gradio
+  - transformers
+  - torch
+  - numpy
+
+## Deployment
+
+This model is deployed using Hugging Face Spaces with a Gradio interface.
+
+## License
+
+MIT License 

README.md

@@ -0,0 +1,66 @@
+# ML/AI Models API Service
+
+A centralized service that hosts various machine learning and AI models using Gradio interfaces, exposed via REST APIs for external frontend clients.
+
+## Features
+
+- Multiple ML/AI model endpoints
+- Gradio interfaces for each model
+- FastAPI backend for API exposure
+- Easy model management and deployment
+- Scalable architecture
+
+## Setup
+
+1. Create a virtual environment:
+```bash
+python -m venv venv
+source venv/bin/activate  # On Windows: venv\Scripts\activate
+```
+
+2. Install dependencies:
+```bash
+pip install -r requirements.txt
+```
+
+3. Create a `.env` file:
+```bash
+cp .env.example .env
+```
+
+4. Run the development server:
+```bash
+python src/main.py
+```
+
+## Project Structure
+
+```
+├── src/
+│   ├── main.py              # Main application entry point
+│   ├── config.py            # Configuration settings
+│   ├── models/              # ML/AI model implementations
+│   │   ├── __init__.py
+│   │   └── base.py         # Base model class
+│   ├── interfaces/          # Gradio interfaces
+│   │   └── __init__.py
+│   └── api/                 # FastAPI routes
+│       └── __init__.py
+├── tests/                   # Test files
+├── .env.example            # Example environment variables
+├── requirements.txt        # Project dependencies
+└── README.md              # This file
+```
+
+## Adding New Models
+
+1. Add your model implementation in `src/models/`
+2. Create a Gradio interface in `src/interfaces/`
+3. Add API endpoints in `src/api/`
+4. Register the model in `src/main.py`
+
+## API Documentation
+
+Once the server is running, visit:
+- API documentation: `http://localhost:8000/docs`
+- Gradio interfaces: `http://localhost:8000/gradio` 

app.py

@@ -0,0 +1,39 @@
+import gradio as gr
+from src.models.text_classification import TextClassificationModel
+import logging
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+def create_demo():
+    try:
+        # Initialize the model
+        logger.info("Initializing Text Classification model...")
+        model = TextClassificationModel()
+        
+        # Create the interface
+        logger.info("Creating Gradio interface...")
+        demo = model.create_interface()
+        
+        logger.info("Gradio interface created successfully")
+        return demo
+        
+    except Exception as e:
+        logger.error(f"Error creating demo: {str(e)}")
+        raise
+
+if __name__ == "__main__":
+    try:
+        logger.info("Starting the application...")
+        demo = create_demo()
+        logger.info("Launching the interface...")
+        demo.launch(
+            server_name="0.0.0.0",  # Allow external connections
+            server_port=7860,       # Specify port explicitly
+            share=True             # Enable public URL
+        )
+        logger.info("Interface launched successfully")
+    except Exception as e:
+        logger.error(f"Application error: {str(e)}")
+        raise 

app_hf.py

@@ -0,0 +1,16 @@
+import gradio as gr
+from src.models.text_classification import TextClassificationModel
+import logging
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Initialize the model
+model = TextClassificationModel()
+
+# Create the interface
+demo = model.create_interface()
+
+# Launch the interface (Hugging Face will handle the server configuration)
+demo.launch() 

requirements-full.txt

@@ -0,0 +1,10 @@
+gradio>=4.19.2
+fastapi>=0.110.0
+uvicorn>=0.27.1
+python-dotenv>=1.0.1
+pydantic>=2.6.3
+numpy>=1.26.4
+torch>=2.2.1
+transformers>=4.38.2
+pillow>=10.2.0
+python-multipart>=0.0.9 

requirements.txt

@@ -0,0 +1,9 @@
+gradio>=4.19.2
+transformers>=4.38.2
+torch>=2.2.1
+numpy>=1.26.4
+fastapi>=0.110.0
+uvicorn>=0.27.1
+pydantic>=2.6.3
+pydantic-settings>=2.2.1
+python-dotenv>=1.0.1 

src/__init__.py

@@ -0,0 +1 @@
+"""ML Models API Service.""" 

src/api/__init__.py

@@ -0,0 +1 @@
+"""API endpoints for ML models.""" 

src/api/models.py

@@ -0,0 +1,42 @@
+from fastapi import APIRouter, HTTPException
+from typing import List, Dict
+from models.registry import GradioRegistry
+
+router = APIRouter()
+registry = GradioRegistry()
+
+@router.get("/models", response_model=List[Dict])
+async def list_models():
+    """List all available models."""
+    return registry.list_models()
+
+@router.get("/models/{model_id}")
+async def get_model_info(model_id: str):
+    """Get information about a specific model."""
+    model_info = registry.get_model_info(model_id)
+    if not model_info:
+        raise HTTPException(status_code=404, detail="Model not found")
+    return model_info
+
+@router.get("/models/{model_id}/status")
+async def get_model_status(model_id: str):
+    """Get the current status of a model."""
+    model_info = registry.get_model_info(model_id)
+    if not model_info:
+        raise HTTPException(status_code=404, detail="Model not found")
+    return {"status": model_info.status}
+
+@router.post("/models/{model_id}/load")
+async def load_model(model_id: str):
+    """Load a model into memory."""
+    model = registry.get_model(model_id)
+    if not model:
+        raise HTTPException(status_code=404, detail="Model not found")
+    
+    try:
+        model.load_model()
+        registry.update_model_status(model_id, "loaded")
+        return {"status": "loaded"}
+    except Exception as e:
+        registry.update_model_status(model_id, "error")
+        raise HTTPException(status_code=500, detail=str(e)) 

src/api/text_classification.py

@@ -0,0 +1,52 @@
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+from typing import Dict, Union, List
+from models.text_classification import TextClassificationModel
+
+router = APIRouter()
+model = TextClassificationModel()
+
+class TextInput(BaseModel):
+    text: str
+
+class BatchTextInput(BaseModel):
+    texts: List[str]
+
+class PredictionResponse(BaseModel):
+    label: str
+    confidence: float
+
+class BatchPredictionResponse(BaseModel):
+    predictions: List[PredictionResponse]
+
+@router.post("/predict", response_model=PredictionResponse)
+async def predict(input_data: TextInput) -> Dict[str, Union[str, float]]:
+    """Make a prediction for a single text."""
+    try:
+        result = await model.predict(input_data.text)
+        return result
+    except Exception as e:
+        raise HTTPException(
+            status_code=500,
+            detail=f"Prediction failed: {str(e)}"
+        )
+
+@router.post("/predict_batch", response_model=BatchPredictionResponse)
+async def predict_batch(input_data: BatchTextInput) -> Dict[str, List[Dict[str, Union[str, float]]]]:
+    """Make predictions for multiple texts."""
+    try:
+        predictions = []
+        for text in input_data.texts:
+            result = await model.predict(text)
+            predictions.append(result)
+        return {"predictions": predictions}
+    except Exception as e:
+        raise HTTPException(
+            status_code=500,
+            detail=f"Batch prediction failed: {str(e)}"
+        )
+
+@router.get("/info")
+async def get_model_info():
+    """Get information about the text classification model."""
+    return model.get_info() 

src/config.py

@@ -0,0 +1,23 @@
+from pydantic_settings import BaseSettings
+from pydantic import Field
+
+
+class Settings(BaseSettings):
+    """Application settings."""
+    app_name: str = "ML Models API"
+    debug: bool = Field(default=False, env="DEBUG")
+    host: str = Field(default="127.0.0.1", env="HOST")
+    port: int = Field(default=8000, env="PORT")
+    
+    # Add model-specific configurations here
+    MODEL_CACHE_DIR: str = "model_cache"
+    MAX_BATCH_SIZE: int = 32
+    
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+
+def get_settings() -> Settings:
+    """Get application settings."""
+    return Settings() 

src/main.py

@@ -0,0 +1,66 @@
+import uvicorn
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+import gradio as gr
+from config import get_settings
+from models.text_classification import TextClassificationModel
+from api.models import router as models_router, registry
+
+app = FastAPI(
+    title=get_settings().app_name,
+    description="API for managing and running ML models",
+    version="1.0.0",
+    docs_url="/docs",
+    redoc_url="/redoc",
+)
+
+# Configure CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Modify this in production
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Register models
+text_classifier = TextClassificationModel()
+registry.register_model(
+    "text-classification",
+    text_classifier,
+    "/gradio/text-classification"
+)
+
+# Mount the models API router
+app.include_router(
+    models_router,
+    prefix="/api/models",
+    tags=["models"]
+)
+
+# Mount Gradio interface
+app = gr.mount_gradio_app(
+    app, 
+    text_classifier.create_interface(), 
+    path="/gradio/text-classification"
+)
+
+@app.get("/")
+async def root():
+    """Root endpoint returning basic API information."""
+    return {
+        "name": get_settings().app_name,
+        "version": "1.0.0",
+        "status": "running"
+    }
+
+if __name__ == "__main__":
+    # Initialize settings
+    settings = get_settings()
+    
+    uvicorn.run(
+        "main:app",
+        host=settings.host,
+        port=settings.port,
+        reload=settings.debug
+    ) 

src/models/__init__.py

@@ -0,0 +1 @@
+"""ML model implementations.""" 

src/models/base.py

@@ -0,0 +1,42 @@
+from abc import ABC, abstractmethod
+from typing import Any, Dict, Optional
+import gradio as gr
+
+
+class BaseModel(ABC):
+    """Base class for all ML/AI models."""
+    
+    def __init__(self, name: str, description: str):
+        self.name = name
+        self.description = description
+        self._model: Optional[Any] = None
+        self._interface: Optional[gr.Interface] = None
+    
+    @abstractmethod
+    def load_model(self) -> None:
+        """Load the model into memory."""
+        pass
+    
+    @abstractmethod
+    def create_interface(self) -> gr.Interface:
+        """Create and return a Gradio interface for the model."""
+        pass
+    
+    @abstractmethod
+    async def predict(self, *args, **kwargs) -> Any:
+        """Make predictions using the model."""
+        pass
+    
+    def get_interface(self) -> gr.Interface:
+        """Get or create the Gradio interface."""
+        if self._interface is None:
+            self._interface = self.create_interface()
+        return self._interface
+    
+    def get_info(self) -> Dict[str, str]:
+        """Get model information."""
+        return {
+            "name": self.name,
+            "description": self.description,
+            "status": "loaded" if self._model is not None else "unloaded"
+        } 

src/models/registry.py

@@ -0,0 +1,79 @@
+from typing import Dict, List, Optional
+import gradio as gr
+from .base import BaseModel
+
+class GradioModelInfo:
+    """Information about a Gradio model."""
+    def __init__(self, 
+                 model_id: str,
+                 name: str, 
+                 description: str,
+                 input_type: str,
+                 output_type: List[str],
+                 examples: List[List[str]],
+                 api_path: str):
+        self.model_id = model_id
+        self.name = name
+        self.description = description
+        self.input_type = input_type
+        self.output_type = output_type
+        self.examples = examples
+        self.api_path = api_path
+        self.status = "unloaded"
+
+class GradioRegistry:
+    """Registry for Gradio models."""
+    
+    def __init__(self):
+        self._models: Dict[str, BaseModel] = {}
+        self._model_info: Dict[str, GradioModelInfo] = {}
+        
+    def register_model(self, 
+                      model_id: str, 
+                      model: BaseModel,
+                      api_path: str) -> None:
+        """Register a new Gradio model."""
+        self._models[model_id] = model
+        
+        # Create interface to extract information
+        interface = model.create_interface()
+        
+        # Store model information
+        self._model_info[model_id] = GradioModelInfo(
+            model_id=model_id,
+            name=model.name,
+            description=model.description,
+            input_type=interface.input_components[0].__class__.__name__,
+            output_type=[comp.__class__.__name__ for comp in interface.output_components],
+            examples=interface.examples or [],
+            api_path=api_path
+        )
+        
+    def get_model(self, model_id: str) -> Optional[BaseModel]:
+        """Get a model by ID."""
+        return self._models.get(model_id)
+    
+    def get_model_info(self, model_id: str) -> Optional[GradioModelInfo]:
+        """Get model information by ID."""
+        return self._model_info.get(model_id)
+    
+    def list_models(self) -> List[Dict]:
+        """List all registered models."""
+        return [
+            {
+                "id": info.model_id,
+                "name": info.name,
+                "description": info.description,
+                "input_type": info.input_type,
+                "output_type": info.output_type,
+                "examples": info.examples,
+                "api_path": info.api_path,
+                "status": info.status
+            }
+            for info in self._model_info.values()
+        ]
+    
+    def update_model_status(self, model_id: str, status: str) -> None:
+        """Update the status of a model."""
+        if model_id in self._model_info:
+            self._model_info[model_id].status = status 

src/models/text_classification.py

@@ -0,0 +1,106 @@
+from typing import Any, Dict, Union, Tuple
+import gradio as gr
+from transformers import pipeline, AutoModelForSequenceClassification, AutoTokenizer
+import logging
+from .base import BaseModel
+
+logger = logging.getLogger(__name__)
+
+class TextClassificationModel(BaseModel):
+    """Lightweight text classification model using tiny BERT."""
+    
+    def __init__(self):
+        super().__init__(
+            name="Lightweight Text Classifier",
+            description="Fast text classification using a tiny BERT model (4.4MB)"
+        )
+        self.model_name = "prajjwal1/bert-tiny"
+        self._model = None
+    
+    def load_model(self) -> None:
+        """Load the classification model."""
+        try:
+            logger.info(f"Loading model: {self.model_name}")
+            
+            # Initialize model with binary classification
+            model = AutoModelForSequenceClassification.from_pretrained(
+                self.model_name,
+                num_labels=2
+            )
+            tokenizer = AutoTokenizer.from_pretrained(self.model_name)
+            
+            self._model = pipeline(
+                "text-classification",
+                model=model,
+                tokenizer=tokenizer,
+                device=-1  # CPU, use device=0 for GPU
+            )
+            
+            # Log model size
+            model_size_mb = sum(p.numel() * p.element_size() for p in model.parameters()) / (1024 * 1024)
+            logger.info(f"Model loaded successfully. Size: {model_size_mb:.2f} MB")
+            
+        except Exception as e:
+            logger.error(f"Error loading model: {str(e)}")
+            raise
+    
+    async def predict(self, text: str) -> Dict[str, Union[str, float]]:
+        """Make a prediction using the model."""
+        try:
+            if self._model is None:
+                self.load_model()
+                
+            logger.info(f"Processing text: {text[:50]}...")
+            result = self._model(text)[0]
+            
+            # Map raw labels to sentiment
+            label_map = {
+                "LABEL_0": "NEGATIVE",
+                "LABEL_1": "POSITIVE"
+            }
+            
+            prediction = {
+                "label": label_map.get(result["label"], result["label"]),
+                "confidence": float(result["score"])
+            }
+            logger.info(f"Prediction result: {prediction}")
+            return prediction
+            
+        except Exception as e:
+            logger.error(f"Prediction error: {str(e)}")
+            raise
+
+    async def predict_for_interface(self, text: str) -> Tuple[str, float]:
+        """Make a prediction and return it in a format suitable for the Gradio interface."""
+        result = await self.predict(text)
+        return result["label"], result["confidence"]
+    
+    def create_interface(self) -> gr.Interface:
+        """Create a Gradio interface for text classification."""
+        if self._model is None:
+            self.load_model()
+            
+        examples = [
+            ["This movie was fantastic! I really enjoyed it."],
+            ["The service was terrible and the food was cold."],
+            ["It was an okay experience, nothing special."],
+            ["The weather is nice today!"],
+            ["I'm feeling sick and tired."]
+        ]
+        
+        return gr.Interface(
+            fn=self.predict_for_interface,  # Use the interface-specific prediction function
+            inputs=gr.Textbox(
+                lines=3,
+                placeholder="Enter text to classify...",
+                label="Input Text"
+            ),
+            outputs=[
+                gr.Label(label="Sentiment"),
+                gr.Number(label="Confidence", precision=4)
+            ],
+            title=self.name,
+            description=self.description + "\n\nThis model is also available via API!",
+            examples=examples,
+            api_name="predict"
+        ) 

test_api.py

@@ -0,0 +1,60 @@
+import requests
+import time
+import logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+def test_api():
+    # Wait a bit for the model to load
+    logger.info("Waiting for the server to start...")
+    time.sleep(10)
+    
+    base_url = "http://127.0.0.1:7860"
+    
+    # Test single prediction
+    test_texts = [
+        "This is amazing! I love it!",
+        "This is terrible, I hate it.",
+        "It's okay, nothing special."
+    ]
+    
+    logger.info("\nTesting single predictions:")
+    for text in test_texts:
+        try:
+            logger.info(f"\nTesting with text: {text}")
+            response = requests.post(
+                f"{base_url}/api/predict",
+                json={"text": text}
+            )
+            
+            if response.status_code == 200:
+                result = response.json()
+                logger.info(f"Result: {result}")
+            else:
+                logger.error(f"Error: {response.status_code} - {response.text}")
+                
+        except Exception as e:
+            logger.error(f"Request failed: {str(e)}")
+        
+        time.sleep(1)  # Small delay between requests
+    
+    # Test batch prediction
+    logger.info("\nTesting batch prediction:")
+    try:
+        response = requests.post(
+            f"{base_url}/api/predict_batch",
+            json={"texts": test_texts}
+        )
+        
+        if response.status_code == 200:
+            result = response.json()
+            logger.info(f"Batch results: {result}")
+        else:
+            logger.error(f"Batch Error: {response.status_code} - {response.text}")
+            
+    except Exception as e:
+        logger.error(f"Batch request failed: {str(e)}")
+
+if __name__ == "__main__":
+    test_api() 

test_client.py

@@ -0,0 +1,130 @@
+import requests
+import json
+import time
+import logging
+from typing import Dict, List, Any, Optional
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class MLModelsClient:
+    """Client for interacting with the ML Models API."""
+    
+    def __init__(self, base_url: str = "http://localhost:8000"):
+        self.base_url = base_url
+        
+    def list_models(self) -> List[Dict]:
+        """List all available models."""
+        try:
+            logger.info("Fetching available models...")
+            response = requests.get(f"{self.base_url}/api/models")
+            response.raise_for_status()
+            models = response.json()
+            logger.info(f"Found {len(models)} models")
+            return models
+        except Exception as e:
+            logger.error(f"Error listing models: {str(e)}")
+            raise
+
+    def get_model_info(self, model_id: str) -> Dict:
+        """Get information about a specific model."""
+        try:
+            logger.info(f"Fetching info for model {model_id}...")
+            response = requests.get(f"{self.base_url}/api/models/{model_id}")
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
+            logger.error(f"Error getting model info: {str(e)}")
+            raise
+
+    def get_model_status(self, model_id: str) -> str:
+        """Get the current status of a model."""
+        try:
+            logger.info(f"Fetching status for model {model_id}...")
+            response = requests.get(f"{self.base_url}/api/models/{model_id}/status")
+            response.raise_for_status()
+            return response.json()["status"]
+        except Exception as e:
+            logger.error(f"Error getting model status: {str(e)}")
+            raise
+
+    def load_model(self, model_id: str) -> str:
+        """Load a model into memory."""
+        try:
+            logger.info(f"Loading model {model_id}...")
+            response = requests.post(f"{self.base_url}/api/models/{model_id}/load")
+            response.raise_for_status()
+            return response.json()["status"]
+        except Exception as e:
+            logger.error(f"Error loading model: {str(e)}")
+            raise
+
+    def predict(self, model_id: str, text: str) -> Dict:
+        """Make a prediction using a model."""
+        try:
+            logger.info(f"Making prediction with model {model_id}...")
+            model_info = self.get_model_info(model_id)
+            response = requests.post(
+                f"{self.base_url}{model_info.get('api_path')}/predict",
+                json={"text": text}
+            )
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
+            logger.error(f"Error making prediction: {str(e)}")
+            raise
+
+def test_model_workflow():
+    """Test the complete model workflow."""
+    client = MLModelsClient()
+    
+    try:
+        # 1. List available models
+        logger.info("\n1. Testing model listing...")
+        models = client.list_models()
+        for model in models:
+            logger.info(f"Found model: {json.dumps(model, indent=2)}")
+
+        if not models:
+            logger.error("No models found!")
+            return
+
+        # Use the first model for testing
+        model_id = models[0]["id"]
+
+        # 2. Get model information
+        logger.info(f"\n2. Testing model info retrieval for {model_id}...")
+        model_info = client.get_model_info(model_id)
+        logger.info(f"Model info: {json.dumps(model_info, indent=2)}")
+
+        # 3. Get model status
+        logger.info(f"\n3. Testing model status retrieval for {model_id}...")
+        status = client.get_model_status(model_id)
+        logger.info(f"Model status: {status}")
+
+        # 4. Load the model
+        logger.info(f"\n4. Testing model loading for {model_id}...")
+        load_status = client.load_model(model_id)
+        logger.info(f"Load status: {load_status}")
+
+        # 5. Test predictions
+        test_texts = [
+            "This is amazing! I really love it!",
+            "This is terrible, I hate it.",
+            "It's okay, nothing special."
+        ]
+
+        logger.info(f"\n5. Testing predictions for {model_id}...")
+        for text in test_texts:
+            logger.info(f"\nPredicting for text: {text}")
+            result = client.predict(model_id, text)
+            logger.info(f"Prediction: {json.dumps(result, indent=2)}")
+            time.sleep(1)  # Small delay between predictions
+
+    except Exception as e:
+        logger.error(f"Test workflow failed: {str(e)}")
+        raise
+
+if __name__ == "__main__":
+    logger.info("Starting model testing workflow...")
+    test_model_workflow()