Spaces:
Running
on
CPU Upgrade
Running
on
CPU Upgrade
File size: 5,200 Bytes
0e34dc4 13efede 0e34dc4 13efede 0e34dc4 13efede |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 |
import os
import logging
import json
from huggingface_hub import model_info, InferenceClient
from dotenv import load_dotenv
# Load environment variables once at the module level
load_dotenv()
# Define preferred providers
PREFERRED_PROVIDERS = ["sambanova", "novita"]
# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
def prioritize_providers(providers):
"""Prioritize preferred providers, keeping all others."""
return sorted(providers, key=lambda provider: provider not in PREFERRED_PROVIDERS)
def test_provider(model_name: str, provider: str, verbose: bool = False) -> bool:
"""
Test if a specific provider is available for a model using InferenceClient
Args:
model_name: Name of the model
provider: Provider to test
verbose: Whether to log detailed information
Returns:
True if the provider is available, False otherwise
"""
try:
# Get HF token from environment
hf_token = os.environ.get("HF_TOKEN")
if not hf_token:
raise ValueError("HF_TOKEN not defined in environment")
if verbose:
logger.info(f"Testing provider {provider} for model {model_name}")
# Initialize the InferenceClient with the specific provider
client = InferenceClient(
model=model_name,
token=hf_token,
provider=provider,
timeout=10 # Increased timeout to allow model loading
)
try:
# Use the chat completions method for testing
response = client.chat_completion(
messages=[{"role": "user", "content": "Hello"}],
max_tokens=5
)
if verbose:
logger.info(f"Provider {provider} is available for {model_name}")
return True
except Exception as e:
if verbose:
error_message = str(e)
logger.error(f"Error with provider {provider}: {error_message}")
# Log specific error types if we can identify them
if "status_code=429" in error_message:
logger.warning(f"Provider {provider} rate limited. You may need to wait or upgrade your plan.")
elif "status_code=401" in error_message:
logger.warning(f"Authentication failed for provider {provider}. Check your token.")
elif "status_code=503" in error_message:
logger.warning(f"Provider {provider} service unavailable. Model may be loading or provider is down.")
elif "timed out" in error_message.lower():
logger.error(f"Timeout error with provider {provider} - request timed out after 10 seconds")
return False
except Exception as e:
if verbose:
logger.error(f"Error in test_provider: {str(e)}")
return False
def get_available_model_provider(model_name, verbose=False):
"""
Get the first available provider for a given model.
Args:
model_name: Name of the model on the Hub
verbose: Whether to log detailed information
Returns:
First available provider or None if none are available
"""
try:
# Get HF token from environment
hf_token = os.environ.get("HF_TOKEN")
if not hf_token:
raise ValueError("HF_TOKEN not defined in environment")
# Get providers for the model and prioritize them
info = model_info(model_name, expand="inferenceProviderMapping")
if not hasattr(info, "inference_provider_mapping"):
if verbose:
logger.info(f"No inference providers found for {model_name}")
return None
providers = list(info.inference_provider_mapping.keys())
if not providers:
if verbose:
logger.info(f"Empty list of providers for {model_name}")
return None
# Prioritize providers
providers = prioritize_providers(providers)
if verbose:
logger.info(f"Available providers for {model_name}: {', '.join(providers)}")
# Test each provider
for provider in providers:
if test_provider(model_name, provider, verbose):
return provider
return None
except Exception as e:
if verbose:
logger.error(f"Error in get_available_model_provider: {str(e)}")
return None
if __name__ == "__main__":
models = [
"Qwen/QwQ-32B",
"Qwen/Qwen2.5-72B-Instruct",
"meta-llama/Llama-3.3-70B-Instruct",
"deepseek-ai/DeepSeek-R1-Distill-Llama-70B",
"mistralai/Mistral-Small-24B-Instruct-2501",
]
providers = []
for model in models:
provider = get_available_model_provider(model, verbose=True)
providers.append(provider)
print(f"Providers {len(providers)}: {providers}")
|