feat: add litellm_history — recent request history with model, cost, tokens, duration, cache hits

New function litellm_history(limit=20, show_cost=True) hits /spend/logs,
sorts by startTime descending, and shows the most recent N requests with:
- Model name (stripped of provider prefix)
- Call type (chat/embed/image)
- Status (✅/❌ based on exception)
- Duration
- Token counts (prompt→completion, compact format)
- Cache hit percentage (💾 indicator)
- Per-request cost (optional)

Limit capped at 50. Bumps version to 1.1.0.
This commit is contained in:
rin.kagamine committed 2026-06-04 21:17:35 -04:00
1 parent 4dd10fab80
commit fd449079be
1 file changed
+142 -1
+142 -1
View File
@@ -5,7 +5,7 @@ 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.0.0
version: 1.1.0
licence: MIT
"""
@@ -241,4 +241,145 @@ class Tools:
lines.append("By Date:")
lines.extend(date_lines)
return "\n".join(lines)
def litellm_history(
self,
limit: int = 20,
show_cost: 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)
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)
recent = sorted_logs[:limit]
total_count = len(data)
lines = []
lines.append(f"LiteLLM Request History — {self._base_url()}")
lines.append(f"Showing {len(recent)} of {total_count} total requests (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 = "❌ FAIL" if exception 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 exception:
# Truncate long error messages
err_short = exception[:80] + "…" if len(exception) > 80 else exception
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"))
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)