From 7e59455361a590ec773fdc29cfc896a386743548 Mon Sep 17 00:00:00 2001 From: wuzhuorong <973204353@qq.com> Date: Tue, 16 Jun 2026 10:58:01 +0800 Subject: [PATCH] =?UTF-8?q?feat(server):=20=E6=96=B0=E5=A2=9E=20LLM=20?= =?UTF-8?q?=E6=88=90=E6=9C=AC=E8=BF=BD=E8=B8=AA=E4=B8=8E=E7=86=94=E6=96=AD?= =?UTF-8?q?=E9=99=8D=E7=BA=A7=E6=9C=BA=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/server/services/llm_cost_tracker.py | 555 +++++++++++++++++++++++ 1 file changed, 555 insertions(+) create mode 100644 apps/server/services/llm_cost_tracker.py diff --git a/apps/server/services/llm_cost_tracker.py b/apps/server/services/llm_cost_tracker.py new file mode 100644 index 0000000..b8acb67 --- /dev/null +++ b/apps/server/services/llm_cost_tracker.py @@ -0,0 +1,555 @@ +"""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", +]