Der Echtzeit-Betrieb eines KI-gestützten Systems bringt eine Herausforderung mit sich, die viele Entwickler unterschätzen: die lückenlose Kontrolle der API-Kosten. Stellen Sie sich folgendes Szenario vor: Ihr E-Commerce-Unternehmen setzt gerade einen KI-Chatbot für den Weihnachtsgeschäft ein. Innerhalb weniger Stunden schießen die Kosten durch die Decke, weil ein unerwarteter Traffic-Ansturm die Token-Quoten sprengt. Genau das ist mir vor zwei Jahren passiert – und diese Erfahrung hat mich gelehrt, wie kritisch ein durchdachtes Monitoring-System ist. In diesem Tutorial zeige ich Ihnen, wie Sie mit HolySheep AI eine umfassende Kostenüberwachung implementieren, die Ihnen volle Kontrolle über Ihre AI-Ausgaben gibt.
Warum ist API-Kostenmonitoring entscheidend?
Bei der Arbeit an einem Enterprise RAG-System für einen großen Finanzdienstleister habe ich am eigenen Leib erfahren, was passiert, wenn das Token-Monitoring fehlt. Das System verarbeitete täglich über 500.000 Anfragen, und innerhalb weniger Tage waren die monatlichen Kosten von geplanten 2.000€ auf über 12.000€ gestiegen. Die Ursache war banal: Ein fehlerhafter Retry- Mechanismus sendete bei jedem Timeout drei bis fünf identische Anfragen. Ohne ein ordentliches Dashboard wäre dieser Fehler erst bei der nächsten Rechnung aufgefallen.
Mit HolySheep AI profitieren Sie von einem Wechselkurs von ¥1 zu $1, was eine Ersparnis von über 85% gegenüber vielen westlichen Anbietern bedeutet. Die Latenz liegt konstant unter 50ms, und Sie erhalten kostenlose Credits beim Start. Die transparenten Preise für 2026 zeigen: GPT-4.1 kostet $8 pro Million Token, Claude Sonnet 4.5 $15, Gemini 2.5 Flash $2.50 und DeepSeek V3.2 nur $0.42 pro Million Token. Gerade DeepSeek V3.2 bietet damit ein exzellentes Preis-Leistungs-Verhältnis für kostensensitive Anwendungen.
Grundstruktur: Token-Zähler mit HolySheep API
Bevor wir zu komplexeren Dashboards kommen, starten wir mit dem fundamentalen Baustein: dem Tracking des tatsächlichen Token-Verbrauchs. Der folgende Python-Client demonstriert eine professionelle Implementierung mit integrerter Kostenberechnung und automatischer Kategorisierung nach Modelltyp.
#!/usr/bin/env python3
"""
HolySheep AI Token-Verbrauchstracker mit Echtzeit-Kostenberechnung
Author: HolySheep AI Technical Blog
Version: 2.0.0
"""
import requests
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from collections import defaultdict
import threading
import json
@dataclass
class TokenUsage:
"""Datenklasse für einzelnen Token-Verbrauchseintrag"""
timestamp: datetime
model: str
prompt_tokens: int
completion_tokens: int
total_tokens: int
cost_usd: float
request_id: str
@dataclass
class BudgetAlert:
"""Konfiguration für Budget-Alarme"""
threshold_usd: float
period_minutes: int
callback: Optional[callable] = None
last_triggered: datetime = field(default_factory=datetime.min)
class HolySheepCostMonitor:
"""
Professioneller Token-Verbrauchsmonitor für HolySheep AI API.
Berechnet automatisch Kosten basierend auf aktuellen 2026-Preisen.
"""
# Offizielle HolySheep AI Preise (2026) in USD pro Million Token
MODEL_PRICES = {
"gpt-4.1": {"prompt": 2.00, "completion": 6.00},
"gpt-4.1-turbo": {"prompt": 1.00, "completion": 4.00},
"claude-sonnet-4.5": {"prompt": 3.00, "completion": 15.00},
"gemini-2.5-flash": {"prompt": 0.35, "completion": 1.75},
"deepseek-v3.2": {"prompt": 0.14, "completion": 0.28},
# Fallback für unbekannte Modelle
"default": {"prompt": 1.00, "completion": 3.00}
}
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.usage_log: List[TokenUsage] = []
self.lock = threading.Lock()
self.alerts: List[BudgetAlert] = []
self._total_cost = 0.0
self._session_start = datetime.now()
def _calculate_cost(self, model: str, prompt_tokens: int,
completion_tokens: int) -> float:
"""Berechnet Kosten basierend auf Modell-Preisen"""
prices = self.MODEL_PRICES.get(model, self.MODEL_PRICES["default"])
prompt_cost = (prompt_tokens / 1_000_000) * prices["prompt"]
completion_cost = (completion_tokens / 1_000_000) * prices["completion"]
return round(prompt_cost + completion_cost, 6)
def track_request(self, model: str, prompt_tokens: int,
completion_tokens: int, request_id: str = None) -> TokenUsage:
"""Verfolgt einen einzelnen API-Request und berechnet Kosten"""
total_tokens = prompt_tokens + completion_tokens
cost = self._calculate_cost(model, prompt_tokens, completion_tokens)
usage = TokenUsage(
timestamp=datetime.now(),
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
cost_usd=cost,
request_id=request_id or f"req_{int(time.time()*1000)}"
)
with self.lock:
self.usage_log.append(usage)
self._total_cost += cost
self._check_alerts()
return usage
def make_request(self, messages: List[Dict], model: str = "deepseek-v3.2",
temperature: float = 0.7, max_tokens: int = 1000) -> Dict:
"""
Führt einen API-Request durch und verfolgt automatisch den Verbrauch.
Verwendung: base_url https://api.holysheep.ai/v1
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
data = response.json()
# Extrahiere Token-Informationen aus der Antwort
usage = data.get("usage", {})
tracked = self.track_request(
model=model,
prompt_tokens=usage.get("prompt_tokens", 0),
completion_tokens=usage.get("completion_tokens", 0),
request_id=data.get("id")
)
return {
"content": data["choices"][0]["message"]["content"],
"usage": tracked,
"model": model
}
def add_alert(self, threshold_usd: float, period_minutes: int,
callback: callable = None):
"""Fügt einen Budget-Alarm hinzu"""
alert = BudgetAlert(threshold_usd, period_minutes, callback)
self.alerts.append(alert)
def _check_alerts(self):
"""Prüft alle aktiven Alarme"""
now = datetime.now()
for alert in self.alerts:
window_start = now - timedelta(minutes=alert.period_minutes)
# Berechne Kosten im Zeitfenster
recent_cost = sum(
u.cost_usd for u in self.usage_log
if u.timestamp >= window_start
)
if (recent_cost >= alert.threshold_usd and
now - alert.last_triggered > timedelta(minutes=5)):
alert.last_triggered = now
if alert.callback:
alert.callback(alert.threshold_usd, recent_cost, alert.period_minutes)
else:
print(f"⚠️ BUDGET-ALARM: {recent_cost:.2f}$ überschreitet "
f"Schwellenwert {alert.threshold_usd}$ "
f"innerhalb von {alert.period_minutes} Minuten")
def get_cost_summary(self, hours: int = 24) -> Dict:
"""Gibt eine Kostenübersicht für den angegebenen Zeitraum zurück"""
cutoff = datetime.now() - timedelta(hours=hours)
with self.lock:
filtered = [u for u in self.usage_log if u.timestamp >= cutoff]
by_model = defaultdict(lambda: {"requests": 0, "tokens": 0, "cost": 0.0})
for usage in filtered:
by_model[usage.model]["requests"] += 1
by_model[usage.model]["tokens"] += usage.total_tokens
by_model[usage.model]["cost"] += usage.cost_usd
return {
"period_hours": hours,
"total_requests": len(filtered),
"total_tokens": sum(u.total_tokens for u in filtered),
"total_cost_usd": round(self._total_cost, 4),
"by_model": dict(by_model),
"avg_cost_per_request": (
round(self._total_cost / len(filtered), 6)
if filtered else 0
)
}
Beispiel-Nutzung
if __name__ == "__main__":
monitor = HolySheepCostMonitor(api_key="YOUR_HOLYSHEEP_API_KEY")
# Budget-Alarm: Benachrichtigung bei 50$ in 60 Minuten
def alert_callback(threshold: float, current: float, period: int):
print(f"🚨 Kritisch: {current:.2f}$ Kosten in {period}min (Schwelle: {threshold}$)")
monitor.add_alert(threshold_usd=50.0, period_minutes=60, callback=alert_callback)
# Test-Request
response = monitor.make_request(
messages=[{"role": "user", "content": "Erkläre die Vorteile von RAG-Systemen"}],
model="deepseek-v3.2"
)
print(f"Antwort: {response['content'][:100]}...")
print(f"Verbrauch: {response['usage'].total_tokens} Token, Kosten: {response['usage'].cost_usd:.6f}$")
# Zusammenfassung
summary = monitor.get_cost_summary(hours=1)
print(f"\n📊 Kostenübersicht (1 Stunde):")
print(f" Gesamtkosten: {summary['total_cost_usd']:.4f}$")
print(f" Anfragen: {summary['total_requests']}")
print(f" Modelle: {summary['by_model']}")Dashboard-Implementierung mit Streamlit
Für ein professionelles Monitoring-Dashboard empfehle ich Streamlit in Kombination mit Plotly. Die folgende vollständige Anwendung bietet Echtzeit-Visualisierung, Kostenprognosen und exportierbare Berichte. Sie können dieses Dashboard auf einem internen Server hosten oder mit Streamlit Cloud kostenlos bereitstellen.
#!/usr/bin/env python3
"""
HolySheep AI Token-Dashboard - Echtzeit-Kostenmonitoring
Kompatibel mit Python 3.8+, Streamlit, Plotly, Pandas
Ausführung: streamlit run dashboard.py
"""
import streamlit as st
import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import pandas as pd
from datetime import datetime, timedelta
import requests
import time
import json
import sqlite3
from threading import Thread
import queue
============================================================
KONFIGURATION - ANPASSEN SIE DIESE WERTE
============================================================
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem HolySheep API-Key
BASE_URL = "https://api.holysheep.ai/v1"
DB_PATH = "token_usage.db"
HolySheep AI Preise 2026 (USD pro Million Token)
MODEL_PRICES = {
"gpt-4.1": {"prompt": 2.00, "completion": 6.00},
"gpt-4.1-turbo": {"prompt": 1.00, "completion": 4.00},
"claude-sonnet-4.5": {"prompt": 3.00, "completion": 15.00},
"gemini-2.5-flash": {"prompt": 0.35, "completion": 1.75},
"deepseek-v3.2": {"prompt": 0.14, "completion": 0.28},
}
Budget-Konfiguration
DAILY_BUDGET = 100.0 # Tägliches Budget in USD
MONTHLY_BUDGET = 1500.0 # Monatliches Budget in USD
============================================================
DATENBANK-SCHEMA
============================================================
def init_database():
"""Initialisiert SQLite-Datenbank für persistente Datenspeicherung"""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS token_usage (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
model TEXT NOT NULL,
prompt_tokens INTEGER,
completion_tokens INTEGER,
total_tokens INTEGER,
cost_usd REAL,
request_id TEXT,
response_time_ms REAL,
status TEXT
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS budget_alerts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
threshold_type TEXT,
threshold_value REAL,
actual_value REAL,
acknowledged INTEGER DEFAULT 0
)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_timestamp
ON token_usage(timestamp)
""")
conn.commit()
return conn
============================================================
API-CLIENT
============================================================
class HolySheepAPIClient:
"""Thread-sicherer API-Client für HolySheep AI"""
def __init__(self, api_key: str, base_url: str):
self.api_key = api_key
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def chat_completion(self, messages: list, model: str = "deepseek-v3.2",
temperature: float = 0.7, max_tokens: int = 2000) -> dict:
"""Führt einen Chat-Completion-Request aus"""
start_time = time.time()
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
response = self.session.post(
f"{self.base_url}/chat/completions",
json=payload,
timeout=60
)
response_time = (time.time() - start_time) * 1000 # ms
response.raise_for_status()
data = response.json()
usage = data.get("usage", {})
prices = MODEL_PRICES.get(model, {"prompt": 1.0, "completion": 3.0})
cost = (usage.get("prompt_tokens", 0) / 1_000_000 * prices["prompt"] +
usage.get("completion_tokens", 0) / 1_000_000 * prices["completion"])
return {
"content": data["choices"][0]["message"]["content"],
"model": model,
"prompt_tokens": usage.get("prompt_tokens", 0),
"completion_tokens": usage.get("completion_tokens", 0),
"total_tokens": usage.get("total_tokens", 0),
"cost_usd": round(cost, 6),
"response_time_ms": round(response_time, 2),
"request_id": data.get("id"),
"status": "success"
}
============================================================
STREAMLIT DASHBOARD
============================================================
def main():
st.set_page_config(
page_title="HolySheep AI - Kostenmonitoring Dashboard",
page_icon="🐑",
layout="wide"
)
# Initialisierung
if 'db_conn' not in st.session_state:
st.session_state.db_conn = init_database()
if 'client' not in st.session_state:
st.session_state.client = HolySheepAPIClient(API_KEY, BASE_URL)
if 'last_refresh' not in st.session_state:
st.session_state.last_refresh = datetime.now()
# Sidebar-Konfiguration
st.sidebar.header("⚙️ Konfiguration")
selected_model = st.sidebar.selectbox(
"Standard-Modell",
options=list(MODEL_PRICES.keys()),
index=list(MODEL_PRICES.keys()).index("deepseek-v3.2")
)
daily_limit = st.sidebar.number_input(
"Tägliches Budget (USD)",
min_value=1.0,
max_value=10000.0,
value=DAILY_BUDGET,
step=10.0
)
auto_refresh = st.sidebar.checkbox("Auto-Refresh (30s)", value=True)
# Haupttitel
st.title("🐑 HolySheep AI - Token-Verbrauchsdashboard")
st.markdown("**Echtzeit-Überwachung Ihrer AI-API-Kosten**")
# Metriken-Zeile
col1, col2, col3, col4 = st.columns(4)
# Daten laden
conn = st.session_state.db_conn
df = load_usage_data(conn)
# Berechnungen
today = datetime.now().date()
today_df = df[pd.to_datetime(df['timestamp']).dt.date == today]
today_cost = today_df['cost_usd'].sum()
today_requests = len(today_df)
today_tokens = today_df['total_tokens'].sum()
avg_latency = today_df['response_time_ms'].mean() if len(today_df) > 0 else 0
# Metriken anzeigen
with col1:
st.metric(
label="💰 Heutige Kosten",
value=f"${today_cost:.2f}",
delta=f"{today_cost/daily_limit*100:.1f}% des Budgets" if daily_limit else None
)
with col2:
st.metric(
label="📊 Heutige Anfragen",
value=f"{today_requests:,}"
)
with col3:
st.metric(
label="🎯 Verbrauchte Token",
value=f"{today_tokens:,}"
)
with col4:
st.metric(
label="⚡ Durchschn. Latenz",
value=f"{avg_latency:.0f}ms"
)
# Budget-Fortschritt
st.progress(min(today_cost/daily_limit, 1.0), text=f"Budget: ${today_cost:.2f} / ${daily_limit:.2f}")
# Warnung bei Budget-Überschreitung
if today_cost >= daily_limit:
st.error(f"🚨 Budget-Überschreitung! Heutige Kosten ({today_cost:.2f}$) "
f"überschreiten das Limit ({daily_limit:.2f}$)")
# Tabs für verschiedene Ansichten
tab1, tab2, tab3, tab4 = st.tabs([
"📈 Zeitverlauf",
"📊 Nach Modell",
"💡 Kostenanalyse",
"📝 API-Test"
])
with tab1:
st.subheader("Token-Verbrauch über Zeit")
# Zeitraum-Selector
time_range = st.selectbox(
"Zeitraum",
options=["Letzte Stunde", "Letzte 24 Stunden", "Letzte 7 Tage", "Letzte 30 Tage"]
)
hours_map = {"Letzte Stunde": 1, "Letzte 24 Stunden": 24,
"Letzte 7 Tage": 168, "Letzte 30 Tage": 720}
cutoff = datetime.now() - timedelta(hours=hours_map[time_range])
filtered_df = df[pd.to_datetime(df['timestamp']) >= cutoff]
# Kosten-Zeitreihen-Diagramm
fig = make_subplots(
rows=2, cols=1,
subplot_titles=("Kostenentwicklung (USD)", "Token-Verbrauch"),
vertical_spacing=0.15
)
# Kosten-Zeitreihe
filtered_df['hour'] = pd.to_datetime(filtered_df['timestamp']).dt.floor('H')
cost_by_hour = filtered_df.groupby('hour')['cost_usd'].sum().reset_index()
fig.add_trace(
go.Scatter(
x=cost_by_hour['hour'],
y=cost_by_hour['cost_usd'],
mode='lines+markers',
name='Kosten ($)',
line=dict(color='#00D09C', width=2),
fill='tozeroy'
),
row=1, col=1
)
# Budget-Linie
budget_per_hour = daily_limit / 24 * hours_map[time_range] / hours_map[time_range]
fig.add_hline(
y=daily_limit,
line_dash="dash",
line_color="red",
annotation_text="Tagesbudget",
row=1, col=1
)
# Token-Zeitreihe
tokens_by_hour = filtered_df.groupby('hour')['total_tokens'].sum().reset_index()
fig.add_trace(
go.Bar(
x=tokens_by_hour['hour'],
y=tokens_by_hour['total_tokens'],
name='Token',
marker_color='#5B8DEF'
),
row=2, col=1
)
fig.update_layout(
height=600,
showlegend=True,
template="plotly_dark"
)
st.plotly_chart(fig, use_container_width=True)
with tab2:
st.subheader("Verbrauch nach Modell")
col_left, col_right = st.columns(2)
with col_left:
# Kosten nach Modell
model_costs = df.groupby('model')['cost_usd'].sum().sort_values(ascending=False)
fig_pie = px.pie(
values=model_costs.values,
names=model_costs.index,
title="Kostenverteilung nach Modell",
color_discrete_sequence=px.colors.qualitative.Set3
)
fig_pie.update_layout(template="plotly_dark")
st.plotly_chart(fig_pie, use_container_width=True)
with col_right:
# Anfragen nach Modell
model_requests = df.groupby('model').size().sort_values(ascending=False)
fig_bar = px.bar(
x=model_requests.index,
y=model_requests.values,
title="Anzahl Anfragen nach Modell",
labels={'x': 'Modell', 'y': 'Anfragen'},
color=model_requests.values,
color_continuous_scale='Viridis'
)
fig_bar.update_layout(template="plotly_dark", showlegend=False)
st.plotly_chart(fig_bar, use_container_width=True)
# Detail-Tabelle
st.subheader("Modell-Performance")
model_stats = df.groupby('model').agg({
'cost_usd': ['sum', 'mean'],
'total_tokens': ['sum', 'mean'],
'response_time_ms': 'mean',
'id': 'count'
}).round(4)
model_stats.columns = ['Gesamtkosten', 'Ø Kosten', 'Gesamttoken',
'Ø Token', 'Ø Latenz (ms)', 'Anfragen']
model_stats = model_stats.sort_values('Gesamtkosten', ascending=False)
st.dataframe(model_stats, use_container_width=True)
with tab3:
st.subheader("Kostenanalyse und Prognose")
# Kosten-Statistiken
col1, col2, col3 = st.columns(3)
weekly_cost = df[pd.to_datetime(df['timestamp']) >=
datetime.now() - timedelta(days=7)]['cost_usd'].sum()
monthly_cost = df[pd.to_datetime(df['timestamp']) >=
datetime.now() - timedelta(days=30)]['cost_usd'].sum()
projected_monthly = (monthly_cost / 30 * 30) if monthly_cost > 0 else 0
with col1:
st.metric("7-Tage-Kosten", f"${weekly_cost:.2f}")
with col2:
st.metric("30-Tage-Kosten", f"${monthly_cost:.2f}")
with col3:
st.metric("Prognostizierte Monatskosten", f"${projected_monthly:.2f}")
# Kostenoptimierungs-Vorschläge
st.subheader("💡 Optimierungsempfehlungen")
current_model = selected_model
best_value_model = min(MODEL_PRICES.keys(),
key=lambda m: MODEL_PRICES[m]["prompt"])
st.info(f"""
**Kostenoptimierung für {current_model}:**
1. **Modell-Auswahl prüfen**: Für einfache Aufgaben könnte **{best_value_model}**
ausreichen und bis zu {MODEL_PRICES.get(current_model, MODEL_PRICES['default'])['prompt'] / MODEL_PRICES[best_value_model]['prompt']:.0f}x günstiger sein.
2. **Max-Tokens optimieren**: Reduzieren Sie die maximalen Antwort-Token
um durchschnittlich 20-30%.
3. **Caching implementieren**: Bei wiederholten Anfragen können Sie bis zu
60% der Kosten einsparen.
4. **Batch-Verarbeitung**: Gruppieren Sie mehrere Anfragen für effizientere
Verarbeitung.
""")
with tab4:
st.subheader("API-Testumgebung")
# Test-Interface
test_col1, test_col2 = st.columns([3, 1])
with test_col1:
prompt = st.text_area(
"Test-Prompt",
value="Erkläre in 3 Sätzen, was ein RAG-System ist.",
height=100
)
with test_col2:
test_model = st.selectbox("Modell", options=list(MODEL_PRICES.keys()))
test_temp = st.slider("Temperatur", 0.0, 1.0, 0.7, 0.1)
if st.button("🚀 Anfrage senden", type="primary"):
with st.spinner("Anfrage wird gesendet..."):
try:
response = st.session_state.client.chat_completion(
messages=[{"role": "user", "content": prompt}],
model=test_model,
temperature=test_temp
)
# Ergebnis anzeigen
st.success("Anfrage erfolgreich!")
result_col1, result_col2 = st.columns(2)
with result_col1:
st.markdown("**Antwort:**")
st.write(response['content'])
with result_col2:
st.markdown("**Verbrauch:**")
st.write(f"- Modell: {response['model']}")
st.write(f"- Prompt-Token: {response['prompt_tokens']:,}")
st.write(f"- Completion-Token: {response['completion_tokens']:,}")
st.write(f"- Gesamt-Token: {response['total_tokens']:,}")
st.write(f"- Kosten: ${response['cost_usd']:.6f}")
st.write(f"- Latenz: {response['response_time_ms']:.0f}ms")
# In Datenbank speichern
save_usage(conn, response)
except Exception as e:
st.error(f"Fehler: {str(e)}")
st.markdown("---")
# Letzte Anfragen
st.subheader("Letzte 10 Anfragen")
recent_df = df.tail(10)[['timestamp', 'model', 'total_tokens',
'cost_usd', 'response_time_ms']].copy()
recent_df['timestamp'] = pd.to_datetime(recent_df['timestamp']).dt.strftime('%H:%M:%S')
st.dataframe(recent_df, use_container_width=True)
# Export-Optionen
st.sidebar.markdown("---")
st.sidebar.subheader("📥 Datenexport")
if st.sidebar.button("CSV-Export (7 Tage)"):
export_df = df[pd.to_datetime(df['timestamp']) >=
datetime.now() - timedelta(days=7)]
csv = export_df.to_csv(index=False)
st.sidebar.download_button(
label="Download CSV",
data=csv,
file_name=f"token_usage_{datetime.now().strftime('%Y%m%d')}.csv",
mime="text/csv"
)
# Footer
st.markdown("---")
st.markdown(
"**🐑 HolySheep AI Dashboard** | "
"Preise basierend auf [HolySheep AI 2026 Preisliste]"
"(https://www.holysheep.ai/pricing) | "
"<50ms Latenz | 85%+ Ersparnis"
)
# Auto-Refresh
if auto_refresh:
time.sleep(30)
st.rerun()
def load_usage_data(conn) -> pd.DataFrame:
"""Lädt Verbrauchsdaten aus der Datenbank"""
query = """
SELECT id, timestamp, model, prompt_tokens, completion_tokens,
total_tokens, cost_usd, request_id, response_time_ms, status
FROM token_usage
WHERE timestamp >= datetime('now', '-30 days')
ORDER BY timestamp DESC
"""
df = pd.read_sql_query(query, conn)
return df
def save_usage(conn, response: dict):
"""Speichert einen API-Verbrauch in der Datenbank"""
cursor = conn.cursor()
cursor.execute("""
INSERT INTO token_usage
(timestamp, model, prompt_tokens, completion_tokens, total_tokens,
cost_usd, request_id, response_time_ms, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
datetime.now().isoformat(),
response['model'],
response['prompt_tokens'],
response['completion_tokens'],
response['total_tokens'],
response['cost_usd'],
response['request_id'],
response['response_time_ms'],
response['status']
))
conn.commit()
if __name__ == "__main__":
main()Budget-Warnungen: Automatische Alarmierung bei Kostenüberschreitung
Ein statisches Dashboard ist gut, aber was passiert, wenn Sie nicht vor dem Bildschirm sitzen? Hier kommen automatisierte Budget-Warnungen ins Spiel. Ich habe für mehrere Kunden Systeme implementiert, die bei Überschreitung definierter Schwellenwerte automatisch per E-Mail, Slack oder SMS benachrichtigen. Das folgende System integriert sich nahtlos in HolySheep AI und reagiert in Echtzeit.
#!/usr/bin/env python3
"""
HolySheep AI Budget-Warnsystem mit mehrkanaliger Alarmierung
- E-Mail-Benachrichtigungen
- Slack-Webhook-Integration
- SMS via Twilio
- Automatische Kostenbegrenzung
"""
import smtplib
import ssl
import json
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from dataclasses import dataclass, field
from typing import List, Optional, Callable
from datetime import datetime, timedelta
from enum import Enum
import requests
import threading
import time
class AlertLevel(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
EMERGENCY = "emergency"
@dataclass
class BudgetThreshold:
"""Definiert einen Budget-Schwellenwert"""
name: str
amount_usd: float
time_window_minutes: int
alert_level: AlertLevel
auto_disable: bool = False
cooldown_minutes: int = 15
@dataclass
class AlertNotification:
"""Struktur für Alarm-Benachrichtigungen"""
level: AlertLevel
title: str
message: str
current_cost: float
threshold: float
model_breakdown: dict
timestamp: datetime
class BudgetAlertManager:
"""
Zentrale Verwaltung für Budget-Warnungen und automatische Reaktionen.
Integriert E-Mail, Slack und SMS-Benachrichtigungen.
"""
def __init__(self):
self.thresholds: List[BudgetThreshold] = []
self.notification_channels: List[dict] = []
self.alert_history: List[AlertNotification] = []
self.cooldowns: dict = {}
self._lock = threading.Lock()
# Standard-Schwellenwerte konfigurieren
self._setup_default_thresholds()
def _setup_default_thresholds(self):
"""Richtet Standard-Budget-Schwellenwerte ein"""
# Sanfte Warnung bei 50% des Tagesbudgets
self.add_threshold(BudgetThreshold(
name="daily_50_percent",
amount_usd=50.0,
time_window_minutes=1440, # 24 Stunden
alert_level=AlertLevel.INFO
))
# Warnung bei 75%
self.add_threshold(BudgetThreshold(
name="daily_75_percent",
amount_usd=75.0,
time_window_minutes=1440,
alert_level=AlertLevel.WARNING
))
# Kritisch bei 90%
self.add_threshold(BudgetThreshold(
name="daily_90_percent",
amount_usd=90.0,
time_window_minutes=1440,
alert_level=AlertLevel.CRITICAL,
cooldown_minutes=30
))
# Notfall bei Überschreitung
self.add_threshold(BudgetThreshold(
name="daily_exceeded",
amount_usd=100.0,
time_window_minutes=1440,
alert_level=AlertLevel.EMERGENCY,
auto_disable=True,
cooldown_minutes=60
))
def add_threshold(self, threshold: BudgetThreshold):
"""Fügt einen neuen Budget-Schwellenwert hinzu"""
with self._lock:
self.thresholds.append(threshold)
def configure_email(self, smtp_server: str, smtp_port: int,
sender_email