Upload 7 files

src/config.py

@@ -0,0 +1,8 @@
+# Model settings
+device = "cuda"
+model_name = "openbmb/MiniCPM-o-2_6"
+
+# Decoding settings
+sampling = False
+stream = False
+repetition_penalty = 1.05

src/exception.py

@@ -0,0 +1,50 @@
+"""
+This module defines a custom exception handling class and a function to get error message with details of the error.
+"""
+
+# Standard Library
+import sys
+
+# Local imports
+from src.logger import logging
+
+
+# Function Definition to get error message with details of the error (file name and line number) when an error occurs in the program
+def get_error_message(error, error_detail: sys):
+    """
+    Get error message with details of the error.
+
+    Args:
+        - error (Exception): The error that occurred.
+        - error_detail (sys): The details of the error.
+
+    Returns:
+        str: A string containing the error message along with the file name and line number where the error occurred.
+    """
+    _, _, exc_tb = error_detail.exc_info()
+
+    # Get error details
+    file_name = exc_tb.tb_frame.f_code.co_filename
+    return "Error occured in python script name [{0}] line number [{1}] error message[{2}]".format(
+        file_name, exc_tb.tb_lineno, str(error)
+    )
+
+
+# Custom Exception Handling Class Definition
+class CustomExceptionHandling(Exception):
+    """
+    Custom Exception Handling:
+        This class defines a custom exception that can be raised when an error occurs in the program.
+        It takes an error message and an error detail as input and returns a formatted error message when the exception is raised.
+    """
+
+    # Constructor
+    def __init__(self, error_message, error_detail: sys):
+        """Initialize the exception"""
+        super().__init__(error_message)
+
+        self.error_message = get_error_message(error_message, error_detail=error_detail)
+
+    def __str__(self):
+        """String representation of the exception"""
+        return self.error_message

src/logger.py

@@ -0,0 +1,21 @@
+# Importing the required modules
+import os
+import logging
+from datetime import datetime
+
+# Creating a log file with the current date and time as the name of the file
+LOG_FILE = f"{datetime.now().strftime('%m_%d_%Y_%H_%M_%S')}.log"
+
+# Creating a logs folder if it does not exist
+logs_path = os.path.join(os.getcwd(), "logs", LOG_FILE)
+os.makedirs(logs_path, exist_ok=True)
+
+# Setting the log file path and the log level
+LOG_FILE_PATH = os.path.join(logs_path, LOG_FILE)
+
+# Configuring the logger
+logging.basicConfig(
+    filename=LOG_FILE_PATH,
+    format="[ %(asctime)s ] %(lineno)d %(name)s - %(levelname)s - %(message)s",
+    level=logging.INFO,
+)

src/minicpm/model.py

@@ -0,0 +1,63 @@
+# Necessary imports
+import os
+import sys
+from dotenv import load_dotenv
+from typing import Any
+import torch
+from transformers import AutoModel, AutoTokenizer, AutoProcessor
+
+# Local imports
+from src.logger import logging
+from src.exception import CustomExceptionHandling
+
+
+# Load the Environment Variables from .env file
+load_dotenv()
+
+# Access token for using the model
+access_token = os.environ.get("ACCESS_TOKEN")
+
+
+def load_model_tokenizer_and_processor(model_name: str, device: str) -> Any:
+    """
+    Load the model, tokenizer and processor.
+
+    Args:
+        - model_name (str): The name of the model to load.
+        - device (str): The device to load the model onto.
+
+    Returns:
+        - model: The loaded model.
+        - tokenizer: The loaded tokenizer.
+        - processor: The loaded processor.
+    """
+    try:
+        # Load the model, tokenizer and processor
+        model = AutoModel.from_pretrained(
+            model_name,
+            trust_remote_code=True,
+            attn_implementation="sdpa",
+            torch_dtype=torch.bfloat16,
+            init_vision=True,
+            init_audio=False,
+            init_tts=False,
+            token=access_token
+        )
+        model = model.eval().to(device=device)
+        tokenizer = AutoTokenizer.from_pretrained(
+            model_name, trust_remote_code=True, token=access_token
+        )
+        processor = AutoProcessor.from_pretrained(
+            model_name, trust_remote_code=True, token=access_token
+        )
+
+        # Log the successful loading of the model, tokenizer and processor
+        logging.info("Model, tokenizer and processor loaded successfully.")
+
+        # Return the model, tokenizer and processor
+        return model, tokenizer, processor
+
+    # Handle exceptions that may occur during model, tokenizer and processor loading
+    except Exception as e:
+        # Custom exception handling
+        raise CustomExceptionHandling(e, sys) from e

src/minicpm/response.py

@@ -0,0 +1,78 @@
+# Necessary imports
+import sys
+import gradio as gr
+import spaces
+
+# Local imports
+from src.config import (
+    device,
+    model_name,
+    sampling,
+    stream,
+    repetition_penalty,
+)
+from src.minicpm.model import load_model_tokenizer_and_processor
+from src.logger import logging
+from src.exception import CustomExceptionHandling
+
+
+# Model, tokenizer and processor
+model, tokenizer, processor = load_model_tokenizer_and_processor(model_name, device)
+
+
+@spaces.GPU(duration=120)
+def describe_image(
+    image: str, 
+    question: str,
+    temperature: float,
+    top_p: float,
+    top_k: int,
+    max_new_tokens: int,
+) -> str:
+    """
+    Generates an answer to a given question based on the provided image and question.
+
+    Args:
+        - image (str): The path to the image file.
+        - question (str): The question text.
+        - temperature (float): The temperature parameter for the model.
+        - top_p (float): The top_p parameter for the model.
+        - top_k (int): The top_k parameter for the model.
+        - max_new_tokens (int): The max tokens to be generated by the model.
+
+    Returns:
+        str: The generated answer to the question.
+    """
+    try:
+        # Check if image or question is None
+        if not image or not question:
+            gr.Warning("Please provide an image and a question.")
+
+        # Message format for the model
+        msgs = [{"role": "user", "content": [image, question]}]
+
+        # Generate the answer
+        answer = model.chat(
+            image=None,
+            msgs=msgs,
+            tokenizer=tokenizer,
+            processor=processor,
+            sampling=sampling,
+            stream=stream,
+            top_p=top_p,
+            top_k=top_k,
+            temperature=temperature,
+            repetition_penalty=repetition_penalty,
+            max_new_tokens=max_new_tokens,
+        )
+
+        # Log the successful generation of the answer
+        logging.info("Answer generated successfully.")
+
+        # Return the answer
+        return "".join(answer)
+
+    # Handle exceptions that may occur during answer generation
+    except Exception as e:
+        # Custom exception handling
+        raise CustomExceptionHandling(e, sys) from e