From 1a656c07952fbb7f420b85e2e14a2d66245d2db7 Mon Sep 17 00:00:00 2001 From: Alejandro Tapia Date: Tue, 24 Feb 2026 17:37:00 -0800 Subject: [PATCH] removed extra comma typo --- .../health_endpoints/_health_endpoints.py | 107 +++++++++++------- 1 file changed, 63 insertions(+), 44 deletions(-) diff --git a/litellm/proxy/health_endpoints/_health_endpoints.py b/litellm/proxy/health_endpoints/_health_endpoints.py index b844a959622..f3bed3656f6 100644 --- a/litellm/proxy/health_endpoints/_health_endpoints.py +++ b/litellm/proxy/health_endpoints/_health_endpoints.py @@ -110,26 +110,31 @@ def _resolve_os_environ_variables(params: dict) -> dict: def get_callback_identifier(callback): """ Get the callback identifier string, handling both strings and objects. - + This function extracts a string identifier from a callback, which can be: - A string (returned as-is) - An object with a callback_name attribute - An object registered in CustomLoggerRegistry - Falls back to callback_name() helper function - + Args: callback: The callback to identify (can be str or object) - + Returns: str: The callback identifier string """ if isinstance(callback, str): return callback - if hasattr(callback, 'callback_name') and callback.callback_name: + if hasattr(callback, "callback_name") and callback.callback_name: return callback.callback_name - if hasattr(callback, '__class__'): - callback_strs = CustomLoggerRegistry.get_all_callback_strs_from_class_type(callback.__class__) - if hasattr(callback, 'callback_name') and callback.callback_name in callback_strs: + if hasattr(callback, "__class__"): + callback_strs = CustomLoggerRegistry.get_all_callback_strs_from_class_type( + callback.__class__ + ) + if ( + hasattr(callback, "callback_name") + and callback.callback_name in callback_strs + ): return callback.callback_name if callback_strs: return callback_strs[0] @@ -151,7 +156,7 @@ services = Union[ "datadog_llm_observability", "generic_api", "arize", - "sqs" + "sqs", ], str, ] @@ -224,7 +229,7 @@ async def health_services_endpoint( # noqa: PLR0915 "datadog_llm_observability", "generic_api", "arize", - "sqs" + "sqs", ]: raise HTTPException( status_code=400, @@ -245,7 +250,7 @@ async def health_services_endpoint( # noqa: PLR0915 if cb_id == service: service_in_success_callbacks = True break - + if ( service == "openmeter" or service == "braintrust" @@ -320,6 +325,7 @@ async def health_services_endpoint( # noqa: PLR0915 ) elif service == "sqs": from litellm.integrations.sqs import SQSLogger + sqs_logger = SQSLogger() response = await sqs_logger.async_health_check() return { @@ -518,12 +524,12 @@ async def _save_health_check_to_db( def _build_model_param_to_info_mapping(model_list: list) -> dict: """ Build a mapping from model parameter to model info (model_name, model_id). - + Multiple models might share the same model parameter, so we use a list. - + Args: model_list: List of model configurations - + Returns: Dictionary mapping model parameter to list of model info dicts """ @@ -534,14 +540,16 @@ def _build_model_param_to_info_mapping(model_list: list) -> dict: model_id = model_info.get("id") litellm_params = model.get("litellm_params", {}) model_param = litellm_params.get("model") - + if model_param and model_name: if model_param not in model_param_to_info: model_param_to_info[model_param] = [] - model_param_to_info[model_param].append({ - "model_name": model_name, - "model_id": model_id, - }) + model_param_to_info[model_param].append( + { + "model_name": model_name, + "model_id": model_id, + } + ) return model_param_to_info @@ -552,19 +560,19 @@ def _aggregate_health_check_results( ) -> dict: """ Aggregate health check results per unique model. - + Uses (model_id, model_name) as key, or (None, model_name) if model_id is None. - + Args: model_param_to_info: Mapping from model parameter to model info healthy_endpoints: List of healthy endpoint results unhealthy_endpoints: List of unhealthy endpoint results - + Returns: Dictionary mapping (model_id, model_name) to aggregated health check results """ model_results = {} - + # Process healthy endpoints for endpoint in healthy_endpoints: model_param = endpoint.get("model") @@ -580,7 +588,7 @@ def _aggregate_health_check_results( "error_message": None, } model_results[key]["healthy_count"] += 1 - + # Process unhealthy endpoints for endpoint in unhealthy_endpoints: model_param = endpoint.get("model") @@ -600,7 +608,7 @@ def _aggregate_health_check_results( # Use the first error message encountered if not model_results[key]["error_message"] and error_message: model_results[key]["error_message"] = str(error_message)[:500] - + return model_results @@ -613,14 +621,14 @@ async def _save_health_check_results_if_changed( ): """ Save health check results to database, but only if status changed or >1 hour since last save. - + OPTIMIZATION: Only saves to database if the status has changed from the last saved check. This dramatically reduces database writes when health status remains stable. - + - Stable systems: ~1 write/hour per model (instead of 12 writes/hour with 5-min intervals) - Status changes: Immediate write (no delay) - Result: ~92% reduction in DB writes for stable systems, while maintaining real-time updates on changes - + Args: prisma_client: Database client model_results: Dictionary of aggregated health check results per model @@ -630,7 +638,7 @@ async def _save_health_check_results_if_changed( """ for result in model_results.values(): new_status = "healthy" if result["healthy_count"] > 0 else "unhealthy" - + # Check if we should save this result should_save = True lookup_key = result["model_id"] if result["model_id"] else result["model_name"] @@ -641,6 +649,7 @@ async def _save_health_check_results_if_changed( # Check if last check was recent (within 1 hour) if last_check.checked_at: from datetime import datetime, timezone + time_since_last_check = ( datetime.now(timezone.utc) - last_check.checked_at ).total_seconds() @@ -648,7 +657,7 @@ async def _save_health_check_results_if_changed( # This ensures we still get periodic updates even if status is stable if time_since_last_check < 3600: # 1 hour threshold should_save = False - + if should_save: asyncio.create_task( prisma_client.save_health_check_result( @@ -675,27 +684,27 @@ async def _save_background_health_checks_to_db( ): """ Save background health check results to database for each model. - + Maps health check endpoints back to their original models to get model_name and model_id. Aggregates results per unique model (by model_id if available, otherwise model_name). - + OPTIMIZATION: Only saves to database if the status has changed from the last saved check. This dramatically reduces database writes when health status remains stable. """ if prisma_client is None: return - + try: # Step 1: Build mapping from model parameter to model info model_param_to_info = _build_model_param_to_info_mapping(model_list) - + # Step 2: Aggregate health check results per unique model model_results = _aggregate_health_check_results( model_param_to_info, healthy_endpoints, unhealthy_endpoints, ) - + # Step 3: Get latest health checks for all models in one query to compare status latest_checks = await prisma_client.get_all_latest_health_checks() latest_checks_map = {} @@ -704,7 +713,7 @@ async def _save_background_health_checks_to_db( key = check.model_id if check.model_id else check.model_name if key not in latest_checks_map: latest_checks_map[key] = check - + # Step 4: Save aggregated results, but only if status changed await _save_health_check_results_if_changed( prisma_client, @@ -729,6 +738,7 @@ async def _perform_health_check_and_save( start_time, user_id, model_id=None, + max_concurrency=None, ): """Helper function to perform health check and save results to database""" healthy_endpoints, unhealthy_endpoints = await perform_health_check( @@ -736,6 +746,7 @@ async def _perform_health_check_and_save( cli_model=cli_model, model=target_model, details=details, + max_concurrency=max_concurrency, model_id=model_id, ) @@ -793,6 +804,7 @@ async def health_endpoint( import time from litellm.proxy.proxy_server import ( + health_check_concurrency, health_check_details, health_check_results, llm_model_list, @@ -845,6 +857,7 @@ async def health_endpoint( start_time=start_time, user_id=user_api_key_dict.user_id, model_id=None, # CLI model doesn't have model_id + max_concurrency=health_check_concurrency, ) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -868,6 +881,7 @@ async def health_endpoint( start_time=start_time, user_id=user_api_key_dict.user_id, model_id=model_id, + max_concurrency=health_check_concurrency, ) except Exception as e: verbose_proxy_logger.error( @@ -1424,11 +1438,11 @@ async def test_model_connection( status_code=500, detail={"error": CommonProxyErrors.db_not_connected_error.value}, ) - + # Get model name from litellm_params request_litellm_params = litellm_params or {} model_name = 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 = {} @@ -1436,34 +1450,39 @@ async def test_model_connection( try: # First try to find by proxy model_name (e.g., "gpt-4o") deployments = llm_router.get_model_list(model_name=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 = llm_router.get_model_list(model_name=None) if all_deployments: for deployment in all_deployments: - if deployment.get("litellm_params", {}).get("model") == model_name: + 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", {})) + config_litellm_params = dict( + deployments[0].get("litellm_params", {}) + ) except Exception as e: verbose_proxy_logger.debug( f"Could not find model {model_name} in router: {e}. " "Proceeding with request params only." ) - + # Merge: config params (from proxy config) as base, request params override # This allows users to override specific params while using config for credentials merged_litellm_params = {**config_litellm_params, **request_litellm_params} - + # Resolve os.environ/ environment variables in any remaining request params # This handles cases where user explicitly passes os.environ/ values to override config litellm_params = _resolve_os_environ_variables(merged_litellm_params) - + ## Auth check await ModelManagementAuthChecks.can_user_make_model_call( model_params=Deployment(