diff --git a/litellm/files/main.py b/litellm/files/main.py index 1d5da29fe6f..cdb7e949a9c 100644 --- a/litellm/files/main.py +++ b/litellm/files/main.py @@ -58,6 +58,7 @@ from litellm.types.llms.openai import ( ) from litellm.types.router import * from litellm.types.utils import ( + FILE_CONTENT_STREAMING_PROVIDERS, OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS, LlmProviders, ) @@ -79,7 +80,22 @@ def _should_sdk_support_streaming( """ Return whether file content streaming is supported for the provider. """ - return custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS + return custom_llm_provider in FILE_CONTENT_STREAMING_PROVIDERS + + +def _file_content_logging_obj(kwargs: dict[str, object], _is_async: bool) -> LiteLLMLoggingObj: + logging_obj: Final = kwargs.get("litellm_logging_obj") + if isinstance(logging_obj, LiteLLMLoggingObj): + return logging_obj + return LiteLLMLoggingObj( + model="", + messages=[], + stream=False, + call_type="afile_content" if _is_async else "file_content", + start_time=time.time(), + litellm_call_id=str(kwargs.get("litellm_call_id") or uuid_module.uuid4()), + function_id=str(kwargs.get("id") or ""), + ) openai_files_instance: Final = OpenAIFilesAPI() @@ -868,18 +884,21 @@ def file_content( ) _is_async: Final = kwargs.pop("afile_content", False) is True + litellm_params_dict["api_key"] = optional_params.api_key + litellm_params_dict["api_base"] = optional_params.api_base if stream and _should_sdk_support_streaming(custom_llm_provider): return file_content_streaming( file_id=file_id, model=model, custom_llm_provider=custom_llm_provider, + file_content_request=_file_content_request, extra_headers=extra_headers, - extra_body=extra_body, chunk_size=chunk_size, optional_params=optional_params, + litellm_params=litellm_params_dict, timeout=timeout, - logging_obj=cast(LiteLLMLoggingObj | None, kwargs.get("litellm_logging_obj")), + logging_obj=_file_content_logging_obj(kwargs, _is_async), _is_async=_is_async, client=client, ) @@ -890,27 +909,12 @@ def file_content( provider=LlmProviders(custom_llm_provider), ) if provider_config is not None: - litellm_params_dict["api_key"] = optional_params.api_key - litellm_params_dict["api_base"] = optional_params.api_base - - logging_obj = kwargs.get("litellm_logging_obj") - if logging_obj is None: - logging_obj = LiteLLMLoggingObj( - model="", - messages=[], - stream=False, - call_type="afile_content" if _is_async else "file_content", - start_time=time.time(), - litellm_call_id=kwargs.get("litellm_call_id", str(uuid_module.uuid4())), - function_id=str(kwargs.get("id") or ""), - ) - response = base_llm_http_handler.retrieve_file_content( file_content_request=_file_content_request, provider_config=provider_config, litellm_params=litellm_params_dict, headers=extra_headers or {}, - logging_obj=logging_obj, + logging_obj=_file_content_logging_obj(kwargs, _is_async), _is_async=_is_async, client=(client if client is not None and isinstance(client, (HTTPHandler, AsyncHTTPHandler)) else None), timeout=timeout, @@ -1000,24 +1004,24 @@ def file_content_streaming( file_id: str, model: str | None, custom_llm_provider: FileContentProvider | str | None, + file_content_request: FileContentRequest, extra_headers: dict[str, str] | None, - extra_body: dict[str, str] | None, chunk_size: int, optional_params: GenericLiteLLMParams, + litellm_params: dict, timeout: float | httpx.Timeout, - logging_obj: LiteLLMLoggingObj | None, + logging_obj: LiteLLMLoggingObj, _is_async: bool, - client: OpenAI | AsyncOpenAI | None, + client: OpenAI | AsyncOpenAI | HTTPHandler | AsyncHTTPHandler | None, ) -> FileContentStreamingResult | Coroutine[object, object, FileContentStreamingResult]: - if logging_obj is not None: - logging_obj.model = model or "" - logging_obj.model_call_details["model"] = model or "" - logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider + logging_obj.model = model or "" + logging_obj.model_call_details["model"] = model or "" + logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider - litellm_params: Final = logging_obj.model_call_details.get("litellm_params", {}) or {} - if optional_params.api_base is not None: - litellm_params["api_base"] = optional_params.api_base - logging_obj.model_call_details["litellm_params"] = litellm_params + logged_litellm_params: Final = logging_obj.model_call_details.get("litellm_params", {}) or {} + if optional_params.api_base is not None: + logged_litellm_params["api_base"] = optional_params.api_base + logging_obj.model_call_details["litellm_params"] = logged_litellm_params def _wrap_streaming_result( response: FileContentStreamingResult, @@ -1044,22 +1048,45 @@ def file_content_streaming( ) response = openai_files_instance.file_content_streaming( _is_async=_is_async, - file_content_request=FileContentRequest( - file_id=file_id, - extra_headers=extra_headers, - extra_body=extra_body, - ), + file_content_request=file_content_request, api_base=openai_creds.api_base, api_key=openai_creds.api_key, timeout=timeout, max_retries=optional_params.max_retries, organization=openai_creds.organization, chunk_size=chunk_size, - client=client, + client=client if isinstance(client, (OpenAI, AsyncOpenAI)) else None, + ) + elif custom_llm_provider == LlmProviders.VERTEX_AI.value: + if not _is_async: + raise litellm.exceptions.BadRequestError( + message="Streaming 'file_content' for vertex_ai is only supported through 'afile_content'.", + model="n/a", + llm_provider=custom_llm_provider, + response=httpx.Response( + status_code=400, + content="Unsupported provider", + request=httpx.Request(method="file_content", url="https://github.com/BerriAI/litellm"), + ), + ) + vertex_files_config: Final = ProviderConfigManager.get_provider_files_config( + model="", + provider=LlmProviders.VERTEX_AI, + ) + assert vertex_files_config is not None + response = base_llm_http_handler.async_retrieve_file_content_streaming( + file_content_request=file_content_request, + provider_config=vertex_files_config, + litellm_params=litellm_params, + headers=extra_headers or {}, + logging_obj=logging_obj, + chunk_size=chunk_size, + client=client if isinstance(client, AsyncHTTPHandler) else None, + timeout=timeout, ) else: raise litellm.exceptions.BadRequestError( - message=f"LiteLLM doesn't support {custom_llm_provider} for streaming 'file_content'. Supported providers are {sorted(OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS)}.", + message=f"LiteLLM doesn't support {custom_llm_provider} for streaming 'file_content'. Supported providers are {sorted(FILE_CONTENT_STREAMING_PROVIDERS)}.", model="n/a", llm_provider=custom_llm_provider, response=httpx.Response( diff --git a/litellm/files/types.py b/litellm/files/types.py index b4ec9996f37..bcb752237fa 100644 --- a/litellm/files/types.py +++ b/litellm/files/types.py @@ -1,4 +1,4 @@ -from collections.abc import AsyncIterator, Iterator +from collections.abc import AsyncIterator, Iterator, Mapping from typing import Literal, NamedTuple FileContentProvider = Literal[ @@ -8,4 +8,4 @@ FileContentProvider = Literal[ class FileContentStreamingResult(NamedTuple): stream_iterator: Iterator[bytes] | AsyncIterator[bytes] - headers: dict[str, str] + headers: Mapping[str, str] diff --git a/litellm/llms/base_llm/files/transformation.py b/litellm/llms/base_llm/files/transformation.py index 6d16a1cea69..254995c028f 100644 --- a/litellm/llms/base_llm/files/transformation.py +++ b/litellm/llms/base_llm/files/transformation.py @@ -1,10 +1,11 @@ from abc import ABC, abstractmethod -from collections.abc import Iterator, Mapping +from collections.abc import AsyncGenerator, Iterator, Mapping from typing import TYPE_CHECKING, Any, Union import httpx from openai.types.file_deleted import FileDeleted +from litellm.files.types import FileContentStreamingResult from litellm.proxy._types import UserAPIKeyAuth from litellm.types.files import TwoStepFileUploadConfig from litellm.types.llms.openai import ( @@ -196,6 +197,18 @@ class BaseFilesConfig(BaseConfig): ) -> "HttpxBinaryResponseContent": """Transform file content response into OpenAI format.""" + async def transform_file_content_stream( + self, + *, + stream_iterator: AsyncGenerator[bytes, None], + headers: Mapping[str, str], + request_url: str, + logging_obj: LiteLLMLoggingObj, + litellm_params: dict, + ) -> FileContentStreamingResult: + """Transform a streamed file content body. Passes the upstream bytes and headers through by default.""" + return FileContentStreamingResult(stream_iterator=stream_iterator, headers=headers) + def transform_request( self, model: str, diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index 857adf5b9f1..98fe0014386 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -1,14 +1,27 @@ import asyncio import json import ssl -from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping, Sequence +from collections.abc import AsyncGenerator, AsyncIterator, Coroutine, Iterator, Mapping, Sequence from contextlib import asynccontextmanager from functools import lru_cache from types import MappingProxyType, ModuleType -from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypedDict, TypeVar, Union, cast, get_type_hints +from typing import ( + TYPE_CHECKING, + Any, + Final, + Literal, + NamedTuple, + Optional, + TypedDict, + TypeVar, + Union, + cast, + get_type_hints, +) from urllib.parse import parse_qs, urlencode, urlparse, urlunparse import httpx +from httpx import USE_CLIENT_DEFAULT from httpx._types import FileContent from openai.types.file_deleted import FileDeleted @@ -19,6 +32,7 @@ import litellm.types.utils from litellm._logging import _redact_string, verbose_logger from litellm.anthropic_beta_headers_manager import update_headers_with_filtered_beta from litellm.constants import MAX_FILE_LIST_LIMIT, REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES +from litellm.files.types import FileContentStreamingResult from litellm.litellm_core_utils.agentic_loop_settings import ( DEFAULT_MAX_AGENTIC_LOOPS, validated_max_agentic_loops, @@ -288,6 +302,39 @@ def _aws_signing_overrides(optional_params: Mapping[str, Any], litellm_params: M ) +class _PreparedFileContentRequest(NamedTuple): + url: str + params: dict + headers: dict + + +async def _aiter_bytes_then_close(response: httpx.Response, *, chunk_size: int) -> AsyncGenerator[bytes, None]: + try: + async for chunk in response.aiter_bytes(chunk_size=chunk_size): + yield chunk + finally: + await response.aclose() + + +_DECODED_BODY_STALE_HEADERS: Final[frozenset[str]] = frozenset({"content-encoding", "content-length"}) + + +def _decoded_body_headers(response: httpx.Response) -> httpx.Headers: + """ + `aiter_bytes` yields the decoded body, so the upstream transfer headers only + describe the bytes on the wire when no content-encoding was applied. + """ + if response.headers.get("content-encoding", "identity").lower() == "identity": + return response.headers + return httpx.Headers( + [ + (name, value) + for name, value in response.headers.multi_items() + if name.lower() not in _DECODED_BODY_STALE_HEADERS + ] + ) + + def _collect_ws_project_quota_callbacks() -> tuple[ProjectQuotaCallback, ...]: """Duck-type discover proxy hooks exposing per-frame project ITPM/OTPM enforcement, so the Responses WebSocket loop can charge every @@ -5080,35 +5127,16 @@ class BaseLLMHTTPHandler: else: sync_httpx_client = client - # Get URL and params from provider config - url, params = provider_config.transform_file_content_request( + prepared: Final = self._prepare_file_content_request( file_content_request=file_content_request, - optional_params={}, + provider_config=provider_config, litellm_params=litellm_params, - ) - - # Validate environment and get headers - headers = provider_config.validate_environment( - api_key=litellm_params.get("api_key"), headers=headers, - model="", - messages=[], - optional_params={}, - litellm_params=litellm_params, - ) - - logging_obj.pre_call( - input="", - api_key="", - additional_args={ - "api_base": url, - "headers": headers, - "file_id": file_content_request.get("file_id"), - }, + logging_obj=logging_obj, ) try: - response: Final = sync_httpx_client.get(url=url, headers=headers, params=params) + response: Final = sync_httpx_client.get(url=prepared.url, headers=prepared.headers, params=prepared.params) except Exception as e: raise self._handle_error(e=e, provider_config=provider_config) @@ -5143,35 +5171,18 @@ class BaseLLMHTTPHandler: else: async_httpx_client = client - # Get URL and params from provider config - url, params = provider_config.transform_file_content_request( + prepared: Final = self._prepare_file_content_request( file_content_request=file_content_request, - optional_params={}, + provider_config=provider_config, litellm_params=litellm_params, - ) - - # Validate environment and get headers - headers = provider_config.validate_environment( - api_key=litellm_params.get("api_key"), headers=headers, - model="", - messages=[], - optional_params={}, - litellm_params=litellm_params, - ) - - logging_obj.pre_call( - input="", - api_key="", - additional_args={ - "api_base": url, - "headers": headers, - "file_id": file_content_request.get("file_id"), - }, + logging_obj=logging_obj, ) try: - response: Final = await async_httpx_client.get(url=url, headers=headers, params=params) + response: Final = await async_httpx_client.get( + url=prepared.url, headers=prepared.headers, params=prepared.params + ) except Exception as e: raise self._handle_error(e=e, provider_config=provider_config) @@ -5188,6 +5199,93 @@ class BaseLLMHTTPHandler: litellm_params=litellm_params, ) + async def async_retrieve_file_content_streaming( + self, + file_content_request: "FileContentRequest", + provider_config: BaseFilesConfig, + litellm_params: dict, + headers: dict, + logging_obj: LiteLLMLoggingObj, + chunk_size: int, + client: AsyncHTTPHandler | None = None, + timeout: float | httpx.Timeout | None = None, + ) -> FileContentStreamingResult: + """ + Async retrieve file content by ID as a byte stream, without buffering the body. + """ + async_httpx_client: Final = ( + client if client is not None else get_async_httpx_client(llm_provider=provider_config.custom_llm_provider) + ) + + prepared: Final = self._prepare_file_content_request( + file_content_request=file_content_request, + provider_config=provider_config, + litellm_params=litellm_params, + headers=headers, + logging_obj=logging_obj, + ) + + request: Final = async_httpx_client.client.build_request( + "GET", + prepared.url, + headers=prepared.headers, + params=httpx.QueryParams(HTTPHandler.extract_query_params(prepared.url)).merge(prepared.params), + timeout=USE_CLIENT_DEFAULT if timeout is None else httpx.Timeout(timeout), + ) + try: + response: Final = await async_httpx_client.client.send(request, stream=True) + except Exception as e: # noqa: BLE001 # _handle_error maps every failure kind, like the buffered fetch + raise self._handle_error(e=e, provider_config=provider_config) + + if response.status_code >= 400: + error_body: Final = await response.aread() + await response.aclose() + raise provider_config.get_error_class( + error_message=error_body.decode("utf-8", errors="replace"), + status_code=response.status_code, + headers=response.headers, + ) + + return await provider_config.transform_file_content_stream( + stream_iterator=_aiter_bytes_then_close(response, chunk_size=chunk_size), + headers=_decoded_body_headers(response), + request_url=str(response.request.url), + logging_obj=logging_obj, + litellm_params=litellm_params, + ) + + @staticmethod + def _prepare_file_content_request( + file_content_request: "FileContentRequest", + provider_config: BaseFilesConfig, + litellm_params: dict, + headers: dict, + logging_obj: LiteLLMLoggingObj, + ) -> "_PreparedFileContentRequest": + url, params = provider_config.transform_file_content_request( + file_content_request=file_content_request, + optional_params={}, + litellm_params=litellm_params, + ) + request_headers: Final = provider_config.validate_environment( + api_key=litellm_params.get("api_key"), + headers=headers, + model="", + messages=[], + optional_params={}, + litellm_params=litellm_params, + ) + logging_obj.pre_call( + input="", + api_key="", + additional_args={ + "api_base": url, + "headers": request_headers, + "file_id": file_content_request.get("file_id"), + }, + ) + return _PreparedFileContentRequest(url=url, params=params, headers=request_headers) + def _prepare_fake_stream_request( self, stream: bool, diff --git a/litellm/llms/vertex_ai/files/transformation.py b/litellm/llms/vertex_ai/files/transformation.py index 263956efc9f..85ec2911464 100644 --- a/litellm/llms/vertex_ai/files/transformation.py +++ b/litellm/llms/vertex_ai/files/transformation.py @@ -5,7 +5,10 @@ import json import os import re import time -from collections.abc import Callable, Iterable, Iterator, Mapping +from collections.abc import AsyncGenerator, Callable, Iterable, Iterator, Mapping +from contextlib import aclosing +from dataclasses import dataclass +from types import MappingProxyType from typing import Any, Final, TypedDict from urllib.parse import quote, unquote @@ -16,6 +19,7 @@ from typing_extensions import ReadOnly, Required import litellm from litellm._uuid import uuid +from litellm.files.types import FileContentStreamingResult from litellm.files.utils import FilesAPIUtils from litellm.litellm_core_utils.cloud_storage_security import ( VERTEX_AI_MANAGED_GCS_PREFIX, @@ -81,6 +85,8 @@ _EMBED_REQUEST_FIELD_BY_GEMINI_PARAM: Final = ( ("title", "title"), ) _VERTEX_BATCH_FANNED_OUT_KEY_PATTERN: Final = re.compile(r"(?P[^#]*)#(?P\d+)/(?P\d+)") +_JSONL_NEWLINE: Final = b"\n" +_BATCH_OUTPUT_FIRST_ROW_PEEK_LIMIT_BYTES: Final = 32 * 1024 * 1024 class _GcsObjectMetadataJson(TypedDict, total=False): @@ -257,6 +263,118 @@ def _is_vertex_embeddings_batch_output_row(vertex_output_row: Mapping[str, objec return bool(vertex_output_row.get("status")) and isinstance(request_data, dict) and "content" in request_data +def _is_vertex_generate_content_batch_output_row(vertex_output_row: Mapping[str, object]) -> bool: + """ + Whether a Vertex batch output row came from a `GenerateContentRequest`. Anything + else (a plain JSON line, an OpenAI batch row) is not a Vertex batch output. + """ + if not ( + "request" in vertex_output_row and "response" in vertex_output_row and "processed_time" in vertex_output_row + ): + return False + response: Final = vertex_output_row.get("response") + return (isinstance(response, dict) and ("candidates" in response or "promptFeedback" in response)) or bool( + vertex_output_row.get("status") + ) + + +def _try_parse_vertex_batch_output_row(line: bytes) -> _VertexBatchRow | None: + try: + row: Final = _parse_vertex_batch_output_row(line.decode("utf-8")) + except (UnicodeDecodeError, ValueError): + return None + return row if isinstance(row, dict) else None + + +def _first_non_empty_jsonl_line(lines: Iterable[bytes]) -> bytes | None: + return next((stripped for line in lines if (stripped := line.strip())), None) + + +async def _peek_first_jsonl_line( + chunks: AsyncGenerator[bytes, None], + *, + peek_limit_bytes: int, +) -> tuple[bytes | None, bytes]: + """ + Reads from `chunks` until the first non-empty line is complete, returning it with + everything read so far so the caller can replay the bytes. Stops peeking once the + buffered prefix exceeds `peek_limit_bytes` without a newline, so a large file that + is not JSONL is never buffered in full. + """ + buffered: bytes = b"" # rebind-ok: accumulates the prefix read while looking for the first newline + async for chunk in chunks: + buffered = buffered + chunk + first_line = _first_non_empty_jsonl_line(buffered.split(_JSONL_NEWLINE)[:-1]) + if first_line is not None: + return first_line, buffered + if len(buffered) > peek_limit_bytes: + return None, buffered + return _first_non_empty_jsonl_line(buffered.split(_JSONL_NEWLINE)), buffered + + +async def _prepend_bytes(prefix: bytes, chunks: AsyncGenerator[bytes, None]) -> AsyncGenerator[bytes, None]: + async with aclosing(chunks): + if prefix: + yield prefix + async for chunk in chunks: + yield chunk + + +async def _aiter_jsonl_lines(chunks: AsyncGenerator[bytes, None]) -> AsyncGenerator[bytes, None]: + """Yields stripped, non-empty JSONL lines from a byte stream, holding at most one partial line.""" + pending: bytes = b"" # rebind-ok: carries the partial trailing line over to the next chunk + async with aclosing(chunks): + async for chunk in chunks: + *complete_lines, pending = (pending + chunk).split(_JSONL_NEWLINE) + for line in complete_lines: + if stripped := line.strip(): + yield stripped + if tail := pending.strip(): + yield tail + + +async def _aiter_single_chunk(content: bytes) -> AsyncGenerator[bytes, None]: + yield content + + +async def _aread_all(chunks: AsyncGenerator[bytes, None]) -> bytes: + async with aclosing(chunks): + return b"".join(tuple([chunk async for chunk in chunks])) + + +def _headers_without_content_length(headers: Mapping[str, str]) -> Mapping[str, str]: + return MappingProxyType({key: value for key, value in headers.items() if key.lower() != "content-length"}) + + +@dataclass(frozen=True, slots=True) +class _VertexBatchOutputRowTransformContext: + vertex_gemini_config: VertexGeminiConfig + logging_obj: Logging + mock_httpx_response: httpx.Response + + +def _new_vertex_batch_output_row_transform_context() -> _VertexBatchOutputRowTransformContext: + batch_transform_logging_obj: Final = Logging( + model="", + messages=[], + stream=False, + call_type="batch_transform", + start_time=time.time(), + litellm_call_id="", + function_id="", + ) + batch_transform_logging_obj.optional_params = {} + return _VertexBatchOutputRowTransformContext( + vertex_gemini_config=VertexGeminiConfig(), + logging_obj=batch_transform_logging_obj, + mock_httpx_response=httpx.Response( + status_code=200, + headers={"content-type": "application/json"}, + request=httpx.Request(method="POST", url="https://example.com"), + ), + ) + + def _openai_batch_output_row( custom_id: str, body: Mapping[str, object] | None = None, @@ -1074,6 +1192,84 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig): return HttpxBinaryResponseContent(response=raw_response) + async def transform_file_content_stream( + self, + *, + stream_iterator: AsyncGenerator[bytes, None], + headers: Mapping[str, str], + request_url: str, + logging_obj: LiteLLMLoggingObj, + litellm_params: dict, + ) -> FileContentStreamingResult: + """ + Streams file content, converting a Vertex AI batch output to OpenAI format row by + row when the first row identifies one, so peak memory stays at about one row. + + Embeddings batch outputs are grouped by entry and so are transformed in full. + Everything else is passed through unchanged, including a row that fails to + transform mid-stream. + """ + if litellm.disable_vertex_batch_output_transformation: + return FileContentStreamingResult(stream_iterator=stream_iterator, headers=headers) + + first_line, buffered = await _peek_first_jsonl_line( + stream_iterator, + peek_limit_bytes=_BATCH_OUTPUT_FIRST_ROW_PEEK_LIMIT_BYTES, + ) + replayed_stream: Final = _prepend_bytes(buffered, stream_iterator) + first_row: Final = None if first_line is None else _try_parse_vertex_batch_output_row(first_line) + if first_row is None: + return FileContentStreamingResult(stream_iterator=replayed_stream, headers=headers) + + if _is_vertex_embeddings_batch_output_row(first_row): + transformed_content: Final = self._try_transform_vertex_batch_output_to_openai( + content=await _aread_all(replayed_stream), + logging_obj=logging_obj, + model=_model_from_managed_gcs_url(request_url), + ) + return FileContentStreamingResult( + stream_iterator=_aiter_single_chunk(transformed_content), + headers=MappingProxyType({**headers, "content-length": str(len(transformed_content))}), + ) + + if not _is_vertex_generate_content_batch_output_row(first_row): + return FileContentStreamingResult(stream_iterator=replayed_stream, headers=headers) + + return FileContentStreamingResult( + stream_iterator=self._aiter_openai_batch_output_rows(_aiter_jsonl_lines(replayed_stream)), + headers=_headers_without_content_length(headers), + ) + + async def _aiter_openai_batch_output_rows(self, lines: AsyncGenerator[bytes, None]) -> AsyncGenerator[bytes, None]: + context: Final = _new_vertex_batch_output_row_transform_context() + async with aclosing(lines): + first_line: Final = await anext(lines, None) + if first_line is None: + return + yield self._transform_vertex_batch_output_line(first_line, context=context) + async for line in lines: + yield _JSONL_NEWLINE + self._transform_vertex_batch_output_line(line, context=context) + + def _transform_vertex_batch_output_line( + self, + line: bytes, + *, + context: _VertexBatchOutputRowTransformContext, + ) -> bytes: + vertex_output: Final = _try_parse_vertex_batch_output_row(line) + if vertex_output is None: + return line + try: + openai_output: Final = self._transform_single_vertex_batch_output_to_openai( + vertex_output=vertex_output, + vertex_gemini_config=context.vertex_gemini_config, + logging_obj=context.logging_obj, + mock_httpx_response=context.mock_httpx_response, + ) + except Exception: # noqa: BLE001 # a row that fails to transform is passed through raw, like the buffered path + return line + return json.dumps(openai_output).encode("utf-8") + def _try_transform_vertex_batch_output_to_openai( self, content: bytes, @@ -1120,38 +1316,13 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig): # first line is not valid UTF-8/JSON) raises and falls through to the # passthrough below, leaving the content untouched. first_row: Final = _parse_vertex_batch_output_row(first_line) - is_vertex_batch_output: Final = _is_vertex_embeddings_batch_output_row(first_row) or ( - "request" in first_row - and "response" in first_row - and "processed_time" in first_row - and ( - "candidates" in first_row.get("response", {}) - or "promptFeedback" in first_row.get("response", {}) - or bool(first_row.get("status")) - ) - ) - if not is_vertex_batch_output: + if not ( + _is_vertex_embeddings_batch_output_row(first_row) + or _is_vertex_generate_content_batch_output_row(first_row) + ): return content - vertex_gemini_config: Final = VertexGeminiConfig() - # Use a fresh Logging object for the per-row transform so we never - # mutate the caller's (which already ran pre_call with its own - # model/start_time/optional_params). - batch_transform_logging_obj: Final = Logging( - model="", - messages=[], - stream=False, - call_type="batch_transform", - start_time=time.time(), - litellm_call_id="", - function_id="", - ) - batch_transform_logging_obj.optional_params = {} - mock_httpx_response: Final = httpx.Response( - status_code=200, - headers={"content-type": "application/json"}, - request=httpx.Request(method="POST", url="https://example.com"), - ) + context: Final = _new_vertex_batch_output_row_transform_context() all_lines = itertools.chain((first_line,), lines) @@ -1173,9 +1344,9 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig): try: openai_output = self._transform_single_vertex_batch_output_to_openai( vertex_output=_parse_vertex_batch_output_row(line), - vertex_gemini_config=vertex_gemini_config, - logging_obj=batch_transform_logging_obj, - mock_httpx_response=mock_httpx_response, + vertex_gemini_config=context.vertex_gemini_config, + logging_obj=context.logging_obj, + mock_httpx_response=context.mock_httpx_response, ) except Exception: return content diff --git a/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py b/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py index fdd984b8aa8..2381a5cc2db 100644 --- a/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py +++ b/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py @@ -5,7 +5,7 @@ from fastapi.responses import StreamingResponse import litellm from litellm.files.types import FileContentProvider, FileContentStreamingResult -from litellm.types.utils import OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS +from litellm.types.utils import FILE_CONTENT_STREAMING_PROVIDERS if TYPE_CHECKING: from litellm.proxy._types import UserAPIKeyAuth @@ -43,6 +43,7 @@ class FileContentStreamingHandler: data=resolved_streaming_data, credentials=credentials, file_id=original_file_id, + include_internal_credentials=True, ) resolved_streaming_data.pop("model", None) resolved_streaming_provider: Final = cast(str, credentials["custom_llm_provider"]) @@ -64,7 +65,7 @@ class FileContentStreamingHandler: *, custom_llm_provider: str, ) -> bool: - return custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS + return custom_llm_provider in FILE_CONTENT_STREAMING_PROVIDERS @staticmethod async def stream_file_content_with_logging( diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 748c91a4792..f48947dc8dc 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -4137,6 +4137,10 @@ OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS: set[str] = { LlmProviders.LITELLM_PROXY.value, } +FILE_CONTENT_STREAMING_PROVIDERS: Final[frozenset[str]] = frozenset( + {*OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS, LlmProviders.VERTEX_AI.value} +) + ListBatchesSupportedProvider = Literal["openai", "azure", "hosted_vllm", "litellm_proxy", "vertex_ai"] LIST_BATCHES_SUPPORTED_PROVIDERS: Final[frozenset[str]] = frozenset(get_args(ListBatchesSupportedProvider)) diff --git a/tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_files_streaming.py b/tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_files_streaming.py index 7383513fb96..b94ea1ea269 100644 --- a/tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_files_streaming.py +++ b/tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_files_streaming.py @@ -15,9 +15,15 @@ replaced by a list-based pipeline: 4. A tuple-wrapped file handle uploaded through the real create_file ordering keeps every row, including entry 0 (no partial upload from a consumed cursor). + 5. Downloading a GCS object through ``async_retrieve_file_content_streaming`` + yields the body as it arrives instead of buffering it, keeps the upstream + ``content-type`` / ``content-length``, transforms a Vertex batch output + row by row, and closes the response when the consumer is done. """ +import asyncio import gc +import gzip import io import json import tempfile @@ -27,20 +33,22 @@ import tracemalloc import httpx import pytest +import litellm +from litellm.files.types import FileContentStreamingResult from litellm.litellm_core_utils.litellm_logging import Logging from litellm.llms.base_llm.files.transformation import BaseFileUploadStream from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler +from litellm.llms.vertex_ai.common_utils import VertexAIError from litellm.llms.vertex_ai.files.transformation import ( VertexAIFilesConfig, - _OpenAIToVertexBatchUploadStream, _get_litellm_batch_custom_id_from_labels, _iter_openai_jsonl_entries, _iter_openai_jsonl_lines, _openai_batch_jsonl_entry_to_vertex_rows, + _OpenAIToVertexBatchUploadStream, ) -from litellm.types.llms.openai import CreateFileRequest -from litellm.llms.vertex_ai.common_utils import VertexAIError +from litellm.types.llms.openai import CreateFileRequest, FileContentRequest def _upload_stream(transformed) -> BaseFileUploadStream: @@ -586,3 +594,321 @@ class TestStreamingMediaUpload: monkeypatch.setattr(tempfile, "TemporaryFile", lambda *a, **k: (created.append(1), real_tempfile(*a, **k))[1]) await self._run(_make_openai_jsonl_bytes(50)) assert created == [] + + +_MANAGED_OUTPUT_FILE_ID = ( + "gs://test-bucket/litellm-vertex-files/publishers/google/models/gemini-2.5-flash/abc/predictions.jsonl" +) + + +def _vertex_batch_output_row(custom_id: str, text: str) -> bytes: + return json.dumps( + { + "status": "", + "processed_time": "2024-11-01T18:13:16.826+00:00", + "request": {"labels": {"litellm_custom_id": custom_id}, "contents": [{"parts": [{"text": "hi"}]}]}, + "response": { + "candidates": [{"content": {"parts": [{"text": text}], "role": "model"}, "finishReason": "STOP"}], + "usageMetadata": {"promptTokenCount": 1, "candidatesTokenCount": 2, "totalTokenCount": 3}, + "modelVersion": "gemini-2.5-flash@default", + }, + } + ).encode("utf-8") + + +def _vertex_embeddings_output_row(key: str, values: list[float]) -> bytes: + return json.dumps( + { + "key": key, + "request": {"content": {"parts": [{"text": "hello world"}]}}, + "response": {"embedding": {"values": values}, "usageMetadata": {"promptTokenCount": 2}}, + } + ).encode("utf-8") + + +def _gcs_download_mock(raw_chunks: list[bytes], headers: dict[str, str]): + """A fake GCS `alt=media` endpoint that serves the object one raw chunk at a + time, recording the request and how many chunks the consumer has pulled so + far, so a test can tell streaming apart from buffering.""" + state = {"urls": [], "headers": [], "served": 0, "closed": False} + + async def body(): + for chunk in raw_chunks: + state["served"] += 1 + yield chunk + await asyncio.sleep(0) + + async def handler(request: httpx.Request) -> httpx.Response: + state["urls"].append(str(request.url)) + state["headers"].append(dict(request.headers)) + response = httpx.Response(200, content=body(), headers=headers) + original_aclose = response.aclose + + async def aclose(): + state["closed"] = True + await original_aclose() + + response.aclose = aclose + return response + + return handler, state + + +class _StaticTokenFilesConfig(VertexAIFilesConfig): + """Vertex files config with a fixed access token, so no ADC lookup runs in tests.""" + + def get_access_token(self, credentials, project_id, _retry_reauth=False): + return "test-token", "test-project" + + +def _stable_row_fields(jsonl: bytes) -> list[tuple]: + """Project OpenAI batch output rows onto the fields the transform derives from + the Vertex row, leaving out the ids and timestamps it generates per call.""" + rows = [json.loads(line) for line in jsonl.split(b"\n") if line] + return [ + ( + row["custom_id"], + row["error"], + row["response"]["status_code"], + row["response"]["body"]["model"], + row["response"]["body"]["choices"][0]["message"]["content"], + row["response"]["body"]["usage"]["total_tokens"], + ) + for row in rows + ] + + +class TestFileContentStreaming: + """End-to-end against a faked GCS media endpoint. These fail if the retrieval + buffers the object before yielding, drops or duplicates bytes across chunk + boundaries, loses the upstream headers, or leaks the httpx response.""" + + async def _open(self, raw_chunks: list[bytes], headers: dict[str, str], chunk_size: int = 16): + mock, state = _gcs_download_mock(raw_chunks, headers) + result = await BaseLLMHTTPHandler().async_retrieve_file_content_streaming( + file_content_request=FileContentRequest(file_id=_MANAGED_OUTPUT_FILE_ID), + provider_config=_StaticTokenFilesConfig(), + litellm_params={"gcs_bucket_name": "test-bucket"}, + headers={}, + logging_obj=_logging_obj(), + chunk_size=chunk_size, + client=_async_handler_with(mock), + ) + return result, state + + async def test_plain_object_streams_through_with_upstream_headers(self): + raw = b'{"line": 1}\n{"line": 2}\n' * 40 + raw_chunks = [raw[i : i + 100] for i in range(0, len(raw), 100)] + upstream = {"content-type": "application/octet-stream", "content-length": str(len(raw))} + + result, state = await self._open(raw_chunks, upstream, chunk_size=7) + + assert state["urls"] == [ + "https://storage.googleapis.com/storage/v1/b/test-bucket/o/" + "litellm-vertex-files%2Fpublishers%2Fgoogle%2Fmodels%2Fgemini-2.5-flash%2Fabc%2Fpredictions.jsonl?alt=media" + ] + assert state["headers"][0]["authorization"] == "Bearer test-token" + assert result.headers["content-type"] == "application/octet-stream" + assert result.headers["content-length"] == str(len(raw)) + + received = [chunk async for chunk in result.stream_iterator] + assert b"".join(received) == raw + assert len(received) > 1 + assert state["closed"] is True + + async def test_body_is_yielded_before_the_object_is_fully_served(self): + raw_chunks = [b'{"line": %d}\n' % i for i in range(50)] + result, state = await self._open(raw_chunks, {"content-type": "application/octet-stream"}, chunk_size=8) + + first = await anext(result.stream_iterator) + + assert first + assert state["served"] < len(raw_chunks) + assert state["closed"] is False + + async def test_gzip_encoded_object_is_decoded_without_stale_transfer_headers(self): + raw = b'{"line": 1}\n{"line": 2}\n' * 200 + encoded = gzip.compress(raw) + upstream = { + "content-type": "application/octet-stream", + "content-encoding": "gzip", + "content-length": str(len(encoded)), + } + + result, state = await self._open([encoded[i : i + 64] for i in range(0, len(encoded), 64)], upstream) + streamed = b"".join([chunk async for chunk in result.stream_iterator]) + + assert streamed == raw + assert result.headers["content-type"] == "application/octet-stream" + assert "content-encoding" not in result.headers + assert "content-length" not in result.headers + assert state["closed"] is True + + async def test_vertex_batch_output_is_transformed_row_by_row(self): + rows = [_vertex_batch_output_row(f"request-{i}", f"answer {i}") for i in range(30)] + raw = b"\n".join(rows) + b"\n" + raw_chunks = [raw[i : i + 333] for i in range(0, len(raw), 333)] + expected = VertexAIFilesConfig()._try_transform_vertex_batch_output_to_openai( + content=raw, logging_obj=_logging_obj(), model="gemini-2.5-flash" + ) + assert expected != raw + + result, state = await self._open( + raw_chunks, + {"content-type": "application/octet-stream", "content-length": str(len(raw))}, + chunk_size=97, + ) + first = await anext(result.stream_iterator) + assert json.loads(first)["custom_id"] == "request-0" + assert state["served"] < len(raw_chunks) + + rest = [chunk async for chunk in result.stream_iterator] + streamed = b"".join([first, *rest]) + assert _stable_row_fields(streamed) == _stable_row_fields(expected) + assert len(_stable_row_fields(streamed)) == len(rows) + assert streamed.count(b"\n") == expected.count(b"\n") + assert len(rest) == len(rows) - 1 + assert result.headers["content-type"] == "application/octet-stream" + assert "content-length" not in result.headers + assert state["closed"] is True + + async def test_last_row_without_trailing_newline_and_unparseable_row_are_kept(self): + broken = b'{"custom_id": "request-1", "response": {"candidates": [}' + rows = [_vertex_batch_output_row("request-0", "first"), broken, _vertex_batch_output_row("request-2", "last")] + raw = b"\n".join(rows) + raw_chunks = [raw[i : i + 41] for i in range(0, len(raw), 41)] + + result, state = await self._open(raw_chunks, {}, chunk_size=29) + streamed_lines = b"".join([chunk async for chunk in result.stream_iterator]).split(b"\n") + + assert len(streamed_lines) == len(rows) + assert json.loads(streamed_lines[0])["custom_id"] == "request-0" + assert json.loads(streamed_lines[0])["response"]["body"]["choices"][0]["message"]["content"] == "first" + assert streamed_lines[1] == broken + assert json.loads(streamed_lines[2])["custom_id"] == "request-2" + assert json.loads(streamed_lines[2])["response"]["body"]["choices"][0]["message"]["content"] == "last" + assert state["closed"] is True + + async def test_transform_opt_out_streams_raw_batch_output(self, monkeypatch): + monkeypatch.setattr("litellm.disable_vertex_batch_output_transformation", True) + raw = b"\n".join(_vertex_batch_output_row(f"request-{i}", "x") for i in range(3)) + b"\n" + + result, _ = await self._open([raw], {"content-length": str(len(raw))}) + + assert b"".join([chunk async for chunk in result.stream_iterator]) == raw + assert result.headers["content-length"] == str(len(raw)) + + async def test_embeddings_batch_output_is_transformed_with_updated_content_length(self): + rows = [_vertex_embeddings_output_row(f"request-{i}", [0.1 * i, 0.2]) for i in range(3)] + raw = b"\n".join(rows) + b"\n" + raw_chunks = [raw[i : i + 50] for i in range(0, len(raw), 50)] + + result, _ = await self._open(raw_chunks, {"content-length": str(len(raw))}, chunk_size=64) + streamed = b"".join([chunk async for chunk in result.stream_iterator]) + + transformed = [json.loads(line) for line in streamed.split(b"\n") if line] + assert [row["custom_id"] for row in transformed] == ["request-0", "request-1", "request-2"] + assert transformed[1]["response"]["body"]["data"][0]["embedding"] == [0.1, 0.2] + assert transformed[1]["response"]["body"]["model"] == "gemini-2.5-flash" + assert result.headers["content-length"] == str(len(streamed)) + + async def test_object_without_newlines_streams_after_the_peek_limit(self): + piece = b"\xff" * (1024 * 1024) + raw_chunks = [piece] * 40 + + result, state = await self._open(raw_chunks, {"content-type": "image/png"}, chunk_size=len(piece)) + first = await anext(result.stream_iterator) + + assert state["served"] < len(raw_chunks) + rest = [chunk async for chunk in result.stream_iterator] + assert len(first) + sum(len(chunk) for chunk in rest) == len(piece) * len(raw_chunks) + assert set(first) == {0xFF} and all(set(chunk) == {0xFF} for chunk in rest) + assert result.headers["content-type"] == "image/png" + + async def test_consumer_stopping_early_closes_the_response(self): + raw_chunks = [b'{"line": %d}\n' % i for i in range(50)] + result, state = await self._open(raw_chunks, {}) + + await anext(result.stream_iterator) + await result.stream_iterator.aclose() + + assert state["closed"] is True + + async def test_gcs_error_raises_and_closes_the_response(self): + state = {"closed": False} + + async def handler(request: httpx.Request) -> httpx.Response: + response = httpx.Response(403, json={"error": {"message": "forbidden"}}) + original_aclose = response.aclose + + async def aclose(): + state["closed"] = True + await original_aclose() + + response.aclose = aclose + return response + + with pytest.raises(VertexAIError) as exc_info: + await BaseLLMHTTPHandler().async_retrieve_file_content_streaming( + file_content_request=FileContentRequest(file_id=_MANAGED_OUTPUT_FILE_ID), + provider_config=_StaticTokenFilesConfig(), + litellm_params={"gcs_bucket_name": "test-bucket"}, + headers={}, + logging_obj=_logging_obj(), + chunk_size=16, + client=_async_handler_with(handler), + ) + + assert exc_info.value.status_code == 403 + assert "forbidden" in str(exc_info.value) + assert state["closed"] is True + + async def test_afile_content_stream_routes_vertex_ai_to_the_gcs_stream(self): + raw = b'{"line": 1}\n{"line": 2}\n' * 20 + mock, state = _gcs_download_mock( + [raw[i : i + 64] for i in range(0, len(raw), 64)], {"content-length": str(len(raw))} + ) + + result = await litellm.afile_content( + file_id=_MANAGED_OUTPUT_FILE_ID, + custom_llm_provider="vertex_ai", + stream=True, + api_key="test-token", + gcs_bucket_name="test-bucket", + client=_async_handler_with(mock), + ) + + assert isinstance(result, FileContentStreamingResult) + assert result.headers["content-length"] == str(len(raw)) + assert state["urls"][0].endswith("predictions.jsonl?alt=media") + assert b"".join([chunk async for chunk in result.stream_iterator]) == raw + assert state["closed"] is True + + async def test_afile_content_without_stream_keeps_buffered_vertex_response(self): + raw = b'{"line": 1}\n{"line": 2}\n' + mock, _ = _gcs_download_mock([raw], {"content-length": str(len(raw))}) + + result = await litellm.afile_content( + file_id=_MANAGED_OUTPUT_FILE_ID, + custom_llm_provider="vertex_ai", + api_key="test-token", + gcs_bucket_name="test-bucket", + client=_async_handler_with(mock), + ) + + assert result.response.content == raw + + def test_sync_file_content_stream_is_rejected_for_vertex_ai(self): + mock, state = _gcs_download_mock([b"x"], {}) + + with pytest.raises(litellm.BadRequestError, match="afile_content"): + litellm.file_content( + file_id=_MANAGED_OUTPUT_FILE_ID, + custom_llm_provider="vertex_ai", + stream=True, + api_key="test-token", + gcs_bucket_name="test-bucket", + client=_async_handler_with(mock), + ) + + assert state["urls"] == [] diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py index 5d8222162a2..aa505c3019b 100644 --- a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py +++ b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py @@ -3384,12 +3384,14 @@ def test_get_file_content_provider_only_resolves_named_vertex_credentials( async def _mock_afile_content(**kwargs): captured_kwargs.update(kwargs) - return HttpxBinaryResponseContent( - response=httpx.Response( - status_code=200, - content=b"vertex-bytes", - headers={"content-type": "application/octet-stream"}, - ) + + async def _stream(): + yield b"vertex-" + yield b"bytes" + + return FileContentStreamingResult( + stream_iterator=_stream(), + headers={"content-type": "application/octet-stream"}, ) monkeypatch.setattr(litellm, "afile_content", _mock_afile_content) @@ -3414,6 +3416,7 @@ def test_get_file_content_provider_only_resolves_named_vertex_credentials( assert response.status_code == 200, response.text assert response.content == b"vertex-bytes" assert captured_kwargs.get("file_id") == "file-abc123" + assert captured_kwargs.get("stream") is True _assert_vertex_named_credentials_attached(captured_kwargs) proxy_logging_obj.post_call_failure_hook.assert_not_called()