rin.kagamine
8a8be608a0
When show_failed=False, filters out requests with metadata.exception or metadata.status='failure' before applying the limit. Useful for hiding auth probe noise and routing failures. Header shows '(failed requests hidden)' when active.
409 lines
18 KiB
Python
409 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.1.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,
|
|
) -> 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")
|
|
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 = []
|
|
for model_name in sorted(model_totals, key=lambda k: model_totals[k]["spend"], reverse=True):
|
|
mt = model_totals[model_name]
|
|
# 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)
|
|
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) |