root
1e63653f52
- Add MANUAL_MONITORED_MODEL_ALIASES entry for deepseek-v4-flash-0731 -> deepseek-v4-flash-ga-260731, since VolcEngine resource pack uses MMDD format while upstream model name uses -ga-YYMMDD (inconsistent naming). - Fix config.yaml auto-discovery to skip lines starting with '#', preventing commented-out api_base lines from leaking into the currently-open entry and misclassifying non-VolcEngine models. This was causing deepseek-v4-flash (via NEWAPI) to be falsely detected as a monitored VolcEngine model.
1404 lines
55 KiB
Python
1404 lines
55 KiB
Python
import os
|
||
import sys
|
||
import time
|
||
import json
|
||
import hmac
|
||
import hashlib
|
||
import datetime
|
||
import requests
|
||
import redis
|
||
import traceback
|
||
import threading
|
||
from urllib.parse import urlparse
|
||
|
||
# ==========================================
|
||
# 📦 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"
|
||
)
|
||
SECRET_KEY = os.getenv(
|
||
"VOLC_SECRET_KEY", "Tm1VNE5XSmlORGs1WXpGbU5EVmtabUUyT1RZeVltTTBNalkxTWpjeE1tTQ=="
|
||
)
|
||
HOST = "open.volcengineapi.com"
|
||
REGION = "cn-beijing"
|
||
|
||
# Daemon Config
|
||
BUFFER = 300000 # User requested 30w buffer
|
||
SYNC_INTERVAL = 60
|
||
CHECK_INTERVAL = 10
|
||
THRESHOLD = 300000
|
||
|
||
# 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 = [
|
||
"_pack_data_collaboration",
|
||
"pack_data_collaboration",
|
||
"_data_collaboration_resource_pack",
|
||
"_data_collaboration",
|
||
"data_collaboration",
|
||
"协作奖励计划资源包",
|
||
"免费在线推理资源包",
|
||
"免费在线推理包",
|
||
"资源包",
|
||
"_free_infer_res_pack",
|
||
"_pack_free_infer",
|
||
"_free_inference_resource_pack",
|
||
"free_inference_resource_pack",
|
||
"_infer_res_pack",
|
||
"-free-infer-res-pack", # Matches: DeepSeek-V3-free-infer-res-pack
|
||
]
|
||
|
||
# Manual package-name aliases for monitored models.
|
||
# 常规版本号模型（如 xxx-260215）会根据 config.yaml 自动推导成 xxx，
|
||
# 这里只保留特殊别名或一对多映射，减少新增/删除监控模型时需要手改的地方。
|
||
MANUAL_MONITORED_MODEL_ALIASES = {
|
||
"doubao-seed-1-6": ["doubao-seed-1-6-250615", "doubao-seed-1-6-251015"],
|
||
"doubao-seed-code": "doubao-seed-code-preview-251028",
|
||
"doubao-seed-translation": "doubao-seed-translation-250915",
|
||
"doubao-seed-2-0-code": "doubao-seed-2-0-code-preview-260215",
|
||
# 火山资源包名用 MMDD（deepseek-v4-flash-0731），上游模型名用 -ga-YYMMDD（-ga-260731），
|
||
# 两者不一致，必须手工映射，否则资源包会被 "Not in monitored models" 跳过。
|
||
"deepseek-v4-flash-0731": "deepseek-v4-flash-ga-260731",
|
||
}
|
||
|
||
# Auto-discovered monitored upstream models and endpoint pairs from config.yaml
|
||
MONITORED_MODEL_PAIRS = set()
|
||
MONITORED_MODEL_ALIASES = {}
|
||
|
||
# ==========================================
|
||
# 🛑 Hard Daily Limits (Local Override / 本地每日硬限额)
|
||
# ==========================================
|
||
# Configured directly here to bypass API latency.
|
||
# These limits are applied LOCALLY and update immediately.
|
||
# 这里的限额是本地强制执行的，不受API延迟影响，用于防止超支。
|
||
#
|
||
# Format: "clean-model-name": limit_in_tokens
|
||
# Or with hash: "clean-model-name:hash": limit_in_tokens
|
||
HARD_DAILY_LIMITS = {
|
||
# Specific Limit for Doubao Code
|
||
"doubao-seed-code-preview-251028": 4_800_000,
|
||
# Specific Limits for Doubao Seed 2.0 models on HUOSHAN_BASE_URL
|
||
"doubao-seed-2-0-pro-260215:ba0383d3": 1_900_000,
|
||
"doubao-seed-2-0-lite-260215:ba0383d3": 1_900_000,
|
||
# Default Limit for all other models (Universal Fallback)
|
||
"__DEFAULT__": 1_800_000,
|
||
}
|
||
|
||
# ==========================================
|
||
# 🔌 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/", "")
|
||
.replace("volcengine/", "")
|
||
)
|
||
|
||
|
||
def strip_model_version_suffix(model_name):
|
||
"""Convert foo-bar-260215 -> foo-bar for common versioned upstream names."""
|
||
if not model_name:
|
||
return ""
|
||
|
||
parts = model_name.rsplit("-", 1)
|
||
if len(parts) == 2 and len(parts[1]) == 6 and parts[1].isdigit():
|
||
return parts[0]
|
||
|
||
return model_name
|
||
|
||
|
||
def build_monitored_model_aliases(monitored_models):
|
||
"""
|
||
Build alias -> monitored upstream model(s) mapping.
|
||
Common versioned names are derived automatically from config.yaml.
|
||
"""
|
||
alias_map = {}
|
||
|
||
def add_alias(alias_name, target_model):
|
||
clean_alias = clean_model_name(alias_name)
|
||
clean_target = clean_model_name(target_model)
|
||
if not clean_alias or not clean_target:
|
||
return
|
||
alias_map.setdefault(clean_alias, [])
|
||
if clean_target not in alias_map[clean_alias]:
|
||
alias_map[clean_alias].append(clean_target)
|
||
|
||
for target_model in sorted(monitored_models):
|
||
add_alias(target_model, target_model)
|
||
|
||
base_alias = strip_model_version_suffix(target_model)
|
||
if base_alias != target_model:
|
||
add_alias(base_alias, target_model)
|
||
|
||
for alias_name, targets in MANUAL_MONITORED_MODEL_ALIASES.items():
|
||
if isinstance(targets, list):
|
||
for target_model in targets:
|
||
add_alias(alias_name, target_model)
|
||
else:
|
||
add_alias(alias_name, targets)
|
||
|
||
return alias_map
|
||
|
||
|
||
def normalize_volc_name(raw_name):
|
||
"""Normalize Volcengine package names before monitored-model resolution."""
|
||
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 clean_name
|
||
|
||
|
||
def sign(key, msg):
|
||
return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()
|
||
|
||
|
||
def get_signature_key(key, date_stamp, region_name, service_name):
|
||
k_date = sign(key.encode("utf-8"), date_stamp)
|
||
k_region = sign(k_date, region_name)
|
||
k_service = sign(k_region, service_name)
|
||
k_signing = sign(k_service, "request")
|
||
return k_signing
|
||
|
||
|
||
def get_signed_headers(action, version, body_dict, service="billing"):
|
||
t = datetime.datetime.utcnow()
|
||
amz_date = t.strftime("%Y%m%dT%H%M%SZ")
|
||
date_stamp = t.strftime("%Y%m%d")
|
||
request_body = json.dumps(body_dict)
|
||
payload_hash = hashlib.sha256(request_body.encode("utf-8")).hexdigest()
|
||
|
||
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}"
|
||
|
||
credential_scope = f"{date_stamp}/{REGION}/{service}/request"
|
||
string_to_sign = f"HMAC-SHA256\n{amz_date}\n{credential_scope}\n{hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()}"
|
||
|
||
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()
|
||
|
||
authorization_header = f"HMAC-SHA256 Credential={ACCESS_KEY}/{credential_scope}, SignedHeaders={signed_headers}, Signature={signature}"
|
||
|
||
return {
|
||
"Content-Type": "application/json",
|
||
"Host": HOST,
|
||
"X-Date": amz_date,
|
||
"X-Content-Sha256": payload_hash,
|
||
"Authorization": authorization_header,
|
||
}, request_body
|
||
|
||
|
||
# ==========================================
|
||
# 🎯 Limit Logic (Unified)
|
||
# ==========================================
|
||
def get_usage_key(clean_model, api_base=None):
|
||
# Enforce Beijing Time (UTC+8) for daily key rotation at 00:00 CST
|
||
# Beijing 00:00 = UTC 16:00 (Previous Day) -> simple offset: UTC_timestamp + 8 hours
|
||
now = time.time()
|
||
beijing_time_ts = now + (8 * 3600)
|
||
date_str = time.strftime("%Y-%m-%d", time.gmtime(beijing_time_ts))
|
||
|
||
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_for_hash(clean_model, current_hash=None):
|
||
"""Check static configuration in HARD_DAILY_LIMITS using an optional endpoint hash."""
|
||
if current_hash:
|
||
specific_key = f"{clean_model}:{current_hash}"
|
||
if specific_key in HARD_DAILY_LIMITS:
|
||
return HARD_DAILY_LIMITS[specific_key]
|
||
|
||
# 2. Exact Model Name
|
||
if clean_model in HARD_DAILY_LIMITS:
|
||
return HARD_DAILY_LIMITS[clean_model]
|
||
|
||
# 3. Fuzzy Match
|
||
for quota_model, limit_val in HARD_DAILY_LIMITS.items():
|
||
if quota_model == "__DEFAULT__":
|
||
continue
|
||
if ":" not in quota_model and quota_model in clean_model:
|
||
return limit_val
|
||
|
||
# 4. Universal Default Fallback
|
||
if "__DEFAULT__" in HARD_DAILY_LIMITS:
|
||
return HARD_DAILY_LIMITS["__DEFAULT__"]
|
||
|
||
return None
|
||
|
||
|
||
def get_static_limit(clean_model, api_base):
|
||
"""Check static configuration in HARD_DAILY_LIMITS."""
|
||
current_hash = generate_api_base_hash(api_base) if api_base else None
|
||
return get_static_limit_for_hash(clean_model, current_hash)
|
||
|
||
|
||
def get_monitored_static_limit(clean_model, model_hashes=None):
|
||
"""
|
||
Resolve the static limit for monitored-model debug output.
|
||
If multiple endpoint hashes exist, show the strictest applicable static cap.
|
||
"""
|
||
specific_limits = []
|
||
|
||
for current_hash in model_hashes or []:
|
||
limit_val = get_static_limit_for_hash(clean_model, current_hash)
|
||
if limit_val is not None:
|
||
specific_limits.append(float(limit_val))
|
||
|
||
if specific_limits:
|
||
return min(specific_limits)
|
||
|
||
return get_static_limit_for_hash(clean_model, None)
|
||
|
||
|
||
def calculate_effective_limit(static_limit, dynamic_limit):
|
||
"""Return the effective cap after combining static and dynamic limits."""
|
||
try:
|
||
if dynamic_limit is None:
|
||
return float(static_limit) if static_limit is not None else None
|
||
|
||
dynamic_limit = float(dynamic_limit)
|
||
if static_limit is None:
|
||
return dynamic_limit
|
||
|
||
return min(float(static_limit), dynamic_limit)
|
||
except (ValueError, TypeError):
|
||
return float(static_limit) if static_limit is not None else None
|
||
|
||
|
||
def format_limit_debug_fields(static_limit=None, stable_limit=None, effective_limit=None):
|
||
"""Render explicit limit labels for debug logs to avoid confusing stable and effective caps."""
|
||
fields = []
|
||
|
||
if static_limit is not None:
|
||
fields.append(f"StaticLimit={int(float(static_limit)):,}")
|
||
if stable_limit is not None:
|
||
fields.append(f"StableLimit={int(float(stable_limit)):,}")
|
||
if effective_limit is not None:
|
||
fields.append(f"EffectiveLimit={int(float(effective_limit)):,}")
|
||
|
||
return " | ".join(fields) if fields else "EffectiveLimit=unknown"
|
||
|
||
|
||
def get_effective_limit(clean_model, api_base, current_usage=None):
|
||
"""
|
||
Calculate effective limit using the Peak-Buffer strategy.
|
||
Target Limit = (Peak Daily Remaining) - Buffer.
|
||
"""
|
||
if not redis_client:
|
||
return get_static_limit(clean_model, api_base)
|
||
|
||
# 1. Get Static Limit (Hard Config)
|
||
static_limit = get_static_limit(clean_model, api_base)
|
||
|
||
# 2. Get Dynamic Stable Limit (Calculated by Daemon)
|
||
# This limit is derived from: Peak_Daily_Balance - Buffer
|
||
# It is stable and immune to API latency spikes.
|
||
stable_limit_key = f"limit:dynamic:volcengine:stable_limit:{clean_model}"
|
||
stable_limit_val = redis_client.get(stable_limit_key)
|
||
|
||
effective_limit = None
|
||
|
||
if stable_limit_val is not None:
|
||
try:
|
||
dynamic_limit = float(stable_limit_val)
|
||
effective_limit = calculate_effective_limit(static_limit, dynamic_limit)
|
||
|
||
except (ValueError, TypeError):
|
||
effective_limit = calculate_effective_limit(static_limit, None)
|
||
else:
|
||
# Fallback: If daemon hasn't calculated stable limit yet, use old logic?
|
||
# Or just use static. Safer to wait for daemon.
|
||
effective_limit = calculate_effective_limit(static_limit, None)
|
||
|
||
# Legacy fallback (Optional: Enable if you want immediate loose limits before first sync)
|
||
# dynamic_remaining = redis_client.get(f"limit:dynamic:volcengine:remaining:{clean_model}")
|
||
# if dynamic_remaining and current_usage:
|
||
# effective_limit = float(current_usage) + float(dynamic_remaining)
|
||
|
||
return effective_limit
|
||
|
||
|
||
# ==========================================
|
||
# 👻 Handler Logic (DailyLimitHandler)
|
||
# ==========================================
|
||
def soft_evict_from_router(target_model, target_api_base):
|
||
"""
|
||
Temporarily disable a model in the router by renaming it.
|
||
"""
|
||
if not LITELLM_AVAILABLE:
|
||
return
|
||
|
||
try:
|
||
from litellm.proxy.proxy_server import llm_router
|
||
|
||
if not llm_router:
|
||
return
|
||
|
||
# Prepare target identifiers
|
||
target_clean = clean_model_name(target_model)
|
||
target_hash = generate_api_base_hash(target_api_base)
|
||
|
||
for i, model_entry in enumerate(llm_router.model_list):
|
||
# Skip if already evicted/backed up
|
||
if "_daily_limit_backup" in model_entry:
|
||
continue
|
||
|
||
params = model_entry.get("litellm_params", {})
|
||
m_model = params.get("model", "")
|
||
m_base = params.get("api_base", "")
|
||
|
||
# Check Match
|
||
m_clean = clean_model_name(m_model)
|
||
m_hash = generate_api_base_hash(m_base)
|
||
|
||
# Match on Name AND Hash (Strict)
|
||
if m_clean == target_clean and m_hash == target_hash:
|
||
dprint(
|
||
f"🚫 Soft-evicting {model_entry.get('model_name')} (Limit Reached)"
|
||
)
|
||
|
||
# Create Backup
|
||
model_entry["_daily_limit_backup"] = {
|
||
"model_name": model_entry.get("model_name"),
|
||
"api_base": m_base,
|
||
}
|
||
|
||
# Disable by renaming (Router won't find it)
|
||
model_entry["model_name"] = (
|
||
f"{model_entry.get('model_name')}-limit-reached"
|
||
)
|
||
|
||
# Optional: break after finding first match?
|
||
# Or continue to evict all instances (e.g. load balancing same model)?
|
||
# Let's evict ALL matches to be safe.
|
||
# return
|
||
|
||
except Exception as e:
|
||
log(f"❌ Soft evict failed: {e}", level="ERROR")
|
||
|
||
|
||
last_restore_check = 0
|
||
|
||
|
||
def restore_blocked_models():
|
||
"""
|
||
Periodically check if blocked models have recovered quota.
|
||
Optimized for Volcengine:
|
||
- Quota usually resets in Beijing Time mornings (uncertain time).
|
||
- Logic:
|
||
- During Beijing Morning (00:00 - 14:00): Check every 30 mins.
|
||
- Other times: Check every 4 hours (low probability of reset).
|
||
"""
|
||
global last_restore_check
|
||
now = time.time()
|
||
|
||
# Calculate Dynamic Interval based on Beijing Time
|
||
# UTC+8 Logic
|
||
utc_now_struct = time.gmtime(now)
|
||
beijing_hour = (utc_now_struct.tm_hour + 8) % 24
|
||
|
||
# Morning Window: 07:00 - 13:00
|
||
if 7 <= beijing_hour < 13:
|
||
interval = 1800 # 30 minutes
|
||
mode_str = "Morning/Window (30m)"
|
||
else:
|
||
interval = 14400 # 4 hours
|
||
mode_str = "Off-Peak (4h)"
|
||
|
||
if now - last_restore_check < interval:
|
||
return
|
||
|
||
last_restore_check = now
|
||
|
||
try:
|
||
from litellm.proxy.proxy_server import llm_router
|
||
# ... (rest of the function) ...
|
||
|
||
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 effective_limit is None:
|
||
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 startup_audit():
|
||
"""
|
||
🚀 Startup Audit:
|
||
System initialization check. Iterate through all models in the router.
|
||
If a model is found to be already over the limit in Redis, kick it out immediately.
|
||
系统启动自检：遍历 Router 中所有模型，如果发现 Redis 中记录已超额，直接踢出。
|
||
"""
|
||
if not redis_client or not LITELLM_AVAILABLE:
|
||
return
|
||
|
||
try:
|
||
from litellm.proxy.proxy_server import llm_router
|
||
|
||
if not llm_router or not llm_router.model_list:
|
||
dprint("⚠️ Startup audit skipped: llm_router not ready.")
|
||
return
|
||
|
||
dprint("🚀 Starting Startup Audit (Pre-emptive Block)...")
|
||
blocked_count = 0
|
||
|
||
# Create a copy to iterate safely
|
||
# We only need to read configs to find targets
|
||
for model_entry in llm_router.model_list:
|
||
# Skip if already blocked (shouldn't happen on fresh start, but good practice)
|
||
if "_daily_limit_backup" in model_entry:
|
||
continue
|
||
|
||
params = model_entry.get("litellm_params", {})
|
||
raw_model = params.get("model", "")
|
||
api_base = params.get("api_base", "")
|
||
|
||
# 1. Clean Name
|
||
clean_model = clean_model_name(raw_model)
|
||
if not clean_model:
|
||
continue
|
||
|
||
# 2. Check Limit & Usage
|
||
# We use effective limit logic which checks both static and dynamic
|
||
effective_limit = get_effective_limit(clean_model, api_base)
|
||
|
||
if effective_limit is None:
|
||
continue
|
||
|
||
usage_key = get_usage_key(clean_model, api_base)
|
||
current_val = redis_client.get(usage_key)
|
||
|
||
if current_val and float(current_val) >= float(effective_limit):
|
||
dprint(
|
||
f"🚫 [STARTUP BLOCK] Found {raw_model} over limit ({int(float(current_val))}/{int(effective_limit)}). Evicting now."
|
||
)
|
||
soft_evict_from_router(raw_model, api_base)
|
||
blocked_count += 1
|
||
|
||
if blocked_count > 0:
|
||
dprint(f"✅ Startup Audit Complete: Blocked {blocked_count} models.")
|
||
else:
|
||
dprint("✅ Startup Audit Complete: All models healthy.")
|
||
|
||
except ImportError:
|
||
dprint("⚠️ Startup audit skipped: litellm not found.")
|
||
except Exception as e:
|
||
log(f"❌ Startup audit failed: {e}", level="ERROR")
|
||
|
||
|
||
def quota_interceptor(kwargs, user_api_key_dict=None):
|
||
try:
|
||
# 1. 尝试恢复（定期）
|
||
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")
|
||
|
||
# FILTER: Only track models explicitly discovered as VolcEngine in config.yaml
|
||
# This prevents tracking NewAPI aliases that point to other providers
|
||
current_hash = generate_api_base_hash(api_base)
|
||
if (
|
||
MONITORED_MODEL_PAIRS
|
||
and (clean_model, current_hash) not in MONITORED_MODEL_PAIRS
|
||
):
|
||
return
|
||
|
||
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 ""
|
||
|
||
# Get Dynamic Remaining for Log
|
||
dynamic_key = f"limit:dynamic:volcengine:remaining:{clean_model}"
|
||
rem_val = redis_client.get(dynamic_key)
|
||
rem_display = ""
|
||
if rem_val is not None:
|
||
try:
|
||
# Display Real Balance (Remaining + Buffer)
|
||
rem_int = int(float(rem_val))
|
||
bal_est = rem_int + BUFFER
|
||
rem_display = f" | Rem: {bal_est:,}"
|
||
except Exception as e:
|
||
print(
|
||
f"DEBUG: Error calculating balance for {clean_model}: {e}"
|
||
)
|
||
pass
|
||
else:
|
||
# Debug why key is missing
|
||
if DEBUG_MODE:
|
||
print(f"DEBUG: Dynamic key not found: {dynamic_key}")
|
||
|
||
key_display = key.split(":", 2)[-1]
|
||
# Display Limit clearly
|
||
limit_str = (
|
||
f" | Limit: {int(effective_limit):,}" if effective_limit is not None else ""
|
||
)
|
||
log(
|
||
f"📊 [Usage] {key_display}: +{total_tokens} (Today: {current_total:,}{rem_display}{limit_str})",
|
||
level="INFO",
|
||
)
|
||
|
||
if effective_limit is not None 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():
|
||
action = "ListResourcePackages"
|
||
version = "2022-01-01"
|
||
all_packages = []
|
||
current_token = ""
|
||
seen_tokens = set()
|
||
|
||
while True:
|
||
try:
|
||
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()
|
||
items = data.get("Result", {}).get("List", [])
|
||
if items:
|
||
all_packages.extend(items)
|
||
next_token = data.get("Result", {}).get("NextToken", "")
|
||
|
||
# Infinite Loop Protection
|
||
if not next_token:
|
||
break
|
||
if next_token == current_token:
|
||
break
|
||
if next_token in seen_tokens:
|
||
if DEBUG_MODE:
|
||
print(
|
||
f"DEBUG: Detected pagination cycle with token {next_token}. Stopping."
|
||
)
|
||
break
|
||
|
||
seen_tokens.add(next_token)
|
||
current_token = next_token
|
||
time.sleep(0.2)
|
||
except Exception as e:
|
||
print(f"Error fetching packages: {e}")
|
||
break
|
||
return all_packages
|
||
|
||
|
||
def process_packages_breakdown(packages):
|
||
model_counts = {}
|
||
seen_package_hashes = set()
|
||
|
||
print(f"\n--- Raw Package Processing ---", flush=True)
|
||
|
||
for pkg in packages:
|
||
try:
|
||
# 0. Deduplication (Exact Match)
|
||
# VolcEngine pagination bugs might return duplicate items or pages.
|
||
# We assume identical JSON objects are duplicates.
|
||
pkg_hash = hashlib.md5(json.dumps(pkg, sort_keys=True).encode()).hexdigest()
|
||
if pkg_hash in seen_package_hashes:
|
||
if DEBUG_MODE:
|
||
print(f"DEBUG: Skipping duplicate package (Hash: {pkg_hash})")
|
||
continue
|
||
seen_package_hashes.add(pkg_hash)
|
||
|
||
raw_code = pkg.get("ConfigurationCode", "")
|
||
raw_name = pkg.get("ConfigurationName", "Unknown")
|
||
target_raw = raw_code if raw_code else raw_name
|
||
|
||
# 1. Normalize name (strips suffixes, handles case/hyphens)
|
||
norm_result = normalize_volc_name(target_raw)
|
||
|
||
status = pkg.get("Status", "")
|
||
if status == "Effective":
|
||
available = pkg.get("AvailableAmount", 0)
|
||
else:
|
||
available = 0
|
||
|
||
unit = pkg.get("Unit", "token")
|
||
|
||
# 2. Resolve to monitored upstream models.
|
||
# 普通版本号模型会自动从 config.yaml 推导；特殊别名走手工覆盖。
|
||
target_models = MONITORED_MODEL_ALIASES.get(norm_result, [norm_result])
|
||
|
||
# Debug log
|
||
if (
|
||
DEBUG_MODE
|
||
or "doubao" in str(target_models)
|
||
or "glm" in str(target_models)
|
||
):
|
||
print(
|
||
f"DEBUG: Pkg Raw='{target_raw}' -> Norm='{norm_result}' -> Targets={target_models}, Avail={int(float(available)):,}",
|
||
flush=True,
|
||
)
|
||
|
||
if unit.lower() == "token":
|
||
for model_name in target_models:
|
||
if model_name == "unknown":
|
||
continue
|
||
# Aggregate balances (Handle float string '123.00' -> float -> int)
|
||
try:
|
||
amount = int(float(available))
|
||
except ValueError:
|
||
print(
|
||
f"Error converting available amount '{available}' to int for {model_name}"
|
||
)
|
||
amount = 0
|
||
|
||
model_counts[model_name] = model_counts.get(model_name, 0) + amount
|
||
|
||
except Exception as e:
|
||
print(f"Error processing package: {e}")
|
||
continue
|
||
|
||
return model_counts
|
||
|
||
|
||
def active_monitor_loop():
|
||
# Config Discovery
|
||
global MONITORED_MODEL_PAIRS, MONITORED_MODEL_ALIASES
|
||
MONITORED_MODEL_PAIRS.clear()
|
||
MONITORED_MODEL_ALIASES.clear()
|
||
MONITORED_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")
|
||
|
||
print(
|
||
"Auto-discovering monitored Volcengine models from config.yaml / 正在从 config.yaml 自动发现纳入监控的火山模型...",
|
||
flush=True,
|
||
)
|
||
|
||
# Robust Parsing Logic
|
||
current_entry = {
|
||
"is_volc": False,
|
||
"upstream_model": None,
|
||
"api_base_val": None,
|
||
}
|
||
|
||
for line in lines:
|
||
sline = line.strip()
|
||
|
||
# Skip commented lines (YAML comments). Otherwise commented-out
|
||
# api_base lines (e.g. a disabled model block) leak into the
|
||
# currently-open entry and can misclassify a NewAPI entry as
|
||
# VolcEngine.
|
||
if sline.startswith("#"):
|
||
continue
|
||
|
||
# Start of a new model definition
|
||
if sline.startswith("- model_name:"):
|
||
# Commit previous entry if valid
|
||
if (
|
||
current_entry.get("is_volc")
|
||
and current_entry.get("upstream_model")
|
||
and current_entry.get("api_base_val")
|
||
):
|
||
MONITORED_MODELS.add(current_entry["upstream_model"])
|
||
# Compute Hash
|
||
h = generate_api_base_hash(current_entry["api_base_val"])
|
||
MONITORED_MODEL_PAIRS.add((current_entry["upstream_model"], h))
|
||
print(
|
||
f"✅ [Config] Added monitored pair: {current_entry['upstream_model']} | Hash: {h} | Base: {str(current_entry['api_base_val'])[:30]}...",
|
||
flush=True,
|
||
)
|
||
|
||
# Reset for new entry
|
||
current_entry = {
|
||
"is_volc": False,
|
||
"upstream_model": None,
|
||
"api_base_val": None,
|
||
}
|
||
continue
|
||
|
||
# Capture API Base & Check for Volcengine
|
||
if "api_base:" in sline:
|
||
try:
|
||
raw_val = sline.split(":", 1)[1].strip()
|
||
# Resolve Env Var
|
||
if raw_val.startswith("os.environ/"):
|
||
env_key = raw_val.replace("os.environ/", "")
|
||
resolved_val = os.getenv(env_key, "")
|
||
if not resolved_val:
|
||
print(
|
||
f"⚠️ [Config] Warning: Env var {env_key} resolved to EMPTY!",
|
||
flush=True,
|
||
)
|
||
else:
|
||
resolved_val = raw_val
|
||
|
||
current_entry["api_base_val"] = resolved_val
|
||
|
||
# Identify as Volcengine if using HUOSHAN env or specific host
|
||
if (
|
||
"HUOSHAN_BASE_URL" in sline
|
||
or "volcengineapi" in resolved_val
|
||
):
|
||
current_entry["is_volc"] = True
|
||
except Exception:
|
||
pass
|
||
|
||
# Capture Upstream Model Name
|
||
# Matches: "model: openai/deepseek..."
|
||
if sline.startswith("model:"):
|
||
try:
|
||
raw_val = sline.split(":", 1)[1].strip()
|
||
# Clean prefixes
|
||
clean = (
|
||
raw_val.replace("openai/", "")
|
||
.replace("azure/", "")
|
||
.replace("anthropic/", "")
|
||
.replace("volcengine/", "")
|
||
)
|
||
current_entry["upstream_model"] = clean
|
||
except IndexError:
|
||
pass
|
||
|
||
# Commit the last entry
|
||
if (
|
||
current_entry.get("is_volc")
|
||
and current_entry.get("upstream_model")
|
||
and current_entry.get("api_base_val")
|
||
):
|
||
MONITORED_MODELS.add(current_entry["upstream_model"])
|
||
h = generate_api_base_hash(current_entry["api_base_val"])
|
||
MONITORED_MODEL_PAIRS.add((current_entry["upstream_model"], h))
|
||
if DEBUG_MODE:
|
||
print(
|
||
f"DEBUG: Added monitored pair: {current_entry['upstream_model']} @ {h}"
|
||
)
|
||
|
||
MONITORED_MODEL_ALIASES = build_monitored_model_aliases(MONITORED_MODELS)
|
||
print(
|
||
f"Auto-discovered / 已自动发现 {len(MONITORED_MODELS)} 个纳入监控的火山上游模型: {MONITORED_MODELS}",
|
||
flush=True,
|
||
)
|
||
print(
|
||
f"Monitored model pair hashes / 监控模型端点哈希: {[p[1] for p in MONITORED_MODEL_PAIRS]}",
|
||
flush=True,
|
||
)
|
||
if DEBUG_MODE:
|
||
print(
|
||
f"DEBUG: Monitored model aliases / 监控模型别名: {MONITORED_MODEL_ALIASES}",
|
||
flush=True,
|
||
)
|
||
|
||
except Exception as e:
|
||
print(f"Config Auto-discovery failed / 配置自动发现失败: {e}", flush=True)
|
||
|
||
if not redis_client:
|
||
print(
|
||
"❌ Redis not connected. Exiting monitor loop / Redis 未连接，退出监控循环.",
|
||
flush=True,
|
||
)
|
||
sys.exit(1)
|
||
|
||
print(
|
||
f"✅ Redis connected at / Redis 已连接于 {REDIS_HOST}:{REDIS_PORT}", flush=True
|
||
)
|
||
|
||
# ==========================================
|
||
# 🚀 Run Startup Audit Immediately
|
||
# ==========================================
|
||
# Give LiteLLM a moment to initialize router if running concurrently
|
||
time.sleep(2)
|
||
startup_audit()
|
||
|
||
last_usage_map = {}
|
||
|
||
last_sync_time = 0
|
||
current_sync_interval = 0
|
||
global_min_remaining = 0
|
||
IDLE_THRESHOLD = 300
|
||
idle_start_time = 0
|
||
|
||
print(
|
||
"Starting Active Monitor Loop (Safety-Margin + Idle Awareness) / 正在启动主动监控循环 (安全余量 + 空闲感知)...",
|
||
flush=True,
|
||
)
|
||
print(f"Buffer / 缓冲: {BUFFER}, Threshold / 阈值: {THRESHOLD}", flush=True)
|
||
|
||
while True:
|
||
try:
|
||
time.sleep(CHECK_INTERVAL)
|
||
now = time.time()
|
||
|
||
# Enforce Beijing Time for daily operations
|
||
beijing_time_ts = now + (8 * 3600)
|
||
today_str = time.strftime("%Y-%m-%d", time.gmtime(beijing_time_ts))
|
||
|
||
# Idle Check - Only exit idle if a "low quota" model has usage
|
||
LOW_QUOTA_THRESHOLD = 800000 # 判定为“低额度”的阈值
|
||
model_usages = {}
|
||
any_low_quota_usage = False
|
||
|
||
for m in MONITORED_MODELS:
|
||
m_usage = 0
|
||
target_hashes = [p[1] for p in MONITORED_MODEL_PAIRS if p[0] == m]
|
||
for h in target_hashes:
|
||
val = redis_client.get(f"daily_usage:{today_str}:{m}:{h}")
|
||
if val: m_usage += float(val)
|
||
val_no_hash = redis_client.get(f"daily_usage:{today_str}:{m}")
|
||
if val_no_hash: m_usage += float(val_no_hash)
|
||
model_usages[m] = m_usage
|
||
|
||
# Calculate delta for this specific model
|
||
m_delta = m_usage - last_usage_map.get(m, 0.0)
|
||
|
||
# Check if this model is considered "low quota"
|
||
rem_key = f"limit:dynamic:volcengine:remaining:{m}"
|
||
rem_val = redis_client.get(rem_key)
|
||
# 未知额度视为充足，避免首次启动时所有模型被误判为"低额度"
|
||
remaining = float(rem_val) if rem_val is not None else float(LOW_QUOTA_THRESHOLD)
|
||
|
||
if m_delta > 0 and remaining < LOW_QUOTA_THRESHOLD:
|
||
any_low_quota_usage = True
|
||
|
||
time_diff = now - last_sync_time
|
||
|
||
if not any_low_quota_usage:
|
||
if idle_start_time == 0:
|
||
idle_start_time = now
|
||
else:
|
||
idle_start_time = 0
|
||
|
||
is_idle = (idle_start_time > 0) and (
|
||
(now - idle_start_time) > IDLE_THRESHOLD
|
||
)
|
||
|
||
# Dynamic Interval
|
||
# Logic: Adjust intervals based on distance from BUFFER
|
||
# User specified thresholds (Distance < X):
|
||
# < 10w -> 2min (120s)
|
||
# < 30w -> 5min (300s)
|
||
# < 50w -> 10min (600s)
|
||
# < 80w -> 30min (1800s)
|
||
# < 100w -> 60min (3600s)
|
||
# >= 100w -> 60min (3600s)
|
||
|
||
# Determine Base Sync Interval based on global risk (Optimized for Hourly Latency)
|
||
# Since API updates are hourly, frequent polling provides no new info and wastes resources.
|
||
# We trust Redis Usage for real-time gating.
|
||
# < 100k -> 5 min (Emergency check for the update moment)
|
||
# < 300k -> 10 min
|
||
# < 500k -> 20 min
|
||
# < 800k -> 45 min
|
||
# >= 800k -> 60 min
|
||
if global_min_remaining < 100_000:
|
||
base_interval = 300 # 5 minutes
|
||
elif global_min_remaining < 300_000:
|
||
base_interval = 600 # 10 minutes
|
||
elif global_min_remaining < 500_000:
|
||
base_interval = 1200 # 20 minutes
|
||
elif global_min_remaining < 800_000:
|
||
base_interval = 2700 # 45 minutes
|
||
else:
|
||
base_interval = 3600 # 60 minutes
|
||
|
||
if is_idle:
|
||
current_sync_interval = min(base_interval * 4, 3600)
|
||
if current_sync_interval < 300:
|
||
current_sync_interval = 300
|
||
else:
|
||
current_sync_interval = base_interval
|
||
|
||
# =====================================================
|
||
# 🚀 Per-Model Burst Detection (精细化突发流量检测)
|
||
# =====================================================
|
||
burst_triggered = False
|
||
burst_reason = ""
|
||
|
||
for model in MONITORED_MODELS:
|
||
# 1. Get Current Usage (calculated above)
|
||
curr_usage = model_usages.get(model, 0.0)
|
||
|
||
# 2. Get Last Synced Usage (Snapshot)
|
||
last_usage = last_usage_map.get(model, 0.0)
|
||
|
||
# 3. Calculate Delta since last sync
|
||
model_delta = curr_usage - last_usage
|
||
|
||
# 4. Get Current Balance for Threshold Calculation
|
||
rem_key = f"limit:dynamic:volcengine:remaining:{model}"
|
||
rem_val = redis_client.get(rem_key)
|
||
current_balance = float(rem_val) if rem_val else 0.0
|
||
|
||
# 5. Determine Dynamic Threshold based on Balance
|
||
# Rich models get loose limits, Poor models get strict limits.
|
||
if current_balance > 1000000:
|
||
model_threshold = 500000
|
||
elif current_balance > 500000:
|
||
model_threshold = 100000
|
||
elif current_balance > 100000:
|
||
model_threshold = 50000
|
||
else:
|
||
model_threshold = 10000 # Strict for poor models
|
||
|
||
# 6. Check Trigger
|
||
if model_delta > model_threshold:
|
||
burst_triggered = True
|
||
burst_reason = f"{model} (+{model_delta:,} > {model_threshold:,})"
|
||
break # One burst is enough to trigger sync
|
||
|
||
should_sync = (
|
||
burst_triggered
|
||
or (time_diff > current_sync_interval)
|
||
or (last_sync_time == 0)
|
||
)
|
||
|
||
if should_sync:
|
||
trigger_reason = (
|
||
f"Burst: {burst_reason}" if burst_triggered else "Timer Expired"
|
||
)
|
||
# For Timer Expired, we don't have a specific delta to show, or we can calculate global delta if needed.
|
||
# But for now, let's just show the reason.
|
||
print(
|
||
f"{'💤' if is_idle else '⚡'} Sync Triggered ({trigger_reason}) / 同步触发... Interval: {current_sync_interval}s",
|
||
flush=True,
|
||
)
|
||
|
||
packages = fetch_all_packages()
|
||
model_balances = process_packages_breakdown(packages)
|
||
|
||
print(
|
||
f"DEBUG: Processed models / 已处理模型: {list(model_balances.keys())}",
|
||
flush=True,
|
||
)
|
||
|
||
temp_min_remaining = float("inf")
|
||
configured_models_found = False
|
||
|
||
for model_name, balance in model_balances.items():
|
||
remaining = balance - BUFFER
|
||
print(
|
||
f"DEBUG: Checking {model_name} (Bal: {int(balance):,}, Rem: {int(remaining):,}) (余额/剩余)",
|
||
flush=True,
|
||
)
|
||
|
||
# Target Resolution (Already handled in process_packages_breakdown mostly)
|
||
# But we'll treat model_name as the target since process_packages_breakdown now handles the list logic
|
||
targets = [model_name]
|
||
|
||
for target in targets:
|
||
# STRICT FILTER: Only sync models explicitly configured with VolcEngine Base URL
|
||
# User Requirement: Do NOT sync or limit models that share names but use other providers (e.g. NewAPI)
|
||
if target not in MONITORED_MODELS:
|
||
if DEBUG_MODE:
|
||
print(
|
||
f"DEBUG: Skipping {target} (Not in monitored models / 未纳入监控模型)",
|
||
flush=True,
|
||
)
|
||
continue
|
||
|
||
# Fetch Current Usage from Redis (to calculate Total Pool)
|
||
# We need to find the correct hash for this model from MONITORED_MODEL_PAIRS
|
||
current_usage = 0
|
||
target_hashes = [
|
||
p[1] for p in MONITORED_MODEL_PAIRS if p[0] == target
|
||
]
|
||
|
||
# Sum usage across all configured endpoints for this model (usually just one)
|
||
for h in target_hashes:
|
||
# Replicate get_usage_key logic (Beijing Time UTC+8)
|
||
now_ts = time.time()
|
||
beijing_time_ts = now_ts + (8 * 3600)
|
||
usage_date = time.strftime(
|
||
"%Y-%m-%d", time.gmtime(beijing_time_ts)
|
||
)
|
||
u_key = f"daily_usage:{usage_date}:{target}:{h}"
|
||
val = redis_client.get(u_key)
|
||
if val:
|
||
current_usage += float(val)
|
||
|
||
# Use normalized key for storage
|
||
# Format: limit:dynamic:volcengine:remaining:{model_name}
|
||
dynamic_key = f"limit:dynamic:volcengine:remaining:{target}"
|
||
|
||
# ==========================================
|
||
# 🏔️ Smart Peak Tracking (Auto-Correction)
|
||
# ==========================================
|
||
# Strategy:
|
||
# 1. Capture Jumps (Recovery): If Balance jumps UP, trust it immediately. (Fixes latency)
|
||
# 2. Capture Drops (Expiry): If Balance drops HUGE (>20%), trust it. (Fixes expiry)
|
||
# - Why trust drops? Because if it's a false drop (glitch), Step 1 will fix it
|
||
# automatically on the next sync when it "Jumps" back up.
|
||
|
||
total_pool = current_usage + balance
|
||
peak_key = f"volc:peak_pool:{today_str}:{target}"
|
||
|
||
# Short-term memory (Last seen value)
|
||
last_pool_key = f"volc:last_seen_pool:{target}"
|
||
last_pool_val = redis_client.get(last_pool_key)
|
||
last_pool = float(last_pool_val) if last_pool_val else 0.0
|
||
|
||
# Long-term memory (Day's High Water Mark)
|
||
old_peak_val = redis_client.get(peak_key)
|
||
historical_peak = float(old_peak_val) if old_peak_val else 0.0
|
||
|
||
current_peak = historical_peak
|
||
|
||
# 1. JUMP DETECTION (Recharge / Latency Resolved)
|
||
# If pool jumps significantly (>5%) from last seen, it's a fresh, true value.
|
||
if total_pool > (last_pool * 1.05) and last_pool > 0:
|
||
if DEBUG_MODE:
|
||
print(
|
||
f"DEBUG: 📈 Balance Jump ({int(last_pool):,} -> {int(total_pool):,}) - Locking New Peak.",
|
||
flush=True,
|
||
)
|
||
current_peak = total_pool
|
||
redis_client.setex(peak_key, 172800, int(current_peak))
|
||
|
||
# 2. DROP DETECTION (Resource Expiry)
|
||
# If pool drops (even small amounts), we accept it.
|
||
# Why? Because keeping a phantom high peak is dangerous.
|
||
# If it was a glitch, Rule #1 will fix it next time it bounces back.
|
||
elif total_pool < historical_peak and historical_peak > 0:
|
||
drop_ratio = (
|
||
historical_peak - total_pool
|
||
) / historical_peak
|
||
if DEBUG_MODE:
|
||
print(
|
||
f"DEBUG: 📉 Pool Drop Detected ({int(historical_peak):,} -> {int(total_pool):,}) - Ratio: {drop_ratio:.1%} - Adjusting Limit Down.",
|
||
flush=True,
|
||
)
|
||
current_peak = total_pool
|
||
redis_client.setex(peak_key, 172800, int(current_peak))
|
||
|
||
# 3. INITIALIZATION / DRIFT
|
||
elif historical_peak == 0 or total_pool > historical_peak:
|
||
current_peak = total_pool
|
||
redis_client.setex(peak_key, 172800, int(current_peak))
|
||
|
||
# Always update "Last Seen" for next comparison
|
||
redis_client.setex(last_pool_key, 86400, int(total_pool))
|
||
|
||
# Calculate Stable Limit
|
||
safety_margin = max(current_peak * 0.15, 250000)
|
||
stable_limit = int(max(0, current_peak - safety_margin))
|
||
|
||
stable_limit_key = (
|
||
f"limit:dynamic:volcengine:stable_limit:{target}"
|
||
)
|
||
|
||
# Store stable limit
|
||
redis_client.setex(stable_limit_key, 86400, int(stable_limit))
|
||
|
||
if DEBUG_MODE:
|
||
static_limit = get_monitored_static_limit(
|
||
target, target_hashes
|
||
)
|
||
effective_limit = calculate_effective_limit(
|
||
static_limit, stable_limit
|
||
)
|
||
print(
|
||
f"DEBUG: {target} PeakPool={int(current_peak):,} | Usage={int(current_usage):,} | {format_limit_debug_fields(static_limit=static_limit, stable_limit=stable_limit, effective_limit=effective_limit)}",
|
||
flush=True,
|
||
)
|
||
|
||
if DEBUG_MODE:
|
||
print(
|
||
f"DEBUG: Checking {target} (Configured / 已配置) (Bal: {int(balance):,}, Rem: {int(remaining):,})",
|
||
flush=True,
|
||
)
|
||
|
||
# Set with expiry (24h)
|
||
redis_client.setex(dynamic_key, 86400, int(remaining))
|
||
if DEBUG_MODE:
|
||
print(
|
||
f"DEBUG: Set Redis {dynamic_key} = {int(remaining):,}",
|
||
flush=True,
|
||
)
|
||
|
||
# Track min remaining for frequency control
|
||
configured_models_found = True
|
||
|
||
# Only track POSITIVE remaining for frequency control
|
||
# If a model is 0 (dead), we shouldn't panic-poll the healthy ones.
|
||
# We rely on the Restore Loop to check dead models.
|
||
if remaining > 0 and remaining < temp_min_remaining:
|
||
temp_min_remaining = remaining
|
||
|
||
# If all models are dead (temp_min_remaining is still inf), set to 0 for logic but handle gracefully
|
||
# If we found models but all were <= 0, then temp_min_remaining is inf.
|
||
|
||
# Logic:
|
||
# 1. No models found at all -> Default to Safe (300k)
|
||
# 2. Models found, but all 0 -> "All Dead Mode"
|
||
# 3. Models found, some > 0 -> Use min of >0 models
|
||
|
||
if not configured_models_found:
|
||
global_min_remaining = 300000 # Default
|
||
elif temp_min_remaining == float("inf"):
|
||
# All active models are <= 0
|
||
# Check time of day to decide if we should panic
|
||
utc_now_struct = time.gmtime(now)
|
||
beijing_hour = (utc_now_struct.tm_hour + 8) % 24
|
||
if 7 <= beijing_hour < 13:
|
||
# Morning Reset Window: Check relatively frequently
|
||
global_min_remaining = (
|
||
200000 # Pretend we have some quota to get ~5min interval
|
||
)
|
||
else:
|
||
# Afternoon/Night: Unlikely to reset
|
||
global_min_remaining = (
|
||
1000000 # Pretend high quota to sleep 60m
|
||
)
|
||
else:
|
||
global_min_remaining = temp_min_remaining
|
||
|
||
# Update snapshots for next iteration delta calculation
|
||
for m, u_val in model_usages.items():
|
||
last_usage_map[m] = u_val
|
||
last_sync_time = now
|
||
|
||
if global_min_remaining > 800000:
|
||
print(
|
||
f"🌊 High Tide Mode: Sleeping for 60m (Effective Min Rem: {int(global_min_remaining):,}) / 高水位模式：休眠 60 分钟",
|
||
flush=True,
|
||
)
|
||
elif global_min_remaining > 500000:
|
||
print(
|
||
f"🌊 Normal Mode: Sleeping for 30m (Min Rem: {int(global_min_remaining):,}) / 正常模式：休眠 30 分钟",
|
||
flush=True,
|
||
)
|
||
elif global_min_remaining > 300000:
|
||
print(
|
||
f"🚤 Speed Mode: Sleeping for 10m (Min Rem: {int(global_min_remaining):,}) / 极速模式：休眠 10 分钟",
|
||
flush=True,
|
||
)
|
||
elif global_min_remaining > 100000:
|
||
print(
|
||
f"⛵ Safe Sailing Mode: Sleeping for 5m (Min Rem: {int(global_min_remaining):,}) / 安全航行模式：休眠 5 分钟",
|
||
flush=True,
|
||
)
|
||
else:
|
||
print(
|
||
f"🆘 Emergency Mode: Sleeping for 2m (Min Rem: {int(global_min_remaining):,}) / 紧急模式：休眠 2 分钟",
|
||
flush=True,
|
||
)
|
||
|
||
except Exception as e:
|
||
print(f"Error in monitor loop / 监控循环出错: {e}", flush=True)
|
||
traceback.print_exc()
|
||
time.sleep(5)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
active_monitor_loop()
|
||
else:
|
||
# Auto-start when loaded as a module/callback
|
||
print(
|
||
"🚀 Initializing VolcSync Background Monitor... / 正在初始化火山引擎同步后台监控...",
|
||
flush=True,
|
||
)
|
||
t = threading.Thread(target=active_monitor_loop, daemon=True)
|
||
t.start()
|