models/llms.py

import requests
import time
import json
import os
import logging
from typing import List, Dict
from utils.logging import log_function, setup_logging
from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type, wait_exponential
from config.load_configs import load_config
from dotenv import load_dotenv
from requests.packages.urllib3.util.retry import Retry
load_dotenv()

setup_logging(level=logging.DEBUG)
logger = logging.getLogger(__name__)

class BaseModel:
    def __init__(self, temperature: float, model: str, json_response: bool, max_retries: int = 3, retry_delay: int = 1):
        self.temperature = temperature
        self.model = model
        self.json_response = json_response
        self.max_retries = max_retries
        self.retry_delay = retry_delay

    # @retry(stop=stop_after_attempt(3), wait=wait_fixed(1), retry=retry_if_exception_type(requests.RequestException))
    @retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=2, max=15), reraise=True)
    def _make_request(self, url, headers, payload):

        response = requests.post(url, headers=headers, data=json.dumps(payload))
        response.raise_for_status()
        return response.json()
    
class MistralModel(BaseModel):
    def __init__(self, temperature: float, model: str, json_response: bool, max_retries: int = 3, retry_delay: int = 1):
        super().__init__(temperature, model, json_response, max_retries, retry_delay)
        # config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'config.yaml')
        # load_config(config_path)
        # load_config()
        self.api_key = os.environ.get("MISTRAL_API_KEY")
        self.headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json',
            'Authorization': f'Bearer {self.api_key}'
        }
        self.model_endpoint = "https://api.mistral.ai/v1/chat/completions"

    @retry(stop=stop_after_attempt(3), wait=wait_fixed(1), retry=retry_if_exception_type(requests.RequestException))
    def _make_request(self, url, headers, payload):
        response = requests.post(url, headers=headers, data=json.dumps(payload))
        response.raise_for_status()
        return response.json()

    def invoke(self, messages: List[Dict[str, str]]) -> str:
        system = messages[0]["content"]
        user = messages[1]["content"]

        payload = {
            "model": self.model,
            "messages": [
                {
                    "role": "system",
                    "content": system
                },
                {
                    "role": "user",
                    "content": user
                }
            ],
            "temperature": self.temperature,
        }

        if self.json_response:
            payload["response_format"] = {"type": "json_object"}

        try:
            request_response_json = self._make_request(self.model_endpoint, self.headers, payload)
            
            if 'choices' not in request_response_json or len(request_response_json['choices']) == 0:
                raise ValueError("No choices in response")

            response_content = request_response_json['choices'][0]['message']['content']
            
            if self.json_response:
                response = json.dumps(json.loads(response_content))
            else:
                response = response_content

            return response
        except requests.RequestException as e:
            return json.dumps({"error": f"Error in invoking model after {self.max_retries} retries: {str(e)}"})
        except (ValueError, KeyError, json.JSONDecodeError) as e:
            return json.dumps({"error": f"Error processing response: {str(e)}"})


class ClaudeModel(BaseModel):
    def __init__(self, temperature: float, model: str, json_response: bool, max_retries: int = 3, retry_delay: int = 1):
        super().__init__(temperature, model, json_response, max_retries, retry_delay)
        # config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'config.yaml')
        # load_config(config_path)
        self.api_key = os.environ.get("ANTHROPIC_API_KEY")
        self.headers = {
            'Content-Type': 'application/json', 
            'x-api-key': self.api_key,
            'anthropic-version': '2023-06-01'
        }
        self.model_endpoint = "https://api.302.ai/v1/chat/completions"

    def invoke(self, messages: List[Dict[str, str]]) -> str:
        # time.sleep(5)
        system = messages[0]["content"]
        user = messages[1]["content"]

        content = f"system:{system}\n\n user:{user}"
        if self.json_response:
            content += ". Your output must be json formatted. Just return the specified json format, do not prepend your response with anything."

        payload = {
            "model": self.model,
            "messages": [
                {
                    "role": "user",
                    "content": content
                }
            ],
            "max_tokens": 4096,
            "temperature": self.temperature,
        }

        try:
            request_response_json = self._make_request(self.model_endpoint, self.headers, payload)
            
            if 'content' not in request_response_json or not request_response_json['content']:
                raise ValueError("No content in response")

            response_content = request_response_json['content'][0]['text']
            
            if self.json_response:
                response = json.dumps(json.loads(response_content))
            else:
                response = response_content

            return response
        except requests.RequestException as e:
            return json.dumps({"error": f"Error in invoking model after {self.max_retries} retries: {str(e)}"})
        except (ValueError, KeyError, json.JSONDecodeError) as e:
            return json.dumps({"error": f"Error processing response: {str(e)}"})

