fix: 消耗任务四类遗漏修复 + 清理死变量
volc_consume.py: - payload 模型参数路由:doubao 系 max_completion_tokens+reasoning_effort minimal, 其他模型 max_tokens(旧 MONITOR_TARGETS 实测配置,否则 doubao 消耗会 400 失效) - _send_ark URL 单源:直接拼 HUOSHAN_BASE_URL(与 config.yaml 同 env),删 ARK_HOST - 速率三档对齐旧 auto_consume(正常 15-40s / 极速 / 动态),避免 plan 大时 60s/次拖过 00:00 - run_consume_all 单次拉包复用,消除 6 次重复分页 volc_limit_sync.py: - 修复 DEBUG 打印残留 current_peak 引用(v2 简化后已删变量,DEBUG 模式会 NameError) - 清理旧遗留死变量 SYNC_INTERVAL / mode_str / display_limit,去重 limit_str
This commit is contained in:
+50
-18
@@ -31,8 +31,7 @@ import volc_limit_sync as vls
|
|||||||
# ==========================================
|
# ==========================================
|
||||||
# 🛠️ Configuration
|
# 🛠️ Configuration
|
||||||
# ==========================================
|
# ==========================================
|
||||||
ARK_HOST = os.getenv("ARK_HOST", "ark.cn-beijing.volces.com")
|
# Ark endpoint 单源:URL 直接拼自 HUOSHAN_BASE_URL(与 config.yaml 同一 env)
|
||||||
# 与 config.yaml 的 os.environ/HUOSHAN_BASE_URL 一致;用于生成与 LiteLLM 路径相同的 usage key
|
|
||||||
HUOSHAN_BASE_URL = os.getenv(
|
HUOSHAN_BASE_URL = os.getenv(
|
||||||
"HUOSHAN_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3"
|
"HUOSHAN_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3"
|
||||||
)
|
)
|
||||||
@@ -295,8 +294,9 @@ def calculate_plan_for_model(packages, model):
|
|||||||
# 🔌 Ark 直连请求
|
# 🔌 Ark 直连请求
|
||||||
# ==========================================
|
# ==========================================
|
||||||
def _send_ark(payload, timeout=120):
|
def _send_ark(payload, timeout=120):
|
||||||
"""直连 Ark Chat Completions(Bearer token,不经过 LiteLLM)。"""
|
"""直连 Ark Chat Completions(Bearer token,不经过 LiteLLM)。
|
||||||
url = f"https://{ARK_HOST}/api/v3/chat/completions"
|
单源配置:endpoint 直接取 HUOSHAN_BASE_URL(与 config.yaml 同一 env),防 region 漂移。"""
|
||||||
|
url = f"{HUOSHAN_BASE_URL.rstrip('/')}/chat/completions"
|
||||||
headers = {
|
headers = {
|
||||||
"Content-Type": "application/json",
|
"Content-Type": "application/json",
|
||||||
"Authorization": f"Bearer {ARK_API_KEY}",
|
"Authorization": f"Bearer {ARK_API_KEY}",
|
||||||
@@ -305,10 +305,14 @@ def _send_ark(payload, timeout=120):
|
|||||||
|
|
||||||
|
|
||||||
def _build_payload(model, source_groups, order):
|
def _build_payload(model, source_groups, order):
|
||||||
"""构造消耗请求:日志语料 + 任务提示词,max_tokens 限制输出。"""
|
"""构造消耗请求:日志语料 + 任务提示词。
|
||||||
|
模型参数与旧 MONITOR_TARGETS 实测配置一致:
|
||||||
|
- doubao-seed 系列 → max_completion_tokens + reasoning_effort minimal(火山方舟规范)
|
||||||
|
- 其他(glm/deepseek)→ max_tokens
|
||||||
|
"""
|
||||||
dynamic_input = generate_dynamic_payload(source_groups, order)
|
dynamic_input = generate_dynamic_payload(source_groups, order)
|
||||||
tpl = random.choice(TASK_PROMPTS)
|
tpl = random.choice(TASK_PROMPTS)
|
||||||
return {
|
payload = {
|
||||||
"model": model,
|
"model": model,
|
||||||
"messages": [
|
"messages": [
|
||||||
{"role": "system", "content": tpl["sys"]},
|
{"role": "system", "content": tpl["sys"]},
|
||||||
@@ -318,8 +322,13 @@ def _build_payload(model, source_groups, order):
|
|||||||
},
|
},
|
||||||
],
|
],
|
||||||
"stream": False,
|
"stream": False,
|
||||||
"max_tokens": 1000,
|
|
||||||
}
|
}
|
||||||
|
if model.startswith("doubao"):
|
||||||
|
payload["reasoning_effort"] = "minimal"
|
||||||
|
payload["max_completion_tokens"] = 1000
|
||||||
|
else:
|
||||||
|
payload["max_tokens"] = 1000
|
||||||
|
return payload
|
||||||
|
|
||||||
|
|
||||||
# ==========================================
|
# ==========================================
|
||||||
@@ -350,19 +359,21 @@ def _record_consumption(model, tokens):
|
|||||||
# ==========================================
|
# ==========================================
|
||||||
# 🎯 单模型消耗执行
|
# 🎯 单模型消耗执行
|
||||||
# ==========================================
|
# ==========================================
|
||||||
def run_consume_model(model, source_groups, order, dry_run=False):
|
def run_consume_model(model, source_groups, order, packages=None, dry_run=False):
|
||||||
"""
|
"""
|
||||||
对单个监控模型执行"洗过期包"消耗。
|
对单个监控模型执行"洗过期包"消耗。
|
||||||
目标 = Σ窗口内包计划(均分×1.1/冲刺),clamp:
|
目标 = Σ窗口内包计划(均分×1.1/冲刺),clamp:
|
||||||
- 采集上限:plan ≤ CONSUME_DAILY_CAP − 当日已消耗
|
- 采集上限:plan ≤ CONSUME_DAILY_CAP − 当日已消耗
|
||||||
- 防扣费:plan ≤ 余额快照 − BUFFER
|
- 防扣费:plan ≤ 余额快照 − BUFFER
|
||||||
停止条件(每轮检查):daily_usage ≥ 2M / 余额快照−已烧 ≤ 0 / 已烧 ≥ plan
|
停止条件(每轮检查):daily_usage ≥ 2M / 余额快照−已烧 ≤ 0 / 已烧 ≥ plan
|
||||||
|
:param packages: 复用 run_consume_all 的一次拉包结果;None 时自行拉取
|
||||||
"""
|
"""
|
||||||
try:
|
if packages is None:
|
||||||
packages = vls.fetch_all_packages()
|
try:
|
||||||
except Exception as e:
|
packages = vls.fetch_all_packages()
|
||||||
vls.log(f"❌ [{model}] 拉取资源包失败: {e}", level="ERROR")
|
except Exception as e:
|
||||||
return
|
vls.log(f"❌ [{model}] 拉取资源包失败: {e}", level="ERROR")
|
||||||
|
return
|
||||||
|
|
||||||
plan, details, balance = calculate_plan_for_model(packages, model)
|
plan, details, balance = calculate_plan_for_model(packages, model)
|
||||||
for d in details:
|
for d in details:
|
||||||
@@ -401,12 +412,26 @@ def _run_requests(model, plan, balance_snapshot, source_groups, order):
|
|||||||
fail_count = 0
|
fail_count = 0
|
||||||
deadline = time.time() + MAX_RUNTIME_SECONDS
|
deadline = time.time() + MAX_RUNTIME_SECONDS
|
||||||
|
|
||||||
# 速率估算:单次 ~4.3 万 token(输入 3-5.5 万字符 + 输出 ~1000)
|
# 速率三档(对齐旧 auto_consume run_stealthy_consume):
|
||||||
|
# 任务轻松(单次间隔 > MAX_SLEEP_BASE)→ 正常模式 15-40s 伪装,不赶时间
|
||||||
|
# 任务极重(间隔 < 2s)→ 极速模式 1-3s(本场景 plan≤2M 实际不会触发)
|
||||||
|
# 其余 → 动态 ±5s
|
||||||
|
# 目标:23:00 触发、00:00 结算,1 小时内完成(plan=2M ≈ 47 次 × 27s ≈ 21 分钟)
|
||||||
est_per_req = (MIN_INPUT_CHARS + MAX_INPUT_CHARS) / 2 + 1000
|
est_per_req = (MIN_INPUT_CHARS + MAX_INPUT_CHARS) / 2 + 1000
|
||||||
est_requests = max(1, plan / est_per_req)
|
est_requests = max(1, plan / est_per_req)
|
||||||
interval = int(min(60, max(2, 3600 / est_requests)))
|
interval = int(3600 / est_requests)
|
||||||
cur_min = max(1, interval - 5)
|
if interval > MAX_SLEEP_BASE:
|
||||||
cur_max = interval + 5
|
cur_min, cur_max = MIN_SLEEP_BASE, MAX_SLEEP_BASE
|
||||||
|
mode_str = "正常模式 🐢"
|
||||||
|
elif interval < 2:
|
||||||
|
cur_min, cur_max = 1, 3
|
||||||
|
mode_str = "极速模式 🐇"
|
||||||
|
else:
|
||||||
|
cur_min, cur_max = max(1, interval - 5), interval + 5
|
||||||
|
mode_str = f"动态调整 🚗 ({cur_min}~{cur_max}s)"
|
||||||
|
vls.log(
|
||||||
|
f"⚡ 速率控制: {mode_str} | 预计 {int(est_requests)} 次请求 (单次 ≈{int(est_per_req):,} token)"
|
||||||
|
)
|
||||||
|
|
||||||
while burned < plan:
|
while burned < plan:
|
||||||
if time.time() >= deadline:
|
if time.time() >= deadline:
|
||||||
@@ -483,9 +508,16 @@ def run_consume_all(dry_run=False):
|
|||||||
f"📄 语料来源: {', '.join(f'{k}({len(v)}行)' for k, v in source_groups.items())}"
|
f"📄 语料来源: {', '.join(f'{k}({len(v)}行)' for k, v in source_groups.items())}"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# 只拉一次资源包,全部模型复用(每模型独立 fetch 会重复分页 6 次)
|
||||||
|
try:
|
||||||
|
packages = vls.fetch_all_packages()
|
||||||
|
except Exception as e:
|
||||||
|
vls.log(f"❌ 拉取资源包失败: {e}", level="ERROR")
|
||||||
|
packages = []
|
||||||
|
|
||||||
for model in models:
|
for model in models:
|
||||||
try:
|
try:
|
||||||
run_consume_model(model, source_groups, order, dry_run=dry_run)
|
run_consume_model(model, source_groups, order, packages=packages, dry_run=dry_run)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
vls.log(f"❌ [{model}] 消耗任务异常: {e}\n{traceback.format_exc()}", level="ERROR")
|
vls.log(f"❌ [{model}] 消耗任务异常: {e}\n{traceback.format_exc()}", level="ERROR")
|
||||||
|
|
||||||
|
|||||||
+4
-9
@@ -91,7 +91,6 @@ REGION = "cn-beijing"
|
|||||||
|
|
||||||
# Daemon Config
|
# Daemon Config
|
||||||
BUFFER = 300000 # User requested 30w buffer
|
BUFFER = 300000 # User requested 30w buffer
|
||||||
SYNC_INTERVAL = 60
|
|
||||||
CHECK_INTERVAL = 10
|
CHECK_INTERVAL = 10
|
||||||
THRESHOLD = 300000
|
THRESHOLD = 300000
|
||||||
|
|
||||||
@@ -539,10 +538,8 @@ def restore_blocked_models():
|
|||||||
# Morning Window: 07:00 - 13:00
|
# Morning Window: 07:00 - 13:00
|
||||||
if 7 <= beijing_hour < 13:
|
if 7 <= beijing_hour < 13:
|
||||||
interval = 1800 # 30 minutes
|
interval = 1800 # 30 minutes
|
||||||
mode_str = "Morning/Window (30m)"
|
|
||||||
else:
|
else:
|
||||||
interval = 14400 # 4 hours
|
interval = 14400 # 4 hours
|
||||||
mode_str = "Off-Peak (4h)"
|
|
||||||
|
|
||||||
if now - last_restore_check < interval:
|
if now - last_restore_check < interval:
|
||||||
return
|
return
|
||||||
@@ -806,7 +803,9 @@ class DailyLimitHandler(CustomLogger):
|
|||||||
effective_limit = get_effective_limit(
|
effective_limit = get_effective_limit(
|
||||||
clean_model, api_base, current_usage=current_total
|
clean_model, api_base, current_usage=current_total
|
||||||
)
|
)
|
||||||
display_limit = f"/{int(effective_limit):,}" if effective_limit else ""
|
limit_str = (
|
||||||
|
f" | Limit: {int(effective_limit):,}" if effective_limit is not None else ""
|
||||||
|
)
|
||||||
|
|
||||||
# Get Dynamic Remaining for Log
|
# Get Dynamic Remaining for Log
|
||||||
dynamic_key = f"limit:dynamic:volcengine:remaining:{clean_model}"
|
dynamic_key = f"limit:dynamic:volcengine:remaining:{clean_model}"
|
||||||
@@ -829,10 +828,6 @@ class DailyLimitHandler(CustomLogger):
|
|||||||
print(f"DEBUG: Dynamic key not found: {dynamic_key}")
|
print(f"DEBUG: Dynamic key not found: {dynamic_key}")
|
||||||
|
|
||||||
key_display = key.split(":", 2)[-1]
|
key_display = key.split(":", 2)[-1]
|
||||||
# Display Limit clearly
|
|
||||||
limit_str = (
|
|
||||||
f" | Limit: {int(effective_limit):,}" if effective_limit is not None else ""
|
|
||||||
)
|
|
||||||
log(
|
log(
|
||||||
f"📊 [Usage] {key_display}: +{total_tokens} (Today: {current_total:,}{rem_display}{limit_str})",
|
f"📊 [Usage] {key_display}: +{total_tokens} (Today: {current_total:,}{rem_display}{limit_str})",
|
||||||
level="INFO",
|
level="INFO",
|
||||||
@@ -1419,7 +1414,7 @@ def active_monitor_loop():
|
|||||||
static_limit, stable_limit
|
static_limit, stable_limit
|
||||||
)
|
)
|
||||||
print(
|
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)}",
|
f"DEBUG: {target} Pool={int(total_pool):,} | Usage={int(current_usage):,} | {format_limit_debug_fields(static_limit=static_limit, stable_limit=stable_limit, effective_limit=effective_limit)}",
|
||||||
flush=True,
|
flush=True,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user