feat(batches): store each batch line's request and response in S3 and GCS

enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py

@@ -735,9 +735,10 @@ async def _track_completed_batch_cost(
 
         """
         from litellm.batches.batch_utils import (
-            count_error_file_failed_requests,
             get_file_content_as_dictionary,
             calculate_batch_cost_and_usage,
+            count_error_file_failed_requests_from_content,
+            fetch_batch_error_file_content,
         )
         from litellm.files.main import afile_content
         from litellm.files.types import FileContentCallOptions
@@ -874,14 +875,15 @@ async def _track_completed_batch_cost(
             model_name=model_name,
             model_info=deployment_model_info,
         )
-        error_file_failed_requests: Final = await count_error_file_failed_requests(
+        error_file_content = await fetch_batch_error_file_content(
             response,
             custom_llm_provider=batch_file_provider,
             litellm_params={
                 **credentials,
                 "_litellm_internal_model_credentials": MappingProxyType(dict(credentials)),
             },
         )
+        error_file_failed_requests: Final = count_error_file_failed_requests_from_content(error_file_content)
         batch_result: Final = (
             output_file_result
             if not error_file_failed_requests
@@ -921,6 +923,7 @@ async def _track_completed_batch_cost(
             optional_params={},
             custom_llm_provider=str(llm_provider) if llm_provider else None,
         )
+        logging_obj._litellm_internal_model_credentials = MappingProxyType(dict(credentials))  # pyright: ignore[reportPrivateUsage]  # trusted credentials transport, consumed by batch_line_item_logging
 
         if not await self._claim_job_for_costing(job):
             verbose_proxy_logger.info(
@@ -939,6 +942,8 @@ async def _track_completed_batch_cost(
                 batch_failed_requests=batch_result.failed_requests,
                 batch_prompt_cost=batch_result.prompt_cost,
                 batch_completion_cost=batch_result.completion_cost,
+                batch_output_file_content=content_bytes,
+                batch_error_file_content=error_file_content,
             )
         except Exception:
             await self._release_job_claim(job)

litellm/__init__.py

@@ -346,6 +346,7 @@ def _dev_env_hot_reload_enabled() -> bool:
 )
 openai_system_messages_first: bool = False
 disable_vertex_batch_output_transformation: bool = False
+store_batch_line_items_in_callbacks: bool = False
 extra_spend_tag_headers: Optional[List[str]] = None
 in_memory_llm_clients_cache: "LLMClientCache"
 safe_memory_mode: bool = False