mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-10 08:28:25 +00:00
Merge 95d87d0d0d into 6254860560
This commit is contained in:
commit
10527b8b1e
3 changed files with 595 additions and 36 deletions
|
|
@ -2199,6 +2199,11 @@ DEFAULT_COMPETITOR_DISCOVERY_MODEL: Final = "gpt-4o-mini"
|
|||
ADVISOR_NATIVE_PROVIDERS: Final[frozenset] = frozenset({"anthropic"})
|
||||
# Hard cap on advisor iterations per request to prevent runaway loops.
|
||||
ADVISOR_MAX_USES: Final[int] = 5
|
||||
# Generic-router call types (non chat-completion) that are safe to mirror to a
|
||||
# `silent_model`. Only side-effect-free inference endpoints belong here: the same
|
||||
# router helper also serves file/fine-tuning/passthrough calls that must never be
|
||||
# replayed against a second deployment.
|
||||
SILENT_MODEL_MIRROR_ALLOWED_CALL_TYPES: Final[frozenset[str]] = frozenset({"aresponses", "anthropic_messages"})
|
||||
# Description injected into the synthetic advisor tool definition sent to non-native providers.
|
||||
ADVISOR_TOOL_DESCRIPTION: Final[str] = (
|
||||
"Consult a highly intelligent advisor model when you need expert guidance, "
|
||||
|
|
|
|||
|
|
@ -88,6 +88,7 @@ from litellm.constants import (
|
|||
ROUTING_REQUEST_TAGS_METADATA_KEY,
|
||||
RUNTIME_UPDATABLE_ROUTER_SETTINGS,
|
||||
SESSION_DEPLOYMENT_AFFINITY_TTL_METADATA_KEY,
|
||||
SILENT_MODEL_MIRROR_ALLOWED_CALL_TYPES,
|
||||
)
|
||||
from litellm.integrations.custom_guardrail import CustomGuardrail, is_guardrail_intervention
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
|
|
@ -1081,6 +1082,13 @@ class _StreamRetryOpened:
|
|||
max_retries: int
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class _GenericMirrorState:
|
||||
"""Shared by the attempts of one request, including mid-stream fallback hops."""
|
||||
|
||||
scheduled: bool = False
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _StreamRetriesExhausted:
|
||||
error: "MidStreamFallbackError"
|
||||
|
|
@ -1274,6 +1282,23 @@ def as_output_cap(value: object) -> int | None:
|
|||
return cap if cap >= 0 else None
|
||||
|
||||
|
||||
def _is_silent_experiment_marker(value: object) -> bool:
|
||||
"""True when a metadata dict carries the silent-experiment marker."""
|
||||
return isinstance(value, Mapping) and bool(
|
||||
cast(Mapping[str, object], value).get("is_silent_experiment", False) # cast-ok: narrowed by isinstance above
|
||||
)
|
||||
|
||||
|
||||
def _has_mcp_tool(tools: object) -> bool:
|
||||
"""True when a Responses/Messages `tools` list carries an MCP tool definition."""
|
||||
if not isinstance(tools, Sequence) or isinstance(tools, str):
|
||||
return False
|
||||
for tool in cast(Sequence[object], tools): # cast-ok: narrowed by isinstance above
|
||||
if isinstance(tool, Mapping) and cast(Mapping[str, object], tool).get("type") == "mcp": # cast-ok: narrowed
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def stamped_client_ceiling(request_kwargs: Mapping[str, object]) -> Mapping[str, int] | None:
|
||||
"""None means unstamped; an empty mapping means the original caller supplied no limit."""
|
||||
stamped: Final = next(
|
||||
|
|
@ -3604,7 +3629,12 @@ class PythonCore(DeprecatedRouterMembers):
|
|||
self._stamp_failed_deployment_id_with_effective_model_info(e, deployment, kwargs)
|
||||
raise e
|
||||
|
||||
def _get_silent_experiment_kwargs(self, **kwargs) -> dict:
|
||||
def _get_silent_experiment_kwargs(
|
||||
self,
|
||||
metadata_variable_name: str = "metadata",
|
||||
/,
|
||||
**kwargs: object, # kwargs-ok: the primary call's kwargs, forwarded verbatim into the silent call
|
||||
) -> dict:
|
||||
"""
|
||||
Prepare kwargs for a silent experiment by ensuring isolation from the primary call.
|
||||
|
||||
|
|
@ -3613,29 +3643,36 @@ class PythonCore(DeprecatedRouterMembers):
|
|||
parent_otel_span — an OTel Span that is not deepcopy-able). Force a shallow
|
||||
copy of the metadata dict so mutations (model_group, is_silent_experiment)
|
||||
never corrupt the main call's metadata.
|
||||
|
||||
`metadata_variable_name` selects the kwargs key that carries the router's own
|
||||
metadata: "metadata" for chat completions, "litellm_metadata" for calls routed
|
||||
through `_ageneric_api_call_with_fallbacks` (Responses API, Anthropic Messages),
|
||||
where `metadata` is the provider request-body field and must not be touched.
|
||||
"""
|
||||
from litellm.litellm_core_utils.core_helpers import safe_deep_copy
|
||||
|
||||
silent_kwargs: Final = safe_deep_copy(kwargs)
|
||||
silent_kwargs: Final[dict[str, object]] = cast( # cast-ok: safe_deep_copy is untyped # mutable-ok: fresh copy
|
||||
dict[str, object], safe_deep_copy(kwargs)
|
||||
)
|
||||
|
||||
# safe_deep_copy may fall back to the original metadata reference when
|
||||
# deepcopy fails (UserAPIKeyAuth.parent_otel_span is not deepcopy-able).
|
||||
# Detect this via identity check and force a shallow copy so that setting
|
||||
# model_group / is_silent_experiment on the silent dict doesn't corrupt
|
||||
# the primary call's metadata.
|
||||
original_metadata: Final = kwargs.get("metadata")
|
||||
if original_metadata is not None and silent_kwargs.get("metadata") is original_metadata:
|
||||
silent_kwargs["metadata"] = dict(original_metadata)
|
||||
|
||||
if "metadata" not in silent_kwargs:
|
||||
silent_kwargs["metadata"] = {}
|
||||
# Always shallow-copy the metadata dict so that setting model_group /
|
||||
# is_silent_experiment on the silent dict doesn't corrupt the primary call's.
|
||||
copied_metadata: Final = silent_kwargs.get(metadata_variable_name)
|
||||
silent_metadata: Final[dict[str, object]] = ( # mutable-ok: the silent call's own metadata dict
|
||||
dict(cast(Mapping[str, object], copied_metadata)) # cast-ok: narrowed by the isinstance check
|
||||
if isinstance(copied_metadata, Mapping)
|
||||
else {}
|
||||
)
|
||||
|
||||
# OTel spans are not safe to use across event loops. The silent
|
||||
# experiment runs in a new event loop, so strip the span to prevent
|
||||
# cross-loop tracing races or span corruption.
|
||||
silent_kwargs["metadata"].pop("litellm_parent_otel_span", None)
|
||||
silent_metadata.pop("litellm_parent_otel_span", None)
|
||||
|
||||
silent_kwargs["metadata"]["is_silent_experiment"] = True
|
||||
silent_metadata["is_silent_experiment"] = True
|
||||
silent_kwargs[metadata_variable_name] = silent_metadata
|
||||
|
||||
# Pop logging objects and call IDs to ensure a fresh logging context
|
||||
# This prevents collisions in the Proxy's database (spend_logs)
|
||||
|
|
@ -3664,13 +3701,18 @@ class PythonCore(DeprecatedRouterMembers):
|
|||
async for _ in response:
|
||||
pass
|
||||
|
||||
def _silent_experiment_completion(self, silent_model: str, messages: Sequence[Mapping[str, str]], **kwargs):
|
||||
def _silent_experiment_completion(
|
||||
self,
|
||||
silent_model: str,
|
||||
messages: Sequence[Mapping[str, str]],
|
||||
**kwargs: object, # kwargs-ok: the primary call's kwargs, forwarded verbatim into the silent call
|
||||
):
|
||||
"""
|
||||
Run a silent experiment in the background (thread).
|
||||
"""
|
||||
try:
|
||||
# Prevent infinite recursion if silent model also has a silent model
|
||||
if kwargs.get("metadata", {}).get("is_silent_experiment", False):
|
||||
if _is_silent_experiment_marker(kwargs.get("metadata")):
|
||||
return
|
||||
|
||||
messages = copy.deepcopy(messages)
|
||||
|
|
@ -4931,13 +4973,72 @@ class PythonCore(DeprecatedRouterMembers):
|
|||
wrapper_ref: Final[weakref.ref[FallbackAwareStreamWrapper]] = weakref.ref(wrapped_response)
|
||||
return wrapped_response
|
||||
|
||||
async def _silent_experiment_acompletion(self, silent_model: str, messages: Sequence[Mapping[str, str]], **kwargs):
|
||||
def _silent_experiment_generic_kwargs(
|
||||
self,
|
||||
**kwargs: object, # kwargs-ok: the primary call's kwargs, snapshotted before the router mutates them
|
||||
) -> Mapping[str, object] | None:
|
||||
"""
|
||||
Snapshot the primary call's kwargs for a generic-path silent experiment, or
|
||||
return None when the request must not be mirrored.
|
||||
|
||||
Called synchronously before the mirror task is scheduled: `create_task` only
|
||||
starts running once the caller yields, by which point
|
||||
`_update_kwargs_with_deployment` has merged the primary deployment's metadata
|
||||
and tags into `kwargs` in place. Copying here keeps the mirror on the caller's
|
||||
request. Router metadata for these call types lives in `litellm_metadata`;
|
||||
`metadata` is the provider request-body field (OpenAI Responses `metadata`,
|
||||
Anthropic `metadata`) and is left untouched so the marker never reaches the
|
||||
provider.
|
||||
"""
|
||||
# Prevent infinite recursion if the silent model also has a silent model
|
||||
if _is_silent_experiment_marker(kwargs.get("metadata")) or _is_silent_experiment_marker(
|
||||
kwargs.get("litellm_metadata")
|
||||
):
|
||||
return None
|
||||
# MCP tools may execute against the provider (Anthropic `mcp_servers`, Responses
|
||||
# MCP tools with require_approval="never"); replaying those is a side effect.
|
||||
if kwargs.get("mcp_servers") or _has_mcp_tool(kwargs.get("tools")):
|
||||
return None
|
||||
return {**self._get_silent_experiment_kwargs("litellm_metadata", **kwargs), "stream": False}
|
||||
|
||||
async def _silent_experiment_ageneric(
|
||||
self,
|
||||
silent_model: str,
|
||||
original_function: Callable[..., Awaitable[object]],
|
||||
silent_kwargs: Mapping[str, object],
|
||||
) -> None:
|
||||
"""
|
||||
Run a generic-path silent experiment (Responses API, Anthropic Messages) in the
|
||||
background. Counterpart of `_silent_experiment_acompletion`; `silent_kwargs`
|
||||
comes from `_silent_experiment_generic_kwargs`. `model_group` is re-stamped to
|
||||
`silent_model` by `_update_kwargs_before_fallbacks` inside the call, so metrics
|
||||
attribute the mirror to the silent model.
|
||||
"""
|
||||
try:
|
||||
verbose_router_logger.info("Starting silent experiment for model %s", silent_model)
|
||||
remaining_kwargs: Final = {key: value for key, value in silent_kwargs.items() if key != "attempt_function"}
|
||||
with mirrored_call():
|
||||
await self._ageneric_api_call_with_fallbacks(
|
||||
model=silent_model,
|
||||
original_function=original_function,
|
||||
attempt_function=None,
|
||||
**remaining_kwargs,
|
||||
)
|
||||
except Exception as e: # noqa: BLE001 - a background mirror must never break the primary request
|
||||
verbose_router_logger.error("Silent experiment failed for model %s: %s", silent_model, e)
|
||||
|
||||
async def _silent_experiment_acompletion(
|
||||
self,
|
||||
silent_model: str,
|
||||
messages: Sequence[Mapping[str, str]],
|
||||
**kwargs: object, # kwargs-ok: the primary call's kwargs, forwarded verbatim into the silent call
|
||||
):
|
||||
"""
|
||||
Run a silent experiment in the background.
|
||||
"""
|
||||
try:
|
||||
# Prevent infinite recursion if silent model also has a silent model
|
||||
if kwargs.get("metadata", {}).get("is_silent_experiment", False):
|
||||
if _is_silent_experiment_marker(kwargs.get("metadata")):
|
||||
return
|
||||
|
||||
messages = copy.deepcopy(messages)
|
||||
|
|
@ -6760,6 +6861,7 @@ class PythonCore(DeprecatedRouterMembers):
|
|||
kwargs["model"] = model
|
||||
kwargs["original_generic_function"] = original_function
|
||||
kwargs["original_function"] = attempt_function or self._ageneric_api_call_with_fallbacks_helper
|
||||
kwargs["_silent_experiment_state"] = _GenericMirrorState() # rebind-ok: request-owned attempt state
|
||||
if attempt_function is not None:
|
||||
controls: Final = per_request_fallback_controls(kwargs)
|
||||
kwargs[MID_STREAM_FALLBACK_CONTROLS_KEY] = controls # rebind-ok: forwarded to every hop
|
||||
|
|
@ -6807,12 +6909,21 @@ class PythonCore(DeprecatedRouterMembers):
|
|||
kwargs["endpoint"] = replace_path_segment(kwargs["endpoint"], model, replacement_model_name)
|
||||
return kwargs
|
||||
|
||||
async def _ageneric_api_call_with_fallbacks_helper(self, model: str, original_generic_function: Callable, **kwargs):
|
||||
async def _ageneric_api_call_with_fallbacks_helper(
|
||||
self,
|
||||
model: str,
|
||||
original_generic_function: Callable,
|
||||
_silent_experiment_state: object = None,
|
||||
**kwargs,
|
||||
):
|
||||
"""
|
||||
Helper function to make a generic LLM API call through the router, this allows you to use retries/fallbacks with litellm router
|
||||
"""
|
||||
|
||||
passthrough_on_no_deployment: Final = kwargs.pop("passthrough_on_no_deployment", False)
|
||||
mirror_state: Final = (
|
||||
_silent_experiment_state if isinstance(_silent_experiment_state, _GenericMirrorState) else None
|
||||
)
|
||||
function_name: Final = "_ageneric_api_call_with_fallbacks"
|
||||
handler_name: Final[str | None] = getattr(original_generic_function, "__name__", None)
|
||||
deployment = None # rebind-ok: pre-init so the except block can stamp a failure with no deployment picked
|
||||
|
|
@ -6834,6 +6945,37 @@ class PythonCore(DeprecatedRouterMembers):
|
|||
return await original_generic_function(model=model, **kwargs)
|
||||
raise e
|
||||
|
||||
deployment_has_mcp: Final = bool(deployment["litellm_params"].get("mcp_servers")) or _has_mcp_tool(
|
||||
deployment["litellm_params"].get("tools")
|
||||
)
|
||||
if deployment_has_mcp and _is_silent_experiment_marker(kwargs.get("litellm_metadata")):
|
||||
return None
|
||||
silent_model: Final = cast( # cast-ok: untyped deployment value is validated by _silent_experiment_targets
|
||||
object, deployment["litellm_params"].get("silent_model")
|
||||
)
|
||||
if (
|
||||
silent_model is not None
|
||||
and handler_name in SILENT_MODEL_MIRROR_ALLOWED_CALL_TYPES
|
||||
and not deployment_has_mcp
|
||||
and (mirror_state is None or not mirror_state.scheduled)
|
||||
):
|
||||
# Snapshot now: `_update_kwargs_with_deployment` below mutates kwargs in place,
|
||||
# and the task only starts running once this coroutine yields.
|
||||
for silent_target in _silent_experiment_targets(silent_model):
|
||||
silent_kwargs = self._silent_experiment_generic_kwargs(**kwargs)
|
||||
if silent_kwargs is not None:
|
||||
if mirror_state is not None:
|
||||
mirror_state.scheduled = True
|
||||
asyncio.create_task(
|
||||
self._silent_experiment_ageneric(
|
||||
silent_model=silent_target,
|
||||
original_function=cast( # cast-ok: generic handlers are async callables; the bare Callable annotation predates this
|
||||
Callable[..., Awaitable[object]], original_generic_function
|
||||
),
|
||||
silent_kwargs=silent_kwargs,
|
||||
)
|
||||
)
|
||||
|
||||
self._update_kwargs_with_deployment(
|
||||
deployment=deployment,
|
||||
kwargs=kwargs,
|
||||
|
|
@ -6843,6 +6985,8 @@ class PythonCore(DeprecatedRouterMembers):
|
|||
)
|
||||
|
||||
data: Final = deployment["litellm_params"].copy()
|
||||
# Router-only setting: must never reach the provider handler (#34890)
|
||||
data.pop("silent_model", None)
|
||||
model_name: Final = data["model"]
|
||||
self.total_calls[model_name] += 1
|
||||
|
||||
|
|
@ -6869,6 +7013,7 @@ class PythonCore(DeprecatedRouterMembers):
|
|||
if custom_llm_provider is not None:
|
||||
response_kwargs["custom_llm_provider"] = custom_llm_provider
|
||||
|
||||
response_kwargs.pop("silent_model", None)
|
||||
compacted_input: Final = await compact_to_fit(
|
||||
self._core_ctx,
|
||||
deployment,
|
||||
|
|
@ -6950,6 +7095,8 @@ class PythonCore(DeprecatedRouterMembers):
|
|||
carry_over_routed_deployment(live_kwargs=kwargs, snapshot=hop_kwargs)
|
||||
if kwargs.get("stream") and isinstance(response, BaseResponsesAPIStreamingIterator):
|
||||
return await self._aresponses_streaming_iterator(response=response, initial_kwargs=hop_kwargs)
|
||||
if response is None:
|
||||
raise TypeError("Responses request returned no response")
|
||||
return response
|
||||
|
||||
async def _aanthropic_messages_streaming_iterator(
|
||||
|
|
@ -7522,6 +7669,8 @@ class PythonCore(DeprecatedRouterMembers):
|
|||
response=cast("AsyncIterator[bytes]", response), # cast-ok: stream=True always returns a byte iterator
|
||||
initial_kwargs=hop_kwargs,
|
||||
)
|
||||
if response is None:
|
||||
raise TypeError("Anthropic Messages request returned no response")
|
||||
return response
|
||||
|
||||
async def _dispatch_generic_call_type(
|
||||
|
|
|
|||
|
|
@ -13,9 +13,10 @@ import pytest
|
|||
import respx
|
||||
|
||||
import litellm
|
||||
from litellm._internal_context import in_mirrored_call
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
|
||||
from litellm.router import Router
|
||||
from litellm.router import Router, _has_mcp_tool, _is_silent_experiment_marker
|
||||
from litellm.router import _silent_experiment_kwargs_snapshot
|
||||
from litellm.router import _silent_experiment_targets
|
||||
|
||||
|
|
@ -784,11 +785,7 @@ async def test_router_silent_experiment_acompletion():
|
|||
|
||||
# Find the silent call
|
||||
silent_call = next(
|
||||
(
|
||||
c
|
||||
for c in call_args_list
|
||||
if c[1].get("metadata", {}).get("is_silent_experiment") is True
|
||||
),
|
||||
(c for c in call_args_list if c[1].get("metadata", {}).get("is_silent_experiment") is True),
|
||||
None,
|
||||
)
|
||||
assert silent_call is not None
|
||||
|
|
@ -796,11 +793,7 @@ async def test_router_silent_experiment_acompletion():
|
|||
|
||||
# Find the primary call
|
||||
primary_call = next(
|
||||
(
|
||||
c
|
||||
for c in call_args_list
|
||||
if not c[1].get("metadata", {}).get("is_silent_experiment")
|
||||
),
|
||||
(c for c in call_args_list if not c[1].get("metadata", {}).get("is_silent_experiment")),
|
||||
None,
|
||||
)
|
||||
assert primary_call is not None
|
||||
|
|
@ -869,11 +862,7 @@ def test_router_silent_experiment_completion():
|
|||
|
||||
# Find the silent call
|
||||
silent_call = next(
|
||||
(
|
||||
c
|
||||
for c in call_args_list
|
||||
if c[1].get("metadata", {}).get("is_silent_experiment") is True
|
||||
),
|
||||
(c for c in call_args_list if c[1].get("metadata", {}).get("is_silent_experiment") is True),
|
||||
None,
|
||||
)
|
||||
assert silent_call is not None
|
||||
|
|
@ -909,8 +898,6 @@ def test_silent_experiment_sends_shadow_request_attributed_to_the_silent_model(r
|
|||
assert primary_metadata == {"model_group": "primary-model"}
|
||||
|
||||
|
||||
|
||||
|
||||
_EMBEDDING_API_BASE: Final = "https://embeddings.example.test/v1"
|
||||
|
||||
|
||||
|
|
@ -986,6 +973,356 @@ def test_silent_experiment_does_not_launch_from_a_shadow_request(run_silent_expe
|
|||
acompletion.assert_not_awaited()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Generic router path (Responses API / Anthropic Messages) — issues #31888, #34890
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _generic_silent_model_list():
|
||||
return [
|
||||
{
|
||||
"model_name": "primary-model",
|
||||
"litellm_params": {
|
||||
"model": "openai/gpt-4o-mini",
|
||||
"api_key": "fake-key",
|
||||
"silent_model": "silent-model",
|
||||
},
|
||||
},
|
||||
{
|
||||
"model_name": "silent-model",
|
||||
"litellm_params": {
|
||||
"model": "openai/gpt-4o",
|
||||
"api_key": "fake-key",
|
||||
},
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def _split_generic_calls(mock):
|
||||
"""Return (primary_call, silent_call) from a mocked generic handler."""
|
||||
silent = next(
|
||||
(c for c in mock.call_args_list if (c.kwargs.get("litellm_metadata") or {}).get("is_silent_experiment")),
|
||||
None,
|
||||
)
|
||||
primary = next(
|
||||
(c for c in mock.call_args_list if not (c.kwargs.get("litellm_metadata") or {}).get("is_silent_experiment")),
|
||||
None,
|
||||
)
|
||||
return primary, silent
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_silent_experiment_aresponses():
|
||||
"""
|
||||
Regression for #31888: silent_model on a deployment must fire a background
|
||||
aresponses call when the primary request goes through /v1/responses.
|
||||
"""
|
||||
mock_aresponses = AsyncMock(return_value=MagicMock())
|
||||
mock_aresponses.__name__ = "aresponses"
|
||||
|
||||
router = Router(model_list=_generic_silent_model_list())
|
||||
router.aresponses = router.factory_function(mock_aresponses, call_type="aresponses")
|
||||
await router.aresponses(
|
||||
model="primary-model",
|
||||
input=[{"role": "user", "content": "hi"}],
|
||||
metadata={"trace": "user-supplied"},
|
||||
)
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
assert mock_aresponses.call_count == 2
|
||||
primary_call, silent_call = _split_generic_calls(mock_aresponses)
|
||||
assert primary_call is not None
|
||||
assert silent_call is not None
|
||||
|
||||
assert primary_call.kwargs["model"] == "openai/gpt-4o-mini"
|
||||
assert primary_call.kwargs["litellm_metadata"]["model_group"] == "primary-model"
|
||||
|
||||
assert silent_call.kwargs["model"] == "openai/gpt-4o"
|
||||
# model_group is overridden so metrics/logging attribute the mirror to the silent model
|
||||
assert silent_call.kwargs["litellm_metadata"]["model_group"] == "silent-model"
|
||||
|
||||
# The router-only setting must never reach the provider handler (#34890)
|
||||
for call in mock_aresponses.call_args_list:
|
||||
assert "silent_model" not in call.kwargs
|
||||
|
||||
# `metadata` is the OpenAI Responses request-body field: the marker must not be
|
||||
# written there, and the caller's value must pass through untouched.
|
||||
assert silent_call.kwargs["metadata"] == {"trace": "user-supplied"}
|
||||
assert primary_call.kwargs["metadata"] == {"trace": "user-supplied"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_silent_experiment_anthropic_messages():
|
||||
"""
|
||||
Regression for #34890: same contract for /v1/messages. Anthropic validates the
|
||||
`metadata` body field strictly, so the marker must live in litellm_metadata.
|
||||
"""
|
||||
mock_messages = AsyncMock(return_value=MagicMock())
|
||||
mock_messages.__name__ = "anthropic_messages"
|
||||
|
||||
router = Router(model_list=_generic_silent_model_list())
|
||||
router.aanthropic_messages = router.factory_function(mock_messages, call_type="anthropic_messages")
|
||||
await router.aanthropic_messages(
|
||||
model="primary-model",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
max_tokens=16,
|
||||
)
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
assert mock_messages.call_count == 2
|
||||
primary_call, silent_call = _split_generic_calls(mock_messages)
|
||||
assert primary_call is not None
|
||||
assert silent_call is not None
|
||||
assert primary_call.kwargs["model"] == "openai/gpt-4o-mini"
|
||||
assert silent_call.kwargs["model"] == "openai/gpt-4o"
|
||||
assert silent_call.kwargs["litellm_metadata"]["model_group"] == "silent-model"
|
||||
for call in mock_messages.call_args_list:
|
||||
assert "silent_model" not in call.kwargs
|
||||
assert "is_silent_experiment" not in (call.kwargs.get("metadata") or {})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_silent_experiment_generic_does_not_corrupt_primary_metadata():
|
||||
"""
|
||||
The primary call's litellm_metadata must not see the silent marker or the
|
||||
silent model_group, even when deepcopy falls back to a shared reference.
|
||||
"""
|
||||
mock_aresponses = AsyncMock(return_value=MagicMock())
|
||||
mock_aresponses.__name__ = "aresponses"
|
||||
|
||||
router = Router(model_list=_generic_silent_model_list())
|
||||
router.aresponses = router.factory_function(mock_aresponses, call_type="aresponses")
|
||||
litellm_metadata = {
|
||||
"user_api_key_auth": _FakeUserAPIKeyAuth(
|
||||
key_alias="primary-key",
|
||||
parent_otel_span=_NonCopyableSpan(),
|
||||
)
|
||||
}
|
||||
await router.aresponses(
|
||||
model="primary-model",
|
||||
input=[{"role": "user", "content": "hi"}],
|
||||
litellm_metadata=litellm_metadata,
|
||||
)
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
assert mock_aresponses.call_count == 2
|
||||
primary_call, silent_call = _split_generic_calls(mock_aresponses)
|
||||
assert primary_call is not None
|
||||
assert silent_call is not None
|
||||
assert primary_call.kwargs["litellm_metadata"]["model_group"] == "primary-model"
|
||||
assert "is_silent_experiment" not in primary_call.kwargs["litellm_metadata"]
|
||||
assert silent_call.kwargs["litellm_metadata"]["model_group"] == "silent-model"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_silent_experiment_skips_non_allowlisted_generic_call_types():
|
||||
"""
|
||||
Only side-effect-free inference call types are mirrored. The same generic
|
||||
helper serves file/fine-tuning/passthrough calls that must not be replayed
|
||||
against the silent deployment.
|
||||
"""
|
||||
sentinel_response = MagicMock()
|
||||
mock_file_content = AsyncMock(return_value=sentinel_response)
|
||||
mock_file_content.__name__ = "afile_content"
|
||||
|
||||
router = Router(model_list=_generic_silent_model_list())
|
||||
router.afile_content = router.factory_function(mock_file_content, call_type="afile_content")
|
||||
response = await router.afile_content(model="primary-model", file_id="file-123")
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
assert response is sentinel_response
|
||||
assert mock_file_content.call_count == 1
|
||||
assert "silent_model" not in mock_file_content.call_args.kwargs
|
||||
|
||||
|
||||
def test_silent_experiment_generic_kwargs_skips_recursion_and_mcp():
|
||||
"""
|
||||
The generic-path snapshot must return None when the request is already a
|
||||
silent experiment (marker in either litellm_metadata or metadata) or when it
|
||||
carries MCP tooling that the provider may execute.
|
||||
"""
|
||||
router = Router(model_list=_generic_silent_model_list())
|
||||
base = {"input": [{"role": "user", "content": "hi"}]}
|
||||
|
||||
assert router._silent_experiment_generic_kwargs(**base, litellm_metadata={"is_silent_experiment": True}) is None
|
||||
assert router._silent_experiment_generic_kwargs(**base, metadata={"is_silent_experiment": True}) is None
|
||||
assert router._silent_experiment_generic_kwargs(**base, mcp_servers=[{"type": "url", "url": "x"}]) is None
|
||||
assert (
|
||||
router._silent_experiment_generic_kwargs(
|
||||
**base, tools=[{"type": "mcp", "server_label": "x", "require_approval": "never"}]
|
||||
)
|
||||
is None
|
||||
)
|
||||
|
||||
snapshot = router._silent_experiment_generic_kwargs(**base, tools=[{"type": "web_search"}])
|
||||
assert snapshot is not None
|
||||
assert snapshot["litellm_metadata"]["is_silent_experiment"] is True
|
||||
assert snapshot["stream"] is False
|
||||
assert "metadata" not in snapshot
|
||||
|
||||
|
||||
def test_silent_experiment_generic_kwargs_snapshots_before_mutation():
|
||||
"""
|
||||
Regression: the snapshot is taken synchronously, so metadata merged into the
|
||||
caller's kwargs afterwards (deployment tags etc.) never reaches the mirror.
|
||||
"""
|
||||
router = Router(model_list=_generic_silent_model_list())
|
||||
litellm_metadata = {"model_group": "primary-model"}
|
||||
snapshot = router._silent_experiment_generic_kwargs(input="hi", litellm_metadata=litellm_metadata)
|
||||
assert snapshot is not None
|
||||
litellm_metadata["tags"] = ["primary-deployment-tag"]
|
||||
assert "tags" not in snapshot["litellm_metadata"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_silent_experiment_ageneric_error_is_caught():
|
||||
"""
|
||||
A failing mirror must never propagate to the caller.
|
||||
"""
|
||||
router = Router(model_list=_generic_silent_model_list())
|
||||
|
||||
with patch.object(
|
||||
router,
|
||||
"_ageneric_api_call_with_fallbacks",
|
||||
new_callable=AsyncMock,
|
||||
side_effect=Exception("downstream failure"),
|
||||
):
|
||||
result = await router._silent_experiment_ageneric(
|
||||
silent_model="silent-model",
|
||||
original_function=litellm.aresponses,
|
||||
silent_kwargs={"input": [{"role": "user", "content": "hi"}]},
|
||||
)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_is_silent_experiment_marker():
|
||||
assert _is_silent_experiment_marker({"is_silent_experiment": True}) is True
|
||||
assert _is_silent_experiment_marker({"is_silent_experiment": False}) is False
|
||||
assert _is_silent_experiment_marker({}) is False
|
||||
assert _is_silent_experiment_marker(None) is False
|
||||
assert _is_silent_experiment_marker("is_silent_experiment") is False
|
||||
|
||||
|
||||
def test_has_mcp_tool():
|
||||
assert _has_mcp_tool([{"type": "mcp", "server_label": "x"}]) is True
|
||||
assert _has_mcp_tool([{"type": "web_search"}, {"type": "mcp"}]) is True
|
||||
assert _has_mcp_tool([{"type": "web_search"}]) is False
|
||||
assert _has_mcp_tool([]) is False
|
||||
assert _has_mcp_tool(None) is False
|
||||
assert _has_mcp_tool("mcp") is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("mcp_deployment", ["primary-model", "silent-model"])
|
||||
@pytest.mark.parametrize(
|
||||
"mcp_config",
|
||||
[
|
||||
{"tools": [{"type": "mcp", "server_label": "test", "server_url": "https://mcp.example.test"}]},
|
||||
{"mcp_servers": [{"type": "url", "name": "test", "url": "https://mcp.example.test"}]},
|
||||
],
|
||||
)
|
||||
async def test_generic_mirroring_skips_deployment_mcp(mcp_deployment, mcp_config):
|
||||
provider: Final = AsyncMock(return_value=MagicMock())
|
||||
provider.__name__ = "aresponses"
|
||||
router: Final = Router(
|
||||
model_list=[
|
||||
{
|
||||
**deployment,
|
||||
"litellm_params": {
|
||||
**deployment["litellm_params"],
|
||||
**(mcp_config if deployment["model_name"] == mcp_deployment else {}),
|
||||
},
|
||||
}
|
||||
for deployment in _generic_silent_model_list()
|
||||
]
|
||||
)
|
||||
endpoint: Final = router.factory_function(provider, call_type="aresponses")
|
||||
await endpoint(model="primary-model", input="hi")
|
||||
mirrors: Final = tuple(
|
||||
task for task in asyncio.all_tasks() if task.get_coro().__qualname__.endswith("._silent_experiment_ageneric")
|
||||
)
|
||||
await asyncio.gather(*mirrors)
|
||||
|
||||
provider.assert_awaited_once()
|
||||
assert provider.await_args.kwargs["model"] == "openai/gpt-4o-mini"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"call_type,attempt_method",
|
||||
[
|
||||
("aresponses", "_ageneric_api_call_with_fallbacks_responses_attempt"),
|
||||
("anthropic_messages", "_ageneric_api_call_with_fallbacks_anthropic_messages_attempt"),
|
||||
],
|
||||
)
|
||||
async def test_generic_primary_attempt_rejects_missing_provider_response(call_type, attempt_method):
|
||||
provider: Final = AsyncMock(return_value=None)
|
||||
provider.__name__ = call_type
|
||||
router: Final = Router(model_list=_generic_silent_model_list())
|
||||
with pytest.raises(TypeError, match="request returned no response"):
|
||||
await getattr(router, attempt_method)(model="silent-model", original_generic_function=provider, input="hi")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("call_type", ["aresponses", "anthropic_messages"])
|
||||
@pytest.mark.parametrize("fail_mirror", [False, True])
|
||||
async def test_generic_mirroring_fans_out_to_each_configured_target(call_type: str, fail_mirror: bool) -> None:
|
||||
observed: Final[asyncio.Queue[tuple[object, bool]]] = asyncio.Queue()
|
||||
|
||||
async def provider_call(**kwargs: object) -> str:
|
||||
observed.put_nowait((kwargs["model"], in_mirrored_call()))
|
||||
if fail_mirror and kwargs["model"] != "openai/gpt-5.4-mini":
|
||||
raise RuntimeError("mirror provider unavailable")
|
||||
return "provider-result"
|
||||
|
||||
provider: Final = AsyncMock(side_effect=provider_call)
|
||||
provider.__name__ = call_type
|
||||
router: Final = Router(model_list=_streaming_model_list(["shadow-a", "shadow-b"]), num_retries=0)
|
||||
endpoint: Final = router.factory_function(provider, call_type=call_type)
|
||||
|
||||
result: Final = await endpoint(model="primary-model", input="hi", metadata={"trace": "original"})
|
||||
mirrors: Final = tuple(
|
||||
task for task in asyncio.all_tasks() if task.get_coro().__qualname__.endswith("._silent_experiment_ageneric")
|
||||
)
|
||||
await asyncio.gather(*mirrors)
|
||||
|
||||
assert result == "provider-result"
|
||||
assert provider.await_count == 3
|
||||
assert {observed.get_nowait() for _ in range(3)} == {
|
||||
("openai/gpt-5.4-mini", False),
|
||||
("openai/gpt-5.4-nano", True),
|
||||
("anthropic/claude-haiku-4-5", True),
|
||||
}
|
||||
assert not in_mirrored_call()
|
||||
assert {call.kwargs["litellm_metadata"]["model_group"] for call in provider.await_args_list} == {
|
||||
"primary-model",
|
||||
"shadow-a",
|
||||
"shadow-b",
|
||||
}
|
||||
for call in provider.await_args_list:
|
||||
assert call.kwargs["metadata"] == {"trace": "original"}
|
||||
assert "silent_model" not in call.kwargs
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generic_mirror_does_not_reuse_primary_attempt_callback():
|
||||
primary_attempt: Final = AsyncMock(return_value="primary-only")
|
||||
provider: Final = AsyncMock(return_value="mirror-result")
|
||||
provider.__name__ = "aresponses"
|
||||
router: Final = Router(model_list=_generic_silent_model_list())
|
||||
|
||||
await router._silent_experiment_ageneric(
|
||||
silent_model="silent-model",
|
||||
original_function=provider,
|
||||
silent_kwargs={"input": "hi", "attempt_function": primary_attempt},
|
||||
)
|
||||
|
||||
primary_attempt.assert_not_awaited()
|
||||
provider.assert_awaited_once()
|
||||
assert provider.await_args.kwargs["model"] == "openai/gpt-4o"
|
||||
|
||||
|
||||
def _judged_router(shadow_outcome: str) -> Router:
|
||||
return Router(
|
||||
model_list=[
|
||||
|
|
@ -1044,3 +1381,71 @@ async def test_calls_a_failure_function_callback_starts_for_a_mirrored_call_are_
|
|||
await _wait_for_judges(sync_and_async_recording_logger, {"shadow-a"})
|
||||
|
||||
assert sync_and_async_recording_logger.judge_flags() == {"shadow-a": None}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("call_type", ["aresponses", "anthropic_messages"])
|
||||
@pytest.mark.parametrize("use_fallback", [False, True], ids=["retry", "fallback"])
|
||||
async def test_generic_mirrors_only_once_across_attempts(call_type: str, use_fallback: bool) -> None:
|
||||
calls: Final[list[tuple[str, bool]]] = []
|
||||
|
||||
async def provider_call(**kwargs):
|
||||
model = kwargs["model"]
|
||||
mirror = in_mirrored_call()
|
||||
calls.append((model, mirror))
|
||||
if not mirror and len([call for call in calls if not call[1]]) == 1:
|
||||
raise litellm.InternalServerError(message="retry me", llm_provider="openai", model=model)
|
||||
return "provider-result"
|
||||
|
||||
provider: Final = AsyncMock(side_effect=provider_call)
|
||||
provider.__name__ = call_type
|
||||
models = _generic_silent_model_list()
|
||||
models.append(
|
||||
{
|
||||
"model_name": "fallback-model",
|
||||
"litellm_params": {"model": "openai/gpt-4.1-mini", "api_key": "fake-key", "silent_model": "silent-model"},
|
||||
}
|
||||
)
|
||||
router: Final = Router(
|
||||
model_list=models,
|
||||
num_retries=0 if use_fallback else 1,
|
||||
retry_after=0,
|
||||
disable_cooldowns=True,
|
||||
fallbacks=[{"primary-model": ["fallback-model"]}] if use_fallback else [],
|
||||
)
|
||||
endpoint: Final = router.factory_function(provider, call_type=call_type)
|
||||
result = await endpoint(model="primary-model", input="hi")
|
||||
mirrors = tuple(
|
||||
task for task in asyncio.all_tasks() if task.get_coro().__qualname__.endswith("._silent_experiment_ageneric")
|
||||
)
|
||||
await asyncio.gather(*mirrors)
|
||||
|
||||
assert result == "provider-result"
|
||||
assert len([call for call in calls if not call[1]]) == 2
|
||||
assert len([call for call in calls if call[1]]) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("call_type", ["aresponses", "anthropic_messages"])
|
||||
async def test_generic_mirror_attempt_state_is_per_request(call_type: str) -> None:
|
||||
provider: Final = AsyncMock(return_value="provider-result")
|
||||
provider.__name__ = call_type
|
||||
router: Final = Router(model_list=_generic_silent_model_list(), num_retries=0)
|
||||
endpoint: Final = router.factory_function(provider, call_type=call_type)
|
||||
|
||||
assert await asyncio.gather(
|
||||
endpoint(model="primary-model", input="request-a"),
|
||||
endpoint(model="primary-model", input="request-b"),
|
||||
) == ["provider-result", "provider-result"]
|
||||
mirrors: Final = tuple(
|
||||
task for task in asyncio.all_tasks() if task.get_coro().__qualname__.endswith("._silent_experiment_ageneric")
|
||||
)
|
||||
await asyncio.gather(*mirrors)
|
||||
|
||||
assert provider.await_count == 4
|
||||
for request in ("request-a", "request-b"):
|
||||
assert {call.kwargs["model"] for call in provider.await_args_list if call.kwargs["input"] == request} == {
|
||||
"openai/gpt-4o-mini",
|
||||
"openai/gpt-4o",
|
||||
}
|
||||
assert all("_silent_experiment_state" not in call.kwargs for call in provider.await_args_list)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue