Skip to content

base

Module for defining abstract classes for instructor client retrieval.

ClientFactory dataclass

Bases: ABC

An abstract base class for defining client factories to route API traffic to different providers.

This class serves as a template for creating specific client factories for various AI model providers (e.g., OpenAI, Anthropic, etc.). It provides a standardized interface for creating clients, formatting messages, and handling completions across different API providers.

Attributes:

Name Type Description
async_client AsyncInstructor

An asynchronous instructor client for making API calls.

The ClientFactory class defines several abstract methods that must be implemented by subclasses: - from_config: For creating a factory instance from configuration parameters. - format_messages: For converting Spark-specific message formats to provider-specific formats. - format_completion: For standardizing completion responses from different providers.

It also provides concrete implementations for creating completions with or without Pydantic models.

Usage

Subclass ClientFactory for each API provider, implementing the abstract methods according to the provider's specific requirements. Then use these subclasses to create provider-specific clients and handle API interactions in a standardized way across your application.

Example 
class OpenAIFactory(ClientFactory):
    @classmethod
    def from_config(cls, mode=None, base_url=None, api_key=None, **kwargs):
        # Implementation for creating an OpenAI client
        ...

    def format_messages(self, messages):
        # Implementation for formatting messages for OpenAI
        ...

    def format_completion(self, completion):
        # Implementation for formatting OpenAI completion
        ...

# Using the factory
openai_factory = OpenAIFactory.from_config(api_key="your-api-key")
completion = await openai_factory.create(...)
Source code in spark_instructor/factory/base.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
@dataclass
class ClientFactory(ABC):
    """An abstract base class for defining client factories to route API traffic to different providers.

    This class serves as a template for creating specific client factories for various AI model providers
    (e.g., OpenAI, Anthropic, etc.). It provides a standardized interface for creating clients,
    formatting messages, and handling completions across different API providers.

    Attributes:
        async_client (instructor.AsyncInstructor): An asynchronous instructor client for making API calls.

    The ClientFactory class defines several abstract methods that must be implemented by subclasses:
    - from_config: For creating a factory instance from configuration parameters.
    - format_messages: For converting Spark-specific message formats to provider-specific formats.
    - format_completion: For standardizing completion responses from different providers.

    It also provides concrete implementations for creating completions with or without Pydantic models.

    Usage:
        Subclass ClientFactory for each API provider, implementing the abstract methods according to
        the provider's specific requirements. Then use these subclasses to create provider-specific
        clients and handle API interactions in a standardized way across your application.

    Example:
        ```python
        class OpenAIFactory(ClientFactory):
            @classmethod
            def from_config(cls, mode=None, base_url=None, api_key=None, **kwargs):
                # Implementation for creating an OpenAI client
                ...

            def format_messages(self, messages):
                # Implementation for formatting messages for OpenAI
                ...

            def format_completion(self, completion):
                # Implementation for formatting OpenAI completion
                ...

        # Using the factory
        openai_factory = OpenAIFactory.from_config(api_key="your-api-key")
        completion = await openai_factory.create(...)
        ```
    """

    async_client: instructor.AsyncInstructor

    @classmethod
    @abstractmethod
    def from_config(
        cls,
        mode: Optional[instructor.Mode] = None,
        base_url: Optional[str] = None,
        api_key: Optional[str] = None,
        **kwargs,
    ) -> "ClientFactory":
        """Create a client factory instance from configuration parameters.

        This method should be implemented to initialize the factory with provider-specific settings.

        Args:
            mode (Optional[instructor.Mode]): The mode of operation for the instructor client.
            base_url (Optional[str]): The base URL for the API endpoint.
            api_key (Optional[str]): The API key for authentication.
            **kwargs: Additional keyword arguments for provider-specific configuration.

        Returns:
            ClientFactory: An instance of the concrete ClientFactory subclass.
        """
        pass

    @abstractmethod
    def format_messages(self, messages: SparkChatCompletionMessages) -> List[ChatCompletionMessageParam]:
        """Format Spark completion messages to provider-specific chat completion messages.

        This method should be implemented to convert the standardized Spark message format
        to the format expected by the specific API provider.

        Args:
            messages (SparkChatCompletionMessages): The messages in Spark format.

        Returns:
            List[ChatCompletionMessageParam]: The formatted messages ready for the API provider.
        """
        pass

    @abstractmethod
    def format_completion(self, completion: Any) -> OpenAICompletion:
        """Format a provider-specific completion to a standardized OpenAI-style completion.

        This method should be implemented to convert the completion response from the
        provider's format to a standardized OpenAICompletion format.

        Args:
            completion (Any): The completion response from the API provider.

        Returns:
            OpenAICompletion: The formatted completion in OpenAI-compatible format.
        """
        pass

    async def create_with_completion(
        self,
        response_model: Type[T],
        messages: SparkChatCompletionMessages,
        model: str,
        max_tokens: int,
        temperature: float,
        max_retries: int,
        **kwargs,
    ) -> Tuple[T, ChatCompletion]:
        """Create a Pydantic model instance along with the full completion response.

        This method sends a request to the API, formats the response into a Pydantic model,
        and returns both the model instance and the full completion details.

        Args:
            response_model (Type[T]): The Pydantic model class for structuring the response.
            messages (SparkChatCompletionMessages): The input messages for the completion.
            model (str): The name or identifier of the AI model to use.
            max_tokens (int): The maximum number of tokens in the completion response.
            temperature (float): The sampling temperature for the model's output.
            max_retries (int): The maximum number of retry attempts for failed requests.
            **kwargs: Additional keyword arguments for the API request.

        Returns:
            Tuple[T, ChatCompletion]: A tuple containing the Pydantic model instance and the full completion.
        """
        pydantic_object, completion = await self.async_client.chat.completions.create_with_completion(
            response_model=response_model,
            messages=self.format_messages(messages),
            model=model,
            max_retries=max_retries,
            max_tokens=max_tokens,
            temperature=temperature,
            **kwargs,
        )
        return pydantic_object, self.format_completion(completion)

    async def create(
        self,
        response_model: Type[T] | None,
        messages: SparkChatCompletionMessages,
        model: str,
        max_tokens: int,
        temperature: float,
        max_retries: int,
        **kwargs,
    ) -> ChatCompletion:
        """Create a completion response.

        This method sends a request to the API and returns the completion response.
        If a response_model is provided, it structures the response accordingly.

        Args:
            response_model (Type[T] | None): Optional Pydantic model class for structuring the response.
            messages (SparkChatCompletionMessages): The input messages for the completion.
            model (str): The name or identifier of the AI model to use.
            max_tokens (int): The maximum number of tokens in the completion response.
            temperature (float): The sampling temperature for the model's output.
            max_retries (int): The maximum number of retry attempts for failed requests.
            **kwargs: Additional keyword arguments for the API request.

        Returns:
            ChatCompletion: The completion response, formatted according to the OpenAI standard.
        """
        completion = await self.async_client.chat.completions.create(
            response_model=response_model,  # type: ignore
            messages=self.format_messages(messages),
            model=model,
            max_retries=max_retries,
            max_tokens=max_tokens,
            temperature=temperature,
            **kwargs,
        )
        return self.format_completion(completion)

