scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-13 08:52:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 174

fix(cache_control): pick the chat-path dialect from the deployment's provider prefix (#45295)

Browse source 
* fix(cache_control): resolve the chat-path dialect with the deployment api_base

* test(cache_control): type the chat-path resolver helpers

* fix(cache_control): pick the chat-path dialect from the deployment's provider prefix

Resolving with the deployment api_base followed the dispatch's azure_ai to
azure flip, so chat on a non-Foundry Azure host sent Anthropic cache_control
while /v1/messages, told azure_ai, sent the OpenAI breakpoint. Use the model's
provider prefix when the caller does not name a provider, so chat matches the
told path whatever AZURE_AI_API_BASE holds, and keep name-only resolution
everywhere else.

* fix(cache_control): keep litellm_proxy routing when the chat path reads the provider prefix

With USE_LITELLM_PROXY or litellm.use_litellm_proxy on, every call goes to the
upstream proxy, so the provider prefix no longer decides the cache format there
and resolution matches main. The plain OpenAI check reads the same provider
value as the hosted check, so one function no longer mixes two lookups

* fix(cache_control): resolve the chat-path dialect with the deployment api_base, as the call is routed

The hosted cache-format check now resolves the provider with the deployment's own api_base when the caller names none, the same provider the call dispatches to and the Responses path passes, so AZURE_AI_API_BASE no longer decides the chat format. The plain OpenAI check keeps main's name-only lookup.

* fix(cache_control): send acompletion fallbacks the caller's request, not the primary's hooked kwargs

acompletion handed async_completion_with_fallbacks the kwargs the cache seed and
prompt hook had already rewritten for the first model, so a Claude fallback behind
a Foundry GPT model received prompt_cache_options and got a 400. Fallbacks now get
a copy of the caller's model, messages, tools, and kwargs taken before the hooks,
as completion already does with its args.

Both cache seeds also read base_url before api_base, since completion sends the
call to base_url when both are set.

* fix(cache_control): keep the caller's tools list intact for acompletion fallbacks

The vector-store hook pops recognized file_search tools from the request's
tools list in place, so the fallback request built from the pre-hook snapshot
lost the retrieved context. Snapshot a copy of the list instead.

* fix(cache_control): skip acompletion's prompt hooks when fallbacks will run them

Each fallback attempt runs the prompt hooks for its own model, so running them
first in the outer call only cost a vector-store search whose result the
hand-off threw away.

* fix(cache_control): run acompletion's prompt hooks for fallbacks and leave only the cache hook to each attempt

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-30 05:11:42 -07:00 • committed by GitHub
parent 87fec5467f
commit a44e44b980
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 294 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
litellm/integrations/anthropic_cache_control_hook.py
View file

