Zum Inhalt

Basis-Provider

llm_client.providers.base_provider

Base provider interface for LLM clients with streaming support.

Classes

BaseProvider

Bases: ABC

Abstract base class for LLM providers.

This class defines the interface that all LLM providers must implement. Each provider handles the specific API communication and response parsing for its respective service.

Attributes:

Name Type Description
llm

Name of the model to use.
temperature

Sampling temperature for generation.
max_tokens

Maximum number of tokens to generate.
client Any

The underlying API client instance.
Source code in llm_client/providers/base_provider.py 
 12
 13
 14
 15
 16
 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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
class BaseProvider(ABC):
    """Abstract base class for LLM providers.

    This class defines the interface that all LLM providers must implement.
    Each provider handles the specific API communication and response parsing
    for its respective service.

    Attributes:
        llm: Name of the model to use.
        temperature: Sampling temperature for generation.
        max_tokens: Maximum number of tokens to generate.
        client: The underlying API client instance.
    """

    def __init__(
        self, llm: str, temperature: float = 0.7, max_tokens: int = 512, **kwargs: Any
    ) -> None:
        """Initialize the provider.

        Args:
            llm (str): Model name to use.
            temperature (float): Sampling temperature (0.0 to 2.0).
            max_tokens (int): Maximum tokens to generate.
            **kwargs: Additional provider-specific parameters.
        """
        self.llm = llm
        self.temperature = temperature
        self.max_tokens = max_tokens
        self.client: Any = None
        self._initialize_client(**kwargs)

    @abstractmethod
    def _initialize_client(self, **kwargs: Any) -> None:
        """Initialize the API client.

        This method should create and configure the provider's API client.

        Args:
            **kwargs: Provider-specific initialization parameters.

        Raises:
            APIKeyNotFoundError: If required API key is missing.
            ProviderNotAvailableError: If required package is not installed.
        """
        pass

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10),
        reraise=True,
    )
    def chat_completion(self, messages: list[dict[str, str]]) -> str:
        """Execute a chat completion request with retry logic.

        This method implements exponential backoff retry logic to handle
        transient API failures. It will retry up to 3 times with increasing
        delays between attempts.

        Args:
            messages (list[dict[str, str]]): List of message dictionaries with 'role' and 'content' keys.

        Returns:
            str: The generated text response.

        Raises:
            ChatCompletionError: If the API call fails after all retries.

        Examples:
            >>> messages = [{"role": "user", "content": "Hello"}]
            >>> response = provider.chat_completion(messages)
        """
        try:
            return self._chat_completion_impl(messages)
        except Exception as e:
            raise ChatCompletionError(self.__class__.__name__, e) from e

    @abstractmethod
    def _chat_completion_impl(self, messages: list[dict[str, str]]) -> str:
        """Internal implementation of chat completion.

        This method should be implemented by concrete providers to perform
        the actual API call without retry logic.

        Args:
            messages (list[dict[str, str]]): List of message dictionaries.

        Returns:
            str: The generated text response.

        Raises:
            Exception: Any API-specific exceptions.
        """
        pass

    def chat_completion_with_tools(
        self,
        messages: list[dict[str, str]],
        tools: list[dict],
        tool_choice: str | dict | None = None,
    ) -> dict:
        """Execute chat completion with function/tool calling support.

        Args:
            messages (list[dict[str, str]]): List of message dictionaries.
            tools (list[dict]): List of tool/function definitions.
            tool_choice (str | dict | None): Controls which tool is called ("auto", "none", specific tool).

        Returns:
            dict: Dictionary with 'content' (str or None) and 'tool_calls' (list or None).

        Raises:
            ChatCompletionError: If the API call fails.
            NotImplementedError: If provider doesn't support tools.

        Examples:
            >>> tools = [{
            ...     "type": "function",
            ...     "function": {
            ...         "name": "get_weather",
            ...         "description": "Get weather for a location",
            ...         "parameters": {
            ...             "type": "object",
            ...             "properties": {
            ...                 "location": {"type": "string"}
            ...             }
            ...         }
            ...     }
            ... }]
            >>> result = provider.chat_completion_with_tools(messages, tools)
        """
        try:
            return self._chat_completion_with_tools_impl(messages, tools, tool_choice)
        except NotImplementedError as err:
            raise NotImplementedError(
                f"{self.__class__.__name__} does not support tool calling"
            ) from err
        except Exception as e:
            raise ChatCompletionError(self.__class__.__name__, e) from e

    def _chat_completion_with_tools_impl(
        self,
        messages: list[dict[str, str]],
        tools: list[dict],
        tool_choice: str | dict | None = None,
    ) -> dict:
        """Internal implementation of tool calling.

        Should be overridden by providers that support tools.

        Args:
            messages (list[dict[str, str]]): List of message dictionaries.
            tools (list[dict]): List of tool definitions.
            tool_choice (str | dict | None): Tool choice parameter.

        Returns:
            dict: Dict with 'content' and 'tool_calls' keys.

        Raises:
            NotImplementedError: If provider doesn't support tools.
        """
        raise NotImplementedError("Tool calling not implemented for this provider")

    def chat_completion_with_files(
        self,
        messages: list[dict[str, str]],
        files: list[str] | None = None,
    ) -> str:
        """Execute chat completion with file uploads.

        This method allows sending files (images, PDFs) along with the chat messages.
        File support varies by provider:
        - OpenAI: Images (PNG, JPEG, WEBP, GIF), PDFs
        - Gemini: Images, PDFs, Videos, Audio
        - Groq: Limited vision support
        - Ollama: Vision models only

        Args:
            messages (list[dict[str, str]]): List of message dictionaries with 'role' and 'content' keys.
            files (list[str] | None): List of file paths to upload. Supported formats depend on provider.

        Returns:
            str: The generated text response.

        Raises:
            FileUploadNotSupportedError: If provider doesn't support file uploads.
            FileNotFoundError: If a specified file doesn't exist.
            ChatCompletionError: If the API call fails.

        Examples:
            >>> messages = [{"role": "user", "content": "Describe this image"}]
            >>> response = provider.chat_completion_with_files(
            ...     messages,
            ...     files=["image.jpg"]
            ... )

            >>> # Multiple files
            >>> response = provider.chat_completion_with_files(
            ...     messages,
            ...     files=["document.pdf", "chart.png"]
            ... )
        """
        try:
            return self._chat_completion_with_files_impl(messages, files)
        except NotImplementedError as err:
            from ..exceptions import FileUploadNotSupportedError

            raise FileUploadNotSupportedError(
                self.__class__.__name__, "Provider does not support file uploads"
            ) from err
        except Exception as e:
            raise ChatCompletionError(self.__class__.__name__, e) from e

    def _chat_completion_with_files_impl(
        self,
        messages: list[dict[str, str]],
        files: list[str] | None = None,
    ) -> str:
        """Internal implementation of file upload chat completion.

        Should be overridden by providers that support file uploads.

        Args:
            messages (list[dict[str, str]]): List of message dictionaries.
            files (list[str] | None): List of file paths.

        Returns:
            str: Generated text response.

        Raises:
            NotImplementedError: If provider doesn't support file uploads.
        """
        raise NotImplementedError("File upload not implemented for this provider")

    def chat_completion_stream(self, messages: list[dict[str, str]]) -> Iterator[str]:
        """Stream response tokens as they arrive.

        This method returns an iterator that yields response tokens as they
        are generated by the LLM, enabling real-time display of responses.

        Args:
            messages (list[dict[str, str]]): List of message dictionaries with 'role' and 'content' keys.

        Yields:
            str: Individual tokens or chunks of the response text.

        Raises:
            StreamingNotSupportedError: If streaming is not supported.
            ChatCompletionError: If the streaming API call fails.

        Examples:
            >>> messages = [{"role": "user", "content": "Tell me a story"}]
            >>> for chunk in provider.chat_completion_stream(messages):
            ...     print(chunk, end="", flush=True)
        """
        try:
            return self._chat_completion_stream_impl(messages)
        except NotImplementedError as err:
            from ..exceptions import StreamingNotSupportedError

            raise StreamingNotSupportedError(
                self.__class__.__name__, "Provider does not implement streaming"
            ) from err
        except Exception as e:
            raise ChatCompletionError(self.__class__.__name__, e) from e

    def _chat_completion_stream_impl(self, messages: list[dict[str, str]]) -> Iterator[str]:
        """Internal implementation of streaming chat completion.

        This method can be overridden by concrete providers that support
        streaming. Default implementation raises NotImplementedError.

        Args:
            messages (list[dict[str, str]]): List of message dictionaries.

        Yields:
            str: Individual tokens or chunks of the response text.

        Raises:
            NotImplementedError: If provider doesn't support streaming.
        """
        raise NotImplementedError("Streaming not implemented for this provider")

    @staticmethod
    @abstractmethod
    def get_default_model() -> str:
        """Get the default model name for this provider.

        Returns:
            Default model name as string.
        """
        pass

    @staticmethod
    @abstractmethod
    def is_available() -> bool:
        """Check if the provider's package is installed.

        Returns:
            True if the provider can be used, False otherwise.
        """
        pass

    def __repr__(self) -> str:
        """Return string representation of the provider.

        Returns:
            String with provider info.
        """
        return (
            f"{self.__class__.__name__}(" f"model={self.llm}, " f"temperature={self.temperature})"
        )
