kim/litellm-config
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: merge daily_limit functionality into volc_limit_sync for single-file dynamic rate limiting

Browse Source 
This commit merges the interception and soft-eviction logic from daily_limit.py into volc_limit_sync.py.
It enables a unified solution where:
1. Running the script directly starts the sync daemon.
2. Importing 'proxy_handler_instance' provides the LiteLLM callback handler.
The handler now respects dynamic limits synced to Redis.
This commit is contained in:
kim
2026-01-29 15:11:14 +08:00
parent 284b33c362
commit 3156bbe431
1 changed files with 447 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volc_limit_sync.py
+446 -201
View File
@@ -7,8 +7,48 @@ import hashlib
import datetime
import requests
import redis
import traceback
from urllib.parse import urlparse
# ================= Configuration =================
# ==========================================
# 📦 LiteLLM Imports (Safe Handling)
# ==========================================
try:
import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm.exceptions import ServiceUnavailableError, ContextWindowExceededError
LITELLM_AVAILABLE = True
except ImportError:
LITELLM_AVAILABLE = False
# Define dummy classes to prevent errors if run standalone without litellm
class CustomLogger:
pass
class ServiceUnavailableError(Exception):
pass
class ContextWindowExceededError(Exception):
pass
# ==========================================
# 🛠️ Configuration & Constants
# ==========================================
# 日志级别配置
LOG_LEVEL = os.getenv("DAILY_LIMIT_LOG_LEVEL", "INFO").upper()
def get_log_level_value(level_name):
levels = {"DEBUG": 10, "INFO": 20, "WARNING": 30, "ERROR": 40}
return levels.get(level_name, 20)
CURRENT_LEVEL_VAL = get_log_level_value(LOG_LEVEL)
DEBUG_MODE = CURRENT_LEVEL_VAL <= 10
# Volcengine API Config
ACCESS_KEY = os.getenv(
"VOLC_ACCESS_KEY", "AKLTNGYxZmUyNzNmZjk3NGMyZmEzNjVjZmY4NGM5ZjcxMjU"
)
@@ -18,11 +58,20 @@ SECRET_KEY = os.getenv(
HOST = "open.volcengineapi.com"
REGION = "cn-beijing"
# Daemon Config
BUFFER = 300000 # User requested 30w buffer
SYNC_INTERVAL = 60
CHECK_INTERVAL = 10
THRESHOLD = 50000
# Redis Config
REDIS_HOST = os.getenv(
"REDIS_HOST", "redis"
) # Default to 'redis' (docker service name)
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", None)
# Suffixes to clean from Volcengine response
SUFFIXES_TO_REMOVE = [
"_data_collaboration_resource_pack",
"协作奖励计划资源包",
@@ -31,40 +80,100 @@ SUFFIXES_TO_REMOVE = [
"资源包",
]
# 火山引擎清洗后名称 -> LiteLLM 模型名称的映射表
# 用户 TODO: 请根据您的实际部署情况手动配置此映射
#
# Key 格式 (火山清洗后名称):
# 1. 移除后缀 (如 "_data_collaboration_resource_pack")
# 2. 替换下划线和点为中划线 ("_.", -> "-")
# 3. 转小写
# 示例: "DeepSeek_V3.2_..." -> "deepseek-v3-2"
#
# Value 格式 (LiteLLM 清洗后名称):
# config.yaml 中定义的 model_name (不带 openai/ 等前缀)
# 类型: 字符串 或 字符串列表
# 示例 (一对一): "deepseek-v3-2-251201"
# 示例 (一对多, 共享额度): ["doubao-seed-1-6-250615", "doubao-seed-1-6-251015"]
# Volcengine Name -> LiteLLM Model Name Map
VOLC_MODEL_MAP = {
"doubao-seed-1-6": ["doubao-seed-1-6-250615", "doubao-seed-1-6-251015"], # 豆包 1.6
# "doubao-seed-1-6-thinking": "", # 豆包 1.6 Thinking
"doubao-seed-1-8": "doubao-seed-1-8-251228", # 豆包 1.8
"doubao-seed-code": "doubao-seed-code-preview-251028", # 豆包 Code
"doubao-seed-translation": "doubao-seed-translation-250915", # 豆包 Translation
"deepseek-v3": "deepseek-v3-250324", # DeepSeek V3 (通用)
"deepseek-v3-1": "deepseek-v3-1-terminus", # DeepSeek V3.1
"deepseek-v3-2": "deepseek-v3-2-251201", # DeepSeek V3.2
"deepseek-r1": "deepseek-r1-250528", # DeepSeek R1
"kimi-k2": "kimi-k2-thinking-251104", # Kimi K2
"doubao-seed-1-6": ["doubao-seed-1-6-250615", "doubao-seed-1-6-251015"],
"doubao-seed-1-8": "doubao-seed-1-8-251228",
"doubao-seed-code": "doubao-seed-code-preview-251028",
"doubao-seed-translation": "doubao-seed-translation-250915",
"deepseek-v3": "deepseek-v3-250324",
"deepseek-v3-1": "deepseek-v3-1-terminus",
"deepseek-v3-2": "deepseek-v3-2-251201",
"deepseek-r1": "deepseek-r1-250528",
"kimi-k2": "kimi-k2-thinking-251104",
"glm-4_7": "glm-4-7-251222",
}
# Redis Configuration
REDIS_HOST = os.getenv("REDIS_HOST", "redis")
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", None)
# Static Quotas (Fallback/Strict Limits)
MODEL_QUOTAS = {
"deepseek-v3-2-251201:ba0383d3": 1600000,
"doubao-seed-1-8-251228:ba0383d3": 1800000,
"glm-4-7-251222:ba0383d3": 1800000,
"doubao-seed-1-8-251228:d57edef4": 1500000,
}
# ==========================================
# 🔌 Redis Connection
# ==========================================
try:
redis_client = redis.Redis(
host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD, decode_responses=True
)
# Check connection only if running as script, to avoid noise on import
# redis_client.ping()
except Exception as e:
print(f"❌ Redis connection failed: {str(e)}", flush=True)
redis_client = None
# ==========================================
# 📝 Logging Helpers
# ==========================================
def log(msg, level="INFO"):
msg_val = get_log_level_value(level)
if msg_val >= CURRENT_LEVEL_VAL:
icon = {"DEBUG": "🔍", "INFO": "ℹ️", "WARNING": "⚠️", "ERROR": "❌"}.get(
level, "📝"
)
print(f"{icon} [VolcSync] {msg}", flush=True)
def dprint(msg):
log(msg, level="INFO")
# ==========================================
# 🧮 Utility Functions (Hash, Clean, Sign)
# ==========================================
def generate_api_base_hash(api_base):
"""Generate robust hash from api_base ignoring path variations."""
if not api_base:
return "default"
try:
s = api_base.strip()
parsed = urlparse(s)
if parsed.scheme and parsed.netloc:
clean_base = f"{parsed.scheme}://{parsed.netloc}".lower()
else:
clean_base = s.rstrip("/").lower()
except Exception:
clean_base = api_base.strip().rstrip("/").lower()
h = hashlib.md5(clean_base.encode()).hexdigest()[:8]
return h
def clean_model_name(name):
"""LiteLLM style cleaning: remove provider prefixes."""
if not name:
return ""
return name.replace("openai/", "").replace("azure/", "").replace("anthropic/", "")
def normalize_volc_name(raw_name):
"""Volcengine API response cleaning."""
if not raw_name:
return "unknown"
clean_name = raw_name
for suffix in sorted(SUFFIXES_TO_REMOVE, key=len, reverse=True):
if clean_name.endswith(suffix):
clean_name = clean_name.replace(suffix, "")
break
clean_name = clean_name.replace("_", "-").replace(".", "-")
clean_name = clean_name.lower().strip()
return VOLC_MODEL_MAP.get(clean_name, clean_name)
# ================= Signing Logic =================
def sign(key, msg):
return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()
@@ -78,43 +187,28 @@ def get_signature_key(key, date_stamp, region_name, service_name):
def get_signed_headers(action, version, body_dict, service="billing"):
# 1. Prepare Time
t = datetime.datetime.utcnow()
amz_date = t.strftime("%Y%m%dT%H%M%SZ")
date_stamp = t.strftime("%Y%m%d")
# 2. Process Body
request_body = json.dumps(body_dict)
payload_hash = hashlib.sha256(request_body.encode("utf-8")).hexdigest()
# 3. Canonical Request
method = "POST"
canonical_uri = "/"
canonical_querystring = f"Action={action}&Version={version}"
content_type = "application/json"
# Headers must be sorted
canonical_headers = f"content-type:{content_type}\nhost:{HOST}\nx-content-sha256:{payload_hash}\nx-date:{amz_date}\n"
canonical_headers = f"content-type:application/json\nhost:{HOST}\nx-content-sha256:{payload_hash}\nx-date:{amz_date}\n"
signed_headers = "content-type;host;x-content-sha256;x-date"
canonical_request = f"POST\n/\nAction={action}&Version={version}\n{canonical_headers}\n{signed_headers}\n{payload_hash}"
canonical_request = f"{method}\n{canonical_uri}\n{canonical_querystring}\n{canonical_headers}\n{signed_headers}\n{payload_hash}"
# 4. String to Sign
algorithm = "HMAC-SHA256"
credential_scope = f"{date_stamp}/{REGION}/{service}/request"
string_to_sign = f"{algorithm}\n{amz_date}\n{credential_scope}\n{hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()}"
string_to_sign = f"HMAC-SHA256\n{amz_date}\n{credential_scope}\n{hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()}"
# 5. Calculate Signature
signing_key = get_signature_key(SECRET_KEY, date_stamp, REGION, service)
signature = hmac.new(
signing_key, string_to_sign.encode("utf-8"), hashlib.sha256
).hexdigest()
# 6. Authorization Header
authorization_header = f"{algorithm} Credential={ACCESS_KEY}/{credential_scope}, SignedHeaders={signed_headers}, Signature={signature}"
authorization_header = f"HMAC-SHA256 Credential={ACCESS_KEY}/{credential_scope}, SignedHeaders={signed_headers}, Signature={signature}"
return {
"Content-Type": content_type,
"Content-Type": "application/json",
"Host": HOST,
"X-Date": amz_date,
"X-Content-Sha256": payload_hash,
@@ -122,146 +216,354 @@ def get_signed_headers(action, version, body_dict, service="billing"):
}, request_body
# ================= Helper Functions =================
def normalize_model_name(raw_name):
# ==========================================
# 🎯 Limit Logic (Unified)
# ==========================================
def get_usage_key(clean_model, api_base=None):
now = time.time()
adjusted_time = now - 3600 # UTC+1 split
date_str = time.strftime("%Y-%m-%d", time.gmtime(adjusted_time))
suffix = f":{generate_api_base_hash(api_base)}" if api_base else ""
return f"daily_usage:{date_str}:{clean_model}{suffix}"
def get_static_limit(clean_model, api_base):
"""Check static configuration in MODEL_QUOTAS."""
current_hash = generate_api_base_hash(api_base)
# 1. Exact Match with Hash
specific_key = f"{clean_model}:{current_hash}"
if specific_key in MODEL_QUOTAS:
return MODEL_QUOTAS[specific_key]
# 2. Exact Model Name
if clean_model in MODEL_QUOTAS:
return MODEL_QUOTAS[clean_model]
# 3. Fuzzy Match
for quota_model, limit_val in MODEL_QUOTAS.items():
if ":" not in quota_model and quota_model in clean_model:
return limit_val
return None
def get_effective_limit(clean_model, api_base, current_usage=None):
"""
Align Volcengine response names with LiteLLM upstream names.
1. Remove suffixes.
2. Replace _ with -.
3. Replace . with -. (Crucial: V3.2 -> v3-2)
4. Lowercase.
5. Check VOLC_MODEL_MAP for explicit overrides.
Calculate effective limit combining Static Quotas and Dynamic Redis Quotas.
Dynamic Limit = Current Usage + Redis Remaining Balance
"""
if not raw_name:
return "unknown"
if not redis_client:
return get_static_limit(clean_model, api_base)
clean_name = raw_name
# 1. Get Static Limit
static_limit = get_static_limit(clean_model, api_base)
# Step 1: Remove suffixes
for suffix in sorted(SUFFIXES_TO_REMOVE, key=len, reverse=True):
if clean_name.endswith(suffix):
clean_name = clean_name.replace(suffix, "")
break
# 2. Get Dynamic Remaining Balance
# Key set by daemon loop: limit:dynamic:volcengine:remaining:{clean_model}
dynamic_key = f"limit:dynamic:volcengine:remaining:{clean_model}"
dynamic_remaining = redis_client.get(dynamic_key)
# Step 2 & 3: Replace _ and . with -
clean_name = clean_name.replace("_", "-").replace(".", "-")
effective_limit = None
# Step 4: Lowercase everything
clean_name = clean_name.lower().strip()
# If we have dynamic data
if dynamic_remaining is not None:
try:
# If usage not provided, fetch it
if current_usage is None:
usage_key = get_usage_key(clean_model, api_base)
current_usage = float(redis_client.get(usage_key) or 0)
# Step 5: Check VOLC_MODEL_MAP for explicit overrides
return VOLC_MODEL_MAP.get(clean_name, clean_name)
# Dynamic Limit = The point where usage would consume all remaining balance
# limit = usage + remaining
dynamic_limit_val = float(current_usage) + float(dynamic_remaining)
# If static limit exists, take the stricter one (minimum)
# UNLESS static is much higher? No, min implies stricter.
if static_limit:
effective_limit = min(float(static_limit), dynamic_limit_val)
else:
effective_limit = dynamic_limit_val
except (ValueError, TypeError):
effective_limit = static_limit
else:
effective_limit = static_limit
return effective_limit
# ==========================================
# 👻 Handler Logic (DailyLimitHandler)
# ==========================================
def soft_evict_from_router(target_upstream_model, target_api_base=None):
try:
from litellm.proxy.proxy_server import llm_router
if not llm_router or not llm_router.model_list:
return
evicted = False
target_upstream_clean = clean_model_name(target_upstream_model)
for i, model_entry in enumerate(llm_router.model_list):
params = model_entry.get("litellm_params", {})
upstream_model = params.get("model", "")
current_api_base = params.get("api_base", "")
if target_upstream_clean not in clean_model_name(upstream_model):
continue
if target_api_base and current_api_base != target_api_base:
continue
current_alias = model_entry.get("model_name")
if current_alias and "blocked" not in current_alias:
log(
f"👻 [ROUTER HIDE] Blocking '{current_alias}' for {target_upstream_model}",
level="INFO",
)
model_entry["_daily_limit_backup"] = {
"model_name": current_alias,
"api_base": current_api_base,
}
llm_router.model_list[i]["model_name"] = f"{current_alias}-blocked"
llm_router.model_list[i]["litellm_params"]["api_base"] = (
"http://0.0.0.0"
)
evicted = True
if evicted:
dprint("✅ Node effectively removed from routing pool.")
except Exception as e:
print(f"❌ Router eviction failed: {e}", flush=True)
last_restore_check = 0
RESTORE_CHECK_INTERVAL = 60
def restore_blocked_models():
global last_restore_check
now = time.time()
if now - last_restore_check < RESTORE_CHECK_INTERVAL:
return
last_restore_check = now
try:
from litellm.proxy.proxy_server import llm_router
if not llm_router or not llm_router.model_list:
return
restored_count = 0
for i, model_entry in enumerate(llm_router.model_list):
backup = model_entry.get("_daily_limit_backup")
if not backup:
continue
params = model_entry.get("litellm_params", {})
upstream_model = params.get("model", "")
clean_model = clean_model_name(upstream_model)
original_api_base = backup.get("api_base")
# Check Quota
# Need current usage to calculate dynamic limit correctly?
# get_effective_limit handles fetching usage if None passed.
effective_limit = get_effective_limit(clean_model, original_api_base)
should_restore = False
if not effective_limit:
should_restore = True
else:
usage_key = get_usage_key(clean_model, original_api_base)
current_val = float(redis_client.get(usage_key) or 0)
if current_val < float(effective_limit):
should_restore = True
if should_restore:
dprint(f"♻️ [RESTORE] Restoring {backup['model_name']} (Quota OK)")
llm_router.model_list[i]["model_name"] = backup["model_name"]
llm_router.model_list[i]["litellm_params"]["api_base"] = (
original_api_base
)
del llm_router.model_list[i]["_daily_limit_backup"]
restored_count += 1
if restored_count > 0:
dprint(f"✅ Restored {restored_count} models.")
except Exception as e:
log(f"❌ Restore check failed: {e}", level="ERROR")
def quota_interceptor(kwargs, user_api_key_dict=None):
try:
restore_blocked_models()
except (ServiceUnavailableError, ContextWindowExceededError):
raise
except Exception:
pass
class DailyLimitHandler(CustomLogger):
def __init__(self):
if LITELLM_AVAILABLE:
if quota_interceptor not in litellm.input_callback:
litellm.input_callback.append(quota_interceptor)
dprint("✅ quota_interceptor registered")
if self not in litellm.callbacks:
litellm.callbacks.append(self)
dprint("✅ DailyLimitHandler registered")
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
self._record_usage(kwargs, response_obj)
def log_success_event(self, kwargs, response_obj, start_time, end_time):
self._record_usage(kwargs, response_obj)
def _record_usage(self, kwargs, response_obj):
if not redis_client:
return
try:
litellm_params = kwargs.get("litellm_params") or {}
raw_model = litellm_params.get("model") or kwargs.get("model")
clean_model = clean_model_name(raw_model)
api_base = litellm_params.get("api_base")
usage = None
if hasattr(response_obj, "usage"):
usage = response_obj.usage
elif isinstance(response_obj, dict):
usage = response_obj.get("usage")
if hasattr(usage, "model_dump"):
usage = usage.model_dump()
elif hasattr(usage, "dict"):
usage = usage.dict()
if not usage:
return
total_tokens = usage.get("total_tokens", 0)
if clean_model and total_tokens > 0:
key = get_usage_key(clean_model, api_base)
current_total = redis_client.incrby(key, int(total_tokens))
redis_client.expire(key, 172800)
# Get Limit (Static + Dynamic)
effective_limit = get_effective_limit(
clean_model, api_base, current_usage=current_total
)
display_limit = f"/{int(effective_limit):,}" if effective_limit else ""
key_display = key.split(":", 2)[-1]
log(
f"📊 [Usage] {key_display}: +{total_tokens} (Today: {current_total:,}{display_limit})",
level="INFO",
)
if effective_limit and float(current_total) >= float(effective_limit):
log(
f"🚫 [BLOCK] {clean_model} Limit Hit: {current_total}/{effective_limit}",
level="WARNING",
)
soft_evict_from_router(raw_model, api_base)
except Exception as e:
log(
f"❌ _record_usage failed: {e}\n{traceback.format_exc()}", level="ERROR"
)
# Module Level Instance
proxy_handler_instance = DailyLimitHandler()
# ==========================================
# 🔄 Daemon Logic (Active Monitor Loop)
# ==========================================
def fetch_all_packages():
"""Fetch all package pages from Volcengine API."""
action = "ListResourcePackages"
version = "2022-01-01"
all_packages = []
current_token = ""
print(f"[{datetime.datetime.now()}] Fetching packages from Volcengine...")
while True:
try:
# User doc: NextToken 首次查询传 ""
payload = {
"ResourceType": "Package",
"MaxResults": "20",
"NextToken": current_token,
}
headers, body = get_signed_headers(action, version, payload)
url = f"https://{HOST}/?Action={action}&Version={version}"
response = requests.post(url, data=body, headers=headers, timeout=60)
response.raise_for_status()
data = response.json()
# Debug: print first item of first page to verify fields
# if not current_token and DEBUG_MODE:
# pass
result_block = data.get("Result", {})
items = result_block.get("List", [])
items = data.get("Result", {}).get("List", [])
if items:
all_packages.extend(items)
next_token = result_block.get("NextToken", "")
next_token = data.get("Result", {}).get("NextToken", "")
if not next_token or next_token == current_token:
break
current_token = next_token
time.sleep(0.2) # Rate limit protection
time.sleep(0.2)
except Exception as e:
print(f"Error fetching packages: {e}")
break
return all_packages
def active_monitor_loop():
# 4. In-Line Config Parser (Config Awareness)
# Always run auto-discovery to populate INTERESTED_MODELS (The Whitelist)
INTERESTED_MODELS = set()
def process_packages_breakdown(packages):
model_counts = {}
for pkg in packages:
raw_code = pkg.get("ConfigurationCode", "")
raw_name = pkg.get("ConfigurationName", "Unknown")
target_raw = raw_code if raw_code else raw_name
model_name = normalize_volc_name(target_raw)
try:
available = int(pkg.get("AvailableAmount", 0))
if pkg.get("Unit", "token").lower() == "token":
model_counts[model_name] = model_counts.get(model_name, 0) + available
except ValueError:
continue
return model_counts
def active_monitor_loop():
# Config Discovery
INTERESTED_MODELS = set()
try:
if os.path.exists("config.yaml"):
with open("config.yaml", "r", encoding="utf-8") as f:
content = f.read()
lines = content.split("\n")
current_model_name = None
print("Auto-discovering Volcengine models from config.yaml...")
for line in lines:
sline = line.strip()
if sline.startswith("- model_name:"):
current_model_name = sline.split(":", 1)[1].strip()
# Remove quotes if present
if current_model_name.startswith(
'"'
) or current_model_name.startswith("'"):
current_model_name = current_model_name[1:-1]
current_model_name = sline.split(":", 1)[1].strip().strip("\"'")
if current_model_name:
# Heuristic: Check for Volcengine signals in the block
if "api_base:" in sline and ("HUOSHAN" in sline or "volc" in sline):
INTERESTED_MODELS.add(current_model_name)
# Also check provider prefix in model name
if "model:" in sline and "volcengine/" in sline:
INTERESTED_MODELS.add(current_model_name)
print(
f"Auto-discovered {len(INTERESTED_MODELS)} Active Volcengine models: {INTERESTED_MODELS}"
)
except Exception as e:
print(f"Config Auto-discovery failed: {e}")
# Connect to Redis
try:
r = redis.Redis(
host=REDIS_HOST,
port=REDIS_PORT,
password=REDIS_PASSWORD,
decode_responses=True,
)
r.ping()
print(f"Connected to Redis at {REDIS_HOST}:{REDIS_PORT}")
except Exception as e:
print(f"Failed to connect to Redis: {e}")
if not redis_client:
print("❌ Redis not connected. Exiting monitor loop.")
sys.exit(1)
last_usage_map = {} # Track usage per model
print(f"✅ Redis connected at {REDIS_HOST}:{REDIS_PORT}")
last_usage_map = {}
last_sync_time = 0
# Dynamic interval state (Start with aggressive check to initialize)
current_sync_interval = 0
global_min_remaining = 0
IDLE_THRESHOLD = 300 # 5 minutes of no usage -> Start relaxing
IDLE_THRESHOLD = 300
idle_start_time = 0
print("Starting Active Monitor Loop (Safety-Margin + Idle Awareness)...")
@@ -270,51 +572,42 @@ def active_monitor_loop():
while True:
try:
time.sleep(CHECK_INTERVAL)
now = time.time()
today_str = datetime.datetime.now().strftime("%Y-%m-%d")
# 1. Fetch current global usage
# Idle Check
usage_key_global = f"daily_usage:{today_str}:global_total"
current_usage_global = int(r.get(usage_key_global) or 0)
current_usage_global = int(redis_client.get(usage_key_global) or 0)
last_usage_global = sum(last_usage_map.values()) if last_usage_map else 0
delta = current_usage_global - last_usage_global
time_diff = now - last_sync_time
# 2. Idle Detection Logic
if delta <= 0: # No traffic
if delta <= 0:
if idle_start_time == 0:
idle_start_time = now
else:
idle_start_time = 0 # Traffic detected, reset idle timer
idle_start_time = 0
is_idle = (idle_start_time > 0) and (
(now - idle_start_time) > IDLE_THRESHOLD
)
# 3. Determine Dynamic Interval
# Logic: Base Interval (based on balance) + Idle Multiplier
# Dynamic Interval
if global_min_remaining > 2_000_000:
base_interval = 1800 # 30 min (>200w)
base_interval = 1800
elif global_min_remaining > 1_000_000:
base_interval = 600 # 10 min (100w-200w)
base_interval = 600
elif global_min_remaining > 600_000:
base_interval = 300 # 5 min (60w-100w)
base_interval = 300
elif global_min_remaining > 300_000:
base_interval = 60 # 1 min (30w-60w)
base_interval = 60
else:
base_interval = 10 # Panic (<30w Remaining)
base_interval = 10
# Apply Idle Relaxation
# If system is idle for >5 min, we can relax the check interval.
# Max idle interval extended to 3600s (1 hour) as requested.
if is_idle:
# Exponential backoff or just aggressive multiplier?
# Let's quadruple the base interval, capped at 3600s
current_sync_interval = min(base_interval * 4, 3600)
if current_sync_interval < 300:
current_sync_interval = 300 # Minimum 5 min sleep if idle
current_sync_interval = 300
else:
current_sync_interval = base_interval
@@ -328,68 +621,45 @@ def active_monitor_loop():
trigger_reason = (
"Burst Traffic" if delta > THRESHOLD else "Timer Expired"
)
status_icon = "💤" if is_idle else "⚡"
print(
f"{status_icon} Sync Triggered ({trigger_reason})... Delta: {delta}, Interval: {current_sync_interval}s"
f"{'💤' if is_idle else '⚡'} Sync Triggered ({trigger_reason})... Delta: {delta}, Interval: {current_sync_interval}s"
)
# Fetch and Process
packages = fetch_all_packages()
model_balances = process_packages_breakdown(packages)
# Update Redis and calculate Global Min Remaining (for next interval)
temp_min_remaining = float("inf")
# Logic Fix: Only active targets affect the panic timer.
# VOLC_MODEL_MAP is for Translation Only.
# INTERESTED_MODELS is the Source of Truth for Whitelist.
configured_models_found = False
for model_name, balance in model_balances.items():
remaining = balance - BUFFER
# If VOLC_MODEL_MAP is not empty, we ignore anything NOT in it.
is_interested = True
# 1. Resolve Targets (Translation)
# Translate Volc Name -> LiteLLM Name(s) using MAP if available
# Target Resolution
targets = []
if model_name in VOLC_MODEL_MAP:
val = VOLC_MODEL_MAP[model_name]
if val:
targets = val if isinstance(val, list) else [val]
else:
# Empty string = Use default name (Auto Match)
targets = [model_name]
if not targets:
targets = [model_name] # Empty string case
else:
targets = [model_name]
# 2. Filter Targets (Whitelist)
# Only keep targets that are actually in INTERESTED_MODELS (config.yaml)
# Whitelist Filter
final_targets_to_update = []
is_active_for_panic = False
for target in targets:
# Check against Whitelist
matched = False
if not INTERESTED_MODELS:
# Fallback: Monitor everything if config discovery failed
matched = True
else:
for interested in INTERESTED_MODELS:
# Exact match (Highest Priority)
if target == interested:
matched = True
# Heuristic match (only if target is raw volc name)
# If target was explicitly mapped, it implies exact intent.
# If target is raw, we allow fuzzy.
elif target == model_name and (
target in interested or interested in target
):
matched = True
if matched:
break
@@ -397,25 +667,21 @@ def active_monitor_loop():
final_targets_to_update.append(target)
is_active_for_panic = True
# 3. Safety Calculation
# Only active models affect the system panic level
if is_active_for_panic:
configured_models_found = True
if remaining < temp_min_remaining:
temp_min_remaining = remaining
# Redis Update: We still update Redis for the active targets
# Update Redis
for target in final_targets_to_update:
key = f"limit:dynamic:volcengine:remaining:{target}"
r.setex(key, 7200, int(remaining))
redis_client.setex(key, 7200, int(remaining))
print(f"Updated {target}: Bal={balance:,} -> Rem={remaining:,}")
# Update State
global_min_remaining = temp_min_remaining if model_balances else 0
last_usage_map["global"] = current_usage_global
last_sync_time = now
# Log next scheduled check
if global_min_remaining > 10_000_000:
print(
f"🌊 High Tide Mode: Sleeping for 30m (Min Rem: {int(global_min_remaining):,})"
@@ -427,30 +693,9 @@ def active_monitor_loop():
except Exception as e:
print(f"Error in monitor loop: {e}")
traceback.print_exc()
time.sleep(5)
def process_packages_breakdown(packages):
"""Return dictionary of {model_name: total_available}"""
model_counts = {}
for pkg in packages:
raw_code = pkg.get("ConfigurationCode", "")
raw_name = pkg.get("ConfigurationName", "Unknown")
target_raw = raw_code if raw_code else raw_name
model_name = normalize_model_name(target_raw)
try:
available = int(pkg.get("AvailableAmount", 0))
unit = pkg.get("Unit", "token").lower()
if unit == "token":
model_counts[model_name] = model_counts.get(model_name, 0) + available
except ValueError:
continue
return model_counts
if __name__ == "__main__":
active_monitor_loop()