fix(openai): download PDF URLs off the event loop and transform async chat messages once (#49176)

* fix(openai): transform async chat messages once, whatever configs were built before

* fix(openai): keep tool-result PDF downloads off the event loop and stop Mistral rewriting retried messages

* fix(openai): pre-download PDF URLs into the sync transform so async requests match sync

Async chat requests now download every PDF URL in user and tool messages at once, then run the sync transform_request on a worker thread that reads those downloads. The request, filename, logged messages, and download count match a sync request, and a failed download raises the sync error with no second attempt. DeepSeek joins the configs that flatten content and skip the up-front download.

* fix(openai): skip up-front PDF downloads past a file part the transform rejects

* fix(openai): keep system-message PDF downloads off the event loop and stop downloads after a dead PDF URL

* fix(openai): let PDF downloads after a dead URL finish and retry rate-limited ones one at a time

Cancelling in-flight downloads after a dead PDF URL could abort a proxy worker on uvloop, so the downloads now run to the end in the background. A URL whose up-front download got a 429 downloads again inside the transform one at a time, as on main. A content part whose type is not a string is no longer read as a PDF part, so it reaches the provider instead of raising a TypeError.

* fix(openai): download a request's PDF URLs one at a time and stop at the first dead one

* fix(openai): prefetch the PDF URLs the sync transform downloads, after role mapping and up to the first unreadable part

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
norkelbyte 2026-11-10 08:37:00 -08:00 • committed by GitHub
parent f8933ecae4
commit c90b5e057f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 747 additions and 51 deletions

View file

@ -5,7 +5,7 @@ Helper functions to handle images passed in messages
import asyncio
import base64
import time
from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
from collections.abc import AsyncIterator, Callable, Iterable, Iterator, Mapping, Sequence
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from functools import partial
@ -531,6 +531,29 @@ async def async_convert_urls_to_base64(
)
async def _download_outcome(url: str, timeout: float | httpx.Timeout | None) -> "str | litellm.ImageFetchError":
try:
return await async_convert_url_to_base64(url, timeout=timeout)
except litellm.ImageFetchError as error:
return error
async def _downloads_through_first_failure(
remote_urls: tuple[str, ...], timeout: float | httpx.Timeout | None
) -> "AsyncIterator[str | litellm.ImageFetchError]":
for url in remote_urls:
yield (outcome := await _download_outcome(url, timeout))
if isinstance(outcome, litellm.ImageFetchError):
return
async def async_fetch_urls_to_base64(
remote_urls: tuple[str, ...], timeout: float | httpx.Timeout | None = None
) -> "Mapping[str, str | litellm.ImageFetchError]":
outcomes: Final = tuple([outcome async for outcome in _downloads_through_first_failure(remote_urls, timeout)])
return MappingProxyType(dict(zip(remote_urls[: len(outcomes)], outcomes, strict=True)))
def _remote_urls_to_inline(
messages: Iterable[AllMessageValues], should_inline: Callable[[RemoteMedia], bool]
) -> tuple[str, ...]:
@ -539,6 +562,41 @@ def _remote_urls_to_inline(
return tuple(dict.fromkeys(remote.url for remote in remotes if should_inline(_remote_media(remote))))
def _message_parts(messages: Iterable[AllMessageValues]) -> Iterator[object]:
return chain.from_iterable(_content_parts(message) for message in messages)
def _file_of_file_part(part: object) -> Mapping[str, object] | None:
fields: Final = _as_mapping(part)
return _as_mapping(fields.get("file")) if fields is not None and fields.get("type") == "file" else None
def _file_id_url(part: object, url_starts: tuple[str, ...]) -> str | None:
file: Final = _file_of_file_part(part)
file_id: Final = file.get("file_id") if file is not None else None
return file_id if isinstance(file_id, str) and file_id.startswith(url_starts) else None
def _is_readable_file_part_or_other_part(part: object) -> bool:
fields: Final = _as_mapping(part)
if fields is None or fields.get("type") != "file":
return fields is not None
file: Final = _file_of_file_part(part)
file_id: Final = file.get("file_id") if file is not None else None
return file is not None and (not file_id or isinstance(file_id, str))
def has_file_id_urls(messages: Iterable[AllMessageValues], url_starts: tuple[str, ...]) -> bool:
return any(_file_id_url(part, url_starts) is not None for part in _message_parts(messages))
def file_id_urls_before_an_unreadable_part(
messages: Iterable[AllMessageValues], url_starts: tuple[str, ...]
) -> tuple[str, ...]:
readable_parts: Final = takewhile(_is_readable_file_part_or_other_part, _message_parts(messages))
return tuple(dict.fromkeys(url for part in readable_parts if (url := _file_id_url(part, url_starts)) is not None))
def inline_remote_media(
messages: list[AllMessageValues], # mutable-ok: every transform_request takes list[AllMessageValues]
should_inline: Callable[[RemoteMedia], bool] = inline_every_remote_url,

View file

@ -34,6 +34,9 @@ def _warn_on_dropped_budget(thinking: object, model: str) -> None:
class DeepSeekChatConfig(OpenAIGPTConfig):
def _flattens_content_lists(self) -> bool:
return True
def get_supported_openai_params(self, model: str) -> list[str]:
"""
DeepSeek reasoner models support thinking parameter.

View file

@ -25,6 +25,9 @@ class DockerModelRunnerChatConfig(OpenAIGPTConfig):
The engine name (e.g., "llama.cpp") is part of the API endpoint path.
"""
def _flattens_content_lists(self) -> bool:
return True
@overload
def _transform_messages(
self, messages: list[AllMessageValues], model: str, is_async: Literal[True]

View file

@ -22,6 +22,9 @@ class HerokuError(Exception):
class HerokuChatConfig(OpenAIGPTConfig):
def _flattens_content_lists(self) -> bool:
return True
@overload
def _transform_messages(
self, messages: list[AllMessageValues], model: str, is_async: Literal[True]

View file

@ -142,8 +142,6 @@ class MistralConfig(OpenAIGPTConfig):
- `response_format` (object or null): An object specifying the format that the model must output. Setting to { "type": "json_object" } enables JSON mode, which guarantees the message the model generates is in JSON. When using JSON mode you MUST also instruct the model to produce JSON yourself with a system or a user message.
"""
_is_base_class = False
temperature: int | None = None
top_p: int | None = None
max_tokens: int | None = None

View file

@ -2,9 +2,11 @@
Support for gpt model family
"""
import asyncio
import json
import os
from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping, Sequence
from contextvars import ContextVar
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, cast, overload
from urllib.parse import urlparse
@ -28,7 +30,10 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
)
from litellm.litellm_core_utils.prompt_templates.image_handling import (
async_convert_url_to_base64,
async_fetch_urls_to_base64,
convert_url_to_base64,
file_id_urls_before_an_unreadable_part,
has_file_id_urls,
)
from litellm.litellm_core_utils.prompt_templates.json_schema_utils import (
drop_non_python_regex_patterns,
@ -85,6 +90,20 @@ else:
_NO_TOOLS_UPDATE: Final[Mapping[str, object]] = MappingProxyType({})
_ROLES_WHOSE_FILES_ARE_DOWNLOADED: Final = ("user", "tool")
_PDF_URL_STARTS: Final = ("https://", "http://", "www.")
_NOTHING_PREFETCHED: Final[Mapping[str, "str | litellm.ImageFetchError"]] = MappingProxyType({})
_prefetched_pdf_urls: Final[ContextVar[Mapping[str, "str | litellm.ImageFetchError"]]] = ContextVar(
"prefetched_pdf_urls"
)
def _pdf_data_url(url: str) -> str:
prefetched: Final = _prefetched_pdf_urls.get(_NOTHING_PREFETCHED).get(url)
if isinstance(prefetched, litellm.ImageFetchError):
raise prefetched
return prefetched if prefetched is not None else convert_url_to_base64(url)
def targets_openai_hosted_endpoint(custom_llm_provider: str | None, api_base: str | None) -> bool:
if custom_llm_provider != "openai":
@ -191,9 +210,6 @@ class OpenAIGPTConfig(BaseLLMModelInfo, BaseConfig):
- `top_p` (number or null): An alternative to sampling with temperature, used for nucleus sampling.
"""
# Add a class variable to track if this is the base class
_is_base_class = True
frequency_penalty: int | None = None
function_call: str | dict | None = None
functions: list | None = None
@ -225,8 +241,6 @@ class OpenAIGPTConfig(BaseLLMModelInfo, BaseConfig):
if key != "self" and value is not None:
setattr(self.__class__, key, value)
self.__class__._is_base_class = False
@classmethod
def get_config(cls):
return super().get_config()
@ -322,9 +336,8 @@ class OpenAIGPTConfig(BaseLLMModelInfo, BaseConfig):
)
def contains_pdf_url(self, content_item: ChatCompletionFileObjectFile) -> bool:
potential_pdf_url_starts: Final = ["https://", "http://", "www."]
file_id: Final = content_item.get("file_id")
if file_id and any(file_id.startswith(start) for start in potential_pdf_url_starts):
if file_id and any(file_id.startswith(start) for start in _PDF_URL_STARTS):
return True
return False
@ -332,7 +345,7 @@ class OpenAIGPTConfig(BaseLLMModelInfo, BaseConfig):
content_copy: Final = content_item.copy()
file_id: Final = content_copy.get("file_id")
if file_id is not None:
base64_data: Final = convert_url_to_base64(file_id)
base64_data: Final = _pdf_data_url(file_id)
content_copy["file_data"] = base64_data
content_copy["filename"] = "my_file.pdf"
content_copy.pop("file_id")
@ -688,6 +701,12 @@ class OpenAIGPTConfig(BaseLLMModelInfo, BaseConfig):
**self._sanitized_tools_update_for_openai(optional_params, litellm_params),
}
def _flattens_content_lists(self) -> bool:
return False
def _with_provider_roles(self, messages: Sequence[AllMessageValues], model: str) -> Sequence[AllMessageValues]:
return messages
async def async_transform_request(
self,
model: str,
@ -696,30 +715,25 @@ class OpenAIGPTConfig(BaseLLMModelInfo, BaseConfig):
litellm_params: dict,
headers: dict,
) -> dict:
transformed_messages, tools = self._messages_and_tools_in_cache_mark_dialect(
model=model,
messages=await self._transform_messages(
messages=self._prompt_cache_ordered_messages(messages, litellm_params), model=model, is_async=True
),
tools=optional_params.get("tools", []),
prompt_cache_options=_prompt_cache_options(optional_params),
litellm_params=litellm_params,
request_params=optional_params,
)
if tools is not None and len(tools) > 0:
optional_params["tools"] = tools
if self.__class__._is_base_class:
if not optional_params.get("tools") and not optional_params.get("functions"):
optional_params.pop("tool_choice", None)
return {
"model": model,
"messages": transformed_messages,
**optional_params,
**self._sanitized_tools_update_for_openai(optional_params, litellm_params),
}
else:
## allow for any object specific behaviour to be handled
if not has_file_id_urls(messages, _PDF_URL_STARTS):
return self.transform_request(model, messages, optional_params, litellm_params, headers)
sent_messages: Final = self._with_provider_roles(
self._prompt_cache_ordered_messages(messages, litellm_params), model
)
pdf_urls: Final = file_id_urls_before_an_unreadable_part(
(message for message in sent_messages if message.get("role") in _ROLES_WHOSE_FILES_ARE_DOWNLOADED),
_PDF_URL_STARTS,
)
prefetched: Final = (
_NOTHING_PREFETCHED if self._flattens_content_lists() else await async_fetch_urls_to_base64(pdf_urls)
)
reset_token: Final = _prefetched_pdf_urls.set(prefetched)
try:
return await asyncio.to_thread(
self.transform_request, model, messages, optional_params, litellm_params, headers
)
finally:
_prefetched_pdf_urls.reset(reset_token)
def _passed_in_tools(self, optional_params: dict) -> bool:
return optional_params.get("tools", None) is not None

View file

@ -122,6 +122,15 @@ class OpenAIOSeriesConfig(OpenAIGPTConfig):
return super()._map_openai_params(non_default_params, optional_params, model, drop_params)
def _with_provider_roles(self, messages: Sequence[AllMessageValues], model: str) -> Sequence[AllMessageValues]:
keeps_system_messages: Final = supports_system_messages(model, "openai")
return tuple(
message
if keeps_system_messages or message["role"] != "system"
else ChatCompletionUserMessage(content=message["content"], role="user")
for message in messages
)
def is_model_o_series_model(self, model: str) -> bool:
model = model.split("/")[-1] # could be "openai/o3" or "o3"
return (
@ -152,13 +161,7 @@ class OpenAIOSeriesConfig(OpenAIGPTConfig):
- modalities: image => drop param (if user opts in to dropping param)
- role: system ==> translate to role 'user'
"""
_supports_system_messages: Final = supports_system_messages(model, "openai")
transformed_messages: Final = list(messages)
for i, message in enumerate(transformed_messages):
if message["role"] == "system" and not _supports_system_messages:
new_message = ChatCompletionUserMessage(content=message["content"], role="user")
transformed_messages[i] = new_message # Replace the old message with the new one
transformed_messages: Final = list(self._with_provider_roles(messages, model))
if is_async:
return super()._transform_messages(transformed_messages, model, is_async=cast(Literal[True], True))
else:

View file

@ -1166,7 +1166,7 @@ class OpenAIChatCompletion(BaseLLM, BaseOpenAILLM):
shared_session: Optional["ClientSession"] = None,
):
response = None
data = provider_config.transform_request(
data = await provider_config.async_transform_request(
model=model,
messages=messages,
optional_params=optional_params,

View file

@ -50,6 +50,9 @@ def create_config_class(provider: SimpleProviderConfig):
base_class: Final[type] = OpenAIGPTConfig if provider.base_class == "openai_gpt" else OpenAILikeChatConfig
class JSONProviderConfig(base_class):
def _flattens_content_lists(self) -> bool:
return bool(provider.special_handling.get("convert_content_list_to_string"))
@overload
def _transform_messages(
self, messages: list[AllMessageValues], model: str, is_async: Literal[True]
@ -69,7 +72,7 @@ def create_config_class(provider: SimpleProviderConfig):
"""Transform messages based on special_handling config"""
# Handle content list to string conversion if configured
if provider.special_handling.get("convert_content_list_to_string"):
if self._flattens_content_lists():
messages = handle_messages_with_content_list_to_str_conversion(messages)
if is_async:

View file

@ -32,6 +32,9 @@ class SambanovaConfig(OpenAIGPTConfig):
response_format: dict | None = None
tools: list | None = None
def _flattens_content_lists(self) -> bool:
return True
def __init__(
self,
max_tokens: int | None = None,

View file

@ -1,15 +1,18 @@
"""
Tests for OpenAI GPT transformation (litellm/llms/openai/chat/gpt_transformation.py)
"""
import asyncio
import base64
import copy
import json
import threading
from collections.abc import Callable
from typing import Final
import httpx
import pytest
import respx
from httpx import Response
from httpx import Request, Response
import litellm
from litellm.litellm_core_utils.prompt_templates.common_utils import (
@ -17,11 +20,21 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
TOOL_RESULT_MEDIA_BOUNDARY,
)
from litellm.litellm_core_utils.prompt_templates.image_handling import in_memory_cache
from litellm.llms.bedrock.chat.chat_completions.transformation import AmazonBedrockRuntimeChatCompletionsConfig
from litellm.llms.deepinfra.chat.transformation import DeepInfraConfig
from litellm.llms.deepseek.chat.transformation import DeepSeekChatConfig
from litellm.llms.docker_model_runner.chat.transformation import DockerModelRunnerChatConfig
from litellm.llms.heroku.chat.transformation import HerokuChatConfig
from litellm.llms.openai.chat.gpt_reasoning_transformation import OpenAIGPT5Config
from litellm.llms.openai.chat.gpt_transformation import (
OpenAIChatCompletionStreamingHandler,
OpenAIGPTConfig,
)
from litellm.llms.openai.chat.o_series_transformation import OpenAIOSeriesConfig
from litellm.llms.openai_like.dynamic_config import create_config_class
from litellm.llms.openai_like.json_loader import SimpleProviderConfig
from litellm.llms.sambanova.chat import SambanovaConfig
from litellm.llms.vertex_ai.vertex_gemma_models.transformation import VertexGemmaConfig
class TestOpenAIGPTConfig:
@ -1348,6 +1361,607 @@ async def test_openai_media_download_preserves_supported_chat_fields(
_PDF_BYTES: Final = b"%PDF-1.7\nmock-pdf"
_DOWNLOADED_PDF_PART: Final = {
"type": "file",
"file": {
"file_data": f"data:application/pdf;base64,{base64.b64encode(_PDF_BYTES).decode('utf-8')}",
"filename": "my_file.pdf",
},
}
def _pdf_url_part(url: str) -> dict[str, object]:
return {"type": "file", "file": {"file_id": url}}
async def _serve_pdf_to_async_clients_only(request: Request) -> Response:
return Response(200, headers={"content-type": "application/pdf"}, content=_PDF_BYTES)
def _tool_result_messages(*content: object) -> list[dict[str, object]]:
return [
{"role": "user", "content": "read the report"},
{
"role": "assistant",
"content": None,
"tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "fetch_report", "arguments": "{}"}}],
},
{"role": "tool", "tool_call_id": "call_1", "content": list(content)},
]
@pytest.fixture
def pdf_urls_reachable(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "user_url_validation", False)
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
_SUMMARIZE: Final = {"type": "text", "text": "Summarize the report."}
_UPLOADED_FILE_PART: Final = {"type": "file", "file": {"file_id": "file-uploaded"}}
_ONCE_URL: Final = "https://files.example.test/download-once.pdf"
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=True)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("messages", "expected_messages"),
[
pytest.param(
[{"role": "user", "content": [_SUMMARIZE, _UPLOADED_FILE_PART, _pdf_url_part(_ONCE_URL)]}],
[{"role": "user", "content": [_SUMMARIZE, _UPLOADED_FILE_PART, _DOWNLOADED_PDF_PART]}],
id="user_message",
),
pytest.param(
_tool_result_messages(_SUMMARIZE, _pdf_url_part(_ONCE_URL)),
[
*_tool_result_messages(_SUMMARIZE)[:2],
{"role": "tool", "tool_call_id": "call_1", "content": [_SUMMARIZE]},
{"role": "user", "content": [{"type": "text", "text": TOOL_RESULT_MEDIA_BOUNDARY}, _DOWNLOADED_PDF_PART]},
],
id="tool_result",
),
],
)
async def test_async_transform_request_downloads_a_pdf_url_once_with_the_async_client(
respx_mock: respx.MockRouter, messages: list[dict[str, object]], expected_messages: list[dict[str, object]]
) -> None:
in_memory_cache.delete_cache(_ONCE_URL)
download: Final = respx_mock.get(_ONCE_URL).mock(side_effect=_serve_pdf_to_async_clients_only)
request: Final = await OpenAIGPTConfig().async_transform_request(
model="gpt-4.1",
messages=copy.deepcopy(messages),
optional_params={},
litellm_params={"custom_llm_provider": "openai"},
headers={},
)
assert download.call_count == 1
assert request["messages"] == expected_messages
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=True)
@pytest.mark.asyncio
async def test_async_transform_request_downloads_a_pdf_too_large_for_the_url_cache_once(
respx_mock: respx.MockRouter,
) -> None:
pdf_url: Final = "https://files.example.test/too-large-to-cache.pdf"
large_pdf: Final = _PDF_BYTES + b"0" * in_memory_cache.max_size_per_item * 1024
async def serve_large_pdf_to_async_clients_only(request: Request) -> Response:
return Response(200, headers={"content-type": "application/pdf"}, content=large_pdf)
download: Final = respx_mock.get(pdf_url).mock(side_effect=serve_large_pdf_to_async_clients_only)
request: Final = await OpenAIGPTConfig().async_transform_request(
model="gpt-4.1",
messages=[{"role": "user", "content": [_SUMMARIZE, _pdf_url_part(pdf_url)]}],
optional_params={},
litellm_params={},
headers={},
)
assert download.call_count == 1
assert request["messages"][0]["content"][1]["file"] == {
"file_data": f"data:application/pdf;base64,{base64.b64encode(large_pdf).decode('utf-8')}",
"filename": "my_file.pdf",
}
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=True)
@pytest.mark.asyncio
async def test_async_transform_request_downloads_pdf_urls_one_at_a_time_for_a_host_that_refuses_bursts(
respx_mock: respx.MockRouter,
) -> None:
pdf_urls: Final = tuple(f"https://files.example.test/in-turn-{index}.pdf" for index in range(3))
for pdf_url in pdf_urls:
in_memory_cache.delete_cache(pdf_url)
download_running: Final = asyncio.Lock()
async def refuse_a_second_download_at_once(request: Request) -> Response:
if download_running.locked():
return Response(429)
async with download_running:
await asyncio.sleep(0.05)
return await _serve_pdf_to_async_clients_only(request)
downloads: Final = respx_mock.route(method="GET", host="files.example.test").mock(
side_effect=refuse_a_second_download_at_once
)
request: Final = await OpenAIGPTConfig().async_transform_request(
model="gpt-4.1",
messages=[{"role": "user", "content": [_SUMMARIZE, *map(_pdf_url_part, pdf_urls)]}],
optional_params={},
litellm_params={"custom_llm_provider": "openai"},
headers={},
)
assert request["messages"] == [{"role": "user", "content": [_SUMMARIZE, *[_DOWNLOADED_PDF_PART] * 3]}]
assert downloads.call_count == len(pdf_urls)
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=True)
@pytest.mark.asyncio
async def test_streamed_completion_downloads_its_pdf_url_with_the_async_client(respx_mock: respx.MockRouter) -> None:
pdf_url: Final = "https://files.example.test/streamed.pdf"
in_memory_cache.delete_cache(pdf_url)
download: Final = respx_mock.get(pdf_url).mock(side_effect=_serve_pdf_to_async_clients_only)
chunk: Final = {
"id": "chatcmpl-stream",
"object": "chat.completion.chunk",
"created": 0,
"model": "gpt-4.1",
"choices": [{"index": 0, "delta": {"role": "assistant", "content": "seen"}, "finish_reason": "stop"}],
}
chat: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
return_value=Response(
200,
headers={"content-type": "text/event-stream"},
content=f"data: {json.dumps(chunk)}\n\ndata: [DONE]\n\n".encode(),
)
)
stream: Final = await litellm.acompletion(
model="gpt-4.1",
messages=[{"role": "user", "content": [_SUMMARIZE, _pdf_url_part(pdf_url)]}],
api_key="sk-openai-test",
stream=True,
)
text: Final = "".join([part.choices[0].delta.content or "" async for part in stream])
assert text == "seen"
assert download.call_count == 1
assert json.loads(chat.calls[0].request.content)["messages"] == [
{"role": "user", "content": [_SUMMARIZE, _DOWNLOADED_PDF_PART]}
]
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=False)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("unreadable_part", "error"),
[
pytest.param(
{"type": "file", "file": None},
"Content block has type='file' but is missing the required 'file' field",
id="file_part_without_its_file",
),
pytest.param(
{"type": "file", "file": "file-uploaded"}, "has no attribute 'items'", id="file_that_is_not_an_object"
),
pytest.param(
{"type": "file", "file": {"file_id": 123}}, "has no attribute 'startswith'", id="file_id_not_a_string"
),
pytest.param("Summarize the report.", "has no attribute 'get'", id="part_that_is_not_an_object"),
],
)
async def test_async_transform_request_rejects_an_unreadable_part_like_the_sync_transform_before_a_later_pdf_url(
respx_mock: respx.MockRouter, unreadable_part: object, error: str
) -> None:
pdf_url: Final = "https://files.example.test/after-a-broken-part.pdf"
in_memory_cache.delete_cache(pdf_url)
download: Final = respx_mock.get(pdf_url).mock(return_value=Response(200, content=b"%PDF-1.4"))
messages: Final = [{"role": "user", "content": [_SUMMARIZE, unreadable_part, _pdf_url_part(pdf_url)]}]
with pytest.raises((litellm.BadRequestError, AttributeError), match=error) as sync_error:
OpenAIGPTConfig().transform_request(
model="gpt-4.1", messages=copy.deepcopy(messages), optional_params={}, litellm_params={}, headers={}
)
with pytest.raises(type(sync_error.value), match=error):
await OpenAIGPTConfig().async_transform_request(
model="gpt-4.1",
messages=copy.deepcopy(messages),
optional_params={},
litellm_params={"custom_llm_provider": "openai"},
headers={},
)
assert not download.called
@pytest.mark.asyncio
@pytest.mark.parametrize("config", [OpenAIGPTConfig(), DeepInfraConfig()], ids=["openai", "deepinfra"])
async def test_async_transform_request_sends_a_plain_string_tool_result_part_like_the_sync_transform(
config: OpenAIGPTConfig,
) -> None:
messages: Final = _tool_result_messages("plain string result")
request: Final = await config.async_transform_request(
model="test-model", messages=copy.deepcopy(messages), optional_params={}, litellm_params={}, headers={}
)
assert request == config.transform_request(
model="test-model", messages=copy.deepcopy(messages), optional_params={}, litellm_params={}, headers={}
)
def _json_provider_config(special_handling: dict[str, bool]) -> OpenAIGPTConfig:
provider: Final = SimpleProviderConfig(
"test-json-provider",
{
"base_url": "https://json-provider.example.test/v1",
"api_key_env": "TEST_JSON_PROVIDER_API_KEY",
"special_handling": special_handling,
},
)
return create_config_class(provider)()
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=False)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"config",
[
pytest.param(SambanovaConfig(), id="sambanova"),
pytest.param(HerokuChatConfig(), id="heroku"),
pytest.param(DockerModelRunnerChatConfig(), id="docker_model_runner"),
pytest.param(_json_provider_config({"convert_content_list_to_string": True}), id="json_provider"),
pytest.param(DeepSeekChatConfig(), id="deepseek"),
],
)
async def test_async_transform_request_does_not_download_a_pdf_url_for_a_provider_that_sends_only_text(
respx_mock: respx.MockRouter, config: OpenAIGPTConfig
) -> None:
pdf_url: Final = "https://files.example.test/dropped.pdf"
in_memory_cache.delete_cache(pdf_url)
download: Final = respx_mock.get(pdf_url).mock(side_effect=_serve_pdf_to_async_clients_only)
request: Final = await config.async_transform_request(
model="test-model",
messages=[{"role": "user", "content": [_SUMMARIZE, _pdf_url_part(pdf_url)]}],
optional_params={},
litellm_params={},
headers={},
)
assert download.call_count == 0
assert request["messages"] == [{"role": "user", "content": "Summarize the report."}]
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=True)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("config", "messages", "expected_messages"),
[
pytest.param(
HerokuChatConfig(),
[{"role": "user", "content": [_pdf_url_part(_ONCE_URL)]}],
[{"role": "user", "content": [_DOWNLOADED_PDF_PART]}],
id="pdf_only_message_for_a_provider_that_sends_only_text",
),
],
)
async def test_async_transform_request_downloads_a_pdf_the_sync_transform_reaches_off_the_event_loop(
respx_mock: respx.MockRouter,
config: OpenAIGPTConfig,
messages: list[dict[str, object]],
expected_messages: list[dict[str, object]],
) -> None:
in_memory_cache.delete_cache(_ONCE_URL)
event_loop_thread: Final = threading.get_ident()
def serve_pdf_off_the_event_loop_only(request: Request) -> Response:
if threading.get_ident() == event_loop_thread:
return Response(503)
return Response(200, headers={"content-type": "application/pdf"}, content=_PDF_BYTES)
respx_mock.get(_ONCE_URL).mock(side_effect=serve_pdf_off_the_event_loop_only)
request: Final = await config.async_transform_request(
model="test-model-without-system-messages",
messages=copy.deepcopy(messages),
optional_params={},
litellm_params={},
headers={},
)
assert request["messages"] == expected_messages
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=True)
@pytest.mark.asyncio
async def test_async_transform_request_downloads_a_system_pdf_with_the_async_client_for_a_model_without_system_messages(
respx_mock: respx.MockRouter,
) -> None:
in_memory_cache.delete_cache(_ONCE_URL)
download: Final = respx_mock.get(_ONCE_URL).mock(side_effect=_serve_pdf_to_async_clients_only)
request: Final = await OpenAIOSeriesConfig().async_transform_request(
model="test-model-without-system-messages",
messages=[{"role": "system", "content": [_pdf_url_part(_ONCE_URL)]}, {"role": "user", "content": [_SUMMARIZE]}],
optional_params={},
litellm_params={},
headers={},
)
assert download.call_count == 1
assert request["messages"] == [
{"role": "user", "content": [_DOWNLOADED_PDF_PART]},
{"role": "user", "content": [_SUMMARIZE]},
]
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=True)
@pytest.mark.asyncio
async def test_async_transform_request_downloads_a_pdf_url_with_the_async_client_for_a_json_provider_that_keeps_lists(
respx_mock: respx.MockRouter,
) -> None:
in_memory_cache.delete_cache(_ONCE_URL)
respx_mock.get(_ONCE_URL).mock(side_effect=_serve_pdf_to_async_clients_only)
request: Final = await _json_provider_config({}).async_transform_request(
model="test-model",
messages=[{"role": "user", "content": [_SUMMARIZE, _pdf_url_part(_ONCE_URL)]}],
optional_params={},
litellm_params={},
headers={},
)
assert request["messages"] == [{"role": "user", "content": [_SUMMARIZE, _DOWNLOADED_PDF_PART]}]
def _serve_pdf(request: Request) -> Response:
return Response(200, headers={"content-type": "application/pdf"}, content=_PDF_BYTES)
def _answer_not_found(request: Request) -> Response:
return Response(404)
def _answer_too_many_requests(request: Request) -> Response:
return Response(429)
def _refuse_connection(request: Request) -> Response:
raise httpx.ConnectError("connection refused", request=request)
_PARITY_URL: Final = "https://files.example.test/parity.pdf"
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=False)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"messages",
[
pytest.param([{"role": "user", "content": [_SUMMARIZE, _pdf_url_part(_PARITY_URL)]}], id="user_message"),
pytest.param(
[
{
"role": "user",
"content": [_SUMMARIZE, {"type": "file", "file": {"file_id": _PARITY_URL, "filename": "report.pdf"}}],
}
],
id="named_file",
),
pytest.param(_tool_result_messages(_SUMMARIZE, _pdf_url_part(_PARITY_URL)), id="tool_result"),
pytest.param(
[
{"role": "system", "content": [_pdf_url_part(_PARITY_URL)]},
{"role": "user", "content": [_SUMMARIZE]},
{"role": "assistant", "content": [_pdf_url_part(_PARITY_URL)]},
{"role": "user", "content": "Go on."},
],
id="system_and_assistant_messages",
),
],
)
@pytest.mark.parametrize(
("config", "model"),
[
pytest.param(OpenAIGPTConfig(), "gpt-4.1", id="openai"),
pytest.param(AmazonBedrockRuntimeChatCompletionsConfig(), "us.xai.grok-4.6", id="bedrock_chat_completions"),
pytest.param(VertexGemmaConfig(), "gemma-3-27b-it", id="vertex_gemma"),
],
)
async def test_async_transform_request_matches_the_sync_request_downloads_and_callers_messages(
respx_mock: respx.MockRouter, messages: list[dict[str, object]], config: OpenAIGPTConfig, model: str
) -> None:
download: Final = respx_mock.get(_PARITY_URL).mock(side_effect=_serve_pdf)
sync_messages: Final = copy.deepcopy(messages)
async_messages: Final = copy.deepcopy(messages)
in_memory_cache.delete_cache(_PARITY_URL)
sync_request: Final = config.transform_request(
model=model, messages=sync_messages, optional_params={}, litellm_params={}, headers={}
)
sync_downloads: Final = download.call_count
in_memory_cache.delete_cache(_PARITY_URL)
async_request: Final = await config.async_transform_request(
model=model, messages=async_messages, optional_params={}, litellm_params={}, headers={}
)
assert async_request == sync_request
assert async_messages == sync_messages
assert download.call_count - sync_downloads == sync_downloads
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=True)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"dead_url_answer",
[
pytest.param(_answer_not_found, id="not_found"),
pytest.param(_answer_too_many_requests, id="rate_limited"),
pytest.param(_refuse_connection, id="refused"),
],
)
async def test_async_transform_request_fails_a_dead_pdf_url_with_the_sync_error_and_download_count(
respx_mock: respx.MockRouter, dead_url_answer: Callable[[Request], Response]
) -> None:
healthy_url: Final = "https://files.example.test/healthy.pdf"
dead_url: Final = "https://files.example.test/dead.pdf"
healthy: Final = respx_mock.get(healthy_url).mock(side_effect=_serve_pdf)
dead: Final = respx_mock.get(dead_url).mock(side_effect=dead_url_answer)
messages: Final = [{"role": "user", "content": [_SUMMARIZE, _pdf_url_part(healthy_url), _pdf_url_part(dead_url)]}]
in_memory_cache.delete_cache(healthy_url)
with pytest.raises(litellm.ImageFetchError) as sync_error:
OpenAIGPTConfig().transform_request(
model="gpt-4.1", messages=copy.deepcopy(messages), optional_params={}, litellm_params={}, headers={}
)
sync_downloads: Final = healthy.call_count + dead.call_count
in_memory_cache.delete_cache(healthy_url)
with pytest.raises(litellm.ImageFetchError) as async_error:
await OpenAIGPTConfig().async_transform_request(
model="gpt-4.1", messages=copy.deepcopy(messages), optional_params={}, litellm_params={}, headers={}
)
assert str(async_error.value) == str(sync_error.value)
assert healthy.call_count + dead.call_count - sync_downloads == sync_downloads
_DEAD_URL: Final = "https://files.example.test/dead-first.pdf"
_AFTER_DEAD_URL: Final = "https://files.example.test/after-dead.pdf"
_WWW_FILE_ID: Final = "www.files.example.test/no-scheme.pdf"
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=False)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("config", "model", "messages"),
[
pytest.param(
OpenAIGPTConfig(),
"gpt-4.1",
[{"role": "user", "content": [_SUMMARIZE, _pdf_url_part(_DEAD_URL), _pdf_url_part(_AFTER_DEAD_URL)]}],
id="dead_url",
),
pytest.param(
OpenAIGPTConfig(),
"gpt-4.1",
[{"role": "user", "content": [_SUMMARIZE, _pdf_url_part(_WWW_FILE_ID), _pdf_url_part(_AFTER_DEAD_URL)]}],
id="www_file_id",
),
pytest.param(
OpenAIOSeriesConfig(),
"test-model-without-system-messages",
[
{"role": "system", "content": [_pdf_url_part(_DEAD_URL)]},
{"role": "user", "content": [_SUMMARIZE, _pdf_url_part(_AFTER_DEAD_URL)]},
],
id="system_message_for_a_model_without_system_messages",
),
],
)
async def test_async_transform_request_fails_a_dead_pdf_url_without_downloading_the_pdf_urls_after_it(
respx_mock: respx.MockRouter, config: OpenAIGPTConfig, model: str, messages: list[dict[str, object]]
) -> None:
in_memory_cache.delete_cache(_AFTER_DEAD_URL)
respx_mock.get(_DEAD_URL).mock(return_value=Response(404))
later: Final = respx_mock.get(_AFTER_DEAD_URL).mock(side_effect=_serve_pdf_to_async_clients_only)
with pytest.raises(litellm.ImageFetchError):
await config.async_transform_request(
model=model, messages=copy.deepcopy(messages), optional_params={}, litellm_params={}, headers={}
)
await asyncio.wait_for(asyncio.gather(*(asyncio.all_tasks() - {asyncio.current_task()})), timeout=5)
assert later.call_count == 0
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=False)
@pytest.mark.asyncio
async def test_async_transform_request_downloads_a_moved_system_pdf_first_when_system_messages_go_first(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "openai_system_messages_first", True)
in_memory_cache.delete_cache(_AFTER_DEAD_URL)
respx_mock.get(_DEAD_URL).mock(return_value=Response(404))
later: Final = respx_mock.get(_AFTER_DEAD_URL).mock(side_effect=_serve_pdf_to_async_clients_only)
with pytest.raises(litellm.ImageFetchError):
await OpenAIOSeriesConfig().async_transform_request(
model="test-model-without-system-messages",
messages=[
{"role": "user", "content": [_SUMMARIZE, _pdf_url_part(_AFTER_DEAD_URL)]},
{"role": "system", "content": [_pdf_url_part(_DEAD_URL)]},
],
optional_params={},
litellm_params={"custom_llm_provider": "openai"},
headers={},
)
assert later.call_count == 0
@pytest.mark.asyncio
@pytest.mark.parametrize("role", ["system", "assistant", "developer"])
@pytest.mark.parametrize("part_type", [pytest.param(["text"], id="list"), pytest.param({"kind": "text"}, id="dict")])
async def test_async_transform_request_passes_a_part_with_a_non_string_type_through_like_the_sync_request(
role: str, part_type: object
) -> None:
messages: Final = [
{"role": role, "content": [{"type": part_type, "text": "Answer briefly."}]},
{"role": "user", "content": "hi"},
]
sync_request: Final = OpenAIGPTConfig().transform_request(
model="gpt-4.1", messages=copy.deepcopy(messages), optional_params={}, litellm_params={}, headers={}
)
async_request: Final = await OpenAIGPTConfig().async_transform_request(
model="gpt-4.1", messages=copy.deepcopy(messages), optional_params={}, litellm_params={}, headers={}
)
assert async_request == sync_request
@pytest.mark.usefixtures("pdf_urls_reachable")
@pytest.mark.respx(assert_all_called=True)
@pytest.mark.asyncio
async def test_a_pdf_url_that_failed_in_an_async_request_downloads_again_in_a_later_request_of_the_same_task(
respx_mock: respx.MockRouter,
) -> None:
in_memory_cache.delete_cache(_ONCE_URL)
respx_mock.get(_ONCE_URL).mock(side_effect=[Response(404), _serve_pdf(Request("GET", _ONCE_URL))])
messages: Final = [{"role": "user", "content": [_SUMMARIZE, _pdf_url_part(_ONCE_URL)]}]
with pytest.raises(litellm.ImageFetchError):
await OpenAIGPTConfig().async_transform_request(
model="gpt-4.1", messages=copy.deepcopy(messages), optional_params={}, litellm_params={}, headers={}
)
request: Final = OpenAIGPTConfig().transform_request(
model="gpt-4.1", messages=copy.deepcopy(messages), optional_params={}, litellm_params={}, headers={}
)
assert request["messages"] == [{"role": "user", "content": [_SUMMARIZE, _DOWNLOADED_PDF_PART]}]
@pytest.mark.respx(assert_all_called=True)
def test_vision_with_custom_model_preserves_base64_image_in_request(respx_mock: respx.MockRouter) -> None:
image_uri: Final = f"data:image/png;base64,{base64.b64encode(b'test-image').decode('utf-8')}"
@ -1560,14 +2174,8 @@ class TestSystemMessagesFirst:
@pytest.mark.asyncio
async def test_async_transform_request_moves_system_messages_first(self, monkeypatch):
class UninstantiatedOpenAIGPTConfig(OpenAIGPTConfig):
_is_base_class = True
def __init__(self) -> None:
pass
monkeypatch.setattr(litellm, "openai_system_messages_first", True)
request = await UninstantiatedOpenAIGPTConfig().async_transform_request(
request = await self.config.async_transform_request(
model="gpt-4.1",
messages=self._messages(),
optional_params={},