diff --git a/litellm/constants.py b/litellm/constants.py index c586a5f767a..a943316b6af 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -1967,6 +1967,8 @@ BACKGROUND_INTERACTION_COST_POLLING_ENABLED: Final = _background_interaction_cos BACKGROUND_INTERACTION_SETTLEMENT_CLAIM_LEASE_SECONDS: Final = 300.0 PROXY_BUDGET_RESCHEDULER_MAX_TIME: Final = get_env_int("PROXY_BUDGET_RESCHEDULER_MAX_TIME", 605) PROXY_BATCH_WRITE_AT: Final = get_env_int("PROXY_BATCH_WRITE_AT", 10) # in seconds, increased from 10 +TEST_MODEL_SPEND_LOG_WAIT_SECONDS: Final = get_env_int("TEST_MODEL_SPEND_LOG_WAIT_SECONDS", PROXY_BATCH_WRITE_AT + 5) +TEST_MODEL_COST_WAIT_SECONDS: Final = get_env_float("TEST_MODEL_COST_WAIT_SECONDS", 3.0) PROXY_CONFIG_RELOAD_INTERVAL_SECONDS: Final = get_env_int("PROXY_CONFIG_RELOAD_INTERVAL_SECONDS", 30) # APScheduler Configuration - MEMORY LEAK FIX diff --git a/litellm/proxy/health_endpoints/_health_endpoints.py b/litellm/proxy/health_endpoints/_health_endpoints.py index 8545e0609db..727d2f0309b 100644 --- a/litellm/proxy/health_endpoints/_health_endpoints.py +++ b/litellm/proxy/health_endpoints/_health_endpoints.py @@ -6,12 +6,13 @@ import os import secrets import time import traceback -from collections.abc import Iterable, Mapping, Sequence +import uuid +from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence from dataclasses import dataclass from datetime import datetime, timedelta, timezone from itertools import chain, groupby from types import MappingProxyType -from typing import Final, Literal, TypedDict +from typing import Annotated, Final, Literal, TypeAlias, TypedDict import fastapi from fastapi import APIRouter, Depends, HTTPException, Request, Response, status @@ -20,13 +21,19 @@ from typing_extensions import ReadOnly import litellm from litellm._logging import verbose_logger, verbose_proxy_logger -from litellm.constants import HEALTH_CHECK_TIMEOUT_SECONDS, PROXY_DB_LOOKUP_STALL_WINDOW_SECONDS +from litellm.constants import ( + HEALTH_CHECK_TIMEOUT_SECONDS, + PROXY_DB_LOOKUP_STALL_WINDOW_SECONDS, + TEST_MODEL_COST_WAIT_SECONDS, + TEST_MODEL_SPEND_LOG_WAIT_SECONDS, +) from litellm.integrations.SlackAlerting.ms_teams import ( MS_TEAMS_ALERT_HEADERS, build_ms_teams_payload, get_ms_teams_webhook_url, ) from litellm.litellm_core_utils.custom_logger_registry import CustomLoggerRegistry +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler from litellm.proxy._types import ( AlertType, @@ -85,6 +92,19 @@ from litellm.proxy.health_check import ( # noqa: F401 # legacy module exports run_with_timeout, update_litellm_params_for_health_check, ) +from litellm.proxy.health_endpoints.model_checks import ( + ENDPOINT_FOR_MODE, + SCRIPTS, + CalledTurn, + EndpointResult, + EndpointScript, + TestModelEndpoint, + TestModelResponse, + endpoint_result_payload, + overall_status, + response_cost, + run_endpoint_check, +) from litellm.proxy.middleware.admission_control_middleware import ( get_admission_control_stats, ) @@ -101,7 +121,7 @@ from litellm.router_utils.clientside_credential_handler import ( from litellm.router_utils.common_utils import resolve_model_group_alias from litellm.secret_managers.main import get_secret_bool from litellm.types.proxy.litellm_pre_call_utils import RedactedDict, SecretFields -from litellm.types.router import reject_server_owned_wif_fields +from litellm.types.router import Deployment, reject_server_owned_wif_fields from litellm.types.utils import AdminApiKeySecretReference, admin_api_key_secret_reference #### Health ENDPOINTS #### @@ -2487,6 +2507,410 @@ async def health_liveliness_options(): return Response(headers=response_headers, status_code=200) +@dataclass(frozen=True, slots=True) +class _ResolvedConnection: + """A connection test's probe parameters, resolved the way ``/health/test_connection`` always has.""" + + request_litellm_params: Mapping[str, object] + config_litellm_params: Mapping[str, object] + litellm_params: dict[str, object] # mutable-ok: the health check helpers mutate the probe params in place + resolved_model_info: dict[str, object] | None # mutable-ok: handed to Deployment, which takes a plain dict + probe_model_info: Mapping[str, object] + deployment_by_id: Deployment | None + stored_model: str | None + + +def _resolve_connection_params( + request_litellm_params: Mapping[str, object], + model_info: Mapping[str, object] | None, + llm_router: Router | None, + general_settings: Mapping[str, object], +) -> _ResolvedConnection: + # Reject request-supplied os.environ/ references. Config values are + # already resolved before reaching this endpoint; any remaining + # reference must have come from the request body. + _reject_os_environ_references(dict(request_litellm_params)) + if model_info: + _reject_os_environ_references(dict(model_info)) + requested_model: Final = request_litellm_params.get("model") + model_name: Final = requested_model if isinstance(requested_model, str) else None + + # Look up model configuration from router if model name is provided + # This gets the litellm_params from proxy config (with resolved env vars) + config_litellm_params: dict = {} + loaded_model_info: dict | None = None + deployment_by_id: Deployment | None = None + if llm_router is not None: + # Prefer disambiguation by deployment id (`model_info.id`) when + # the caller supplies it. This is required when multiple + # deployments share a `model_name` (e.g. wildcard `openai/*` + # with multiple `api_base` values for failover): the UI's + # "Test Connection" button targets a specific row, and that + # row's id is the only thing that uniquely identifies which + # deployment to probe. Without this, all duplicates collapse + # onto `deployments[0]`. + requested_id: Final = (model_info or {}).get("id") + request_model_id: Final = requested_id if isinstance(requested_id, str) else None + try: + if request_model_id: + deployment_by_id = llm_router.registry().deployment(request_model_id) + + if deployment_by_id is not None: + config_litellm_params = deployment_by_id.litellm_params.model_dump(exclude_none=True) + loaded_model_info = deployment_by_id.model_info.model_dump(exclude_none=True) + elif model_name: + # Fall back to model_name lookup for callers (e.g. the + # "Add Model" wizard, or curl) that don't supply an id. + # First try to find by proxy model_name (e.g., "gpt-4o") + deployments = llm_router.registry().deployments(model_name) + + # If not found, try to find by litellm model name + # (e.g., "azure/gpt-4o") + if not deployments or len(deployments) == 0: + all_deployments: Final = llm_router.registry().deployments(None) + if all_deployments: + for deployment in all_deployments: + if deployment.get("litellm_params", {}).get("model") == model_name: + deployments = [deployment] + break + + if deployments and len(deployments) > 0: + # Use the first deployment's litellm_params as base + # config. These already have resolved environment + # variables from proxy config. + config_litellm_params = dict(deployments[0].get("litellm_params", {})) + loaded_model_info = dict(deployments[0].get("model_info") or {}) + except Exception as e: + verbose_proxy_logger.debug( + "Could not find model %s in router: %s. Proceeding with request params only.", model_name, e + ) + + selected_by_id: Final = deployment_by_id is not None + reject_server_owned_wif_fields( + MappingProxyType( + { + key: value + for key, value in _OBJECT_MAPPING.validate_python(request_litellm_params).items() + if key != "chatgpt_auth_file" + } + ) + ) + # Merge: config params (from proxy config) as base, request params override + merged_litellm_params: Final = { + **_config_base_for_health_check( + config_litellm_params, + request_litellm_params, + allow_client_side_credentials=general_settings.get("allow_client_side_credentials") is True, + selected_by_id=selected_by_id, + ), + **request_litellm_params, + } + + resolved_model_info: Final = ( + loaded_model_info if loaded_model_info is not None else (dict(model_info) if model_info is not None else None) + ) + probe_model_info: Final = _model_info_for_mode_resolution( + _OBJECT_MAPPING.validate_python(resolved_model_info or {}), + stored_params=_OBJECT_MAPPING.validate_python(config_litellm_params), + request_params=_OBJECT_MAPPING.validate_python(request_litellm_params), + ) + stored_model: Final = _OBJECT_MAPPING.validate_python(config_litellm_params).get("model") + return _ResolvedConnection( + request_litellm_params=_OBJECT_MAPPING.validate_python(request_litellm_params), + config_litellm_params=_OBJECT_MAPPING.validate_python(config_litellm_params), + litellm_params=merged_litellm_params, + resolved_model_info=resolved_model_info, + probe_model_info=probe_model_info, + deployment_by_id=deployment_by_id, + stored_model=stored_model if isinstance(stored_model, str) else None, + ) + + +_RawQuery: TypeAlias = Callable[..., Awaitable[Sequence[object]]] + + +class _SpendLogRow(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + tag: str + spend: float + + +_SPEND_LOG_ROWS: Final = TypeAdapter(list[_SpendLogRow]) +_TEST_MODEL_TAG_PREFIX: Final = "litellm_test_model:" +_SPEND_LOG_BY_TAG_SQL: Final = """ + SELECT tag, "spend" + FROM "LiteLLM_SpendLogs" sl, jsonb_array_elements_text(sl."request_tags"::jsonb) AS tag + WHERE tag IN (SELECT jsonb_array_elements_text($1::jsonb)) +""" + + +def _test_model_tag(call_id: str) -> str: + return f"{_TEST_MODEL_TAG_PREFIX}{call_id}" + + +async def _spend_rows_once(run_query: _RawQuery, request_ids: frozenset[str]) -> Mapping[str, float]: + """Spend per request id. The row's request_id is the provider's response id, so each turn is found by the + tag the check put on it, which the proxy writes to request_tags like any client tag.""" + rows: Final = _SPEND_LOG_ROWS.validate_python( + await run_query(_SPEND_LOG_BY_TAG_SQL, json.dumps(sorted(_test_model_tag(call_id) for call_id in request_ids))) + ) + return MappingProxyType({row.tag.removeprefix(_TEST_MODEL_TAG_PREFIX): row.spend for row in rows}) + + +async def _spend_log_rows( + run_query: _RawQuery, + request_ids: Sequence[str], + *, + wait_seconds: float = TEST_MODEL_SPEND_LOG_WAIT_SECONDS, + poll_seconds: float = 1.0, +) -> Mapping[str, float]: + """Spend rows for ``request_ids``, polled until every row landed or ``wait_seconds`` passed. + + The proxy flushes spend logs in batches every ``PROXY_BATCH_WRITE_AT`` seconds, so a row for a request + that just finished is expected to take that long to appear. + """ + deadline: Final = time.monotonic() + wait_seconds + wanted: Final = frozenset(request_ids) + while not wanted <= (found := await _spend_rows_once(run_query, wanted)).keys() and time.monotonic() < deadline: + await asyncio.sleep(poll_seconds) + return found + + +def _test_model_endpoints( + requested: Sequence[TestModelEndpoint] | None, mode: str | None +) -> tuple[TestModelEndpoint, ...]: + if requested: + return tuple(dict.fromkeys(requested)) + native: Final = ENDPOINT_FOR_MODE.get(mode or "chat") + if native is None: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail={ + "error": f"/health/test_model covers chat models only and this model's mode is {mode!r}. " + "Use /health/test_connection for it." + }, + ) + return (native,) + + +async def _logged_cost(logging_obj: LiteLLMLoggingObj, fallback: float | None) -> float | None: + """The cost the success handler logged for the call, which lands a moment after the call returns.""" + deadline: Final = time.monotonic() + TEST_MODEL_COST_WAIT_SECONDS + while (logged := _OBJECT_MAPPING.validate_python(logging_obj.model_call_details).get("response_cost")) is None and ( + time.monotonic() < deadline + ): + await asyncio.sleep(0.05) + return float(logged) if isinstance(logged, int | float) else fallback + + +def _turn_caller( + llm_router: Router | None, + *, + pinned_model_id: str | None, + call_params: Mapping[str, object], + proxy_metadata: Mapping[str, object], +) -> Callable[[EndpointScript, Mapping[str, object], str], Awaitable[CalledTurn]]: + """The two ways a turn reaches the provider: pinned to a saved deployment through the router, so the + deployment's own credentials and pricing apply, or straight through litellm with the submitted params. + Either way the call carries the proxy's own logging object, the way every proxied request does, so the + cost read back is the one the spend log row gets.""" + + async def call(script: EndpointScript, body: Mapping[str, object], call_id: str) -> CalledTurn: + target_params: Final = {"model": pinned_model_id} if pinned_model_id is not None else call_params + logging_obj, prepared = litellm.utils.function_setup( + original_function=script.route_type, + rules_obj=litellm.utils.Rules(), + start_time=datetime.now(), # noqa: DTZ005 # Logging pairs it with the naive end_time it records + **{ # pyright: ignore[reportArgumentType] # takes the untyped request dict, like the proxy + **target_params, + **body, + script.metadata_key: {**proxy_metadata, "tags": [_test_model_tag(call_id)]}, + "litellm_call_id": call_id, + "drop_params": True, + }, + ) + prepared["litellm_logging_obj"] = logging_obj + target: Final = llm_router if pinned_model_id is not None and llm_router is not None else litellm + response: Final[object] = await asyncio.wait_for( + getattr(target, script.route_type)(**prepared), HEALTH_CHECK_TIMEOUT_SECONDS + ) + return CalledTurn(response=response, cost=await _logged_cost(logging_obj, response_cost(response))) + + return call + + +@router.post( + "/health/test_model", + tags=["health"], + dependencies=[Depends(user_api_key_auth)], +) +async def test_model( + request: Request, + user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], + model_id: Annotated[ + str | None, + fastapi.Body( + description="The deployment to test, by its model_info.id. The saved credentials and pricing are used." + ), + ] = None, + litellm_params: Annotated[ + Mapping[str, object] | None, + fastapi.Body( + description=( + "Parameters of a model that is not saved yet (model, api_key, api_base, api_version), as sent to " + "/model/new. When model_id is also given these override the saved deployment's values." + ) + ), + ] = None, + model_info: Annotated[ + Mapping[str, object] | None, + fastapi.Body(description="Model info of an unsaved model, for example mode or custom pricing."), + ] = None, + endpoints: Annotated[ + Sequence[TestModelEndpoint] | None, + fastapi.Body( + description=( + "Which endpoints to test: /v1/chat/completions, /v1/responses, /v1/messages. " + "Defaults to the model's native endpoint." + ) + ), + ] = None, +) -> TestModelResponse: + """ + Run the live checks on a model, the way a client would use it, and report what works. + + Each selected endpoint gets one two-turn conversation with a long system prompt and a tool. Turn 1 asks + a question the tool answers, turn 2 sends the tool result back. Three checks come out of it: + + - `conversation`: both turns parse, turn 1 calls the tool, turn 2 uses the tool result + - `prompt_caching`: turn 2 reports cached input tokens, with no cache markers in the request + - `cost_tracking`: every turn carries a non-zero response cost, and the spend log row lands when a + database is connected + + Example: + ```bash + curl -X POST 'http://localhost:4000/health/test_model' \\ + -H "Authorization: Bearer $LITELLM_MASTER_KEY" \\ + -H 'Content-Type: application/json' \\ + -d '{"model_id": "a1b2c3", "endpoints": ["/v1/chat/completions", "/v1/responses"]}' + ``` + + Returns `status` (`success`, `warning` or `error`), the tested `model`, and one entry per endpoint + with its `status`, the three `checks` (`name`, `status`, `detail`), `cache_read_tokens`, + `cache_read_percent`, `spend_usd` and `duration_ms`. + """ + from litellm.proxy.litellm_pre_call_utils import LiteLLMProxyRequestSetup + from litellm.proxy.management_endpoints.model_management_endpoints import ( + ModelManagementAuthChecks, + ) + from litellm.proxy.proxy_server import ( + general_settings, + llm_router, + premium_user, + prisma_client, + ) + from litellm.types.router import Deployment, LiteLLM_Params + + try: + request_litellm_params: Final = _OBJECT_MAPPING.validate_python(litellm_params or {}) + request_model_info: Final[Mapping[str, object]] = { + **(model_info or {}), + **({"id": model_id} if model_id else {}), + } + if not model_id and not request_litellm_params.get("model"): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail={"error": "Send model_id for a saved deployment or litellm_params.model for a new one."}, + ) + resolved: Final = _resolve_connection_params( + request_litellm_params=request_litellm_params, + model_info=request_model_info, + llm_router=llm_router, + general_settings=_OBJECT_MAPPING.validate_python(general_settings), + ) + if model_id and resolved.deployment_by_id is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"error": f"No deployment with model_info.id {model_id!r} is loaded on this proxy."}, + ) + if prisma_client is not None: + await ModelManagementAuthChecks.can_user_make_model_call( + model_params=Deployment( + model_name="test_model", + litellm_params=LiteLLM_Params.model_validate(resolved.litellm_params), + model_info=resolved.resolved_model_info, + ), + user_api_key_dict=user_api_key_dict, + prisma_client=prisma_client, + premium_user=premium_user, + incoming_params=resolved.request_litellm_params, + ) + elif not _is_proxy_admin(user_api_key_dict): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, detail={"error": "Only proxy admins can test models."} + ) + + call_params: Final = {key: value for key, value in resolved.litellm_params.items() if key != "mode"} + mode: Final = _string_mode_or_bad_request(resolved.litellm_params.get("mode")) or resolve_health_check_mode( + resolved.probe_model_info, _OBJECT_MAPPING.validate_python(call_params) + ) + selected: Final = _test_model_endpoints(endpoints, mode) + overrides_saved_deployment: Final = any(key != "model" for key in resolved.request_litellm_params) + pinned_model_id: Final = ( + resolved.deployment_by_id.model_info.id + if resolved.deployment_by_id is not None and not overrides_saved_deployment + else None + ) + proxy_metadata: Final = { + **LiteLLMProxyRequestSetup.get_sanitized_user_information_from_key(user_api_key_dict=user_api_key_dict), + "user_api_key": LiteLLMProxyRequestSetup.get_logged_api_key(user_api_key_dict), + "endpoint": "/health/test_model", + } + call: Final = _turn_caller( + llm_router, + pinned_model_id=pinned_model_id, + call_params={ + **_OBJECT_MAPPING.validate_python(call_params), + **_admin_api_key_from_request( + resolved.request_litellm_params, _OBJECT_MAPPING.validate_python(call_params) + ), + }, + proxy_metadata=proxy_metadata, + ) + run_query: Final[_RawQuery | None] = prisma_client.db.query_raw if prisma_client is not None else None + + async def spend_lookup(request_ids: Sequence[str]) -> Mapping[str, float]: + return await _spend_log_rows(run_query, request_ids) if run_query is not None else {} + + results: Final[Sequence[EndpointResult]] = await asyncio.gather( + *( + run_endpoint_check( + SCRIPTS[endpoint], + call, + new_call_id=lambda: str(uuid.uuid4()), + spend_lookup=spend_lookup if prisma_client is not None else None, + ) + for endpoint in selected + ) + ) + tested_model: Final = resolved.litellm_params.get("model") + return TestModelResponse( + status=overall_status(results), + model=tested_model if isinstance(tested_model, str) else None, + model_id=pinned_model_id, + endpoints=[endpoint_result_payload(result) for result in results], + ) + except (HTTPException, ProxyException): + raise + except Exception as e: # noqa: BLE001 # any other failure becomes the endpoint's 500, like test_connection + verbose_proxy_logger.debug("litellm.proxy.health_endpoints.test_model(): Exception occurred - %s", e) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail={"error": f"Failed to test model: {e}"}, + ) + + @router.post( "/health/test_connection", tags=["health"], @@ -2599,94 +3023,17 @@ async def test_model_connection( detail={"error": CommonProxyErrors.db_not_connected_error.value}, ) - # Get model name from litellm_params - request_litellm_params: Final = litellm_params or {} - # Reject request-supplied os.environ/ references. Config values are - # already resolved before reaching this endpoint; any remaining - # reference must have come from the request body. - _reject_os_environ_references(request_litellm_params) - if model_info: - _reject_os_environ_references(model_info) - model_name: Final = request_litellm_params.get("model") - - # Look up model configuration from router if model name is provided - # This gets the litellm_params from proxy config (with resolved env vars) - config_litellm_params: dict = {} - loaded_model_info: dict | None = None - deployment_by_id: Deployment | None = None - if llm_router is not None: - # Prefer disambiguation by deployment id (`model_info.id`) when - # the caller supplies it. This is required when multiple - # deployments share a `model_name` (e.g. wildcard `openai/*` - # with multiple `api_base` values for failover): the UI's - # "Test Connection" button targets a specific row, and that - # row's id is the only thing that uniquely identifies which - # deployment to probe. Without this, all duplicates collapse - # onto `deployments[0]`. - request_model_info: Final = model_info or {} - request_model_id: Final = request_model_info.get("id") - try: - if request_model_id: - deployment_by_id = llm_router.registry().deployment(request_model_id) - - if deployment_by_id is not None: - config_litellm_params = deployment_by_id.litellm_params.model_dump(exclude_none=True) - loaded_model_info = deployment_by_id.model_info.model_dump(exclude_none=True) - elif model_name: - # Fall back to model_name lookup for callers (e.g. the - # "Add Model" wizard, or curl) that don't supply an id. - # First try to find by proxy model_name (e.g., "gpt-4o") - deployments = llm_router.registry().deployments(model_name) - - # If not found, try to find by litellm model name - # (e.g., "azure/gpt-4o") - if not deployments or len(deployments) == 0: - all_deployments: Final = llm_router.registry().deployments(None) - if all_deployments: - for deployment in all_deployments: - if deployment.get("litellm_params", {}).get("model") == model_name: - deployments = [deployment] - break - - if deployments and len(deployments) > 0: - # Use the first deployment's litellm_params as base - # config. These already have resolved environment - # variables from proxy config. - config_litellm_params = dict(deployments[0].get("litellm_params", {})) - loaded_model_info = dict(deployments[0].get("model_info") or {}) - except Exception as e: - verbose_proxy_logger.debug( - "Could not find model %s in router: %s. Proceeding with request params only.", model_name, e - ) - - selected_by_id: Final = deployment_by_id is not None - reject_server_owned_wif_fields( - MappingProxyType( - { - key: value - for key, value in _OBJECT_MAPPING.validate_python(request_litellm_params).items() - if key != "chatgpt_auth_file" - } - ) + resolved: Final = _resolve_connection_params( + request_litellm_params=_OBJECT_MAPPING.validate_python(litellm_params or {}), + model_info=_OBJECT_MAPPING.validate_python(model_info) if model_info else None, + llm_router=llm_router, + general_settings=_OBJECT_MAPPING.validate_python(general_settings), ) - # Merge: config params (from proxy config) as base, request params override - litellm_params = { - **_config_base_for_health_check( - config_litellm_params, - request_litellm_params, - allow_client_side_credentials=general_settings.get("allow_client_side_credentials") is True, - selected_by_id=selected_by_id, - ), - **request_litellm_params, - } - - resolved_model_info: Final = loaded_model_info if loaded_model_info is not None else model_info - probe_model_info: Final = _model_info_for_mode_resolution( - _OBJECT_MAPPING.validate_python(resolved_model_info or {}), - stored_params=_OBJECT_MAPPING.validate_python(config_litellm_params), - request_params=_OBJECT_MAPPING.validate_python(request_litellm_params), - ) - stored_model: Final = _OBJECT_MAPPING.validate_python(config_litellm_params).get("model") + request_litellm_params: Final = resolved.request_litellm_params + resolved_model_info: Final = resolved.resolved_model_info + probe_model_info: Final = resolved.probe_model_info + stored_model: Final = resolved.stored_model + litellm_params = resolved.litellm_params litellm_params = update_litellm_params_for_health_check( model_info=dict(probe_model_info), litellm_params=litellm_params, diff --git a/litellm/proxy/health_endpoints/model_checks.py b/litellm/proxy/health_endpoints/model_checks.py new file mode 100644 index 00000000000..719243cc032 --- /dev/null +++ b/litellm/proxy/health_endpoints/model_checks.py @@ -0,0 +1,729 @@ +"""Live checks behind ``POST /health/test_model``. + +One two-turn tool-call conversation per selected endpoint, scripted in that endpoint's native request shape +with no prompt-cache markers, so the second turn can only report cached tokens when the provider caches the +repeated prefix on its own. Three verdicts come out of the two turns: the conversation itself (both turns +parse, the tool call lands, the tool result is used), prompt caching, and cost tracking. +""" + +from __future__ import annotations + +import json +import time +from collections.abc import Awaitable, Callable, Mapping, Sequence +from dataclasses import dataclass +from types import MappingProxyType +from typing import Final, Literal, TypeAlias + +from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError + +from litellm.router_utils.add_retry_fallback_headers import get_hidden_params_dict + +TestModelEndpoint: TypeAlias = Literal["/v1/chat/completions", "/v1/responses", "/v1/messages"] +TestModelRouteType: TypeAlias = Literal["acompletion", "aresponses", "anthropic_messages"] +Verdict: TypeAlias = Literal["pass", "warn", "fail"] +EndpointStatus: TypeAlias = Literal["success", "warning", "error"] + +TEST_MODEL_ENDPOINTS: Final[tuple[TestModelEndpoint, ...]] = ("/v1/chat/completions", "/v1/responses", "/v1/messages") +WEATHER_TOOL_NAME: Final = "get_weather" +WEATHER_TOOL_RESULT: Final = "72F and sunny" +WEATHER_QUESTION: Final = "What is the weather in Boston right now? Use the get_weather tool to find out." +MAX_OUTPUT_TOKENS: Final = 2048 +THINKING_BUDGET_TOKENS: Final = 1024 + +_WEATHER_PARAMETERS: Final[Mapping[str, object]] = MappingProxyType( + { + "type": "object", + "properties": { + "location": {"type": "string", "description": "City name, for example Boston"}, + "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}, + }, + "required": ["location"], + } +) +_WEATHER_DESCRIPTION: Final = "Get the current weather for a city" + +_CITIES: Final = ("Boston", "Denver", "Lisbon", "Osaka", "Nairobi", "Perth", "Quito", "Reykjavik") +_SECTION_SENTENCES: Final = ( + "Section {n}: deliveries routed through the {city} hub are scanned at intake, weighed on the calibrated " + "scale, and matched against the manifest before any label is printed or any pallet leaves the dock.", + "Agents handling {city} tickets confirm the tracking number, the declared value, and the delivery window " + "with the sender before promising a reroute, and record the promise in the ticket with a timestamp.", + "If a parcel bound for {city} misses its scheduled scan by more than {hours} hours, the agent opens a " + "trace request, notifies the recipient in plain language, and schedules a follow-up for the next shift.", + "Refunds for {city} shipments above {amount} dollars need a supervisor code, a photo of the damage when " + "one exists, and the original invoice number, and are paid to the original payment method within ten days.", +) +_SECTION_COUNT: Final = 32 + +SYSTEM_PROMPT: Final = "\n\n".join( + ( + "You are the support assistant for Northwind Logistics. Follow the playbook below exactly.", + *( + " ".join( + sentence.format(n=n, city=_CITIES[n % len(_CITIES)], hours=4 + (n % 5), amount=100 * (1 + (n % 7))) + for sentence in _SECTION_SENTENCES + ) + for n in range(1, _SECTION_COUNT + 1) + ), + f"When the user asks about the weather, call the {WEATHER_TOOL_NAME} tool with the city name. After a " + "tool result arrives, answer in one short sentence that repeats the temperature exactly as the tool " + "reported it.", + ) +) + + +@dataclass(frozen=True, slots=True) +class ToolCallSeen: + id: str + name: str + raw_arguments: str + arguments: Mapping[str, object] | None + + +@dataclass(frozen=True, slots=True) +class ParsedTurn: + tool_call: ToolCallSeen | None + text: str + input_tokens: int + cache_read_tokens: int + assistant_echo: object + + +@dataclass(frozen=True, slots=True) +class EndpointScript: + endpoint: TestModelEndpoint + route_type: TestModelRouteType + metadata_key: Literal["metadata", "litellm_metadata"] + turn_one: Callable[[], Mapping[str, object]] + turn_two: Callable[[ParsedTurn], Mapping[str, object]] + parse: Callable[[object], ParsedTurn | str] + + +@dataclass(frozen=True, slots=True) +class CheckResult: + name: Literal["conversation", "prompt_caching", "cost_tracking"] + status: Verdict + detail: str + + +@dataclass(frozen=True, slots=True) +class EndpointResult: + endpoint: TestModelEndpoint + status: EndpointStatus + checks: tuple[CheckResult, ...] + cache_read_tokens: int + cache_read_percent: float | None + spend_usd: float + duration_ms: int + + +@dataclass(frozen=True, slots=True) +class CalledTurn: + """One turn's provider response and the cost the proxy logged for it (what the spend log row carries).""" + + response: object + cost: float | None + + +@dataclass(frozen=True, slots=True) +class TurnOutcome: + response: object | None + cost: float | None + error: str | None + call_id: str + + +TurnCaller: TypeAlias = Callable[[EndpointScript, Mapping[str, object], str], Awaitable[CalledTurn]] +SpendLookup: TypeAlias = Callable[[Sequence[str]], Awaitable[Mapping[str, float]]] +CallIdFactory: TypeAlias = Callable[[], str] + + +class _Loose(BaseModel): + model_config = ConfigDict(frozen=True, extra="ignore") + + +class _ChatFunction(_Loose): + name: str + arguments: str = "" + + +class _ChatToolCall(_Loose): + id: str = "" + function: _ChatFunction + + +class _PromptTokensDetails(_Loose): + cached_tokens: int | None = None + + +class _ChatUsage(_Loose): + prompt_tokens: int = 0 + prompt_tokens_details: _PromptTokensDetails | None = None + + +class _ChatMessage(_Loose): + content: str | Sequence[object] | None = None + tool_calls: Sequence[_ChatToolCall] | None = None + + +class _ChatChoice(_Loose): + message: _ChatMessage + + +class _ChatResponse(_Loose): + choices: Sequence[_ChatChoice] + usage: _ChatUsage | None = None + + +class _ResponsesContent(_Loose): + type: str = "" + text: str = "" + + +class _ResponsesItem(_Loose): + type: str = "" + name: str = "" + arguments: str = "" + call_id: str = "" + content: Sequence[_ResponsesContent] | None = None + + +class _InputTokensDetails(_Loose): + cached_tokens: int | None = None + + +class _ResponsesUsage(_Loose): + input_tokens: int = 0 + input_tokens_details: _InputTokensDetails | None = None + + +class _ResponsesResponse(_Loose): + output: Sequence[_ResponsesItem] + usage: _ResponsesUsage | None = None + + +class _MessagesBlock(_Loose): + type: str = "" + text: str = "" + id: str = "" + name: str = "" + input: Mapping[str, object] | None = None + + +class _MessagesUsage(_Loose): + input_tokens: int = 0 + cache_read_input_tokens: int | None = None + cache_creation_input_tokens: int | None = None + + +class _MessagesResponse(_Loose): + content: Sequence[_MessagesBlock] + usage: _MessagesUsage + + +_CHAT: Final = TypeAdapter(_ChatResponse) +_RESPONSES: Final = TypeAdapter(_ResponsesResponse) +_MESSAGES: Final = TypeAdapter(_MessagesResponse) +_PLAIN_OBJECT: Final = TypeAdapter(dict[str, object]) +_OBJECT_LIST: Final = TypeAdapter(list[object]) +_CHOICES: Final = TypeAdapter(list[dict[str, object]]) +_JSON_DOCUMENT: Final = TypeAdapter(object) +_EMPTY_ASSISTANT: Final[Mapping[str, object]] = MappingProxyType({"role": "assistant", "content": None}) + + +def _plain(response: object) -> Mapping[str, object] | None: + if isinstance(response, BaseModel): + return _PLAIN_OBJECT.validate_python(response.model_dump(exclude_none=True)) + if isinstance(response, dict): + return _PLAIN_OBJECT.validate_python(response) + return None + + +def _validation_detail(error: ValidationError) -> str: + first: Final = error.errors()[0] + location: Final = ".".join(str(part) for part in first["loc"]) + return f"{location}: {first['msg']}" if location else first["msg"] + + +def _arguments(raw: str) -> Mapping[str, object] | None: + try: + parsed: Final = _JSON_DOCUMENT.validate_json(raw) + except ValidationError: + return None + return MappingProxyType(_PLAIN_OBJECT.validate_python(parsed)) if isinstance(parsed, dict) else None + + +def _chat_turn_one() -> Mapping[str, object]: + return { + "messages": [ + {"role": "system", "content": SYSTEM_PROMPT}, + {"role": "user", "content": WEATHER_QUESTION}, + ], + "tools": [ + { + "type": "function", + "function": { + "name": WEATHER_TOOL_NAME, + "description": _WEATHER_DESCRIPTION, + "parameters": dict(_WEATHER_PARAMETERS), + }, + } + ], + "reasoning_effort": "low", + "max_tokens": MAX_OUTPUT_TOKENS, + } + + +def _chat_turn_two(turn_one: ParsedTurn) -> Mapping[str, object]: + first: Final = _chat_turn_one() + tool_call_id: Final = turn_one.tool_call.id if turn_one.tool_call is not None else "" + return { + **first, + "messages": [ + *_message_list(first["messages"]), + turn_one.assistant_echo, + {"role": "tool", "tool_call_id": tool_call_id, "content": WEATHER_TOOL_RESULT}, + ], + } + + +def _message_list(messages: object) -> Sequence[object]: + return _OBJECT_LIST.validate_python(messages) if isinstance(messages, list) else [] + + +def _first_choice_message(plain: Mapping[str, object]) -> object: + try: + choices: Final = _CHOICES.validate_python(plain.get("choices")) + except ValidationError: + return dict(_EMPTY_ASSISTANT) + return choices[0].get("message", dict(_EMPTY_ASSISTANT)) if choices else dict(_EMPTY_ASSISTANT) + + +def _parse_chat(response: object) -> ParsedTurn | str: + plain: Final = _plain(response) + if plain is None: + return f"response is a {type(response).__name__}, not a chat completion" + try: + parsed: Final = _CHAT.validate_python(plain) + except ValidationError as error: + return f"chat completion did not parse: {_validation_detail(error)}" + if not parsed.choices: + return "chat completion returned no choices" + message: Final = parsed.choices[0].message + first_call: Final = message.tool_calls[0] if message.tool_calls else None + usage: Final = parsed.usage + details: Final = usage.prompt_tokens_details if usage is not None else None + return ParsedTurn( + tool_call=( + ToolCallSeen( + id=first_call.id, + name=first_call.function.name, + raw_arguments=first_call.function.arguments, + arguments=_arguments(first_call.function.arguments), + ) + if first_call is not None + else None + ), + text=message.content if isinstance(message.content, str) else "", + input_tokens=usage.prompt_tokens if usage is not None else 0, + cache_read_tokens=(details.cached_tokens or 0) if details is not None else 0, + assistant_echo=_first_choice_message(plain), + ) + + +def _responses_turn_one() -> Mapping[str, object]: + return { + "instructions": SYSTEM_PROMPT, + "input": [{"role": "user", "content": WEATHER_QUESTION}], + "tools": [ + { + "type": "function", + "name": WEATHER_TOOL_NAME, + "description": _WEATHER_DESCRIPTION, + "parameters": dict(_WEATHER_PARAMETERS), + } + ], + "reasoning": {"effort": "low"}, + "max_output_tokens": MAX_OUTPUT_TOKENS, + } + + +def _responses_turn_two(turn_one: ParsedTurn) -> Mapping[str, object]: + first: Final = _responses_turn_one() + call_id: Final = turn_one.tool_call.id if turn_one.tool_call is not None else "" + return { + **first, + "input": [ + *_message_list(first["input"]), + *_message_list(turn_one.assistant_echo), + {"type": "function_call_output", "call_id": call_id, "output": WEATHER_TOOL_RESULT}, + ], + } + + +def _parse_responses(response: object) -> ParsedTurn | str: + plain: Final = _plain(response) + if plain is None: + return f"response is a {type(response).__name__}, not a Responses API response" + try: + parsed: Final = _RESPONSES.validate_python(plain) + except ValidationError as error: + return f"Responses API response did not parse: {_validation_detail(error)}" + function_call: Final = next((item for item in parsed.output if item.type == "function_call"), None) + texts: Final = tuple( + content.text + for item in parsed.output + if item.type == "message" and item.content is not None + for content in item.content # comprehension-ok: flattening message content blocks + if content.type == "output_text" + ) + usage: Final = parsed.usage + details: Final = usage.input_tokens_details if usage is not None else None + return ParsedTurn( + tool_call=( + ToolCallSeen( + id=function_call.call_id, + name=function_call.name, + raw_arguments=function_call.arguments, + arguments=_arguments(function_call.arguments), + ) + if function_call is not None + else None + ), + text=" ".join(texts), + input_tokens=usage.input_tokens if usage is not None else 0, + cache_read_tokens=(details.cached_tokens or 0) if details is not None else 0, + assistant_echo=plain.get("output", []), + ) + + +def _messages_turn_one() -> Mapping[str, object]: + return { + "system": SYSTEM_PROMPT, + "messages": [{"role": "user", "content": WEATHER_QUESTION}], + "tools": [ + { + "name": WEATHER_TOOL_NAME, + "description": _WEATHER_DESCRIPTION, + "input_schema": dict(_WEATHER_PARAMETERS), + } + ], + "thinking": {"type": "enabled", "budget_tokens": THINKING_BUDGET_TOKENS}, + "max_tokens": MAX_OUTPUT_TOKENS, + } + + +def _messages_turn_two(turn_one: ParsedTurn) -> Mapping[str, object]: + first: Final = _messages_turn_one() + tool_use_id: Final = turn_one.tool_call.id if turn_one.tool_call is not None else "" + return { + **first, + "messages": [ + *_message_list(first["messages"]), + {"role": "assistant", "content": turn_one.assistant_echo}, + { + "role": "user", + "content": [{"type": "tool_result", "tool_use_id": tool_use_id, "content": WEATHER_TOOL_RESULT}], + }, + ], + } + + +def _parse_messages(response: object) -> ParsedTurn | str: + plain: Final = _plain(response) + if plain is None: + return f"response is a {type(response).__name__}, not a Messages API response" + try: + parsed: Final = _MESSAGES.validate_python(plain) + except ValidationError as error: + return f"Messages API response did not parse: {_validation_detail(error)}" + tool_use: Final = next((block for block in parsed.content if block.type == "tool_use"), None) + cache_read: Final = parsed.usage.cache_read_input_tokens or 0 + cache_creation: Final = parsed.usage.cache_creation_input_tokens or 0 + return ParsedTurn( + tool_call=( + ToolCallSeen( + id=tool_use.id, + name=tool_use.name, + raw_arguments=json.dumps(tool_use.input or {}), + arguments=MappingProxyType(dict(tool_use.input or {})), + ) + if tool_use is not None + else None + ), + text=" ".join(block.text for block in parsed.content if block.type == "text"), + input_tokens=parsed.usage.input_tokens + cache_read + cache_creation, + cache_read_tokens=cache_read, + assistant_echo=plain.get("content", []), + ) + + +CHAT_COMPLETIONS_SCRIPT: Final = EndpointScript( + endpoint="/v1/chat/completions", + route_type="acompletion", + metadata_key="metadata", + turn_one=_chat_turn_one, + turn_two=_chat_turn_two, + parse=_parse_chat, +) +RESPONSES_SCRIPT: Final = EndpointScript( + endpoint="/v1/responses", + route_type="aresponses", + metadata_key="litellm_metadata", + turn_one=_responses_turn_one, + turn_two=_responses_turn_two, + parse=_parse_responses, +) +MESSAGES_SCRIPT: Final = EndpointScript( + endpoint="/v1/messages", + route_type="anthropic_messages", + metadata_key="litellm_metadata", + turn_one=_messages_turn_one, + turn_two=_messages_turn_two, + parse=_parse_messages, +) +SCRIPTS: Final[Mapping[TestModelEndpoint, EndpointScript]] = MappingProxyType( + {script.endpoint: script for script in (CHAT_COMPLETIONS_SCRIPT, RESPONSES_SCRIPT, MESSAGES_SCRIPT)} +) +ENDPOINT_FOR_MODE: Final[Mapping[str, TestModelEndpoint]] = MappingProxyType( + {"chat": "/v1/chat/completions", "responses": "/v1/responses", "anthropic_messages": "/v1/messages"} +) + + +def response_cost(response: object) -> float | None: + cost: Final = get_hidden_params_dict(response).get("response_cost") + return float(cost) if isinstance(cost, int | float) else None + + +async def _turn(script: EndpointScript, body: Mapping[str, object], call: TurnCaller, call_id: str) -> TurnOutcome: + try: + called: Final = await call(script, body, call_id) + except Exception as error: # noqa: BLE001 # a provider or translation failure is the check's finding + return TurnOutcome(response=None, cost=None, error=f"{type(error).__name__}: {error}", call_id=call_id) + return TurnOutcome(response=called.response, cost=called.cost, error=None, call_id=call_id) + + +def _conversation_check( + turn_one: TurnOutcome, + parsed_one: ParsedTurn | str | None, + turn_two: TurnOutcome | None, + parsed_two: ParsedTurn | str | None, +) -> CheckResult: + if turn_one.error is not None: + return CheckResult("conversation", "fail", f"turn 1 request failed: {turn_one.error}") + if isinstance(parsed_one, str): + return CheckResult("conversation", "fail", f"turn 1 {parsed_one}") + if parsed_one is None or parsed_one.tool_call is None: + shown: Final = parsed_one.text[:160] if parsed_one is not None else "" + return CheckResult("conversation", "fail", f"turn 1 returned no tool call, the model answered: {shown!r}") + call: Final = parsed_one.tool_call + if call.name != WEATHER_TOOL_NAME: + return CheckResult("conversation", "fail", f"turn 1 called {call.name!r} instead of {WEATHER_TOOL_NAME!r}") + if call.arguments is None: + return CheckResult( + "conversation", "fail", f"turn 1 tool arguments are not a JSON object: {call.raw_arguments!r}" + ) + if "location" not in call.arguments: + return CheckResult("conversation", "fail", f"turn 1 tool arguments lack 'location': {call.raw_arguments!r}") + if turn_two is None or turn_two.error is not None: + reason: Final = turn_two.error if turn_two is not None else "not sent" + return CheckResult("conversation", "fail", f"turn 2 request with the tool result failed: {reason}") + if isinstance(parsed_two, str): + return CheckResult("conversation", "fail", f"turn 2 {parsed_two}") + if parsed_two is None or "72" not in parsed_two.text: + answer: Final = parsed_two.text[:160] if parsed_two is not None else "" + return CheckResult( + "conversation", "fail", f"turn 2 did not use the tool result {WEATHER_TOOL_RESULT!r}: {answer!r}" + ) + return CheckResult( + "conversation", + "pass", + f"turn 1 called {WEATHER_TOOL_NAME}({call.raw_arguments}), turn 2 answered {parsed_two.text[:120]!r}", + ) + + +def _cache_percent(parsed_two: ParsedTurn | None) -> float | None: + if parsed_two is None or parsed_two.input_tokens <= 0: + return None + return round(100.0 * parsed_two.cache_read_tokens / parsed_two.input_tokens, 1) + + +def _missed_cache(parsed: ParsedTurn | str | None) -> bool: + return isinstance(parsed, ParsedTurn) and parsed.cache_read_tokens == 0 + + +def _hit_cache(parsed: ParsedTurn | str | None) -> bool: + return isinstance(parsed, ParsedTurn) and parsed.cache_read_tokens > 0 + + +def _prompt_caching_check(parsed_two: ParsedTurn | None, attempts: int) -> CheckResult: + if parsed_two is None: + return CheckResult("prompt_caching", "fail", "not checked, the conversation did not reach turn 2") + percent: Final = _cache_percent(parsed_two) + attempt_note: Final = " on the second attempt" if attempts > 1 else "" + if parsed_two.cache_read_tokens > 0: + return CheckResult( + "prompt_caching", + "pass", + f"turn 2 read {parsed_two.cache_read_tokens} of {parsed_two.input_tokens} input tokens from cache " + f"({percent}%){attempt_note}", + ) + return CheckResult( + "prompt_caching", + "fail", + f"turn 2 reported 0 cached tokens on {parsed_two.input_tokens} input tokens in {attempts} attempts, the " + "provider did not reuse the repeated system prompt", + ) + + +def _cost_verdict(turn_number: int, cost: float | None) -> CheckResult | None: + if cost is None: + return CheckResult( + "cost_tracking", + "fail", + f"turn {turn_number} carried no response_cost, no price resolved for this model (not in the cost map and " + "no custom pricing on the deployment)", + ) + if cost == 0: + return CheckResult( + "cost_tracking", "warn", f"turn {turn_number} response_cost is $0, this model is priced at zero" + ) + return None + + +def _cost_tracking_check( + turn_one: TurnOutcome, turn_two: TurnOutcome | None, spend_rows: Mapping[str, float] | None +) -> CheckResult: + turns: Final = tuple(turn for turn in (turn_one, turn_two) if turn is not None and turn.response is not None) + if not turns: + return CheckResult("cost_tracking", "fail", "not checked, no turn returned a response") + costs: Final = tuple(turn.cost for turn in turns) + failed: Final = next( + (verdict for number, cost in enumerate(costs, start=1) if (verdict := _cost_verdict(number, cost)) is not None), + None, + ) + if failed is not None: + return failed + priced: Final = ", ".join(f"turn {number} ${cost:.6f}" for number, cost in enumerate(costs, start=1)) + if spend_rows is None: + return CheckResult("cost_tracking", "pass", f"{priced}, spend log not checked (no database)") + missing: Final = tuple(turn.call_id for turn in turns if turn.call_id not in spend_rows) + if missing: + return CheckResult( + "cost_tracking", + "fail", + f"{priced}, but no LiteLLM_SpendLogs row landed for request_id {', '.join(missing)}", + ) + logged: Final = ", ".join(f"${spend_rows[turn.call_id]:.6f}" for turn in turns) + return CheckResult("cost_tracking", "pass", f"{priced}, spend log rows written ({logged})") + + +def _status(checks: Sequence[CheckResult]) -> EndpointStatus: + statuses: Final = frozenset(check.status for check in checks) + if "fail" in statuses: + return "error" + return "warning" if "warn" in statuses else "success" + + +def overall_status(results: Sequence[EndpointResult]) -> EndpointStatus: + statuses: Final = frozenset(result.status for result in results) + if "error" in statuses: + return "error" + return "warning" if "warning" in statuses else "success" + + +async def run_endpoint_check( + script: EndpointScript, + call: TurnCaller, + *, + new_call_id: CallIdFactory, + spend_lookup: SpendLookup | None, +) -> EndpointResult: + started: Final = time.monotonic() + turn_one: Final = await _turn(script, script.turn_one(), call, new_call_id()) + parsed_one: Final = script.parse(turn_one.response) if turn_one.response is not None else None + ready_for_turn_two: Final = isinstance(parsed_one, ParsedTurn) and parsed_one.tool_call is not None + turn_two: Final = ( + await _turn(script, script.turn_two(parsed_one), call, new_call_id()) + if ready_for_turn_two and isinstance(parsed_one, ParsedTurn) + else None + ) + first_parsed_two: Final = ( + script.parse(turn_two.response) if turn_two is not None and turn_two.response is not None else None + ) + retried_turn_two: Final = ( + await _turn(script, script.turn_two(parsed_one), call, new_call_id()) + if isinstance(parsed_one, ParsedTurn) and _missed_cache(first_parsed_two) + else None + ) + retried_parsed_two: Final = ( + script.parse(retried_turn_two.response) + if retried_turn_two is not None and retried_turn_two.response is not None + else None + ) + turn_two_used: Final = retried_turn_two if _hit_cache(retried_parsed_two) else turn_two + parsed_two: Final = retried_parsed_two if _hit_cache(retried_parsed_two) else first_parsed_two + cache_attempts: Final = 1 if retried_turn_two is None else 2 + answered: Final = tuple( + turn for turn in (turn_one, turn_two_used) if turn is not None and turn.response is not None + ) + spend_rows: Final = ( + await spend_lookup(tuple(turn.call_id for turn in answered)) if spend_lookup is not None and answered else None + ) + second: Final = parsed_two if isinstance(parsed_two, ParsedTurn) else None + checks: Final = ( + _conversation_check(turn_one, parsed_one, turn_two_used, parsed_two), + _prompt_caching_check(second, cache_attempts), + _cost_tracking_check(turn_one, turn_two_used, spend_rows), + ) + return EndpointResult( + endpoint=script.endpoint, + status=_status(checks), + checks=checks, + cache_read_tokens=second.cache_read_tokens if second is not None else 0, + cache_read_percent=_cache_percent(second), + spend_usd=sum(turn.cost for turn in answered if turn.cost is not None), + duration_ms=int((time.monotonic() - started) * 1000), + ) + + +class TestModelCheck(BaseModel): + model_config = ConfigDict(frozen=True) + + name: Literal["conversation", "prompt_caching", "cost_tracking"] + status: Verdict + detail: str + + +class TestModelEndpointResult(BaseModel): + model_config = ConfigDict(frozen=True) + + endpoint: TestModelEndpoint + status: EndpointStatus + checks: Sequence[TestModelCheck] + cache_read_tokens: int + cache_read_percent: float | None + spend_usd: float + duration_ms: int + + +class TestModelResponse(BaseModel): + model_config = ConfigDict(frozen=True) + + status: EndpointStatus + model: str | None + model_id: str | None + endpoints: Sequence[TestModelEndpointResult] + + +def endpoint_result_payload(result: EndpointResult) -> TestModelEndpointResult: + return TestModelEndpointResult( + endpoint=result.endpoint, + status=result.status, + checks=[TestModelCheck(name=check.name, status=check.status, detail=check.detail) for check in result.checks], + cache_read_tokens=result.cache_read_tokens, + cache_read_percent=result.cache_read_percent, + spend_usd=result.spend_usd, + duration_ms=result.duration_ms, + ) diff --git a/tests/integration/management/test_model_health_check.py b/tests/integration/management/test_model_health_check.py index 992c1e8e875..36572d9d5b8 100644 --- a/tests/integration/management/test_model_health_check.py +++ b/tests/integration/management/test_model_health_check.py @@ -1,6 +1,7 @@ import json import os import uuid +from collections.abc import Mapping from email.message import Message from email.parser import BytesParser from email.policy import HTTP @@ -10,7 +11,15 @@ from typing import Final import httpx import pytest import yaml -from integration._support.client import JSON_OBJECT, Gateway, Scenario, eventually, object_value, string_value +from integration._support.client import ( + JSON_OBJECT, + Gateway, + Scenario, + eventually, + list_value, + object_value, + string_value, +) from integration._support.database import scratch_database from integration._support.process import owned_proxy from integration._support.upstream import ScenarioHandle, delete_scenario, register_scenario @@ -1426,3 +1435,139 @@ def test_health_check_of_gpt_transcribe_probes_without_verbose_json_and_reports_ (request.method, request.target, tuple(sorted(_multipart_text_fields(_multipart_parts(request)).items()))) for request in requests } == {("POST", "/v1/audio/transcriptions", (("model", "gpt-transcribe"),))} + + +_TEST_MODEL_TOOL_CALL_ID: Final = "call_boston_wire" + + +def _test_model_chat_reply(request: Request, *, cached_tokens: int) -> Reply: + assert request.target == "/v1/chat/completions", request.target + body: Final = JSON_OBJECT.validate_json(request.body) + assert "cache_control" not in request.body.decode(), request.body.decode() + messages: Final = body["messages"] + assert isinstance(messages, list) and isinstance(messages[0], dict), body + assert messages[0]["role"] == "system", messages[0] + tool_turn: Final = any(isinstance(message, dict) and message.get("role") == "tool" for message in messages) + if tool_turn: + assert isinstance(messages[-1], dict), messages + assert messages[-1]["tool_call_id"] == _TEST_MODEL_TOOL_CALL_ID, messages[-1] + message: dict[str, JsonValue] = {"role": "assistant", "content": "Boston is 72F and sunny."} + else: + assert body["tools"], body + message = { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": _TEST_MODEL_TOOL_CALL_ID, + "type": "function", + "function": {"name": "get_weather", "arguments": json.dumps({"location": "Boston"})}, + } + ], + } + return Reply( + body=json.dumps( + { + "id": f"chatcmpl-{uuid.uuid4().hex}", + "object": "chat.completion", + "created": 1, + "model": body["model"], + "choices": [{"index": 0, "finish_reason": "stop", "message": message}], + "usage": { + "prompt_tokens": 5000, + "completion_tokens": 12, + "total_tokens": 5012, + "prompt_tokens_details": {"cached_tokens": cached_tokens if tool_turn else 0}, + }, + } + ).encode() + ) + + +def _check_statuses(endpoint_result: Mapping[str, JsonValue]) -> dict[str, tuple[str, str]]: + return { + string_value(object_value(check)["name"]): ( + string_value(object_value(check)["status"]), + string_value(object_value(check)["detail"]), + ) + for check in list_value(endpoint_result["checks"]) + } + + +@pytest.mark.timeout(240) +def test_health_test_model_runs_the_tool_conversation_against_a_saved_deployment_and_finds_its_spend_rows( + gateway: Gateway, +) -> None: + with ( + gateway.scenario() as scenario, + wire_server(lambda request: _test_model_chat_reply(request, cached_tokens=4800)) as wire, + ): + model_name: Final = scenario.model(api_base=f"{wire.url}/v1", api_key="synthetic-test-model-key") + model_id: Final = next( + string_value(object_value(info["model_info"])["id"]) + for info in (object_value(entry) for entry in list_value(gateway.get("/v1/model/info")["data"])) + if info["model_name"] == model_name + ) + + response: Final = gateway.request( + "POST", "/health/test_model", {"model_id": model_id, "endpoints": ["/v1/chat/completions"]} + ) + + assert response.status_code == 200, response.text + payload: Final = JSON_OBJECT.validate_json(response.content) + assert payload["status"] == "success", response.text + assert payload["model_id"] == model_id, response.text + endpoint_result: Final = object_value(list_value(payload["endpoints"])[0]) + assert endpoint_result["endpoint"] == "/v1/chat/completions", response.text + checks: Final = _check_statuses(endpoint_result) + assert checks["conversation"][0] == "pass", response.text + assert checks["prompt_caching"] == ("pass", "turn 2 read 4800 of 5000 input tokens from cache (96.0%)"), ( + response.text + ) + assert checks["cost_tracking"][0] == "pass", response.text + assert "spend log rows written" in checks["cost_tracking"][1], response.text + assert endpoint_result["cache_read_tokens"] == 4800, response.text + requests: Final = wire.drain() + assert [request.headers["authorization"] for request in requests] == ["Bearer synthetic-test-model-key"] * 2, ( + response.text + ) + rows: Final = gateway.rows( + 'SELECT "spend", "request_tags"::text AS tags FROM "LiteLLM_SpendLogs" WHERE "model_id" = %s', + (model_id,), + ) + assert len(rows) == 2, rows + assert all(isinstance(row["spend"], int | float) and row["spend"] > 0 for row in rows), rows + assert all("litellm_test_model:" in string_value(row["tags"]) for row in rows), rows + + +@pytest.mark.timeout(240) +def test_health_test_model_names_the_check_that_failed_for_an_unsaved_model_that_never_caches( + gateway: Gateway, +) -> None: + with wire_server(lambda request: _test_model_chat_reply(request, cached_tokens=0)) as wire: + response: Final = gateway.request( + "POST", + "/health/test_model", + { + "litellm_params": { + "model": f"openai/unpriced-{uuid.uuid4().hex}", + "api_base": f"{wire.url}/v1", + "api_key": "synthetic-unsaved-key", + }, + "endpoints": ["/v1/chat/completions"], + }, + ) + + assert response.status_code == 200, response.text + payload: Final = JSON_OBJECT.validate_json(response.content) + assert payload["status"] == "error", response.text + checks: Final = _check_statuses(object_value(list_value(payload["endpoints"])[0])) + assert checks["conversation"][0] == "pass", response.text + assert checks["prompt_caching"] == ( + "fail", + "turn 2 reported 0 cached tokens on 5000 input tokens in 2 attempts, the provider did not reuse the " + "repeated system prompt", + ), response.text + assert checks["cost_tracking"][0] == "fail", response.text + assert "no price resolved for this model" in checks["cost_tracking"][1], response.text + assert len(wire.drain()) >= 2, response.text diff --git a/tests/unit/proxy/health_endpoints/test_model_checks.py b/tests/unit/proxy/health_endpoints/test_model_checks.py new file mode 100644 index 00000000000..35094b8f32e --- /dev/null +++ b/tests/unit/proxy/health_endpoints/test_model_checks.py @@ -0,0 +1,373 @@ +import asyncio +import json +from collections.abc import Mapping, Sequence +from typing import Final + +import pytest + +from litellm.proxy.health_endpoints.model_checks import ( + CHAT_COMPLETIONS_SCRIPT, + MESSAGES_SCRIPT, + RESPONSES_SCRIPT, + WEATHER_TOOL_RESULT, + CalledTurn, + EndpointScript, + endpoint_result_payload, + overall_status, + run_endpoint_check, +) +from litellm.types.llms.openai import ResponsesAPIResponse +from litellm.types.utils import ModelResponse + +_TOOL_CALL_ID: Final = "call_boston" + + +def _chat_reply(*, tool_call: bool, text: str = "", cached_tokens: int = 0, prompt_tokens: int = 5000) -> ModelResponse: + message: Final = ( + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": _TOOL_CALL_ID, + "type": "function", + "function": {"name": "get_weather", "arguments": json.dumps({"location": "Boston"})}, + } + ], + } + if tool_call + else {"role": "assistant", "content": text} + ) + return ModelResponse( + id="chatcmpl-1", + model="gpt-test", + choices=[{"index": 0, "finish_reason": "tool_calls" if tool_call else "stop", "message": message}], + usage={ + "prompt_tokens": prompt_tokens, + "completion_tokens": 12, + "total_tokens": prompt_tokens + 12, + "prompt_tokens_details": {"cached_tokens": cached_tokens}, + }, + ) + + +def _responses_reply(*, tool_call: bool, text: str = "", cached_tokens: int = 0) -> ResponsesAPIResponse: + output: Final = ( + [ + {"type": "reasoning", "id": "rs_1", "summary": []}, + { + "type": "function_call", + "id": "fc_1", + "call_id": _TOOL_CALL_ID, + "name": "get_weather", + "arguments": json.dumps({"location": "Boston"}), + "status": "completed", + }, + ] + if tool_call + else [ + { + "type": "message", + "id": "msg_1", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": text, "annotations": []}], + } + ] + ) + return ResponsesAPIResponse.model_validate( + { + "id": "resp_1", + "object": "response", + "created_at": 1, + "status": "completed", + "model": "gpt-test", + "output": output, + "parallel_tool_calls": True, + "tool_choice": "auto", + "tools": [], + "usage": { + "input_tokens": 5000, + "output_tokens": 12, + "total_tokens": 5012, + "input_tokens_details": {"cached_tokens": cached_tokens}, + }, + } + ) + + +def _messages_reply(*, tool_call: bool, text: str = "", cached_tokens: int = 0) -> Mapping[str, object]: + content: Final = ( + [ + {"type": "thinking", "thinking": "look it up", "signature": "sig"}, + {"type": "tool_use", "id": _TOOL_CALL_ID, "name": "get_weather", "input": {"location": "Boston"}}, + ] + if tool_call + else [{"type": "text", "text": text}] + ) + return { + "id": "msg_1", + "type": "message", + "role": "assistant", + "model": "claude-test", + "content": content, + "stop_reason": "tool_use" if tool_call else "end_turn", + "usage": { + "input_tokens": 5000 - cached_tokens, + "output_tokens": 12, + "cache_read_input_tokens": cached_tokens, + "cache_creation_input_tokens": 0, + }, + } + + +class _ScriptedProvider: + """Answers turn 1 with a tool call and every later turn from ``second_turns``, recording what was sent.""" + + def __init__( + self, + first: CalledTurn, + second_turns: Sequence[CalledTurn | Exception], + ) -> None: + self.first = first + self.second_turns = list(second_turns) + self.sent: list[Mapping[str, object]] = [] + + async def __call__(self, script: EndpointScript, body: Mapping[str, object], call_id: str) -> CalledTurn: + self.sent.append(body) + if len(self.sent) == 1: + return self.first + outcome: Final = self.second_turns.pop(0) + if isinstance(outcome, Exception): + raise outcome + return outcome + + +def _ids() -> Sequence[str]: + return ("turn-1", "turn-2", "turn-2-retry") + + +def _run(script: EndpointScript, provider: _ScriptedProvider, spend_rows: Mapping[str, float] | None = None): + ids: Final = iter(_ids()) + + async def spend_lookup(request_ids: Sequence[str]) -> Mapping[str, float]: + return spend_rows or {} + + return asyncio.run( + run_endpoint_check( + script, + provider, + new_call_id=lambda: next(ids), + spend_lookup=spend_lookup if spend_rows is not None else None, + ) + ) + + +def _checks(result) -> Mapping[str, tuple[str, str]]: + return {check.name: (check.status, check.detail) for check in result.checks} + + +@pytest.mark.parametrize( + ("script", "first", "second"), + [ + ( + CHAT_COMPLETIONS_SCRIPT, + _chat_reply(tool_call=True), + _chat_reply(tool_call=False, text="Boston is 72F and sunny.", cached_tokens=4800), + ), + ( + RESPONSES_SCRIPT, + _responses_reply(tool_call=True), + _responses_reply(tool_call=False, text="Boston is 72F and sunny.", cached_tokens=4800), + ), + ( + MESSAGES_SCRIPT, + _messages_reply(tool_call=True), + _messages_reply(tool_call=False, text="Boston is 72F and sunny.", cached_tokens=4800), + ), + ], + ids=["chat", "responses", "messages"], +) +def test_every_endpoint_passes_all_three_checks_and_feeds_turn_one_back_in_its_own_shape(script, first, second): + provider: Final = _ScriptedProvider(CalledTurn(first, 0.004), [CalledTurn(second, 0.0005)]) + + result: Final = _run(script, provider, spend_rows={"turn-1": 0.004, "turn-2": 0.0005}) + + assert result.status == "success", result + checks: Final = _checks(result) + assert checks["conversation"][0] == "pass", checks + assert checks["prompt_caching"] == ("pass", "turn 2 read 4800 of 5000 input tokens from cache (96.0%)"), checks + assert checks["cost_tracking"] == ( + "pass", + "turn 1 $0.004000, turn 2 $0.000500, spend log rows written ($0.004000, $0.000500)", + ), checks + assert result.cache_read_tokens == 4800 + assert result.cache_read_percent == 96.0 + assert result.spend_usd == pytest.approx(0.0045) + assert len(provider.sent) == 2 + turn_two: Final = provider.sent[1] + sent_json: Final = json.dumps(turn_two) + assert _TOOL_CALL_ID in sent_json and WEATHER_TOOL_RESULT in sent_json, turn_two + assert "cache_control" not in sent_json and "cachePoint" not in sent_json, turn_two + if script is CHAT_COMPLETIONS_SCRIPT: + tool_message: Final = turn_two["messages"][-1] + assert tool_message == {"role": "tool", "tool_call_id": _TOOL_CALL_ID, "content": WEATHER_TOOL_RESULT} + assert turn_two["messages"][-2]["tool_calls"][0]["id"] == _TOOL_CALL_ID + elif script is RESPONSES_SCRIPT: + assert turn_two["input"][-1] == { + "type": "function_call_output", + "call_id": _TOOL_CALL_ID, + "output": WEATHER_TOOL_RESULT, + } + assert [item["type"] for item in turn_two["input"][1:-1]] == ["reasoning", "function_call"] + else: + assert turn_two["messages"][-1]["content"] == [ + {"type": "tool_result", "tool_use_id": _TOOL_CALL_ID, "content": WEATHER_TOOL_RESULT} + ] + assert turn_two["messages"][-2]["content"][0]["type"] == "thinking" + + +def test_a_model_that_answers_in_text_instead_of_calling_the_tool_fails_the_conversation_and_stops(): + provider: Final = _ScriptedProvider(CalledTurn(_chat_reply(tool_call=False, text="It is sunny."), 0.004), []) + + result: Final = _run(CHAT_COMPLETIONS_SCRIPT, provider) + + assert result.status == "error" + checks: Final = _checks(result) + assert checks["conversation"] == ("fail", "turn 1 returned no tool call, the model answered: 'It is sunny.'") + assert checks["prompt_caching"] == ("fail", "not checked, the conversation did not reach turn 2") + assert checks["cost_tracking"] == ("pass", "turn 1 $0.004000, spend log not checked (no database)") + assert len(provider.sent) == 1 + + +def test_a_turn_two_provider_error_is_reported_as_the_conversation_failure(): + provider: Final = _ScriptedProvider( + CalledTurn(_chat_reply(tool_call=True), 0.004), [RuntimeError("400 invalid tool message")] + ) + + result: Final = _run(CHAT_COMPLETIONS_SCRIPT, provider) + + checks: Final = _checks(result) + assert checks["conversation"] == ( + "fail", + "turn 2 request with the tool result failed: RuntimeError: 400 invalid tool message", + ) + assert result.status == "error" + + +def test_an_answer_that_ignores_the_tool_result_fails_the_conversation(): + provider: Final = _ScriptedProvider( + CalledTurn(_chat_reply(tool_call=True), 0.004), + [CalledTurn(_chat_reply(tool_call=False, text="I cannot tell.", cached_tokens=4800), 0.0005)], + ) + + checks: Final = _checks(_run(CHAT_COMPLETIONS_SCRIPT, provider)) + + assert checks["conversation"] == ("fail", "turn 2 did not use the tool result '72F and sunny': 'I cannot tell.'") + assert checks["prompt_caching"][0] == "pass" + + +def test_zero_cached_tokens_is_retried_once_and_fails_when_the_retry_misses_too(): + provider: Final = _ScriptedProvider( + CalledTurn(_chat_reply(tool_call=True), 0.004), + [ + CalledTurn(_chat_reply(tool_call=False, text="72F", cached_tokens=0), 0.004), + CalledTurn(_chat_reply(tool_call=False, text="72F", cached_tokens=0), 0.004), + ], + ) + + result: Final = _run(CHAT_COMPLETIONS_SCRIPT, provider) + + assert len(provider.sent) == 3 + assert _checks(result)["prompt_caching"] == ( + "fail", + "turn 2 reported 0 cached tokens on 5000 input tokens in 2 attempts, the provider did not reuse the " + "repeated system prompt", + ) + assert result.status == "error" + assert result.cache_read_tokens == 0 + + +def test_a_cache_hit_on_the_retry_passes_and_says_so(): + provider: Final = _ScriptedProvider( + CalledTurn(_chat_reply(tool_call=True), 0.004), + [ + CalledTurn(_chat_reply(tool_call=False, text="72F", cached_tokens=0), 0.004), + CalledTurn(_chat_reply(tool_call=False, text="72F", cached_tokens=5000), 0.0005), + ], + ) + + result: Final = _run(CHAT_COMPLETIONS_SCRIPT, provider) + + assert _checks(result)["prompt_caching"] == ( + "pass", + "turn 2 read 5000 of 5000 input tokens from cache (100.0%) on the second attempt", + ) + assert result.spend_usd == pytest.approx(0.0045) + + +def test_a_model_without_a_price_fails_cost_tracking_and_a_zero_priced_model_only_warns(): + unpriced: Final = _ScriptedProvider( + CalledTurn(_chat_reply(tool_call=True), None), + [CalledTurn(_chat_reply(tool_call=False, text="72F", cached_tokens=4800), None)], + ) + free: Final = _ScriptedProvider( + CalledTurn(_chat_reply(tool_call=True), 0.0), + [CalledTurn(_chat_reply(tool_call=False, text="72F", cached_tokens=4800), 0.0)], + ) + + unpriced_result: Final = _run(CHAT_COMPLETIONS_SCRIPT, unpriced) + free_result: Final = _run(CHAT_COMPLETIONS_SCRIPT, free) + + assert _checks(unpriced_result)["cost_tracking"] == ( + "fail", + "turn 1 carried no response_cost, no price resolved for this model (not in the cost map and no custom " + "pricing on the deployment)", + ) + assert unpriced_result.status == "error" + assert _checks(free_result)["cost_tracking"] == ( + "warn", + "turn 1 response_cost is $0, this model is priced at zero", + ) + assert free_result.status == "warning" + assert overall_status([unpriced_result, free_result]) == "error" + assert overall_status([free_result]) == "warning" + + +def test_a_missing_spend_log_row_fails_cost_tracking_even_when_the_response_was_priced(): + provider: Final = _ScriptedProvider( + CalledTurn(_chat_reply(tool_call=True), 0.004), + [CalledTurn(_chat_reply(tool_call=False, text="72F", cached_tokens=4800), 0.0005)], + ) + + checks: Final = _checks(_run(CHAT_COMPLETIONS_SCRIPT, provider, spend_rows={"turn-1": 0.004})) + + assert checks["cost_tracking"] == ( + "fail", + "turn 1 $0.004000, turn 2 $0.000500, but no LiteLLM_SpendLogs row landed for request_id turn-2", + ) + + +def test_a_response_that_is_not_the_endpoint_shape_fails_the_conversation_with_the_parse_error(): + provider: Final = _ScriptedProvider(CalledTurn({"id": "x", "object": "nonsense"}, 0.001), []) + + checks: Final = _checks(_run(MESSAGES_SCRIPT, provider)) + + assert checks["conversation"][0] == "fail" + assert checks["conversation"][1].startswith("turn 1 Messages API response did not parse: content"), checks + + +def test_payload_lists_the_checks_in_order_with_their_details(): + provider: Final = _ScriptedProvider( + CalledTurn(_chat_reply(tool_call=True), 0.004), + [CalledTurn(_chat_reply(tool_call=False, text="72F", cached_tokens=4800), 0.0005)], + ) + + payload: Final = endpoint_result_payload(_run(CHAT_COMPLETIONS_SCRIPT, provider)).model_dump() + + assert payload["endpoint"] == "/v1/chat/completions" + assert payload["status"] == "success" + assert [check["name"] for check in payload["checks"]] == ["conversation", "prompt_caching", "cost_tracking"] + assert all(check["status"] == "pass" for check in payload["checks"]) + assert payload["cache_read_percent"] == 96.0 + assert isinstance(payload["duration_ms"], int) diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 96c7c34621b..cfdee1cf59f 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -7967,6 +7967,46 @@ export interface paths { patch?: never; trace?: never; }; + "/health/test_model": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** + * Test Model + * @description Run the live checks on a model, the way a client would use it, and report what works. + * + * Each selected endpoint gets one two-turn conversation with a long system prompt and a tool. Turn 1 asks + * a question the tool answers, turn 2 sends the tool result back. Three checks come out of it: + * + * - `conversation`: both turns parse, turn 1 calls the tool, turn 2 uses the tool result + * - `prompt_caching`: turn 2 reports cached input tokens, with no cache markers in the request + * - `cost_tracking`: every turn carries a non-zero response cost, and the spend log row lands when a + * database is connected + * + * Example: + * ```bash + * curl -X POST 'http://localhost:4000/health/test_model' \ + * -H "Authorization: Bearer $LITELLM_MASTER_KEY" \ + * -H 'Content-Type: application/json' \ + * -d '{"model_id": "a1b2c3", "endpoints": ["/v1/chat/completions", "/v1/responses"]}' + * ``` + * + * Returns `status` (`success`, `warning` or `error`), the tested `model`, and one entry per endpoint + * with its `status`, the three `checks` (`name`, `status`, `detail`), `cache_read_tokens`, + * `cache_read_percent`, `spend_usd` and `duration_ms`. + */ + post: operations["test_model_health_test_model_post"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/images/edits": { parameters: { query?: never; @@ -27753,6 +27793,33 @@ export interface components { [key: string]: unknown; }; }; + /** Body_test_model_health_test_model_post */ + Body_test_model_health_test_model_post: { + /** + * Endpoints + * @description Which endpoints to test: /v1/chat/completions, /v1/responses, /v1/messages. Defaults to the model's native endpoint. + */ + endpoints?: ("/v1/chat/completions" | "/v1/responses" | "/v1/messages")[] | null; + /** + * Litellm Params + * @description Parameters of a model that is not saved yet (model, api_key, api_base, api_version), as sent to /model/new. When model_id is also given these override the saved deployment's values. + */ + litellm_params?: { + [key: string]: unknown; + } | null; + /** + * Model Id + * @description The deployment to test, by its model_info.id. The saved credentials and pricing are used. + */ + model_id?: string | null; + /** + * Model Info + * @description Model info of an unsaved model, for example mode or custom pricing. + */ + model_info?: { + [key: string]: unknown; + } | null; + }; /** Body_token_endpoint__mcp_server_name__token_post */ Body_token_endpoint__mcp_server_name__token_post: { /** Assertion */ @@ -47363,6 +47430,58 @@ export interface components { /** Success */ success: boolean; }; + /** TestModelCheck */ + TestModelCheck: { + /** Detail */ + detail: string; + /** + * Name + * @enum {string} + */ + name: "conversation" | "prompt_caching" | "cost_tracking"; + /** + * Status + * @enum {string} + */ + status: "pass" | "warn" | "fail"; + }; + /** TestModelEndpointResult */ + TestModelEndpointResult: { + /** Cache Read Percent */ + cache_read_percent: number | null; + /** Cache Read Tokens */ + cache_read_tokens: number; + /** Checks */ + checks: components["schemas"]["TestModelCheck"][]; + /** Duration Ms */ + duration_ms: number; + /** + * Endpoint + * @enum {string} + */ + endpoint: "/v1/chat/completions" | "/v1/responses" | "/v1/messages"; + /** Spend Usd */ + spend_usd: number; + /** + * Status + * @enum {string} + */ + status: "success" | "warning" | "error"; + }; + /** TestModelResponse */ + TestModelResponse: { + /** Endpoints */ + endpoints: components["schemas"]["TestModelEndpointResult"][]; + /** Model */ + model: string | null; + /** Model Id */ + model_id: string | null; + /** + * Status + * @enum {string} + */ + status: "success" | "warning" | "error"; + }; /** * TestPoliciesAndGuardrailsRequest * @description Request body for POST /utils/test_policies_and_guardrails. @@ -63001,6 +63120,39 @@ export interface operations { }; }; }; + test_model_health_test_model_post: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: { + content: { + "application/json": components["schemas"]["Body_test_model_health_test_model_post"]; + }; + }; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["TestModelResponse"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; image_edit_api_images_edits_post: { parameters: { query?: {