335 lines
16 KiB
Python
335 lines
16 KiB
Python
"""
|
|
title: LiteLLM Dashboard
|
|
author: Kagamine Rin & VAI
|
|
author_url: https://rin.captcouch.xyz
|
|
description: Professional LiteLLM observability — Grouped health, wildcard resolution, and Billion-scale token tracking.
|
|
required_open_webui_version: 0.4.0
|
|
requirements: requests
|
|
version: 1.6.2
|
|
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 = {}
|
|
self._model_info_cache = []
|
|
|
|
class Valves(BaseModel):
|
|
"""LiteLLM proxy connection settings."""
|
|
litellm_url: str = Field(
|
|
default="http://localhost:4000",
|
|
description="LiteLLM proxy base URL",
|
|
)
|
|
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.",
|
|
)
|
|
timezone: str = Field(
|
|
default="UTC",
|
|
description="IANA Timezone name for date resolution.",
|
|
)
|
|
enable_dynamic_pricing: bool = Field(
|
|
default=True,
|
|
description="When True, attempts to fetch live pricing from /v1/model/info.",
|
|
)
|
|
|
|
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 _compact_name(self, name: str, max_len: int = 28) -> str:
|
|
"""Strip redundant provider prefixes and truncate carefully."""
|
|
prefixes = ["gemini/", "openai/", "ollama-cloud/", "anthropic/", "mistral/", "azure/"]
|
|
for p in prefixes:
|
|
if name.startswith(p):
|
|
name = name[len(p):]
|
|
break
|
|
return (name[:max_len-1] + "…") if len(name) > max_len else name
|
|
|
|
def _format_tokens(self, n: int) -> str:
|
|
"""Scale tokens to human-readable units (k, M, B)."""
|
|
if n >= 1_000_000_000:
|
|
return f"{n / 1_000_000_000:.2f}B"
|
|
if n >= 1_000_000:
|
|
return f"{n / 1_000_000:.2f}M"
|
|
if n >= 1_000:
|
|
return f"{n / 1_000:.1f}k"
|
|
return str(n)
|
|
|
|
def _fetch_live_data(self):
|
|
"""Sync internal price and model cache."""
|
|
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:
|
|
self._model_info_cache = resp.json().get("data", [])
|
|
for m in self._model_info_cache:
|
|
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)),
|
|
"search_fee": info.get("search_context_cost_per_query", {}).get("search_context_size_low", 0)
|
|
}
|
|
except Exception: pass
|
|
|
|
def _audit_cost(self, model: str, prompt: int, cached: int, completion: int, reported_cost: float, reasoning: int = 0, has_search: bool = False) -> str:
|
|
self._fetch_live_data()
|
|
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))
|
|
|
|
if has_search:
|
|
calc += pricing.get("search_fee", 0)
|
|
|
|
delta = reported_cost - calc
|
|
symbol = "⚠️" if abs(delta) > 0.005 else "✓"
|
|
return f" [calc: ${calc:.4f} {symbol}]"
|
|
|
|
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
|
|
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 litellm_health(self) -> str:
|
|
"""Check LiteLLM proxy health, proactive budget, and grouped model readiness."""
|
|
self._fetch_live_data()
|
|
lines = [f"LiteLLM Health & Budget — {self._base_url()}"]
|
|
lines.append("=" * 70)
|
|
|
|
# 1. Budget Sentinel
|
|
if self.valves.litellm_master_key:
|
|
try:
|
|
r = requests.get(f"{self._base_url()}/spend/keys", headers=self._headers(), timeout=10)
|
|
if r.status_code == 200:
|
|
for k in r.json():
|
|
spend, limit = k.get("spend", 0), k.get("max_budget")
|
|
if limit:
|
|
pct = (spend / limit) * 100
|
|
alert = "🚨 [CRITICAL]" if pct > 90 else "💳"
|
|
lines.append(f"{alert} Key: {k.get('key_name')} | Spend: ${spend:.3f} / ${limit:.2f} ({pct:.1f}%)")
|
|
except Exception: pass
|
|
|
|
# 2. Infrastructure Readiness
|
|
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
|
|
|
|
# 3. Model Fleet (Grouped by Provider + Wildcard Linkage)
|
|
try:
|
|
url = f"{self._base_url()}/health"
|
|
resp = requests.get(url, headers=self._headers(), timeout=15)
|
|
if resp.status_code == 200:
|
|
hd = resp.json()
|
|
healthy = hd.get("healthy_endpoints", [])
|
|
unhealthy = hd.get("unhealthy_endpoints", [])
|
|
|
|
groups = {}
|
|
for ep in healthy + unhealthy:
|
|
m_raw = ep.get("model", "unknown")
|
|
m_display = self._compact_name(m_raw, 32)
|
|
|
|
# Wildcard Linkage
|
|
found_alias = False
|
|
for info in self._model_info_cache:
|
|
if info.get("model_info", {}).get("health_check_model") == m_raw:
|
|
alias = info.get("model_name", "unknown")
|
|
m_display = f"{self._compact_name(alias, 28)} (via {m_display})"
|
|
found_alias = True
|
|
break
|
|
|
|
if not found_alias and "*" in m_raw:
|
|
meta = ep.get("litellm_metadata", {}) or {}
|
|
probe = meta.get("health_check_model", "generic")
|
|
m_display = f"{m_display} (via {probe.split('/')[-1]})"
|
|
|
|
# Provider Determination
|
|
provider = "other"
|
|
if "ollama" in ep.get("api_base", "").lower() or "ollama" in m_raw.lower():
|
|
provider = "ollama"
|
|
elif "/" in m_raw:
|
|
provider = m_raw.split("/")[0]
|
|
|
|
if provider not in groups: groups[provider] = {"ok": [], "fail": []}
|
|
|
|
icon = "✅" if ep in healthy else "❌"
|
|
info_line = f" {icon} {m_display:40s} | rpm: {ep.get('rpm','?'):>3}"
|
|
if ep in healthy: groups[provider]["ok"].append(info_line)
|
|
else: groups[provider]["fail"].append(info_line)
|
|
|
|
lines.append(f"\nFleet Status: {len(healthy)} healthy, {len(unhealthy)} unhealthy")
|
|
for prov in sorted(groups.keys()):
|
|
lines.append(f"\n--- {prov.upper()} ---")
|
|
lines.extend(groups[prov]["ok"] + groups[prov]["fail"])
|
|
else: lines.append(f"\n❌ Fleet Check Failed: Status {resp.status_code}")
|
|
except Exception as e: lines.append(f"\n❌ Fleet Check Failed: {str(e)}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
def litellm_cost(self, start_date: str, end_date: str, hide_zero_spend: bool = False) -> str:
|
|
"""Comprehensive cost report with Billion-scale token tracking and sparklines."""
|
|
if not self.valves.litellm_master_key: return "❌ Master key required."
|
|
effective_hide = hide_zero_spend or self.valves.default_hide_zero_spend
|
|
r_start, r_end = self._resolve_date(start_date), self._resolve_date(end_date)
|
|
|
|
try:
|
|
url = f"{self._base_url()}/user/daily/activity"
|
|
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"❌ Error {resp.status_code}"
|
|
data = resp.json()
|
|
except Exception as e: return f"❌ 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}."
|
|
|
|
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("=" * 85)
|
|
lines.append(f"Total Spend: ${metadata.get('total_spend', 0):.2f}{get_spark(spends)}")
|
|
|
|
t_in, t_out = metadata.get('total_prompt_tokens',0), metadata.get('total_completion_tokens',0)
|
|
lines.append(f"Total Tokens: {self._format_tokens(metadata.get('total_tokens', 0))} ({self._format_tokens(t_in)} in + {self._format_tokens(t_out)} out)")
|
|
|
|
c_read = metadata.get("total_cache_read_input_tokens", 0)
|
|
if t_in > 0 and c_read > 0:
|
|
lines.append(f"Global Cache: {(c_read/t_in*100):.1f}% ({self._format_tokens(c_read)} saved)")
|
|
|
|
lines.append(f"Total Requests: {metadata.get('total_successful_requests', 0)} ok, {metadata.get('total_failed_requests', 0)} fail")
|
|
|
|
# Model Table
|
|
model_totals = {}
|
|
for day in results:
|
|
for m_name, m_data in day.get("breakdown", {}).get("models", {}).items():
|
|
m = 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.get("spend", 0)
|
|
model_totals[m_name]["reqs"] += m.get("successful_requests", 0)
|
|
model_totals[m_name]["tokens"] += m.get("total_tokens", 0)
|
|
model_totals[m_name]["cache"] += m.get("cache_read_input_tokens", 0)
|
|
|
|
lines.append(f"\n{'Model':32s} | {'Spend':>10s} | {'Requests':>12s} | {'Tokens':>12s}")
|
|
lines.append("-" * 85)
|
|
hidden = 0
|
|
for name, mt in sorted(model_totals.items(), key=lambda x: x[1]["spend"], reverse=True):
|
|
if effective_hide and mt["spend"] < 0.001:
|
|
hidden += 1; continue
|
|
c_inf = f" 💾{int(mt['cache']/(mt['tokens'] or 1)*100)}%" if mt["cache"] else ""
|
|
lines.append(f" {self._compact_name(name, 30):30s} | ${mt['spend']:9.2f} | {mt['reqs']:8d} ok | {self._format_tokens(mt['tokens']):>10s}{c_inf}")
|
|
if hidden: lines.append(f" ... ({hidden} models with sub-$0.001 spend hidden)")
|
|
|
|
return "\n".join(lines)
|
|
|
|
def litellm_history(self, limit: int = 20, show_failed: bool = True) -> str:
|
|
"""Deep forensic history including Thought detection, Latency, and Cost Audit."""
|
|
if not self.valves.litellm_master_key: return "❌ Master key required."
|
|
try:
|
|
resp = requests.get(f"{self._base_url()}/spend/logs", headers=self._headers(), timeout=30)
|
|
if resp.status_code != 200: return f"❌ Error {resp.status_code}"
|
|
logs = sorted(resp.json(), key=lambda x: x.get("startTime", ""), reverse=True)
|
|
except Exception as e: return f"❌ 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 Forensic History — {self._base_url()}"]
|
|
lines.append(f"Showing {len(recent)} of {len(logs)} total requests")
|
|
lines.append("=" * 115)
|
|
lines.append(f" # {'Time':8s} | {'Model':28s} | {'Status':7s} | {'Dur':>6s} | {'Spend':>10s} | {'Tokens':>12s}")
|
|
lines.append("-" * 115)
|
|
|
|
for i, entry in enumerate(recent, 1):
|
|
ts = entry.get("startTime", "?")[5:16]
|
|
model = self._compact_name(entry.get("model", "unknown"), 28)
|
|
spend = entry.get("spend", 0)
|
|
p_tok = entry.get("prompt_tokens", 0) or 0
|
|
c_tok = entry.get("completion_tokens", 0) or 0
|
|
|
|
meta = entry.get("metadata", {}) or {}
|
|
is_failed = bool(meta.get("exception")) or meta.get("status") == "failure"
|
|
status = "❌ FAIL" if is_failed else "✅ OK"
|
|
|
|
dur = entry.get("request_duration_ms", 0) or 0
|
|
dur_s = f"{(dur/1000):.1f}s" if dur > 1000 else f"{dur}ms"
|
|
slow = "⚠️" if dur > 15000 else ""
|
|
|
|
usage = entry.get("usage", {}) or {}
|
|
thought = usage.get("completion_tokens_details", {}).get("reasoning_tokens", 0)
|
|
thought_s = f" [💭{self._format_tokens(thought)}]" if thought else ""
|
|
has_search = bool(meta.get("web_search_options")) or "search" in entry.get("call_type", "")
|
|
search_s = " 🔍" if has_search else ""
|
|
|
|
cached = entry.get("cache_read_input_tokens", 0) or 0
|
|
cache_s = f" 💾{int(cached/p_tok*100)}%" if p_tok and cached else ""
|
|
|
|
p_f = f"{p_tok/1000:.1f}k" if p_tok >= 1000 else str(p_tok)
|
|
c_f = f"{c_tok/1000:.1f}k" if c_tok >= 1000 else str(c_tok)
|
|
|
|
audit = self._audit_cost(entry.get("model"), p_tok, cached, c_tok, spend, reasoning=thought, has_search=has_search)
|
|
|
|
line = f" {i:2d}. {ts} | {model:28s} | {status} | {dur_s:>6s}{slow} | ${spend:.4f}{audit:18s} | {self._format_tokens(p_tok):>6s}→{self._format_tokens(c_tok):<6s}{thought_s}{cache_s}{search_s}"
|
|
lines.append(line)
|
|
if is_failed: lines.append(f" └─ Error: {meta.get('exception', 'unknown')[:85]}")
|
|
|
|
return "\n".join(lines)
|