02da23ef42
Edge端: - 新增 SQLite 异步任务队列 (queue_manager + consumer) - 新增 TokenBucket 速率限制器 (Gemini 1000 RPM, NVIDIA 40 RPM, burst 2x) - 新增 /api/edge/video/enqueue + /api/edge/results 端点 - 消费者线程从队列消费任务，按速率限制调用AI模型 - orchestrator 集成 rate_limiter，Gemini优先→NVIDIA兜底 NAS端: - Dispatcher 重构为 enqueue 模式（上传后立即返回，不等结果） - 新增 Poller 线程（定期从Edge拉取结果写 MariaDB） - app.py 启动 Poller，config.yaml 新增 poller 配置 - db_layer 更新 valid_stages 添加 'process'
49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
"""
|
|
Rate Limiter - Token Bucket 算法
|
|
|
|
按 API 限制速度的 2 倍设置突发容量，按 API 限制速度持续补充。
|
|
"""
|
|
import time
|
|
import threading
|
|
|
|
|
|
class TokenBucket:
|
|
def __init__(self, rpm: int, burst_factor: int = 2):
|
|
self.capacity = rpm * burst_factor
|
|
self.refill_rate = rpm / 60.0
|
|
self.tokens = float(self.capacity)
|
|
self.last_refill = time.monotonic()
|
|
self._lock = threading.Lock()
|
|
|
|
def acquire(self, tokens: int = 1, timeout: float = 300.0) -> bool:
|
|
deadline = time.monotonic() + timeout
|
|
while True:
|
|
with self._lock:
|
|
now = time.monotonic()
|
|
elapsed = now - self.last_refill
|
|
self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
|
|
self.last_refill = now
|
|
if self.tokens >= tokens:
|
|
self.tokens -= tokens
|
|
return True
|
|
wait = (tokens - self.tokens) / self.refill_rate
|
|
if time.monotonic() + wait > deadline:
|
|
return False
|
|
time.sleep(min(wait, 1.0))
|
|
|
|
|
|
class RateLimiter:
|
|
"""多 API 速率限制管理"""
|
|
|
|
def __init__(self):
|
|
self._buckets = {}
|
|
|
|
def register(self, name: str, rpm: int, burst_factor: int = 2):
|
|
self._buckets[name] = TokenBucket(rpm, burst_factor)
|
|
|
|
def acquire(self, name: str, tokens: int = 1, timeout: float = 300.0) -> bool:
|
|
bucket = self._buckets.get(name)
|
|
if bucket is None:
|
|
return True
|
|
return bucket.acquire(tokens, timeout)
|