Functions
__init__(llm, temperature=0.7, max_tokens=512, **kwargs)

Initialize the provider.

Parameters:

Name Type Description Default
llm str

Model name to use.

 required
temperature float

Sampling temperature (0.0 to 2.0).

 0.7
max_tokens int

Maximum tokens to generate.

 512
**kwargs Any

Additional provider-specific parameters.

 {}
Source code in llm_client/providers/base_provider.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def __init__(
    self, llm: str, temperature: float = 0.7, max_tokens: int = 512, **kwargs: Any
) -> None:
    """Initialize the provider.

    Args:
        llm (str): Model name to use.
        temperature (float): Sampling temperature (0.0 to 2.0).
        max_tokens (int): Maximum tokens to generate.
        **kwargs: Additional provider-specific parameters.
    """
    self.llm = llm
    self.temperature = temperature
    self.max_tokens = max_tokens
    self.client: Any = None
    self._initialize_client(**kwargs)
__repr__()

Return string representation of the provider.

Returns:

Type Description
str

String with provider info.
Source code in llm_client/providers/base_provider.py 
314
315
316
317
318
319
320
321
322
def __repr__(self) -> str:
    """Return string representation of the provider.

    Returns:
        String with provider info.
    """
    return (
        f"{self.__class__.__name__}(" f"model={self.llm}, " f"temperature={self.temperature})"
    )
