Spaces:
Sleeping
Sleeping
File size: 7,397 Bytes
469eae6 |
1 2 3 4 5 6 7 8 9 10 11 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 |
"""
Common helpers / utils across al OpenAI endpoints
"""
import hashlib
import json
from typing import Any, Dict, List, Literal, Optional, Union
import httpx
import openai
from openai import AsyncAzureOpenAI, AsyncOpenAI, AzureOpenAI, OpenAI
import litellm
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.custom_httpx.http_handler import _DEFAULT_TTL_FOR_HTTPX_CLIENTS
class OpenAIError(BaseLLMException):
def __init__(
self,
status_code: int,
message: str,
request: Optional[httpx.Request] = None,
response: Optional[httpx.Response] = None,
headers: Optional[Union[dict, httpx.Headers]] = None,
body: Optional[dict] = None,
):
self.status_code = status_code
self.message = message
self.headers = headers
if request:
self.request = request
else:
self.request = httpx.Request(method="POST", url="https://api.openai.com/v1")
if response:
self.response = response
else:
self.response = httpx.Response(
status_code=status_code, request=self.request
)
super().__init__(
status_code=status_code,
message=self.message,
headers=self.headers,
request=self.request,
response=self.response,
body=body,
)
####### Error Handling Utils for OpenAI API #######################
###################################################################
def drop_params_from_unprocessable_entity_error(
e: Union[openai.UnprocessableEntityError, httpx.HTTPStatusError],
data: Dict[str, Any],
) -> Dict[str, Any]:
"""
Helper function to read OpenAI UnprocessableEntityError and drop the params that raised an error from the error message.
Args:
e (UnprocessableEntityError): The UnprocessableEntityError exception
data (Dict[str, Any]): The original data dictionary containing all parameters
Returns:
Dict[str, Any]: A new dictionary with invalid parameters removed
"""
invalid_params: List[str] = []
if isinstance(e, httpx.HTTPStatusError):
error_json = e.response.json()
error_message = error_json.get("error", {})
error_body = error_message
else:
error_body = e.body
if (
error_body is not None
and isinstance(error_body, dict)
and error_body.get("message")
):
message = error_body.get("message", {})
if isinstance(message, str):
try:
message = json.loads(message)
except json.JSONDecodeError:
message = {"detail": message}
detail = message.get("detail")
if isinstance(detail, List) and len(detail) > 0 and isinstance(detail[0], dict):
for error_dict in detail:
if (
error_dict.get("loc")
and isinstance(error_dict.get("loc"), list)
and len(error_dict.get("loc")) == 2
):
invalid_params.append(error_dict["loc"][1])
new_data = {k: v for k, v in data.items() if k not in invalid_params}
return new_data
class BaseOpenAILLM:
"""
Base class for OpenAI LLMs for getting their httpx clients and SSL verification settings
"""
@staticmethod
def get_cached_openai_client(
client_initialization_params: dict, client_type: Literal["openai", "azure"]
) -> Optional[Union[OpenAI, AsyncOpenAI, AzureOpenAI, AsyncAzureOpenAI]]:
"""Retrieves the OpenAI client from the in-memory cache based on the client initialization parameters"""
_cache_key = BaseOpenAILLM.get_openai_client_cache_key(
client_initialization_params=client_initialization_params,
client_type=client_type,
)
_cached_client = litellm.in_memory_llm_clients_cache.get_cache(_cache_key)
return _cached_client
@staticmethod
def set_cached_openai_client(
openai_client: Union[OpenAI, AsyncOpenAI, AzureOpenAI, AsyncAzureOpenAI],
client_type: Literal["openai", "azure"],
client_initialization_params: dict,
):
"""Stores the OpenAI client in the in-memory cache for _DEFAULT_TTL_FOR_HTTPX_CLIENTS SECONDS"""
_cache_key = BaseOpenAILLM.get_openai_client_cache_key(
client_initialization_params=client_initialization_params,
client_type=client_type,
)
litellm.in_memory_llm_clients_cache.set_cache(
key=_cache_key,
value=openai_client,
ttl=_DEFAULT_TTL_FOR_HTTPX_CLIENTS,
)
@staticmethod
def get_openai_client_cache_key(
client_initialization_params: dict, client_type: Literal["openai", "azure"]
) -> str:
"""Creates a cache key for the OpenAI client based on the client initialization parameters"""
hashed_api_key = None
if client_initialization_params.get("api_key") is not None:
hash_object = hashlib.sha256(
client_initialization_params.get("api_key", "").encode()
)
# Hexadecimal representation of the hash
hashed_api_key = hash_object.hexdigest()
# Create a more readable cache key using a list of key-value pairs
key_parts = [
f"hashed_api_key={hashed_api_key}",
f"is_async={client_initialization_params.get('is_async')}",
]
LITELLM_CLIENT_SPECIFIC_PARAMS = [
"timeout",
"max_retries",
"organization",
"api_base",
]
openai_client_fields = (
BaseOpenAILLM.get_openai_client_initialization_param_fields(
client_type=client_type
)
+ LITELLM_CLIENT_SPECIFIC_PARAMS
)
for param in openai_client_fields:
key_parts.append(f"{param}={client_initialization_params.get(param)}")
_cache_key = ",".join(key_parts)
return _cache_key
@staticmethod
def get_openai_client_initialization_param_fields(
client_type: Literal["openai", "azure"]
) -> List[str]:
"""Returns a list of fields that are used to initialize the OpenAI client"""
import inspect
from openai import AzureOpenAI, OpenAI
if client_type == "openai":
signature = inspect.signature(OpenAI.__init__)
else:
signature = inspect.signature(AzureOpenAI.__init__)
# Extract parameter names, excluding 'self'
param_names = [param for param in signature.parameters if param != "self"]
return param_names
@staticmethod
def _get_async_http_client() -> Optional[httpx.AsyncClient]:
if litellm.aclient_session is not None:
return litellm.aclient_session
return httpx.AsyncClient(
limits=httpx.Limits(max_connections=1000, max_keepalive_connections=100),
verify=litellm.ssl_verify,
)
@staticmethod
def _get_sync_http_client() -> Optional[httpx.Client]:
if litellm.client_session is not None:
return litellm.client_session
return httpx.Client(
limits=httpx.Limits(max_connections=1000, max_keepalive_connections=100),
verify=litellm.ssl_verify,
)
|