tjdify/api/core/model_providers/providers/openai_provider.py

import json
import logging
from json import JSONDecodeError
from typing import Type, Optional

from flask import current_app
from openai.error import AuthenticationError, OpenAIError

import openai

from core.helper import encrypter
from core.model_providers.models.entity.provider import ModelFeature
from core.model_providers.models.speech2text.openai_whisper import OpenAIWhisper
from core.model_providers.models.base import BaseProviderModel
from core.model_providers.models.embedding.openai_embedding import OpenAIEmbedding
from core.model_providers.models.entity.model_params import ModelKwargsRules, KwargRule, ModelType, ModelMode
from core.model_providers.models.llm.openai_model import OpenAIModel, COMPLETION_MODELS
from core.model_providers.models.moderation.openai_moderation import OpenAIModeration
from core.model_providers.providers.base import BaseModelProvider, CredentialsValidateFailedError
from core.model_providers.providers.hosted import hosted_model_providers
from models.provider import ProviderType, ProviderQuotaType


class OpenAIProvider(BaseModelProvider):

    @property
    def provider_name(self):
        """
        Returns the name of a provider.
        """
        return 'openai'

    def _get_fixed_model_list(self, model_type: ModelType) -> list[dict]:
        if model_type == ModelType.TEXT_GENERATION:
            models = [
                {
                    'id': 'gpt-3.5-turbo',
                    'name': 'gpt-3.5-turbo',
                    'mode': ModelMode.CHAT.value,
                    'features': [
                        ModelFeature.AGENT_THOUGHT.value
                    ]
                },
                {
                    'id': 'gpt-3.5-turbo-instruct',
                    'name': 'GPT-3.5-Turbo-Instruct',
                    'mode': ModelMode.COMPLETION.value,
                },
                {
                    'id': 'gpt-3.5-turbo-16k',
                    'name': 'gpt-3.5-turbo-16k',
                    'mode': ModelMode.CHAT.value,
                    'features': [
                        ModelFeature.AGENT_THOUGHT.value
                    ]
                },
                {
                    'id': 'gpt-4',
                    'name': 'gpt-4',
                    'mode': ModelMode.CHAT.value,
                    'features': [
                        ModelFeature.AGENT_THOUGHT.value
                    ]
                },
                {
                    'id': 'gpt-4-32k',
                    'name': 'gpt-4-32k',
                    'mode': ModelMode.CHAT.value,
                    'features': [
                        ModelFeature.AGENT_THOUGHT.value
                    ]
                },
                {
                    'id': 'text-davinci-003',
                    'name': 'text-davinci-003',
                    'mode': ModelMode.COMPLETION.value,
                }
            ]

            if self.provider.provider_type == ProviderType.SYSTEM.value \
                    and self.provider.quota_type == ProviderQuotaType.TRIAL.value:
                models = [item for item in models if item['id'] not in ['gpt-4', 'gpt-4-32k']]

            return models
        elif model_type == ModelType.EMBEDDINGS:
            return [
                {
                    'id': 'text-embedding-ada-002',
                    'name': 'text-embedding-ada-002'
                }
            ]
        elif model_type == ModelType.SPEECH_TO_TEXT:
            return [
                {
                    'id': 'whisper-1',
                    'name': 'whisper-1'
                }
            ]
        elif model_type == ModelType.MODERATION:
            return [
                {
                    'id': 'text-moderation-stable',
                    'name': 'text-moderation-stable'
                }
            ]
        else:
            return []

    def _get_text_generation_model_mode(self, model_name) -> str:
        if model_name in COMPLETION_MODELS:
            return ModelMode.COMPLETION.value
        else:
            return ModelMode.CHAT.value

    def get_model_class(self, model_type: ModelType) -> Type[BaseProviderModel]:
        """
        Returns the model class.

        :param model_type:
        :return:
        """
        if model_type == ModelType.TEXT_GENERATION:
            model_class = OpenAIModel
        elif model_type == ModelType.EMBEDDINGS:
            model_class = OpenAIEmbedding
        elif model_type == ModelType.MODERATION:
            model_class = OpenAIModeration
        elif model_type == ModelType.SPEECH_TO_TEXT:
            model_class = OpenAIWhisper
        else:
            raise NotImplementedError

        return model_class

    def get_model_parameter_rules(self, model_name: str, model_type: ModelType) -> ModelKwargsRules:
        """
        get model parameter rules.

        :param model_name:
        :param model_type:
        :return:
        """
        model_max_tokens = {
            'gpt-4': 8192,
            'gpt-4-32k': 32768,
            'gpt-3.5-turbo': 4096,
            'gpt-3.5-turbo-instruct': 4097,
            'gpt-3.5-turbo-16k': 16384,
            'text-davinci-003': 4097,
        }

        return ModelKwargsRules(
            temperature=KwargRule[float](min=0, max=2, default=1, precision=2),
            top_p=KwargRule[float](min=0, max=1, default=1, precision=2),
            presence_penalty=KwargRule[float](min=-2, max=2, default=0, precision=2),
            frequency_penalty=KwargRule[float](min=-2, max=2, default=0, precision=2),
            max_tokens=KwargRule[int](min=10, max=model_max_tokens.get(model_name, 4097), default=16, precision=0),
        )

    @classmethod
    def is_provider_credentials_valid_or_raise(cls, credentials: dict):
        """
        Validates the given credentials.
        """
        if 'openai_api_key' not in credentials:
            raise CredentialsValidateFailedError('OpenAI API key is required')

        try:
            credentials_kwargs = {
                "api_key": credentials['openai_api_key']
            }

            if 'openai_api_base' in credentials and credentials['openai_api_base']:
                credentials_kwargs['api_base'] = credentials['openai_api_base'] + '/v1'

            if 'openai_organization' in credentials:
                credentials_kwargs['organization'] = credentials['openai_organization']

            openai.ChatCompletion.create(
                messages=[{"role": "user", "content": 'ping'}],
                model='gpt-3.5-turbo',
                timeout=10,
                request_timeout=(5, 30),
                max_tokens=20,
                **credentials_kwargs
            )
        except (AuthenticationError, OpenAIError) as ex:
            raise CredentialsValidateFailedError(str(ex))
        except Exception as ex:
            logging.exception('OpenAI config validation failed')
            raise ex

    @classmethod
    def encrypt_provider_credentials(cls, tenant_id: str, credentials: dict) -> dict:
        credentials['openai_api_key'] = encrypter.encrypt_token(tenant_id, credentials['openai_api_key'])
        return credentials

    def get_provider_credentials(self, obfuscated: bool = False) -> dict:
        if self.provider.provider_type == ProviderType.CUSTOM.value:
            try:
                credentials = json.loads(self.provider.encrypted_config)
            except JSONDecodeError:
                credentials = {
                    'openai_api_base': None,
                    'openai_api_key': self.provider.encrypted_config,
                    'openai_organization': None
                }

            if credentials['openai_api_key']:
                credentials['openai_api_key'] = encrypter.decrypt_token(
                    self.provider.tenant_id,
                    credentials['openai_api_key']
                )

                if obfuscated:
                    credentials['openai_api_key'] = encrypter.obfuscated_token(credentials['openai_api_key'])

            if 'openai_api_base' not in credentials or not credentials['openai_api_base']:
                credentials['openai_api_base'] = None
            else:
                credentials['openai_api_base'] = credentials['openai_api_base'] + '/v1'

            if 'openai_organization' not in credentials:
                credentials['openai_organization'] = None

            return credentials
        else:
            if hosted_model_providers.openai:
                return {
                    'openai_api_base': hosted_model_providers.openai.api_base,
                    'openai_api_key': hosted_model_providers.openai.api_key,
                    'openai_organization': hosted_model_providers.openai.api_organization
                }
            else:
                return {
                    'openai_api_base': None,
                    'openai_api_key': None,
                    'openai_organization': None
                }

    @classmethod
    def is_provider_type_system_supported(cls) -> bool:
        if current_app.config['EDITION'] != 'CLOUD':
            return False

        if hosted_model_providers.openai:
            return True

        return False

    def should_deduct_quota(self):
        if hosted_model_providers.openai \
                and hosted_model_providers.openai.quota_limit and hosted_model_providers.openai.quota_limit > 0:
            return True

        return False

    def get_payment_info(self) -> Optional[dict]:
        """
        get payment info if it payable.

        :return:
        """
        if hosted_model_providers.openai \
                and hosted_model_providers.openai.paid_enabled:
            return {
                'product_id': hosted_model_providers.openai.paid_stripe_price_id,
                'increase_quota': hosted_model_providers.openai.paid_increase_quota,
            }

        return None

    @classmethod
    def is_model_credentials_valid_or_raise(cls, model_name: str, model_type: ModelType, credentials: dict):
        """
        check model credentials valid.

        :param model_name:
        :param model_type:
        :param credentials:
        """
        return

    @classmethod
    def encrypt_model_credentials(cls, tenant_id: str, model_name: str, model_type: ModelType, credentials: dict) -> dict:
        """
        encrypt model credentials for save.

        :param tenant_id:
        :param model_name:
        :param model_type:
        :param credentials:
        :return:
        """
        return {}

    def get_model_credentials(self, model_name: str, model_type: ModelType, obfuscated: bool = False) -> dict:
        """
        get credentials for llm use.

        :param model_name:
        :param model_type:
        :param obfuscated:
        :return:
        """
        return self.get_provider_credentials(obfuscated)