class GeminiModel(BaseModel):
    def __init__(self, temperature: float, model: str, json_response: bool, max_retries: int = 3, retry_delay: int = 1):
        super().__init__(temperature, model, json_response, max_retries, retry_delay)
        # config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'config.yaml')
        # load_config(config_path)
        self.api_key = os.environ.get("GEMINI_API_KEY")
        self.headers = {
            'Content-Type': 'application/json'
        }
        self.model_endpoint = f"https://generativelanguage.googleapis.com/v1/models/{model}:generateContent?key={self.api_key}"

    def invoke(self, messages: List[Dict[str, str]]) -> str:
        system = messages[0]["content"]
        user = messages[1]["content"]

        content = f"system:{system}\n\nuser:{user}"
        if self.json_response:
            content += ". Your output must be JSON formatted. Just return the specified JSON format, do not prepend your response with anything."

        payload = {
            "contents": [
                {
                    "parts": [
                        {
                            "text": content
                        }
                    ]
                }
            ],
            "generationConfig": {
                "temperature": self.temperature
            },
        }

        if self.json_response:
                payload = {
                    "contents": [
                        {
                            "parts": [
                                {
                                    "text": content
                                }
                            ]
                        }
                    ],
                    "generationConfig": {
                        "response_mime_type": "application/json",
                        "temperature": self.temperature
                    },
                }
            # payload["generationConfig"]["response_mime_type"] = "application/json"

        try:
            request_response_json = self._make_request(self.model_endpoint, self.headers, payload)

            if 'candidates' not in request_response_json or not request_response_json['candidates']:
                raise ValueError("No content in response")

            response_content = request_response_json['candidates'][0]['content']['parts'][0]['text']
            
            if self.json_response:
                response = json.dumps(json.loads(response_content))
            else:
                response = response_content

            return response
        except requests.RequestException as e:
            return json.dumps({"error": f"Error in invoking model after {self.max_retries} retries: {str(e)}"})
        except (ValueError, KeyError, json.JSONDecodeError) as e:
            return json.dumps({"error": f"Error processing response: {str(e)}"})

class GroqModel(BaseModel):
    def __init__(self, temperature: float, model: str, json_response: bool, max_retries: int = 3, retry_delay: int = 1):
        super().__init__(temperature, model, json_response, max_retries, retry_delay)
        # config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'config.yaml')
        # load_config(config_path)
        self.api_key = os.environ.get("GROQ_API_KEY")
        self.headers = {
            'Content-Type': 'application/json', 
            'Authorization': f'Bearer {self.api_key}'
        }
        self.model_endpoint = "https://api.groq.com/openai/v1/chat/completions"

    def invoke(self, messages: List[Dict[str, str]]) -> str:
        system = messages[0]["content"]
        user = messages[1]["content"]

        payload = {
            "model": self.model,
            "messages": [
                {
                    "role": "user",
                    "content": f"system:{system}\n\n user:{user}"
                }
            ],
            "temperature": self.temperature,
        }

        time.sleep(10)

        if self.json_response:
            payload["response_format"] = {"type": "json_object"}

        try:
            request_response_json = self._make_request(self.model_endpoint, self.headers, payload)
            
            if 'choices' not in request_response_json or len(request_response_json['choices']) == 0:
                raise ValueError("No choices in response")

            response_content = request_response_json['choices'][0]['message']['content']
            
            if self.json_response:
                response = json.dumps(json.loads(response_content))
            else:
                response = response_content

            return response
        except requests.RequestException as e:
            return json.dumps({"error": f"Error in invoking model after {self.max_retries} retries: {str(e)}"})
        except (ValueError, KeyError, json.JSONDecodeError) as e:
            return json.dumps({"error": f"Error processing response: {str(e)}"})

class OllamaModel(BaseModel):
    def __init__(self, temperature: float, model: str, json_response: bool, max_retries: int = 3, retry_delay: int = 1):
        super().__init__(temperature, model, json_response, max_retries, retry_delay)
        self.headers = {"Content-Type": "application/json"}
        self.ollama_host = os.getenv("OLLAMA_HOST", "http://localhost:11434")
        self.model_endpoint = f"{self.ollama_host}/api/generate"

    def _check_and_pull_model(self):
        # Check if the model exists
        response = requests.get(f"{self.ollama_host}/api/tags")
        if response.status_code == 200:
            models = response.json().get("models", [])
            if not any(model["name"] == self.model for model in models):
                print(f"Model {self.model} not found. Pulling the model...")
                self._pull_model()
            else:
                print(f"Model {self.model} is already available.")
        else:
            print(f"Failed to check models. Status code: {response.status_code}")

    def _pull_model(self):
        pull_endpoint = f"{self.ollama_host}/api/pull"
        payload = {"name": self.model}
        response = requests.post(pull_endpoint, json=payload, stream=True)
        
        if response.status_code == 200:
            for line in response.iter_lines():
                if line:
                    status = json.loads(line.decode('utf-8'))
                    print(f"Pulling model: {status.get('status')}")
            print(f"Model {self.model} pulled successfully.")
        else:
            print(f"Failed to pull model. Status code: {response.status_code}")

    def invoke(self, messages: List[Dict[str, str]]) -> str:
        self._check_and_pull_model()  # Check and pull the model if necessary

        system = messages[0]["content"]
        user = messages[1]["content"]

        payload = {
            "model": self.model,
            "prompt": user,
            "system": system,
            "stream": False,
            "temperature": self.temperature,
        }

        if self.json_response:
            payload["format"] = "json"
        
        try:
            request_response_json = self._make_request(self.model_endpoint, self.headers, payload)
            
            if self.json_response:
                response = json.dumps(json.loads(request_response_json['response']))
            else:
                response = str(request_response_json['response'])

            return response
        except requests.RequestException as e:
            return json.dumps({"error": f"Error in invoking model after {self.max_retries} retries: {str(e)}"})
        except json.JSONDecodeError as e:
            return json.dumps({"error": f"Error processing response: {str(e)}"})
