""" title: LiteLLM Dashboard author: Kagamine Rin & VAI author_url: https://rin.captcouch.xyz description: High-performance LiteLLM observability — Comprehensive health, cost breakdowns, and request history. required_open_webui_version: 0.4.0 requirements: requests version: 1.6.0 licence: MIT """ import requests import json from datetime import datetime, timedelta from zoneinfo import ZoneInfo from pydantic import BaseModel, Field from typing import Optional, List, Dict, Any class Tools: def __init__(self): self.valves = self.Valves() self._price_cache = {} class Valves(BaseModel): """LiteLLM proxy connection settings.""" litellm_url: str = Field( default="http://localhost:4000", description="LiteLLM proxy base URL (e.g. http://localhost:4000)", ) litellm_master_key: str = Field( default="", description="LiteLLM master API key for authentication", json_schema_extra={"input": {"type": "password"}}, ) default_hide_zero_spend: bool = Field( default=True, description="Default behavior for zero-spend model filtering. When True, models with sub-$0.001 spend are hidden by default.", ) timezone: str = Field( default="UTC", description="IANA Timezone name (e.g. America/New_York) for date resolution.", ) enable_dynamic_pricing: bool = Field( default=True, description="When True, attempts to fetch live pricing from /v1/model/info for the Audit Engine.", ) def _headers(self) -> dict: if not self.valves.litellm_master_key: return {} return {"Authorization": f"Bearer {self.valves.litellm_master_key}"} def _base_url(self) -> str: return self.valves.litellm_url.rstrip("/") def _fetch_live_prices(self): """Sync internal price cache with LiteLLM's live model registry.""" if not self.valves.enable_dynamic_pricing or self._price_cache: return try: url = f"{self._base_url()}/v1/model/info" resp = requests.get(url, headers=self._headers(), timeout=10) if resp.status_code == 200: data = resp.json().get("data", []) for m in data: name = m.get("model_name") info = m.get("model_info", {}) if name and info: self._price_cache[name] = { "input": info.get("input_cost_per_token", 0), "input_cached": info.get("cache_read_input_token_cost", 0), "output": info.get("output_cost_per_token", 0), "reasoning": info.get("output_cost_per_reasoning_token", info.get("output_cost_per_token", 0)) } except Exception: pass def _resolve_date(self, date_str: str) -> str: date_str = date_str.strip().lower() try: today = datetime.now(ZoneInfo(self.valves.timezone)).date() except Exception: today = datetime.utcnow().date() if date_str == "today": return today.isoformat() if date_str == "yesterday": return (today - timedelta(days=1)).isoformat() if "days ago" in date_str: try: n = int(date_str.split()[0]) return (today - timedelta(days=n)).isoformat() except Exception: pass # Day names days = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"] if date_str in days: target_day = days.index(date_str) days_back = (today.weekday() - target_day) % 7 return (today - timedelta(days=days_back)).isoformat() try: datetime.strptime(date_str, "%Y-%m-%d") return date_str except ValueError: pass return date_str def _audit_cost(self, model: str, prompt: int, cached: int, completion: int, reported_cost: float, reasoning: int = 0) -> str: """Audit reported cost against dynamic pricing.""" self._fetch_live_prices() pricing = self._price_cache.get(model) if not pricing: return "" uncached = max(0, prompt - cached) actual_output = max(0, completion - reasoning) calc = (cached * (pricing.get("input_cached") or 0)) + \ (uncached * (pricing.get("input") or 0)) + \ (actual_output * (pricing.get("output") or 0)) + \ (reasoning * (pricing.get("reasoning") or pricing.get("output") or 0)) delta = reported_cost - calc symbol = "⚠️" if abs(delta) > 0.005 else "✓" return f" [calc: ${calc:.4f} {symbol}]" def litellm_health(self) -> str: """ Check LiteLLM proxy health and authentication status. Returns detailed model health, rate limits, and infrastructure readiness. """ lines = [f"LiteLLM Health Check — {self._base_url()}"] lines.append("=" * 60) # Readiness Header try: r = requests.get(f"{self._base_url()}/health/readiness", timeout=5) if r.status_code == 200: d = r.json() lines.append(f"✅ Proxy: {d.get('status')} | DB: {d.get('db')} | v{d.get('litellm_version')}") except Exception: pass if not self.valves.litellm_master_key: lines.append("⚠️ No master key configured — showing public health only") else: lines.append("✅ Authenticated with master key") try: url = f"{self._base_url()}/health" resp = requests.get(url, headers=self._headers(), timeout=15) if resp.status_code != 200: return f"❌ Unexpected status {resp.status_code}: {resp.text[:500]}" data = resp.json() healthy = data.get("healthy_endpoints", []) unhealthy = data.get("unhealthy_endpoints", []) lines.append(f"\nHealthy: {len(healthy)} | Unhealthy: {len(unhealthy)}") if healthy: lines.append(f"\n--- Healthy Models ({len(healthy)}) ---") for ep in healthy: model = ep.get("model", "unknown") display = model.replace("gemini/", "").replace("openai/", "") rpm = ep.get("rpm", "?") timeout = ep.get("timeout", "?") lines.append(f" ✅ {display:30s} — rpm: {rpm:>3}, timeout: {timeout}s") if unhealthy: lines.append(f"\n--- Unhealthy Models ({len(unhealthy)}) ---") for ep in unhealthy: lines.append(f" ❌ {ep.get('model', 'unknown')}") except Exception as e: return f"❌ Connection failed: {str(e)}" return "\n".join(lines) def litellm_cost(self, start_date: str, end_date: str, hide_zero_spend: bool = False) -> str: """ Get LiteLLM cost breakdown for a date range. Returns per-model and per-day spend, token counts, cache hit rate, and trend sparklines. """ if not self.valves.litellm_master_key: return "❌ Master key required for cost data." effective_hide = hide_zero_spend or self.valves.default_hide_zero_spend r_start = self._resolve_date(start_date) r_end = self._resolve_date(end_date) url = f"{self._base_url()}/user/daily/activity" try: resp = requests.get(url, headers=self._headers(), params={"start_date": r_start, "end_date": r_end}, timeout=30) if resp.status_code != 200: return f"❌ Unexpected status {resp.status_code}" data = resp.json() except Exception as e: return f"❌ Connection failed: {str(e)}" results = data.get("results", []) metadata = data.get("metadata", {}) if not results: return f"No cost data found for {r_start} to {r_end}." # Totals total_spend = metadata.get("total_spend", 0) total_prompt = metadata.get("total_prompt_tokens", 0) total_completion = metadata.get("total_completion_tokens", 0) total_tokens = metadata.get("total_tokens", 0) cache_read = metadata.get("total_cache_read_input_tokens", 0) # Sparkline spends = [day.get("metrics", {}).get("spend", 0) for day in results] def get_spark(vals): if len(vals) < 2: return "" chars = " ▂▃▄▅▆▇█" low, high = min(vals), max(vals) rng = (high - low) or 1 return " [" + "".join(chars[int((v - low) / rng * 7)] for v in vals) + "]" lines = [f"LiteLLM Cost Report: {r_start} to {r_end}"] lines.append("=" * 75) lines.append(f"Total Spend: ${total_spend:.2f}{get_spark(spends)}") lines.append(f"Total Tokens: {total_tokens:,} ({total_prompt:,} prompt + {total_completion:,} completion)") if total_prompt > 0 and cache_read > 0: hit_rate = (cache_read / total_prompt) * 100 lines.append(f"Cache Hit Rate: {hit_rate:.1f}% ({cache_read:,} / {total_prompt:,} input tokens)") lines.append(f"Requests: {metadata.get('total_successful_requests', 0)} ok, {metadata.get('total_failed_requests', 0)} fail") # Model Aggregation model_totals = {} for day in results: models = day.get("breakdown", {}).get("models", {}) for m_name, m_data in models.items(): m_metrics = m_data.get("metrics", {}) if m_name not in model_totals: model_totals[m_name] = {"spend": 0, "reqs": 0, "tokens": 0, "cache": 0} model_totals[m_name]["spend"] += m_metrics.get("spend", 0) model_totals[m_name]["reqs"] += m_metrics.get("successful_requests", 0) model_totals[m_name]["tokens"] += m_metrics.get("total_tokens", 0) model_totals[m_name]["cache"] += m_metrics.get("cache_read_input_tokens", 0) lines.append(f"\n{'Model':30s} | {'Spend':>10s} | {'Requests':>14s} | {'Tokens':>12s}") lines.append("-" * 75) hidden_count = 0 sorted_models = sorted(model_totals.items(), key=lambda x: x[1]["spend"], reverse=True) for name, mt in sorted_models: if effective_hide and mt["spend"] < 0.001: hidden_count += 1 continue display = name.replace("gemini/", "").replace("openai/", "")[:30] cache_info = "" if mt["cache"] > 0: hit = (mt["cache"] / (mt["tokens"] or 1) * 100) cache_info = f" 💾{hit:.0f}%" lines.append(f" {display:28s} | ${mt['spend']:9.2f} | {mt['reqs']:6d} ok | {mt['tokens']:10,d}{cache_info}") if hidden_count: lines.append(f" ... ({hidden_count} model(s) with sub-$0.001 spend hidden)") # Date breakdown lines.append("\nBy Date:") for day in results: m = day.get("metrics", {}) lines.append(f" {day.get('date')}: ${m.get('spend',0):.2f} | {m.get('successful_requests',0)} ok | {m.get('total_tokens',0):,} tokens") return "\n".join(lines) def litellm_history(self, limit: int = 20, show_failed: bool = True) -> str: """ Get recent LiteLLM request history with audit data and thought detection. Returns a high-density table of recent requests. """ if not self.valves.litellm_master_key: return "❌ Master key required for request history." try: resp = requests.get(f"{self._base_url()}/spend/logs", headers=self._headers(), timeout=30) if resp.status_code != 200: return f"❌ Unexpected status {resp.status_code}" logs = sorted(resp.json(), key=lambda x: x.get("startTime", ""), reverse=True) except Exception as e: return f"❌ Connection failed: {str(e)}" if not show_failed: logs = [l for l in logs if not l.get("metadata", {}).get("exception")] recent = logs[:max(1, min(50, limit))] lines = [f"LiteLLM Request History — {self._base_url()}"] lines.append(f"Showing {len(recent)} of {len(logs)} total requests") lines.append("=" * 105) for i, entry in enumerate(recent, 1): start_time = entry.get("startTime", "?")[5:16] # MM-DD HH:MM model = entry.get("model", "unknown").split("/")[-1][:28] spend = entry.get("spend", 0) prompt = entry.get("prompt_tokens", 0) or 0 completion = entry.get("completion_tokens", 0) or 0 # Metadata & Status meta = entry.get("metadata", {}) or {} is_failed = bool(meta.get("exception")) or meta.get("status") == "failure" status = "❌ FAIL" if is_failed else "✅ OK" # Advanced Tokens usage = entry.get("usage", {}) or {} thought = usage.get("completion_tokens_details", {}).get("reasoning_tokens", 0) thought_str = f" [💭{thought}]" if thought else "" cached = entry.get("cache_read_input_tokens", 0) or 0 cache_str = f" 💾{int(cached/prompt*100)}%" if prompt and cached else "" # Format numbers p_s = f"{prompt/1000:.1f}k" if prompt >= 1000 else str(prompt) c_s = f"{completion/1000:.1f}k" if completion >= 1000 else str(completion) audit = self._audit_cost(entry.get("model"), prompt, cached, completion, spend, reasoning=thought) line = f" {i:2d}. {start_time} | {model:28s} | {status} | ${spend:.4f}{audit:18s} | {p_s:>6s}→{c_s:<6s}{thought_str}{cache_str}" lines.append(line) if is_failed: err = meta.get("exception", "unknown error") lines.append(f" └─ {err[:80]}") return "\n".join(lines)