Spaces:
Runtime error
Runtime error
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 | |
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() | |
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() | |
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] |