From 96b81d52c4603085d4c2e72a89cfdfadd972d490 Mon Sep 17 00:00:00 2001 From: vibe-code Date: Tue, 11 Aug 2026 13:15:24 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=90=88=E5=85=A5=E7=81=AB=E5=B1=B1?= =?UTF-8?q?=E5=85=8D=E8=B4=B9=E9=A2=9D=E5=BA=A6=E8=87=AA=E5=8A=A8=E6=B6=88?= =?UTF-8?q?=E8=80=97=E4=BB=BB=E5=8A=A1=EF=BC=88=E6=AF=8F=E6=97=A523?= =?UTF-8?q?=E7=82=B9=E6=B4=97=E8=BF=87=E6=9C=9F=E5=8C=85=E9=98=B2=E6=89=A3?= =?UTF-8?q?=E8=B4=B9=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 volc_consume.py：消耗任务模块，直连 Ark 洗 9 天内过期资源包 （旧 calculate_plan 均分×1.1/24h冲刺保留 + 双 clamp：2M 采集上限、余额−BUFFER 防扣费） - volc_limit_sync.py：限额 v2 简化（stable_limit = usage+balance−BUFFER，删除峰值状态机 ~90 行）， HARD_DAILY_LIMITS 统一 2M（pre-block 不滞后，1.8M 折扣作废）， 新增 _maybe_trigger_daily_consume（北京时间 23:00 ± 抖动，SETNX 去重，延迟导入） - docker-compose 挂载 volc_consume.py；AGENTS.md/文档同步 --- AGENTS.md | 13 +- docker-compose.yml | 4 +- merge-auto-consume.md | 48 +++- volc_consume.py | 512 ++++++++++++++++++++++++++++++++++++++++++ volc_limit_sync.py | 151 ++++++------- 5 files changed, 630 insertions(+), 98 deletions(-) create mode 100644 volc_consume.py diff --git a/AGENTS.md b/AGENTS.md index 32395a6..b6ba14d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -5,9 +5,10 @@ ## 1. 项目上下文 此仓库包含用于 **LiteLLM** 的自定义回调脚本，核心功能是限制模型用量（特别是火山引擎/VolcEngine）。 - **核心逻辑**：通过 `quota_interceptor` 拦截请求，并通过 `DailyLimitHandler` 记录用量到 Redis。后台 `active_monitor_loop` 守护线程定期从火山引擎 API 同步配额到 Redis。 -- **架构**：所有逻辑统一在 `volc_limit_sync.py` 单文件中。LiteLLM 通过 `config.yaml` 中 `callbacks: ["volc_limit_sync.proxy_handler_instance"]` 加载回调，模块加载时自动启动后台同步线程。 +- **架构**：主逻辑在 `volc_limit_sync.py`（限额 + 同步守护 + 每日消耗调度），消耗任务拆分在 `volc_consume.py`（每日北京时间 23:00 由守护线程延迟导入触发；见 `merge-auto-consume.md`）。LiteLLM 通过 `config.yaml` 中 `callbacks: ["volc_limit_sync.proxy_handler_instance"]` 加载回调，模块加载时自动启动后台同步线程。 - **关键文件**： - - `volc_limit_sync.py`: **唯一核心文件**，包含同步守护线程（`active_monitor_loop`）、LiteLLM 回调（`DailyLimitHandler` + `quota_interceptor`）、限额计算、模型屏蔽/恢复等全部逻辑。 + - `volc_limit_sync.py`: **核心文件**，包含同步守护线程（`active_monitor_loop`）、LiteLLM 回调（`DailyLimitHandler` + `quota_interceptor`）、限额计算（v2 简化：`stable_limit = usage + balance − BUFFER`）、模型屏蔽/恢复、以及每日 23:00 消耗任务调度（`_maybe_trigger_daily_consume`，SETNX 去重）。 + - `volc_consume.py`: 消耗任务模块（每日 23:00 触发；手动 `python volc_consume.py --dry-run` / `--force`）。复用 volc_limit_sync 的拉包/归一化/用量记账设施，直连 Ark API 洗即将过期的免费资源包（防 30 天过期蒸发）。 - `config.yaml`: LiteLLM 配置文件，包含模型定义和回调配置。 - `docker-compose.yml`: 容器编排，仅挂载 `config.yaml` 和 `volc_limit_sync.py`。 - `debug_volc_sync.py` / `debug_volc_sync_manual.py`: 调试脚本，用于手动验证 API 和 Redis 状态。 @@ -38,6 +39,14 @@ pip install litellm redis requests - **调试模式**： 设置环境变量 `DAILY_LIMIT_LOG_LEVEL=DEBUG` 可启用详细日志。 +- **消耗任务（手动触发）**： + ```bash + # 需要 HUOSHAN_API_KEY（Ark API Key）与 HUOSHAN_BASE_URL + python volc_consume.py --dry-run # 只计算每个模型的消耗计划，不发送请求 + python volc_consume.py --force # 立即执行（绕过 23:00 时间限制） + ``` + 每日 23:00 由 `volc_limit_sync.py` 守护线程自动触发（`consume:last_run:{日期}` 去重）。 + ## 3. 代码风格指南 (Code Style) ### 3.1 基础规范 diff --git a/docker-compose.yml b/docker-compose.yml index 90e0893..20a7677 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -8,7 +8,9 @@ services: volumes: - ./config.yaml:/app/config.yaml # 【关键】挂载自定义脚本到容器根目录 - - ./volc_limit_sync.py:/app/volc_limit_sync.py + - ./volc_limit_sync.py:/app/volc_limit_sync.py + # 自动消耗任务（每日 23:00 由 volc_limit_sync 调度触发） + - ./volc_consume.py:/app/volc_consume.py command: - "--config=/app/config.yaml" ports: diff --git a/merge-auto-consume.md b/merge-auto-consume.md index 3390071..9ffa81b 100644 --- a/merge-auto-consume.md +++ b/merge-auto-consume.md @@ -155,21 +155,51 @@ daily_usage ≥ 2M → 停（采集上限，再烧不返还 --- -## 四、限额设计（防扣费） +## 四、限额设计（防扣费）——v2 简化逻辑 -### 4.1 effective_limit 语义收敛 +### 4.1 为什么废弃 Peak-Buffer 峰值状态机（存量场景产物） + +旧逻辑 `stable_limit = PeakPool × 0.85`（PeakPool = usage + balance，首锁/门控重锁/DROP 状态机维护）有两个问题： + +1. **比例 margin 在 usage 接近 balance 时数学失效**（致命）： + - `u=0.5b` 时限额允许再烧 `0.775b`，实际余额只剩 `0.5b`（超 55%） + - `u=0.8b` 时允许再烧 `0.73b`，实际只剩 `0.2b`（超 3.6 倍）→ **烧穿扣费** + - 存量场景没爆是因为 usage 远小于 balance；新模型"返还 = 消耗"使 usage 天天逼近 balance → 必炸 +2. **峰值状态机防的是旧 bug**（存量包余额只减不增时，滞后虚高把 PeakPool 漂移、限额越算越多 399,836 → 599,511）。新模型 balance 每天真实跳涨（返还），锁峰值会把真实发放当异常压住——**跟随现值即可，状态机删除** + +### 4.1b 1.7M/1.8M 保守值的来历：post-hoc 滞后时代的遗产 + +旧拦截机制是"请求成功后记账 + 踢模型"（`log_success_event`），**超限请求本身已放行**，存在"滞后一个请求"的漏网量。因此限额必须打折留安全垫：`HARD_DAILY_LIMITS` 默认 1.8M（2M 打 9 折）、`stable_limit` 打 8.5 折（≈1.7M 效应）。 + +**现状已改为请求前闸门**：`async_pre_call_hook` → `_check_request_block`（429 拒绝，不进 router，不触发 retries/fallback）——主拦截不再滞后，"滞后一个请求"的折扣失去存在必要。新公式精确到 2M；BUFFER 30 万保留，但它防的是**余额统计滞后**（实测 20 万内不动，固定量非比例），与请求滞后无关。 + +### 4.2 新公式 ``` -effective_limit = min(HARD_DAILY_LIMITS, stable_limit) -HARD_DAILY_LIMITS → 统一改为 2M（含 doubao-seed-code-preview-251028 的 480 万遗留，全部模型 2M 上限） -stable_limit → PeakPool − 15% margin（保留，防余额统计滞后：实测 20 万内余额不动） +effective_limit = min(HARD_DAILY_LIMITS, max(0, usage + balance − BUFFER)) +HARD_DAILY_LIMITS → 统一 2M（采集上限；doubao-seed-code-preview-251028 480 万遗留一并改为 2M） ``` -- PeakPool = usage + balance 在"日返还"模型下仍成立（≈ 当日可用总额） -- 真实用户被限在 ~1.7M，消耗任务洗过期额度 → 两者不抢额度（FIFO 方向一致） -- 返还延迟/发放不足时：balance 低 → `remaining` 低 → limit 低 → 真实用户提前被限 ✓ +**`usage + balance` = 当日初始可用池，天然稳定**： +- 消耗中：usage ↑ / balance ↓ → 池恒定 → 限额恒定 ✓ +- 返还：balance 跳涨 → 池涨 → 限额次日自动解锁（无需 RECHARGE）✓ +- 过期：balance 跳跌 → 池跌 → 限额收紧（无需 DROP）✓ +- 余额滞后（实测 20 万内不动）：池虚高 ≤ 20 万 → BUFFER 30 万覆盖 ✓ +- 消耗任务 23:00 烧（usage 计入）：池不变 → 滞后窗口内限额偏松 ≤ BUFFER ✓ +- 余额耗尽：池 − 30 万 ≤ 0 → 限额 0 → 全拦，扣费结构性杜绝 ✓ -### 4.2 空窗期防扣费闭环（消耗量计入 daily_usage 的理由） +**与消耗任务同构**：限额放行条件 = `balance − BUFFER ≥ 0`；消耗任务停止条件 = `余额快照 − 已烧 ≤ 0`——同一把尺子。 + +### 4.3 简化后保留/删除清单 + +| 保留 | 删除（~100 行） | +|---|---| +| 同步频率控制（驱动改为 balance，余额越低同步越勤） | `volc:peak_pool:*` 状态机 | +| `remaining = balance − BUFFER` 写入 | RECHARGE / DROP / 首锁 / 门控重锁分支 | +| 软踢 / 恢复 / 审计 / pre-block / usage 记账 | `volc:last_seen_pool` / `volc:last_balance` | +| `get_usage_key` 北京时间滚动 | stable_limit 改直接写 `max(0, usage + balance − BUFFER)` | + +### 4.4 空窗期防扣费闭环（消耗量计入 daily_usage 的理由） 23:00 消耗任务把额度洗掉后余额可能很低，此时真实用户请求必须被拦。拦截依据 = `daily_usage vs effective_limit`： diff --git a/volc_consume.py b/volc_consume.py new file mode 100644 index 0000000..5a912b0 --- /dev/null +++ b/volc_consume.py @@ -0,0 +1,512 @@ +# ========================================== +# 🚀 VolcConsume — 火山免费额度自动消耗任务 +# ========================================== +# 与 volc_limit_sync.py 合并方案的一部分（见 merge-auto-consume.md）： +# - 每日北京时间 23:00 由 volc_limit_sync 守护线程触发（可手动 --force） +# - 洗掉 CONSUME_MONITOR_DAYS 天内过期的资源包额度（FIFO 下即队列头部）， +# 防止 30 天过期蒸发导致"启动资金"缩水（免费额度规则：消耗 X → 次日返还 X） +# - 直连 Ark API（Bearer token，绕过 LiteLLM 限额，避免消耗任务自杀） +# - 消耗量手动 INCR 进 daily_usage（与 LiteLLM 回调同 key）： +# ① 算准采集上限缺口，② 空窗期限额能正确拦截真实用户（防扣费） +# - 三条停止条件：daily_usage ≥ 2M / 余额快照 − 已烧 ≤ 0 / 已烧 ≥ plan +# +# 手动运行： +# python volc_consume.py --dry-run # 只算计划不消耗 +# python volc_consume.py --force # 忽略 23:00 时间限制立即执行 +import os +import sys +import time +import math +import random +import uuid +import re +import traceback +from datetime import datetime, timezone + +import requests + +# 复用 volc_limit_sync 的拉包/归一化/Redis/日志/用量 key +import volc_limit_sync as vls + +# ========================================== +# 🛠️ Configuration +# ========================================== +ARK_HOST = os.getenv("ARK_HOST", "ark.cn-beijing.volces.com") +# 与 config.yaml 的 os.environ/HUOSHAN_BASE_URL 一致；用于生成与 LiteLLM 路径相同的 usage key +HUOSHAN_BASE_URL = os.getenv( + "HUOSHAN_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3" +) +ARK_API_KEY = os.getenv("HUOSHAN_API_KEY", "") + +# 采集上限：火山免费额度规则，每模型每天最多返还 2M（2026-08 确认所有模型统一） +CONSUME_DAILY_CAP = int(os.getenv("CONSUME_DAILY_CAP", "2000000")) +# 计划低于该阈值不消耗（避免无谓请求） +CONSUME_MIN_PLAN = int(os.getenv("CONSUME_MIN_PLAN", "50000")) +# 洗包窗口：ExpiryTime - now <= N 天的包视为"快过期" +CONSUME_MONITOR_DAYS = int(os.getenv("CONSUME_MONITOR_DAYS", "9")) +# 计划超额系数（旧脚本逻辑：多烧 10% 防积压） +PLAN_OVERAGE = 1.1 + +# 请求伪装参数（移植自 auto_consume.py） +MIN_INPUT_CHARS = 30000 +MAX_INPUT_CHARS = 55000 +MIN_SLEEP_BASE = 15 +MAX_SLEEP_BASE = 40 +MAX_RUNTIME_SECONDS = 21600 # 单模型任务最长 6h +MAX_CONSEC_FAILS = 5 + +# 任务提示词池（移植自 auto_consume config.py） +TASK_PROMPTS = [ + {"sys": "你是资深运维专家。", "user": "请分析以下日志数据，找出潜在的风险点。"}, + {"sys": "你是数据结构化助手。", "user": "请忽略乱码，提取以下文本中的关键信息。"}, + {"sys": "你是安全审计员。", "user": "扫描以下文本，查找是否存在敏感操作记录。"}, + { + "sys": "你是系统可靠性工程师。", + "user": "请判断以下日志片段中是否出现服务降级或异常重启的迹象。", + }, + {"sys": "你是日志分析工具。", "user": "汇总以下日志中的错误级别事件，并给出时间分布。"}, + {"sys": "你是性能优化顾问。", "user": "根据以下日志评估系统负载特征，指出瓶颈迹象。"}, +] + +# 日志源候选（容器内大概率不存在，由合成兜底） +LOG_SOURCES = [ + "/var/log/syslog", + "/var/log/messages", + "/var/log/kern.log", + "/var/log/auth.log", +] + +# 敏感信息脱敏规则：(正则, 替换值) +_SENSITIVE_PATTERNS = [ + (re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"), "x.x.x.x"), + (re.compile(r"([0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}"), "xx:xx:xx:xx:xx:xx"), + (re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+"), "user@example.com"), + (re.compile(r"(/home/|/Users/)[^\s\"':]+"), r"\1user"), +] + + +# ========================================== +# 📄 语料生成（移植自 auto_consume.py，原样保留） +# ========================================== +def _sanitize_line(line): + """过滤日志行中的敏感信息（IP、MAC、邮箱、用户目录等）""" + for pattern, repl in _SENSITIVE_PATTERNS: + line = pattern.sub(repl, line) + return line + + +def _synthetic_lines(count=5000): + """ + 生成时间线连续、格式多样的合成日志（语料不足时的兜底）。 + 从约 2 小时前开始，每行间隔约 1.5 秒，模拟真实系统日志流。 + """ + hosts = ["web-01", "web-02", "api-01", "db-01", "cache-01", "gateway-01"] + templates = [ + "kernel: [TCP] Possible SYN flooding on port {port}. Sending cookies.", + "sshd: Accepted publickey for deploy from x.x.x.x port {port} ssh2", + "nginx: worker process, pid={pid}, GET /api/v2/status HTTP/1.1 200 0.00{ms}", + "systemd: Started Daily apt download and clean activities.", + "python3[{pid}]: heartbeat ok, workers={n}, queue_depth={q}", + "postgres[{pid}]: LOG: checkpoint complete: wrote {n} buffers", + "redis-server[{pid}]: Saving DB to disk complete", + "kernel: EXT4-fs (sda1): mounted filesystem with ordered data mode", + ] + start_ts = datetime.now().timestamp() - count * 1.5 + lines = [] + for i in range(count): + ts = datetime.fromtimestamp(start_ts + i * 1.5).strftime("%b %d %H:%M:%S") + host = random.choice(hosts) + tpl = random.choice(templates) + msg = tpl.format( + port=random.randint(1024, 65535), + pid=random.randint(1000, 9999), + n=random.randint(10, 500), + q=random.randint(0, 200), + ms=random.randint(1, 9), + ) + lines.append(f"{ts} {host} {msg}") + return lines + + +def load_base_logs(): + """ + 加载日志语料（按来源分组，保留行序，去除相邻重复、过滤敏感信息）。 + :return: (来源列表, {来源: 日志行列表}) + """ + source_groups = {} + order = [] + + for log_path in LOG_SOURCES: + if os.path.exists(log_path) and os.access(log_path, os.R_OK): + try: + with open(log_path, "r", errors="ignore") as f: + lines = [l.strip() for l in f.readlines()[-8000:] if l.strip()] + if lines: + key = os.path.basename(log_path) + source_groups[key] = [_sanitize_line(l) for l in lines] + order.append(key) + except Exception as e: + vls.log(f"⚠️ 读取 {log_path} 失败: {e}", level="WARNING") + continue + + # 语料不足 1000 行时，用时间线连续的合成日志补充 + total_lines = sum(len(v) for v in source_groups.values()) + if total_lines < 1000: + vls.log("📄 本地日志不足，使用合成日志补齐语料") + source_groups["synthetic"] = _synthetic_lines(5000) + order.append("synthetic") + + # 去除相邻重复行（日志轮转/重复打印的常见现象） + for key in source_groups: + deduped = [] + for line in source_groups[key]: + if not deduped or line != deduped[-1]: + deduped.append(line) + source_groups[key] = deduped + + return order, source_groups + + +def generate_dynamic_payload(source_groups, order): + """ + 生成随机会话头 + 时间线连贯的日志片段。 + 从单个来源中随机起点取连续窗口（保持时间线连续性），混入少量轮转标记。 + """ + target_chars = random.randint(MIN_INPUT_CHARS, MAX_INPUT_CHARS) + + available = [k for k in order if len(source_groups[k]) >= 50] + if not available: + available = [k for k in order] + source_key = random.choice(available) + source_lines = source_groups[source_key] + + window_size = min(len(source_lines), 5000) + start_idx = random.randint(0, max(0, len(source_lines) - window_size)) + sampled = source_lines[start_idx : start_idx + window_size] + + content_lines = [] + total_chars = 0 + for line in sampled: + content_lines.append(line) + total_chars += len(line) + 1 + if total_chars >= target_chars: + break + + markers = [ + "-- Reboot --", + "-- 日志轮转，新文件开始 --", + "-- 会话超时，连接已断开 --", + "# 服务健康检查快照", + "", + ] + for _ in range(random.randint(1, 3)): + if len(content_lines) > 10: + pos = random.randint(0, len(content_lines) - 1) + content_lines.insert(pos, random.choice(markers)) + + content = "\n".join(content_lines) + + session_id = uuid.uuid4() + ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + header = f"[LOG SESSION {session_id} | 来源: {source_key} | 采集时间: {ts} UTC]" + + return f"{header}\n{content}\n[END OF SESSION {session_id}]" + + +# ========================================== +# 📊 消耗计划（旧 calculate_plan 核心 + v2 clamp） +# ========================================== +def _pkg_models(pkg): + """解析资源包归属的监控模型（与 process_packages_breakdown 同口径）。""" + raw_code = pkg.get("ConfigurationCode", "") + raw_name = pkg.get("ConfigurationName", "Unknown") + target_raw = raw_code if raw_code else raw_name + norm = vls.normalize_volc_name(target_raw) + targets = vls.MONITORED_MODEL_ALIASES.get(norm, [norm]) + if isinstance(targets, str): + targets = [targets] + return targets + + +def calculate_plan_for_model(packages, model): + """ + 计算某模型的当日消耗计划与余额快照。 + :return: (plan, details, balance) balance = 该模型全部 Effective token 包剩余 + 逻辑（保留旧脚本）： + - 窗口内包（ExpiryTime - now ≤ CONSUME_MONITOR_DAYS）： + days_left ≤ 1 → 冲刺全量；否则 remaining/ceil(days)×1.1，多包相加 + - 窗口外包不消耗（留给真实用户，FIFO 下被自然消耗） + """ + total_plan = 0 + balance = 0 + details = [] + now = datetime.now(timezone.utc) + + for pkg in packages: + if model not in _pkg_models(pkg): + continue + if pkg.get("Status") != "Effective": + continue + unit = pkg.get("Unit", "token") + if str(unit).lower() != "token": + continue + try: + remaining = int(float(pkg.get("AvailableAmount", 0))) + except Exception: + continue + if remaining < 1000: + continue + balance += remaining + + expiry_str = pkg.get("ExpiryTime", "") + if not expiry_str: + continue + try: + if expiry_str.endswith("Z"): + expiry_str = expiry_str[:-1] + "+00:00" + expiry_date = datetime.fromisoformat(expiry_str) + except Exception: + continue + + days_left = (expiry_date - now).total_seconds() / 86400.0 + if days_left <= 0 or days_left > CONSUME_MONITOR_DAYS: + continue + + if days_left <= 1.0: + plan = remaining + strategy = "🚨 冲刺(最后24h)" + else: + div = math.ceil(days_left) + plan = int((remaining / div) * PLAN_OVERAGE) + if plan > remaining: + plan = remaining + strategy = f"📉 分{div}天(加量10%)" + + if plan > 0: + total_plan += plan + details.append( + f" - [{pkg.get('ConfigurationName')}] 剩:{remaining:,} | 计划:{plan:,} | {strategy}" + ) + + return total_plan, details, balance + + +# ========================================== +# 🔌 Ark 直连请求 +# ========================================== +def _send_ark(payload, timeout=120): + """直连 Ark Chat Completions（Bearer token，不经过 LiteLLM）。""" + url = f"https://{ARK_HOST}/api/v3/chat/completions" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {ARK_API_KEY}", + } + return requests.post(url, headers=headers, json=payload, timeout=timeout) + + +def _build_payload(model, source_groups, order): + """构造消耗请求：日志语料 + 任务提示词，max_tokens 限制输出。""" + dynamic_input = generate_dynamic_payload(source_groups, order) + tpl = random.choice(TASK_PROMPTS) + return { + "model": model, + "messages": [ + {"role": "system", "content": tpl["sys"]}, + { + "role": "user", + "content": f"{tpl['user']}\n\n[DATA START]\n{dynamic_input}\n[DATA END]", + }, + ], + "stream": False, + "max_tokens": 1000, + } + + +# ========================================== +# 📈 用量记账（与 LiteLLM 回调同 key） +# ========================================== +def _usage_key(model): + return vls.get_usage_key(model, HUOSHAN_BASE_URL) + + +def _current_daily_usage(model): + if not vls.redis_client: + return 0 + val = vls.redis_client.get(_usage_key(model)) + return float(val or 0) + + +def _record_consumption(model, tokens): + if not vls.redis_client: + return + try: + key = _usage_key(model) + vls.redis_client.incrby(key, int(tokens)) + vls.redis_client.expire(key, 172800) + except Exception as e: + vls.log(f"❌ 消耗记账失败: {e}", level="ERROR") + + +# ========================================== +# 🎯 单模型消耗执行 +# ========================================== +def run_consume_model(model, source_groups, order, dry_run=False): + """ + 对单个监控模型执行"洗过期包"消耗。 + 目标 = Σ窗口内包计划（均分×1.1/冲刺），clamp： + - 采集上限：plan ≤ CONSUME_DAILY_CAP − 当日已消耗 + - 防扣费：plan ≤ 余额快照 − BUFFER + 停止条件（每轮检查）：daily_usage ≥ 2M / 余额快照−已烧 ≤ 0 / 已烧 ≥ plan + """ + try: + packages = vls.fetch_all_packages() + except Exception as e: + vls.log(f"❌ [{model}] 拉取资源包失败: {e}", level="ERROR") + return + + plan, details, balance = calculate_plan_for_model(packages, model) + for d in details: + vls.log(d) + + if plan <= 0: + vls.log(f"💤 [{model}] 无快过期包，跳过") + return + + usage_now = _current_daily_usage(model) + # clamp 1: 采集上限（超出部分不返还，白烧） + plan = min(plan, CONSUME_DAILY_CAP - usage_now) + # clamp 2: 余额快照 − BUFFER（脚本自身不烧穿余额，防扣费） + plan = min(plan, max(0, balance - vls.BUFFER)) + plan = int(max(0, plan)) + + if plan < CONSUME_MIN_PLAN: + vls.log(f"💤 [{model}] 计划 {plan:,} < 阈值 {CONSUME_MIN_PLAN:,}，跳过") + return + + vls.log( + f"▶️ [{model}] 目标消耗 {plan:,} | 余额 {balance:,} | 当日已用 {usage_now:.0f}" + ) + if dry_run: + vls.log(f"🔍 [dry-run] 仅计算，不发送请求") + return + + _run_requests(model, plan, balance, source_groups, order) + + +def _run_requests(model, plan, balance_snapshot, source_groups, order): + """ + 循环发送消耗请求。速率：目标 1 小时内完成（23:00 触发，00:00 结算）。 + """ + burned = 0 + fail_count = 0 + deadline = time.time() + MAX_RUNTIME_SECONDS + + # 速率估算：单次 ~4.3 万 token（输入 3-5.5 万字符 + 输出 ~1000） + est_per_req = (MIN_INPUT_CHARS + MAX_INPUT_CHARS) / 2 + 1000 + est_requests = max(1, plan / est_per_req) + interval = int(min(60, max(2, 3600 / est_requests))) + cur_min = max(1, interval - 5) + cur_max = interval + 5 + + while burned < plan: + if time.time() >= deadline: + vls.log(f"⏰ [{model}] 已达运行时长上限，停止本次任务", level="WARNING") + break + if fail_count > MAX_CONSEC_FAILS: + vls.log(f"❌ [{model}] 失败次数过多，跳过剩余任务", level="ERROR") + break + # 停止条件 1: 采集上限 + if _current_daily_usage(model) >= CONSUME_DAILY_CAP: + vls.log(f"🛑 [{model}] 当日采集已达上限 {CONSUME_DAILY_CAP:,}，停止", level="WARNING") + break + # 停止条件 2: 余额快照耗尽（防扣费） + if (balance_snapshot - burned) <= 0: + vls.log(f"🛑 [{model}] 余额快照已耗尽，停止（防扣费）", level="WARNING") + break + # 停止条件 3: 计划完成 + if burned >= plan: + break + + payload = _build_payload(model, source_groups, order) + try: + resp = _send_ark(payload) + if resp.status_code == 200: + fail_count = 0 + total = resp.json().get("usage", {}).get("total_tokens", 0) + if total > 0: + _record_consumption(model, total) + burned += total + percent = min(100, burned / plan * 100) + vls.log( + f"✨ [{model}] 进度 {percent:.1f}% | 本次 {total:,} | 累计 {burned:,}/{plan:,}" + ) + if burned < plan: + time.sleep(random.randint(cur_min, cur_max)) + else: + if resp.status_code == 401: + vls.log(f"❌ [{model}] 认证失败 (401)，API Key 可能已失效，中止", level="ERROR") + break + fail_count += 1 + backoff = min(60, 2**fail_count) + random.uniform(0, 1) + vls.log( + f"⏳ [{model}] API 错误 {resp.status_code}，{backoff:.1f}s 后重试", + level="ERROR", + ) + time.sleep(backoff) + except Exception as e: + fail_count += 1 + vls.log(f"❌ [{model}] 请求异常: {e}", level="ERROR") + time.sleep(10) + + vls.log(f"✅ [{model}] 消耗结束。实际消耗: {burned:,}/{plan:,}") + + +# ========================================== +# 🚀 总入口 +# ========================================== +def run_consume_all(dry_run=False): + """遍历所有监控模型执行消耗。模型列表复用 volc_limit_sync 的 config.yaml 发现。""" + if not ARK_API_KEY: + vls.log("❌ 未配置 HUOSHAN_API_KEY，无法消耗", level="ERROR") + return + + if not vls.MONITORED_MODEL_PAIRS: + vls.discover_monitored_models() + models = sorted({m for m, _h in vls.MONITORED_MODEL_PAIRS}) + if not models: + vls.log("❌ 未发现监控模型（config.yaml 无 volcengine 条目）", level="ERROR") + return + + vls.log(f"📚 加载语料库...") + order, source_groups = load_base_logs() + vls.log( + f"📄 语料来源: {', '.join(f'{k}({len(v)}行)' for k, v in source_groups.items())}" + ) + + for model in models: + try: + run_consume_model(model, source_groups, order, dry_run=dry_run) + except Exception as e: + vls.log(f"❌ [{model}] 消耗任务异常: {e}\n{traceback.format_exc()}", level="ERROR") + + vls.log("🎉 消耗任务完成") + + +def main(): + dry_run = "--dry-run" in sys.argv + force = "--force" in sys.argv + + if not force and not dry_run: + beijing_hour = (time.gmtime(time.time()).tm_hour + 8) % 24 + if beijing_hour != 23: + vls.log( + f"⏰ 当前北京时间 {beijing_hour} 时，非 23:00 触发窗口。加 --force 强制执行", + level="WARNING", + ) + return + + run_consume_all(dry_run=dry_run) + + +if __name__ == "__main__": + main() diff --git a/volc_limit_sync.py b/volc_limit_sync.py index 3d0a2f2..097077f 100644 --- a/volc_limit_sync.py +++ b/volc_limit_sync.py @@ -3,6 +3,8 @@ import sys import time import json import hmac +import math +import random import hashlib import datetime import requests @@ -146,13 +148,10 @@ MONITORED_MODEL_ALIASES = {} # 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, + # 火山免费额度规则（2026-08）：每模型每日采集上限 2M，次日按采集量发放资源包。 + # 请求前闸门（async_pre_call_hook）已不滞后 → 无需打折（旧 1.8M/480 万为 post-hoc 时代遗产）。 + # 统一 2M，余额相关的动态限额由 stable_limit（usage+balance−BUFFER）表达。 + "__DEFAULT__": 2_000_000, } # ========================================== @@ -973,8 +972,12 @@ def process_packages_breakdown(packages): return model_counts -def active_monitor_loop(): - # Config Discovery +def discover_monitored_models(): + """ + 🔎 从 config.yaml 自动发现纳入监控的火山模型，填充 MONITORED_MODEL_PAIRS / + MONITORED_MODEL_ALIASES。返回监控模型名 set。 + 独立函数化：volc_consume.py 手动运行（无守护线程）时也可复用。 + """ global MONITORED_MODEL_PAIRS, MONITORED_MODEL_ALIASES MONITORED_MODEL_PAIRS.clear() MONITORED_MODEL_ALIASES.clear() @@ -1107,6 +1110,50 @@ def active_monitor_loop(): except Exception as e: print(f"Config Auto-discovery failed / 配置自动发现失败: {e}", flush=True) + return MONITORED_MODELS + + +def _maybe_trigger_daily_consume(): + """ + 🕰️ 每日北京时间 23:00 触发免费额度消耗任务（洗过期资源包）。 + - consume:last_run:{北京日期} 去重（SETNX，当天只触发一次） + - daemon 线程执行，不阻塞主循环；线程内 0-10 分钟随机抖动，避免固定 23:00 被风控识别 + - 延迟 import volc_consume，避免模块级循环 import + """ + if not redis_client: + return + try: + now = time.time() + beijing_ts = now + (8 * 3600) + beijing_hour = time.gmtime(beijing_ts).tm_hour + if beijing_hour != 23: + return + date_str = time.strftime("%Y-%m-%d", time.gmtime(beijing_ts)) + lock_key = f"consume:last_run:{date_str}" + if redis_client.set(lock_key, "1", nx=True, ex=172800) is None: + return # 今天已触发过 + + log("🚀 触发每日消耗任务（洗过期资源包防退费）...", level="INFO") + + def _run(): + try: + time.sleep(random.randint(0, 600)) + import volc_consume # 延迟导入，避免模块级循环 import + + volc_consume.run_consume_all() + except Exception as e: + log(f"❌ 消耗任务异常: {e}\n{traceback.format_exc()}", level="ERROR") + + t = threading.Thread(target=_run, daemon=True) + t.start() + except Exception as e: + log(f"❌ 消耗任务调度失败: {e}", level="ERROR") + + +def active_monitor_loop(): + # Config Discovery + MONITORED_MODELS = discover_monitored_models() + if not redis_client: print( "❌ Redis not connected. Exiting monitor loop / Redis 未连接，退出监控循环.", @@ -1142,6 +1189,7 @@ def active_monitor_loop(): while True: try: time.sleep(CHECK_INTERVAL) + _maybe_trigger_daily_consume() now = time.time() # Enforce Beijing Time for daily operations @@ -1336,86 +1384,17 @@ def active_monitor_loop(): dynamic_key = f"limit:dynamic:volcengine:remaining:{target}" # ========================================== - # 🏔️ Daily Peak Lock (First-Sync + Balance-Gated Re-Lock) + # 📏 v2 简化限额：stable_limit = usage + balance − BUFFER # ========================================== - # 背景：AvailableAmount 是各资源包实例的实时剩余，但火山统计有 - # 更新滞后（高用量时段会短暂冻结，实测 20 万 token 消耗内余额不动）。 - # 旧逻辑把每次用量增长误判为"余额恢复"（Jump）锁更高峰值， - # 导致 stable_limit 越算越多（当天 399,836 -> 599,511 一路抬升）。 - # 修正策略（日内限额单调可控）： - # 1. 首次同步锁定当日峰值（= usage + balance 的总配额估计）； - # 2. 仅当 balance 自身净上涨（新资源包发放/对账回补，≥50k） - # 时，按新增额度抬高峰值 —— 幅度 = 历史峰值 + 发放增量， - # 绝不把滞后虚高的 usage 一起算进去； - # 3. balance 回落（额度扣减追上用量）时 Drop 收紧； - # 4. usage 增长驱动的 pool 上涨一律不得抬高限额。 - # 实测：协作奖励计划包每自然日发放、偶发一天 2 个，故必须有第 2 条； - # 次日 00:00 (CST) 自动重建当日峰值。 + # 旧 Peak-Buffer（峰值首锁/门控重锁/DROP，~90 行状态机）为存量 + # 资源包场景设计；日返还模型下 balance 每天真实跳涨，锁峰值会把 + # 真实发放当异常压住，且比例 margin 在 usage 逼近 balance 时 + # 数学失效（允许再烧量 > 剩余余额 → 烧穿扣费）。 + # usage + balance = 当日初始可用池，消耗中恒定；返还/过期自然增减。 + # BUFFER 覆盖余额统计滞后（实测 20 万内不动）。见 merge-auto-consume.md §4。 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 - last_balance_key = f"volc:last_balance:{target}" - last_balance_val = redis_client.get(last_balance_key) - last_balance = float(last_balance_val) if last_balance_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. INITIALIZATION (First sync of the day): lock once - if historical_peak == 0: - if DEBUG_MODE and last_pool > 0: - print( - f"DEBUG: 📌 Locking Daily Peak ({int(last_pool):,} -> {int(total_pool):,}) - First Sync Today.", - flush=True, - ) - current_peak = total_pool - redis_client.setex(peak_key, 172800, int(current_peak)) - - # 2. RECHARGE (Balance-driven upward re-lock) - # 新资源包发放 / API 对账回补 → balance 自身净上涨 ≥50k。 - # 只把"发放增量"加进已锁定的峰值，不重算整个滞后 pool。 - elif last_balance > 0 and (balance - last_balance) >= 50000: - grant_delta = int(balance - last_balance) - current_peak = historical_peak + grant_delta - if DEBUG_MODE: - print( - f"DEBUG: 🆕 New Quota Detected (balance {int(last_balance):,} -> {int(balance):,}, +{grant_delta:,}) - Re-locking Daily Peak to {int(current_peak):,}.", - flush=True, - ) - redis_client.setex(peak_key, 172800, int(current_peak)) - - # 3. DROP DETECTION (Resource Expiry / latency catch-up) - # pool 回落时收紧限额；回落之后保持不变，绝不回升。 - 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)) - - # 4. NO USAGE-DRIVEN RE-LOCK: 用量增长（balance 不变）不得抬高限额。 - # （旧逻辑的 Bug 就在这里：把滞后虚高当余额恢复。） - - # Always update "Last Seen" for next comparison - redis_client.setex(last_pool_key, 86400, int(total_pool)) - redis_client.setex(last_balance_key, 86400, int(balance)) - - # Calculate Stable Limit - safety_margin = max(current_peak * 0.15, 250000) - stable_limit = int(max(0, current_peak - safety_margin)) + stable_limit = int(max(0, total_pool - BUFFER)) stable_limit_key = ( f"limit:dynamic:volcengine:stable_limit:{target}"