create(response_model, messages, model, max_tokens, temperature, max_retries, **kwargs) async

Create a completion response.

This method sends a request to the API and returns the completion response. If a response_model is provided, it structures the response accordingly.

Parameters:

Name Type Description Default
response_model Type[T] | None

Optional Pydantic model class for structuring the response.

 required
messages SparkChatCompletionMessages

The input messages for the completion.

 required
model str

The name or identifier of the AI model to use.

 required
max_tokens int

The maximum number of tokens in the completion response.

 required
temperature float

The sampling temperature for the model's output.

 required
max_retries int

The maximum number of retry attempts for failed requests.

 required
**kwargs

Additional keyword arguments for the API request.

 {}

Returns:

Name Type Description
ChatCompletion ChatCompletion

The completion response, formatted according to the OpenAI standard.
Source code in spark_instructor/factory/base.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
async def create(
    self,
    response_model: Type[T] | None,
    messages: SparkChatCompletionMessages,
    model: str,
    max_tokens: int,
    temperature: float,
    max_retries: int,
    **kwargs,
) -> ChatCompletion:
    """Create a completion response.

    This method sends a request to the API and returns the completion response.
    If a response_model is provided, it structures the response accordingly.

    Args:
        response_model (Type[T] | None): Optional Pydantic model class for structuring the response.
        messages (SparkChatCompletionMessages): The input messages for the completion.
        model (str): The name or identifier of the AI model to use.
        max_tokens (int): The maximum number of tokens in the completion response.
        temperature (float): The sampling temperature for the model's output.
        max_retries (int): The maximum number of retry attempts for failed requests.
        **kwargs: Additional keyword arguments for the API request.

    Returns:
        ChatCompletion: The completion response, formatted according to the OpenAI standard.
    """
    completion = await self.async_client.chat.completions.create(
        response_model=response_model,  # type: ignore
        messages=self.format_messages(messages),
        model=model,
        max_retries=max_retries,
        max_tokens=max_tokens,
        temperature=temperature,
        **kwargs,
    )
    return self.format_completion(completion)

create_with_completion(response_model, messages, model, max_tokens, temperature, max_retries, **kwargs) async

Create a Pydantic model instance along with the full completion response.

This method sends a request to the API, formats the response into a Pydantic model, and returns both the model instance and the full completion details.

Parameters:

Name Type Description Default
response_model Type[T]

The Pydantic model class for structuring the response.

 required
messages SparkChatCompletionMessages

The input messages for the completion.

 required
model str

The name or identifier of the AI model to use.

 required
max_tokens int

