Compare commits
4 Commits
8feb38db37
...
5e40a311d1
| Author | SHA1 | Date | |
|---|---|---|---|
 vibe-code
|
5e40a311d1 | ||
|
vibe-code
|
b7df36bb4e | ||
|
vibe-code
|
96b81d52c4 | ||
|
vibe-code
|
2cc7c7d08b |
@@ -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`，dynamic 缺失时 fallback `FALLBACK_LIMIT` 90 万兜底）、模型屏蔽/恢复、以及每日 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 基础规范
|
||||
|
||||
@@ -9,6 +9,8 @@ services:
|
||||
- ./config.yaml:/app/config.yaml
|
||||
# 【关键】挂载自定义脚本到容器根目录
|
||||
- ./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:
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
# 火山免费额度自动消耗：auto_consume 合入 volc_limit_sync 设计方案
|
||||
|
||||
> 日期：2026-08-11
|
||||
> 背景：`/root/workspace/huoshan-api-auto_consume/`（自动消耗脚本）与 `/root/workspace/litellm-config/`（LiteLLM 限流回调 + 火山余额同步）合并，使每日 23:00 自动消耗免费额度，防止资源包过期蒸发，避免真实扣费。
|
||||
> 前置文档：《merge-volc-limit-into-newapi.md》（2026-08-05，方向相反：把 volc_limit_sync 功能迁往 new-api，未实施）。
|
||||
|
||||
---
|
||||
|
||||
## 一、火山免费额度业务模型（一切设计的事实基础）
|
||||
|
||||
### 1.1 日返还循环
|
||||
|
||||
```
|
||||
每天任意时刻 真实用户 + 消耗任务 的消耗 = 当日采集量（火山实时统计，每模型单独计算）
|
||||
每天 23:00 CST 消耗任务补齐缺口（可选，见 §4）
|
||||
次日 00:00 起 火山按昨日采集量发放资源包（不定时，可能延迟到上午）
|
||||
```
|
||||
|
||||
### 1.2 五个硬规则（用户确认）
|
||||
|
||||
| 规则 | 值 | 影响 |
|
||||
|---|---|---|
|
||||
| 采集上限 | **2M token/模型/天** | 消耗超过 2M 的部分**不返还**（白烧）；所有模型统一 |
|
||||
| 发放额度 | = 昨日采集量（≤2M） | 消耗 X → 次日返还 X，用多少补多少，延迟一天到账 |
|
||||
| 发放时间 | **不定时** | 次日 0:00 至发放时点存在余额空窗 |
|
||||
| 资源包有效期 | **30 天** | 过期额度实打实消失，失去"启动资金" |
|
||||
| 消耗顺序 | **FIFO（先消耗获得时间早的包）** | 队列头部 = 最早获得 = 最早过期（效期统一 30 天时两者等价） |
|
||||
| 余额耗尽 | **请求不失败，按真实价格扣费** | 全部防护的终极目标：**避免扣费** |
|
||||
| 首次开通 | **每模型赠送 2M** | 启动资金的种子（无对应前日采集的免费午餐） |
|
||||
|
||||
### 1.3 关键账目：启动资金守恒
|
||||
|
||||
```
|
||||
余额_{N+1} = 余额_N − 消耗_N + 返还_{N+1}
|
||||
= 余额_N − 消耗_N + 采集_N
|
||||
= 余额_N （消耗即采集，≤2M 内全额返还）
|
||||
```
|
||||
|
||||
**推论**：
|
||||
- **使用不损失额度**（用多少补多少，仅延迟一天）
|
||||
- **过期才损失额度**（额度蒸发 = 启动资金永久减少）
|
||||
- 消耗任务的价值 = 防止"启动资金"被 30 天过期侵蚀，把额度滚成新包永续循环
|
||||
- 真实用量 ≥ 66.7k/天（= 2M/30 天）时，赠送包被自然消耗，脚本自动闲置；低于此值必须靠脚本在最后 9 天窗口洗掉
|
||||
|
||||
### 1.4 扣费风险链（限流系统的真正职责）
|
||||
|
||||
```
|
||||
免费额度耗尽（余额 = 0） → 继续请求 → 真实扣费
|
||||
```
|
||||
|
||||
触发扣费的两种路径：
|
||||
1. **真实用户单日用量 > 当日可用余额**（含空窗期 0:00–发放时点余额不足）
|
||||
2. **消耗任务烧穿余额**（脚本自身引入的扣费，必须结构性杜绝）
|
||||
|
||||
---
|
||||
|
||||
## 二、两项目现状与合并可行性
|
||||
|
||||
### 2.1 当前项目 `volc_limit_sync.py`（1536 行，LiteLLM 回调 + 常驻守护）
|
||||
|
||||
| 能力 | 实现 |
|
||||
|---|---|
|
||||
| 火山余额同步 | SigV4 签名调 `ListResourcePackages`（`Status: Effective` 过滤 + `seen_tokens` 防分页死循环），`normalize_volc_name` 归一化，按监控模型聚合 |
|
||||
| 动态限额 | Peak-Buffer：`PeakPool = 当日用量 + 余额`；`stable_limit = PeakPool − 15% margin（下限 25w）`；与静态 `HARD_DAILY_LIMITS` 取小得 `effective_limit` |
|
||||
| 当日用量计数 | LiteLLM 成功回调 `INCRBY total_tokens` → Redis `daily_usage:{北京日期}:{model}:{api_base_hash}`，北京时间 00:00 滚动，TTL 2 天 |
|
||||
| 超限停用/恢复 | `soft_evict_from_router` 重命名软踢；北京时间 7–13 点每 30 分钟、其余 4 小时检查恢复；`startup_audit` 启动预踢 |
|
||||
| 模型发现 | 解析 `config.yaml` 中 `volcengine/` 前缀 + `HUOSHAN_BASE_URL` 条目 → `MONITORED_MODEL_PAIRS`；版本号自动推导别名 + `MANUAL_MONITORED_MODEL_ALIASES` 手工覆盖 |
|
||||
|
||||
### 2.2 目标项目 `huoshan-api-auto_consume/`（420 行 + 4 辅助）
|
||||
|
||||
| 能力 | 实现 |
|
||||
|---|---|
|
||||
| 语料 | `/var/log` 系统日志分组 + 脱敏（IP/MAC/邮箱/路径）+ 不足时合成日志 |
|
||||
| 计划 | `calculate_plan`：9 天窗口内包 `remaining/ceil(days_left)×1.1`，最后 24h 冲刺全量，多包相加 |
|
||||
| 执行 | 直连 Ark API（`ark.cn-beijing.volces.com/api/v3/chat/completions`），速率伪装（15–40s 随机），指数退避，6h 上限，连续失败 5 次跳过 |
|
||||
| 配置 | `config.py`：`MONITOR_TARGETS`（id/keyword/alias/extra_params） |
|
||||
|
||||
### 2.3 可复用资产与冲突点
|
||||
|
||||
**重叠（合并即消除重复）**：
|
||||
- SigV4 签名（`volc_signer.py` ≈ `get_signed_headers`）——**同一账户同一组 AK/SK**
|
||||
- `fetch_all_packages`（目标版缺 `seen_tokens` 防循环保护，保留当前版）
|
||||
- 资源包→模型映射（当前版 `normalize_volc_name` + 别名体系更健壮，目标版 `keyword` 模糊匹配退役）
|
||||
|
||||
**关键事实：模型配置天然单一化**。`config.yaml` 的 volcengine 上游模型名（剥掉 `volcengine/` 前缀）就是 Ark endpoint id，与 `MONITOR_TARGETS` 的 id 一一对应（`deepseek-v4-flash` 的 `-ga-260731` vs `-0731` 差异已由 `MANUAL_MONITORED_MODEL_ALIASES` 覆盖）→ **`MONITOR_TARGETS` 退役，消耗任务直接由 config.yaml 驱动**。
|
||||
|
||||
**冲突点与解法**：
|
||||
1. 限额 vs 消耗：消耗请求**必须直连 Ark，绝不改走 LiteLLM**（否则被 `quota_interceptor` 拦下自杀）
|
||||
2. 消耗量**必须计入 `daily_usage`**（手动 INCR 同 key）——空窗期防扣费闭环的关键，见 §5.3
|
||||
3. 脚本自身不能烧穿余额（结构性防扣费，见 §4.4）
|
||||
|
||||
---
|
||||
|
||||
## 三、消耗策略设计
|
||||
|
||||
### 3.1 为什么不做"烧满 2M"
|
||||
|
||||
API 请求的 token 消耗不可精确控制（输入按字符估算、输出不确定）。若目标 = 补满 2M：
|
||||
- 烧超采集上限的部分不返还 → 白烧
|
||||
- 烧到余额 0 仍继续 → **直接扣费**
|
||||
- 精确校准需要逐请求调整 → 复杂度不可接受
|
||||
|
||||
### 3.2 目标：洗掉"即将过期的额度"（过期风险驱动）
|
||||
|
||||
```
|
||||
消耗量 = Σ(该模型 ExpiryTime − now ≤ MONITOR_DAYS 的资源包剩余额度)
|
||||
```
|
||||
|
||||
- 消耗量 ≤ 快过期包剩余 ≤ 当前可用余额 → **结构上不可能烧穿余额**（估算偏差只决定多烧/少烧几笔请求，多烧部分来自非过期包，次日照常返还，无害）
|
||||
- FIFO 先烧最早获得 = 最早过期的包 → 脚本烧的恰好是最该洗的包，顺序自动正确
|
||||
- 不追求 2M 精确值 → 无校准问题
|
||||
|
||||
### 3.3 均分逻辑（保留旧 `calculate_plan`，经真实使用验证）
|
||||
|
||||
```
|
||||
对每个 9 天窗口内的包：
|
||||
days_left ≤ 1.0 → plan = remaining # 24h 冲刺，全量兜底
|
||||
days_left > 1.0 → plan = remaining / ceil(days_left) × 1.1 # 均分 + 10% 提前量
|
||||
总 plan = Σ(各包 plan)
|
||||
```
|
||||
|
||||
**为什么均分优于一次性洗光**（在新模型下）：
|
||||
1. **最小干扰真实用户**：只在额度即将死亡（进窗口）后才每天接管 1/D，其余留给真实用户 → 免费额度利用率最大化，空窗风险最小
|
||||
2. **请求负载分摊**：每天 3–5 次请求（而非 25–40 次），速率伪装自然，风控友好
|
||||
3. **余额下降平缓**：23:00 后余额仍保留大部分 → 扣费只发生在余额真正归零时，均分下脚本几乎不可能制造归零
|
||||
4. **FIFO 自动一致**：效期统一 30 天 → 获得早 = 过期早 → 队列头部 = 窗口内包，计划与执行不打架（分包明细仅用于定总量）
|
||||
|
||||
**与日返还循环的契合**：旧脚本本就是"每天跑一次、分 D 天烧完"的设计；自动化后每天重算（余额/天数变化 → div 递减 → 当日份额递增），节奏天然吻合。9 天窗口 + 1.1 系数 + 24h 冲刺：只要每天触发成功，任何进窗口的包都赶在过期前清空。
|
||||
|
||||
### 3.4 三个约束（旧逻辑缺失，必须补）
|
||||
|
||||
```
|
||||
plan = min(plan, 2M − daily_usage) # 采集上限：超出不返还，白烧
|
||||
plan = min(plan, 余额快照 − 本次会话已烧 − BUFFER) # 防扣费：脚本不烧穿余额
|
||||
```
|
||||
|
||||
**执行停止条件（每轮请求后检查，3 条闭环）**：
|
||||
```
|
||||
daily_usage ≥ 2M → 停（采集上限，再烧不返还）
|
||||
余额快照 − 本次已烧 ≤ 0 → 停（防扣费）
|
||||
本次已烧 ≥ plan → 停（任务完成）
|
||||
```
|
||||
|
||||
### 3.5 消耗量估算
|
||||
|
||||
- 单次请求：输入 3–5.5 万字符 ≈ 5–11 万 token（中文 1 字 ≈ 1–2 token）+ 输出 ~1000
|
||||
- 极值 2M / 8 万 ≈ 25 次请求 → 15–40s 间隔 ≈ 10–17 分钟，1 小时窗口绰绰有余
|
||||
- 均分模式每天 3–5 次请求，更宽松
|
||||
|
||||
### 3.6 触发调度
|
||||
|
||||
- 北京时间 **23:00 ± 随机抖动**（防固定时间被风控识别），Redis `consume:last_run:{日期}` 去重，当天未跑过才触发
|
||||
- 守护线程（`active_monitor_loop` 已有主循环）检测时点 → spawn daemon 线程执行，不阻塞主循环
|
||||
- 23:00 的时点语义：距 00:00 采集结算 1 小时，此时窗口内包状态已定（真实用户当天用量已发生），洗包目标最准
|
||||
|
||||
---
|
||||
|
||||
## 四、限额设计（防扣费）——v2 简化逻辑
|
||||
|
||||
### 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, max(0, usage + balance − BUFFER))
|
||||
HARD_DAILY_LIMITS → 统一 2M（采集上限；doubao-seed-code-preview-251028 480 万遗留一并改为 2M）
|
||||
```
|
||||
|
||||
**`usage + balance` = 当日初始可用池，天然稳定**：
|
||||
- 消耗中：usage ↑ / balance ↓ → 池恒定 → 限额恒定 ✓
|
||||
- 返还：balance 跳涨 → 池涨 → 限额次日自动解锁（无需 RECHARGE）✓
|
||||
- 过期：balance 跳跌 → 池跌 → 限额收紧（无需 DROP）✓
|
||||
- 余额滞后（实测 20 万内不动）：池虚高 ≤ 20 万 → BUFFER 30 万覆盖 ✓
|
||||
- 消耗任务 23:00 烧（usage 计入）：池不变 → 滞后窗口内限额偏松 ≤ BUFFER ✓
|
||||
- 余额耗尽：池 − 30 万 ≤ 0 → 限额 0 → 全拦，扣费结构性杜绝 ✓
|
||||
|
||||
**fallback（stable_limit 缺失窗口）**：免费模型余额常态 < 2M（昨日采多少返多少），若 dynamic 缺失（首次同步前/新增模型/TTL 过期）时裸奔到静态 2M，余额不足的模型会被烧穿扣费。故 fallback 用保守值 `FALLBACK_LIMIT = min(static, 90万)`（env 可调），宁可短时收紧不可裸奔。
|
||||
|
||||
**与消耗任务同构**：限额放行条件 = `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`：
|
||||
|
||||
- **若消耗量不计入 daily_usage**：限额看到"真实用户才用 100k，未超限" → 放行 → **扣费**
|
||||
- **若计入**：daily_usage ≈ 真实 + 消耗任务 ≈ 2M → 超限拦截 → 空窗期安全
|
||||
|
||||
同步线程检测 balance 归零 → limit 收紧有 2–60min 滞后；消耗量计入让拦截在**滞后窗口内也有效**，双保险。
|
||||
|
||||
### 4.3 限额与消耗共用数据、方向相反
|
||||
|
||||
| 数据 | 限额（拦真实用户） | 消耗任务（洗过期额度） |
|
||||
|---|---|---|
|
||||
| `daily_usage` | 超限踢出 | 停止条件（≥2M 停） |
|
||||
| `remaining`（余额−缓冲） | 限额收紧 | 防烧穿上限 |
|
||||
| `stable_limit` | effective_limit 组成 | 不适用（直连旁路） |
|
||||
|
||||
---
|
||||
|
||||
## 五、架构与实现
|
||||
|
||||
### 5.1 代码形态
|
||||
|
||||
```
|
||||
litellm-config/
|
||||
├── volc_limit_sync.py # 现状：签名/拉取/归一化/限额/守护线程
|
||||
│ # 新增：23:00 调度钩子（检测时点 + spawn 消耗线程 + 去重）
|
||||
├── volc_consume.py # 新增：消耗任务（从 auto_consume 移植 + 改造）
|
||||
│ ├── load_base_logs / generate_dynamic_payload / _sanitize_line # 原样
|
||||
│ ├── calculate_plan # 保留 + 两个 clamp
|
||||
│ ├── run_stealthy_consume # 保留（停止条件 + 手动 INCR）
|
||||
│ └── consume_once(model) # 新封装：查缺口 → 执行 → 记账
|
||||
├── config.yaml # 模型入口不变（消耗任务读同一份监控发现）
|
||||
└── .env # 密钥统一走 env（HUOSHAN_API_KEY 即 Ark key）
|
||||
```
|
||||
|
||||
请求直连 Ark（`HUOSHAN_API_KEY`），复用 `MONITORED_MODEL_PAIRS` / `MONITORED_MODEL_ALIASES` / `fetch_all_packages` / 签名。`MONITOR_TARGETS` 与 `config.py` 不再需要。
|
||||
|
||||
### 5.2 Redis key 一览
|
||||
|
||||
| key | 读写方 | 说明 |
|
||||
|---|---|---|
|
||||
| `daily_usage:{日期}:{model}:{hash}` | LiteLLM 回调 INCR + 消耗任务 INCR | 当日用量，TTL 2 天 |
|
||||
| `limit:dynamic:volcengine:remaining:{model}` | 守护线程写 | 余额 − BUFFER |
|
||||
| `limit:dynamic:volcengine:stable_limit:{model}` | 守护线程写 | 稳定限额 |
|
||||
| `volc:peak_pool:{日期}:{model}` | 守护线程 | 当日峰池 |
|
||||
| `consume:last_run:{日期}` | 消耗任务 | 当日去重，新增 |
|
||||
|
||||
### 5.3 消耗任务执行流程
|
||||
|
||||
```
|
||||
23:00 触发（守护线程）
|
||||
→ consume:last_run:{今天} 已存在？ → 跳过
|
||||
→ 遍历 MONITORED_MODELS（config.yaml 发现的 volcengine 模型）
|
||||
→ fetch_all_packages（复用）→ 按模型过滤窗口内包
|
||||
→ plan = calculate_plan 结果，应用两个 clamp
|
||||
→ plan > 阈值(如 5 万)？
|
||||
→ 读 remaining 快照 → spawn 消耗线程
|
||||
→ 循环：生成语料 payload → 直连 Ark（速率伪装/退避/6h 上限）
|
||||
→ 成功 → INCR daily_usage；每轮检查 3 停止条件
|
||||
→ 失败 > 5 次 → 跳过该模型
|
||||
→ 写 consume:last_run:{今天}
|
||||
```
|
||||
|
||||
### 5.4 代码资产迁移清单
|
||||
|
||||
| 资产 | 来源 | 处理 |
|
||||
|---|---|---|
|
||||
| 签名 / `fetch_all_packages` / 归一化 / 模型发现 | volc_limit_sync | 原样复用 |
|
||||
| `calculate_plan`（9 天窗口/均分×1.1/冲刺/相加） | auto_consume | 保留，加 2 个 clamp |
|
||||
| 语料生成/脱敏/合成 | auto_consume | 原样移植 |
|
||||
| 速率伪装/退避/失败上限/6h 上限 | auto_consume | 原样移植 |
|
||||
| `_record_usage` 的 key 计算（`get_usage_key`） | volc_limit_sync | 消耗任务复用做 INCR |
|
||||
| 调度钩子 | 新增 | 守护线程 23:00 检测 + 去重 |
|
||||
|
||||
---
|
||||
|
||||
## 六、风险与边界（已确认/已知）
|
||||
|
||||
| 项 | 结论 |
|
||||
|---|---|
|
||||
| 空窗期（0:00–次日发放不定时） | **用户确认可接受**；均分策略使空窗概率最小化（余额平缓下降） |
|
||||
| 烧满 2M 的精确性 | 放弃该策略，改为过期风险驱动，结构性规避 |
|
||||
| 物理上限 | 窗口内快过期额度 > 2M 时每天最多洗 2M，极少数过期蒸发不可避（返还上限硬约束） |
|
||||
| 触发连续性 | 依赖每天 23:00 成功；连续失败 ~8 天耗尽窗口缓冲（冲刺兜底），守护线程常驻风险低 |
|
||||
| 风控 | 23:00 ± 抖动 + 15–40s 随机间隔 + 每天 3–5 次请求，接近真实行为 |
|
||||
| 容器语料 | LiteLLM 容器内无 `/var/log` 系统日志 → 合成语料兜底（5000 行够用） |
|
||||
| 消耗任务自身扣费 | 结构性杜绝：plan ≤ 余额快照 − 缓冲，且每轮检查余额 |
|
||||
|
||||
---
|
||||
|
||||
## 七、待确认事项
|
||||
|
||||
1. `MONITOR_DAYS` 沿用 9 天（推荐：每天触发 + 冲刺兜底下足够，且与旧逻辑一致）
|
||||
2. 实现落点：拆 `volc_consume.py` 模块（推荐，单文件将超 2000 行）还是严格并入 `volc_limit_sync.py`
|
||||
3. 消耗触发阈值（plan 小于多少跳过）：建议 5 万 token，避免无谓请求
|
||||
4. 消耗任务与真实用户的额度优先级：均分策略已隐含"真实用户优先"，无需额外开关
|
||||
|
||||
---
|
||||
|
||||
## 八、实施步骤（确认后执行）
|
||||
|
||||
1. 移植 `auto_consume.py` 的语料/执行逻辑到 `volc_consume.py`，请求层改为直连 Ark（env key）
|
||||
2. `calculate_plan` 加两个 clamp + 停止条件；执行循环加 `daily_usage` INCR 与余额检查
|
||||
3. `volc_limit_sync.py` 守护线程加 23:00 调度钩子 + `consume:last_run` 去重
|
||||
4. `HARD_DAILY_LIMITS` 统一为 2M；`MONITOR_TARGETS`/`config.py` 退役
|
||||
5. 用 `test_connection.py` 等价方式验证 Ark 连通 → 手动触发一次消耗任务（dry-run 模式验证 plan 计算与 clamp）→ 观察 Redis key 变化
|
||||
6. 全量回归：守护线程限额逻辑不受影响（消耗线程独立 daemon）
|
||||
+544
@@ -0,0 +1,544 @@
|
||||
# ==========================================
|
||||
# 🚀 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 endpoint 单源：URL 直接拼自 HUOSHAN_BASE_URL（与 config.yaml 同一 env）
|
||||
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）。
|
||||
单源配置：endpoint 直接取 HUOSHAN_BASE_URL（与 config.yaml 同一 env），防 region 漂移。"""
|
||||
url = f"{HUOSHAN_BASE_URL.rstrip('/')}/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):
|
||||
"""构造消耗请求：日志语料 + 任务提示词。
|
||||
模型参数与旧 MONITOR_TARGETS 实测配置一致：
|
||||
- doubao-seed 系列 → max_completion_tokens + reasoning_effort minimal（火山方舟规范）
|
||||
- 其他（glm/deepseek）→ max_tokens
|
||||
"""
|
||||
dynamic_input = generate_dynamic_payload(source_groups, order)
|
||||
tpl = random.choice(TASK_PROMPTS)
|
||||
payload = {
|
||||
"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,
|
||||
}
|
||||
if model.startswith("doubao"):
|
||||
payload["reasoning_effort"] = "minimal"
|
||||
payload["max_completion_tokens"] = 1000
|
||||
else:
|
||||
payload["max_tokens"] = 1000
|
||||
return payload
|
||||
|
||||
|
||||
# ==========================================
|
||||
# 📈 用量记账（与 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, packages=None, dry_run=False):
|
||||
"""
|
||||
对单个监控模型执行"洗过期包"消耗。
|
||||
目标 = Σ窗口内包计划（均分×1.1/冲刺），clamp：
|
||||
- 采集上限：plan ≤ CONSUME_DAILY_CAP − 当日已消耗
|
||||
- 防扣费：plan ≤ 余额快照 − BUFFER
|
||||
停止条件（每轮检查）：daily_usage ≥ 2M / 余额快照−已烧 ≤ 0 / 已烧 ≥ plan
|
||||
:param packages: 复用 run_consume_all 的一次拉包结果；None 时自行拉取
|
||||
"""
|
||||
if packages is None:
|
||||
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
|
||||
|
||||
# 速率三档（对齐旧 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_requests = max(1, plan / est_per_req)
|
||||
interval = int(3600 / est_requests)
|
||||
if interval > MAX_SLEEP_BASE:
|
||||
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:
|
||||
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())}"
|
||||
)
|
||||
|
||||
# 只拉一次资源包，全部模型复用（每模型独立 fetch 会重复分页 6 次）
|
||||
try:
|
||||
packages = vls.fetch_all_packages()
|
||||
except Exception as e:
|
||||
vls.log(f"❌ 拉取资源包失败: {e}", level="ERROR")
|
||||
packages = []
|
||||
|
||||
for model in models:
|
||||
try:
|
||||
run_consume_model(model, source_groups, order, packages=packages, 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()
|
||||
+80
-98
@@ -3,6 +3,8 @@ import sys
|
||||
import time
|
||||
import json
|
||||
import hmac
|
||||
import math
|
||||
import random
|
||||
import hashlib
|
||||
import datetime
|
||||
import requests
|
||||
@@ -89,10 +91,14 @@ REGION = "cn-beijing"
|
||||
|
||||
# Daemon Config
|
||||
BUFFER = 300000 # User requested 30w buffer
|
||||
SYNC_INTERVAL = 60
|
||||
CHECK_INTERVAL = 10
|
||||
THRESHOLD = 300000
|
||||
|
||||
# stable_limit 缺失时（首次同步前/新增模型/TTL 过期）的保守兜底限额。
|
||||
# 免费模型余额常态 < 2M（昨日采多少返多少），若 fallback 裸奔到采集上限 2M，
|
||||
# 余额不足的模型会被烧穿扣费。宁可短时收紧，不可裸奔。
|
||||
FALLBACK_LIMIT = int(os.getenv("FALLBACK_LIMIT", "900000"))
|
||||
|
||||
# Redis Config
|
||||
REDIS_HOST = os.getenv(
|
||||
"REDIS_HOST", "redis"
|
||||
@@ -146,13 +152,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,
|
||||
}
|
||||
|
||||
# ==========================================
|
||||
@@ -438,9 +441,12 @@ def get_effective_limit(clean_model, api_base, current_usage=None):
|
||||
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)
|
||||
# Fallback: 如果 daemon 尚未计算 stable limit（首次同步前/新增模型/TTL 过期），
|
||||
# 用保守兜底（FALLBACK_LIMIT），绝不裸奔到采集上限 —— 免费模型余额常态 < 2M，
|
||||
# 裸奔 2M 会把余额不足的模型烧穿扣费。
|
||||
fallback = float(static_limit) if static_limit is not None else float(FALLBACK_LIMIT)
|
||||
fallback = min(fallback, float(FALLBACK_LIMIT))
|
||||
effective_limit = calculate_effective_limit(fallback, 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}")
|
||||
@@ -532,10 +538,8 @@ def restore_blocked_models():
|
||||
# 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
|
||||
@@ -799,7 +803,9 @@ class DailyLimitHandler(CustomLogger):
|
||||
effective_limit = get_effective_limit(
|
||||
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
|
||||
dynamic_key = f"limit:dynamic:volcengine:remaining:{clean_model}"
|
||||
@@ -822,10 +828,6 @@ class DailyLimitHandler(CustomLogger):
|
||||
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",
|
||||
@@ -973,8 +975,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 +1113,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 +1192,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 +1387,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}"
|
||||
@@ -1432,7 +1414,7 @@ def active_monitor_loop():
|
||||
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)}",
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user