chat_completion(messages)

Execute a chat completion request with retry logic.

This method implements exponential backoff retry logic to handle transient API failures. It will retry up to 3 times with increasing delays between attempts.

Parameters:

Name Type Description Default
messages list[dict[str, str]]

List of message dictionaries with 'role' and 'content' keys.

 required

Returns:

Name Type Description
str str

The generated text response.

Raises:

Type Description
ChatCompletionError

If the API call fails after all retries.

Examples:

>>> messages = [{"role": "user", "content": "Hello"}]
>>> response = provider.chat_completion(messages)
Source code in llm_client/providers/base_provider.py 
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
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    reraise=True,
)
def chat_completion(self, messages: list[dict[str, str]]) -> str:
    """Execute a chat completion request with retry logic.

    This method implements exponential backoff retry logic to handle
    transient API failures. It will retry up to 3 times with increasing
    delays between attempts.

    Args:
        messages (list[dict[str, str]]): List of message dictionaries with 'role' and 'content' keys.

    Returns:
        str: The generated text response.

    Raises:
        ChatCompletionError: If the API call fails after all retries.

    Examples:
        >>> messages = [{"role": "user", "content": "Hello"}]
        >>> response = provider.chat_completion(messages)
    """
    try:
        return self._chat_completion_impl(messages)
    except Exception as e:
        raise ChatCompletionError(self.__class__.__name__, e) from e
