perf: move Anthropic, Vertex Anthropic, Ollama and HF template fetches off the event loop (#40311)

Direct Anthropic http image inlining, Vertex AI Anthropic forced base64 conversion, Ollama completion image download and the watsonx GPT-OSS Hugging Face chat template lookup all ran synchronous HTTP inside the async request path. Each provider config now transforms through async_inline_remote_media on the async path, the Anthropic handler awaits the config's async_transform_request before dispatch and in the Rust fallback, and watsonx text exposes async_transform_request and always awaits ahf_chat_template

Resolves LIT-7028

Co-authored-by: yassin <yassin@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-09-08 15:53:43 -07:00 • committed by GitHub
parent b8d573c5f9
commit 0f886d9006
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 380 additions and 130 deletions

View file

@ -183,6 +183,7 @@ class _RemoteSource:
class RemoteMedia:
url: str
fields: Mapping[str, object]
part_type: str
_NO_FIELDS: Final[Mapping[str, object]] = MappingProxyType({})
@ -192,6 +193,10 @@ def inline_every_remote_url(_media: RemoteMedia) -> bool:
return True
def inline_remote_image_urls(media: RemoteMedia) -> bool:
return media.part_type == "image_url"
def _parse_remote_image(fields: Mapping[str, object]) -> _RemoteImage | None:
if fields.get("type") != "image_url":
return None
@ -223,11 +228,11 @@ def _parse_remote_part(part: object) -> _RemoteImage | _RemoteFile | _RemoteSour
def _remote_media(remote: _RemoteImage | _RemoteFile | _RemoteSource) -> RemoteMedia:
match remote:
case _RemoteImage(_, image_url, url):
return RemoteMedia(url, image_url if image_url is not None else _NO_FIELDS)
return RemoteMedia(url, image_url if image_url is not None else _NO_FIELDS, "image_url")
case _RemoteFile(_, file, url):
return RemoteMedia(url, file)
case _RemoteSource(_, source, url):
return RemoteMedia(url, source)
return RemoteMedia(url, file, "file")
case _RemoteSource(part, source, url):
return RemoteMedia(url, source, str(part.get("type")))
_PDF_FORMAT: Final = MappingProxyType({"format": "application/pdf"})

View file

@ -368,27 +368,92 @@ class AnthropicChatCompletion(BaseLLM):
if config is None:
raise ValueError(f"Provider config not found for model: {model} and provider: {custom_llm_provider}")
def build_request() -> tuple[dict, dict]: # mutable-ok: rewritten in place downstream
"""Translate the request the Python way, returning `(headers, data)`.
transform_params: Final = {**optional_params, "is_vertex_request": is_vertex_request}
def finish_request(request_data: dict) -> tuple[dict, dict]: # mutable-ok: rewritten in place downstream
"""Filter beta headers and emit pre_call, returning `(headers, data)`.
The pair stays mutable because the streaming path rewrites it in
place (`data["stream"] = True`) before sending.
Shared by the normal path and by the Rust path's fallback, which
builds it only when the Rust call did not serve the request.
place (`data["stream"] = True`) before sending. A Rust attempt that
declined already emitted pre_call for this request, so skip it there.
"""
request_data: Final = config.transform_request(
model=model,
messages=messages,
optional_params={**optional_params, "is_vertex_request": is_vertex_request},
litellm_params=litellm_params,
headers=headers,
)
return update_request_with_filtered_beta(
request_headers, data = update_request_with_filtered_beta(
headers=headers,
request_data=request_data,
provider=custom_llm_provider,
)
if not serves_via_rust:
logging_obj.pre_call(
input=messages,
api_key=api_key,
additional_args={
"complete_input_dict": data,
"api_base": api_base,
"headers": request_headers,
},
)
print_verbose(f"_is_function_call: {_is_function_call}")
return request_headers, data
async def acompletion_dispatch() -> "ModelResponse | CustomStreamWrapper":
"""Translate then send, so the provider config can inline remote media off the event loop."""
request_headers, data = finish_request(
await config.async_transform_request(
model=model,
messages=messages,
optional_params=transform_params,
litellm_params=litellm_params,
headers=headers,
)
)
if (
stream is True
): # if function call - fake the streaming (need complete blocks for output parsing in openai format)
print_verbose("makes async anthropic streaming POST request")
data["stream"] = stream
return await self.acompletion_stream_function(
model=model,
messages=messages,
data=data,
api_base=api_base,
custom_prompt_dict=custom_prompt_dict,
model_response=model_response,
print_verbose=print_verbose,
encoding=encoding,
api_key=api_key,
logging_obj=logging_obj,
optional_params=optional_params,
stream=stream,
_is_function_call=_is_function_call,
json_mode=json_mode,
litellm_params=litellm_params,
logger_fn=logger_fn,
headers=request_headers,
timeout=timeout,
client=(client if client is not None and isinstance(client, AsyncHTTPHandler) else None),
)
return await self.acompletion_function(
model=model,
messages=messages,
data=data,
api_base=api_base,
custom_prompt_dict=custom_prompt_dict,
model_response=model_response,
print_verbose=print_verbose,
encoding=encoding,
api_key=api_key,
provider_config=config,
logging_obj=logging_obj,
optional_params=optional_params,
stream=stream,
_is_function_call=_is_function_call,
litellm_params=litellm_params,
logger_fn=logger_fn,
headers=request_headers,
client=client,
json_mode=json_mode,
timeout=timeout,
)
# The Rust core owns the whole call for the subset it accepts, so ask
# before transforming: whichever path runs emits pre_call exactly once.
@ -424,35 +489,6 @@ class AnthropicChatCompletion(BaseLLM):
additional_args=rust_logging_args,
)
if acompletion is True:
async def python_fallback() -> "ModelResponse | CustomStreamWrapper":
# pre_call already fired for this request above. The Rust
# path only declines before the provider is called, so this
# is the same attempt continuing, not a second one.
fallback_headers, fallback_data = build_request()
return await self.acompletion_function(
model=model,
messages=messages,
data=fallback_data,
api_base=api_base,
custom_prompt_dict=custom_prompt_dict,
model_response=model_response,
print_verbose=print_verbose,
encoding=encoding,
api_key=api_key,
provider_config=config,
logging_obj=logging_obj,
optional_params=optional_params,
stream=stream,
_is_function_call=_is_function_call,
litellm_params=litellm_params,
logger_fn=logger_fn,
headers=fallback_headers,
client=client,
json_mode=json_mode,
timeout=timeout,
)
return rust_chat_completions_bridge.achat_completions_or_fallback(
model=model,
messages=messages,
@ -464,7 +500,7 @@ class AnthropicChatCompletion(BaseLLM):
extra_headers=headers,
timeout=timeout,
on_response=log_rust_post_call,
python_fallback=python_fallback,
python_fallback=acompletion_dispatch,
)
rust_response: Final = rust_chat_completions_bridge.chat_completions(
model=model,
@ -481,74 +517,18 @@ class AnthropicChatCompletion(BaseLLM):
if rust_response is not None:
return rust_response
headers, data = build_request()
## LOGGING
# Reaching here with `serves_via_rust` set means the Rust attempt
# declined at call time, before the provider was called, and already
# logged this request. That is the same attempt continuing.
if not serves_via_rust:
logging_obj.pre_call(
input=messages,
api_key=api_key,
additional_args={
"complete_input_dict": data,
"api_base": api_base,
"headers": headers,
},
)
print_verbose(f"_is_function_call: {_is_function_call}")
if acompletion is True:
if (
stream is True
): # if function call - fake the streaming (need complete blocks for output parsing in openai format)
print_verbose("makes async anthropic streaming POST request")
data["stream"] = stream
return self.acompletion_stream_function(
model=model,
messages=messages,
data=data,
api_base=api_base,
custom_prompt_dict=custom_prompt_dict,
model_response=model_response,
print_verbose=print_verbose,
encoding=encoding,
api_key=api_key,
logging_obj=logging_obj,
optional_params=optional_params,
stream=stream,
_is_function_call=_is_function_call,
json_mode=json_mode,
litellm_params=litellm_params,
logger_fn=logger_fn,
headers=headers,
timeout=timeout,
client=(client if client is not None and isinstance(client, AsyncHTTPHandler) else None),
)
else:
return self.acompletion_function(
model=model,
messages=messages,
data=data,
api_base=api_base,
custom_prompt_dict=custom_prompt_dict,
model_response=model_response,
print_verbose=print_verbose,
encoding=encoding,
api_key=api_key,
provider_config=config,
logging_obj=logging_obj,
optional_params=optional_params,
stream=stream,
_is_function_call=_is_function_call,
litellm_params=litellm_params,
logger_fn=logger_fn,
headers=headers,
client=client,
json_mode=json_mode,
timeout=timeout,
)
return acompletion_dispatch()
else:
headers, data = finish_request(
config.transform_request(
model=model,
messages=messages,
optional_params=transform_params,
litellm_params=litellm_params,
headers=headers,
)
)
## COMPLETION CALL
if (
stream is True

View file

@ -26,6 +26,11 @@ from litellm.litellm_core_utils.core_helpers import map_finish_reason
from litellm.litellm_core_utils.prompt_templates.common_utils import (
sanitize_input_schema_for_anthropic,
)
from litellm.litellm_core_utils.prompt_templates.image_handling import (
RemoteMedia,
async_inline_remote_media,
inline_remote_image_urls,
)
from litellm.llms.base_llm.base_utils import type_to_response_format_param
from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
from litellm.types.llms.anthropic import (
@ -1840,6 +1845,25 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
break
return headers
def inlines_remote_media(self, media: RemoteMedia) -> bool:
return inline_remote_image_urls(media) and media.url.startswith("http://")
async def async_transform_request(
self,
model: str,
messages: list[AllMessageValues], # mutable-ok: BaseConfig signature
optional_params: dict[str, object], # mutable-ok: BaseConfig signature
litellm_params: dict[str, object], # mutable-ok: BaseConfig signature
headers: dict[str, object], # mutable-ok: BaseConfig signature
) -> dict[str, object]: # mutable-ok: BaseConfig signature
return self.transform_request(
model=model,
messages=await async_inline_remote_media(messages, should_inline=self.inlines_remote_media),
optional_params=optional_params,
litellm_params=litellm_params,
headers=headers,
)
def transform_request(
self,
model: str,

View file

@ -16,6 +16,10 @@ from litellm.litellm_core_utils.prompt_templates.factory import (
custom_prompt,
ollama_pt,
)
from litellm.litellm_core_utils.prompt_templates.image_handling import (
async_inline_remote_media,
inline_remote_image_urls,
)
from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator
from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
from litellm.types.llms.openai import AllMessageValues, ChatCompletionUsageBlock
@ -344,6 +348,26 @@ class OllamaConfig(BaseConfig):
)
return model_response
@property
def uses_async_transform_request(self) -> bool:
return True
async def async_transform_request(
self,
model: str,
messages: list[AllMessageValues], # mutable-ok: BaseConfig signature
optional_params: dict[str, object], # mutable-ok: BaseConfig signature
litellm_params: dict[str, object], # mutable-ok: BaseConfig signature
headers: dict[str, object], # mutable-ok: BaseConfig signature
) -> dict[str, object]: # mutable-ok: BaseConfig signature
return self.transform_request(
model=model,
messages=await async_inline_remote_media(messages, should_inline=inline_remote_image_urls),
optional_params=optional_params,
litellm_params=litellm_params,
headers=headers,
)
def transform_request(
self,
model: str,

View file

@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, Final
import httpx
import litellm
from litellm.litellm_core_utils.prompt_templates.image_handling import RemoteMedia, inline_remote_image_urls
from litellm.llms.base_llm.chat.transformation import LiteLLMLoggingObj
from litellm.types.llms.openai import AllMessageValues
from litellm.types.utils import ModelResponse
@ -51,6 +52,9 @@ class VertexAIAnthropicConfig(AnthropicConfig):
def custom_llm_provider(self) -> str | None:
return "vertex_ai"
def inlines_remote_media(self, media: RemoteMedia) -> bool:
return inline_remote_image_urls(media)
def should_strip_billing_metadata(self) -> bool:
return True

View file

@ -157,11 +157,9 @@ class IBMWatsonXChatConfig(IBMWatsonXMixin, OpenAIGPTConfig):
@staticmethod
async def aapply_prompt_template(model: str, messages: list[dict[str, str]]) -> str | None:
"""Apply prompt template (async version)"""
import litellm
from litellm.litellm_core_utils.prompt_templates.factory import (
ahf_chat_template,
custom_prompt,
hf_chat_template,
ibm_granite_pt,
mistral_instruct_pt,
)
@ -179,11 +177,7 @@ class IBMWatsonXChatConfig(IBMWatsonXMixin, OpenAIGPTConfig):
else:
hf_model = model
try:
# Use sync if cached, async if not
if hf_model in litellm.known_tokenizer_config:
result = hf_chat_template(model=hf_model, messages=messages)
else:
result = await ahf_chat_template(model=hf_model, messages=messages)
result = await ahf_chat_template(model=hf_model, messages=messages)
# Return result if it's truthy (not None and not empty string)
# The caller (_aconvert_watsonx_messages_core) will handle None/empty by falling back to default
if result:

View file

@ -16,6 +16,7 @@ from ..common_utils import (
IBMWatsonXMixin,
WatsonXAIError,
_get_api_params,
aconvert_watsonx_messages_to_prompt,
convert_watsonx_messages_to_prompt,
)
@ -236,7 +237,11 @@ class IBMWatsonXAIConfig(IBMWatsonXMixin, BaseConfig):
**watsonx_auth_payload,
}
async def atransform_request(
@property
def uses_async_transform_request(self) -> bool:
return True
async def async_transform_request(
self,
model: str,
messages: list[AllMessageValues],
@ -244,11 +249,6 @@ class IBMWatsonXAIConfig(IBMWatsonXMixin, BaseConfig):
litellm_params: dict,
headers: dict,
) -> dict:
"""Async version of transform_request"""
from litellm.llms.watsonx.common_utils import (
aconvert_watsonx_messages_to_prompt,
)
provider: Final = model.split("/")[0]
prompt: Final = await aconvert_watsonx_messages_to_prompt(
model=model, messages=messages, provider=provider, custom_prompt_dict={}

View file

@ -9,7 +9,8 @@ import pytest
import litellm
from litellm.constants import RESPONSE_FORMAT_TOOL_NAME
from litellm.llms.anthropic.chat.handler import ModelResponseIterator, make_call
from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm._uuid import uuid
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.types.llms.openai import (
ChatCompletionToolCallChunk,
ChatCompletionToolCallFunctionChunk,
@ -85,6 +86,54 @@ def test_anthropic_completion_does_not_send_deployment_default_limits():
assert "default_api_key_tpm_limit" not in request_body
async def test_anthropic_async_completion_inlines_http_images_off_the_event_loop(async_only_image_fetch):
http_image_url = f"http://img.example/{uuid.uuid4()}.png"
https_image_url = f"https://img.example/{uuid.uuid4()}.png"
captured = {}
def handle(request):
captured["body"] = json.loads(request.content)
return httpx.Response(
200,
json={
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-6",
"content": [{"type": "text", "text": "Green"}],
"stop_reason": "end_turn",
"usage": {"input_tokens": 1, "output_tokens": 1},
},
)
client = AsyncHTTPHandler()
client.client = httpx.AsyncClient(transport=httpx.MockTransport(handle))
response = await litellm.acompletion(
model="anthropic/claude-sonnet-4-6",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "What colour is this?"},
{"type": "image_url", "image_url": {"url": http_image_url}},
{"type": "image_url", "image_url": {"url": https_image_url}},
],
}
],
api_key="test-key",
client=client,
)
assert response.choices[0].message.content == "Green"
assert async_only_image_fetch.fetched == [http_image_url]
sources = [part["source"] for part in captured["body"]["messages"][0]["content"] if part["type"] == "image"]
assert sources == [
{"type": "base64", "media_type": "image/png", "data": async_only_image_fetch.base64_png},
{"type": "url", "url": https_image_url},
]
def test_redacted_thinking_content_block_delta():
chunk = {
"type": "content_block_start",

View file

@ -2,9 +2,11 @@ import json
from litellm._uuid import uuid
from unittest.mock import MagicMock, patch
import httpx
import pytest
import litellm
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.llms.ollama.completion.transformation import (
OllamaConfig,
OllamaTextCompletionResponseIterator,
@ -502,3 +504,43 @@ class TestOllamaTextCompletionResponseIterator:
assert result["usage"]["prompt_tokens"] == 10
assert result["usage"]["completion_tokens"] == 5
assert result["usage"]["total_tokens"] == 15
async def test_ollama_async_completion_inlines_remote_images_off_the_event_loop(async_only_image_fetch):
image_url = f"https://img.example/{uuid.uuid4()}.png"
captured = {}
def handle(request):
captured["body"] = json.loads(request.content)
return httpx.Response(
200,
json={
"model": "llava",
"response": "Green",
"done": True,
"prompt_eval_count": 1,
"eval_count": 1,
},
)
client = AsyncHTTPHandler()
client.client = httpx.AsyncClient(transport=httpx.MockTransport(handle))
response = await litellm.acompletion(
model="ollama/llava",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "What colour is this?"},
{"type": "image_url", "image_url": {"url": image_url}},
],
}
],
api_base="http://ollama.example:11434",
client=client,
)
assert response.choices[0].message.content == "Green"
assert async_only_image_fetch.fetched == [image_url]
assert captured["body"]["images"] == [async_only_image_fetch.base64_png]

View file

@ -6,11 +6,15 @@ Vertex AI Anthropic models don't support URL sources for images.
LiteLLM should convert image URLs to base64 when using Vertex AI Anthropic.
"""
import json
import sys
from unittest.mock import patch, MagicMock
import httpx
import pytest
import litellm
from litellm._uuid import uuid
from litellm.litellm_core_utils.prompt_templates.factory import (
anthropic_messages_pt,
convert_to_anthropic_tool_result,
@ -371,3 +375,59 @@ class TestToolMessageImageURLHandling:
assert item["source"]["type"] == "url"
return
pytest.fail("Could not find image in tool result")
async def test_vertex_ai_anthropic_async_completion_inlines_https_images_off_the_event_loop(async_only_image_fetch):
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
image_url = f"https://img.example/{uuid.uuid4()}.png"
captured = {}
def handle(request):
captured["body"] = json.loads(request.content)
return httpx.Response(
200,
json={
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-6",
"content": [{"type": "text", "text": "Green"}],
"stop_reason": "end_turn",
"usage": {"input_tokens": 1, "output_tokens": 1},
},
)
client = AsyncHTTPHandler()
client.client = httpx.AsyncClient(transport=httpx.MockTransport(handle))
vertexai = MagicMock()
vertexai.preview.language_models = MagicMock()
with (
patch.dict(sys.modules, {"vertexai": vertexai}),
patch.object( # test-quality-ok: litellm.acompletion has no seam for Vertex token minting
litellm.main.vertex_partner_models_chat_completion,
"_ensure_access_token",
return_value=("token", "test-project"),
),
):
response = await litellm.acompletion(
model="vertex_ai/claude-sonnet-4-6",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "What colour is this?"},
{"type": "image_url", "image_url": {"url": image_url}},
],
}
],
vertex_project="test-project",
vertex_location="us-east5",
client=client,
)
assert response.choices[0].message.content == "Green"
assert async_only_image_fetch.fetched == [image_url]
sources = [part["source"] for part in captured["body"]["messages"][0]["content"] if part["type"] == "image"]
assert sources == [{"type": "base64", "media_type": "image/png", "data": async_only_image_fetch.base64_png}]

