fix(router): pin /v1/responses requests for prompt caching affinity (#50321)

* fix(router): pin /v1/responses requests for prompt caching affinity

* fix(router): pin previous_response_id turns and keep native tools in the Responses pin key

A Responses turn that continues a stored response now goes to the deployment
that answered it, unless that deployment already failed this request. Native
Responses tools are hashed into the pin key as sent instead of converted, so
two requests with different native tools no longer share a pin and a native
request no longer logs a false dropped-tool warning. The pin keeps its
cacheable deployment ids in a sorted tuple so loggers can serialize request
metadata.

* fix(router): leave a Responses request with a malformed tool to the provider

* fix(router): keep the Responses prompt cache pin out of logged metadata

The pre-call check now keeps the pin in a router-owned in-flight store and
puts only a random token in litellm_metadata, so loggers that ship metadata
never see the routing key or deployment ids. The store is shared across check
instances, so turning prompt_caching off and on at runtime keeps pinning, and
a retry drops its previous attempt's token. The skipped-input debug log now
names only the error type.

* fix(router): apply the prompt caching pin after tag, region and pre-call filters, and leave it on retries

The prompt caching check no longer narrows the candidates inside the callback
filters. It names the deployment it prefers, and the router core applies that
preference at the pinned deployment step, after tag, region, pre-call and plugin
filters, only when the deployment is still a candidate. A pin can no longer
send a request to a deployment its tags do not allow.

A retry skips the pin when the pinned deployment was already tried in this
request, read from the retry breadcrumbs in both metadata buckets and the
excluded and retry-skipped ids. When the retry count outruns the breadcrumbs
the check stands down, so a pinned deployment that keeps failing is not
retried on every attempt.

* test(router): list the prompt caching check as an in-memory cache caller

The check's in-flight Responses pins live in a process-local InMemoryCache, so it opens no Redis span and needs no key family, like the other in-memory callers

* fix(router): pin a lone healthy deployment, free every in-flight Responses pin, and clear a reused metadata dict's retry records

* test(router): lift test cooldowns through the cooldown store's evict_local

* fix(router): key Responses tools like their chat twin and drop the pin token after success

A Responses function tool now hashes only the function fields it sent, so a chat request and a Responses request with the same tool share one pin. The success event also removes the pin token from the caller's metadata dict.

* test(router): cover untyped tool parameters and custom tool grammars in the Responses pin key

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
tessellark 2026-11-06 12:56:54 -08:00 • committed by GitHub
parent aeede53112
commit b5a72c2042
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 1486 additions and 113 deletions

View file

@ -20,10 +20,10 @@ use std::{
use async_trait::async_trait;
use litellm_router_core::{
Attempt, AttemptError, Blocked, BuildAsk, BuildPath, Built, Deployment, Error, Host, Output,
Payload, PreRoute, PreRouted, Provided, Randomness, RouterCache, RouterEvent, RouterRequest,
RouterResponse, RoutingContext, SecretRef, SelectStage, ThreadRandomness,
build::BuiltDeployment, host::ModelInfoLookup, model_info::MergedModelInfo,
Attempt, AttemptError, Blocked, BuildAsk, BuildPath, Built, CallbackFiltered, Deployment,
Error, Host, Output, Payload, PreRoute, PreRouted, Provided, Randomness, RouterCache,
RouterEvent, RouterRequest, RouterResponse, RoutingContext, SecretRef, SelectStage,
ThreadRandomness, build::BuiltDeployment, host::ModelInfoLookup, model_info::MergedModelInfo,
precall::PreCallFacts,
};
use pyo3::{PyResult, Python, types::PyAnyMethods};
@ -298,8 +298,9 @@ pub(crate) enum HostCallResponse {
PreCall(PreCallFacts),
/// The candidate models a `routing_plugins` select step reported.
Models(Option<Vec<String>>),
/// The ids a `callback_filters` select step left, or the error a filter raised.
Filtered(Result<Vec<String>, AttemptError>),
/// The ids a `callback_filters` select step left with the one a filter prefers, or the error
/// a filter raised.
Filtered(Result<(Vec<String>, Option<String>), AttemptError>),
/// What a pick's select or pre-route step, or a fallback check, raised on the host instead of
/// answering: the walk takes it as the try's failure.
Raised(AttemptError),
@ -352,8 +353,10 @@ impl HostCallResponse {
got: self.kind(),
});
}
if let (Self::Select(ids) | Self::Filtered(Ok(ids)), HostCall::Select { candidates, .. }) =
(self, call)
if let (
Self::Select(ids) | Self::Filtered(Ok((ids, _))),
HostCall::Select { candidates, .. },
) = (self, call)
&& let Some(id) = ids.iter().find(|id| !candidates.contains(id))
{
return Err(DriveError::UnknownCandidate { id: id.clone() });
@ -890,7 +893,7 @@ impl Host for StepHost {
&self,
ctx: &RoutingContext<'_>,
candidates: Vec<Arc<Deployment>>,
) -> Result<Vec<Arc<Deployment>>, Error> {
) -> Result<CallbackFiltered, Error> {
let (scope, group) = ctx.scope.wire();
let joined = |own: &[String], walked: &[String]| [own, walked].concat();
let call = HostCall::Select {
@ -909,8 +912,11 @@ impl Host for StepHost {
let HostCallResponse::Filtered(result) = call_host(call).await else {
unreachable!("{CHECKED}");
};
let ids = result.map_err(Error::Attempt)?;
Ok(answered_by_id(&candidates, &ids))
let (ids, preferred) = result.map_err(Error::Attempt)?;
Ok(CallbackFiltered {
kept: answered_by_id(&candidates, &ids),
preferred,
})
}
async fn allow_fallback(

View file

@ -1181,7 +1181,8 @@ fn host_call_response(
stage: Stage::CallbackFilters { .. },
..
} => {
// The remaining ids, or a failure mapping as an attempt reports one.
// The remaining ids, alone or paired with the id a filter prefers, or a failure
// mapping as an attempt reports one.
let outcome = outcome.ok_or_else(missing)?;
if let Ok(failure) = outcome.cast::<PyDict>() {
return match attempt_outcome(failure)? {
@ -1191,7 +1192,10 @@ fn host_call_response(
)),
};
}
Ok(HostCallResponse::Filtered(Ok(outcome.extract()?)))
if let Ok(answer) = outcome.cast::<PyTuple>() {
return Ok(HostCallResponse::Filtered(Ok(answer.extract()?)));
}
Ok(HostCallResponse::Filtered(Ok((outcome.extract()?, None))))
}
HostCall::Select {
stage: Stage::Strategy { records, .. },

View file

@ -405,6 +405,12 @@ pub enum SelectStage {
OverrideCheck(RoutingStrategy),
}
#[derive(Debug)]
pub struct CallbackFiltered {
pub kept: Vec<Arc<Deployment>>,
pub preferred: Option<String>,
}
/// Routing events, delivered through [`Host::observe`]. They serialize as JSON objects tagged by
/// `event` (the variant name), durations in seconds.
#[derive(Clone, Debug, PartialEq, serde::Serialize)]
@ -714,14 +720,19 @@ pub trait Host: Send + Sync {
}
/// `async_callback_filter_deployments`: the host's routing callbacks narrow `candidates` in
/// turn. A filter that raises fails the try with [`Error::Attempt`], which the call retries and
/// falls back from like any attempt failure. A host without filters keeps every candidate.
/// turn, and may name one deployment they prefer, which the pick takes only when the filters
/// after the callbacks keep it. A filter that raises fails the try with [`Error::Attempt`],
/// which the call retries and falls back from like any attempt failure. A host without
/// filters keeps every candidate and prefers none.
async fn callback_filters(
&self,
_ctx: &RoutingContext<'_>,
candidates: Vec<Arc<Deployment>>,
) -> Result<Vec<Arc<Deployment>>, Error> {
Ok(candidates)
) -> Result<CallbackFiltered, Error> {
Ok(CallbackFiltered {
kept: candidates,
preferred: None,
})
}
/// The values of `refs`, one per ref, in order: what the deployment build substitutes for
/// each `os.environ/` reference it found in entry `index` ([`crate::build`]). Asked once per

View file

@ -124,8 +124,8 @@ pub use crate::{
error::{AttemptError, ConfigError, Error, FailureKind, Payload, Unsupported},
health::{Clock, Health, Outcome, SystemClock},
host::{
Attempt, Blocked, BuildAsk, BuildPath, Built, Host, Output, Provided, RetryCause,
RouterEvent, RouterRequest, RoutingContext, SecretAt, SecretRef, SelectStage,
Attempt, Blocked, BuildAsk, BuildPath, Built, CallbackFiltered, Host, Output, Provided,
RetryCause, RouterEvent, RouterRequest, RoutingContext, SecretAt, SecretRef, SelectStage,
},
outcome::{OutcomeCallbacks, OutcomeWrite},
prerouting::{PreRoute, PreRouted},

View file

@ -456,7 +456,7 @@ impl Core {
let request_order = request.target_order.filter(|_| !inputs.fallback_hop);
// `async_callback_filter_deployments` runs after the cooldown and blocked filters on every
// async candidate list but a deployment named by id and a listed pass-through pick.
if settings.callback_filters
let preferred = if settings.callback_filters
&& !inputs.count_only
&& !resolution.kind.is_single()
&& request.pass_through != Some(PassThrough::Listed)
@ -472,8 +472,12 @@ impl Core {
ctx.excluded_ids = &inputs.excluded_ids;
ctx.skipped_ids = &inputs.skipped_ids;
ctx.target_order = inputs.target_order.or(request_order);
healthy = self.host.callback_filters(&ctx, healthy).await?;
}
let filtered = self.host.callback_filters(&ctx, healthy).await?;
healthy = filtered.kept;
filtered.preferred
} else {
None
};
if inputs.count_only {
return Ok(Candidates {
name: name.to_owned(),
@ -648,8 +652,13 @@ impl Core {
healthy.retain(|d| !d.cannot_run_batches());
}
if let Some(pinned) = request.pinned_deployment_id.as_deref()
&& healthy.iter().any(|d| d.id == pinned)
if let Some(pinned) = [
request.pinned_deployment_id.as_deref(),
preferred.as_deref(),
]
.into_iter()
.flatten()
.find(|id| healthy.iter().any(|d| d.id == *id))
{
healthy.retain(|d| d.id == pinned);
}

View file

@ -199,6 +199,8 @@ ResponseTools: TypeAlias = Sequence[Mapping[str, object]] | None
ChatToolParam: TypeAlias = ChatCompletionToolParam | OpenAIMcpServerTool | AnthropicToolSearchToolRegex
NAMESPACE_DESCRIPTION_SEPARATOR: Final = "\n\n"
NAMESPACE_MEMBER_TYPES_WITH_CHAT_TOOLS: Final = frozenset({"function", "custom"})
RESPONSES_WEB_SEARCH_TOOL_TYPES: Final = frozenset({"web_search", "web_search_preview"})
RESPONSES_TOOL_TYPES_WITHOUT_CHAT_FORM: Final = frozenset({"computer_use", "image_generation", "local_shell", "shell"})
_INCOMPLETE_REASON_BY_FINISH_REASON: Final[Mapping[str, Literal["max_output_tokens", "content_filter"]]] = (
MappingProxyType({"length": "max_output_tokens", "content_filter": "content_filter", "refusal": "content_filter"})
)
@ -2948,7 +2950,7 @@ class LiteLLMCompletionResponsesConfig:
tool_type: Final = tool.get("type")
if tool_type == "mcp":
return ResponsesToolChatForm(chat_tools=(cast(OpenAIMcpServerTool, tool),), web_search_options=None)
if tool_type == "web_search_preview" or tool_type == "web_search":
if tool_type in RESPONSES_WEB_SEARCH_TOOL_TYPES:
_search_context_size: Final[Literal["low", "medium", "high"]] = cast(
Literal["low", "medium", "high"], tool.get("search_context_size")
)
@ -3001,7 +3003,7 @@ class LiteLLMCompletionResponsesConfig:
return ResponsesToolChatForm(chat_tools=(client_tool_search_chat_tool(tool),), web_search_options=None)
if tool_type == "tool_search":
return ResponsesToolChatForm(chat_tools=(anthropic_tool_search_chat_tool(),), web_search_options=None)
if tool_type in ("computer_use", "image_generation", "local_shell", "shell"):
if tool_type in RESPONSES_TOOL_TYPES_WITHOUT_CHAT_FORM:
verbose_logger.warning(
"Dropping Responses API tool of type '%s': it has no Chat Completions "
"equivalent and the target provider would reject the request.",

View file

@ -39,6 +39,7 @@ _WALK_KEYS: Final = (
_SERVED_KEYS: Final = ("model_info", "deployment")
# The walk state `run_async_fallback` hands every hop as kwargs.
_HOP_KWARGS: Final = ("fallback_depth", "max_fallbacks", "attempted_targets", "include_fallback_errors")
_RETRY_RECORD_KEYS: Final = ("previous_models", "request_retry_count")
def _reentrant_hop_stamps(request: Mapping[str, object], field: str) -> Mapping[str, object]:
@ -179,11 +180,13 @@ class WalkMetadata:
return
sibling: Final = request.get("metadata" if self._field == "litellm_metadata" else "litellm_metadata")
if isinstance(sibling, dict):
sibling.pop("attempted_fallbacks", None) # pyright: ignore[reportUnknownMemberType] # the caller's metadata dict
sibling.pop("original_model_group", None) # pyright: ignore[reportUnknownMemberType] # see above
for key in ("attempted_fallbacks", "original_model_group", *_RETRY_RECORD_KEYS):
sibling.pop(key, None) # pyright: ignore[reportUnknownMemberType] # the caller's metadata dict
bucket: Final = request.get(self._field)
if not isinstance(bucket, dict):
return
for key in _RETRY_RECORD_KEYS:
bucket.pop(key, None) # pyright: ignore[reportUnknownMemberType] # see above
bucket["attempted_fallbacks"] = 0
model: Final = request.get("model")
if model is not None:

View file

@ -44,7 +44,10 @@ from litellm.router_utils.pre_call_checks.deployment_affinity_check import (
)
from litellm.router_utils.pre_call_checks.encrypted_content_affinity_check import EncryptedContentAffinityCheck
from litellm.router_utils.pre_call_checks.model_rate_limit_check import ModelRateLimitingCheck
from litellm.router_utils.pre_call_checks.prompt_caching_deployment_check import PromptCachingDeploymentCheck
from litellm.router_utils.pre_call_checks.prompt_caching_deployment_check import (
PromptCachingDeploymentCheck,
new_responses_pins_in_flight,
)
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging
@ -130,6 +133,7 @@ class OptionalChecks:
redis_cache=cache.redis_cache,
in_memory_cache=InMemoryCache(max_size_in_memory=PROMPT_CACHE_PIN_CACHE_MAX_SIZE),
)
self.prompt_caching_pins_in_flight: Final = new_responses_pins_in_flight()
self._request_scoped_selectors: tuple[CustomLogger, ...] = ()
self._request_scoped_policies: tuple[CustomLogger, ...] = ()
# ``_slack_alerting_loggers``: the Slack alerting logger ``alerting_config`` builds, if any.
@ -167,6 +171,9 @@ class OptionalChecks:
group: Final = hosted.get_routing_group(resolved)
return group is not None and group.routing_strategy == "priority"
def _group_has_alternatives(self, model: str) -> bool:
return len(self._hosted._core_deployments_named(model) or ()) > 1 # pyright: ignore[reportPrivateUsage] # the host registry's name index
async def repin_session_after_success(
self, litellm_params: Mapping[str, object], response: object, cooldowns: "CooldownCache"
) -> None:
@ -291,7 +298,10 @@ class OptionalChecks:
callback: CustomLogger | None = None
if name == "prompt_caching":
callback = PromptCachingDeploymentCheck(
cache=self.prompt_caching_pin_cache, is_priority_group=self._is_priority_group
cache=self.prompt_caching_pin_cache,
is_priority_group=self._is_priority_group,
responses_pins_in_flight=self.prompt_caching_pins_in_flight,
group_has_alternatives=self._group_has_alternatives,
)
elif name == "router_budget_limiting":
if self.budget_limiter() is not None:
@ -448,16 +458,21 @@ class OptionalChecks:
deployments: list[dict[str, object]], # mutable-ok: CustomLogger deployment-list contract
messages: object,
request: Mapping[str, object],
) -> list[object]: # mutable-ok: as the filters return it
) -> tuple[list[object], str | None]: # mutable-ok: as the filters return it
"""``async_callback_filter_deployments``: each routing callback narrows the deployments in
turn; a filter's error is raised for the Rust core to decide on."""
turn, then the prompt caching check names the deployment it prefers among the rest, which
the Rust core takes only if its later filters keep it; a filter's error is raised for the
Rust core to decide on."""
remaining: list[object] = list(deployments) # mutable-ok: filter contract # rebind-ok: narrowed
base: Final[object] = CustomLogger.async_filter_deployments # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # untyped deployment dicts
request_kwargs: Final = dict(request)
for callback in tuple(self.routing_callbacks()):
callbacks: Final = tuple(self.routing_callbacks())
for callback in callbacks:
# The base filter returns its input unchanged: only overriding filters are run.
if not isinstance(callback, CustomLogger) or type(callback).async_filter_deployments is base: # pyright: ignore[reportUnknownMemberType] # see above
continue
if isinstance(callback, PromptCachingDeploymentCheck):
continue
remaining = await callback.async_filter_deployments( # pyright: ignore[reportUnknownMemberType, reportAssignmentType, reportUnknownVariableType] # see above
model=model,
healthy_deployments=remaining,
@ -465,7 +480,15 @@ class OptionalChecks:
request_kwargs=request_kwargs,
parent_otel_span=None,
)
return remaining
prompt_caching: Final = next((cb for cb in callbacks if isinstance(cb, PromptCachingDeploymentCheck)), None)
if prompt_caching is None:
return remaining, None
return remaining, await prompt_caching.async_preferred_deployment_id(
model,
remaining, # pyright: ignore[reportArgumentType] # RustCore's deployment dicts
messages, # pyright: ignore[reportArgumentType] # the request's messages, as given
request_kwargs,
)
# -- release --------------------------------------------------------------------------------
@ -670,15 +693,15 @@ class RustOptionalChecks:
async def _answer_callback_filters(
self, step: Mapping[str, object], request: Mapping[str, object], messages: object
) -> tuple[object, BaseException | None]:
"""A ``callback_filters`` step: the remaining ids, or a filter's error as an attempt
failure with the error itself, which the caller keeps to raise."""
"""A ``callback_filters`` step: the remaining ids and the one a filter prefers, or a
filter's error as an attempt failure with the error itself, which the caller keeps to raise."""
from litellm.router_rust.core import attempt_failure
candidates: Final = [str(id) for id in step.get("candidates") or ()] # pyright: ignore[reportGeneralTypeIssues, reportUnknownVariableType, reportUnknownArgumentType] # the step's id list
records: Final = self._records_of(candidates) # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType, reportUnknownVariableType] # RustCore's deployment dicts
narrowed: Final = _request_with_failed_ids(request, step)
try:
remaining: Final = await self._opt.filter_deployments(
remaining, preferred_id = await self._opt.filter_deployments(
str(step.get("model_group")),
records, # pyright: ignore[reportUnknownArgumentType] # see above
messages,
@ -689,7 +712,8 @@ class RustOptionalChecks:
# A filter's pin lands on the request kwargs the pick goes on to read.
if narrowed is not request and narrowed.get(_ENCRYPTED_PIN) and isinstance(request, dict):
request[_ENCRYPTED_PIN] = True # rebind-ok: the request dict the pick goes on to read
return filtered_ids(remaining, candidates), None # pyright: ignore[reportUnknownArgumentType] # see above
kept: Final = filtered_ids(remaining, candidates) # pyright: ignore[reportUnknownArgumentType] # see above
return (kept, preferred_id), None
__all__ = [

View file

@ -4,23 +4,37 @@ Check if prompt caching is valid for a given deployment
Route to previously cached model id, if valid
"""
from collections.abc import Callable, Mapping, Sequence
from typing import Final, TypedDict, cast
import secrets
from collections.abc import Awaitable, Callable, Iterator, Mapping, Sequence
from itertools import chain
from types import MappingProxyType
from typing import Final, NamedTuple, Protocol, TypedDict, cast
from pydantic import TypeAdapter, ValidationError
from typing_extensions import ReadOnly
from litellm import verbose_logger
from litellm.caching.dual_cache import DualCache
from litellm.caching.in_memory_cache import InMemoryCache
from litellm.constants import DEFAULT_MINIMUM_PROMPT_CACHE_TOKEN_COUNT
from litellm.integrations.anthropic_cache_control_hook import (
AllToolParamValues,
AnthropicCacheControlHook,
)
from litellm.integrations.custom_logger import CustomLogger, Span
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.litellm_core_utils.routing_common_utils import deployment_ids_failed_in_request
from litellm.litellm_core_utils.token_counter import offload_token_count
from litellm.llms.bedrock.common_utils import bedrock_cache_point_ttl_supported, bedrock_capability_model
from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolParam
from litellm.responses.litellm_completion_transformation.transformation import (
RESPONSES_TOOL_TYPES_WITHOUT_CHAT_FORM,
RESPONSES_WEB_SEARCH_TOOL_TYPES,
LiteLLMCompletionResponsesConfig,
ResponsesToolChatForm,
ResponseTools,
)
from litellm.responses.utils import ResponsesAPIRequestUtils
from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolParam, ResponseInputParam
from litellm.types.utils import CallTypes, StandardLoggingPayload
from litellm.utils import (
get_base_model_from_metadata,
@ -28,7 +42,7 @@ from litellm.utils import (
is_prompt_caching_valid_prompt,
)
from ..prompt_caching_cache import PromptCachingCache
from ..prompt_caching_cache import PROMPT_CACHE_PIN_TTL_SECONDS, PromptCachingCache
class _PromptCachingSuccessEventKwargs(TypedDict, total=False):
@ -36,7 +50,165 @@ class _PromptCachingSuccessEventKwargs(TypedDict, total=False):
litellm_params: ReadOnly[Mapping[str, object] | None]
def _get_min_token_count_for_deployments(healthy_deployments: list[dict]) -> int:
class _ResponsesPin(NamedTuple):
cache_key: str
marked_ttl_seconds: int
cacheable_deployment_ids: tuple[str, ...]
class ResponsesPinsInFlight(Protocol):
def get_cache(self, key: str) -> object: ...
def set_cache(self, key: str, value: object) -> None: ...
def delete_cache(self, key: str) -> None: ...
class _ResponsesPrompt(NamedTuple):
messages: list[AllMessageValues] # mutable-ok: the prefix helpers take chat messages as a list
tools: list[AllToolParamValues] | None # mutable-ok: the prefix helpers take chat tools as a list
key_tools: list[ChatCompletionToolParam] | None # mutable-ok: the prefix helpers take chat tools as a list
_RESPONSES_PIN_METADATA_KEY: Final = "responses_prompt_cache_pin"
_RESPONSES_PINS_IN_FLIGHT_LIMIT: Final = 10_000
_RESPONSES_PIN_LIFETIME_SECONDS: Final = 3600
_RETRY_RECORDS: Final = TypeAdapter(tuple[object, ...])
_RESPONSES_CALL_TYPES: Final = frozenset((CallTypes.responses.value, CallTypes.aresponses.value))
_RESPONSES_TOOL_TYPES_KEPT_AS_SENT: Final = RESPONSES_TOOL_TYPES_WITHOUT_CHAT_FORM | RESPONSES_WEB_SEARCH_TOOL_TYPES
def new_responses_pins_in_flight() -> ResponsesPinsInFlight:
return InMemoryCache(
max_size_in_memory=_RESPONSES_PINS_IN_FLIGHT_LIMIT, default_ttl=_RESPONSES_PIN_LIFETIME_SECONDS
)
def _key_chat_tool(source: Mapping[str, object], chat_tool: Mapping[str, object]) -> Mapping[str, object]:
function: Final = as_str_mapping(chat_tool.get("function"))
if source.get("type") != "function" or function is None:
return chat_tool
return {**chat_tool, "function": {key: source[key] for key in function if key in source}}
def _key_chat_tools(
sources: Sequence[Mapping[str, object]], forms: Sequence[ResponsesToolChatForm]
) -> Iterator[Mapping[str, object]]:
for source, form in zip(sources, forms, strict=True):
yield from (_key_chat_tool(source, chat_tool) for chat_tool in form.chat_tools)
def _responses_prompt(request_data: Mapping[str, object]) -> _ResponsesPrompt | None:
request_input: Final = request_data.get("input")
request_tools: Final = request_data.get("tools")
if request_data.get("previous_response_id") is not None or not isinstance(request_input, str | list):
return None
listed_tools: Final = (
cast(ResponseTools, request_tools) # cast-ok: a bad tool raises inside the guard below and is caught
if isinstance(request_tools, list)
else None
)
try:
tools_kept_as_sent: Final = tuple(
tool for tool in listed_tools or () if tool.get("type") in _RESPONSES_TOOL_TYPES_KEPT_AS_SENT
)
convertible_tools: Final = (
[tool for tool in listed_tools if tool.get("type") not in _RESPONSES_TOOL_TYPES_KEPT_AS_SENT]
if listed_tools is not None
else None
)
messages: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages( # pyright: ignore[reportUnknownMemberType] # its input type is untyped
input=cast(str | ResponseInputParam, request_input), # cast-ok: a bad item raises and is caught here
responses_api_request={"instructions": request_data.get("instructions")},
item_cache_control=True,
only_since_compaction=True,
)
forms: Final = LiteLLMCompletionResponsesConfig.responses_tools_to_chat_forms(convertible_tools)
except Exception as error: # noqa: BLE001 # the provider judges the input, a routing preference never does
verbose_logger.debug("Prompt Caching: skipping an input with no chat form: %s", type(error).__name__)
return None
tools: Final = list(chain.from_iterable(form.chat_tools for form in forms))
key_tools: Final = [*_key_chat_tools(convertible_tools or (), forms), *tools_kept_as_sent]
return _ResponsesPrompt(
messages=cast(list[AllMessageValues], messages), # cast-ok: the bridge's chat messages, read as plain JSON
tools=cast(list[AllToolParamValues], tools) or None, # cast-ok: the bridge's chat tools, read as plain JSON
key_tools=cast(list[ChatCompletionToolParam], key_tools) or None, # cast-ok: tools are hashed as plain JSON
)
def _retry_records(metadata: Mapping[str, object]) -> tuple[object, ...]:
try:
return _RETRY_RECORDS.validate_python(metadata.get("previous_models", ()))
except ValidationError:
return ()
def _record_deployment_ids(records: tuple[object, ...]) -> Iterator[str]:
deployment_ids: Final = ((as_str_mapping(record) or {}).get("deployment_id") for record in records)
return (deployment_id for deployment_id in deployment_ids if isinstance(deployment_id, str))
def _retried_deployment_ids(metadata: Mapping[str, object]) -> frozenset[str] | None:
records: Final = _retry_records(metadata)
retry_count: Final = metadata.get("request_retry_count")
if type(retry_count) is int and retry_count > len(records):
return None
return frozenset(_record_deployment_ids(records))
def _deployment_ids_tried_in_request(request_data: Mapping[str, object]) -> frozenset[str] | None:
retried: Final = tuple(
_retried_deployment_ids(as_str_mapping(request_data.get(bucket)) or {})
for bucket in ("litellm_metadata", "metadata")
)
known: Final = tuple(ids for ids in retried if ids is not None)
if len(known) < len(retried):
return None
return deployment_ids_failed_in_request(request_data).union(*known)
def _previous_response_model_id(request_data: Mapping[str, object]) -> str | None:
previous_response_id: Final = request_data.get("previous_response_id")
model_id: Final = (
ResponsesAPIRequestUtils.get_model_id_from_response_id(previous_response_id)
if isinstance(previous_response_id, str)
else None
)
tried: Final = _deployment_ids_tried_in_request(request_data) if model_id is not None else None
return model_id if model_id is not None and tried is not None and model_id not in tried else None
def _deployment_min_tokens(deployment: Mapping[str, Mapping[str, object]], fallback: int) -> int:
deployment_model: Final = deployment.get("litellm_params", {}).get("model")
return get_prompt_cache_min_tokens(model=deployment_model) if isinstance(deployment_model, str) else fallback
async def _cacheable_deployment_ids(
healthy_deployments: Sequence[Mapping[str, Mapping[str, object]]],
reached_minimum: int,
reaches: Callable[[int], Awaitable[bool]],
) -> tuple[str, ...]:
minimums: Final = tuple(
(str(deployment["model_info"]["id"]), _deployment_min_tokens(deployment, reached_minimum))
for deployment in healthy_deployments
)
reached: Final = MappingProxyType(
{
minimum: minimum <= reached_minimum or await reaches(minimum)
for minimum in frozenset(minimum for _, minimum in minimums)
}
)
return tuple(sorted(deployment_id for deployment_id, minimum in minimums if reached[minimum]))
def _deployment_models(healthy_deployments: Sequence[Mapping[str, object]]) -> Iterator[str]:
models: Final = (
(as_str_mapping(deployment.get("litellm_params")) or {}).get("model") for deployment in healthy_deployments
)
return (model for model in models if isinstance(model, str) and model)
def _get_min_token_count_for_deployments(healthy_deployments: Sequence[Mapping[str, object]]) -> int:
"""
Returns the lowest minimum cacheable prefix across a model group.
@ -52,11 +224,7 @@ def _get_min_token_count_for_deployments(healthy_deployments: list[dict]) -> int
losing a cache hit it had earned. The lowest can only cost a lookup that finds nothing.
"""
return min(
(
get_prompt_cache_min_tokens(model=deployment["litellm_params"]["model"])
for deployment in healthy_deployments
if deployment.get("litellm_params", {}).get("model")
),
(get_prompt_cache_min_tokens(model=model) for model in _deployment_models(healthy_deployments)),
default=DEFAULT_MINIMUM_PROMPT_CACHE_TOKEN_COUNT,
)
@ -75,10 +243,20 @@ def _honors_one_hour_cache_ttl(
class PromptCachingDeploymentCheck(CustomLogger):
def __init__(self, cache: DualCache, is_priority_group: Callable[[str], bool] | None = None):
def __init__(
self,
cache: DualCache,
is_priority_group: Callable[[str], bool] | None = None,
responses_pins_in_flight: ResponsesPinsInFlight | None = None,
group_has_alternatives: Callable[[str], bool] | None = None,
):
super().__init__()
self.cache = cache
self.is_priority_group = is_priority_group
self.group_has_alternatives: Final = group_has_alternatives
self.responses_pins_in_flight: Final = (
responses_pins_in_flight if responses_pins_in_flight is not None else new_responses_pins_in_flight()
)
async def async_filter_deployments(
self,
@ -88,43 +266,81 @@ class PromptCachingDeploymentCheck(CustomLogger):
request_kwargs: dict | None = None,
parent_otel_span: Span | None = None,
) -> list[dict]:
if self.is_priority_group is not None and self.is_priority_group(model):
return healthy_deployments
if request_kwargs is not None and request_kwargs.get("_target_order") is not None:
return healthy_deployments
if not healthy_deployments[1:]:
return healthy_deployments
preferred_id: Final = await self.async_preferred_deployment_id(
model, healthy_deployments, messages, request_kwargs
)
preferred: Final = [
deployment for deployment in healthy_deployments if deployment["model_info"]["id"] == preferred_id
]
return preferred[:1] or healthy_deployments
request_data: Final = (
cast( # cast-ok: request_kwargs is a bare dict; this only reads its keys
Mapping[str, object], request_kwargs
)
if request_kwargs is not None
async def async_preferred_deployment_id(
self,
model: str,
healthy_deployments: Sequence[Mapping[str, Mapping[str, object]]],
messages: list[AllMessageValues] | None, # mutable-ok: the request's messages list, read as is
request_data: Mapping[str, object] | None,
) -> str | None:
if self.is_priority_group is not None and self.is_priority_group(model):
return None
if request_data is not None and request_data.get("_target_order") is not None:
return None
pin_metadata: Final = (
request_data.get("litellm_metadata") if messages is None and request_data is not None else None
)
single_healthy: Final = not healthy_deployments[1:]
if single_healthy and not (
isinstance(pin_metadata, dict)
and self.group_has_alternatives is not None
and self.group_has_alternatives(model)
):
return None
previous_attempt_token: Final = (as_str_mapping(pin_metadata) or {}).get(_RESPONSES_PIN_METADATA_KEY)
previous_response_model_id: Final = (
_previous_response_model_id(request_data)
if request_data is not None and isinstance(pin_metadata, dict)
else None
)
if previous_response_model_id is not None:
return previous_response_model_id
responses_prompt: Final = (
await offload_token_count(_responses_prompt)(request_data)
if request_data is not None and isinstance(pin_metadata, dict)
else None
)
request_tools: Final = (
cast( # cast-ok: request_kwargs is untyped; the stand-down scan duck-types every tool it reads
responses_prompt.tools
if responses_prompt is not None
else cast( # cast-ok: request_kwargs is untyped; the stand-down scan duck-types every tool it reads
list[AllToolParamValues] | None, request_data.get("tools")
)
if request_data is not None
if request_data is not None and messages is not None
else None
)
request_messages: Final = responses_prompt.messages if responses_prompt is not None else messages
prompt_messages: Final = (
PromptCachingCache.prepend_system_prompt(
messages, request_data.get("system") if request_data is not None else None
request_messages, request_data.get("system") if request_data is not None else None
)
if messages is not None
if request_messages is not None
else None
)
if prompt_messages is not None and await offload_token_count(is_prompt_caching_valid_prompt)(
messages=prompt_messages,
model=model,
min_token_count=_get_min_token_count_for_deployments(healthy_deployments),
tools=cast( # cast-ok: AllToolParamValues widens the union for Anthropic tool shapes the counter duck-types
list[ChatCompletionToolParam] | None,
request_tools,
),
):
async def reaches(minimum: int) -> bool:
return await offload_token_count(
is_prompt_caching_valid_prompt
)(
messages=prompt_messages,
model=model,
min_token_count=minimum,
tools=cast( # cast-ok: AllToolParamValues widens the union for Anthropic tool shapes the counter duck-types
list[ChatCompletionToolParam] | None,
request_tools,
),
)
group_minimum: Final = _get_min_token_count_for_deployments(healthy_deployments)
if prompt_messages is not None and await reaches(group_minimum):
prompt_cache: Final = PromptCachingCache(
cache=self.cache,
)
@ -134,34 +350,49 @@ class PromptCachingDeploymentCheck(CustomLogger):
## be derived from the messages as they will be sent, not as they arrive here.
affinity_messages: Final = AnthropicCacheControlHook.messages_with_default_injections(
messages=prompt_messages,
models=(
deployment["litellm_params"]["model"]
for deployment in healthy_deployments
if isinstance(deployment.get("litellm_params"), dict) and deployment["litellm_params"].get("model")
),
models=_deployment_models(healthy_deployments),
tools=request_tools,
enable_prompt_caching=(
request_kwargs.get("enable_prompt_caching") is True if request_kwargs is not None else None
request_data.get("enable_prompt_caching") is True if request_data is not None else None
),
request_kwargs=request_kwargs,
request_kwargs=request_data,
)
model_id_dict: Final = await prompt_cache.async_get_model_id(
messages=affinity_messages,
tools=cast( # cast-ok: AllToolParamValues widens the union for Anthropic tool shapes
Sequence[ChatCompletionToolParam] | None,
request_tools,
),
prefix_tools: Final = (
responses_prompt.key_tools
if responses_prompt is not None
else cast( # cast-ok: AllToolParamValues widens the union for Anthropic tool shapes
Sequence[ChatCompletionToolParam] | None, request_tools
)
)
if model_id_dict is not None and model_id_dict["model_id"] not in deployment_ids_failed_in_request(
request_data or {}
):
model_id: Final = model_id_dict["model_id"]
for deployment in healthy_deployments:
if deployment["model_info"]["id"] == model_id:
return [deployment]
positions: Final = await PromptCachingCache.async_prefix_positions(affinity_messages, prefix_tools)
if responses_prompt is not None and isinstance(pin_metadata, dict) and positions:
if isinstance(previous_attempt_token, str):
self.responses_pins_in_flight.delete_cache(previous_attempt_token)
pin_token: Final = secrets.token_hex(16)
self.responses_pins_in_flight.set_cache(
pin_token,
_ResponsesPin(
cache_key=positions[-1].cache_key,
marked_ttl_seconds=await offload_token_count(PromptCachingCache.pin_ttl_seconds)(
affinity_messages, prefix_tools
),
cacheable_deployment_ids=await _cacheable_deployment_ids(
healthy_deployments,
group_minimum,
reaches,
),
),
)
pin_metadata[_RESPONSES_PIN_METADATA_KEY] = pin_token
if single_healthy:
return None
return healthy_deployments
model_id_dict: Final = await prompt_cache.async_get_model_id_at(positions)
tried: Final = _deployment_ids_tried_in_request(request_data or {}) if model_id_dict is not None else None
if model_id_dict is not None and tried is not None and model_id_dict["model_id"] not in tried:
return model_id_dict["model_id"]
return None
def exempt_from_message_redaction(self) -> bool:
return True
@ -173,6 +404,7 @@ class PromptCachingDeploymentCheck(CustomLogger):
start_time: object,
end_time: object,
) -> None:
recorded_pin: Final = self._take_responses_pin(kwargs)
standard_logging_object: Final[StandardLoggingPayload | None] = kwargs.get("standard_logging_object", None)
if standard_logging_object is None:
@ -180,6 +412,10 @@ class PromptCachingDeploymentCheck(CustomLogger):
call_type: Final = standard_logging_object["call_type"]
if call_type in _RESPONSES_CALL_TYPES:
await self._pin_responses_request(recorded_pin, kwargs, standard_logging_object)
return
if (
call_type != CallTypes.completion.value
and call_type != CallTypes.acompletion.value
@ -241,3 +477,33 @@ class PromptCachingDeploymentCheck(CustomLogger):
)
return
def _take_responses_pin(self, kwargs: _PromptCachingSuccessEventKwargs) -> object:
litellm_params: Final = kwargs.get("litellm_params") or {}
metadata: Final = litellm_params.get("litellm_metadata")
pin_token: Final = (as_str_mapping(metadata) or {}).get(_RESPONSES_PIN_METADATA_KEY)
if not isinstance(pin_token, str):
return None
if isinstance(metadata, dict):
metadata.pop(_RESPONSES_PIN_METADATA_KEY, None) # pyright: ignore[reportUnknownMemberType] # the caller's metadata dict
recorded_pin: Final = self.responses_pins_in_flight.get_cache(pin_token)
self.responses_pins_in_flight.delete_cache(pin_token)
return recorded_pin
async def _pin_responses_request(
self,
recorded_pin: object,
kwargs: _PromptCachingSuccessEventKwargs,
standard_logging_object: StandardLoggingPayload,
) -> None:
litellm_params: Final = kwargs.get("litellm_params") or {}
model_id: Final = standard_logging_object["model_id"]
if not isinstance(recorded_pin, _ResponsesPin) or model_id not in recorded_pin.cacheable_deployment_ids:
return
await PromptCachingCache(cache=self.cache).async_set_pin(
model_id,
recorded_pin.cache_key,
recorded_pin.marked_ttl_seconds
if _honors_one_hour_cache_ttl(standard_logging_object, litellm_params)
else PROMPT_CACHE_PIN_TTL_SECONDS,
)

View file

@ -432,18 +432,18 @@ class PromptCachingCache:
if not positions:
return
ttl: Final = (
await self.async_set_pin(
model_id,
positions[-1].cache_key,
await offload_token_count(PromptCachingCache.pin_ttl_seconds)(messages, tools)
if honors_one_hour_ttl
else PROMPT_CACHE_PIN_TTL_SECONDS
else PROMPT_CACHE_PIN_TTL_SECONDS,
)
self.cache.in_memory_cache.delete_cache(positions[-1].cache_key)
async def async_set_pin(self, model_id: str, cache_key: str, ttl: int) -> None:
self.cache.in_memory_cache.delete_cache(cache_key)
with service_target(_PROMPT_CACHE_PINS_TARGET):
await self.cache.async_set_cache(
positions[-1].cache_key,
PromptCachingCacheValue(model_id=model_id),
ttl=ttl,
)
await self.cache.async_set_cache(cache_key, PromptCachingCacheValue(model_id=model_id), ttl=ttl)
async def async_get_model_id(
self,
@ -455,7 +455,10 @@ class PromptCachingCache:
same way the provider cache does, so a breakpoint that moved forward since the last
turn still lands on the deployment whose cache holds the earlier prefix.
"""
cache_keys: Final = _lookback_keys(await PromptCachingCache.async_prefix_positions(messages, tools))
return await self.async_get_model_id_at(await PromptCachingCache.async_prefix_positions(messages, tools))
async def async_get_model_id_at(self, positions: tuple[PrefixPosition, ...]) -> PromptCachingCacheValue | None:
cache_keys: Final = _lookback_keys(positions)
if not cache_keys:
return None

View file

@ -2,7 +2,7 @@ from pathlib import Path
from typing import Final
import pytest
from integration._support.client import Gateway, eventually
from integration._support.client import Gateway, eventually, object_value
from integration._support.database import read_rows, scratch_database
from integration._support.process import owned_proxy_process
from integration._support.wire import Wire, wire_server
@ -24,6 +24,7 @@ from pydantic import JsonValue
_MODEL: Final = "affinity-claude"
_FOLLOW_UPS: Final = 24
_PINS_ROUTE: Final = "/memory-usage-in-mem-cache-items"
_LONG_SYSTEM: Final = "Answer from the tool results below. " + "lorem ipsum " * 1500
@ -91,3 +92,68 @@ def test_router_affinity_is_lost_when_auto_caching_stands_down_for_tool_call_mar
assert [status for status, _, _ in follow_ups] == [200] * _FOLLOW_UPS, [text for _, _, text in follow_ups]
assert {first_marker, *follow_up_markers} == served["left"] | served["right"]
assert all(served[side] & set(follow_up_markers) for side in served), served
def _responses_first_turn(marker: str) -> list[JsonValue]:
return [
{"role": "user", "content": "What is the weather in London?"},
{"type": "function_call", "call_id": "call_london", "name": "lookup_weather", "arguments": "{}"},
{"type": "function_call_output", "call_id": "call_london", "output": "sunny"},
{"role": "user", "content": final_text(marker)},
]
def _responses_body(session: str, request_input: list[JsonValue]) -> dict[str, JsonValue]:
return {
"model": _MODEL,
"instructions": f"{_LONG_SYSTEM} session {session}",
"tools": [{"type": "function", "name": "lookup_weather", "parameters": {"type": "object", "properties": {}}}],
"input": request_input,
}
def _pins(gateway: Gateway) -> dict[str, JsonValue]:
return object_value(object_value(gateway.request("GET", _PINS_ROUTE).json())["prompt_caching_pin_cache"])
@pytest.mark.timeout(240)
def test_responses_thread_holds_its_deployment_when_the_bridge_logs_chat_messages(
gateway: Gateway, tmp_path: Path
) -> None:
session: Final = new_marker()
first_marker: Final = new_marker()
follow_up_markers: Final = tuple(new_marker() for _ in range(_FOLLOW_UPS))
first_turn: Final = _responses_first_turn(first_marker)
with scratch_database() as database_url, wire_server(anthropic_peer) as left, wire_server(anthropic_peer) as right:
config: Final = owned_config(
tmp_path,
[
{**anthropic_deployment(_MODEL, left.url), "model_info": {"id": f"responses-left-{session}"}},
{**anthropic_deployment(_MODEL, right.url), "model_info": {"id": f"responses-right-{session}"}},
],
router_settings={"optional_pre_call_checks": ["prompt_caching"]},
)
with (
owned_proxy_process(
gateway,
tmp_path,
{"DATABASE_URL": database_url},
config=config,
remove_environment=("DATABASE_URL_READ_REPLICA",),
workers=1,
) as owned,
owned.gateway.scenario() as scenario,
):
key: Final = scenario.key(metadata={"enable_prompt_caching": True})
first: Final = owned.gateway.request("POST", "/v1/responses", _responses_body(session, first_turn), key=key)
assert first.status_code == 200, first.text
eventually(lambda: _pins(owned.gateway), bool, seconds=30)
follow_ups: Final = tuple(
owned.gateway.request(
"POST", "/v1/responses", _responses_body(session, _follow_up(first_turn, marker)), key=key
)
for marker in follow_up_markers
)
served: Final = {"left": _served(left), "right": _served(right)}
assert [response.status_code for response in follow_ups] == [200] * _FOLLOW_UPS, follow_ups[0].text
assert {first_marker, *follow_up_markers} in tuple(served.values()), served

View file

@ -81,6 +81,7 @@ _IN_MEMORY_ONLY_CALLERS: Final = frozenset(
"litellm/proxy/ui_crud_endpoints/latest_release_endpoints.py",
"litellm/responses/litellm_completion_transformation/transformation.py",
"litellm/router_utils/client_initalization_utils.py",
"litellm/router_utils/pre_call_checks/prompt_caching_deployment_check.py",
"litellm/router_utils/router_callbacks/track_deployment_metrics.py",
"litellm/secret_managers/cyberark_secret_manager.py",
"litellm/secret_managers/google_secret_manager.py",