diff --git a/litellm/batches/batch_utils.py b/litellm/batches/batch_utils.py index 42ff534c28..b7d147d8e1 100644 --- a/litellm/batches/batch_utils.py +++ b/litellm/batches/batch_utils.py @@ -9,12 +9,12 @@ from litellm._logging import verbose_logger from litellm._uuid import uuid from litellm.types.llms.openai import Batch from litellm.types.utils import CallTypes, ModelResponse, Usage -from litellm.utils import token_counter +from litellm.utils import token_counter, ProviderConfigManager async def calculate_batch_cost_and_usage( file_content_dictionary: List[dict], - custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm"], + custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"], model_name: Optional[str] = None, ) -> Tuple[float, Usage, List[str]]: """ @@ -30,14 +30,16 @@ async def calculate_batch_cost_and_usage( custom_llm_provider=custom_llm_provider, model_name=model_name, ) - batch_models = _get_batch_models_from_file_content(file_content_dictionary, model_name) + batch_models = _get_batch_models_from_file_content( + file_content_dictionary, model_name, custom_llm_provider=custom_llm_provider + ) return batch_cost, batch_usage, batch_models async def _handle_completed_batch( batch: Batch, - custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm"], + custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"], model_name: Optional[str] = None, ) -> Tuple[float, Usage, List[str]]: """Helper function to process a completed batch and handle logging""" @@ -58,14 +60,136 @@ async def _handle_completed_batch( model_name=model_name, ) - batch_models = _get_batch_models_from_file_content(file_content_dictionary, model_name) + batch_models = _get_batch_models_from_file_content( + file_content_dictionary, model_name, custom_llm_provider=custom_llm_provider + ) return batch_cost, batch_usage, batch_models +def transform_raw_provider_response_to_openai( + raw_response: dict, + custom_llm_provider: str, + model: Optional[str] = None, + messages: Optional[list] = None, +) -> ModelResponse: + """ + Unified method to transform any raw LLM provider response to OpenAI format. + + Args: + raw_response: Raw response dictionary from any provider (Anthropic, OpenAI, etc.) + custom_llm_provider: Provider name (e.g., "anthropic", "openai", "vertex_ai") + model: Model name (optional, will try to extract from raw_response if not provided) + messages: Original messages list (optional, defaults to empty list) + + Returns: + ModelResponse: OpenAI-compatible response object + """ + # Lazy import to avoid circular dependency + from litellm.litellm_core_utils.litellm_logging import Logging + + # Extract model from response if not provided + if model is None: + model = raw_response.get("model", "unknown-model") + + # Default messages if not provided + if messages is None: + messages = [] + + # Get provider config using ProviderConfigManager + provider_config = ProviderConfigManager.get_provider_chat_config( + model=model, + provider=litellm.LlmProviders(custom_llm_provider) + ) + + if provider_config is None: + raise ValueError(f"Could not get config for provider: {custom_llm_provider}") + + # Create a mock httpx.Response from the dict + response_text = json.dumps(raw_response) + mock_httpx_response = httpx.Response( + status_code=200, + content=response_text.encode('utf-8'), + headers={"content-type": "application/json"} + ) + + # Create a minimal logging object + logging_obj = Logging( + model=model, + messages=messages, + stream=False, + call_type=CallTypes.completion.value, + start_time=time.time(), + litellm_call_id=None, + function_id=None, + ) + + # Create empty ModelResponse to be populated + model_response = ModelResponse() + + # Call transform_response on the provider config + transformed_response = provider_config.transform_response( + model=model, + raw_response=mock_httpx_response, + model_response=model_response, + logging_obj=logging_obj, + request_data={}, + messages=messages, + optional_params={}, + litellm_params={}, + encoding=litellm.encoding, + api_key=None, + json_mode=None, + ) + + return transformed_response + + +def _extract_raw_response_from_batch_item( + batch_item: dict, + custom_llm_provider: str, +) -> Optional[dict]: + """ + Extract the raw provider response from a batch output file item. + + Handles different batch output formats: + - Anthropic: {"result": {"type": "succeeded", "message": {...}}} + - Vertex AI: {"status": "JOB_STATE_SUCCEEDED", "response": {...}} + - OpenAI/Azure: {"response": {"status_code": 200, "body": {...}}} + + Args: + batch_item: A single item from the batch output file + custom_llm_provider: Provider name (e.g., "anthropic", "openai", "vertex_ai") + + Returns: + Raw response dict or None if not successful + """ + # Anthropic format: {"result": {"type": "succeeded", "message": {...}}} + if custom_llm_provider == "anthropic": + result = batch_item.get("result", {}) + if result.get("type") == "succeeded": + return result.get("message", {}) + return None + + # Vertex AI format: {"status": "JOB_STATE_SUCCEEDED", "response": {...}} + if custom_llm_provider == "vertex_ai": + if batch_item.get("status") == "JOB_STATE_SUCCEEDED": + return batch_item.get("response", {}) + return None + + # OpenAI/Azure format: {"response": {"status_code": 200, "body": {...}}} + # Default to OpenAI format for openai, azure, hosted_vllm, etc. + response = batch_item.get("response", {}) + if response.get("status_code") == 200: + return response.get("body", {}) + + return None + + def _get_batch_models_from_file_content( file_content_dictionary: List[dict], model_name: Optional[str] = None, + custom_llm_provider: str = "openai", ) -> List[str]: """ Get the models from the file content @@ -74,119 +198,79 @@ def _get_batch_models_from_file_content( return [model_name] batch_models = [] for _item in file_content_dictionary: - if _batch_response_was_successful(_item): - _response_body = _get_response_from_batch_job_output_file(_item) - _model = _response_body.get("model") - if _model: - batch_models.append(_model) + if _batch_response_was_successful(_item, custom_llm_provider=custom_llm_provider): + # Extract raw response using generalized method + raw_response = _extract_raw_response_from_batch_item( + batch_item=_item, + custom_llm_provider=custom_llm_provider, + ) + if raw_response: + _model = raw_response.get("model") + if _model: + batch_models.append(_model) return batch_models def _batch_cost_calculator( file_content_dictionary: List[dict], - custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm"] = "openai", + custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"] = "openai", model_name: Optional[str] = None, ) -> float: """ Calculate the cost of a batch based on the output file id """ - # Handle Vertex AI with specialized method - if custom_llm_provider == "vertex_ai" and model_name: - batch_cost, _ = calculate_vertex_ai_batch_cost_and_usage(file_content_dictionary, model_name) - verbose_logger.debug("vertex_ai_total_cost=%s", batch_cost) - return batch_cost + total_cost: float = 0.0 + + for batch_item in file_content_dictionary: + if not _batch_response_was_successful(batch_item, custom_llm_provider=custom_llm_provider): + continue + + # Extract raw response from batch item + raw_response = _extract_raw_response_from_batch_item( + batch_item=batch_item, + custom_llm_provider=custom_llm_provider, + ) + + if raw_response is None: + continue + + # Extract model from response if not provided + actual_model = model_name or raw_response.get("model") + if actual_model is None: + verbose_logger.warning("Could not determine model for batch item, skipping cost calculation") + continue + + try: + # Transform to OpenAI format using generalized method + openai_format_response = transform_raw_provider_response_to_openai( + raw_response=raw_response, + custom_llm_provider=custom_llm_provider, + model=actual_model, + messages=[], # Messages not needed for cost calculation + ) + + # Calculate cost using standard OpenAI cost calculation + cost = litellm.completion_cost( + completion_response=openai_format_response, + custom_llm_provider=custom_llm_provider, + call_type=CallTypes.aretrieve_batch.value, + ) + total_cost += cost + verbose_logger.debug("item_cost=%s, total_cost=%s", cost, total_cost) + except Exception as e: + verbose_logger.warning( + f"Error calculating cost for batch item: {e}. Skipping this item." + ) + continue - # For other providers, use the existing logic - total_cost = _get_batch_job_cost_from_file_content( - file_content_dictionary=file_content_dictionary, - custom_llm_provider=custom_llm_provider, - ) verbose_logger.debug("total_cost=%s", total_cost) return total_cost -def calculate_vertex_ai_batch_cost_and_usage( - vertex_ai_batch_responses: List[dict], - model_name: Optional[str] = None, -) -> Tuple[float, Usage]: - """ - Calculate both cost and usage from Vertex AI batch responses - """ - from litellm.litellm_core_utils.litellm_logging import Logging - from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import ( - VertexGeminiConfig, - ) - total_cost = 0.0 - total_tokens = 0 - prompt_tokens = 0 - completion_tokens = 0 - - for response in vertex_ai_batch_responses: - if response.get("status") == "JOB_STATE_SUCCEEDED": # Check if response was successful - # Transform Vertex AI response to OpenAI format if needed - - # Create required arguments for the transformation method - model_response = ModelResponse() - - # Ensure model_name is not None - actual_model_name = model_name or "gemini-2.5-flash" - - # Create a real LiteLLM logging object - logging_obj = Logging( - model=actual_model_name, - messages=[{"role": "user", "content": "batch_request"}], - stream=False, - call_type=CallTypes.aretrieve_batch, - start_time=time.time(), - litellm_call_id="batch_" + str(uuid.uuid4()), - function_id="batch_processing", - litellm_trace_id=str(uuid.uuid4()), - kwargs={"optional_params": {}} - ) - - # Add the optional_params attribute that the Vertex AI transformation expects - logging_obj.optional_params = {} - raw_response = httpx.Response(200) # Mock response object - - openai_format_response = VertexGeminiConfig()._transform_google_generate_content_to_openai_model_response( - completion_response=response["response"], - model_response=model_response, - model=actual_model_name, - logging_obj=logging_obj, - raw_response=raw_response, - ) - - # Calculate cost using existing function - cost = litellm.completion_cost( - completion_response=openai_format_response, - custom_llm_provider="vertex_ai", - call_type=CallTypes.aretrieve_batch.value, - ) - total_cost += cost - - # Extract usage from the transformed response - usage_obj = getattr(openai_format_response, 'usage', None) - if usage_obj: - usage = usage_obj - else: - # Fallback: create usage from response dict - response_dict = openai_format_response.dict() if hasattr(openai_format_response, 'dict') else {} - usage = _get_batch_job_usage_from_response_body(response_dict) - - total_tokens += usage.total_tokens - prompt_tokens += usage.prompt_tokens - completion_tokens += usage.completion_tokens - - return total_cost, Usage( - total_tokens=total_tokens, - prompt_tokens=prompt_tokens, - completion_tokens=completion_tokens, - ) - async def _get_batch_output_file_content_as_dictionary( batch: Batch, - custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm"] = "openai", + custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"] = "openai", ) -> List[dict]: """ Get the batch output file content as a list of dictionaries @@ -223,62 +307,72 @@ def _get_file_content_as_dictionary(file_content: bytes) -> List[dict]: raise e -def _get_batch_job_cost_from_file_content( - file_content_dictionary: List[dict], - custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm"] = "openai", -) -> float: - """ - Get the cost of a batch job from the file content - """ - try: - total_cost: float = 0.0 - # parse the file content as json - verbose_logger.debug( - "file_content_dictionary=%s", json.dumps(file_content_dictionary, indent=4) - ) - for _item in file_content_dictionary: - if _batch_response_was_successful(_item): - _response_body = _get_response_from_batch_job_output_file(_item) - total_cost += litellm.completion_cost( - completion_response=_response_body, - custom_llm_provider=custom_llm_provider, - call_type=CallTypes.aretrieve_batch.value, - ) - verbose_logger.debug("total_cost=%s", total_cost) - return total_cost - except Exception as e: - verbose_logger.error("error in _get_batch_job_cost_from_file_content", e) - raise e - - def _get_batch_job_total_usage_from_file_content( file_content_dictionary: List[dict], - custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm"] = "openai", + custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"] = "openai", model_name: Optional[str] = None, ) -> Usage: """ Get the tokens of a batch job from the file content """ - # Handle Vertex AI with specialized method - if custom_llm_provider == "vertex_ai" and model_name: - _, batch_usage = calculate_vertex_ai_batch_cost_and_usage(file_content_dictionary, model_name) - return batch_usage + from litellm.cost_calculator import BaseTokenUsageProcessor - # For other providers, use the existing logic - total_tokens: int = 0 - prompt_tokens: int = 0 - completion_tokens: int = 0 - for _item in file_content_dictionary: - if _batch_response_was_successful(_item): - _response_body = _get_response_from_batch_job_output_file(_item) - usage: Usage = _get_batch_job_usage_from_response_body(_response_body) - total_tokens += usage.total_tokens - prompt_tokens += usage.prompt_tokens - completion_tokens += usage.completion_tokens + all_usage: List[Usage] = [] + + for batch_item in file_content_dictionary: + if not _batch_response_was_successful(batch_item, custom_llm_provider=custom_llm_provider): + continue + + # Extract raw response from batch item + raw_response = _extract_raw_response_from_batch_item( + batch_item=batch_item, + custom_llm_provider=custom_llm_provider, + ) + + if raw_response is None: + continue + + # Extract model from response if not provided + actual_model = model_name or raw_response.get("model") + if actual_model is None: + verbose_logger.warning("Could not determine model for batch item, skipping usage calculation") + continue + + try: + # Transform to OpenAI format using generalized method + openai_format_response = transform_raw_provider_response_to_openai( + raw_response=raw_response, + custom_llm_provider=custom_llm_provider, + model=actual_model, + messages=[], # Messages not needed for usage extraction + ) + + # Extract usage from transformed response + usage_obj = getattr(openai_format_response, 'usage', None) + if usage_obj: + all_usage.append(usage_obj) + else: + # Fallback: try to extract from response dict + response_dict = openai_format_response.dict() if hasattr(openai_format_response, 'dict') else {} + usage = _get_batch_job_usage_from_response_body(response_dict) + if usage and usage.total_tokens > 0: + all_usage.append(usage) + except Exception as e: + verbose_logger.warning( + f"Error extracting usage for batch item: {e}. Skipping this item." + ) + continue + + # Combine all usage objects + if all_usage: + combined_usage = BaseTokenUsageProcessor.combine_usage_objects(all_usage) + return combined_usage + + # Return empty usage if no valid responses return Usage( - total_tokens=total_tokens, - prompt_tokens=prompt_tokens, - completion_tokens=completion_tokens, + total_tokens=0, + prompt_tokens=0, + completion_tokens=0, ) def _get_batch_job_input_file_usage( @@ -318,18 +412,30 @@ def _get_batch_job_usage_from_response_body(response_body: dict) -> Usage: return usage -def _get_response_from_batch_job_output_file(batch_job_output_file: dict) -> Any: +def _batch_response_was_successful( + batch_job_output_file: dict, + custom_llm_provider: str = "openai", +) -> bool: """ - Get the response from the batch job output file - """ - _response: dict = batch_job_output_file.get("response", None) or {} - _response_body = _response.get("body", None) or {} - return _response_body - - -def _batch_response_was_successful(batch_job_output_file: dict) -> bool: - """ - Check if the batch job response status == 200 + Check if the batch job response was successful. + + Args: + batch_job_output_file: A single item from the batch output file + custom_llm_provider: Provider name (e.g., "anthropic", "openai", "vertex_ai") + + Returns: + True if the batch response was successful, False otherwise """ + # Anthropic format: {"result": {"type": "succeeded", "message": {...}}} + if custom_llm_provider == "anthropic": + result = batch_job_output_file.get("result", {}) + return result.get("type") == "succeeded" + + # Vertex AI format: {"status": "JOB_STATE_SUCCEEDED", "response": {...}} + if custom_llm_provider == "vertex_ai": + return batch_job_output_file.get("status") == "JOB_STATE_SUCCEEDED" + + # OpenAI/Azure format: {"response": {"status_code": 200, "body": {...}}} + # Default to OpenAI format for openai, azure, hosted_vllm, etc. _response: dict = batch_job_output_file.get("response", None) or {} return _response.get("status_code", None) == 200 diff --git a/litellm/batches/main.py b/litellm/batches/main.py index b99f4a628d..126eb09a51 100644 --- a/litellm/batches/main.py +++ b/litellm/batches/main.py @@ -22,6 +22,7 @@ from openai.types.batch import BatchRequestCounts import litellm from litellm._logging import verbose_logger from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.anthropic.batches.handler import AnthropicBatchesHandler from litellm.llms.azure.batches.handler import AzureBatchesAPI from litellm.llms.bedrock.batches.handler import BedrockBatchesHandler from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler @@ -53,6 +54,7 @@ from litellm.utils import ( openai_batches_instance = OpenAIBatchesAPI() azure_batches_instance = AzureBatchesAPI() vertex_ai_batches_instance = VertexAIBatchPrediction(gcs_bucket_name="") +anthropic_batches_instance = AnthropicBatchesHandler() base_llm_http_handler = BaseLLMHTTPHandler() ################################################# @@ -355,7 +357,7 @@ def create_batch( @client async def aretrieve_batch( batch_id: str, - custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm"] = "openai", + custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic"] = "openai", metadata: Optional[Dict[str, str]] = None, extra_headers: Optional[Dict[str, str]] = None, extra_body: Optional[Dict[str, str]] = None, @@ -401,7 +403,7 @@ def _handle_retrieve_batch_providers_without_provider_config( litellm_params: dict, _retrieve_batch_request: RetrieveBatchRequest, _is_async: bool, - custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm"] = "openai", + custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic"] = "openai", ): api_base: Optional[str] = None if custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS: @@ -498,6 +500,27 @@ def _handle_retrieve_batch_providers_without_provider_config( timeout=timeout, max_retries=optional_params.max_retries, ) + elif custom_llm_provider == "anthropic": + api_base = ( + optional_params.api_base + or litellm.api_base + or get_secret_str("ANTHROPIC_API_BASE") + ) + api_key = ( + optional_params.api_key + or litellm.api_key + or litellm.azure_key + or get_secret_str("ANTHROPIC_API_KEY") + ) + + response = anthropic_batches_instance.retrieve_batch( + _is_async=_is_async, + batch_id=batch_id, + api_base=api_base, + api_key=api_key, + 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( @@ -517,7 +540,7 @@ def _handle_retrieve_batch_providers_without_provider_config( @client def retrieve_batch( batch_id: str, - custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm"] = "openai", + custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic"] = "openai", metadata: Optional[Dict[str, str]] = None, extra_headers: Optional[Dict[str, str]] = None, extra_body: Optional[Dict[str, str]] = None, @@ -608,7 +631,7 @@ def retrieve_batch( api_key=optional_params.api_key, logging_obj=litellm_logging_obj or LiteLLMLoggingObj( - model=model or "bedrock/unknown", + model=model or f"{custom_llm_provider}/unknown", messages=[], stream=False, call_type="batch_retrieve", diff --git a/litellm/files/main.py b/litellm/files/main.py index 9378715a47..acf545e431 100644 --- a/litellm/files/main.py +++ b/litellm/files/main.py @@ -17,6 +17,7 @@ import litellm from litellm import get_secret_str from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.anthropic.files.handler import AnthropicFilesHandler from litellm.llms.azure.files.handler import AzureOpenAIFilesAPI from litellm.llms.bedrock.files.handler import BedrockFilesHandler from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler @@ -49,6 +50,7 @@ openai_files_instance = OpenAIFilesAPI() azure_files_instance = AzureOpenAIFilesAPI() vertex_ai_files_instance = VertexAIFilesHandler() bedrock_files_instance = BedrockFilesHandler() +anthropic_files_instance = AnthropicFilesHandler() ################################################# @@ -757,7 +759,7 @@ def file_list( @client async def afile_content( file_id: str, - custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm"] = "openai", + custom_llm_provider: Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic"] = "openai", extra_headers: Optional[Dict[str, str]] = None, extra_body: Optional[Dict[str, str]] = None, **kwargs, @@ -802,7 +804,7 @@ def file_content( file_id: str, model: Optional[str] = None, custom_llm_provider: Optional[ - Union[Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm"], str] + Union[Literal["openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic"], str] ] = None, extra_headers: Optional[Dict[str, str]] = None, extra_body: Optional[Dict[str, str]] = None, @@ -849,6 +851,18 @@ def file_content( _is_async = kwargs.pop("afile_content", False) is True + # Check if this is an Anthropic batch results request + if custom_llm_provider == "anthropic": + response = anthropic_files_instance.file_content( + _is_async=_is_async, + file_content_request=_file_content_request, + api_base=optional_params.api_base, + api_key=optional_params.api_key, + timeout=timeout, + max_retries=optional_params.max_retries, + ) + return response + if custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS: # for deepinfra/perplexity/anyscale/groq we check in get_llm_provider and pass in the api base from there api_base = ( diff --git a/litellm/llms/anthropic/batches/__init__.py b/litellm/llms/anthropic/batches/__init__.py new file mode 100644 index 0000000000..66d1a8f77f --- /dev/null +++ b/litellm/llms/anthropic/batches/__init__.py @@ -0,0 +1,5 @@ +from .handler import AnthropicBatchesHandler +from .transformation import AnthropicBatchesConfig + +__all__ = ["AnthropicBatchesHandler", "AnthropicBatchesConfig"] + diff --git a/litellm/llms/anthropic/batches/handler.py b/litellm/llms/anthropic/batches/handler.py new file mode 100644 index 0000000000..3061abda96 --- /dev/null +++ b/litellm/llms/anthropic/batches/handler.py @@ -0,0 +1,169 @@ +""" +Anthropic Batches API Handler +""" + +import asyncio +from typing import TYPE_CHECKING, Any, Coroutine, Optional, Union + +import httpx + +from litellm.llms.custom_httpx.http_handler import ( + get_async_httpx_client, +) +from litellm.types.llms.openai import RetrieveBatchRequest +from litellm.types.utils import LiteLLMBatch + +if TYPE_CHECKING: + from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +else: + LiteLLMLoggingObj = Any + +from ..common_utils import AnthropicModelInfo +from .transformation import AnthropicBatchesConfig + + +class AnthropicBatchesHandler: + """ + Handler for Anthropic Message Batches API. + + Supports: + - retrieve_batch() - Retrieve batch status and information + """ + + def __init__(self): + self.anthropic_model_info = AnthropicModelInfo() + self.provider_config = AnthropicBatchesConfig() + + async def aretrieve_batch( + self, + batch_id: str, + api_base: Optional[str], + api_key: Optional[str], + timeout: Union[float, httpx.Timeout], + max_retries: Optional[int], + logging_obj: Optional[LiteLLMLoggingObj] = None, + ) -> LiteLLMBatch: + """ + Async: Retrieve a batch from Anthropic. + + Args: + batch_id: The batch ID to retrieve + api_base: Anthropic API base URL + api_key: Anthropic API key + timeout: Request timeout + max_retries: Max retry attempts (unused for now) + logging_obj: Optional logging object + + Returns: + LiteLLMBatch: Batch information in OpenAI format + """ + # Resolve API credentials + api_base = api_base or self.anthropic_model_info.get_api_base(api_base) + api_key = api_key or self.anthropic_model_info.get_api_key() + + if not api_key: + raise ValueError("Missing Anthropic API Key") + + # Create a minimal logging object if not provided + if logging_obj is None: + from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObjClass + logging_obj = LiteLLMLoggingObjClass( + model="anthropic/unknown", + messages=[], + stream=False, + call_type="batch_retrieve", + start_time=None, + litellm_call_id=f"batch_retrieve_{batch_id}", + function_id="batch_retrieve", + ) + + # Get the complete URL for batch retrieval + retrieve_url = self.provider_config.get_retrieve_batch_url( + api_base=api_base, + batch_id=batch_id, + optional_params={}, + litellm_params={}, + ) + + # Validate environment and get headers + headers = self.provider_config.validate_environment( + headers={}, + model="", + messages=[], + optional_params={}, + litellm_params={}, + api_key=api_key, + api_base=api_base, + ) + + logging_obj.pre_call( + input=batch_id, + api_key=api_key, + additional_args={ + "api_base": retrieve_url, + "headers": headers, + "complete_input_dict": {}, + }, + ) + # Make the request + async_client = get_async_httpx_client(llm_provider="anthropic") + response = await async_client.get( + url=retrieve_url, + headers=headers + ) + response.raise_for_status() + + # Transform response to LiteLLM format + return self.provider_config.transform_retrieve_batch_response( + model=None, + raw_response=response, + logging_obj=logging_obj, + litellm_params={}, + ) + + def retrieve_batch( + self, + _is_async: bool, + batch_id: str, + api_base: Optional[str], + api_key: Optional[str], + timeout: Union[float, httpx.Timeout], + max_retries: Optional[int], + logging_obj: Optional[LiteLLMLoggingObj] = None, + ) -> Union[LiteLLMBatch, Coroutine[Any, Any, LiteLLMBatch]]: + """ + Retrieve a batch from Anthropic. + + Args: + _is_async: Whether to run asynchronously + batch_id: The batch ID to retrieve + api_base: Anthropic API base URL + api_key: Anthropic API key + timeout: Request timeout + max_retries: Max retry attempts (unused for now) + logging_obj: Optional logging object + + Returns: + LiteLLMBatch or Coroutine: Batch information in OpenAI format + """ + if _is_async: + return self.aretrieve_batch( + batch_id=batch_id, + api_base=api_base, + api_key=api_key, + timeout=timeout, + max_retries=max_retries, + logging_obj=logging_obj, + ) + else: + return asyncio.run( + self.aretrieve_batch( + batch_id=batch_id, + api_base=api_base, + api_key=api_key, + timeout=timeout, + max_retries=max_retries, + logging_obj=logging_obj, + ) + ) + diff --git a/litellm/llms/anthropic/batches/transformation.py b/litellm/llms/anthropic/batches/transformation.py index c20136894b..02c490f532 100644 --- a/litellm/llms/anthropic/batches/transformation.py +++ b/litellm/llms/anthropic/batches/transformation.py @@ -1,10 +1,14 @@ import json -from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast +import time +from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Union, cast -from httpx import Response +import httpx +from httpx import Headers, Response -from litellm.types.llms.openai import AllMessageValues -from litellm.types.utils import ModelResponse +from litellm.llms.base_llm.batches.transformation import BaseBatchesConfig +from litellm.llms.base_llm.chat.transformation import BaseLLMException +from litellm.types.llms.openai import AllMessageValues, CreateBatchRequest +from litellm.types.utils import LiteLLMBatch, LlmProviders, ModelResponse if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj @@ -14,11 +18,225 @@ else: LoggingClass = Any -class AnthropicBatchesConfig: +class AnthropicBatchesConfig(BaseBatchesConfig): def __init__(self): from ..chat.transformation import AnthropicConfig + from ..common_utils import AnthropicError, AnthropicModelInfo self.anthropic_chat_config = AnthropicConfig() # initialize once + self.anthropic_model_info = AnthropicModelInfo() + + @property + def custom_llm_provider(self) -> LlmProviders: + """Return the LLM provider type for this configuration.""" + return LlmProviders.ANTHROPIC + + def validate_environment( + self, + headers: dict, + model: str, + messages: List[AllMessageValues], + optional_params: dict, + litellm_params: dict, + api_key: Optional[str] = None, + api_base: Optional[str] = None, + ) -> dict: + """Validate and prepare environment-specific headers and parameters.""" + # Resolve api_key from environment if not provided + api_key = api_key or self.anthropic_model_info.get_api_key() + if api_key is None: + raise ValueError( + "Missing Anthropic API Key - A call is being made to anthropic but no key is set either in the environment variables or via params" + ) + _headers = { + "accept": "application/json", + "anthropic-version": "2023-06-01", + "content-type": "application/json", + "x-api-key": api_key, + } + # Add beta header for message batches + if "anthropic-beta" not in headers: + headers["anthropic-beta"] = "message-batches-2024-09-24" + headers.update(_headers) + return headers + + def get_complete_batch_url( + self, + api_base: Optional[str], + api_key: Optional[str], + model: str, + optional_params: Dict, + litellm_params: Dict, + data: CreateBatchRequest, + ) -> str: + """Get the complete URL for batch creation request.""" + api_base = api_base or self.anthropic_model_info.get_api_base(api_base) + if not api_base.endswith("/v1/messages/batches"): + api_base = f"{api_base.rstrip('/')}/v1/messages/batches" + return api_base + + def transform_create_batch_request( + self, + model: str, + create_batch_data: CreateBatchRequest, + optional_params: dict, + litellm_params: dict, + ) -> Union[bytes, str, Dict[str, Any]]: + """ + Transform the batch creation request to Anthropic format. + + Not currently implemented - placeholder to satisfy abstract base class. + """ + raise NotImplementedError("Batch creation not yet implemented for Anthropic") + + def transform_create_batch_response( + self, + model: Optional[str], + raw_response: httpx.Response, + logging_obj: LoggingClass, + litellm_params: dict, + ) -> LiteLLMBatch: + """ + Transform Anthropic MessageBatch creation response to LiteLLM format. + + Not currently implemented - placeholder to satisfy abstract base class. + """ + raise NotImplementedError("Batch creation not yet implemented for Anthropic") + + def get_retrieve_batch_url( + self, + api_base: Optional[str], + batch_id: str, + optional_params: Dict, + litellm_params: Dict, + ) -> str: + """ + Get the complete URL for batch retrieval request. + + Args: + api_base: Base API URL (optional, will use default if not provided) + batch_id: Batch ID to retrieve + optional_params: Optional parameters + litellm_params: LiteLLM parameters + + Returns: + Complete URL for Anthropic batch retrieval: {api_base}/v1/messages/batches/{batch_id} + """ + api_base = api_base or self.anthropic_model_info.get_api_base(api_base) + return f"{api_base.rstrip('/')}/v1/messages/batches/{batch_id}" + + def transform_retrieve_batch_request( + self, + batch_id: str, + optional_params: dict, + litellm_params: dict, + ) -> Union[bytes, str, Dict[str, Any]]: + """ + Transform batch retrieval request for Anthropic. + + For Anthropic, the URL is constructed by get_retrieve_batch_url(), + so this method returns an empty dict (no additional request params needed). + """ + # No additional request params needed - URL is handled by get_retrieve_batch_url + return {} + + def transform_retrieve_batch_response( + self, + model: Optional[str], + raw_response: httpx.Response, + logging_obj: LoggingClass, + litellm_params: dict, + ) -> LiteLLMBatch: + """Transform Anthropic MessageBatch retrieval response to LiteLLM format.""" + try: + response_data = raw_response.json() + except Exception as e: + raise ValueError(f"Failed to parse Anthropic batch response: {e}") + + # Map Anthropic MessageBatch to OpenAI Batch format + batch_id = response_data.get("id", "") + processing_status = response_data.get("processing_status", "in_progress") + + # Map Anthropic processing_status to OpenAI status + status_mapping: Dict[str, Literal["validating", "failed", "in_progress", "finalizing", "completed", "expired", "cancelling", "cancelled"]] = { + "in_progress": "in_progress", + "canceling": "cancelling", + "ended": "completed", + } + openai_status = status_mapping.get(processing_status, "in_progress") + + # Parse timestamps + def parse_timestamp(ts_str: Optional[str]) -> Optional[int]: + if not ts_str: + return None + try: + from datetime import datetime + dt = datetime.fromisoformat(ts_str.replace('Z', '+00:00')) + return int(dt.timestamp()) + except Exception: + return None + + created_at = parse_timestamp(response_data.get("created_at")) + ended_at = parse_timestamp(response_data.get("ended_at")) + expires_at = parse_timestamp(response_data.get("expires_at")) + cancel_initiated_at = parse_timestamp(response_data.get("cancel_initiated_at")) + archived_at = parse_timestamp(response_data.get("archived_at")) + + # Extract request counts + request_counts_data = response_data.get("request_counts", {}) + from openai.types.batch import BatchRequestCounts + request_counts = BatchRequestCounts( + total=sum([ + request_counts_data.get("processing", 0), + request_counts_data.get("succeeded", 0), + request_counts_data.get("errored", 0), + request_counts_data.get("canceled", 0), + request_counts_data.get("expired", 0), + ]), + completed=request_counts_data.get("succeeded", 0), + failed=request_counts_data.get("errored", 0), + ) + + # Extract results_url - this will be used for file content retrieval + results_url = response_data.get("results_url") + # Store results_url in output_file_id for later retrieval + # We'll encode it in a way that we can detect it's an Anthropic results URL + output_file_id = None + if results_url: + # Encode the batch_id and results_url so we can retrieve it later + # Format: anthropic_batch_results:{batch_id} + output_file_id = f"anthropic_batch_results:{batch_id}" + + return LiteLLMBatch( + id=batch_id, + object="batch", + endpoint="/v1/messages", + errors=None, + input_file_id=None, + completion_window="24h", + status=openai_status, + output_file_id=output_file_id, + error_file_id=None, + created_at=created_at or int(time.time()), + in_progress_at=created_at if processing_status == "in_progress" else None, + expires_at=expires_at, + finalizing_at=None, + completed_at=ended_at if processing_status == "ended" else None, + failed_at=None, + expired_at=archived_at if archived_at else None, + cancelling_at=cancel_initiated_at if processing_status == "canceling" else None, + cancelled_at=ended_at if processing_status == "canceling" and ended_at else None, + request_counts=request_counts, + metadata={}, + ) + + def get_error_class( + self, error_message: str, status_code: int, headers: Union[Dict, Headers] + ) -> "BaseLLMException": + """Get the appropriate error class for Anthropic.""" + from ..common_utils import AnthropicError + + return AnthropicError(status_code=status_code, message=error_message, headers=headers) def transform_response( self, diff --git a/litellm/llms/anthropic/files/__init__.py b/litellm/llms/anthropic/files/__init__.py new file mode 100644 index 0000000000..b8b538ffb6 --- /dev/null +++ b/litellm/llms/anthropic/files/__init__.py @@ -0,0 +1,4 @@ +from .handler import AnthropicFilesHandler + +__all__ = ["AnthropicFilesHandler"] + diff --git a/litellm/llms/anthropic/files/handler.py b/litellm/llms/anthropic/files/handler.py new file mode 100644 index 0000000000..c45868f3bd --- /dev/null +++ b/litellm/llms/anthropic/files/handler.py @@ -0,0 +1,142 @@ +import asyncio +from typing import Any, Coroutine, Optional, Union + +import httpx + +from litellm.llms.custom_httpx.http_handler import ( + AsyncHTTPHandler, + HTTPHandler, + _get_httpx_client, + get_async_httpx_client, +) +from litellm.types.llms.openai import ( + FileContentRequest, + HttpxBinaryResponseContent, +) + +from ..common_utils import AnthropicModelInfo + + +class AnthropicFilesHandler: + """ + Handles Anthropic Files API operations. + + Currently supports: + - file_content() for retrieving Anthropic Message Batch results + """ + + def __init__(self): + self.anthropic_model_info = AnthropicModelInfo() + + async def afile_content( + self, + file_content_request: FileContentRequest, + api_base: Optional[str] = None, + api_key: Optional[str] = None, + timeout: Union[float, httpx.Timeout] = 600.0, + max_retries: Optional[int] = None, + ) -> HttpxBinaryResponseContent: + """ + Async: Retrieve file content from Anthropic. + + For batch results, the file_id should be the batch_id. + This will call Anthropic's /v1/messages/batches/{batch_id}/results endpoint. + + Args: + file_content_request: Contains file_id (batch_id for batch results) + api_base: Anthropic API base URL + api_key: Anthropic API key + timeout: Request timeout + max_retries: Max retry attempts (unused for now) + + Returns: + HttpxBinaryResponseContent: Binary content wrapped in compatible response format + """ + file_id = file_content_request.get("file_id") + if not file_id: + raise ValueError("file_id is required in file_content_request") + + # Extract batch_id from file_id + # Handle both formats: "anthropic_batch_results:{batch_id}" or just "{batch_id}" + if file_id.startswith("anthropic_batch_results:"): + batch_id = file_id.replace("anthropic_batch_results:", "", 1) + else: + batch_id = file_id + + # Get Anthropic API credentials + api_base = self.anthropic_model_info.get_api_base(api_base) + api_key = api_key or self.anthropic_model_info.get_api_key() + + if not api_key: + raise ValueError("Missing Anthropic API Key") + + # Construct the Anthropic batch results URL + results_url = f"{api_base.rstrip('/')}/v1/messages/batches/{batch_id}/results" + + # Prepare headers + headers = { + "accept": "application/json", + "anthropic-version": "2023-06-01", + "x-api-key": api_key, + } + + # Make the request to Anthropic + async_client = get_async_httpx_client(llm_provider="anthropic") + try: + anthropic_response = await async_client.get( + url=results_url, + headers=headers + ) + anthropic_response.raise_for_status() + + # Return the response content + return HttpxBinaryResponseContent(response=anthropic_response) + finally: + await async_client.aclose() + + def file_content( + self, + _is_async: bool, + file_content_request: FileContentRequest, + api_base: Optional[str] = None, + api_key: Optional[str] = None, + timeout: Union[float, httpx.Timeout] = 600.0, + max_retries: Optional[int] = None, + ) -> Union[ + HttpxBinaryResponseContent, Coroutine[Any, Any, HttpxBinaryResponseContent] + ]: + """ + Retrieve file content from Anthropic. + + For batch results, the file_id should be the batch_id. + This will call Anthropic's /v1/messages/batches/{batch_id}/results endpoint. + + Args: + _is_async: Whether to run asynchronously + file_content_request: Contains file_id (batch_id for batch results) + api_base: Anthropic API base URL + api_key: Anthropic API key + timeout: Request timeout + max_retries: Max retry attempts (unused for now) + + Returns: + HttpxBinaryResponseContent or Coroutine: Binary content wrapped in compatible response format + """ + if _is_async: + return self.afile_content( + file_content_request=file_content_request, + api_base=api_base, + api_key=api_key, + max_retries=max_retries, + ) + else: + return asyncio.run( + self.afile_content( + file_content_request=file_content_request, + api_base=api_base, + api_key=api_key, + timeout=timeout, + max_retries=max_retries, + ) + ) +