The maximum number of tokens in the completion response.

 required
temperature float

The sampling temperature for the model's output.

 required
max_retries int

The maximum number of retry attempts for failed requests.

 required
**kwargs

Additional keyword arguments for the API request.

 {}

Returns:

Type Description
Tuple[T, ChatCompletion]

Tuple[T, ChatCompletion]: A tuple containing the Pydantic model instance and the full completion.
Source code in spark_instructor/factory/base.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
async def create_with_completion(
    self,
    response_model: Type[T],
    messages: SparkChatCompletionMessages,
    model: str,
    max_tokens: int,
    temperature: float,
    max_retries: int,
    **kwargs,
) -> Tuple[T, ChatCompletion]:
    """Create a Pydantic model instance along with the full completion response.

    This method sends a request to the API, formats the response into a Pydantic model,
    and returns both the model instance and the full completion details.

    Args:
        response_model (Type[T]): The Pydantic model class for structuring the response.
        messages (SparkChatCompletionMessages): The input messages for the completion.
        model (str): The name or identifier of the AI model to use.
        max_tokens (int): The maximum number of tokens in the completion response.
        temperature (float): The sampling temperature for the model's output.
        max_retries (int): The maximum number of retry attempts for failed requests.
        **kwargs: Additional keyword arguments for the API request.

    Returns:
        Tuple[T, ChatCompletion]: A tuple containing the Pydantic model instance and the full completion.
    """
    pydantic_object, completion = await self.async_client.chat.completions.create_with_completion(
        response_model=response_model,
        messages=self.format_messages(messages),
        model=model,
        max_retries=max_retries,
        max_tokens=max_tokens,
        temperature=temperature,
        **kwargs,
    )
    return pydantic_object, self.format_completion(completion)

format_completion(completion) abstractmethod

Format a provider-specific completion to a standardized OpenAI-style completion.

This method should be implemented to convert the completion response from the provider's format to a standardized OpenAICompletion format.

Parameters:

Name Type Description Default
completion Any

The completion response from the API provider.

 required

Returns:

Name Type Description
OpenAICompletion OpenAICompletion

The formatted completion in OpenAI-compatible format.
Source code in spark_instructor/factory/base.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@abstractmethod
def format_completion(self, completion: Any) -> OpenAICompletion:
    """Format a provider-specific completion to a standardized OpenAI-style completion.

    This method should be implemented to convert the completion response from the
    provider's format to a standardized OpenAICompletion format.

    Args:
        completion (Any): The completion response from the API provider.

    Returns:
        OpenAICompletion: The formatted completion in OpenAI-compatible format.
    """
    pass

format_messages(messages) abstractmethod

Format Spark completion messages to provider-specific chat completion messages.

This method should be implemented to convert the standardized Spark message format to the format expected by the specific API provider.

Parameters:

Name Type Description Default
messages SparkChatCompletionMessages

The messages in Spark format.

 required

Returns:

Type Description
List[ChatCompletionMessageParam]

List[ChatCompletionMessageParam]: The formatted messages ready for the API provider.
Source code in spark_instructor/factory/base.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@abstractmethod
def format_messages(self, messages: SparkChatCompletionMessages) -> List[ChatCompletionMessageParam]:
    """Format Spark completion messages to provider-specific chat completion messages.

    This method should be implemented to convert the standardized Spark message format
    to the format expected by the specific API provider.

    Args:
        messages (SparkChatCompletionMessages): The messages in Spark format.

    Returns:
        List[ChatCompletionMessageParam]: The formatted messages ready for the API provider.
    """
    pass

from_config(mode=None, base_url=None, api_key=None, **kwargs) abstractmethod classmethod

Create a client factory instance from configuration parameters.

This method should be implemented to initialize the factory with provider-specific settings.

Parameters:

Name Type Description Default
mode Optional[Mode]

The mode of operation for the instructor client.

 None
base_url Optional[str]

The base URL for the API endpoint.

 None
api_key Optional[str]

The API key for authentication.

 None
**kwargs

Additional keyword arguments for provider-specific configuration.

 {}

Returns:

Name Type Description
ClientFactory ClientFactory

An instance of the concrete ClientFactory subclass.
Source code in spark_instructor/factory/base.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@classmethod
@abstractmethod
def from_config(
    cls,
    mode: Optional[instructor.Mode] = None,
    base_url: Optional[str] = None,
    api_key: Optional[str] = None,
    **kwargs,
) -> "ClientFactory":
    """Create a client factory instance from configuration parameters.

    This method should be implemented to initialize the factory with provider-specific settings.

    Args:
        mode (Optional[instructor.Mode]): The mode of operation for the instructor client.
        base_url (Optional[str]): The base URL for the API endpoint.
        api_key (Optional[str]): The API key for authentication.
        **kwargs: Additional keyword arguments for provider-specific configuration.

    Returns:
        ClientFactory: An instance of the concrete ClientFactory subclass.
    """
    pass