mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-21 00:21:49 +00:00
b74e696615
Types provider request and response bodies at their boundaries with TypedDicts and Protocols instead of dict[str, Any], so the untyped-to-typed crossing is paid once per boundary rather than once per field read. Removes 1,700 reportAny/reportExplicitAny errors and 1,953 basedpyright errors overall, plus 310 ruff strict-rule and 106 LIT-rule violations. No cast, type: ignore, noqa, or new Any annotations anywhere in the diff. Ratchets the basedpyright, ruff-strict, and type-discipline budgets to the new counts so the cleared headroom cannot silently grow back.
86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
## This is a template base class to be used for adding new LLM providers via API calls
|
|
from typing import TYPE_CHECKING, Any, Union
|
|
|
|
import httpx
|
|
|
|
import litellm
|
|
|
|
if TYPE_CHECKING:
|
|
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
|
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
|
|
from litellm.types.utils import ModelResponse, TextCompletionResponse
|
|
|
|
|
|
class BaseLLM:
|
|
_client_session: httpx.Client | None = None
|
|
|
|
def process_response(
|
|
self,
|
|
model: str,
|
|
response: httpx.Response,
|
|
model_response: "ModelResponse",
|
|
stream: bool,
|
|
logging_obj: "LiteLLMLoggingObj",
|
|
optional_params: dict,
|
|
api_key: str,
|
|
data: dict | str,
|
|
messages: list,
|
|
print_verbose,
|
|
encoding,
|
|
) -> Union["ModelResponse", "CustomStreamWrapper"]:
|
|
"""
|
|
Helper function to process the response across sync + async completion calls
|
|
"""
|
|
return model_response
|
|
|
|
def process_text_completion_response(
|
|
self,
|
|
model: str,
|
|
response: httpx.Response,
|
|
model_response: "TextCompletionResponse",
|
|
stream: bool,
|
|
logging_obj: "LiteLLMLoggingObj",
|
|
optional_params: dict,
|
|
api_key: str,
|
|
data: dict | str,
|
|
messages: list,
|
|
print_verbose,
|
|
encoding,
|
|
) -> Union["TextCompletionResponse", "CustomStreamWrapper"]:
|
|
"""
|
|
Helper function to process the response across sync + async completion calls
|
|
"""
|
|
return model_response
|
|
|
|
def create_client_session(self):
|
|
if litellm.client_session:
|
|
_client_session = litellm.client_session
|
|
else:
|
|
_client_session = httpx.Client()
|
|
|
|
return _client_session
|
|
|
|
def create_aclient_session(self):
|
|
if litellm.aclient_session:
|
|
_aclient_session = litellm.aclient_session
|
|
else:
|
|
_aclient_session = httpx.AsyncClient()
|
|
|
|
return _aclient_session
|
|
|
|
def __exit__(self):
|
|
if hasattr(self, "_client_session") and self._client_session is not None:
|
|
self._client_session.close()
|
|
|
|
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
|
if hasattr(self, "_aclient_session"):
|
|
await self._aclient_session.aclose()
|
|
|
|
def validate_environment(self, *args, **kwargs) -> Any | None: # set up the environment required to run the model
|
|
return None
|
|
|
|
def completion(self, *args, **kwargs) -> Any: # logic for parsing in - calling - parsing out model completion calls
|
|
return None
|
|
|
|
def embedding(self, *args, **kwargs) -> Any: # logic for parsing in - calling - parsing out model embedding calls
|
|
return None
|