OpenAI chat & Embeddings

OpenAI chat & Embeddings.py

import httpx
from pprint import pprint
from dotenv import load_dotenv
from openai.types import CreateEmbeddingResponse

_ = load_dotenv("../.env")

from openai import (
    OpenAI, APIError, RateLimitError, APITimeoutError, APIConnectionError, ChatCompletion,
)
import time
from app_constants import AppConstants


class OpenAIModel:
    RETRIES = 3
    TIMEOUT = 60

    def __init__(self, dimenssions: int = 3072):
        self.client = OpenAI(
            api_key=AppConstants.API_KEY,
            base_url=AppConstants.GATE_WAY_URL,
            http_client=httpx.Client(verify=False),
            timeout=self.TIMEOUT,
        )
        self.header = {
            'api_key': AppConstants.API_KEY,
        }
        self.openai_text_embedding_model = "text-embedding-3-small"
        self.model = AppConstants.OPENAI_MODEL
        self.dimenssion = dimenssions
        self.temprature = 0.2


    def openai_chat(self, system_prompt: str, user_prompt: str) -> ChatCompletion:
        try:
            completion: ChatCompletion = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt},
                ],
                extra_headers=self.header,
            )
            return completion
        except RateLimitError as e:
            pprint(f"RateLimitError: {str(e)}")
        except APITimeoutError as e:
            pprint(f"APITimeoutError: {str(e)}")
        except APIConnectionError as e:
            pprint(f"APIConnectionError: {str(e)}")
        except Exception as e:
            pprint(f"Error: {str(e)}")

    def create_embeddings(self, text: str):
        """
        Create a vector embedding for the given text using OpenAI's embedding models.

        Args:
            text (str): The input text to embed.
            model (str): The embedding model to use. Options:
                         - "text-embedding-3-small" (faster, cheaper, 1536 dimensions)
                         - "text-embedding-3-large" (more accurate, 3072 dimensions)

        Returns:
            list[float]: The embedding vector.
        """
        try:
            # Validate input
            if not isinstance(text, str) or not text.strip():
                raise ValueError("Input text must be a non-empty string.")

            response: CreateEmbeddingResponse = self.client.embeddings.create(
                model=self.openai_text_embedding_model,
                input=text,
                dimensions=self.dimenssion,
                timeout=self.TIMEOUT
            )

            # Extract and return the embedding vector
            return response.data[0].embedding

        except ValueError as ve:
            print(f"Validation Error: {ve}")
        except RateLimitError:
            print("Rate limit exceeded. Please retry after some time.")
        except APITimeoutError:
            print("Request timed out. Try again later.")
        except APIConnectionError:
            print("Network connection error. Check your internet connection.")
        except APIError as e:
            print(f"API returned an error: {e}")
        except Exception as e:
            print(f"Unexpected error: {e}")


obj = OpenAIModel()
chat: ChatCompletion = obj.openai_chat(system_prompt=",,,", user_prompt="...")