@ -116,12 +116,12 @@ def configured_injection_points(value: object) -> Sequence[CacheControlInjection
return tuple(cast(CacheControlInjectionPoint, entry) for entry in value if isinstance(entry, dict))
def _served_provider_for_model(model: str) -> str | None:
def _served_provider_for_model(model: str, api_base: object = None) -> str | None:
from litellm.exceptions import BadRequestError
from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
try:
_, provider, _, _ = get_llm_provider(model=model)
_, provider, _, _ = get_llm_provider(model=model, api_base=api_base if isinstance(api_base, str) else None)
except BadRequestError:
return None
return provider
@ -536,7 +536,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
if model is None:
return False
hosted_flag: Final = _hosted_openai_dialect_flag(
model, custom_llm_provider, AnthropicCacheControlHook._resolve_provider
model, custom_llm_provider, lambda name: AnthropicCacheControlHook._resolve_provider(name, api_base)
)
if hosted_flag is not None:
return hosted_flag
@ -559,8 +559,8 @@ class AnthropicCacheControlHook(CustomPromptManagement):
)
@staticmethod
def _resolve_provider(model: str) -> str | None:
return _served_provider_for_model(model)
def _resolve_provider(model: str, api_base: object = None) -> str | None:
return _served_provider_for_model(model, api_base)
@staticmethod
def count_request_cache_breakpoints(

40
litellm/main.py
View file

@ -681,7 +681,7 @@ async def acompletion(
- The `completion` function is called using `run_in_executor` to execute synchronously in the event loop.
- If `stream` is True, the function returns an async generator that yields completion lines.
"""
fallbacks = kwargs.get("fallbacks", None)
fallbacks: Final = kwargs.get("fallbacks", None) or litellm.model_fallbacks
mock_timeout = kwargs.get("mock_timeout", None)
if mock_timeout is True:
@ -697,6 +697,7 @@ async def acompletion(
loop: Final = asyncio.get_event_loop()
custom_llm_provider = kwargs.get("custom_llm_provider", None)
_ = _resolve_control_options(kwargs, model)
caller_kwargs: Final = cast("Mapping[str, object]", {**kwargs}) # cast-ok: acompletion takes untyped kwargs
## PROMPT MANAGEMENT HOOKS ##
#########################################################
@ -715,15 +716,34 @@ async def acompletion(
custom_llm_provider=cast(str | None, custom_llm_provider), # cast-ok: read from untyped kwargs
tools=tools,
enable_prompt_caching=cast(bool | None, kwargs.get("enable_prompt_caching")), # cast-ok: untyped kwargs
api_base=kwargs.get("api_base") or base_url,
api_base=base_url or kwargs.get("api_base"),
)
if isinstance(litellm_logging_obj, LiteLLMLoggingObj) and (
litellm_logging_obj.should_run_prompt_management_hooks(
prompt_id=kwargs.get("prompt_id", None),
prompt_id: Final = cast(str | None, kwargs.get("prompt_id", None)) # cast-ok: read from untyped kwargs
prompt_management_logger: Final = (
litellm_logging_obj.get_custom_logger_for_prompt_management(
model=model,
non_default_params=kwargs,
tools=tools,
prompt_id=prompt_id,
dynamic_callback_params=litellm_logging_obj.standard_callback_dynamic_params,
)
if isinstance(litellm_logging_obj, LiteLLMLoggingObj)
and litellm_logging_obj.should_run_prompt_management_hooks(
prompt_id=prompt_id,
non_default_params=kwargs,
tools=tools,
)
else None
)
each_fallback_writes_its_own_cache_dialect: Final = fallbacks is not None and isinstance(
prompt_management_logger, AnthropicCacheControlHook
)
if (
isinstance(litellm_logging_obj, LiteLLMLoggingObj)
and prompt_management_logger is not None
and not each_fallback_writes_its_own_cache_dialect
):
(
model,
@ -733,8 +753,9 @@ async def acompletion(
model=model,
messages=messages,
non_default_params=kwargs,
prompt_id=kwargs.get("prompt_id", None),
prompt_id=prompt_id,
prompt_variables=kwargs.get("prompt_variables", None),
prompt_management_logger=prompt_management_logger,
tools=tools,
prompt_label=kwargs.get("prompt_label", None),
prompt_version=kwargs.get("prompt_version", None),
@ -826,9 +847,10 @@ async def acompletion(
litellm_params=(GenericLiteLLMParams.model_validate(_supplemental) if _supplemental else None),
)
fallbacks = fallbacks or litellm.model_fallbacks
if fallbacks is not None:
response = await async_completion_with_fallbacks(**completion_kwargs, kwargs={"fallbacks": fallbacks, **kwargs})
response = await async_completion_with_fallbacks(
**completion_kwargs, kwargs={"fallbacks": fallbacks, **caller_kwargs}
)
if response is None:
raise Exception(
"No response from fallbacks. Got none. Turn on `litellm.set_verbose=True` to see more details."
@ -5765,7 +5787,7 @@ def completion(
custom_llm_provider=cast(str | None, kwargs.get("custom_llm_provider")), # cast-ok: untyped kwargs
tools=tools,
enable_prompt_caching=cast(bool | None, kwargs.get("enable_prompt_caching")), # cast-ok: untyped kwargs
api_base=kwargs.get("api_base") or base_url,
api_base=base_url or kwargs.get("api_base"),
)
if isinstance(litellm_logging_obj, LiteLLMLoggingObj) and (

71
tests/unit/integrations/test_anthropic_cache_control_hook.py
View file

@ -3998,6 +3998,77 @@ class TestChatPathProviderStamp:
assert params["prompt_cache_options"] == {"mode": "explicit"}
class TestChatPathResolvesTheDeploymentProvider:
MODEL = "azure_ai/gpt-6-astra"
AMBIENT_API_BASE = "https://other-deployment.openai.azure.com"
FOUNDRY_API_BASE = "https://foundry.services.ai.azure.com"
UPSTREAM_PROXY_API_BASE = "https://upstream-proxy.example.com"
POINTS = [{"location": "message", "role": "system"}]
MESSAGES = [{"role": "system", "content": "sys"}, {"role": "user", "content": "hi"}]
def _pin_ambient_api_base(self, monkeypatch: pytest.MonkeyPatch, pinned: bool) -> None:
if pinned:
monkeypatch.setenv("AZURE_AI_API_BASE", self.AMBIENT_API_BASE)
else:
monkeypatch.delenv("AZURE_AI_API_BASE", raising=False)
def _chat(
self, custom_llm_provider: str | None, api_base: str, model: str = MODEL
) -> tuple[list[AllMessageValues], dict[str, object]]:
seeded: Final[dict[str, object]] = {"cache_control_injection_points": copy.deepcopy(self.POINTS)}
AnthropicCacheControlHook.maybe_seed_default_injection_points(
non_default_params=seeded,
messages=copy.deepcopy(self.MESSAGES),
model=model,
custom_llm_provider=custom_llm_provider,
api_base=api_base,
)
_, out, params = AnthropicCacheControlHook().get_chat_completion_prompt(
model=model,
messages=copy.deepcopy(self.MESSAGES),
non_default_params=seeded,
prompt_id=None,
prompt_variables=None,
dynamic_callback_params={},
)
return out, params
def test_chat_path_stamps_the_breakpoint_on_a_foundry_deployment_while_the_ambient_base_is_azure_openai(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
self._pin_ambient_api_base(monkeypatch, True)
messages, params = self._chat(None, self.FOUNDRY_API_BASE)
assert (messages, params) == self._chat("azure_ai", self.FOUNDRY_API_BASE)
assert AnthropicCacheControlHook.count_request_cache_breakpoints(messages) == 1
assert ("prompt_cache_options" in params) == supports_openai_prompt_cache_breakpoint(self.MODEL, "azure_ai")
@pytest.mark.parametrize("pinned", [True, False])
@pytest.mark.parametrize(
"api_base",
[FOUNDRY_API_BASE, "https://resource.cognitiveservices.azure.com", "https://resource.openai.azure.com"],
)
def test_chat_path_marks_the_deployment_like_the_provider_its_api_base_dispatches_to(
self, monkeypatch: pytest.MonkeyPatch, api_base: str, pinned: bool
) -> None:
self._pin_ambient_api_base(monkeypatch, pinned)
dispatched_provider: Final = litellm.get_llm_provider(model=self.MODEL, api_base=api_base)[1]
assert self._chat(None, api_base) == self._chat(dispatched_provider, api_base)
@pytest.mark.parametrize("model", [MODEL, "bedrock_mantle/openai.gpt-5.6-sol"])
@pytest.mark.parametrize("set_globally", ["env", "module"])
def test_chat_path_marks_the_deployment_like_litellm_proxy_when_calls_route_to_one(
self, monkeypatch: pytest.MonkeyPatch, model: str, set_globally: str
) -> None:
self._pin_ambient_api_base(monkeypatch, False)
if set_globally == "env":
monkeypatch.setenv("USE_LITELLM_PROXY", "True")
else:
monkeypatch.setattr(litellm, "use_litellm_proxy", True)
assert self._chat(None, self.UPSTREAM_PROXY_API_BASE, model) == self._chat(
"litellm_proxy", self.UPSTREAM_PROXY_API_BASE, model
)
class TestClientBreakpointsCountedOnce:
def test_client_message_breakpoints_are_not_double_counted(self):
messages = [

187
tests/unit/litellm_core_utils/test_fallback_utils.py
View file

@ -1,6 +1,7 @@
"""Tests for litellm.litellm_core_utils.fallback_utils."""
import json
import pathlib
from typing import Final
import httpx
@ -8,12 +9,14 @@ import pytest
import respx
import litellm
from litellm.litellm_core_utils import litellm_logging
from litellm.litellm_core_utils.core_helpers import process_response_headers
from litellm.litellm_core_utils.fallback_utils import (
async_completion_with_fallbacks,
)
from litellm.main import acompletion as python_acompletion
from litellm.main import completion as python_completion
from litellm.vector_stores.vector_store_registry import LiteLLM_ManagedVectorStore, VectorStoreRegistry
@pytest.mark.asyncio
@ -444,3 +447,187 @@ def test_completion_fallbacks_sync(
assert response.choices[0].message.content == "sync fallback answer"
assert tuple(json.loads(call.request.content)["model"] for call in route.calls) == ("first", "second")
CACHE_POINTS: Final = ({"location": "message", "role": "system"},)
CACHED_MESSAGES: Final = (
{"role": "system", "content": "You are a careful assistant."},
{"role": "user", "content": "hello"},
)
OPENAI_ANSWER: Final = {
"id": "chatcmpl-1",
"object": "chat.completion",
"created": 1,
"model": "gpt-6.1-sol",
"choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
}
ANTHROPIC_ANSWER: Final = {
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "claude-haiku-4-5",
"content": [{"type": "text", "text": "ok"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 1, "output_tokens": 1},
}
def _cached_request(**kwargs: object) -> dict[str, object]:
return {
"messages": [dict(message) for message in CACHED_MESSAGES],
"cache_control_injection_points": [dict(point) for point in CACHE_POINTS],
**kwargs,
}
@pytest.mark.asyncio
async def test_acompletion_fallback_sends_the_fallback_model_the_request_a_direct_call_sends(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
openai_route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
return_value=httpx.Response(
401,
json={"error": {"message": "invalid key", "type": "authentication_error", "param": None, "code": None}},
)
)
anthropic_route: Final = respx_mock.post("https://api.anthropic.com/v1/messages").mock(
return_value=httpx.Response(200, json=ANTHROPIC_ANSWER)
)
await python_acompletion(**_cached_request(model="anthropic/claude-haiku-4-5", api_key="anthropic-key"))
await python_acompletion(
**_cached_request(
model="openai/gpt-6.1-sol",
api_key="openai-key",
fallbacks=[{"model": "anthropic/claude-haiku-4-5", "api_key": "anthropic-key"}],
)
)
assert "prompt_cache_options" in json.loads(openai_route.calls[0].request.content)
direct_body, fallback_body = (json.loads(call.request.content) for call in anthropic_route.calls)
assert fallback_body == direct_body
@pytest.mark.parametrize("is_async", [True, False])
@pytest.mark.asyncio
async def test_cache_dialect_follows_base_url_over_an_api_base_the_call_does_not_use(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, is_async: bool
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
return_value=httpx.Response(200, json=OPENAI_ANSWER)
)
async def send(**kwargs: object) -> None:
request: Final = _cached_request(
model="openai/gpt-6.1-sol", api_key="openai-key", base_url="https://api.openai.com/v1", **kwargs
)
if is_async:
await python_acompletion(**request)
else:
python_completion(**request)
await send()
await send(api_base="https://unused-gateway.example.com/v1")
base_url_only, with_unused_api_base = (json.loads(call.request.content) for call in route.calls)
assert with_unused_api_base == base_url_only
@pytest.mark.asyncio
async def test_acompletion_fallback_sends_the_context_a_vector_store_tool_retrieves(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
monkeypatch.setattr(
litellm,
"vector_store_registry",
VectorStoreRegistry(
vector_stores=[
LiteLLM_ManagedVectorStore(
vector_store_id="vs_docs", custom_llm_provider="openai", litellm_params={"api_key": "openai-key"}
)
]
),
)
search_route: Final = respx_mock.post("https://api.openai.com/v1/vector_stores/vs_docs/search").mock(
return_value=httpx.Response(
200,
json={
"object": "vector_store.search_results.page",
"search_query": "what is litellm?",
"data": [{"score": 1.0, "content": [{"type": "text", "text": "litellm is an LLM gateway"}]}],
},
)
)
openai_route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
return_value=httpx.Response(
401,
json={"error": {"message": "invalid key", "type": "authentication_error", "param": None, "code": None}},
)
)
anthropic_route: Final = respx_mock.post("https://api.anthropic.com/v1/messages").mock(
return_value=httpx.Response(200, json=ANTHROPIC_ANSWER)
)
async def send(**kwargs: object) -> None:
await python_acompletion(
messages=[{"role": "user", "content": "what is litellm?"}],
tools=[{"type": "file_search", "vector_store_ids": ["vs_docs"]}],
**kwargs,
)
await send(model="anthropic/claude-haiku-4-5", api_key="anthropic-key")
await send(
model="openai/gpt-6.1-sol",
api_key="openai-key",
fallbacks=[{"model": "anthropic/claude-haiku-4-5", "api_key": "anthropic-key"}],
)
direct_body, fallback_body = (json.loads(call.request.content) for call in anthropic_route.calls)
assert "litellm is an LLM gateway" in json.dumps(direct_body)
assert fallback_body == direct_body
assert search_route.call_count <= openai_route.call_count + anthropic_route.call_count
@pytest.mark.asyncio
async def test_acompletion_fallback_sends_the_prompt_the_primary_model_prefix_template_renders(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
) -> None:
_ = (tmp_path / "brief.prompt").write_text(
"---\ninput:\n schema:\n question: string\n---\nSystem: Answer in one word.\n\nUser: {{question}}\n"
)
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
monkeypatch.setattr(litellm, "global_prompt_directory", str(tmp_path))
monkeypatch.setattr(litellm_logging, "_in_memory_loggers", [])
_ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
return_value=httpx.Response(
401,
json={"error": {"message": "invalid key", "type": "authentication_error", "param": None, "code": None}},
)
)
anthropic_route: Final = respx_mock.post("https://api.anthropic.com/v1/messages").mock(
return_value=httpx.Response(200, json=ANTHROPIC_ANSWER)
)
async def send(**kwargs: object) -> None:
await python_acompletion(
prompt_id="brief",
prompt_variables={"question": "What color is the sky?"},
messages=[{"role": "user", "content": "placeholder"}],
**kwargs,
)
await send(model="dotprompt/anthropic/claude-haiku-4-5", api_key="anthropic-key")
await send(
model="dotprompt/openai/gpt-6.1-sol",
api_key="openai-key",
fallbacks=[{"model": "anthropic/claude-haiku-4-5", "api_key": "anthropic-key"}],
)
direct_body, fallback_body = (json.loads(call.request.content) for call in anthropic_route.calls)
assert "What color is the sky?" in json.dumps(direct_body)
assert fallback_body == direct_body