View file

@ -356,6 +356,74 @@ async def test_watsonx_gpt_oss_uses_async_http_handler():
assert result["status"] == "success", "Should return success status"
@pytest.mark.parametrize("tokenizer_config_cached", [False, True], ids=["tokenizer_config", "cached_config_jinja"])
async def test_watsonx_text_gpt_oss_async_completion_fetches_hf_template_off_the_event_loop(
monkeypatch, tokenizer_config_cached
):
import httpx
from litellm._uuid import uuid
from litellm.litellm_core_utils.prompt_templates import huggingface_template_handler
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
hf_model = f"openai/gpt-oss-{uuid.uuid4()}"
chat_template = "{% for m in messages %}<|{{ m['role'] }}|>{{ m['content'] }}{% endfor %}"
if tokenizer_config_cached:
cached_config = {"status": "success", "tokenizer": {"bos_token": None, "eos_token": None}}
monkeypatch.setattr(litellm, "known_tokenizer_config", {hf_model: cached_config})
expected_fetch = f"https://huggingface.co/{hf_model}/raw/main/chat_template.jinja"
else:
monkeypatch.setattr(litellm, "known_tokenizer_config", {})
expected_fetch = f"https://huggingface.co/{hf_model}/raw/main/tokenizer_config.json"
hf_fetched = []
captured = {}
def forbid_sync_client():
raise AssertionError("sync HuggingFace fetch ran on the request path")
async def serve_hf_file(url, **kwargs):
hf_fetched.append(url)
if url.endswith(".jinja"):
return httpx.Response(200, content=chat_template.encode())
return httpx.Response(200, json={"chat_template": chat_template, "bos_token": None, "eos_token": None})
monkeypatch.setattr(huggingface_template_handler, "_get_httpx_client", forbid_sync_client)
monkeypatch.setattr(huggingface_template_handler, "get_async_httpx_client", lambda **kwargs: Mock(get=serve_hf_file))
def handle(request):
captured["body"] = json.loads(request.content)
return httpx.Response(
200,
json={
"model_id": hf_model,
"results": [
{
"generated_text": "Hi",
"generated_token_count": 1,
"input_token_count": 1,
"stop_reason": "eos_token",
}
],
},
)
client = AsyncHTTPHandler()
client.client = httpx.AsyncClient(transport=httpx.MockTransport(handle))
response = await litellm.acompletion(
model=f"watsonx_text/{hf_model}",
messages=[{"role": "user", "content": "Hi there"}],
api_base="https://test-api.watsonx.ai",
project_id="test-project-id",
token="test-token",
client=client,
)
assert response.choices[0].message.content == "Hi"
assert hf_fetched == [expected_fetch]
assert captured["body"]["input"] == "<|user|>Hi there"
def test_watsonx_chat_completion_with_reasoning_effort(monkeypatch):
"""
Test that 'reasoning_effort' is correctly passed through to the WatsonX API payload.