diff --git a/litellm/llms/anthropic/batches/transformation.py b/litellm/llms/anthropic/batches/transformation.py index 02c490f5326..ec82e6cdaeb 100644 --- a/litellm/llms/anthropic/batches/transformation.py +++ b/litellm/llms/anthropic/batches/transformation.py @@ -197,25 +197,15 @@ class AnthropicBatchesConfig(BaseBatchesConfig): 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, + input_file_id="None", completion_window="24h", status=openai_status, - output_file_id=output_file_id, + output_file_id=batch_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, @@ -236,7 +226,13 @@ class AnthropicBatchesConfig(BaseBatchesConfig): """Get the appropriate error class for Anthropic.""" from ..common_utils import AnthropicError - return AnthropicError(status_code=status_code, message=error_message, headers=headers) + # Convert Dict to Headers if needed + if isinstance(headers, dict): + headers_obj: Optional[Headers] = Headers(headers) + else: + headers_obj = headers if isinstance(headers, Headers) else None + + return AnthropicError(status_code=status_code, message=error_message, headers=headers_obj) def transform_response( self, diff --git a/litellm/llms/anthropic/files/handler.py b/litellm/llms/anthropic/files/handler.py index c45868f3bdf..aecd66f15ff 100644 --- a/litellm/llms/anthropic/files/handler.py +++ b/litellm/llms/anthropic/files/handler.py @@ -1,21 +1,41 @@ import asyncio -from typing import Any, Coroutine, Optional, Union +import json +import time +from typing import Any, Coroutine, Dict, List, Optional, Union import httpx +import litellm +from litellm._logging import verbose_logger +from litellm._uuid import uuid from litellm.llms.custom_httpx.http_handler import ( - AsyncHTTPHandler, - HTTPHandler, - _get_httpx_client, get_async_httpx_client, ) +from litellm.litellm_core_utils.litellm_logging import Logging from litellm.types.llms.openai import ( FileContentRequest, HttpxBinaryResponseContent, + OpenAIBatchResult, + OpenAIChatCompletionResponse, + OpenAIErrorBody, ) +from litellm.types.utils import CallTypes, LlmProviders, ModelResponse +from ..chat.transformation import AnthropicConfig from ..common_utils import AnthropicModelInfo +# Map Anthropic error types to HTTP status codes +ANTHROPIC_ERROR_STATUS_CODE_MAP = { + "invalid_request_error": 400, + "authentication_error": 401, + "permission_error": 403, + "not_found_error": 404, + "rate_limit_error": 429, + "api_error": 500, + "overloaded_error": 503, + "timeout_error": 504, +} + class AnthropicFilesHandler: """ @@ -81,18 +101,29 @@ class AnthropicFilesHandler: } # 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() + async_client = get_async_httpx_client(llm_provider=LlmProviders.ANTHROPIC) + anthropic_response = await async_client.get( + url=results_url, + headers=headers + ) + anthropic_response.raise_for_status() + + # Transform Anthropic batch results to OpenAI format + transformed_content = self._transform_anthropic_batch_results_to_openai_format( + anthropic_response.content + ) + + # Create a new response with transformed content + transformed_response = httpx.Response( + status_code=anthropic_response.status_code, + headers=anthropic_response.headers, + content=transformed_content, + request=anthropic_response.request, + ) + + # Return the transformed response content + return HttpxBinaryResponseContent(response=transformed_response) - # Return the response content - return HttpxBinaryResponseContent(response=anthropic_response) - finally: - await async_client.aclose() def file_content( self, @@ -140,3 +171,197 @@ class AnthropicFilesHandler: ) ) + def _transform_anthropic_batch_results_to_openai_format( + self, anthropic_content: bytes + ) -> bytes: + """ + Transform Anthropic batch results JSONL to OpenAI batch results JSONL format. + + Anthropic format: + { + "custom_id": "...", + "result": { + "type": "succeeded", + "message": { ... } // Anthropic message format + } + } + + OpenAI format: + { + "custom_id": "...", + "response": { + "status_code": 200, + "request_id": "...", + "body": { ... } // OpenAI chat completion format + } + } + """ + try: + anthropic_config = AnthropicConfig() + transformed_lines = [] + + # Parse JSONL content + content_str = anthropic_content.decode("utf-8") + for line in content_str.strip().split("\n"): + if not line.strip(): + continue + + anthropic_result = json.loads(line) + custom_id = anthropic_result.get("custom_id", "") + result = anthropic_result.get("result", {}) + result_type = result.get("type", "") + + # Transform based on result type + if result_type == "succeeded": + # Transform Anthropic message to OpenAI format + anthropic_message = result.get("message", {}) + if anthropic_message: + openai_response_body = self._transform_anthropic_message_to_openai_format( + anthropic_message=anthropic_message, + anthropic_config=anthropic_config, + ) + + # Create OpenAI batch result format + openai_result: OpenAIBatchResult = { + "custom_id": custom_id, + "response": { + "status_code": 200, + "request_id": anthropic_message.get("id", ""), + "body": openai_response_body, + }, + } + transformed_lines.append(json.dumps(openai_result)) + elif result_type == "errored": + # Handle error case + error = result.get("error", {}) + error_obj = error.get("error", {}) + error_message = error_obj.get("message", "Unknown error") + error_type = error_obj.get("type", "api_error") + + status_code = ANTHROPIC_ERROR_STATUS_CODE_MAP.get(error_type, 500) + + error_body_errored: OpenAIErrorBody = { + "error": { + "message": error_message, + "type": error_type, + } + } + openai_result_errored: OpenAIBatchResult = { + "custom_id": custom_id, + "response": { + "status_code": status_code, + "request_id": error.get("request_id", ""), + "body": error_body_errored, + }, + } + transformed_lines.append(json.dumps(openai_result_errored)) + elif result_type in ["canceled", "expired"]: + # Handle canceled/expired cases + error_body_canceled: OpenAIErrorBody = { + "error": { + "message": f"Batch request was {result_type}", + "type": "invalid_request_error", + } + } + openai_result_canceled: OpenAIBatchResult = { + "custom_id": custom_id, + "response": { + "status_code": 400, + "request_id": "", + "body": error_body_canceled, + }, + } + transformed_lines.append(json.dumps(openai_result_canceled)) + + # Join lines and encode back to bytes + transformed_content = "\n".join(transformed_lines) + if transformed_lines: + transformed_content += "\n" # Add trailing newline for JSONL format + return transformed_content.encode("utf-8") + except Exception as e: + verbose_logger.error( + f"Error transforming Anthropic batch results to OpenAI format: {e}" + ) + # Return original content if transformation fails + return anthropic_content + + def _transform_anthropic_message_to_openai_format( + self, anthropic_message: dict, anthropic_config: AnthropicConfig + ) -> OpenAIChatCompletionResponse: + """ + Transform a single Anthropic message to OpenAI chat completion format. + """ + try: + # Create a mock httpx.Response for transformation + mock_response = httpx.Response( + status_code=200, + content=json.dumps(anthropic_message).encode("utf-8"), + ) + + # Create a ModelResponse object + model_response = ModelResponse() + # Initialize with required fields - will be populated by transform_parsed_response + model_response.choices = [ + litellm.Choices( + finish_reason="stop", + index=0, + message=litellm.Message(content="", role="assistant"), + ) + ] # type: ignore + + # Create a logging object for transformation + logging_obj = Logging( + model=anthropic_message.get("model", "claude-3-5-sonnet-20241022"), + 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": {}}, + ) + logging_obj.optional_params = {} + + # Transform using AnthropicConfig + transformed_response = anthropic_config.transform_parsed_response( + completion_response=anthropic_message, + raw_response=mock_response, + model_response=model_response, + json_mode=False, + prefix_prompt=None, + ) + + # Convert ModelResponse to OpenAI format dict - it's already in OpenAI format + openai_body: OpenAIChatCompletionResponse = transformed_response.model_dump(exclude_none=True) + + # Ensure id comes from anthropic_message if not set + if not openai_body.get("id"): + openai_body["id"] = anthropic_message.get("id", "") + + return openai_body + except Exception as e: + verbose_logger.error( + f"Error transforming Anthropic message to OpenAI format: {e}" + ) + # Return a basic error response if transformation fails + error_response: OpenAIChatCompletionResponse = { + "id": anthropic_message.get("id", ""), + "object": "chat.completion", + "created": int(time.time()), + "model": anthropic_message.get("model", ""), + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": ""}, + "finish_reason": "error", + } + ], + "usage": { + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + }, + } + return error_response + diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py index 59397a62a8b..d0e4bbf4a42 100644 --- a/litellm/types/llms/openai.py +++ b/litellm/types/llms/openai.py @@ -437,10 +437,12 @@ class ListBatchRequest(TypedDict, total=False): """ after: Union[str, NotGiven] - limit: Union[int, NotGiven] - extra_headers: Optional[Dict[str, str]] - extra_body: Optional[Dict[str, str]] - timeout: Optional[float] + + +# OpenAI Batch Result Types +class OpenAIErrorBody(TypedDict, total=False): + """Error body in OpenAI batch response format.""" + error: Dict[str, str] BatchJobStatus = Literal[ @@ -1824,6 +1826,20 @@ class OpenAIChatCompletionResponse(TypedDict, total=False): service_tier: str +# OpenAI Batch Result Types (defined after OpenAIChatCompletionResponse for forward reference) +class OpenAIBatchResponse(TypedDict, total=False): + """Response wrapper in OpenAI batch result format.""" + status_code: int + request_id: str + body: Union[OpenAIChatCompletionResponse, OpenAIErrorBody] + + +class OpenAIBatchResult(TypedDict, total=False): + """OpenAI batch result format.""" + custom_id: str + response: OpenAIBatchResponse + + OpenAIChatCompletionFinishReason = Literal[ "stop", "content_filter", "function_call", "tool_calls", "length" ]