mirror of
https://github.com/xtekky/gpt4free.git
synced 2024-11-26 09:57:24 +03:00
5756586cde
Add doctypes to many functions Add file upload for text files Add alternative url to FreeChatgpt Add webp to allowed image types
115 lines
3.5 KiB
Python
115 lines
3.5 KiB
Python
from abc import ABC, abstractmethod
|
|
from typing import Union, List, Dict, Type
|
|
from .typing import Messages, CreateResult
|
|
|
|
class BaseProvider(ABC):
|
|
"""
|
|
Abstract base class for a provider.
|
|
|
|
Attributes:
|
|
url (str): URL of the provider.
|
|
working (bool): Indicates if the provider is currently working.
|
|
needs_auth (bool): Indicates if the provider needs authentication.
|
|
supports_stream (bool): Indicates if the provider supports streaming.
|
|
supports_gpt_35_turbo (bool): Indicates if the provider supports GPT-3.5 Turbo.
|
|
supports_gpt_4 (bool): Indicates if the provider supports GPT-4.
|
|
supports_message_history (bool): Indicates if the provider supports message history.
|
|
params (str): List parameters for the provider.
|
|
"""
|
|
|
|
url: str = None
|
|
working: bool = False
|
|
needs_auth: bool = False
|
|
supports_stream: bool = False
|
|
supports_gpt_35_turbo: bool = False
|
|
supports_gpt_4: bool = False
|
|
supports_message_history: bool = False
|
|
params: str
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def create_completion(
|
|
cls,
|
|
model: str,
|
|
messages: Messages,
|
|
stream: bool,
|
|
**kwargs
|
|
) -> CreateResult:
|
|
"""
|
|
Create a completion with the given parameters.
|
|
|
|
Args:
|
|
model (str): The model to use.
|
|
messages (Messages): The messages to process.
|
|
stream (bool): Whether to use streaming.
|
|
**kwargs: Additional keyword arguments.
|
|
|
|
Returns:
|
|
CreateResult: The result of the creation process.
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
async def create_async(
|
|
cls,
|
|
model: str,
|
|
messages: Messages,
|
|
**kwargs
|
|
) -> str:
|
|
"""
|
|
Asynchronously create a completion with the given parameters.
|
|
|
|
Args:
|
|
model (str): The model to use.
|
|
messages (Messages): The messages to process.
|
|
**kwargs: Additional keyword arguments.
|
|
|
|
Returns:
|
|
str: The result of the creation process.
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
@classmethod
|
|
def get_dict(cls) -> Dict[str, str]:
|
|
"""
|
|
Get a dictionary representation of the provider.
|
|
|
|
Returns:
|
|
Dict[str, str]: A dictionary with provider's details.
|
|
"""
|
|
return {'name': cls.__name__, 'url': cls.url}
|
|
|
|
class BaseRetryProvider(BaseProvider):
|
|
"""
|
|
Base class for a provider that implements retry logic.
|
|
|
|
Attributes:
|
|
providers (List[Type[BaseProvider]]): List of providers to use for retries.
|
|
shuffle (bool): Whether to shuffle the providers list.
|
|
exceptions (Dict[str, Exception]): Dictionary of exceptions encountered.
|
|
last_provider (Type[BaseProvider]): The last provider used.
|
|
"""
|
|
|
|
__name__: str = "RetryProvider"
|
|
supports_stream: bool = True
|
|
|
|
def __init__(
|
|
self,
|
|
providers: List[Type[BaseProvider]],
|
|
shuffle: bool = True
|
|
) -> None:
|
|
"""
|
|
Initialize the BaseRetryProvider.
|
|
|
|
Args:
|
|
providers (List[Type[BaseProvider]]): List of providers to use.
|
|
shuffle (bool): Whether to shuffle the providers list.
|
|
"""
|
|
self.providers = providers
|
|
self.shuffle = shuffle
|
|
self.working = True
|
|
self.exceptions: Dict[str, Exception] = {}
|
|
self.last_provider: Type[BaseProvider] = None
|
|
|
|
ProviderType = Union[Type[BaseProvider], BaseRetryProvider] |