chat_completion_stream(messages)

Stream response tokens as they arrive.

This method returns an iterator that yields response tokens as they are generated by the LLM, enabling real-time display of responses.

Parameters:

Name Type Description Default
messages list[dict[str, str]]

List of message dictionaries with 'role' and 'content' keys.

 required

Yields:

Name Type Description
str str

Individual tokens or chunks of the response text.

Raises:

Type Description
StreamingNotSupportedError

If streaming is not supported.
ChatCompletionError

If the streaming API call fails.

Examples:

>>> messages = [{"role": "user", "content": "Tell me a story"}]
>>> for chunk in provider.chat_completion_stream(messages):
...     print(chunk, end="", flush=True)
Source code in llm_client/providers/base_provider.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def chat_completion_stream(self, messages: list[dict[str, str]]) -> Iterator[str]:
    """Stream response tokens as they arrive.

    This method returns an iterator that yields response tokens as they
    are generated by the LLM, enabling real-time display of responses.

    Args:
        messages (list[dict[str, str]]): List of message dictionaries with 'role' and 'content' keys.

    Yields:
        str: Individual tokens or chunks of the response text.

    Raises:
        StreamingNotSupportedError: If streaming is not supported.
        ChatCompletionError: If the streaming API call fails.

    Examples:
        >>> messages = [{"role": "user", "content": "Tell me a story"}]
        >>> for chunk in provider.chat_completion_stream(messages):
        ...     print(chunk, end="", flush=True)
    """
    try:
        return self._chat_completion_stream_impl(messages)
    except NotImplementedError as err:
        from ..exceptions import StreamingNotSupportedError

        raise StreamingNotSupportedError(
            self.__class__.__name__, "Provider does not implement streaming"
        ) from err
    except Exception as e:
        raise ChatCompletionError(self.__class__.__name__, e) from e
chat_completion_with_files(messages, files=None)

Execute chat completion with file uploads.

This method allows sending files (images, PDFs) along with the chat messages. File support varies by provider: - OpenAI: Images (PNG, JPEG, WEBP, GIF), PDFs - Gemini: Images, PDFs, Videos, Audio - Groq: Limited vision support - Ollama: Vision models only

Parameters:

Name Type Description Default
messages list[dict[str, str]]

List of message dictionaries with 'role' and 'content' keys.

 required
files list[str] | None

List of file paths to upload. Supported formats depend on provider.

 None

Returns:

Name Type Description
str str

The generated text response.

Raises:

Type Description
FileUploadNotSupportedError

If provider doesn't support file uploads.
FileNotFoundError

If a specified file doesn't exist.
ChatCompletionError

If the API call fails.

Examples:

>>> messages = [{"role": "user", "content": "Describe this image"}]
>>> response = provider.chat_completion_with_files(
...     messages,
...     files=["image.jpg"]
... )

>>> # Multiple files
>>> response = provider.chat_completion_with_files(
...     messages,
...     files=["document.pdf", "chart.png"]
... )
Source code in llm_client/providers/base_provider.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def chat_completion_with_files(
    self,
    messages: list[dict[str, str]],
    files: list[str] | None = None,
) -> str:
    """Execute chat completion with file uploads.

    This method allows sending files (images, PDFs) along with the chat messages.
    File support varies by provider:
    - OpenAI: Images (PNG, JPEG, WEBP, GIF), PDFs
    - Gemini: Images, PDFs, Videos, Audio
    - Groq: Limited vision support
    - Ollama: Vision models only

    Args:
        messages (list[dict[str, str]]): List of message dictionaries with 'role' and 'content' keys.
        files (list[str] | None): List of file paths to upload. Supported formats depend on provider.

    Returns:
        str: The generated text response.

    Raises:
        FileUploadNotSupportedError: If provider doesn't support file uploads.
        FileNotFoundError: If a specified file doesn't exist.
        ChatCompletionError: If the API call fails.

    Examples:
        >>> messages = [{"role": "user", "content": "Describe this image"}]
        >>> response = provider.chat_completion_with_files(
        ...     messages,
        ...     files=["image.jpg"]
        ... )

        >>> # Multiple files
        >>> response = provider.chat_completion_with_files(
        ...     messages,
        ...     files=["document.pdf", "chart.png"]
        ... )
    """
    try:
        return self._chat_completion_with_files_impl(messages, files)
    except NotImplementedError as err:
        from ..exceptions import FileUploadNotSupportedError

        raise FileUploadNotSupportedError(
            self.__class__.__name__, "Provider does not support file uploads"
        ) from err
    except Exception as e:
        raise ChatCompletionError(self.__class__.__name__, e) from e