class VllmModel(BaseModel):
    def __init__(self, temperature: float, model: str, model_endpoint: str, json_response: bool, stop: str = None, max_retries: int = 5, retry_delay: int = 1):
        super().__init__(temperature, model, json_response, max_retries, retry_delay)
        self.headers = {"Content-Type": "application/json"}
        self.model_endpoint = model_endpoint + 'v1/chat/completions'
        self.stop = stop

    def invoke(self, messages: List[Dict[str, str]], guided_json: dict = None) -> str:
        system = messages[0]["content"]
        user = messages[1]["content"]

        prefix = self.model.split('/')[0]

        if prefix == "mistralai":
            payload = {
                "model": self.model,
                "messages": [
                    {
                        "role": "user",
                        "content": f"system:{system}\n\n user:{user}"
                    }
                ],
                "temperature": self.temperature,
                "stop": None,
            }
        else:
            payload = {
                "model": self.model,
                "messages": [
                    {
                        "role": "system",
                        "content": system
                    },
                    {
                        "role": "user",
                        "content": user
                    }
                ],
                "temperature": self.temperature,
                "stop": self.stop,
            }

        if self.json_response:
            payload["response_format"] = {"type": "json_object"}
            payload["guided_json"] = guided_json
        
        try:
            request_response_json = self._make_request(self.model_endpoint, self.headers, payload)
            response_content = request_response_json['choices'][0]['message']['content']
            
            if self.json_response:
                response = json.dumps(json.loads(response_content))
            else:
                response = str(response_content)
            
            return response
        except requests.RequestException as e:
            return json.dumps({"error": f"Error in invoking model after {self.max_retries} retries: {str(e)}"})
        except json.JSONDecodeError as e:
            return json.dumps({"error": f"Error processing response: {str(e)}"})

class OpenAIModel(BaseModel):
    def __init__(self, temperature: float, model: str, json_response: bool, max_retries: int = 3, retry_delay: int = 1):
        super().__init__(temperature, model, json_response, max_retries, retry_delay)
        # config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'config.yaml')
        # load_config(config_path)
        load_dotenv()
        self.model_endpoint = 'https://api.302.ai/v1/chat/completions'
        
        self.api_key = os.environ.get('OPENAI_API_KEY')
        self.headers = {
            'Content-Type': 'application/json',
            'Authorization': f'Bearer {self.api_key}'
        }

    def invoke(self, messages: List[Dict[str, str]]) -> str:
        system = messages[0]["content"]
        user = messages[1]["content"]

        if self.model == "o1-preview" or self.model == "o1-mini":

            payload = {
            "model": self.model,
            "messages": [
                {
                    "role": "user",
                    "content": f"{system}\n\n{user}"
                }
            ]
        }

        else:
            payload = {
                "model": self.model,
            "messages": [
                {
                    "role": "system",
                    "content": system
                },
                {
                    "role": "user",
                    "content": user
                }
            ],
            "stream": False,
            "temperature": self.temperature,
        }
        
        if self.json_response:
            payload["response_format"] = {"type": "json_object"}
            payload["messages"][0]["content"] = f"{system}\n\nYou must respond in JSON format."
                    
        try:
            response_json = self._make_request(self.model_endpoint, self.headers, payload)

            if self.json_response:
                response = json.dumps(json.loads(response_json['choices'][0]['message']['content']))
            else:
                response = response_json['choices'][0]['message']['content']

            return response
        except requests.RequestException as e:
            return json.dumps({"error": f"Error in invoking model after {self.max_retries} retries: {str(e)}"})
        except json.JSONDecodeError as e:
            return json.dumps({"error": f"Error processing response: {str(e)}"})