mirror of
https://github.com/BerriAI/litellm.git
synced 2025-04-25 18:54:30 +00:00
* test: add initial e2e test * fix(vertex_ai/files): initial commit adding sync file create support * refactor: initial commit of vertex ai non-jsonl files reaching gcp endpoint * fix(vertex_ai/files/transformation.py): initial working commit of non-jsonl file call reaching backend endpoint * fix(vertex_ai/files/transformation.py): working e2e non-jsonl file upload * test: working e2e jsonl call * test: unit testing for jsonl file creation * fix(vertex_ai/transformation.py): reset file pointer after read allow multiple reads on same file object * fix: fix linting errors * fix: fix ruff linting errors * fix: fix import * fix: fix linting error * fix: fix linting error * fix(vertex_ai/files/transformation.py): fix linting error * test: update test * test: update tests * fix: fix linting errors * fix: fix test * fix: fix linting error
52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
from typing import List, Optional, Union
|
|
|
|
from httpx import Headers
|
|
|
|
from litellm.secret_managers.main import get_secret_str
|
|
from litellm.types.llms.openai import AllMessageValues
|
|
|
|
from ..base_llm.chat.transformation import BaseLLMException
|
|
|
|
|
|
class FireworksAIException(BaseLLMException):
|
|
pass
|
|
|
|
|
|
class FireworksAIMixin:
|
|
"""
|
|
Common Base Config functions across Fireworks AI Endpoints
|
|
"""
|
|
|
|
def get_error_class(
|
|
self, error_message: str, status_code: int, headers: Union[dict, Headers]
|
|
) -> BaseLLMException:
|
|
return FireworksAIException(
|
|
status_code=status_code,
|
|
message=error_message,
|
|
headers=headers,
|
|
)
|
|
|
|
def _get_api_key(self, api_key: Optional[str]) -> Optional[str]:
|
|
dynamic_api_key = api_key or (
|
|
get_secret_str("FIREWORKS_API_KEY")
|
|
or get_secret_str("FIREWORKS_AI_API_KEY")
|
|
or get_secret_str("FIREWORKSAI_API_KEY")
|
|
or get_secret_str("FIREWORKS_AI_TOKEN")
|
|
)
|
|
return dynamic_api_key
|
|
|
|
def validate_environment(
|
|
self,
|
|
headers: dict,
|
|
model: str,
|
|
messages: List[AllMessageValues],
|
|
optional_params: dict,
|
|
litellm_params: dict,
|
|
api_key: Optional[str] = None,
|
|
api_base: Optional[str] = None,
|
|
) -> dict:
|
|
api_key = self._get_api_key(api_key)
|
|
if api_key is None:
|
|
raise ValueError("FIREWORKS_API_KEY is not set")
|
|
|
|
return {"Authorization": "Bearer {}".format(api_key), **headers}
|