""" FALLBACK MANAGEMENT ENDPOINTS Dedicated endpoints for managing model fallbacks separately from general config. POST /fallback - Create or update fallbacks for a specific model GET /fallback/{model} - Get fallbacks for a specific model DELETE /fallback/{model} - Delete fallbacks for a specific model """ # pyright: reportMissingImports=false import json from typing import TYPE_CHECKING, Dict, List, Literal from litellm._logging import verbose_proxy_logger from litellm.proxy._types import UserAPIKeyAuth from litellm.proxy.auth.model_checks import get_all_fallbacks from litellm.proxy.auth.user_api_key_auth import user_api_key_auth if TYPE_CHECKING: from fastapi import APIRouter, Depends, HTTPException, status else: try: from fastapi import APIRouter, Depends, HTTPException, status except ImportError: # fastapi is only required for proxy, not for SDK usage pass from litellm.types.management_endpoints.router_settings_endpoints import ( FallbackCreateRequest, FallbackDeleteResponse, FallbackGetResponse, FallbackResponse, ) router = APIRouter() @router.post( "/fallback", tags=["Fallback Management"], dependencies=[Depends(user_api_key_auth)], response_model=FallbackResponse, status_code=status.HTTP_200_OK, ) async def create_fallback( data: FallbackCreateRequest, user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), ): """ Create or update fallbacks for a specific model. This endpoint allows you to configure fallback models separately from the general config. Fallbacks are triggered when a model call fails after retries. **Example Request:** ```json { "model": "gpt-3.5-turbo", "fallback_models": ["gpt-4", "claude-3-haiku"], "fallback_type": "general" } ``` **Fallback Types:** - `general`: Standard fallbacks for any error (default) - `context_window`: Fallbacks specifically for context window exceeded errors - `content_policy`: Fallbacks specifically for content policy violations """ from litellm.proxy.proxy_server import ( llm_router, prisma_client, proxy_config, store_model_in_db, ) try: # Validate that we have a router if llm_router is None: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={"error": "Router not initialized"}, ) # Validate that the model exists in the router model_names = llm_router.model_names if data.model not in model_names: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={ "error": f"Model '{data.model}' not found in router", "available_models": list(model_names), }, ) # Validate that all fallback models exist in the router invalid_fallback_models = [ m for m in data.fallback_models if m not in model_names ] if invalid_fallback_models: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail={ "error": f"Invalid fallback models: {invalid_fallback_models}", "available_models": list(model_names), }, ) # Check if fallback model is the same as the primary model if data.model in data.fallback_models: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail={ "error": f"Model '{data.model}' cannot be its own fallback" }, ) # Check if we need to store in DB if store_model_in_db is not True or prisma_client is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail={ "error": "Database storage not enabled. Set 'STORE_MODEL_IN_DB=True' in your environment to use this feature." }, ) # Load existing config config = await proxy_config.get_config() router_settings = config.get("router_settings", {}) # Get the appropriate fallback list based on type fallback_key = "fallbacks" if data.fallback_type == "context_window": fallback_key = "context_window_fallbacks" elif data.fallback_type == "content_policy": fallback_key = "content_policy_fallbacks" # Get existing fallbacks existing_fallbacks: List[Dict[str, List[str]]] = router_settings.get( fallback_key, [] ) # Update or add the fallback configuration fallback_updated = False for i, fallback_dict in enumerate(existing_fallbacks): if data.model in fallback_dict: # Update existing fallback existing_fallbacks[i] = {data.model: data.fallback_models} fallback_updated = True break if not fallback_updated: # Add new fallback existing_fallbacks.append({data.model: data.fallback_models}) # Update router settings router_settings[fallback_key] = existing_fallbacks # Save to database - convert router_settings to JSON string router_settings_json = json.dumps(router_settings) await prisma_client.db.litellm_config.upsert( where={"param_name": "router_settings"}, data={ "create": { "param_name": "router_settings", "param_value": router_settings_json, }, "update": { "param_value": router_settings_json }, }, ) # Update the in-memory router configuration setattr(llm_router, fallback_key, existing_fallbacks) verbose_proxy_logger.info( f"Fallback configured: {data.model} -> {data.fallback_models} (type: {data.fallback_type})" ) return FallbackResponse( model=data.model, fallback_models=data.fallback_models, fallback_type=data.fallback_type, message=f"Fallback configuration {'updated' if fallback_updated else 'created'} successfully", ) except HTTPException: raise except Exception as e: verbose_proxy_logger.error(f"Error creating fallback: {str(e)}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={"error": f"Failed to create fallback: {str(e)}"}, ) @router.get( "/fallback/{model}", tags=["Fallback Management"], dependencies=[Depends(user_api_key_auth)], response_model=FallbackGetResponse, ) async def get_fallback( model: str, fallback_type: Literal["general", "context_window", "content_policy"] = "general", user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), ): """ Get fallback configuration for a specific model. **Parameters:** - `model`: The model name to get fallbacks for - `fallback_type`: Type of fallback to retrieve (query parameter) **Example:** ``` GET /fallback/gpt-3.5-turbo?fallback_type=general ``` """ from litellm.proxy.proxy_server import llm_router try: if llm_router is None: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={"error": "Router not initialized"}, ) # Get fallbacks using the existing utility function fallback_models = get_all_fallbacks( model=model, llm_router=llm_router, fallback_type=fallback_type ) if not fallback_models: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={ "error": f"No {fallback_type} fallbacks configured for model '{model}'" }, ) return FallbackGetResponse( model=model, fallback_models=fallback_models, fallback_type=fallback_type, ) except HTTPException: raise except Exception as e: verbose_proxy_logger.error(f"Error getting fallback: {str(e)}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={"error": f"Failed to get fallback: {str(e)}"}, ) @router.delete( "/fallback/{model}", tags=["Fallback Management"], dependencies=[Depends(user_api_key_auth)], response_model=FallbackDeleteResponse, ) async def delete_fallback( model: str, fallback_type: Literal["general", "context_window", "content_policy"] = "general", user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), ): """ Delete fallback configuration for a specific model. **Parameters:** - `model`: The model name to delete fallbacks for - `fallback_type`: Type of fallback to delete (query parameter) **Example:** ``` DELETE /fallback/gpt-3.5-turbo?fallback_type=general ``` """ from litellm.proxy.proxy_server import ( llm_router, prisma_client, proxy_config, store_model_in_db, ) try: if llm_router is None: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={"error": "Router not initialized"}, ) if store_model_in_db is not True or prisma_client is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail={ "error": "Database storage not enabled. Set 'STORE_MODEL_IN_DB=True' in your environment to use this feature." }, ) # Load existing config config = await proxy_config.get_config() router_settings = config.get("router_settings", {}) # Get the appropriate fallback list based on type fallback_key = "fallbacks" if fallback_type == "context_window": fallback_key = "context_window_fallbacks" elif fallback_type == "content_policy": fallback_key = "content_policy_fallbacks" # Get existing fallbacks existing_fallbacks: List[Dict[str, List[str]]] = router_settings.get( fallback_key, [] ) # Find and remove the fallback configuration fallback_found = False updated_fallbacks = [] for fallback_dict in existing_fallbacks: if model not in fallback_dict: updated_fallbacks.append(fallback_dict) else: fallback_found = True if not fallback_found: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={ "error": f"No {fallback_type} fallbacks configured for model '{model}'" }, ) # Update router settings router_settings[fallback_key] = updated_fallbacks # Save to database - convert router_settings to JSON string router_settings_json = json.dumps(router_settings) await prisma_client.db.litellm_config.upsert( where={"param_name": "router_settings"}, data={ "create": { "param_name": "router_settings", "param_value": router_settings_json, }, "update": { "param_value": router_settings_json }, }, ) # Update the in-memory router configuration setattr(llm_router, fallback_key, updated_fallbacks) verbose_proxy_logger.info( f"Fallback deleted: {model} (type: {fallback_type})" ) return FallbackDeleteResponse( model=model, fallback_type=fallback_type, message="Fallback configuration deleted successfully", ) except HTTPException: raise except Exception as e: verbose_proxy_logger.error(f"Error deleting fallback: {str(e)}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={"error": f"Failed to delete fallback: {str(e)}"}, )