scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-03 07:23:33 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 166

feat(proxy): add POST /health/test_model live two-turn model checks

Browse source 
Runs one tool-calling conversation per selected endpoint (/v1/chat/completions,
/v1/responses, /v1/messages) against a saved deployment or unsaved litellm_params
and reports conversation, prompt_caching and cost_tracking verdicts, including
whether LiteLLM_SpendLogs rows landed
This commit is contained in:
devin-ai-integration[bot] 2026-10-30 21:55:12 +00:00
parent 467847b2e9
commit 6e52ee2818
6 changed files with 1840 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/constants.py
View file

@ -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

529
litellm/proxy/health_endpoints/_health_endpoints.py
View file

@ -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,

729
litellm/proxy/health_endpoints/model_checks.py Normal file
View file

@ -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,
)

147
tests/integration/management/test_model_health_check.py
View file

@ -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

373
tests/unit/proxy/health_endpoints/test_model_checks.py Normal file
View file

@ -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)

152
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -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?: {