scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

fix(responses): let cache-control injection reach the system prompt from instructions (#38120)

Browse source 
`AnthropicCacheControlHook` spends the configured injection points on the first
message list it is shown and drops the message points that matched nothing. That is
right when the messages it sees are the ones going upstream. It is wrong for
/v1/responses: the system prompt lives in `instructions`, which only becomes a system
message once the chat-completion bridge builds one, so a role-targeted point matched
nothing and was thrown away before the message it wanted existed. Injection silently
did nothing across the whole surface.

Hand those points back instead, stamped as judged, when the caller says its message
list is provisional. The stamp is what makes carrying them safe: without it the next
pass re-judges the points against messages this pass has already marked and stands the
whole configuration down. Callers holding the final messages -- /chat/completions and
/v1/messages -- do not raise the signal and keep dropping unmatched points as before.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
tin-berri 2026-08-26 16:43:26 -07:00 • committed by GitHub
parent d8edfb69c2
commit cebf0d6f21
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 319 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
litellm/integrations/anthropic_cache_control_hook.py
View file

@ -104,6 +104,13 @@ def _accepts_prompt_cache_breakpoint(block: object) -> bool:
return isinstance(block, dict) and block.get("type") in OPENAI_PROMPT_CACHE_BREAKPOINT_BLOCK_TYPES
# Set by a caller whose message list is not the one that goes upstream -- today the
# Responses API layer, whose `instructions` only becomes a system message further down.
# Tells this hook to hand role-targeted points to the pass holding the final messages
# rather than spending them on a list that is still missing some of their targets.
CARRY_UNMATCHED_MESSAGE_POINTS: Final = "_litellm_carry_unmatched_cache_control_points"
class AnthropicCacheControlHook(CustomPromptManagement):
def get_chat_completion_prompt(
self,
@ -128,6 +135,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
- non_default_params: dict - params with any global cache controls
"""
# Extract cache control injection points
carry_unmatched: Final = bool(non_default_params.pop(CARRY_UNMATCHED_MESSAGE_POINTS, False))
injection_points: Final[list[CacheControlInjectionPoint]] = non_default_params.pop(
"cache_control_injection_points", []
)
@ -161,12 +169,25 @@ class AnthropicCacheControlHook(CustomPromptManagement):
non_default_params.get("prompt_cache_options"),
)
)
# A provisional message list defers every role-targeted point to the pass holding
# the final one: a role with no message here may have one there, and settling all
# of them in one pass is what lets config order decide the shared breakpoint
# budget. An ordinal names a different message once a later layer builds its own
# list, so it is placed here or not at all.
carried_message_points: Final[Sequence[CacheControlMessageInjectionPoint]] = (
tuple(point for point in message_points if point.get("index") is None) if carry_unmatched else ()
)
applied_message_points: Final[Sequence[CacheControlMessageInjectionPoint]] = (
tuple(point for point in message_points if point.get("index") is not None)
if carry_unmatched
else tuple(message_points)
)
reserved_blocks: Final = (
1 if not openai_dialect and any(p.get("location") == "tool_config" for p in remaining_points) else 0
)
breakpoints_before: Final = AnthropicCacheControlHook._count_request_cache_breakpoints(processed_messages)
processed_messages = self._apply_message_injections(
points=message_points,
points=applied_message_points,
messages=processed_messages,
max_blocks=MAX_CACHE_CONTROL_BLOCKS - reserved_blocks,
openai_dialect=openai_dialect,
@ -177,10 +198,15 @@ class AnthropicCacheControlHook(CustomPromptManagement):
):
non_default_params.setdefault("prompt_cache_options", PromptCacheOptions(mode="explicit"))
# Pass through non-message injection points for provider-specific handling
if remaining_points:
# Points this pass did not place: non-message ones for the provider transform, and
# the deferred role-targeted ones. Deferring is what reaches the Responses API's
# `instructions`, which is only a system message once the bridge builds one. The
# judged stamp is what makes it safe: the next pass must not re-judge points
# against messages this pass already marked (see `_should_stand_down`).
carried_points: Final[Sequence[CacheControlInjectionPoint]] = (*remaining_points, *carried_message_points)
if carried_points:
non_default_params["cache_control_injection_points"] = AnthropicCacheControlHook._stamped_as_judged(
remaining_points
carried_points
)
return model, processed_messages, non_default_params
@ -218,7 +244,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
@staticmethod
def _apply_message_injections(
points: list[CacheControlMessageInjectionPoint],
points: Sequence[CacheControlMessageInjectionPoint],
messages: list[AllMessageValues],
max_blocks: int,
openai_dialect: bool = False,

124
litellm/responses/main.py
View file

@ -1,6 +1,7 @@
import asyncio
import contextvars
from collections.abc import Coroutine, Iterable, Mapping
from collections.abc import Coroutine, Generator, Iterable, Mapping
from contextlib import contextmanager
from functools import partial
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, cast
@ -13,6 +14,7 @@ from litellm.completion_extras.litellm_responses_transformation.transformation i
LiteLLMResponsesTransformationHandler,
)
from litellm.constants import request_timeout
from litellm.integrations.anthropic_cache_control_hook import CARRY_UNMATCHED_MESSAGE_POINTS
from litellm.litellm_core_utils.asyncify import run_async_function
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.prompt_templates.common_utils import (
@ -390,6 +392,60 @@ async def aresponses_api_with_mcp(
return response
def _bridges_to_chat_completions(
responses_api_provider_config: BaseResponsesAPIConfig | None, use_chat_completions_api: bool
) -> bool:
"""Whether the request reaches its provider as a chat completion, not a Responses call."""
return responses_api_provider_config is None or use_chat_completions_api is True
def _will_bridge_to_chat_completions(
model: str, custom_llm_provider: str | None, use_chat_completions_api: bool
) -> bool:
"""``_bridges_to_chat_completions`` for callers running before the provider config is resolved.
Resolving the config is a pure lookup, so this asks the same question the dispatch
asks rather than restating its condition. Both callers resolve the provider before
this runs, so the only way to be wrong is a prompt manager that moves the model
across the bridge boundary, which would leave the deferred points to a pass that
never comes.
"""
normalized_model: Final = _normalize_openai_chat_completions_responses_model(model)
if custom_llm_provider is None:
return True
return _bridges_to_chat_completions(
ProviderConfigManager.get_provider_responses_api_config(
model=normalized_model[0], provider=custom_llm_provider
),
use_chat_completions_api or normalized_model[1],
)
@contextmanager
def _prompt_management_sees_a_provisional_message_list(
kwargs: dict[str, Any], # mutable-ok: the signal is read and popped out of the caller's own kwargs
bridged: bool,
) -> Generator[None, None]:
"""Tell the cache-control hook that this layer's messages are not the ones sent upstream.
A Responses request keeps its system prompt in ``instructions``, which only becomes a
system message when the chat-completion bridge builds one, so a role-targeted point
is placed by the bridge's pass rather than this one.
Only raised for a request that will be bridged. A provider serving Responses natively
gets no second pass, so this layer is the last one that can place anything and handing
a point forward there drops it.
"""
if not bridged:
yield
return
kwargs[CARRY_UNMATCHED_MESSAGE_POINTS] = True
try:
yield
finally:
kwargs.pop(CARRY_UNMATCHED_MESSAGE_POINTS, None)
@client
async def aresponses(
input: str | ResponseInputParam,
@ -467,19 +523,25 @@ async def aresponses(
client_input: list[AllMessageValues] = [{"role": "user", "content": input}]
else:
client_input = [item for item in input if isinstance(item, dict) and "role" in item]
(
model,
merged_input,
merged_optional_params,
) = await litellm_logging_obj.async_get_chat_completion_prompt(
model=model,
messages=client_input,
non_default_params=kwargs,
prompt_id=prompt_id,
prompt_variables=prompt_variables,
prompt_label=kwargs.get("prompt_label", None),
prompt_version=kwargs.get("prompt_version", None),
)
with _prompt_management_sees_a_provisional_message_list(
kwargs,
bridged=_will_bridge_to_chat_completions(
model, custom_llm_provider, bool(kwargs.get("use_chat_completions_api"))
),
):
(
model,
merged_input,
merged_optional_params,
) = await litellm_logging_obj.async_get_chat_completion_prompt(
model=model,
messages=client_input,
non_default_params=kwargs,
prompt_id=prompt_id,
prompt_variables=prompt_variables,
prompt_label=kwargs.get("prompt_label", None),
prompt_version=kwargs.get("prompt_version", None),
)
input = cast(
str | ResponseInputParam,
ResponsesAPIRequestUtils.merge_prompt_management_input(
@ -566,6 +628,7 @@ def _apply_prompt_management_to_responses_call(
litellm_logging_obj: LiteLLMLoggingObj | None,
kwargs: dict[str, Any],
local_vars: dict[str, object],
use_chat_completions_api: bool,
) -> tuple[str | ResponseInputParam, str, str | None]:
async_merged: Final[Mapping[str, object] | None] = kwargs.pop("_async_prompt_merged_params", None)
if async_merged is not None:
@ -585,19 +648,23 @@ def _apply_prompt_management_to_responses_call(
if isinstance(litellm_logging_obj, LiteLLMLoggingObj) and litellm_logging_obj.should_run_prompt_management_hooks(
prompt_id=prompt_id, non_default_params=kwargs
):
(
model,
merged_input,
merged_optional_params,
) = litellm_logging_obj.get_chat_completion_prompt(
model=model,
messages=client_input,
non_default_params=kwargs,
prompt_id=prompt_id,
prompt_variables=prompt_variables,
prompt_label=kwargs.get("prompt_label", None),
prompt_version=kwargs.get("prompt_version", None),
)
with _prompt_management_sees_a_provisional_message_list(
kwargs,
bridged=_will_bridge_to_chat_completions(model, custom_llm_provider, use_chat_completions_api),
):
(
model,
merged_input,
merged_optional_params,
) = litellm_logging_obj.get_chat_completion_prompt(
model=model,
messages=client_input,
non_default_params=kwargs,
prompt_id=prompt_id,
prompt_variables=prompt_variables,
prompt_label=kwargs.get("prompt_label", None),
prompt_version=kwargs.get("prompt_version", None),
)
input = cast(
str | ResponseInputParam,
ResponsesAPIRequestUtils.merge_prompt_management_input(
@ -961,6 +1028,7 @@ def responses(
litellm_logging_obj=litellm_logging_obj,
kwargs=kwargs,
local_vars=local_vars,
use_chat_completions_api=use_chat_completions_api,
)
#########################################################
@ -1063,7 +1131,7 @@ def responses(
if _file_search_dispatch is not None:
return _file_search_dispatch
if responses_api_provider_config is None or use_chat_completions_api is True:
if _bridges_to_chat_completions(responses_api_provider_config, use_chat_completions_api):
return litellm_completion_transformation_handler.response_api_handler(
model=model,
input=input,

202
tests/test_litellm/responses/test_responses_api_request_body.py
View file

@ -41,9 +41,7 @@ def _minimal_responses_api_payload(response_id: str, model: str) -> dict:
"id": "msg_1",
"status": "completed",
"role": "assistant",
"content": [
{"type": "output_text", "text": "Done.", "annotations": []}
],
"content": [{"type": "output_text", "text": "Done.", "annotations": []}],
}
],
"parallel_tool_calls": True,
@ -83,9 +81,9 @@ class MockResponse:
def _assert_request_body_matches(request_body: dict, expected_body: dict) -> None:
for key, expected_value in expected_body.items():
assert key in request_body, f"Missing key in request body: {key}"
assert (
request_body[key] == expected_value
), f"Mismatch for key {key}: got {request_body[key]!r}, expected {expected_value!r}"
assert request_body[key] == expected_value, (
f"Mismatch for key {key}: got {request_body[key]!r}, expected {expected_value!r}"
)
@pytest.mark.asyncio
@ -100,9 +98,7 @@ async def test_aresponses_context_management_and_shell_request_body_matches_expe
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new_callable=AsyncMock,
) as mock_post:
mock_post.return_value = MockResponse(
_minimal_responses_api_payload("resp_ctx_shell_test", "gpt-4o"), 200
)
mock_post.return_value = MockResponse(_minimal_responses_api_payload("resp_ctx_shell_test", "gpt-4o"), 200)
await litellm.aresponses(
model="openai/gpt-4o",
@ -426,7 +422,18 @@ async def test_aresponses_websocket_strips_responses_routing_prefix_from_openai_
_INJECTION_POINT_INPUT = [{"role": "system", "content": "You are terse."}, {"role": "user", "content": "hi"}]
_SYSTEM_INJECTION_POINT = [{"location": "message", "role": "system"}]
_SYSTEM_POINT = {"location": "message", "role": "system"}
_USER_POINT = {"location": "message", "role": "user"}
_SYSTEM_INJECTION_POINT = [_SYSTEM_POINT]
_ANTHROPIC_MESSAGES_PAYLOAD = {
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-5",
"content": [{"type": "text", "text": "Done."}],
"stop_reason": "end_turn",
"usage": {"input_tokens": 10, "output_tokens": 5},
}
def _sent_body(mock_post) -> dict:
@ -598,3 +605,178 @@ def test_responses_custom_api_base_sends_no_openai_markers():
body = _sent_body(mock_post)
assert body["input"] == _INJECTION_POINT_INPUT
assert "prompt_cache_options" not in body
@pytest.mark.asyncio
async def test_injection_points_still_reach_a_native_responses_provider():
"""Providers that serve Responses natively never reach the chat-completions bridge,
so this layer is their only chance to inject and must keep doing so."""
injected_client = AsyncHTTPHandler()
mock_post = AsyncMock(return_value=MockResponse(_minimal_responses_api_payload("resp_native", "gpt-5.6"), 200))
injected_client.post = mock_post
await litellm.aresponses(
model="openai/gpt-5.6",
api_key="fake-api-key",
input=copy.deepcopy(_INJECTION_POINT_INPUT),
cache_control_injection_points=copy.deepcopy(_SYSTEM_INJECTION_POINT),
client=injected_client,
)
body = _sent_body(mock_post)
assert body["input"][0]["content"][0]["prompt_cache_breakpoint"] == {"mode": "explicit"}
assert "cache_control_injection_points" not in body
async def _bridged_body(mock_post, *, points, input, instructions="You are a documentation assistant."):
injected_client = AsyncHTTPHandler()
injected_client.post = mock_post
await litellm.aresponses(
model="anthropic/claude-sonnet-4-5",
api_key="fake-api-key",
instructions=instructions,
input=copy.deepcopy(input),
cache_control_injection_points=copy.deepcopy(points),
client=injected_client,
)
return _sent_body(mock_post)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"content",
[
pytest.param("hi", id="string-content"),
pytest.param([{"type": "input_text", "text": "hi there friend"}], id="list-content"),
],
)
@pytest.mark.parametrize(
"points",
[
pytest.param([_SYSTEM_POINT], id="system-only"),
pytest.param([_USER_POINT, _SYSTEM_POINT], id="mixed-user-and-system"),
],
)
async def test_instructions_are_marked_when_the_bridge_builds_the_system_message(points, content):
"""The system prompt lives in `instructions`, which is not a message until the bridge
builds one, so the point targeting it matches nothing at the Responses layer.
Carrying it forward is what marks it at all. Carrying it *stamped* is what keeps a
second point that did match from stranding it: without the stamp the next pass reads
litellm's own marks as client breakpoints and stands the whole configuration down.
"""
mock_post = AsyncMock(return_value=MockResponse(_ANTHROPIC_MESSAGES_PAYLOAD, 200))
body = await _bridged_body(mock_post, points=points, input=[{"role": "user", "content": content}])
assert body["system"][0]["cache_control"] == {"type": "ephemeral"}
@pytest.mark.asyncio
@pytest.mark.parametrize("instructions", [None, "You are a documentation assistant."])
async def test_positional_points_address_the_input_item_the_caller_indexed(instructions):
"""`index` counts the caller's `input` items, and the Responses layer is where that
list still is, so a matched positional point must be spent there and never re-resolved
against the bridge's list, where the system message shifts every ordinal by one."""
mock_post = AsyncMock(return_value=MockResponse(_ANTHROPIC_MESSAGES_PAYLOAD, 200))
body = await _bridged_body(
mock_post,
points=[{"location": "message", "index": 0}],
input=[{"role": "user", "content": [{"type": "input_text", "text": "hi there friend"}]}],
instructions=instructions,
)
assert body["messages"][0]["content"][0]["cache_control"] == {"type": "ephemeral"}
if instructions:
assert "cache_control" not in json.dumps(body["system"])
@pytest.mark.asyncio
async def test_out_of_bounds_positional_points_are_not_revived_by_a_longer_list():
"""An ordinal addresses the list in front of the pass that reads it.
Carrying one forward would re-resolve it against the bridge's longer list, where an
index that named nothing in the caller's `input` can land on a real message -- the
system prompt included. Positional points are resolved where they were written or not
at all.
"""
mock_post = AsyncMock(return_value=MockResponse(_ANTHROPIC_MESSAGES_PAYLOAD, 200))
body = await _bridged_body(
mock_post,
points=[{"location": "message", "index": 1}],
input=[{"role": "user", "content": [{"type": "input_text", "text": "only item"}]}],
)
assert "cache_control" not in json.dumps(body["system"])
assert "cache_control" not in json.dumps(body["messages"])
def _four_user_turns() -> list:
return [
item
for i in range(4)
for item in (
{"role": "user", "content": [{"type": "input_text", "text": f"msg{i}"}]},
{"role": "assistant", "content": [{"type": "output_text", "text": f"reply{i}", "annotations": []}]},
)
]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"points,instructions,system_marked,marked_messages",
[
pytest.param([_SYSTEM_POINT, _USER_POINT], "You are terse.", True, [0, 2, 4], id="earlier-point-wins"),
pytest.param([_USER_POINT, _SYSTEM_POINT], "You are terse.", False, [0, 2, 4, 6], id="reversed-order-reverses"),
pytest.param([_USER_POINT, _SYSTEM_POINT], None, False, [0, 2, 4, 6], id="target-never-built-costs-nothing"),
],
)
async def test_config_order_decides_who_wins_the_shared_breakpoint_budget(
points, instructions, system_marked, marked_messages
):
"""Injection points are honoured in config order, earlier ones winning scarce slots.
A role-targeted point is placed a pass later than a positional one, so the four
breakpoints it competes for are shared across both passes. Every role point being
settled in the pass that holds the final list -- rather than the earlier pass holding
a slot for one it cannot place -- is what keeps that competition ordered in both
directions, and what stops a point whose target is never built from costing anything.
"""
mock_post = AsyncMock(return_value=MockResponse(_ANTHROPIC_MESSAGES_PAYLOAD, 200))
body = await _bridged_body(mock_post, points=points, input=_four_user_turns(), instructions=instructions)
assert ("cache_control" in json.dumps(body.get("system", []))) is system_marked
assert [i for i, msg in enumerate(body["messages"]) if "cache_control" in json.dumps(msg)] == marked_messages
@pytest.mark.asyncio
async def test_a_native_responses_provider_places_every_point_itself():
"""A provider serving Responses natively gets no second pass.
This layer is the last one that can place anything, so handing a point forward here
drops it -- and an unmatchable point must not cost a matching one its slot either.
The request has to be known to be bridged before anything is deferred.
"""
input_items = _four_user_turns()
async def _marked_indices(points):
injected_client = AsyncHTTPHandler()
mock_post = AsyncMock(return_value=MockResponse(_minimal_responses_api_payload("resp_native", "gpt-5.6"), 200))
injected_client.post = mock_post
await litellm.aresponses(
model="openai/gpt-5.6",
api_key="fake-api-key",
input=copy.deepcopy(input_items),
cache_control_injection_points=copy.deepcopy(points),
client=injected_client,
)
body = _sent_body(mock_post)
return [i for i, item in enumerate(body["input"]) if "prompt_cache_breakpoint" in json.dumps(item)]
user_only = await _marked_indices([_USER_POINT])
# The system point can never match here: nothing turns `instructions` into a message
# on the native path, so it must not cost the user point a slot.
with_unmatchable_system = await _marked_indices([_SYSTEM_POINT, _USER_POINT])
assert user_only == [0, 2, 4, 6]
assert with_unmatchable_system == user_only