rin.kagamine
6adcd79c1a
Adds optional hide_zero_spend parameter (default False) to filter out models with bash.00 spend from the per-model breakdown. Useful when free-tier or subscription models (e.g. Ollama Cloud) dominate the breakdown with high token counts but zero cost. - Updated litellm_cost() signature with hide_zero_spend parameter - Added filtering logic in model line builder - Added hidden model count summary line - Updated README with parameter documentation - Bumped version to 1.2.0
418 lines
18 KiB
Python
418 lines
18 KiB
Python
"""
|
|
title: LiteLLM Dashboard
|
|
author: Kagamine Rin
|
|
author_url: https://rin.captcouch.xyz
|
|
description: Check LiteLLM proxy health + auth status and get cost breakdowns with configurable timerange. Two functions, that's it.
|
|
required_open_webui_version: 0.4.0
|
|
requirements: requests
|
|
version: 1.2.0
|
|
licence: MIT
|
|
"""
|
|
|
|
import requests
|
|
from pydantic import BaseModel, Field
|
|
from typing import Optional
|
|
|
|
|
|
class Tools:
|
|
def __init__(self):
|
|
self.valves = self.Valves()
|
|
|
|
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"}},
|
|
)
|
|
|
|
def _headers(self) -> dict:
|
|
"""Build auth headers from valves."""
|
|
if not self.valves.litellm_master_key:
|
|
return {}
|
|
return {"Authorization": f"Bearer {self.valves.litellm_master_key}"}
|
|
|
|
def _base_url(self) -> str:
|
|
"""Get base URL, strip trailing slash."""
|
|
return self.valves.litellm_url.rstrip("/")
|
|
|
|
def litellm_health(self) -> str:
|
|
"""
|
|
Check LiteLLM proxy health and authentication status.
|
|
Returns which models are healthy/unhealthy, their rate limits, and any auth errors.
|
|
Use this to verify your LiteLLM proxy is up and your master key works.
|
|
"""
|
|
url = f"{self._base_url()}/health"
|
|
headers = self._headers()
|
|
|
|
try:
|
|
resp = requests.get(url, headers=headers, timeout=15)
|
|
except requests.exceptions.ConnectionError:
|
|
return f"❌ Connection failed — LiteLLM proxy not reachable at {self._base_url()}"
|
|
except requests.exceptions.Timeout:
|
|
return f"❌ Connection timed out after 15s — proxy at {self._base_url()} may be overloaded"
|
|
|
|
if resp.status_code == 401:
|
|
return "❌ Authentication failed — check your litellm_master_key valve. The proxy rejected the key."
|
|
if resp.status_code == 403:
|
|
return "❌ Forbidden — your key doesn't have permission to access /health"
|
|
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", [])
|
|
healthy_count = data.get("healthy_count", len(healthy))
|
|
unhealthy_count = data.get("unhealthy_count", len(unhealthy))
|
|
|
|
# Strip litellm_metadata — it's ~2KB per model of auth context we don't need
|
|
# Also strip other internal fields that bloat the response
|
|
STRIP_KEYS = {
|
|
"litellm_metadata", "safety_settings", "cache_control_injection_points",
|
|
"use_in_pass_through", "use_litellm_proxy", "merge_reasoning_content_in_choices",
|
|
}
|
|
|
|
lines = []
|
|
lines.append(f"LiteLLM Health Check — {self._base_url()}")
|
|
lines.append("=" * 50)
|
|
|
|
if not self.valves.litellm_master_key:
|
|
lines.append("⚠️ No master key configured — showing public health only")
|
|
else:
|
|
lines.append("✅ Authenticated with master key")
|
|
|
|
lines.append(f"\nHealthy: {healthy_count} | Unhealthy: {unhealthy_count}")
|
|
|
|
if healthy:
|
|
lines.append(f"\n--- Healthy Models ({healthy_count}) ---")
|
|
for ep in healthy:
|
|
model = ep.get("model", "unknown")
|
|
# Strip gemini/ prefix for readability
|
|
display = model.replace("gemini/", "") if model.startswith("gemini/") else model
|
|
rpm = ep.get("rpm", "?")
|
|
timeout = ep.get("timeout", "?")
|
|
extras = []
|
|
if ep.get("reasoning_effort") and ep["reasoning_effort"] != "none":
|
|
extras.append(f"reasoning: {ep['reasoning_effort']}")
|
|
if ep.get("service_tier"):
|
|
extras.append(f"tier: {ep['service_tier']}")
|
|
extra_str = f" ({', '.join(extras)})" if extras else ""
|
|
lines.append(f" ✅ {display} — rpm: {rpm}, timeout: {timeout}s{extra_str}")
|
|
|
|
if unhealthy:
|
|
lines.append(f"\n--- Unhealthy Models ({unhealthy_count}) ---")
|
|
for ep in unhealthy:
|
|
model = ep.get("model", "unknown")
|
|
display = model.replace("gemini/", "") if model.startswith("gemini/") else model
|
|
lines.append(f" ❌ {display}")
|
|
|
|
if not healthy and not unhealthy:
|
|
lines.append("\n(no endpoint data returned)")
|
|
|
|
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.
|
|
:param start_date: Start date in YYYY-MM-DD format (e.g. "2026-06-01")
|
|
:param end_date: End date in YYYY-MM-DD format (e.g. "2026-06-04")
|
|
:param hide_zero_spend: Hide models with $0.00 spend (default False). Useful when free-tier or subscription models (e.g. Ollama Cloud) dominate the breakdown with high token counts but zero cost.
|
|
Returns formatted cost report with per-model and per-day breakdowns, token counts, cache hit rate, and request stats.
|
|
"""
|
|
if not self.valves.litellm_master_key:
|
|
return "❌ Master key required for cost data. Set the litellm_master_key valve."
|
|
|
|
url = f"{self._base_url()}/user/daily/activity"
|
|
headers = self._headers()
|
|
params = {"start_date": start_date, "end_date": end_date}
|
|
|
|
try:
|
|
resp = requests.get(url, headers=headers, params=params, timeout=30)
|
|
except requests.exceptions.ConnectionError:
|
|
return f"❌ Connection failed — LiteLLM proxy not reachable at {self._base_url()}"
|
|
except requests.exceptions.Timeout:
|
|
return f"❌ Connection timed out after 30s — large date ranges may be slow"
|
|
|
|
if resp.status_code == 401:
|
|
return "❌ Authentication failed — check your litellm_master_key valve."
|
|
if resp.status_code == 403:
|
|
return "❌ Forbidden — your key doesn't have permission to access cost data"
|
|
if resp.status_code != 200:
|
|
return f"❌ Unexpected status {resp.status_code}: {resp.text[:500]}"
|
|
|
|
data = resp.json()
|
|
results = data.get("results", [])
|
|
metadata = data.get("metadata", {})
|
|
|
|
if not results:
|
|
return f"No cost data found for {start_date} to {end_date}."
|
|
|
|
# Aggregate totals from metadata
|
|
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)
|
|
total_successful = metadata.get("total_successful_requests", 0)
|
|
total_failed = metadata.get("total_failed_requests", 0)
|
|
total_api = metadata.get("total_api_requests", 0)
|
|
cache_read = metadata.get("total_cache_read_input_tokens", 0)
|
|
cache_created = metadata.get("total_cache_creation_input_tokens", 0)
|
|
|
|
# Cache hit rate
|
|
total_input_tokens = total_prompt # cache_read is a subset of prompt
|
|
if total_input_tokens > 0 and cache_read > 0:
|
|
cache_pct = (cache_read / total_input_tokens) * 100
|
|
cache_str = f"{cache_pct:.1f}% ({cache_read:,} / {total_input_tokens:,} input tokens)"
|
|
else:
|
|
cache_str = "N/A"
|
|
|
|
# Build per-model breakdown across all dates
|
|
model_totals = {}
|
|
for day in results:
|
|
breakdown = day.get("breakdown", {})
|
|
models = breakdown.get("models", {})
|
|
for model_name, model_data in models.items():
|
|
m_metrics = model_data.get("metrics", {})
|
|
if model_name not in model_totals:
|
|
model_totals[model_name] = {
|
|
"spend": 0, "requests": 0, "prompt": 0,
|
|
"completion": 0, "total": 0, "failed": 0,
|
|
}
|
|
model_totals[model_name]["spend"] += m_metrics.get("spend", 0)
|
|
model_totals[model_name]["requests"] += m_metrics.get("successful_requests", 0)
|
|
model_totals[model_name]["failed"] += m_metrics.get("failed_requests", 0)
|
|
model_totals[model_name]["prompt"] += m_metrics.get("prompt_tokens", 0)
|
|
model_totals[model_name]["completion"] += m_metrics.get("completion_tokens", 0)
|
|
model_totals[model_name]["total"] += m_metrics.get("total_tokens", 0)
|
|
|
|
# Build per-date breakdown
|
|
date_lines = []
|
|
for day in results:
|
|
date = day.get("date", "?")
|
|
m = day.get("metrics", {})
|
|
date_lines.append(
|
|
f" {date}: ${m.get('spend', 0):.2f} | "
|
|
f"{m.get('successful_requests', 0)} ok, {m.get('failed_requests', 0)} fail | "
|
|
f"{m.get('total_tokens', 0):,} tokens"
|
|
)
|
|
|
|
# Build model lines sorted by spend
|
|
model_lines = []
|
|
hidden_count = 0
|
|
for model_name in sorted(model_totals, key=lambda k: model_totals[k]["spend"], reverse=True):
|
|
mt = model_totals[model_name]
|
|
if hide_zero_spend and mt["spend"] == 0:
|
|
hidden_count += 1
|
|
continue
|
|
# Strip common provider prefixes for readability
|
|
display = model_name
|
|
for prefix in ("gemini/", "openai/", "anthropic/", "mistral/", "cohere/"):
|
|
if display.startswith(prefix):
|
|
display = display[len(prefix):]
|
|
break
|
|
# Truncate long model names to fit table, appending ellipsis
|
|
max_name = 24
|
|
if len(display) > max_name:
|
|
display = display[:max_name - 1] + "…"
|
|
model_lines.append(
|
|
f" {display:25s} ${mt['spend']:7.2f} | "
|
|
f"{mt['requests']:3d} ok, {mt['failed']:d} fail | "
|
|
f"{mt['total']:>10,} tokens"
|
|
)
|
|
|
|
# Assemble report
|
|
lines = []
|
|
lines.append(f"LiteLLM Cost Report: {start_date} to {end_date}")
|
|
lines.append("=" * 60)
|
|
lines.append(f"Total Spend: ${total_spend:.2f}")
|
|
lines.append(f"Total Tokens: {total_tokens:,} ({total_prompt:,} prompt + {total_completion:,} completion)")
|
|
lines.append(f"Cache Hit Rate: {cache_str}")
|
|
lines.append(f"Requests: {total_successful} successful, {total_failed} failed ({total_api} total)")
|
|
if cache_created > 0:
|
|
lines.append(f"Cache Created: {cache_created:,} tokens")
|
|
lines.append("")
|
|
lines.append("By Model:")
|
|
lines.append(f' {"Model":25s} {"Spend":>7s} | {"Requests":>14s} | {"Tokens":>10s}')
|
|
lines.extend(model_lines)
|
|
if hidden_count > 0:
|
|
lines.append(f" ... ({hidden_count} model(s) with $0.00 spend hidden)")
|
|
lines.append("")
|
|
lines.append("By Date:")
|
|
lines.extend(date_lines)
|
|
|
|
return "\n".join(lines)
|
|
|
|
def litellm_history(
|
|
self,
|
|
limit: int = 20,
|
|
show_cost: bool = True,
|
|
show_failed: bool = True,
|
|
) -> str:
|
|
"""
|
|
Get recent LiteLLM proxy request history.
|
|
:param limit: Number of recent requests to show (1-50, default 20)
|
|
:param show_cost: Include per-request cost (default True)
|
|
:param show_failed: Include failed requests in output (default True). Set False to hide auth failures, routing errors, etc.
|
|
Returns a formatted table of recent LLM requests with model, status, tokens, duration, and optional cost.
|
|
"""
|
|
if not self.valves.litellm_master_key:
|
|
return "❌ Master key required for request history. Set the litellm_master_key valve."
|
|
|
|
limit = max(1, min(50, limit))
|
|
|
|
url = f"{self._base_url()}/spend/logs"
|
|
headers = self._headers()
|
|
|
|
try:
|
|
resp = requests.get(url, headers=headers, timeout=30)
|
|
except requests.exceptions.ConnectionError:
|
|
return f"❌ Connection failed — LiteLLM proxy not reachable at {self._base_url()}"
|
|
except requests.exceptions.Timeout:
|
|
return f"❌ Connection timed out after 30s — request history may be large"
|
|
|
|
if resp.status_code == 401:
|
|
return "❌ Authentication failed — check your litellm_master_key valve."
|
|
if resp.status_code == 403:
|
|
return "❌ Forbidden — your key doesn't have permission to access request logs"
|
|
if resp.status_code != 200:
|
|
return f"❌ Unexpected status {resp.status_code}: {resp.text[:500]}"
|
|
|
|
data = resp.json()
|
|
if not isinstance(data, list) or len(data) == 0:
|
|
return "No request history found."
|
|
|
|
# Sort by startTime descending and take the most recent `limit` entries
|
|
sorted_logs = sorted(data, key=lambda e: e.get("startTime", ""), reverse=True)
|
|
|
|
# Filter out failed requests if show_failed is False
|
|
if not show_failed:
|
|
sorted_logs = [
|
|
e for e in sorted_logs
|
|
if not (
|
|
(e.get("metadata", {}) or {}).get("exception", "")
|
|
or (e.get("metadata", {}) or {}).get("status", "") == "failure"
|
|
)
|
|
]
|
|
|
|
recent = sorted_logs[:limit]
|
|
total_count = len(data)
|
|
filtered_note = " (failed requests hidden)" if not show_failed else ""
|
|
|
|
lines = []
|
|
lines.append(f"LiteLLM Request History — {self._base_url()}")
|
|
lines.append(f"Showing {len(recent)} of {total_count} total requests{filtered_note} (most recent first)")
|
|
lines.append("=" * 90)
|
|
|
|
for i, entry in enumerate(recent, 1):
|
|
# Timestamp
|
|
start_time = entry.get("startTime", "?")[:19] # YYYY-MM-DDTHH:MM:SS
|
|
start_short = start_time[5:] if len(start_time) > 5 else start_time # MM-DDTHH:MM:SS
|
|
|
|
# Model — strip provider prefix for readability
|
|
model = entry.get("model", "unknown")
|
|
for prefix in ("gemini/", "openai/", "anthropic/", "mistral/", "cohere/"):
|
|
if model.startswith(prefix):
|
|
model = model[len(prefix):]
|
|
break
|
|
|
|
# Call type
|
|
call_type = entry.get("call_type", "?")
|
|
# Shorten common call types
|
|
call_type = call_type.replace("acompletion", "chat").replace("aembedding", "embed").replace("aimage_generation", "image").replace("_", " ")
|
|
|
|
# Status — check for errors
|
|
metadata = entry.get("metadata", {}) or {}
|
|
exception = metadata.get("exception", "")
|
|
status_flag = metadata.get("status", "")
|
|
is_failed = bool(exception) or status_flag == "failure"
|
|
status = "❌ FAIL" if is_failed else "✅ OK"
|
|
|
|
# Duration
|
|
duration_ms = entry.get("request_duration_ms", 0) or 0
|
|
if duration_ms >= 1000:
|
|
duration_str = f"{duration_ms / 1000:.1f}s"
|
|
else:
|
|
duration_str = f"{duration_ms}ms"
|
|
|
|
# Tokens
|
|
prompt_tokens = entry.get("prompt_tokens", 0) or 0
|
|
completion_tokens = entry.get("completion_tokens", 0) or 0
|
|
total_tokens = entry.get("total_tokens", 0) or 0
|
|
|
|
# Cache
|
|
cache_read = entry.get("cache_read_input_tokens", 0) or 0
|
|
|
|
# Spend
|
|
spend = entry.get("spend", 0) or 0
|
|
|
|
# Build line
|
|
base = f" {i:2d}. {start_short} | {model[:28]:28s} | {call_type[:10]:10s} | {status} | {duration_str:>7s}"
|
|
|
|
# Token info
|
|
if total_tokens > 0:
|
|
# Compact token display
|
|
if prompt_tokens >= 1000:
|
|
p_str = f"{prompt_tokens / 1000:.1f}k"
|
|
else:
|
|
p_str = str(prompt_tokens)
|
|
if completion_tokens >= 1000:
|
|
c_str = f"{completion_tokens / 1000:.1f}k"
|
|
else:
|
|
c_str = str(completion_tokens)
|
|
token_str = f"{p_str}→{c_str}"
|
|
|
|
# Cache indicator
|
|
if cache_read > 0:
|
|
cache_pct = (cache_read / prompt_tokens * 100) if prompt_tokens > 0 else 0
|
|
token_str += f" 💾{cache_pct:.0f}%"
|
|
|
|
base += f" | {token_str}"
|
|
|
|
# Cost
|
|
if show_cost and spend > 0:
|
|
base += f" | ${spend:.4f}"
|
|
|
|
# Error info
|
|
if is_failed:
|
|
# Priority: exception string > error_information.error_message > generic status
|
|
if exception:
|
|
err_short = exception[:80] + "…" if len(exception) > 80 else exception
|
|
else:
|
|
err_info = metadata.get("error_information", {}) or {}
|
|
err_msg = err_info.get("error_message", "")
|
|
if err_msg:
|
|
err_short = err_msg[:80] + "…" if len(err_msg) > 80 else err_msg
|
|
else:
|
|
err_short = "status: failure"
|
|
base += f"\n └─ {err_short}"
|
|
|
|
lines.append(base)
|
|
|
|
# Summary
|
|
if len(recent) > 0:
|
|
recent_spend = sum(e.get("spend", 0) or 0 for e in recent)
|
|
recent_prompt = sum(e.get("prompt_tokens", 0) or 0 for e in recent)
|
|
recent_completion = sum(e.get("completion_tokens", 0) or 0 for e in recent)
|
|
recent_cache = sum(e.get("cache_read_input_tokens", 0) or 0 for e in recent)
|
|
recent_failed = sum(1 for e in recent if ((e.get("metadata", {}) or {}).get("exception")) or ((e.get("metadata", {}) or {}).get("status", "")) == "failure")
|
|
|
|
lines.append("")
|
|
lines.append(f" Summary ({len(recent)} shown): ${recent_spend:.4f} spent | "
|
|
f"{recent_prompt:,} prompt + {recent_completion:,} completion tokens | "
|
|
f"{recent_failed} failed | "
|
|
f"cache hit {recent_cache / recent_prompt * 100:.0f}%" if recent_prompt > 0 and recent_cache > 0
|
|
else f" Summary ({len(recent)} shown): ${recent_spend:.4f} spent | "
|
|
f"{recent_prompt:,} prompt + {recent_completion:,} completion tokens | "
|
|
f"{recent_failed} failed")
|
|
|
|
return "\n".join(lines)
|