Implementation:BerriAI Litellm LLM Base Class

Attribute	Value
Sources	litellm/llms/base.py
Domains	LLM Providers, Base Class, Extensibility
Last Updated	2026-02-15 16:00 GMT

Overview

BaseLLM is the foundational template base class for adding new LLM provider implementations via API calls in LiteLLM, providing default method stubs for completion, embedding, and response processing.

Description

The BaseLLM class defines the interface contract that all LLM provider implementations should follow. It provides default (no-op) implementations for validate_environment, completion, and embedding methods, as well as process_response and process_text_completion_response helper methods for handling response parsing across both synchronous and asynchronous code paths. The class also manages HTTP client sessions via create_client_session (sync) and create_aclient_session (async), with proper cleanup through __exit__ and __aexit__ methods. If litellm.client_session or litellm.aclient_session is set globally, those shared sessions are used; otherwise, new httpx.Client or httpx.AsyncClient instances are created.

Usage

Import and subclass BaseLLM when implementing a new LLM provider integration. Override the completion, embedding, and process_response methods to implement provider-specific logic. For custom chat LLMs with richer support (streaming, async, image generation), consider using CustomLLM which extends this class.

Code Reference

Source Location

litellm/llms/base.py

Signature

class BaseLLM:
    _client_session: Optional[httpx.Client] = None

    def process_response(self, model: str, response: httpx.Response, model_response: "ModelResponse",
                         stream: bool, logging_obj: Any, optional_params: dict, api_key: str,
                         data: Union[dict, str], messages: list, print_verbose, encoding
                         ) -> Union["ModelResponse", "CustomStreamWrapper"]
    def process_text_completion_response(self, model: str, response: httpx.Response,
                                          model_response: "TextCompletionResponse", stream: bool,
                                          logging_obj: Any, optional_params: dict, api_key: str,
                                          data: Union[dict, str], messages: list, print_verbose, encoding
                                          ) -> Union["TextCompletionResponse", "CustomStreamWrapper"]
    def create_client_session(self) -> httpx.Client
    def create_aclient_session(self) -> httpx.AsyncClient
    def __exit__(self)
    async def __aexit__(self, exc_type, exc_val, exc_tb)
    def validate_environment(self, *args, **kwargs) -> Optional[Any]
    def completion(self, *args, **kwargs) -> Any
    def embedding(self, *args, **kwargs) -> Any

Import

from litellm.llms.base import BaseLLM

I/O Contract

Inputs

Parameter	Type	Description
model	str	The model identifier.
response	httpx.Response	The raw HTTP response from the provider.
model_response	ModelResponse or TextCompletionResponse	The response object to populate.
stream	bool	Whether the response is streaming.
logging_obj	Any	The logging object for the call.
optional_params	dict	Optional parameters for the call.
api_key	str	The API key used for the call.
data	Union[dict, str]	The request data/payload.
messages	list	The messages sent in the request.

Outputs

Method	Return Type	Description
process_response	Union[ModelResponse, CustomStreamWrapper]	The processed model response or stream wrapper.
process_text_completion_response	Union[TextCompletionResponse, CustomStreamWrapper]	The processed text completion response.
create_client_session	httpx.Client	A synchronous HTTP client session.
create_aclient_session	httpx.AsyncClient	An asynchronous HTTP client session.
completion	Any	Returns None by default; override for provider-specific completion.
embedding	Any	Returns None by default; override for provider-specific embeddings.

Usage Examples

from litellm.llms.base import BaseLLM
import httpx

class MyProviderLLM(BaseLLM):
    def validate_environment(self, api_key=None, **kwargs):
        if not api_key:
            raise ValueError("API key required for MyProvider")
        return {"Authorization": f"Bearer {api_key}"}

    def completion(self, model, messages, api_key=None, **kwargs):
        headers = self.validate_environment(api_key=api_key)
        client = self.create_client_session()
        response = client.post(
            "https://api.myprovider.com/v1/completions",
            json={"model": model, "messages": messages},
            headers=headers,
        )
        return self.process_response(
            model=model,
            response=response,
            model_response=kwargs.get("model_response"),
            stream=False,
            logging_obj=kwargs.get("logging_obj"),
            optional_params=kwargs.get("optional_params", {}),
            api_key=api_key,
            data={"model": model, "messages": messages},
            messages=messages,
            print_verbose=kwargs.get("print_verbose"),
            encoding=kwargs.get("encoding"),
        )

Related Pages

• Implementation:BerriAI_Litellm_Custom_LLM_Handler - Extended custom LLM handler that builds on BaseLLM