556 lines
19 KiB
Python
556 lines
19 KiB
Python
"""LLM 成本追踪器 + 增强降级策略 (MVP-3 / D34)
|
||
|
||
职责
|
||
----
|
||
|
||
1. 记录每次 LLM 调用的 token / 费用 / 延迟，提供按时段聚合的统计
|
||
2. 实现"硬熔断"降级策略: 当达到日预算 / 错误率阈值时自动禁用 LLM
|
||
3. 提供成本仪表盘所需的导出接口（供 ``api/llm.py`` 暴露给前端）
|
||
|
||
设计要点
|
||
--------
|
||
|
||
* **职责分离**：成本追踪本身只负责"记账 + 状态机"，不直接决定能否调用 LLM；
|
||
调用方 (例如检测管道) 在每次调用前调用 ``can_call()``，根据结果决定降级
|
||
* **滑动窗口**：使用按 UTC 日历日的桶 (``deque``) 保存最近 N 天的统计，
|
||
避免长时间运行内存无限增长
|
||
* **熔断**：达到错误率阈值 (默认连续 ``error_threshold`` 次调用失败率 >50%)
|
||
自动进入 ``CIRCUIT_OPEN`` 状态，``cooldown_seconds`` 后自动半开尝试恢复
|
||
* **Token 估算**：未拿到真实 usage 时按"图像 + prompt 长度"做粗估，
|
||
保证账单不会因为 provider 不返回 usage 而归零
|
||
|
||
接入方式
|
||
--------
|
||
|
||
::
|
||
|
||
tracker = LLMCostTracker(daily_budget_usd=5.0)
|
||
if not tracker.can_call():
|
||
return None # 降级处理
|
||
result = await llm_service.analyze_with_fallback(req)
|
||
tracker.record(result, request=req)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import time
|
||
from collections import deque
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timezone
|
||
from typing import Any, Deque, Dict, List, Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 常量 / 数据结构
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
# 不同 provider 的近似定价 (USD per 1M tokens)，仅供估算
|
||
DEFAULT_PRICING: Dict[str, Dict[str, float]] = {
|
||
# OpenAI GPT-4o
|
||
"openai": {"prompt": 5.0, "completion": 15.0, "image": 0.005},
|
||
# 豆包 / Qwen / GLM 国内 VLM 大致价位（仅估算）
|
||
"qwen": {"prompt": 0.8, "completion": 2.0, "image": 0.001},
|
||
"glm": {"prompt": 1.0, "completion": 3.0, "image": 0.001},
|
||
"doubao": {"prompt": 0.8, "completion": 2.0, "image": 0.001},
|
||
# mock provider 不计费
|
||
"mock": {"prompt": 0.0, "completion": 0.0, "image": 0.0},
|
||
# 兜底
|
||
"default": {"prompt": 1.0, "completion": 3.0, "image": 0.002},
|
||
}
|
||
|
||
|
||
class CircuitState:
|
||
CLOSED = "closed" # 正常
|
||
OPEN = "open" # 熔断
|
||
HALF_OPEN = "half_open" # 试探恢复
|
||
|
||
|
||
@dataclass
|
||
class CallRecord:
|
||
"""单次 LLM 调用记录。"""
|
||
|
||
timestamp: float
|
||
provider: str
|
||
model: str
|
||
success: bool
|
||
confirmed: Optional[bool]
|
||
latency_ms: float
|
||
prompt_tokens: int
|
||
completion_tokens: int
|
||
image_count: int
|
||
cost_usd: float
|
||
error: Optional[str] = None
|
||
|
||
|
||
@dataclass
|
||
class _DailyBucket:
|
||
"""按 UTC 日历日聚合的桶。"""
|
||
|
||
date: str
|
||
call_count: int = 0
|
||
success_count: int = 0
|
||
failure_count: int = 0
|
||
total_prompt_tokens: int = 0
|
||
total_completion_tokens: int = 0
|
||
total_image_count: int = 0
|
||
total_cost_usd: float = 0.0
|
||
total_latency_ms: float = 0.0
|
||
|
||
def add(self, record: CallRecord) -> None:
|
||
self.call_count += 1
|
||
if record.success:
|
||
self.success_count += 1
|
||
else:
|
||
self.failure_count += 1
|
||
self.total_prompt_tokens += record.prompt_tokens
|
||
self.total_completion_tokens += record.completion_tokens
|
||
self.total_image_count += record.image_count
|
||
self.total_cost_usd += record.cost_usd
|
||
self.total_latency_ms += record.latency_ms
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {
|
||
"date": self.date,
|
||
"call_count": self.call_count,
|
||
"success_count": self.success_count,
|
||
"failure_count": self.failure_count,
|
||
"prompt_tokens": self.total_prompt_tokens,
|
||
"completion_tokens": self.total_completion_tokens,
|
||
"image_count": self.total_image_count,
|
||
"cost_usd": round(self.total_cost_usd, 6),
|
||
"avg_latency_ms": round(
|
||
self.total_latency_ms / self.call_count, 2
|
||
)
|
||
if self.call_count
|
||
else 0.0,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# LLMCostTracker
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class LLMCostTracker:
|
||
"""LLM 成本追踪器 + 熔断降级。
|
||
|
||
Args:
|
||
daily_budget_usd: 日预算，0 表示不限
|
||
history_days: 保留多少天的日级统计
|
||
recent_window: 用于错误率计算的最近调用窗口
|
||
error_rate_threshold: 触发熔断的错误率阈值 [0, 1]
|
||
cooldown_seconds: 熔断进入半开状态的冷却时间
|
||
pricing_overrides: 自定义价格表 (覆盖默认)
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
daily_budget_usd: float = 0.0,
|
||
history_days: int = 7,
|
||
recent_window: int = 20,
|
||
error_rate_threshold: float = 0.5,
|
||
cooldown_seconds: float = 60.0,
|
||
pricing_overrides: Optional[Dict[str, Dict[str, float]]] = None,
|
||
) -> None:
|
||
if daily_budget_usd < 0:
|
||
raise ValueError("daily_budget_usd 必须 >= 0")
|
||
if history_days < 1:
|
||
raise ValueError("history_days 必须 >= 1")
|
||
if recent_window < 1:
|
||
raise ValueError("recent_window 必须 >= 1")
|
||
if not 0.0 < error_rate_threshold <= 1.0:
|
||
raise ValueError("error_rate_threshold 必须在 (0, 1]")
|
||
if cooldown_seconds < 0:
|
||
raise ValueError("cooldown_seconds 必须 >= 0")
|
||
|
||
self.daily_budget_usd = daily_budget_usd
|
||
self.history_days = history_days
|
||
self.recent_window = recent_window
|
||
self.error_rate_threshold = error_rate_threshold
|
||
self.cooldown_seconds = cooldown_seconds
|
||
|
||
# 合并定价
|
||
self._pricing: Dict[str, Dict[str, float]] = {
|
||
k: dict(v) for k, v in DEFAULT_PRICING.items()
|
||
}
|
||
if pricing_overrides:
|
||
for provider, table in pricing_overrides.items():
|
||
self._pricing.setdefault(provider, {})
|
||
self._pricing[provider].update(table)
|
||
|
||
# 状态
|
||
self._buckets: Deque[_DailyBucket] = deque(maxlen=history_days)
|
||
self._recent: Deque[bool] = deque(maxlen=recent_window)
|
||
self._records: Deque[CallRecord] = deque(maxlen=200)
|
||
self._circuit_state = CircuitState.CLOSED
|
||
self._circuit_opened_at: Optional[float] = None
|
||
self._manually_disabled = False
|
||
self._lock = asyncio.Lock()
|
||
|
||
# ------------------------------------------------------------------
|
||
# 准入检查
|
||
# ------------------------------------------------------------------
|
||
|
||
def can_call(self, now: Optional[float] = None) -> bool:
|
||
"""同步判断当前是否允许调用 LLM。
|
||
|
||
触发拒绝的条件按优先级:
|
||
1. 手动禁用
|
||
2. 日预算超限
|
||
3. 熔断器处于 OPEN 且未到冷却结束
|
||
"""
|
||
|
||
if self._manually_disabled:
|
||
return False
|
||
if self._budget_exhausted():
|
||
return False
|
||
|
||
ts = now if now is not None else time.time()
|
||
if self._circuit_state == CircuitState.OPEN:
|
||
if (
|
||
self._circuit_opened_at is not None
|
||
and (ts - self._circuit_opened_at) >= self.cooldown_seconds
|
||
):
|
||
# 进入半开，允许一次试探
|
||
self._circuit_state = CircuitState.HALF_OPEN
|
||
logger.info("LLMCostTracker 熔断器进入 HALF_OPEN")
|
||
return True
|
||
return False
|
||
return True
|
||
|
||
def reason_for_block(self, now: Optional[float] = None) -> Optional[str]:
|
||
"""返回当前不可调用的原因 (调试用)，None 表示可以调用。"""
|
||
|
||
if self._manually_disabled:
|
||
return "manually_disabled"
|
||
if self._budget_exhausted():
|
||
return f"daily_budget_exceeded({self._today_cost():.4f}/{self.daily_budget_usd})"
|
||
ts = now if now is not None else time.time()
|
||
if self._circuit_state == CircuitState.OPEN:
|
||
if (
|
||
self._circuit_opened_at is not None
|
||
and (ts - self._circuit_opened_at) < self.cooldown_seconds
|
||
):
|
||
return "circuit_open"
|
||
return None
|
||
|
||
# ------------------------------------------------------------------
|
||
# 记录
|
||
# ------------------------------------------------------------------
|
||
|
||
def record(
|
||
self,
|
||
result: Any,
|
||
prompt_tokens: Optional[int] = None,
|
||
completion_tokens: Optional[int] = None,
|
||
image_count: int = 1,
|
||
provider_override: Optional[str] = None,
|
||
) -> CallRecord:
|
||
"""记录一次 LLM 调用。
|
||
|
||
``result`` 可以是 ``LLMAnalysisResult`` 实例或鸭子类型 (含 confirmed
|
||
/ provider / model / latency_ms / error / metadata)。
|
||
"""
|
||
|
||
provider = provider_override or getattr(result, "provider", "default")
|
||
model = getattr(result, "model", "")
|
||
success = getattr(result, "error", None) is None and getattr(
|
||
result, "confirmed", None
|
||
) is not None
|
||
latency_ms = float(getattr(result, "latency_ms", 0.0) or 0.0)
|
||
|
||
usage = self._extract_usage(result)
|
||
prompt_tokens = (
|
||
int(prompt_tokens)
|
||
if prompt_tokens is not None
|
||
else int(usage.get("prompt_tokens") or 0)
|
||
)
|
||
completion_tokens = (
|
||
int(completion_tokens)
|
||
if completion_tokens is not None
|
||
else int(usage.get("completion_tokens") or 0)
|
||
)
|
||
|
||
# token 估算兜底
|
||
if prompt_tokens == 0 and completion_tokens == 0 and provider != "mock":
|
||
prompt_tokens = self._estimate_prompt_tokens(image_count)
|
||
completion_tokens = self._estimate_completion_tokens(result)
|
||
|
||
cost = self._calc_cost(provider, prompt_tokens, completion_tokens, image_count)
|
||
|
||
record = CallRecord(
|
||
timestamp=time.time(),
|
||
provider=provider,
|
||
model=model,
|
||
success=success,
|
||
confirmed=getattr(result, "confirmed", None),
|
||
latency_ms=latency_ms,
|
||
prompt_tokens=prompt_tokens,
|
||
completion_tokens=completion_tokens,
|
||
image_count=image_count,
|
||
cost_usd=cost,
|
||
error=getattr(result, "error", None),
|
||
)
|
||
|
||
self._append_record(record)
|
||
self._update_circuit(record)
|
||
return record
|
||
|
||
# ------------------------------------------------------------------
|
||
# 控制
|
||
# ------------------------------------------------------------------
|
||
|
||
def disable(self) -> None:
|
||
"""手动禁用 LLM 调用 (运维场景)。"""
|
||
|
||
self._manually_disabled = True
|
||
logger.warning("LLMCostTracker 已被手动禁用")
|
||
|
||
def enable(self) -> None:
|
||
"""手动重新启用。"""
|
||
|
||
self._manually_disabled = False
|
||
self._circuit_state = CircuitState.CLOSED
|
||
self._circuit_opened_at = None
|
||
logger.info("LLMCostTracker 已重新启用")
|
||
|
||
def reset(self) -> None:
|
||
"""清空全部统计 (谨慎使用)。"""
|
||
|
||
self._buckets.clear()
|
||
self._recent.clear()
|
||
self._records.clear()
|
||
self._circuit_state = CircuitState.CLOSED
|
||
self._circuit_opened_at = None
|
||
|
||
# ------------------------------------------------------------------
|
||
# 导出
|
||
# ------------------------------------------------------------------
|
||
|
||
@property
|
||
def is_enabled(self) -> bool:
|
||
return not self._manually_disabled
|
||
|
||
@property
|
||
def circuit_state(self) -> str:
|
||
return self._circuit_state
|
||
|
||
def daily_summary(self) -> List[Dict[str, Any]]:
|
||
return [bucket.to_dict() for bucket in self._buckets]
|
||
|
||
def today_summary(self) -> Dict[str, Any]:
|
||
bucket = self._today_bucket(create=False)
|
||
if bucket is None:
|
||
return {
|
||
"date": _today_str(),
|
||
"call_count": 0,
|
||
"cost_usd": 0.0,
|
||
"budget_usd": self.daily_budget_usd,
|
||
"budget_used_ratio": 0.0,
|
||
}
|
||
data = bucket.to_dict()
|
||
data["budget_usd"] = self.daily_budget_usd
|
||
data["budget_used_ratio"] = (
|
||
data["cost_usd"] / self.daily_budget_usd
|
||
if self.daily_budget_usd > 0
|
||
else 0.0
|
||
)
|
||
return data
|
||
|
||
def recent_records(self, limit: int = 20) -> List[Dict[str, Any]]:
|
||
records = list(self._records)[-limit:]
|
||
return [
|
||
{
|
||
"timestamp": r.timestamp,
|
||
"provider": r.provider,
|
||
"model": r.model,
|
||
"success": r.success,
|
||
"confirmed": r.confirmed,
|
||
"latency_ms": round(r.latency_ms, 2),
|
||
"prompt_tokens": r.prompt_tokens,
|
||
"completion_tokens": r.completion_tokens,
|
||
"image_count": r.image_count,
|
||
"cost_usd": round(r.cost_usd, 6),
|
||
"error": r.error,
|
||
}
|
||
for r in reversed(records)
|
||
]
|
||
|
||
def stats(self) -> Dict[str, Any]:
|
||
recent_total = len(self._recent)
|
||
recent_failed = sum(1 for ok in self._recent if not ok)
|
||
recent_error_rate = (
|
||
recent_failed / recent_total if recent_total > 0 else 0.0
|
||
)
|
||
|
||
return {
|
||
"enabled": self.is_enabled,
|
||
"circuit_state": self._circuit_state,
|
||
"circuit_opened_at": self._circuit_opened_at,
|
||
"block_reason": self.reason_for_block(),
|
||
"today": self.today_summary(),
|
||
"history_days": self.history_days,
|
||
"recent_window": self.recent_window,
|
||
"recent_error_rate": round(recent_error_rate, 4),
|
||
"error_rate_threshold": self.error_rate_threshold,
|
||
}
|
||
|
||
# ------------------------------------------------------------------
|
||
# 内部
|
||
# ------------------------------------------------------------------
|
||
|
||
def _append_record(self, record: CallRecord) -> None:
|
||
bucket = self._today_bucket(create=True)
|
||
assert bucket is not None
|
||
bucket.add(record)
|
||
self._recent.append(record.success)
|
||
self._records.append(record)
|
||
|
||
def _today_bucket(self, *, create: bool) -> Optional[_DailyBucket]:
|
||
today = _today_str()
|
||
if self._buckets and self._buckets[-1].date == today:
|
||
return self._buckets[-1]
|
||
if not create:
|
||
for b in self._buckets:
|
||
if b.date == today:
|
||
return b
|
||
return None
|
||
bucket = _DailyBucket(date=today)
|
||
self._buckets.append(bucket)
|
||
return bucket
|
||
|
||
def _today_cost(self) -> float:
|
||
bucket = self._today_bucket(create=False)
|
||
return bucket.total_cost_usd if bucket else 0.0
|
||
|
||
def _budget_exhausted(self) -> bool:
|
||
if self.daily_budget_usd <= 0:
|
||
return False
|
||
return self._today_cost() >= self.daily_budget_usd
|
||
|
||
def _update_circuit(self, record: CallRecord) -> None:
|
||
# HALF_OPEN: 一次结果决定回到 CLOSED 还是 OPEN
|
||
if self._circuit_state == CircuitState.HALF_OPEN:
|
||
if record.success:
|
||
self._circuit_state = CircuitState.CLOSED
|
||
self._circuit_opened_at = None
|
||
logger.info("LLMCostTracker 熔断器恢复 CLOSED")
|
||
else:
|
||
self._circuit_state = CircuitState.OPEN
|
||
self._circuit_opened_at = time.time()
|
||
logger.warning("LLMCostTracker 半开试探失败，回到 OPEN")
|
||
return
|
||
|
||
# CLOSED: 检查最近窗口的错误率
|
||
if (
|
||
self._circuit_state == CircuitState.CLOSED
|
||
and len(self._recent) >= max(3, self.recent_window // 2)
|
||
):
|
||
failures = sum(1 for ok in self._recent if not ok)
|
||
error_rate = failures / len(self._recent)
|
||
if error_rate >= self.error_rate_threshold:
|
||
self._circuit_state = CircuitState.OPEN
|
||
self._circuit_opened_at = time.time()
|
||
logger.warning(
|
||
"LLMCostTracker 触发熔断: 最近 %d 次调用错误率 %.2f >= %.2f",
|
||
len(self._recent),
|
||
error_rate,
|
||
self.error_rate_threshold,
|
||
)
|
||
|
||
def _calc_cost(
|
||
self,
|
||
provider: str,
|
||
prompt_tokens: int,
|
||
completion_tokens: int,
|
||
image_count: int,
|
||
) -> float:
|
||
rates = self._pricing.get(provider) or self._pricing["default"]
|
||
cost = (
|
||
prompt_tokens / 1_000_000.0 * rates.get("prompt", 0.0)
|
||
+ completion_tokens / 1_000_000.0 * rates.get("completion", 0.0)
|
||
+ image_count * rates.get("image", 0.0)
|
||
)
|
||
return max(0.0, cost)
|
||
|
||
@staticmethod
|
||
def _estimate_prompt_tokens(image_count: int) -> int:
|
||
# 经验值：每张图 ~ 200 token，prompt 模板 ~ 120 token
|
||
return 120 + 200 * max(0, image_count)
|
||
|
||
@staticmethod
|
||
def _estimate_completion_tokens(result: Any) -> int:
|
||
reasoning = getattr(result, "reasoning", "") or ""
|
||
# 粗略 1.3 字符 = 1 token
|
||
return max(16, int(len(reasoning) / 1.3))
|
||
|
||
@staticmethod
|
||
def _extract_usage(result: Any) -> Dict[str, Any]:
|
||
"""从结果对象上尽力获取 usage (provider 可能放 metadata 里)。"""
|
||
|
||
usage = getattr(result, "usage", None)
|
||
if isinstance(usage, dict):
|
||
return usage
|
||
metadata = getattr(result, "metadata", None)
|
||
if isinstance(metadata, dict):
|
||
md_usage = metadata.get("usage")
|
||
if isinstance(md_usage, dict):
|
||
return md_usage
|
||
return {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 全局单例
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
_global_tracker: Optional[LLMCostTracker] = None
|
||
|
||
|
||
def init_global_tracker(
|
||
daily_budget_usd: float = 0.0,
|
||
history_days: int = 7,
|
||
recent_window: int = 20,
|
||
error_rate_threshold: float = 0.5,
|
||
cooldown_seconds: float = 60.0,
|
||
) -> LLMCostTracker:
|
||
"""初始化并返回全局成本追踪器。"""
|
||
|
||
global _global_tracker
|
||
_global_tracker = LLMCostTracker(
|
||
daily_budget_usd=daily_budget_usd,
|
||
history_days=history_days,
|
||
recent_window=recent_window,
|
||
error_rate_threshold=error_rate_threshold,
|
||
cooldown_seconds=cooldown_seconds,
|
||
)
|
||
return _global_tracker
|
||
|
||
|
||
def get_global_tracker() -> Optional[LLMCostTracker]:
|
||
return _global_tracker
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 工具
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _today_str() -> str:
|
||
return datetime.now(tz=timezone.utc).strftime("%Y-%m-%d")
|
||
|
||
|
||
__all__ = [
|
||
"LLMCostTracker",
|
||
"CallRecord",
|
||
"CircuitState",
|
||
"DEFAULT_PRICING",
|
||
"init_global_tracker",
|
||
"get_global_tracker",
|
||
]
|