chat_completion_with_tools(messages, tools, tool_choice=None)

Execute chat completion with function/tool calling support.

Parameters:

Name Type Description Default
messages list[dict[str, str]]

List of message dictionaries.

 required
tools list[dict]

List of tool/function definitions.

 required
tool_choice str | dict | None

Controls which tool is called ("auto", "none", specific tool).

 None

Returns:

Name Type Description
dict dict

Dictionary with 'content' (str or None) and 'tool_calls' (list or None).

Raises:

Type Description
ChatCompletionError

If the API call fails.
NotImplementedError

If provider doesn't support tools.

Examples:

>>> tools = [{
...     "type": "function",
...     "function": {
...         "name": "get_weather",
...         "description": "Get weather for a location",
...         "parameters": {
...             "type": "object",
...             "properties": {
...                 "location": {"type": "string"}
...             }
...         }
...     }
... }]
>>> result = provider.chat_completion_with_tools(messages, tools)
Source code in llm_client/providers/base_provider.py 
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
def chat_completion_with_tools(
    self,
    messages: list[dict[str, str]],
    tools: list[dict],
    tool_choice: str | dict | None = None,
) -> dict:
    """Execute chat completion with function/tool calling support.

    Args:
        messages (list[dict[str, str]]): List of message dictionaries.
        tools (list[dict]): List of tool/function definitions.
        tool_choice (str | dict | None): Controls which tool is called ("auto", "none", specific tool).

    Returns:
        dict: Dictionary with 'content' (str or None) and 'tool_calls' (list or None).

    Raises:
        ChatCompletionError: If the API call fails.
        NotImplementedError: If provider doesn't support tools.

    Examples:
        >>> tools = [{
        ...     "type": "function",
        ...     "function": {
        ...         "name": "get_weather",
        ...         "description": "Get weather for a location",
        ...         "parameters": {
        ...             "type": "object",
        ...             "properties": {
        ...                 "location": {"type": "string"}
        ...             }
        ...         }
        ...     }
        ... }]
        >>> result = provider.chat_completion_with_tools(messages, tools)
    """
    try:
        return self._chat_completion_with_tools_impl(messages, tools, tool_choice)
    except NotImplementedError as err:
        raise NotImplementedError(
            f"{self.__class__.__name__} does not support tool calling"
        ) from err
    except Exception as e:
        raise ChatCompletionError(self.__class__.__name__, e) from e
get_default_model() abstractmethod staticmethod

Get the default model name for this provider.

Returns:

Type Description
str

Default model name as string.
Source code in llm_client/providers/base_provider.py 
294
295
296
297
298
299
300
301
302
@staticmethod
@abstractmethod
def get_default_model() -> str:
    """Get the default model name for this provider.

    Returns:
        Default model name as string.
    """
    pass
is_available() abstractmethod staticmethod

Check if the provider's package is installed.

Returns:

Type Description
bool

True if the provider can be used, False otherwise.
Source code in llm_client/providers/base_provider.py 
304
305
306
307
308
309
310
311
312
@staticmethod
@abstractmethod
def is_available() -> bool:
    """Check if the provider's package is installed.

    Returns:
        True if the provider can be used, False otherwise.
    """
    pass