From b9515c3b9641ecd4df294bac04e2863bb02e8b0b Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 12:53:46 -0700 Subject: [PATCH 01/20] feat - add batches types --- litellm/types/llms/openai.py | 52 +++++++++++++++++++++++++++++++++++- 1 file changed, 51 insertions(+), 1 deletion(-) diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py index 1c60ad6dbb..60f7a90417 100644 --- a/litellm/types/llms/openai.py +++ b/litellm/types/llms/openai.py @@ -6,7 +6,7 @@ from typing import ( Literal, Iterable, ) -from typing_extensions import override, Required +from typing_extensions import override, Required, Dict from pydantic import BaseModel from openai.types.beta.threads.message_content import MessageContent @@ -146,3 +146,53 @@ class Thread(BaseModel): object: Literal["thread"] """The object type, which is always `thread`.""" + + +# OpenAI Batches Types +class CreateBatchRequest(BaseModel): + """ + CreateBatchRequest + """ + + completion_window: Literal["24h"] + endpoint: Literal["/v1/chat/completions", "/v1/embeddings", "/v1/completions"] + input_file_id: str + metadata: Optional[Dict[str, str]] = None + extra_headers: Optional[Dict[str, str]] = None + extra_body: Optional[Dict[str, str]] = None + timeout: Optional[float] = None + + +class RetrieveBatchRequest(BaseModel): + """ + RetrieveBatchRequest + """ + + batch_id: str + extra_headers: Optional[Dict[str, str]] = None + extra_body: Optional[Dict[str, str]] = None + timeout: Optional[float] = None + + +class CancelBatchRequest(BaseModel): + """ + CancelBatchRequest + """ + + batch_id: str + extra_headers: Optional[Dict[str, str]] = None + extra_body: Optional[Dict[str, str]] = None + timeout: Optional[float] = None + + +class ListBatchRequest(BaseModel): + """ + ListBatchRequest - List your organization's batches + Calls https://api.openai.com/v1/batches + """ + + after: Optional[str] = None + limit: Optional[int] = 20 + extra_headers: Optional[Dict[str, str]] = None + extra_body: Optional[Dict[str, str]] = None + timeout: Optional[float] = None From 0af4c9206fe94d06d8aacd33de63f8fa5963a87c Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 14:54:58 -0700 Subject: [PATCH 02/20] test - openai batches file --- litellm/tests/openai_batch_completions.jsonl | 2 ++ 1 file changed, 2 insertions(+) create mode 100644 litellm/tests/openai_batch_completions.jsonl diff --git a/litellm/tests/openai_batch_completions.jsonl b/litellm/tests/openai_batch_completions.jsonl new file mode 100644 index 0000000000..05448952a0 --- /dev/null +++ b/litellm/tests/openai_batch_completions.jsonl @@ -0,0 +1,2 @@ +{"custom_id": "request-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-3.5-turbo-0125", "messages": [{"role": "system", "content": "You are a helpful assistant."},{"role": "user", "content": "Hello world!"}],"max_tokens": 10}} +{"custom_id": "request-2", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "gpt-3.5-turbo-0125", "messages": [{"role": "system", "content": "You are an unhelpful assistant."},{"role": "user", "content": "Hello world!"}],"max_tokens": 10}} \ No newline at end of file From d5dbf084ed2458a7fa6605e889c50fac9fde867e Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 15:35:11 -0700 Subject: [PATCH 03/20] feat - import batches in __init__ --- litellm/__init__.py | 1 + litellm/batches/main.py | 239 +++++++++++++++++++++++++++ litellm/llms/openai.py | 183 ++++++++++++++++++++ litellm/tests/test_openai_batches.py | 58 +++++++ litellm/types/llms/openai.py | 78 ++++++--- 5 files changed, 539 insertions(+), 20 deletions(-) create mode 100644 litellm/batches/main.py create mode 100644 litellm/tests/test_openai_batches.py diff --git a/litellm/__init__.py b/litellm/__init__.py index 3c78c9b270..56a2088e7e 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -797,3 +797,4 @@ from .budget_manager import BudgetManager from .proxy.proxy_cli import run_server from .router import Router from .assistants.main import * +from .batches.main import * diff --git a/litellm/batches/main.py b/litellm/batches/main.py new file mode 100644 index 0000000000..6aa0d8d884 --- /dev/null +++ b/litellm/batches/main.py @@ -0,0 +1,239 @@ +""" +Main File for Batches API implementation + +https://platform.openai.com/docs/api-reference/batch + +- create_batch() +- retrieve_batch() +- cancel_batch() +- list_batch() + +""" + +from typing import Iterable +import os +import litellm +from openai import OpenAI +import httpx +from litellm import client +from litellm.utils import supports_httpx_timeout +from ..types.router import * +from ..llms.openai import OpenAIBatchesAPI, OpenAIFilesAPI +from ..types.llms.openai import ( + CreateBatchRequest, + RetrieveBatchRequest, + CancelBatchRequest, + CreateFileRequest, + FileTypes, + FileObject, +) + +from typing import Literal, Optional, Dict + +####### ENVIRONMENT VARIABLES ################### +openai_batches_instance = OpenAIBatchesAPI() +openai_files_instance = OpenAIFilesAPI() +################################################# + + +def create_file( + file: FileTypes, + purpose: Literal["assistants", "batch", "fine-tune"], + custom_llm_provider: Literal["openai"] = "openai", + extra_headers: Optional[Dict[str, str]] = None, + extra_body: Optional[Dict[str, str]] = None, + **kwargs, +) -> FileObject: + try: + optional_params = GenericLiteLLMParams(**kwargs) + if custom_llm_provider == "openai": + # for deepinfra/perplexity/anyscale/groq we check in get_llm_provider and pass in the api base from there + api_base = ( + optional_params.api_base + or litellm.api_base + or os.getenv("OPENAI_API_BASE") + or "https://api.openai.com/v1" + ) + organization = ( + optional_params.organization + or litellm.organization + or os.getenv("OPENAI_ORGANIZATION", None) + or None # default - https://github.com/openai/openai-python/blob/284c1799070c723c6a553337134148a7ab088dd8/openai/util.py#L105 + ) + # set API KEY + api_key = ( + optional_params.api_key + or litellm.api_key # for deepinfra/perplexity/anyscale we check in get_llm_provider and pass in the api key from there + or litellm.openai_key + or os.getenv("OPENAI_API_KEY") + ) + ### TIMEOUT LOGIC ### + timeout = ( + optional_params.timeout or kwargs.get("request_timeout", 600) or 600 + ) + # set timeout for 10 minutes by default + + if ( + timeout is not None + and isinstance(timeout, httpx.Timeout) + and supports_httpx_timeout(custom_llm_provider) == False + ): + read_timeout = timeout.read or 600 + timeout = read_timeout # default 10 min timeout + elif timeout is not None and not isinstance(timeout, httpx.Timeout): + timeout = float(timeout) # type: ignore + elif timeout is None: + timeout = 600.0 + + _create_file_request = CreateFileRequest( + file=file, + purpose=purpose, + extra_headers=extra_headers, + extra_body=extra_body, + ) + + response = openai_files_instance.create_file( + api_base=api_base, + api_key=api_key, + timeout=timeout, + max_retries=optional_params.max_retries, + organization=organization, + create_file_data=_create_file_request, + ) + else: + raise litellm.exceptions.BadRequestError( + message="LiteLLM doesn't support {} for 'create_batch'. Only 'openai' is supported.".format( + custom_llm_provider + ), + model="n/a", + llm_provider=custom_llm_provider, + response=httpx.Response( + status_code=400, + content="Unsupported provider", + request=httpx.Request(method="create_thread", url="https://github.com/BerriAI/litellm"), # type: ignore + ), + ) + return response + except Exception as e: + raise e + + +def create_batch( + completion_window: Literal["24h"], + endpoint: Literal["/v1/chat/completions", "/v1/embeddings", "/v1/completions"], + input_file_id: str, + custom_llm_provider: Literal["openai"] = "openai", + metadata: Optional[Dict[str, str]] = None, + extra_headers: Optional[Dict[str, str]] = None, + extra_body: Optional[Dict[str, str]] = None, + **kwargs, +): + """ + Creates and executes a batch from an uploaded file of request + + LiteLLM Equivalent of POST: https://api.openai.com/v1/batches + """ + try: + optional_params = GenericLiteLLMParams(**kwargs) + if custom_llm_provider == "openai": + + # for deepinfra/perplexity/anyscale/groq we check in get_llm_provider and pass in the api base from there + api_base = ( + optional_params.api_base + or litellm.api_base + or os.getenv("OPENAI_API_BASE") + or "https://api.openai.com/v1" + ) + organization = ( + optional_params.organization + or litellm.organization + or os.getenv("OPENAI_ORGANIZATION", None) + or None # default - https://github.com/openai/openai-python/blob/284c1799070c723c6a553337134148a7ab088dd8/openai/util.py#L105 + ) + # set API KEY + api_key = ( + optional_params.api_key + or litellm.api_key # for deepinfra/perplexity/anyscale we check in get_llm_provider and pass in the api key from there + or litellm.openai_key + or os.getenv("OPENAI_API_KEY") + ) + ### TIMEOUT LOGIC ### + timeout = ( + optional_params.timeout or kwargs.get("request_timeout", 600) or 600 + ) + # set timeout for 10 minutes by default + + if ( + timeout is not None + and isinstance(timeout, httpx.Timeout) + and supports_httpx_timeout(custom_llm_provider) == False + ): + read_timeout = timeout.read or 600 + timeout = read_timeout # default 10 min timeout + elif timeout is not None and not isinstance(timeout, httpx.Timeout): + timeout = float(timeout) # type: ignore + elif timeout is None: + timeout = 600.0 + + _create_batch_request = CreateBatchRequest( + completion_window=completion_window, + endpoint=endpoint, + input_file_id=input_file_id, + metadata=metadata, + extra_headers=extra_headers, + extra_body=extra_body, + ) + + response = openai_batches_instance.create_batch( + api_base=api_base, + api_key=api_key, + organization=organization, + create_batch_data=_create_batch_request, + timeout=timeout, + max_retries=optional_params.max_retries, + ) + else: + raise litellm.exceptions.BadRequestError( + message="LiteLLM doesn't support {} for 'create_batch'. Only 'openai' is supported.".format( + custom_llm_provider + ), + model="n/a", + llm_provider=custom_llm_provider, + response=httpx.Response( + status_code=400, + content="Unsupported provider", + request=httpx.Request(method="create_thread", url="https://github.com/BerriAI/litellm"), # type: ignore + ), + ) + return response + except Exception as e: + raise e + + +def retrieve_batch(): + pass + + +def cancel_batch(): + pass + + +def list_batch(): + pass + + +# Async Functions +async def acreate_batch(): + pass + + +async def aretrieve_batch(): + pass + + +async def acancel_batch(): + pass + + +async def alist_batch(): + pass diff --git a/litellm/llms/openai.py b/litellm/llms/openai.py index 05e6566ffa..e5f229eb5c 100644 --- a/litellm/llms/openai.py +++ b/litellm/llms/openai.py @@ -1497,6 +1497,189 @@ class OpenAITextCompletion(BaseLLM): yield transformed_chunk +class OpenAIFilesAPI(BaseLLM): + """ + OpenAI methods to support for batches + - create_file() + - retrieve_file() + - list_files() + - delete_file() + - file_content() + - update_file() + """ + + def __init__(self) -> None: + super().__init__() + + def get_openai_client( + self, + api_key: Optional[str], + api_base: Optional[str], + timeout: Union[float, httpx.Timeout], + max_retries: Optional[int], + organization: Optional[str], + client: Optional[OpenAI] = None, + ) -> OpenAI: + received_args = locals() + if client is None: + data = {} + for k, v in received_args.items(): + if k == "self" or k == "client": + pass + elif k == "api_base" and v is not None: + data["base_url"] = v + elif v is not None: + data[k] = v + openai_client = OpenAI(**data) # type: ignore + else: + openai_client = client + + return openai_client + + def create_file( + self, + create_file_data: CreateFileRequest, + api_base: str, + api_key: Optional[str], + timeout: Union[float, httpx.Timeout], + max_retries: Optional[int], + organization: Optional[str], + client: Optional[OpenAI] = None, + ) -> FileObject: + openai_client: OpenAI = self.get_openai_client( + api_key=api_key, + api_base=api_base, + timeout=timeout, + max_retries=max_retries, + organization=organization, + client=client, + ) + response = openai_client.files.create(**create_file_data) + return response + + +class OpenAIBatchesAPI(BaseLLM): + """ + OpenAI methods to support for batches + - create_batch() + - retrieve_batch() + - cancel_batch() + - list_batch() + """ + + def __init__(self) -> None: + super().__init__() + + def get_openai_client( + self, + api_key: Optional[str], + api_base: Optional[str], + timeout: Union[float, httpx.Timeout], + max_retries: Optional[int], + organization: Optional[str], + client: Optional[OpenAI] = None, + ) -> OpenAI: + received_args = locals() + if client is None: + data = {} + for k, v in received_args.items(): + if k == "self" or k == "client": + pass + elif k == "api_base" and v is not None: + data["base_url"] = v + elif v is not None: + data[k] = v + openai_client = OpenAI(**data) # type: ignore + else: + openai_client = client + + return openai_client + + def create_batch( + self, + create_batch_data: CreateBatchRequest, + api_key: Optional[str], + api_base: Optional[str], + timeout: Union[float, httpx.Timeout], + max_retries: Optional[int], + organization: Optional[str], + client: Optional[OpenAI] = None, + ): + openai_client: OpenAI = self.get_openai_client( + api_key=api_key, + api_base=api_base, + timeout=timeout, + max_retries=max_retries, + organization=organization, + client=client, + ) + response = openai_client.batches.create(**create_batch_data) + return response + + def retrieve_batch( + self, + retrieve_batch_data: RetrieveBatchRequest, + api_key: Optional[str], + api_base: Optional[str], + timeout: Union[float, httpx.Timeout], + max_retries: Optional[int], + organization: Optional[str], + client: Optional[OpenAI] = None, + ): + openai_client: OpenAI = self.get_openai_client( + api_key=api_key, + api_base=api_base, + timeout=timeout, + max_retries=max_retries, + organization=organization, + client=client, + ) + response = openai_client.batches.retrieve(**retrieve_batch_data) + return response + + def cancel_batch( + self, + cancel_batch_data: CancelBatchRequest, + api_key: Optional[str], + api_base: Optional[str], + timeout: Union[float, httpx.Timeout], + max_retries: Optional[int], + organization: Optional[str], + client: Optional[OpenAI] = None, + ): + openai_client: OpenAI = self.get_openai_client( + api_key=api_key, + api_base=api_base, + timeout=timeout, + max_retries=max_retries, + organization=organization, + client=client, + ) + response = openai_client.batches.cancel(**cancel_batch_data) + return response + + # def list_batch( + # self, + # list_batch_data: ListBatchRequest, + # api_key: Optional[str], + # api_base: Optional[str], + # timeout: Union[float, httpx.Timeout], + # max_retries: Optional[int], + # organization: Optional[str], + # client: Optional[OpenAI] = None, + # ): + # openai_client: OpenAI = self.get_openai_client( + # api_key=api_key, + # api_base=api_base, + # timeout=timeout, + # max_retries=max_retries, + # organization=organization, + # client=client, + # ) + # response = openai_client.batches.list(**list_batch_data) + # return response + + class OpenAIAssistantsAPI(BaseLLM): def __init__(self) -> None: super().__init__() diff --git a/litellm/tests/test_openai_batches.py b/litellm/tests/test_openai_batches.py new file mode 100644 index 0000000000..ac282e0ed4 --- /dev/null +++ b/litellm/tests/test_openai_batches.py @@ -0,0 +1,58 @@ +# What is this? +## Unit Tests for OpenAI Batches API +import sys, os, json +import traceback +from dotenv import load_dotenv + +load_dotenv() +sys.path.insert( + 0, os.path.abspath("../..") +) # Adds the parent directory to the system path +import pytest, logging, asyncio +import litellm +from litellm import ( + create_batch, + create_file, +) + + +def test_create_batch(): + """ + 1. Create File for Batch completion + 2. Create Batch Request + """ + file_obj = litellm.create_file( + file=open("openai_batch_completions.jsonl", "rb"), + purpose="batch", + custom_llm_provider="openai", + ) + print("Response from creating file=", file_obj) + + batch_input_file_id = file_obj.id + assert ( + batch_input_file_id is not None + ), "Failed to create file, expected a non null file_id but got {batch_input_file_id}" + + print("response from creating file=", file_obj) + # response = create_batch( + # completion_window="24h", + # endpoint="/v1/chat/completions", + # input_file_id="1", + # custom_llm_provider="openai", + # metadata={"key1": "value1", "key2": "value2"}, + # ) + + print("response") + pass + + +def test_retrieve_batch(): + pass + + +def test_cancel_batch(): + pass + + +def test_list_batch(): + pass diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py index 60f7a90417..c65fb2d5f7 100644 --- a/litellm/types/llms/openai.py +++ b/litellm/types/llms/openai.py @@ -18,8 +18,23 @@ from openai.types.beta.assistant_tool_param import AssistantToolParam from openai.types.beta.threads.run import Run from openai.types.beta.assistant import Assistant from openai.pagination import SyncCursorPage +from os import PathLike +from openai.types import FileObject -from typing import TypedDict, List, Optional +from typing import TypedDict, List, Optional, Tuple, Mapping, IO + +FileContent = Union[IO[bytes], bytes, PathLike[str]] + +FileTypes = Union[ + # file (or bytes) + FileContent, + # (filename, file (or bytes)) + Tuple[Optional[str], FileContent], + # (filename, file (or bytes), content_type) + Tuple[Optional[str], FileContent, Optional[str]], + # (filename, file (or bytes), content_type, headers) + Tuple[Optional[str], FileContent, Optional[str], Mapping[str, str]], +] class NotGiven: @@ -148,8 +163,31 @@ class Thread(BaseModel): """The object type, which is always `thread`.""" +# OpenAI Files Types +class CreateFileRequest(TypedDict, total=False): + """ + CreateFileRequest + Used by Assistants API, Batches API, and Fine-Tunes API + + Required Params: + file: FileTypes + purpose: Literal['assistants', 'batch', 'fine-tune'] + + Optional Params: + extra_headers: Optional[Dict[str, str]] + extra_body: Optional[Dict[str, str]] = None + timeout: Optional[float] = None + """ + + file: FileTypes + purpose: Literal["assistants", "batch", "fine-tune"] + extra_headers: Optional[Dict[str, str]] + extra_body: Optional[Dict[str, str]] + timeout: Optional[float] + + # OpenAI Batches Types -class CreateBatchRequest(BaseModel): +class CreateBatchRequest(TypedDict, total=False): """ CreateBatchRequest """ @@ -157,42 +195,42 @@ class CreateBatchRequest(BaseModel): completion_window: Literal["24h"] endpoint: Literal["/v1/chat/completions", "/v1/embeddings", "/v1/completions"] input_file_id: str - metadata: Optional[Dict[str, str]] = None - extra_headers: Optional[Dict[str, str]] = None - extra_body: Optional[Dict[str, str]] = None - timeout: Optional[float] = None + metadata: Optional[Dict[str, str]] + extra_headers: Optional[Dict[str, str]] + extra_body: Optional[Dict[str, str]] + timeout: Optional[float] -class RetrieveBatchRequest(BaseModel): +class RetrieveBatchRequest(TypedDict, total=False): """ RetrieveBatchRequest """ batch_id: str - extra_headers: Optional[Dict[str, str]] = None - extra_body: Optional[Dict[str, str]] = None - timeout: Optional[float] = None + extra_headers: Optional[Dict[str, str]] + extra_body: Optional[Dict[str, str]] + timeout: Optional[float] -class CancelBatchRequest(BaseModel): +class CancelBatchRequest(TypedDict, total=False): """ CancelBatchRequest """ batch_id: str - extra_headers: Optional[Dict[str, str]] = None - extra_body: Optional[Dict[str, str]] = None - timeout: Optional[float] = None + extra_headers: Optional[Dict[str, str]] + extra_body: Optional[Dict[str, str]] + timeout: Optional[float] -class ListBatchRequest(BaseModel): +class ListBatchRequest(TypedDict, total=False): """ ListBatchRequest - List your organization's batches Calls https://api.openai.com/v1/batches """ - after: Optional[str] = None - limit: Optional[int] = 20 - extra_headers: Optional[Dict[str, str]] = None - extra_body: Optional[Dict[str, str]] = None - timeout: Optional[float] = None + after: Union[str, NotGiven] + limit: Union[int, NotGiven] + extra_headers: Optional[Dict[str, str]] + extra_body: Optional[Dict[str, str]] + timeout: Optional[float] From 38285e53c340418a1decfe5fd1d958885d23803d Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 15:45:23 -0700 Subject: [PATCH 04/20] working create_batch --- litellm/batches/main.py | 8 +++++++- litellm/llms/openai.py | 2 +- litellm/tests/test_openai_batches.py | 28 +++++++++++++++++++--------- litellm/types/llms/openai.py | 2 +- 4 files changed, 28 insertions(+), 12 deletions(-) diff --git a/litellm/batches/main.py b/litellm/batches/main.py index 6aa0d8d884..3bd1d07a47 100644 --- a/litellm/batches/main.py +++ b/litellm/batches/main.py @@ -26,6 +26,7 @@ from ..types.llms.openai import ( CreateFileRequest, FileTypes, FileObject, + Batch, ) from typing import Literal, Optional, Dict @@ -44,6 +45,11 @@ def create_file( extra_body: Optional[Dict[str, str]] = None, **kwargs, ) -> FileObject: + """ + Files are used to upload documents that can be used with features like Assistants, Fine-tuning, and Batch API. + + LiteLLM Equivalent of POST: POST https://api.openai.com/v1/files + """ try: optional_params = GenericLiteLLMParams(**kwargs) if custom_llm_provider == "openai": @@ -127,7 +133,7 @@ def create_batch( extra_headers: Optional[Dict[str, str]] = None, extra_body: Optional[Dict[str, str]] = None, **kwargs, -): +) -> Batch: """ Creates and executes a batch from an uploaded file of request diff --git a/litellm/llms/openai.py b/litellm/llms/openai.py index e5f229eb5c..5c5b837ea6 100644 --- a/litellm/llms/openai.py +++ b/litellm/llms/openai.py @@ -1604,7 +1604,7 @@ class OpenAIBatchesAPI(BaseLLM): max_retries: Optional[int], organization: Optional[str], client: Optional[OpenAI] = None, - ): + ) -> Batch: openai_client: OpenAI = self.get_openai_client( api_key=api_key, api_base=api_base, diff --git a/litellm/tests/test_openai_batches.py b/litellm/tests/test_openai_batches.py index ac282e0ed4..b99991baba 100644 --- a/litellm/tests/test_openai_batches.py +++ b/litellm/tests/test_openai_batches.py @@ -20,6 +20,7 @@ def test_create_batch(): """ 1. Create File for Batch completion 2. Create Batch Request + 3. Retrieve the specific batch """ file_obj = litellm.create_file( file=open("openai_batch_completions.jsonl", "rb"), @@ -33,16 +34,25 @@ def test_create_batch(): batch_input_file_id is not None ), "Failed to create file, expected a non null file_id but got {batch_input_file_id}" - print("response from creating file=", file_obj) - # response = create_batch( - # completion_window="24h", - # endpoint="/v1/chat/completions", - # input_file_id="1", - # custom_llm_provider="openai", - # metadata={"key1": "value1", "key2": "value2"}, - # ) + response = litellm.create_batch( + completion_window="24h", + endpoint="/v1/chat/completions", + input_file_id=batch_input_file_id, + custom_llm_provider="openai", + metadata={"key1": "value1", "key2": "value2"}, + ) - print("response") + print("response from litellm.create_batch=", response) + + assert ( + response.id is not None + ), f"Failed to create batch, expected a non null batch_id but got {response.id}" + assert ( + response.endpoint == "/v1/chat/completions" + ), f"Failed to create batch, expected endpoint to be /v1/chat/completions but got {response.endpoint}" + assert ( + response.input_file_id == batch_input_file_id + ), f"Failed to create batch, expected input_file_id to be {batch_input_file_id} but got {response.input_file_id}" pass diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py index c65fb2d5f7..81a50db6d2 100644 --- a/litellm/types/llms/openai.py +++ b/litellm/types/llms/openai.py @@ -19,7 +19,7 @@ from openai.types.beta.threads.run import Run from openai.types.beta.assistant import Assistant from openai.pagination import SyncCursorPage from os import PathLike -from openai.types import FileObject +from openai.types import FileObject, Batch from typing import TypedDict, List, Optional, Tuple, Mapping, IO From 938f4703c3da18580222058750d9010cdf769c3e Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 15:55:32 -0700 Subject: [PATCH 05/20] feat- add retrieve_batch() --- litellm/batches/main.py | 87 +++++++++++++++++++++++++++- litellm/tests/test_openai_batches.py | 28 ++++++--- 2 files changed, 105 insertions(+), 10 deletions(-) diff --git a/litellm/batches/main.py b/litellm/batches/main.py index 3bd1d07a47..3963a4e114 100644 --- a/litellm/batches/main.py +++ b/litellm/batches/main.py @@ -216,8 +216,91 @@ def create_batch( raise e -def retrieve_batch(): - pass +def retrieve_batch( + batch_id: str, + custom_llm_provider: Literal["openai"] = "openai", + metadata: Optional[Dict[str, str]] = None, + extra_headers: Optional[Dict[str, str]] = None, + extra_body: Optional[Dict[str, str]] = None, + **kwargs, +): + """ + Retrieves a batch. + + LiteLLM Equivalent of GET https://api.openai.com/v1/batches/{batch_id} + """ + try: + optional_params = GenericLiteLLMParams(**kwargs) + if custom_llm_provider == "openai": + + # for deepinfra/perplexity/anyscale/groq we check in get_llm_provider and pass in the api base from there + api_base = ( + optional_params.api_base + or litellm.api_base + or os.getenv("OPENAI_API_BASE") + or "https://api.openai.com/v1" + ) + organization = ( + optional_params.organization + or litellm.organization + or os.getenv("OPENAI_ORGANIZATION", None) + or None # default - https://github.com/openai/openai-python/blob/284c1799070c723c6a553337134148a7ab088dd8/openai/util.py#L105 + ) + # set API KEY + api_key = ( + optional_params.api_key + or litellm.api_key # for deepinfra/perplexity/anyscale we check in get_llm_provider and pass in the api key from there + or litellm.openai_key + or os.getenv("OPENAI_API_KEY") + ) + ### TIMEOUT LOGIC ### + timeout = ( + optional_params.timeout or kwargs.get("request_timeout", 600) or 600 + ) + # set timeout for 10 minutes by default + + if ( + timeout is not None + and isinstance(timeout, httpx.Timeout) + and supports_httpx_timeout(custom_llm_provider) == False + ): + read_timeout = timeout.read or 600 + timeout = read_timeout # default 10 min timeout + elif timeout is not None and not isinstance(timeout, httpx.Timeout): + timeout = float(timeout) # type: ignore + elif timeout is None: + timeout = 600.0 + + _retrieve_batch_request = RetrieveBatchRequest( + batch_id=batch_id, + extra_headers=extra_headers, + extra_body=extra_body, + ) + + response = openai_batches_instance.retrieve_batch( + retrieve_batch_data=_retrieve_batch_request, + api_base=api_base, + api_key=api_key, + organization=organization, + timeout=timeout, + max_retries=optional_params.max_retries, + ) + else: + raise litellm.exceptions.BadRequestError( + message="LiteLLM doesn't support {} for 'create_batch'. Only 'openai' is supported.".format( + custom_llm_provider + ), + model="n/a", + llm_provider=custom_llm_provider, + response=httpx.Response( + status_code=400, + content="Unsupported provider", + request=httpx.Request(method="create_thread", url="https://github.com/BerriAI/litellm"), # type: ignore + ), + ) + return response + except Exception as e: + raise e def cancel_batch(): diff --git a/litellm/tests/test_openai_batches.py b/litellm/tests/test_openai_batches.py index b99991baba..fc797635b0 100644 --- a/litellm/tests/test_openai_batches.py +++ b/litellm/tests/test_openai_batches.py @@ -14,6 +14,7 @@ from litellm import ( create_batch, create_file, ) +import time def test_create_batch(): @@ -34,7 +35,7 @@ def test_create_batch(): batch_input_file_id is not None ), "Failed to create file, expected a non null file_id but got {batch_input_file_id}" - response = litellm.create_batch( + create_batch_response = litellm.create_batch( completion_window="24h", endpoint="/v1/chat/completions", input_file_id=batch_input_file_id, @@ -42,17 +43,28 @@ def test_create_batch(): metadata={"key1": "value1", "key2": "value2"}, ) - print("response from litellm.create_batch=", response) + print("response from litellm.create_batch=", create_batch_response) assert ( - response.id is not None - ), f"Failed to create batch, expected a non null batch_id but got {response.id}" + create_batch_response.id is not None + ), f"Failed to create batch, expected a non null batch_id but got {create_batch_response.id}" assert ( - response.endpoint == "/v1/chat/completions" - ), f"Failed to create batch, expected endpoint to be /v1/chat/completions but got {response.endpoint}" + create_batch_response.endpoint == "/v1/chat/completions" + ), f"Failed to create batch, expected endpoint to be /v1/chat/completions but got {create_batch_response.endpoint}" assert ( - response.input_file_id == batch_input_file_id - ), f"Failed to create batch, expected input_file_id to be {batch_input_file_id} but got {response.input_file_id}" + create_batch_response.input_file_id == batch_input_file_id + ), f"Failed to create batch, expected input_file_id to be {batch_input_file_id} but got {create_batch_response.input_file_id}" + + time.sleep(30) + + retrieved_batch = litellm.retrieve_batch( + batch_id=create_batch_response.id, custom_llm_provider="openai" + ) + print("retrieved batch=", retrieved_batch) + # just assert that we retrieved a non None batch + + assert retrieved_batch.id == create_batch_response.id + pass From 758ed9e923cd5e794fc2a997c9358b97215f97dc Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 16:47:27 -0700 Subject: [PATCH 06/20] feat - add litellm.acreate_file --- litellm/batches/main.py | 55 +++++++++++++++++++++++++--- litellm/llms/openai.py | 44 ++++++++++++++++++---- litellm/tests/test_openai_batches.py | 54 +++++++++++++++++++++++++++ 3 files changed, 139 insertions(+), 14 deletions(-) diff --git a/litellm/batches/main.py b/litellm/batches/main.py index 3963a4e114..056318c8dd 100644 --- a/litellm/batches/main.py +++ b/litellm/batches/main.py @@ -10,11 +10,14 @@ https://platform.openai.com/docs/api-reference/batch """ -from typing import Iterable import os -import litellm -from openai import OpenAI +import asyncio +from functools import partial +import contextvars +from typing import Literal, Optional, Dict, Coroutine, Any, Union import httpx + +import litellm from litellm import client from litellm.utils import supports_httpx_timeout from ..types.router import * @@ -29,14 +32,51 @@ from ..types.llms.openai import ( Batch, ) -from typing import Literal, Optional, Dict - ####### ENVIRONMENT VARIABLES ################### openai_batches_instance = OpenAIBatchesAPI() openai_files_instance = OpenAIFilesAPI() ################################################# +async def acreate_file( + file: FileTypes, + purpose: Literal["assistants", "batch", "fine-tune"], + custom_llm_provider: Literal["openai"] = "openai", + extra_headers: Optional[Dict[str, str]] = None, + extra_body: Optional[Dict[str, str]] = None, + **kwargs, +) -> Coroutine[Any, Any, FileObject]: + """ + Files are used to upload documents that can be used with features like Assistants, Fine-tuning, and Batch API. + + LiteLLM Equivalent of POST: POST https://api.openai.com/v1/files + """ + loop = asyncio.get_event_loop() + kwargs["acreate_file"] = True + + # Use a partial function to pass your keyword arguments + func = partial( + create_file, + file, + purpose, + custom_llm_provider, + extra_headers, + extra_body, + **kwargs, + ) + + # Add the context to the function + ctx = contextvars.copy_context() + func_with_context = partial(ctx.run, func) + init_response = await loop.run_in_executor(None, func_with_context) + if asyncio.iscoroutine(init_response): + response = await init_response + else: + response = init_response # type: ignore + + return response + + def create_file( file: FileTypes, purpose: Literal["assistants", "batch", "fine-tune"], @@ -44,7 +84,7 @@ def create_file( extra_headers: Optional[Dict[str, str]] = None, extra_body: Optional[Dict[str, str]] = None, **kwargs, -) -> FileObject: +) -> Union[FileObject | Coroutine[Any, Any, FileObject]]: """ Files are used to upload documents that can be used with features like Assistants, Fine-tuning, and Batch API. @@ -98,7 +138,10 @@ def create_file( extra_body=extra_body, ) + _is_async = kwargs.pop("acreate_file", False) is True + response = openai_files_instance.create_file( + _is_async=_is_async, api_base=api_base, api_key=api_key, timeout=timeout, diff --git a/litellm/llms/openai.py b/litellm/llms/openai.py index 5c5b837ea6..05fc5784b6 100644 --- a/litellm/llms/openai.py +++ b/litellm/llms/openai.py @@ -21,7 +21,7 @@ from litellm.utils import ( TranscriptionResponse, TextCompletionResponse, ) -from typing import Callable, Optional +from typing import Callable, Optional, Coroutine import litellm from .prompt_templates.factory import prompt_factory, custom_prompt from openai import OpenAI, AsyncOpenAI @@ -1518,42 +1518,70 @@ class OpenAIFilesAPI(BaseLLM): timeout: Union[float, httpx.Timeout], max_retries: Optional[int], organization: Optional[str], - client: Optional[OpenAI] = None, - ) -> OpenAI: + client: Optional[Union[OpenAI, AsyncOpenAI]] = None, + _is_async: bool = False, + ) -> Optional[Union[OpenAI, AsyncOpenAI]]: received_args = locals() + openai_client: Optional[Union[OpenAI, AsyncOpenAI]] = None if client is None: data = {} for k, v in received_args.items(): - if k == "self" or k == "client": + if k == "self" or k == "client" or k == "_is_async": pass elif k == "api_base" and v is not None: data["base_url"] = v elif v is not None: data[k] = v - openai_client = OpenAI(**data) # type: ignore + if _is_async is True: + openai_client = AsyncOpenAI(**data) + else: + openai_client = OpenAI(**data) # type: ignore else: openai_client = client return openai_client + async def acreate_file( + self, + create_file_data: CreateFileRequest, + openai_client: AsyncOpenAI, + ) -> FileObject: + response = await openai_client.files.create(**create_file_data) + return response + def create_file( self, + _is_async: bool, create_file_data: CreateFileRequest, api_base: str, api_key: Optional[str], timeout: Union[float, httpx.Timeout], max_retries: Optional[int], organization: Optional[str], - client: Optional[OpenAI] = None, - ) -> FileObject: - openai_client: OpenAI = self.get_openai_client( + client: Optional[Union[OpenAI, AsyncOpenAI]] = None, + ) -> Union[FileObject, Coroutine[Any, Any, FileObject]]: + openai_client: Optional[Union[OpenAI, AsyncOpenAI]] = self.get_openai_client( api_key=api_key, api_base=api_base, timeout=timeout, max_retries=max_retries, organization=organization, client=client, + _is_async=_is_async, ) + if openai_client is None: + raise ValueError( + "OpenAI client is not initialized. Make sure api_key is passed or OPENAI_API_KEY is set in the environment." + ) + + if _is_async is True: + if not isinstance(openai_client, AsyncOpenAI): + raise ValueError( + "OpenAI client is not an instance of AsyncOpenAI. Make sure you passed an AsyncOpenAI client." + ) + return self.acreate_file( # type: ignore + create_file_data=create_file_data, openai_client=openai_client + ) response = openai_client.files.create(**create_file_data) return response diff --git a/litellm/tests/test_openai_batches.py b/litellm/tests/test_openai_batches.py index fc797635b0..2de417619b 100644 --- a/litellm/tests/test_openai_batches.py +++ b/litellm/tests/test_openai_batches.py @@ -2,6 +2,7 @@ ## Unit Tests for OpenAI Batches API import sys, os, json import traceback +import asyncio from dotenv import load_dotenv load_dotenv() @@ -68,6 +69,59 @@ def test_create_batch(): pass +@pytest.mark.asyncio() +async def test_async_create_batch(): + """ + 1. Create File for Batch completion + 2. Create Batch Request + 3. Retrieve the specific batch + """ + print("Testing async create batch") + file_obj = await litellm.acreate_file( + file=open("openai_batch_completions.jsonl", "rb"), + purpose="batch", + custom_llm_provider="openai", + ) + print("Response from creating file=", file_obj) + + batch_input_file_id = file_obj.id + assert ( + batch_input_file_id is not None + ), "Failed to create file, expected a non null file_id but got {batch_input_file_id}" + + # create_batch_response = litellm.create_batch( + # completion_window="24h", + # endpoint="/v1/chat/completions", + # input_file_id=batch_input_file_id, + # custom_llm_provider="openai", + # metadata={"key1": "value1", "key2": "value2"}, + # ) + + # print("response from litellm.create_batch=", create_batch_response) + + # assert ( + # create_batch_response.id is not None + # ), f"Failed to create batch, expected a non null batch_id but got {create_batch_response.id}" + # assert ( + # create_batch_response.endpoint == "/v1/chat/completions" + # ), f"Failed to create batch, expected endpoint to be /v1/chat/completions but got {create_batch_response.endpoint}" + # assert ( + # create_batch_response.input_file_id == batch_input_file_id + # ), f"Failed to create batch, expected input_file_id to be {batch_input_file_id} but got {create_batch_response.input_file_id}" + + # time.sleep(30) + + # retrieved_batch = litellm.retrieve_batch( + # batch_id=create_batch_response.id, custom_llm_provider="openai" + # ) + # print("retrieved batch=", retrieved_batch) + # # just assert that we retrieved a non None batch + + # assert retrieved_batch.id == create_batch_response.id + + pass + + def test_retrieve_batch(): pass From 1ef7cd923cf11ceb85ad8da3019ae5975285e331 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 17:03:29 -0700 Subject: [PATCH 07/20] feat - add acreate_batch --- litellm/batches/main.py | 104 ++++++++++++++++++++------- litellm/llms/openai.py | 58 ++++++++++++--- litellm/tests/test_openai_batches.py | 34 ++++----- 3 files changed, 143 insertions(+), 53 deletions(-) diff --git a/litellm/batches/main.py b/litellm/batches/main.py index 056318c8dd..05a6dfd517 100644 --- a/litellm/batches/main.py +++ b/litellm/batches/main.py @@ -51,30 +51,33 @@ async def acreate_file( LiteLLM Equivalent of POST: POST https://api.openai.com/v1/files """ - loop = asyncio.get_event_loop() - kwargs["acreate_file"] = True + try: + loop = asyncio.get_event_loop() + kwargs["acreate_file"] = True - # Use a partial function to pass your keyword arguments - func = partial( - create_file, - file, - purpose, - custom_llm_provider, - extra_headers, - extra_body, - **kwargs, - ) + # Use a partial function to pass your keyword arguments + func = partial( + create_file, + file, + purpose, + custom_llm_provider, + extra_headers, + extra_body, + **kwargs, + ) - # Add the context to the function - ctx = contextvars.copy_context() - func_with_context = partial(ctx.run, func) - init_response = await loop.run_in_executor(None, func_with_context) - if asyncio.iscoroutine(init_response): - response = await init_response - else: - response = init_response # type: ignore + # Add the context to the function + ctx = contextvars.copy_context() + func_with_context = partial(ctx.run, func) + init_response = await loop.run_in_executor(None, func_with_context) + if asyncio.iscoroutine(init_response): + response = await init_response + else: + response = init_response # type: ignore - return response + return response + except Exception as e: + raise e def create_file( @@ -167,6 +170,52 @@ def create_file( raise e +async def acreate_batch( + completion_window: Literal["24h"], + endpoint: Literal["/v1/chat/completions", "/v1/embeddings", "/v1/completions"], + input_file_id: str, + custom_llm_provider: Literal["openai"] = "openai", + metadata: Optional[Dict[str, str]] = None, + extra_headers: Optional[Dict[str, str]] = None, + extra_body: Optional[Dict[str, str]] = None, + **kwargs, +) -> Coroutine[Any, Any, Batch]: + """ + Creates and executes a batch from an uploaded file of request + + LiteLLM Equivalent of POST: https://api.openai.com/v1/batches + """ + try: + loop = asyncio.get_event_loop() + kwargs["acreate_batch"] = True + + # Use a partial function to pass your keyword arguments + func = partial( + create_batch, + completion_window, + endpoint, + input_file_id, + custom_llm_provider, + metadata, + extra_headers, + extra_body, + **kwargs, + ) + + # Add the context to the function + ctx = contextvars.copy_context() + func_with_context = partial(ctx.run, func) + init_response = await loop.run_in_executor(None, func_with_context) + if asyncio.iscoroutine(init_response): + response = await init_response + else: + response = init_response # type: ignore + + return response + except Exception as e: + raise e + + def create_batch( completion_window: Literal["24h"], endpoint: Literal["/v1/chat/completions", "/v1/embeddings", "/v1/completions"], @@ -176,7 +225,7 @@ def create_batch( extra_headers: Optional[Dict[str, str]] = None, extra_body: Optional[Dict[str, str]] = None, **kwargs, -) -> Batch: +) -> Union[Batch, Coroutine[Any, Any, Batch]]: """ Creates and executes a batch from an uploaded file of request @@ -224,6 +273,8 @@ def create_batch( elif timeout is None: timeout = 600.0 + _is_async = kwargs.pop("acreate_batch", False) is True + _create_batch_request = CreateBatchRequest( completion_window=completion_window, endpoint=endpoint, @@ -240,6 +291,7 @@ def create_batch( create_batch_data=_create_batch_request, timeout=timeout, max_retries=optional_params.max_retries, + _is_async=_is_async, ) else: raise litellm.exceptions.BadRequestError( @@ -320,7 +372,10 @@ def retrieve_batch( extra_body=extra_body, ) + _is_async = kwargs.pop("aretrieve_batch", False) is True + response = openai_batches_instance.retrieve_batch( + _is_async=_is_async, retrieve_batch_data=_retrieve_batch_request, api_base=api_base, api_key=api_key, @@ -354,11 +409,6 @@ def list_batch(): pass -# Async Functions -async def acreate_batch(): - pass - - async def aretrieve_batch(): pass diff --git a/litellm/llms/openai.py b/litellm/llms/openai.py index 05fc5784b6..fa1f13c70a 100644 --- a/litellm/llms/openai.py +++ b/litellm/llms/openai.py @@ -1605,47 +1605,76 @@ class OpenAIBatchesAPI(BaseLLM): timeout: Union[float, httpx.Timeout], max_retries: Optional[int], organization: Optional[str], - client: Optional[OpenAI] = None, - ) -> OpenAI: + client: Optional[Union[OpenAI, AsyncOpenAI]] = None, + _is_async: bool = False, + ) -> Optional[Union[OpenAI, AsyncOpenAI]]: received_args = locals() + openai_client: Optional[Union[OpenAI, AsyncOpenAI]] = None if client is None: data = {} for k, v in received_args.items(): - if k == "self" or k == "client": + if k == "self" or k == "client" or k == "_is_async": pass elif k == "api_base" and v is not None: data["base_url"] = v elif v is not None: data[k] = v - openai_client = OpenAI(**data) # type: ignore + if _is_async is True: + openai_client = AsyncOpenAI(**data) + else: + openai_client = OpenAI(**data) # type: ignore else: openai_client = client return openai_client + async def acreate_batch( + self, + create_batch_data: CreateBatchRequest, + openai_client: AsyncOpenAI, + ) -> Batch: + response = await openai_client.batches.create(**create_batch_data) + return response + def create_batch( self, + _is_async: bool, create_batch_data: CreateBatchRequest, api_key: Optional[str], api_base: Optional[str], timeout: Union[float, httpx.Timeout], max_retries: Optional[int], organization: Optional[str], - client: Optional[OpenAI] = None, - ) -> Batch: - openai_client: OpenAI = self.get_openai_client( + client: Optional[Union[OpenAI, AsyncOpenAI]] = None, + ) -> Union[Batch, Coroutine[Any, Any, Batch]]: + openai_client: Optional[Union[OpenAI, AsyncOpenAI]] = self.get_openai_client( api_key=api_key, api_base=api_base, timeout=timeout, max_retries=max_retries, organization=organization, client=client, + _is_async=_is_async, ) + if openai_client is None: + raise ValueError( + "OpenAI client is not initialized. Make sure api_key is passed or OPENAI_API_KEY is set in the environment." + ) + + if _is_async is True: + if not isinstance(openai_client, AsyncOpenAI): + raise ValueError( + "OpenAI client is not an instance of AsyncOpenAI. Make sure you passed an AsyncOpenAI client." + ) + return self.acreate_batch( # type: ignore + create_batch_data=create_batch_data, openai_client=openai_client + ) response = openai_client.batches.create(**create_batch_data) return response def retrieve_batch( self, + _is_async: bool, retrieve_batch_data: RetrieveBatchRequest, api_key: Optional[str], api_base: Optional[str], @@ -1654,19 +1683,25 @@ class OpenAIBatchesAPI(BaseLLM): organization: Optional[str], client: Optional[OpenAI] = None, ): - openai_client: OpenAI = self.get_openai_client( + openai_client: Optional[Union[OpenAI, AsyncOpenAI]] = self.get_openai_client( api_key=api_key, api_base=api_base, timeout=timeout, max_retries=max_retries, organization=organization, client=client, + _is_async=_is_async, ) + if openai_client is None: + raise ValueError( + "OpenAI client is not initialized. Make sure api_key is passed or OPENAI_API_KEY is set in the environment." + ) response = openai_client.batches.retrieve(**retrieve_batch_data) return response def cancel_batch( self, + _is_async: bool, cancel_batch_data: CancelBatchRequest, api_key: Optional[str], api_base: Optional[str], @@ -1675,14 +1710,19 @@ class OpenAIBatchesAPI(BaseLLM): organization: Optional[str], client: Optional[OpenAI] = None, ): - openai_client: OpenAI = self.get_openai_client( + openai_client: Optional[Union[OpenAI, AsyncOpenAI]] = self.get_openai_client( api_key=api_key, api_base=api_base, timeout=timeout, max_retries=max_retries, organization=organization, client=client, + _is_async=_is_async, ) + if openai_client is None: + raise ValueError( + "OpenAI client is not initialized. Make sure api_key is passed or OPENAI_API_KEY is set in the environment." + ) response = openai_client.batches.cancel(**cancel_batch_data) return response diff --git a/litellm/tests/test_openai_batches.py b/litellm/tests/test_openai_batches.py index 2de417619b..497662006d 100644 --- a/litellm/tests/test_openai_batches.py +++ b/litellm/tests/test_openai_batches.py @@ -89,25 +89,25 @@ async def test_async_create_batch(): batch_input_file_id is not None ), "Failed to create file, expected a non null file_id but got {batch_input_file_id}" - # create_batch_response = litellm.create_batch( - # completion_window="24h", - # endpoint="/v1/chat/completions", - # input_file_id=batch_input_file_id, - # custom_llm_provider="openai", - # metadata={"key1": "value1", "key2": "value2"}, - # ) + create_batch_response = await litellm.acreate_batch( + completion_window="24h", + endpoint="/v1/chat/completions", + input_file_id=batch_input_file_id, + custom_llm_provider="openai", + metadata={"key1": "value1", "key2": "value2"}, + ) - # print("response from litellm.create_batch=", create_batch_response) + print("response from litellm.create_batch=", create_batch_response) - # assert ( - # create_batch_response.id is not None - # ), f"Failed to create batch, expected a non null batch_id but got {create_batch_response.id}" - # assert ( - # create_batch_response.endpoint == "/v1/chat/completions" - # ), f"Failed to create batch, expected endpoint to be /v1/chat/completions but got {create_batch_response.endpoint}" - # assert ( - # create_batch_response.input_file_id == batch_input_file_id - # ), f"Failed to create batch, expected input_file_id to be {batch_input_file_id} but got {create_batch_response.input_file_id}" + assert ( + create_batch_response.id is not None + ), f"Failed to create batch, expected a non null batch_id but got {create_batch_response.id}" + assert ( + create_batch_response.endpoint == "/v1/chat/completions" + ), f"Failed to create batch, expected endpoint to be /v1/chat/completions but got {create_batch_response.endpoint}" + assert ( + create_batch_response.input_file_id == batch_input_file_id + ), f"Failed to create batch, expected input_file_id to be {batch_input_file_id} but got {create_batch_response.input_file_id}" # time.sleep(30) From 6688215c186256f975c4dccbdb179c237f5b1b9e Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 17:12:41 -0700 Subject: [PATCH 08/20] feat - add aretrieve_batch --- litellm/batches/main.py | 52 ++++++++++++++++++++++++---- litellm/llms/openai.py | 17 +++++++++ litellm/tests/test_openai_batches.py | 16 ++++----- 3 files changed, 69 insertions(+), 16 deletions(-) diff --git a/litellm/batches/main.py b/litellm/batches/main.py index 05a6dfd517..119043e4c1 100644 --- a/litellm/batches/main.py +++ b/litellm/batches/main.py @@ -47,7 +47,7 @@ async def acreate_file( **kwargs, ) -> Coroutine[Any, Any, FileObject]: """ - Files are used to upload documents that can be used with features like Assistants, Fine-tuning, and Batch API. + Async: Files are used to upload documents that can be used with features like Assistants, Fine-tuning, and Batch API. LiteLLM Equivalent of POST: POST https://api.openai.com/v1/files """ @@ -181,7 +181,7 @@ async def acreate_batch( **kwargs, ) -> Coroutine[Any, Any, Batch]: """ - Creates and executes a batch from an uploaded file of request + Async: Creates and executes a batch from an uploaded file of request LiteLLM Equivalent of POST: https://api.openai.com/v1/batches """ @@ -311,6 +311,48 @@ def create_batch( raise e +async def aretrieve_batch( + batch_id: str, + custom_llm_provider: Literal["openai"] = "openai", + metadata: Optional[Dict[str, str]] = None, + extra_headers: Optional[Dict[str, str]] = None, + extra_body: Optional[Dict[str, str]] = None, + **kwargs, +) -> Coroutine[Any, Any, Batch]: + """ + Async: Retrieves a batch. + + LiteLLM Equivalent of GET https://api.openai.com/v1/batches/{batch_id} + """ + try: + loop = asyncio.get_event_loop() + kwargs["aretrieve_batch"] = True + + # Use a partial function to pass your keyword arguments + func = partial( + retrieve_batch, + batch_id, + custom_llm_provider, + metadata, + extra_headers, + extra_body, + **kwargs, + ) + + # Add the context to the function + ctx = contextvars.copy_context() + func_with_context = partial(ctx.run, func) + init_response = await loop.run_in_executor(None, func_with_context) + if asyncio.iscoroutine(init_response): + response = await init_response + else: + response = init_response # type: ignore + + return response + except Exception as e: + raise e + + def retrieve_batch( batch_id: str, custom_llm_provider: Literal["openai"] = "openai", @@ -318,7 +360,7 @@ def retrieve_batch( extra_headers: Optional[Dict[str, str]] = None, extra_body: Optional[Dict[str, str]] = None, **kwargs, -): +) -> Union[Batch, Coroutine[Any, Any, Batch]]: """ Retrieves a batch. @@ -409,10 +451,6 @@ def list_batch(): pass -async def aretrieve_batch(): - pass - - async def acancel_batch(): pass diff --git a/litellm/llms/openai.py b/litellm/llms/openai.py index fa1f13c70a..43d088f0db 100644 --- a/litellm/llms/openai.py +++ b/litellm/llms/openai.py @@ -1672,6 +1672,14 @@ class OpenAIBatchesAPI(BaseLLM): response = openai_client.batches.create(**create_batch_data) return response + async def aretrieve_batch( + self, + retrieve_batch_data: RetrieveBatchRequest, + openai_client: AsyncOpenAI, + ) -> Batch: + response = await openai_client.batches.retrieve(**retrieve_batch_data) + return response + def retrieve_batch( self, _is_async: bool, @@ -1696,6 +1704,15 @@ class OpenAIBatchesAPI(BaseLLM): raise ValueError( "OpenAI client is not initialized. Make sure api_key is passed or OPENAI_API_KEY is set in the environment." ) + + if _is_async is True: + if not isinstance(openai_client, AsyncOpenAI): + raise ValueError( + "OpenAI client is not an instance of AsyncOpenAI. Make sure you passed an AsyncOpenAI client." + ) + return self.aretrieve_batch( # type: ignore + retrieve_batch_data=retrieve_batch_data, openai_client=openai_client + ) response = openai_client.batches.retrieve(**retrieve_batch_data) return response diff --git a/litellm/tests/test_openai_batches.py b/litellm/tests/test_openai_batches.py index 497662006d..2bf0090128 100644 --- a/litellm/tests/test_openai_batches.py +++ b/litellm/tests/test_openai_batches.py @@ -109,17 +109,15 @@ async def test_async_create_batch(): create_batch_response.input_file_id == batch_input_file_id ), f"Failed to create batch, expected input_file_id to be {batch_input_file_id} but got {create_batch_response.input_file_id}" - # time.sleep(30) + await asyncio.sleep(1) - # retrieved_batch = litellm.retrieve_batch( - # batch_id=create_batch_response.id, custom_llm_provider="openai" - # ) - # print("retrieved batch=", retrieved_batch) - # # just assert that we retrieved a non None batch + retrieved_batch = await litellm.aretrieve_batch( + batch_id=create_batch_response.id, custom_llm_provider="openai" + ) + print("retrieved batch=", retrieved_batch) + # just assert that we retrieved a non None batch - # assert retrieved_batch.id == create_batch_response.id - - pass + assert retrieved_batch.id == create_batch_response.id def test_retrieve_batch(): From c6eb004ed1bb0ce73d323c4b12311df5f7d6ea7b Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 17:21:59 -0700 Subject: [PATCH 09/20] fix python 3.8 error --- litellm/types/llms/openai.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py index 81a50db6d2..50ac1335ec 100644 --- a/litellm/types/llms/openai.py +++ b/litellm/types/llms/openai.py @@ -23,7 +23,7 @@ from openai.types import FileObject, Batch from typing import TypedDict, List, Optional, Tuple, Mapping, IO -FileContent = Union[IO[bytes], bytes, PathLike[str]] +FileContent = Union[IO[bytes], bytes, PathLike] FileTypes = Union[ # file (or bytes) From 215f19440c4d08f6b33a4b951aba6c1a8c882866 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 17:25:08 -0700 Subject: [PATCH 10/20] fix python3.8 error --- litellm/batches/main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/litellm/batches/main.py b/litellm/batches/main.py index 119043e4c1..917ad45e10 100644 --- a/litellm/batches/main.py +++ b/litellm/batches/main.py @@ -87,7 +87,7 @@ def create_file( extra_headers: Optional[Dict[str, str]] = None, extra_body: Optional[Dict[str, str]] = None, **kwargs, -) -> Union[FileObject | Coroutine[Any, Any, FileObject]]: +) -> Union[FileObject, Coroutine[Any, Any, FileObject]]: """ Files are used to upload documents that can be used with features like Assistants, Fine-tuning, and Batch API. From 5e9d024c2cf8269c4e70bc44226230f462f1b502 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 17:29:13 -0700 Subject: [PATCH 11/20] fix - test openai batches --- litellm/tests/test_openai_batches.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/litellm/tests/test_openai_batches.py b/litellm/tests/test_openai_batches.py index 2bf0090128..fc29331a5a 100644 --- a/litellm/tests/test_openai_batches.py +++ b/litellm/tests/test_openai_batches.py @@ -24,8 +24,12 @@ def test_create_batch(): 2. Create Batch Request 3. Retrieve the specific batch """ + file_name = "openai_batch_completions.jsonl" + _current_dir = os.path.dirname(os.path.abspath(__file__)) + file_path = os.path.join(_current_dir, file_name) + file_obj = litellm.create_file( - file=open("openai_batch_completions.jsonl", "rb"), + file=open(file_path, "rb"), purpose="batch", custom_llm_provider="openai", ) @@ -77,8 +81,12 @@ async def test_async_create_batch(): 3. Retrieve the specific batch """ print("Testing async create batch") + + file_name = "openai_batch_completions.jsonl" + _current_dir = os.path.dirname(os.path.abspath(__file__)) + file_path = os.path.join(_current_dir, file_name) file_obj = await litellm.acreate_file( - file=open("openai_batch_completions.jsonl", "rb"), + file=open(file_path, "rb"), purpose="batch", custom_llm_provider="openai", ) From 37d350b46675051e29eba8fc61c825493af16f19 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 17:39:17 -0700 Subject: [PATCH 12/20] add batches, files to routes --- litellm/proxy/_types.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index 1b97c68366..07812a756d 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -99,6 +99,14 @@ class LiteLLMRoutes(enum.Enum): # moderations "/moderations", "/v1/moderations", + # batches + "/v1/batches", + "/batches", + "/v1/batches{batch_id}", + "/batches{batch_id}", + # files + "/v1/files", + "/files", # models "/models", "/v1/models", @@ -1215,6 +1223,7 @@ class InvitationModel(LiteLLMBase): updated_at: datetime updated_by: str + class ConfigFieldInfo(LiteLLMBase): field_name: str field_value: Any From c2e24b4ed85ab8638538b92faa0dc6b4a4ac0665 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 17:49:36 -0700 Subject: [PATCH 13/20] feat add v1/batches --- litellm/proxy/proxy_server.py | 161 ++++++++++++++++++++++++++++++++++ 1 file changed, 161 insertions(+) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 6efcb2a702..f74e53f947 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -100,6 +100,13 @@ from litellm.proxy.utils import ( encrypt_value, decrypt_value, ) +from litellm import ( + CreateBatchRequest, + RetrieveBatchRequest, + ListBatchRequest, + CancelBatchRequest, + CreateFileRequest, +) from litellm.proxy.secret_managers.google_kms import load_google_kms from litellm.proxy.secret_managers.aws_secret_manager import load_aws_secret_manager import pydantic @@ -5025,6 +5032,160 @@ async def audio_transcriptions( ) +###################################################################### + +# /v1/batches Endpoints + + +###################################################################### +@router.post( + "/v1/batches", + dependencies=[Depends(user_api_key_auth)], + tags=["Batch"], +) +@router.post( + "/batches", + dependencies=[Depends(user_api_key_auth)], + tags=["Batch"], +) +async def create_batch( + request: Request, + fastapi_response: Response, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """ + Create large batches of API requests for asynchronous processing. + This is the equivalent of POST https://api.openai.com/v1/batch + Supports Identical Params as: https://platform.openai.com/docs/api-reference/batch + + Example Curl + ``` + curl http://localhost:4000/v1/batches \ + -H "Authorization: Bearer sk-1234" \ + -H "Content-Type: application/json" \ + -d '{ + "input_file_id": "file-abc123", + "endpoint": "/v1/chat/completions", + "completion_window": "24h" + }' + ``` + """ + global proxy_logging_obj + data: Dict = {} + try: + # Use orjson to parse JSON data, orjson speeds up requests significantly + form_data = await request.form() + data = {key: value for key, value in form_data.items() if key != "file"} + + # Include original request and headers in the data + data["proxy_server_request"] = { # type: ignore + "url": str(request.url), + "method": request.method, + "headers": dict(request.headers), + "body": copy.copy(data), # use copy instead of deepcopy + } + + if data.get("user", None) is None and user_api_key_dict.user_id is not None: + data["user"] = user_api_key_dict.user_id + + if "metadata" not in data: + data["metadata"] = {} + data["metadata"]["user_api_key"] = user_api_key_dict.api_key + data["metadata"]["user_api_key_metadata"] = user_api_key_dict.metadata + _headers = dict(request.headers) + _headers.pop( + "authorization", None + ) # do not store the original `sk-..` api key in the db + data["metadata"]["headers"] = _headers + data["metadata"]["user_api_key_alias"] = getattr( + user_api_key_dict, "key_alias", None + ) + data["metadata"]["user_api_key_user_id"] = user_api_key_dict.user_id + data["metadata"]["user_api_key_team_id"] = getattr( + user_api_key_dict, "team_id", None + ) + data["metadata"]["global_max_parallel_requests"] = general_settings.get( + "global_max_parallel_requests", None + ) + data["metadata"]["user_api_key_team_alias"] = getattr( + user_api_key_dict, "team_alias", None + ) + data["metadata"]["endpoint"] = str(request.url) + + ### TEAM-SPECIFIC PARAMS ### + if user_api_key_dict.team_id is not None: + team_config = await proxy_config.load_team_config( + team_id=user_api_key_dict.team_id + ) + if len(team_config) == 0: + pass + else: + team_id = team_config.pop("team_id", None) + data["metadata"]["team_id"] = team_id + data = { + **team_config, + **data, + } # add the team-specific configs to the completion call + + _create_batch_data = CreateBatchRequest(**data) + + # for now use custom_llm_provider=="openai" -> this will change as LiteLLM adds more providers for acreate_batch + response = await litellm.acreate_batch( + custom_llm_provider="openai", **_create_batch_data + ) + + ### ALERTING ### + data["litellm_status"] = "success" # used for alerting + + ### RESPONSE HEADERS ### + hidden_params = getattr(response, "_hidden_params", {}) or {} + model_id = hidden_params.get("model_id", None) or "" + cache_key = hidden_params.get("cache_key", None) or "" + api_base = hidden_params.get("api_base", None) or "" + + fastapi_response.headers.update( + get_custom_headers( + user_api_key_dict=user_api_key_dict, + model_id=model_id, + cache_key=cache_key, + api_base=api_base, + version=version, + model_region=getattr(user_api_key_dict, "allowed_model_region", ""), + ) + ) + + return response + except Exception as e: + data["litellm_status"] = "fail" # used for alerting + await proxy_logging_obj.post_call_failure_hook( + user_api_key_dict=user_api_key_dict, original_exception=e, request_data=data + ) + traceback.print_exc() + if isinstance(e, HTTPException): + raise ProxyException( + message=getattr(e, "message", str(e.detail)), + type=getattr(e, "type", "None"), + param=getattr(e, "param", "None"), + code=getattr(e, "status_code", status.HTTP_400_BAD_REQUEST), + ) + else: + error_traceback = traceback.format_exc() + error_msg = f"{str(e)}" + raise ProxyException( + message=getattr(e, "message", error_msg), + type=getattr(e, "type", "None"), + param=getattr(e, "param", "None"), + code=getattr(e, "status_code", 500), + ) + + +###################################################################### + +# END OF /v1/batches Endpoints Implementation + +###################################################################### + + @router.post( "/v1/moderations", dependencies=[Depends(user_api_key_auth)], From 0020672c1937f6f93db01aafee9f3a0eec912223 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 17:58:34 -0700 Subject: [PATCH 14/20] fear support GET /v1/batches{batch_id} --- litellm/proxy/proxy_server.py | 143 ++++++++++++++++++++++++++++++++++ 1 file changed, 143 insertions(+) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index f74e53f947..eaf53f0bd7 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -149,6 +149,7 @@ from fastapi import ( Request, HTTPException, status, + Path, Depends, Header, Response, @@ -5179,6 +5180,148 @@ async def create_batch( ) +@router.get( + "/v1/batches{batch_id}", + dependencies=[Depends(user_api_key_auth)], + tags=["Batch"], +) +@router.get( + "/batches{batch_id}", + dependencies=[Depends(user_api_key_auth)], + tags=["Batch"], +) +async def retrieve_batch( + request: Request, + fastapi_response: Response, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), + batch_id: str = Path( + title="Batch ID to retrieve", description="The ID of the batch to retrieve" + ), +): + """ + Retrieves a batch. + This is the equivalent of GET https://api.openai.com/v1/batches/{batch_id} + Supports Identical Params as: https://platform.openai.com/docs/api-reference/batch/retrieve + + Example Curl + ``` + curl http://localhost:4000/v1/batches/batch_abc123 \ + -H "Authorization: Bearer sk-1234" \ + -H "Content-Type: application/json" \ + + ``` + """ + global proxy_logging_obj + data: Dict = {} + try: + # Use orjson to parse JSON data, orjson speeds up requests significantly + form_data = await request.form() + data = {key: value for key, value in form_data.items() if key != "file"} + + # Include original request and headers in the data + data["proxy_server_request"] = { # type: ignore + "url": str(request.url), + "method": request.method, + "headers": dict(request.headers), + "body": copy.copy(data), # use copy instead of deepcopy + } + + if data.get("user", None) is None and user_api_key_dict.user_id is not None: + data["user"] = user_api_key_dict.user_id + + if "metadata" not in data: + data["metadata"] = {} + data["metadata"]["user_api_key"] = user_api_key_dict.api_key + data["metadata"]["user_api_key_metadata"] = user_api_key_dict.metadata + _headers = dict(request.headers) + _headers.pop( + "authorization", None + ) # do not store the original `sk-..` api key in the db + data["metadata"]["headers"] = _headers + data["metadata"]["user_api_key_alias"] = getattr( + user_api_key_dict, "key_alias", None + ) + data["metadata"]["user_api_key_user_id"] = user_api_key_dict.user_id + data["metadata"]["user_api_key_team_id"] = getattr( + user_api_key_dict, "team_id", None + ) + data["metadata"]["global_max_parallel_requests"] = general_settings.get( + "global_max_parallel_requests", None + ) + data["metadata"]["user_api_key_team_alias"] = getattr( + user_api_key_dict, "team_alias", None + ) + data["metadata"]["endpoint"] = str(request.url) + + ### TEAM-SPECIFIC PARAMS ### + if user_api_key_dict.team_id is not None: + team_config = await proxy_config.load_team_config( + team_id=user_api_key_dict.team_id + ) + if len(team_config) == 0: + pass + else: + team_id = team_config.pop("team_id", None) + data["metadata"]["team_id"] = team_id + data = { + **team_config, + **data, + } # add the team-specific configs to the completion call + + _retrieve_batch_request = RetrieveBatchRequest( + batch_id=batch_id, + ) + + # for now use custom_llm_provider=="openai" -> this will change as LiteLLM adds more providers for acreate_batch + response = await litellm.aretrieve_batch( + custom_llm_provider="openai", **_retrieve_batch_request + ) + + ### ALERTING ### + data["litellm_status"] = "success" # used for alerting + + ### RESPONSE HEADERS ### + hidden_params = getattr(response, "_hidden_params", {}) or {} + model_id = hidden_params.get("model_id", None) or "" + cache_key = hidden_params.get("cache_key", None) or "" + api_base = hidden_params.get("api_base", None) or "" + + fastapi_response.headers.update( + get_custom_headers( + user_api_key_dict=user_api_key_dict, + model_id=model_id, + cache_key=cache_key, + api_base=api_base, + version=version, + model_region=getattr(user_api_key_dict, "allowed_model_region", ""), + ) + ) + + return response + except Exception as e: + data["litellm_status"] = "fail" # used for alerting + await proxy_logging_obj.post_call_failure_hook( + user_api_key_dict=user_api_key_dict, original_exception=e, request_data=data + ) + traceback.print_exc() + if isinstance(e, HTTPException): + raise ProxyException( + message=getattr(e, "message", str(e.detail)), + type=getattr(e, "type", "None"), + param=getattr(e, "param", "None"), + code=getattr(e, "status_code", status.HTTP_400_BAD_REQUEST), + ) + else: + error_traceback = traceback.format_exc() + error_msg = f"{str(e)}" + raise ProxyException( + message=getattr(e, "message", error_msg), + type=getattr(e, "type", "None"), + param=getattr(e, "param", "None"), + code=getattr(e, "status_code", 500), + ) + + ###################################################################### # END OF /v1/batches Endpoints Implementation From bffa79a8c5715648b07b8e0006bb4659901551d7 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Tue, 28 May 2024 18:09:22 -0700 Subject: [PATCH 15/20] feat(proxy_server.py): give request-level breakdown if ttft metric is selected for ju st that day --- litellm/proxy/proxy_server.py | 94 ++++++++++++++++++++++++----------- 1 file changed, 66 insertions(+), 28 deletions(-) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 6efcb2a702..f827528485 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -493,7 +493,7 @@ async def user_api_key_auth( if route in LiteLLMRoutes.public_routes.value: # check if public endpoint - return UserAPIKeyAuth() + return UserAPIKeyAuth(user_role="app_owner") if general_settings.get("enable_jwt_auth", False) == True: is_jwt = jwt_handler.is_jwt(token=api_key) @@ -1385,7 +1385,9 @@ async def user_api_key_auth( api_key=api_key, user_role="app_owner", **valid_token_dict ) else: - return UserAPIKeyAuth(api_key=api_key, **valid_token_dict) + return UserAPIKeyAuth( + api_key=api_key, user_role="app_owner", **valid_token_dict + ) else: raise Exception() except Exception as e: @@ -9579,28 +9581,54 @@ async def model_streaming_metrics( startTime = startTime or datetime.now() - timedelta(days=7) # show over past week endTime = endTime or datetime.now() - sql_query = """ - SELECT - api_base, - model_group, - model, - DATE_TRUNC('day', "startTime")::DATE AS day, - AVG(EXTRACT(epoch FROM ("completionStartTime" - "startTime"))) AS time_to_first_token - FROM - "LiteLLM_SpendLogs" - WHERE - "startTime" BETWEEN $2::timestamp AND $3::timestamp - AND "model_group" = $1 AND "cache_hit" != 'True' - AND "completionStartTime" IS NOT NULL - AND "completionStartTime" != "endTime" - GROUP BY - api_base, - model_group, - model, - day - ORDER BY - time_to_first_token DESC; - """ + is_same_day = startTime.date() == endTime.date() + if is_same_day: + sql_query = """ + SELECT + api_base, + model_group, + model, + "startTime", + request_id, + EXTRACT(epoch FROM ("completionStartTime" - "startTime")) AS time_to_first_token + FROM + "LiteLLM_SpendLogs" + WHERE + "model_group" = $1 AND "cache_hit" != 'True' + AND "completionStartTime" IS NOT NULL + AND "completionStartTime" != "endTime" + AND DATE("startTime") = DATE($2::timestamp) + GROUP BY + api_base, + model_group, + model, + request_id + ORDER BY + time_to_first_token DESC; + """ + else: + sql_query = """ + SELECT + api_base, + model_group, + model, + DATE_TRUNC('day', "startTime")::DATE AS day, + AVG(EXTRACT(epoch FROM ("completionStartTime" - "startTime"))) AS time_to_first_token + FROM + "LiteLLM_SpendLogs" + WHERE + "startTime" BETWEEN $2::timestamp AND $3::timestamp + AND "model_group" = $1 AND "cache_hit" != 'True' + AND "completionStartTime" IS NOT NULL + AND "completionStartTime" != "endTime" + GROUP BY + api_base, + model_group, + model, + day + ORDER BY + time_to_first_token DESC; + """ _all_api_bases = set() db_response = await prisma_client.db.query_raw( @@ -9611,10 +9639,19 @@ async def model_streaming_metrics( for model_data in db_response: _api_base = model_data["api_base"] _model = model_data["model"] - _day = model_data["day"] time_to_first_token = model_data["time_to_first_token"] - if _day not in _daily_entries: - _daily_entries[_day] = {} + unique_key = "" + if is_same_day: + _request_id = model_data["request_id"] + unique_key = _request_id + if _request_id not in _daily_entries: + _daily_entries[_request_id] = {} + else: + _day = model_data["day"] + unique_key = _day + time_to_first_token = model_data["time_to_first_token"] + if _day not in _daily_entries: + _daily_entries[_day] = {} _combined_model_name = str(_model) if "https://" in _api_base: _combined_model_name = str(_api_base) @@ -9622,7 +9659,8 @@ async def model_streaming_metrics( _combined_model_name = _combined_model_name.split("/openai/")[0] _all_api_bases.add(_combined_model_name) - _daily_entries[_day][_combined_model_name] = time_to_first_token + + _daily_entries[unique_key][_combined_model_name] = time_to_first_token """ each entry needs to be like this: From 37e18a42fc9bb4ea0b60ecb77955dbb944369142 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 20:03:13 -0700 Subject: [PATCH 16/20] feat - add file endpoints proxy --- litellm/proxy/proxy_server.py | 149 +++++++++++++++++++++++++++++++++- 1 file changed, 147 insertions(+), 2 deletions(-) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index eaf53f0bd7..edc1b832cc 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -5042,12 +5042,12 @@ async def audio_transcriptions( @router.post( "/v1/batches", dependencies=[Depends(user_api_key_auth)], - tags=["Batch"], + tags=["batch"], ) @router.post( "/batches", dependencies=[Depends(user_api_key_auth)], - tags=["Batch"], + tags=["batch"], ) async def create_batch( request: Request, @@ -5329,6 +5329,151 @@ async def retrieve_batch( ###################################################################### +###################################################################### + +# /v1/files Endpoints + + +###################################################################### +@router.post( + "/v1/files", + dependencies=[Depends(user_api_key_auth)], + tags=["files"], +) +@router.post( + "/files", + dependencies=[Depends(user_api_key_auth)], + tags=["files"], +) +async def create_file( + request: Request, + fastapi_response: Response, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """ + Upload a file that can be used across - Assistants API, Batch API + This is the equivalent of POST https://api.openai.com/v1/files + + Supports Identical Params as: https://platform.openai.com/docs/api-reference/files/create + + Example Curl + ``` + curl https://api.openai.com/v1/files \ + -H "Authorization: Bearer sk-1234" \ + -F purpose="batch" \ + -F file="@mydata.jsonl" + + ``` + """ + global proxy_logging_obj + data: Dict = {} + try: + # Use orjson to parse JSON data, orjson speeds up requests significantly + form_data = await request.form() + data = {key: value for key, value in form_data.items() if key != "file"} + + # Include original request and headers in the data + data["proxy_server_request"] = { # type: ignore + "url": str(request.url), + "method": request.method, + "headers": dict(request.headers), + "body": copy.copy(data), # use copy instead of deepcopy + } + + if data.get("user", None) is None and user_api_key_dict.user_id is not None: + data["user"] = user_api_key_dict.user_id + + if "metadata" not in data: + data["metadata"] = {} + data["metadata"]["user_api_key"] = user_api_key_dict.api_key + data["metadata"]["user_api_key_metadata"] = user_api_key_dict.metadata + _headers = dict(request.headers) + _headers.pop( + "authorization", None + ) # do not store the original `sk-..` api key in the db + data["metadata"]["headers"] = _headers + data["metadata"]["user_api_key_alias"] = getattr( + user_api_key_dict, "key_alias", None + ) + data["metadata"]["user_api_key_user_id"] = user_api_key_dict.user_id + data["metadata"]["user_api_key_team_id"] = getattr( + user_api_key_dict, "team_id", None + ) + data["metadata"]["global_max_parallel_requests"] = general_settings.get( + "global_max_parallel_requests", None + ) + data["metadata"]["user_api_key_team_alias"] = getattr( + user_api_key_dict, "team_alias", None + ) + data["metadata"]["endpoint"] = str(request.url) + + ### TEAM-SPECIFIC PARAMS ### + if user_api_key_dict.team_id is not None: + team_config = await proxy_config.load_team_config( + team_id=user_api_key_dict.team_id + ) + if len(team_config) == 0: + pass + else: + team_id = team_config.pop("team_id", None) + data["metadata"]["team_id"] = team_id + data = { + **team_config, + **data, + } # add the team-specific configs to the completion call + + _create_file_request = CreateFileRequest() + + # for now use custom_llm_provider=="openai" -> this will change as LiteLLM adds more providers for acreate_batch + response = await litellm.acreate_file( + custom_llm_provider="openai", **_create_file_request + ) + + ### ALERTING ### + data["litellm_status"] = "success" # used for alerting + + ### RESPONSE HEADERS ### + hidden_params = getattr(response, "_hidden_params", {}) or {} + model_id = hidden_params.get("model_id", None) or "" + cache_key = hidden_params.get("cache_key", None) or "" + api_base = hidden_params.get("api_base", None) or "" + + fastapi_response.headers.update( + get_custom_headers( + user_api_key_dict=user_api_key_dict, + model_id=model_id, + cache_key=cache_key, + api_base=api_base, + version=version, + model_region=getattr(user_api_key_dict, "allowed_model_region", ""), + ) + ) + + return response + except Exception as e: + data["litellm_status"] = "fail" # used for alerting + await proxy_logging_obj.post_call_failure_hook( + user_api_key_dict=user_api_key_dict, original_exception=e, request_data=data + ) + traceback.print_exc() + if isinstance(e, HTTPException): + raise ProxyException( + message=getattr(e, "message", str(e.detail)), + type=getattr(e, "type", "None"), + param=getattr(e, "param", "None"), + code=getattr(e, "status_code", status.HTTP_400_BAD_REQUEST), + ) + else: + error_traceback = traceback.format_exc() + error_msg = f"{str(e)}" + raise ProxyException( + message=getattr(e, "message", error_msg), + type=getattr(e, "type", "None"), + param=getattr(e, "param", "None"), + code=getattr(e, "status_code", 500), + ) + + @router.post( "/v1/moderations", dependencies=[Depends(user_api_key_auth)], From ca8163bbbacb93afc16d723d8f4a3745e08617b6 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 20:58:22 -0700 Subject: [PATCH 17/20] feat - add afile_content, file_content --- litellm/batches/main.py | 130 +++++++++++++++++++++++++++ litellm/llms/openai.py | 48 ++++++++++ litellm/tests/test_openai_batches.py | 25 +++++- litellm/types/llms/openai.py | 21 +++++ 4 files changed, 222 insertions(+), 2 deletions(-) diff --git a/litellm/batches/main.py b/litellm/batches/main.py index 917ad45e10..5d9a3a1411 100644 --- a/litellm/batches/main.py +++ b/litellm/batches/main.py @@ -30,6 +30,8 @@ from ..types.llms.openai import ( FileTypes, FileObject, Batch, + FileContentRequest, + HttpxBinaryResponseContent, ) ####### ENVIRONMENT VARIABLES ################### @@ -170,6 +172,134 @@ def create_file( raise e +async def afile_content( + file_id: str, + custom_llm_provider: Literal["openai"] = "openai", + extra_headers: Optional[Dict[str, str]] = None, + extra_body: Optional[Dict[str, str]] = None, + **kwargs, +) -> Coroutine[Any, Any, HttpxBinaryResponseContent]: + """ + Async: Get file contents + + LiteLLM Equivalent of GET https://api.openai.com/v1/files + """ + try: + loop = asyncio.get_event_loop() + kwargs["afile_content"] = True + + # Use a partial function to pass your keyword arguments + func = partial( + file_content, + file_id, + custom_llm_provider, + extra_headers, + extra_body, + **kwargs, + ) + + # Add the context to the function + ctx = contextvars.copy_context() + func_with_context = partial(ctx.run, func) + init_response = await loop.run_in_executor(None, func_with_context) + if asyncio.iscoroutine(init_response): + response = await init_response + else: + response = init_response # type: ignore + + return response + except Exception as e: + raise e + + +def file_content( + file_id: str, + custom_llm_provider: Literal["openai"] = "openai", + extra_headers: Optional[Dict[str, str]] = None, + extra_body: Optional[Dict[str, str]] = None, + **kwargs, +) -> Union[HttpxBinaryResponseContent, Coroutine[Any, Any, HttpxBinaryResponseContent]]: + """ + Returns the contents of the specified file. + + LiteLLM Equivalent of POST: POST https://api.openai.com/v1/files + """ + try: + optional_params = GenericLiteLLMParams(**kwargs) + if custom_llm_provider == "openai": + # for deepinfra/perplexity/anyscale/groq we check in get_llm_provider and pass in the api base from there + api_base = ( + optional_params.api_base + or litellm.api_base + or os.getenv("OPENAI_API_BASE") + or "https://api.openai.com/v1" + ) + organization = ( + optional_params.organization + or litellm.organization + or os.getenv("OPENAI_ORGANIZATION", None) + or None # default - https://github.com/openai/openai-python/blob/284c1799070c723c6a553337134148a7ab088dd8/openai/util.py#L105 + ) + # set API KEY + api_key = ( + optional_params.api_key + or litellm.api_key # for deepinfra/perplexity/anyscale we check in get_llm_provider and pass in the api key from there + or litellm.openai_key + or os.getenv("OPENAI_API_KEY") + ) + ### TIMEOUT LOGIC ### + timeout = ( + optional_params.timeout or kwargs.get("request_timeout", 600) or 600 + ) + # set timeout for 10 minutes by default + + if ( + timeout is not None + and isinstance(timeout, httpx.Timeout) + and supports_httpx_timeout(custom_llm_provider) == False + ): + read_timeout = timeout.read or 600 + timeout = read_timeout # default 10 min timeout + elif timeout is not None and not isinstance(timeout, httpx.Timeout): + timeout = float(timeout) # type: ignore + elif timeout is None: + timeout = 600.0 + + _file_content_request = FileContentRequest( + file_id=file_id, + extra_headers=extra_headers, + extra_body=extra_body, + ) + + _is_async = kwargs.pop("afile_content", False) is True + + response = openai_files_instance.file_content( + _is_async=_is_async, + file_content_request=_file_content_request, + api_base=api_base, + api_key=api_key, + timeout=timeout, + max_retries=optional_params.max_retries, + organization=organization, + ) + else: + raise litellm.exceptions.BadRequestError( + message="LiteLLM doesn't support {} for 'create_batch'. Only 'openai' is supported.".format( + custom_llm_provider + ), + model="n/a", + llm_provider=custom_llm_provider, + response=httpx.Response( + status_code=400, + content="Unsupported provider", + request=httpx.Request(method="create_thread", url="https://github.com/BerriAI/litellm"), # type: ignore + ), + ) + return response + except Exception as e: + raise e + + async def acreate_batch( completion_window: Literal["24h"], endpoint: Literal["/v1/chat/completions", "/v1/embeddings", "/v1/completions"], diff --git a/litellm/llms/openai.py b/litellm/llms/openai.py index 43d088f0db..1a1dc4e6dd 100644 --- a/litellm/llms/openai.py +++ b/litellm/llms/openai.py @@ -1585,6 +1585,54 @@ class OpenAIFilesAPI(BaseLLM): response = openai_client.files.create(**create_file_data) return response + async def afile_content( + self, + file_content_request: FileContentRequest, + openai_client: AsyncOpenAI, + ) -> HttpxBinaryResponseContent: + response = await openai_client.files.content(**file_content_request) + return response + + def file_content( + self, + _is_async: bool, + file_content_request: FileContentRequest, + api_base: str, + api_key: Optional[str], + timeout: Union[float, httpx.Timeout], + max_retries: Optional[int], + organization: Optional[str], + client: Optional[Union[OpenAI, AsyncOpenAI]] = None, + ) -> Union[ + HttpxBinaryResponseContent, Coroutine[Any, Any, HttpxBinaryResponseContent] + ]: + openai_client: Optional[Union[OpenAI, AsyncOpenAI]] = self.get_openai_client( + api_key=api_key, + api_base=api_base, + timeout=timeout, + max_retries=max_retries, + organization=organization, + client=client, + _is_async=_is_async, + ) + if openai_client is None: + raise ValueError( + "OpenAI client is not initialized. Make sure api_key is passed or OPENAI_API_KEY is set in the environment." + ) + + if _is_async is True: + if not isinstance(openai_client, AsyncOpenAI): + raise ValueError( + "OpenAI client is not an instance of AsyncOpenAI. Make sure you passed an AsyncOpenAI client." + ) + return self.afile_content( # type: ignore + file_content_request=file_content_request, + openai_client=openai_client, + ) + response = openai_client.files.content(**file_content_request) + + return response + class OpenAIBatchesAPI(BaseLLM): """ diff --git a/litellm/tests/test_openai_batches.py b/litellm/tests/test_openai_batches.py index fc29331a5a..d7e3e18098 100644 --- a/litellm/tests/test_openai_batches.py +++ b/litellm/tests/test_openai_batches.py @@ -60,8 +60,6 @@ def test_create_batch(): create_batch_response.input_file_id == batch_input_file_id ), f"Failed to create batch, expected input_file_id to be {batch_input_file_id} but got {create_batch_response.input_file_id}" - time.sleep(30) - retrieved_batch = litellm.retrieve_batch( batch_id=create_batch_response.id, custom_llm_provider="openai" ) @@ -70,6 +68,17 @@ def test_create_batch(): assert retrieved_batch.id == create_batch_response.id + file_content = litellm.file_content( + file_id=batch_input_file_id, custom_llm_provider="openai" + ) + + result = file_content.content + + result_file_name = "batch_job_results_furniture.jsonl" + + with open(result_file_name, "wb") as file: + file.write(result) + pass @@ -127,6 +136,18 @@ async def test_async_create_batch(): assert retrieved_batch.id == create_batch_response.id + # try to get file content for our original file + + file_content = await litellm.afile_content( + file_id=batch_input_file_id, custom_llm_provider="openai" + ) + + print("file content = ", file_content) + + # # write this file content to a file + # with open("file_content.json", "w") as f: + # json.dump(file_content, f) + def test_retrieve_batch(): pass diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py index 50ac1335ec..77791b8ece 100644 --- a/litellm/types/llms/openai.py +++ b/litellm/types/llms/openai.py @@ -20,6 +20,7 @@ from openai.types.beta.assistant import Assistant from openai.pagination import SyncCursorPage from os import PathLike from openai.types import FileObject, Batch +from openai._legacy_response import HttpxBinaryResponseContent from typing import TypedDict, List, Optional, Tuple, Mapping, IO @@ -186,6 +187,26 @@ class CreateFileRequest(TypedDict, total=False): timeout: Optional[float] +class FileContentRequest(TypedDict, total=False): + """ + FileContentRequest + Used by Assistants API, Batches API, and Fine-Tunes API + + Required Params: + file_id: str + + Optional Params: + extra_headers: Optional[Dict[str, str]] + extra_body: Optional[Dict[str, str]] = None + timeout: Optional[float] = None + """ + + file_id: str + extra_headers: Optional[Dict[str, str]] + extra_body: Optional[Dict[str, str]] + timeout: Optional[float] + + # OpenAI Batches Types class CreateBatchRequest(TypedDict, total=False): """ From 473ec66b848a4e7c4b7fc785d55302f9515ddd68 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 22:19:33 -0700 Subject: [PATCH 18/20] feat - router add abatch_completion --- litellm/router.py | 65 +++++++++++++++---- litellm/tests/test_router_batch_completion.py | 44 +++++++++++++ 2 files changed, 96 insertions(+), 13 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index e2ebea37fa..a2a03da86f 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -356,7 +356,8 @@ class Router: raise ValueError(f"Item '{fallback_dict}' is not a dictionary.") if len(fallback_dict) != 1: raise ValueError( - f"Dictionary '{fallback_dict}' must have exactly one key, but has {len(fallback_dict)} keys.") + f"Dictionary '{fallback_dict}' must have exactly one key, but has {len(fallback_dict)} keys." + ) def routing_strategy_init(self, routing_strategy: str, routing_strategy_args: dict): if routing_strategy == "least-busy": @@ -662,12 +663,17 @@ class Router: raise e async def abatch_completion( - self, models: List[str], messages: List[Dict[str, str]], **kwargs + self, + models: List[str], + messages: Union[List[Dict[str, str]], List[List[Dict[str, str]]]], + **kwargs, ): """ - Async Batch Completion - Batch Process 1 request to multiple model_group on litellm.Router - Use this for sending the same request to N models + Async Batch Completion. Used for 2 scenarios: + 1. Batch Process 1 request to N models on litellm.Router. Pass messages as List[Dict[str, str]] to use this + 2. Batch Process N requests to M models on litellm.Router. Pass messages as List[List[Dict[str, str]]] to use this """ + ############## Helpers for async completion ################## async def _async_completion_no_exceptions( model: str, messages: List[Dict[str, str]], **kwargs @@ -680,17 +686,50 @@ class Router: except Exception as e: return e - _tasks = [] - for model in models: - # add each task but if the task fails - _tasks.append( - _async_completion_no_exceptions( - model=model, messages=messages, **kwargs + async def _async_completion_no_exceptions_return_idx( + model: str, + messages: List[Dict[str, str]], + idx: int, # index of message this response corresponds to + **kwargs, + ): + """ + Wrapper around self.async_completion that catches exceptions and returns them as a result + """ + try: + return ( + await self.acompletion(model=model, messages=messages, **kwargs), + idx, ) - ) + except Exception as e: + return e, idx - response = await asyncio.gather(*_tasks) - return response + ############## Helpers for async completion ################## + + if isinstance(messages, list) and all(isinstance(m, dict) for m in messages): + _tasks = [] + for model in models: + # add each task but if the task fails + _tasks.append(_async_completion_no_exceptions(model=model, messages=messages, **kwargs)) # type: ignore + response = await asyncio.gather(*_tasks) + return response + elif isinstance(messages, list) and all(isinstance(m, list) for m in messages): + _tasks = [] + for idx, message in enumerate(messages): + for model in models: + # Request Number X, Model Number Y + _tasks.append( + _async_completion_no_exceptions_return_idx( + model=model, idx=idx, messages=message, **kwargs # type: ignore + ) + ) + responses = await asyncio.gather(*_tasks) + final_responses: List[List[Any]] = [[] for _ in range(len(messages))] + for response in responses: + if isinstance(response, tuple): + final_responses[response[1]].append(response[0]) + else: + final_responses[0].append(response) + return final_responses async def abatch_completion_one_model_multiple_requests( self, model: str, messages: List[List[Dict[str, str]]], **kwargs diff --git a/litellm/tests/test_router_batch_completion.py b/litellm/tests/test_router_batch_completion.py index f2873b18d5..0925a38358 100644 --- a/litellm/tests/test_router_batch_completion.py +++ b/litellm/tests/test_router_batch_completion.py @@ -58,3 +58,47 @@ async def test_batch_completion_multiple_models(): # assert both models are different assert models_in_responses[0] != models_in_responses[1] + + +@pytest.mark.asyncio +async def test_batch_completion_multiple_models_multiple_messages(): + litellm.set_verbose = True + + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-3.5-turbo", + "litellm_params": { + "model": "gpt-3.5-turbo", + }, + }, + { + "model_name": "groq-llama", + "litellm_params": { + "model": "groq/llama3-8b-8192", + }, + }, + ] + ) + + response = await router.abatch_completion( + models=["gpt-3.5-turbo", "groq-llama"], + messages=[ + [{"role": "user", "content": "is litellm becoming a better product ?"}], + [{"role": "user", "content": "who is this"}], + ], + max_tokens=15, + ) + + print("response from batches =", response) + assert len(response) == 2 + assert len(response[0]) == 2 + assert isinstance(response[0][0], litellm.ModelResponse) + + # models_in_responses = [] + # for individual_response in response: + # _model = individual_response["model"] + # models_in_responses.append(_model) + + # # assert both models are different + # assert models_in_responses[0] != models_in_responses[1] From 9ab96e12ed002facaa57fc7fed7f10a8ee1c52a7 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 22:27:09 -0700 Subject: [PATCH 19/20] fix - update abatch_completion docstring --- litellm/router.py | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/litellm/router.py b/litellm/router.py index a2a03da86f..9c9f81dc88 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -672,6 +672,29 @@ class Router: Async Batch Completion. Used for 2 scenarios: 1. Batch Process 1 request to N models on litellm.Router. Pass messages as List[Dict[str, str]] to use this 2. Batch Process N requests to M models on litellm.Router. Pass messages as List[List[Dict[str, str]]] to use this + + Example Request for 1 request to N models: + ``` + response = await router.abatch_completion( + models=["gpt-3.5-turbo", "groq-llama"], + messages=[ + {"role": "user", "content": "is litellm becoming a better product ?"} + ], + max_tokens=15, + ) + ``` + + + Example Request for N requests to M models: + ``` + response = await router.abatch_completion( + models=["gpt-3.5-turbo", "groq-llama"], + messages=[ + [{"role": "user", "content": "is litellm becoming a better product ?"}], + [{"role": "user", "content": "who is this"}], + ], + ) + ``` """ ############## Helpers for async completion ################## From 7671a65f1bc7a8a8900b9be099f4f0cfda4cfe4a Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 28 May 2024 22:35:00 -0700 Subject: [PATCH 20/20] =?UTF-8?q?bump:=20version=201.39.0=20=E2=86=92=201.?= =?UTF-8?q?39.1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pyproject.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 0fb6b3269f..d124ea4a1a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "litellm" -version = "1.39.0" +version = "1.39.1" description = "Library to easily interface with LLM API providers" authors = ["BerriAI"] license = "MIT" @@ -79,7 +79,7 @@ requires = ["poetry-core", "wheel"] build-backend = "poetry.core.masonry.api" [tool.commitizen] -version = "1.39.0" +version = "1.39.1" version_files = [ "pyproject.toml:^version" ]