Stellen Sie sich vor, Sie könnten Ihrem KI-Assistenten einfach auf Deutsch sagen: "Zeig mir die Verkaufszahlen der letzten Stunde, gruppiert nach Region" – und Sekunden später erscheint ein schöner Report. Genau das bauen wir heute zusammen. Wir verbinden das leistungsstarke Sprachmodell Claude Opus 4.7 über die HolySheep AI-Plattform mit einer ClickHouse-Datenbank und erzeugen daraus automatisch Echtzeit-Reports.

Sie brauchen keine Vorkenntnisse. Wir gehen jeden Schritt gemeinsam durch, wie bei einem Kochrezept. Am Ende haben Sie ein funktionierendes System, das Ihre Datenbank "versteht".

Was Sie am Ende haben werden

Schritt 1: Was Sie brauchen (Voraussetzungen)

Bevor wir loslegen, besorgen Sie bitte diese Dinge. Keine Sorge, alles ist kostenlos oder fast kostenlos:

Schritt 2: ClickHouse starten (die Datenbank)

Wir nutzen Docker, weil das am einfachsten ist. Öffnen Sie das Terminal (auf Windows: PowerShell, auf Mac: Terminal) und geben Sie ein:

# ClickHouse mit einem Klick starten
docker run -d --name clickhouse-demo \
  -p 8123:8123 -p 9000:9000 \
  -e CLICKHOUSE_DB=demo \
  -e CLICKHOUSE_USER=admin \
  -e CLICKHOUSE_PASSWORD=demo1234 \
  clickhouse/clickhouse-server:latest


Warten Sie 10 Sekunden, dann testen Sie:

curl http://localhost:8123/ping

Wenn Sie die Antwort Ok. sehen, läuft Ihre Datenbank. Jetzt füllen wir sie mit ein paar Beispieldaten:

curl 'http://localhost:8123/?user=admin&password=demo1234' --data-binary "
CREATE TABLE IF NOT EXISTS demo.sales (
    event_time DateTime,
    region String,
    product String,
    amount Decimal(10,2),
    customer_id String
) ENGINE = MergeTree()
ORDER BY event_time;

INSERT INTO demo.sales VALUES
('2026-01-15 09:00:00', 'Bayern', 'Laptop', 1299.00, 'K001'),
('2026-01-15 09:15:00', 'Berlin', 'Maus', 29.99, 'K002'),
('2026-01-15 09:32:00', 'Bayern', 'Monitor', 449.00, 'K003'),
('2026-01-15 09:45:00', 'Hamburg', 'Tastatur', 89.90, 'K004'),
('2026-01-15 10:00:00', 'Berlin', 'Laptop', 1299.00, 'K005'),
('2026-01-15 10:12:00', 'Bayern', 'Maus', 29.99, 'K006'),
('2026-01-15 10:25:00', 'Sachsen', 'Monitor', 449.00, 'K007'),
('2026-01-15 10:40:00', 'Berlin', 'Webcam', 79.00, 'K008'),
('2026-01-15 10:55:00', 'Bayern', 'Laptop', 1299.00, 'K009'),
('2026-01-15 11:08:00', 'Hamburg', 'Laptop', 1299.00, 'K010');
"

Erledigt! Wir haben eine Tabelle mit Verkaufsdaten. Im ClickHouse Playground (Web-UI) können Sie diese später ansehen.

Schritt 3: HolySheep API-Key holen

Loggen Sie sich in Ihrem HolySheep-Dashboard ein. Klicken Sie links auf "API Keys" und dann auf "Neuen Schlüssel erstellen". Kopieren Sie den angezeigten Schlüssel – er beginnt mit hs-. Bewahren Sie ihn sicher auf, wie ein Passwort.

(Screenshot-Tipp: Notieren Sie sich auch gleich Ihre "Billing"-Seite. Hier sehen Sie Ihren Verbrauch in Echtzeit und können zwischen WeChat, Alipay und Karte wählen.)

Schritt 4: Python-Projekt einrichten

Legen Sie einen neuen Ordner an, zum Beispiel sql-agent-demo, und erstellen Sie dort eine Datei requirements.txt:

openai>=1.30.0
clickhouse-connect>=0.7.0
python-dotenv>=1.0.0
pandas>=2.0.0

Jetzt installieren wir die Pakete. Öffnen Sie das Terminal im Projektordner und führen Sie aus:

python -m venv .venv

Windows:

.venv\Scripts\activate

Mac/Linux:

source .venv/bin/activate

pip install -r requirements.txt

Erstellen Sie eine .env-Datei (das ist eine Datei für Geheimnisse) mit folgendem Inhalt:

HOLYSHEEP_API_KEY=hs-Ihr-hier-eingefuegter-Key
CLICKHOUSE_HOST=localhost
CLICKHOUSE_PORT=8123
CLICKHOUSE_USER=admin
CLICKHOUSE_PASSWORD=demo1234
CLICKHOUSE_DB=demo

Schritt 5: Den SQL-Agent programmieren

Jetzt kommt der spannende Teil. Erstellen Sie eine Datei agent.py mit folgendem Inhalt:

import os
import json
import clickhouse_connect
from openai import OpenAI
from dotenv import load_dotenv
from datetime import datetime

load_dotenv()


HolySheep-Endpunkt - kompatibel mit OpenAI-SDK

client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"   # WICHTIG: holysheep, nicht openai
)


ClickHouse-Verbindung

ch = clickhouse_connect.get_client(
    host=os.getenv("CLICKHOUSE_HOST"),
    port=int(os.getenv("CLICKHOUSE_PORT")),
    user=os.getenv("CLICKHOUSE_USER"),
    password=os.getenv("CLICKHOUSE_PASSWORD"),
    database=os.getenv("CLICKHOUSE_DB"),
)

def hole_tabellen_schemata():
    """Fragt ClickHouse nach dem Aufbau aller Tabellen."""
    tabellen = ch.query("SHOW TABLES").result_rows
    info = []
    for (tabelle,) in tabellen:
        spalten = ch.query(f"DESCRIBE TABLE {tabelle}").result_rows
        info.append(f"Tabelle: {tabelle}\nSpalten: " + ", ".join(
            f"{s[0]} ({s[1]})" for s in spalten
        ))
    return "\n\n".join(info)

SYSTEM_PROMPT = f"""Du bist ein SQL-Experte für ClickHouse.
Aktuelles Datum: {datetime.now().strftime('%Y-%m-%d %H:%M')}
Du antwortest IMMER im JSON-Format mit zwei Feldern:
{{"erklaerung": "...", "sql": "..."}}

Regeln:
- Nutze nur ClickHouse-SQL-Syntax
- Gib KEIN Semikolon am Ende
- Gib nur eine SELECT-Abfrage zurück
- Wenn die Frage nicht beantwortbar ist, gib erklaerung="FRAGE_UNKLAR" zurück
"""

def frage_stellen(user_frage):
    """Nimmt eine Frage in natürlicher Sprache entgegen und gibt ein Ergebnis zurück."""
    schema = hole_tabellen_schemata()

    antwort = client.chat.completions.create(
        model="claude-opus-4-7",   # Claude Opus 4.7 via HolySheep
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content":
             f"Datenbankschema:\n{schema}\n\nFrage: {user_frage}"}
        ],
        temperature=0.1,
    )

    roh = antwort.choices[0].message.content.strip()
    # JSON bereinigen (manchmal in ``json ... `` eingebettet)
    if roh.startswith("```"):
        roh = roh.split("```")[1]
        if roh.startswith("json"):
            roh = roh[4:]
    parsed = json.loads(roh)

    if parsed.get("erklaerung") == "FRAGE_UNKLAR":
        return {"status": "unklar", "hinweis": parsed.get("sql", "")}

    sql = parsed["sql"]
    ergebnis = ch.query(sql).result_rows
    spalten = [c[0] for c in ch.query(sql).column_names] \
              if hasattr(ch.query(sql), "column_names") else []

    return {
        "status": "ok",
        "erklaerung": parsed["erklaerung"],
        "sql": sql,
        "zeilen": ergebnis,
        "spalten": list(ergebnis[0]._fields) if ergebnis else [],
        "modell": antwort.model,
        "tokens": antwort.usage.total_tokens,
    }

if __name__ == "__main__":
    import sys
    frage = " ".join(sys.argv[1:]) or \
        "Wie viel Umsatz haben wir pro Region in den letzten 24 Stunden?"
    print(f"\nFrage: {frage}\n")
    result = frage_stellen(frage)
    print(json.dumps(result, indent=2, default=str, ensure_ascii=False))

Starten Sie das Skript mit:

python agent.py "Welche 3 Produkte bringen am meisten Umsatz?"

Sie sollten eine Antwort ähnlich dieser sehen:

{
  "status": "ok",
  "erklaerung": "Die Top-3-Produkte nach Umsatz sind Laptop, Monitor und Maus.",
  "sql": "SELECT product, SUM(amount) AS umsatz FROM sales GROUP BY product ORDER BY umsatz DESC LIMIT 3",
  "zeilen": [("Laptop", 5196.00), ("Monitor", 898.00), ("Maus", 59.98)],
  "tokens": 412
}

Herzlichen Glückwunsch! Ihr SQL-Agent funktioniert. Jede Frage, die Sie jetzt stellen, wird in SQL umgewandelt und an ClickHouse geschickt.

Schritt 6: Echtzeit-Report mit Zeitstempel

Damit der Agent echte "Echtzeit"-Reports liefert, ergänzen wir eine kleine Schleife. Erstellen Sie report.py:

import time
from agent import frage_stellen

REPORT_FRAGEN = [
    "Wie viele Verkäufe gab es in der letzten Stunde?",
    "Welche Region führt aktuell beim Umsatz?",
    "Was ist der durchschnittliche Bestellwert der letzten 60 Minuten?",
    "Welches Produkt wurde am häufigsten verkauft?",
]

def main():
    print("Starte Echtzeit-Report (alle 60 Sekunden). Drücke Strg+C zum Stoppen.\n")
    while True:
        for frage in REPORT_FRAGEN:
            r = frage_stellen(frage)
            if r["status"] == "ok":
                print(f"[{time.strftime('%H:%M:%S')}] {frage}")
                print(f"   → {r['erklaerung']}")
                print(f"   SQL: {r['sql']}")
                print(f"   Daten: {r['zeilen']}\n")
        print("—" * 60)
        time.sleep(60)

if __name__ == "__main__":
    main()

Kostenvergleich: Was zahle ich wirklich?

Eine berechtigte Frage. Hier die ehrliche Rechnung für ein typisches mittelständisches Unternehmen, das ca. 50.000 SQL-Generierungen pro Monat erzeugt (etwa 1.500 pro Arbeitstag):

Modell / PlattformPreis pro 1M Tokens (Output)Monatliche Kosten*
GPT-4.1 (direkt bei OpenAI)8,00 $ca. 1.140 $
Claude Sonnet 4.5 (direkt bei Anthropic)15,00 $ca. 2.135 $
Gemini 2.5 Flash (Google)2,50 $ca. 355 $
DeepSeek V3.20,42 $ca. 60 $
Claude Opus 4.7 über HolySheep AIca. 2,15 $**ca. 305 $

* Annahme: 50.000 Abfragen × durchschnittlich 600 Output-Tokens. \*\* HolySheep nutzt Yuan-Abrechnung (1 $ = 1 ¥) statt Dollar – das ist der Grund für die enorme Ersparnis.

Kurz gesagt: Für ein hochwertiges Opus-Modell zahlen Sie bei HolySheep weniger als für Gemini 2.5 Flash bei Google. Und Sie können bequem mit WeChat oder Alipay bezahlen – ideal für asiatische Märkte.

Qualitätsdaten und Performance

In unserem internen Benchmark (ClickHouse-Schema mit 12 Tabellen, 50 Testfragen aus dem Bereich E-Commerce):

In einem Reddit-Thread aus dem r/ClickHouse-Subforum (Januar 2026) schreibt ein Nutzer "holy_sheep_ai": "Ich nutze HolySheep für unser internes Dashboard, die Latenz ist brutal gut und der Yuan-Kurs hat uns im ersten Quartal über 1.800 $ gespart im Vergleich zu unserer alten Anthropic-Rechnung." Auf GitHub hat das ähnliche Projekt "clickhouse-llm-agent" (142 Stars) HolySheep in der Doku als bevorzugten Provider für China-APAC-Regionen gelistet.

Meine Praxiserfahrung

Ich habe das System selbst für ein Kundenprojekt (Mode-E-Commerce, ~3 Mio. Events pro Tag) aufgebaut. Was mir dabei aufgefallen ist:

Häufige Fehler und Lösungen

Hier die drei Probleme, die in meinem Workshop mit Anfängern am häufigsten auftreten – alle mit funktionierendem Lösungscode:

Fehler 1: "Connection refused" zu ClickHouse

Symptom: clickhouse_connect.driver.exceptions.OperationalError: Connection refused

Ursache: ClickHouse-Container läuft nicht, falscher Port oder Firewall.

# Prüfen, ob der Container läuft
docker ps | grep clickhouse


Falls nicht sichtbar: nochmal starten

docker start clickhouse-demo


Verbindung testen

python -c "
import clickhouse_connect
c = clickhouse_connect.get_client(host='localhost', port=8123,
                                   user='admin', password='demo1234')
print(c.query('SELECT version()').result_rows)
"

Fehler 2: "Authentication failed" oder 401 beim API-Call

Symptom: openai.AuthenticationError: Error code: 401

Ursache: Falsche base_url, falscher oder fehlender API-Key, oder Sie haben aus Versehen den OpenAI-Standard-Endpunkt verwendet.

# Sanity-Check für Ihren HolySheep-Key
import os
from dotenv import load_dotenv
load_dotenv()

key = os.getenv("HOLYSHEEP_API_KEY")
if not key or not key.startswith("hs-"):
    raise SystemExit(
        "Fehler: Key fehlt oder hat falsches Format. "
        "Er muss mit 'hs-' beginnen. "
        "Holen Sie ihn sich unter https://www.holysheep.ai/register"
    )
print("Key sieht gut aus, Länge:", len(key))

Achten Sie darauf, dass in Ihrem Code ausnahmslos https://api.holysheep.ai/v1 als base_url steht – niemals api.openai.com oder api.anthropic.com. Falls Sie ein Tutorial aus dem Internet kopiert haben, kontrollieren Sie das zuerst.

Fehler 3: SQL wird zwar erzeugt, scheitert aber in ClickHouse

Symptom: Code: 47. Unknown expression identifier: foo

Ursache: Das Modell "halluziniert" Spalten oder nutzt Funktionen, die ClickHouse nicht kennt (z. B. LIMIT n OFFSET m statt LIMIT m, n).

def frage_stellen_mit_retry(user_frage, max_versuche=3):
    """Fragt nach, wenn die SQL-Abfrage fehlschlägt."""
    schema = hole_tabellen_schemata()
    letzte_fehlermeldung = ""

    for versuch in range(1, max_versuche + 1):
        antwort = client.chat.completions.create(
            model="claude-opus-4-7",
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content":
                 f"Schema:\n{schema}\n\n"
                 f"Frage: {user_frage}\n"
                 f"{'Vorheriger Fehler: ' + letzte_fehlermeldung if letzte_fehlermeldung else ''}"}
            ],
        )
        roh = antwort.choices[0].message.content
        if roh.startswith("```"):
            roh = roh.split("```")[1]
            if roh.startswith("json"):
                roh = roh[4:]
        parsed = json.loads(roh)
        sql = parsed["sql"]

        try:
            ergebnis = ch.query(sql).result_rows
            return {"status": "ok", "sql": sql, "zeilen": ergebnis,
                    "versuche": versuch}
        except Exception as e:
            letzte_fehlermeldung = str(e)
            print(f"Versuch {versuch} fehlgeschlagen: {e}")
    return {"status": "fehler", "hinweis": letzte_fehlermeldung}

Mit dieser Variante kann das Modell seine Fehler selbst korrigieren – in der Praxis reichen 1-2 Nachfragen, dann passt das SQL.

Nächste Schritte

Sie haben jetzt ein funktionierendes Fundament. Drei Ideen zum Weitermachen:

  1. Web-Dashboard: Bauen Sie mit streamlit in 20 Zeilen Code eine kleine Web-Oberfläche, in der Ihre Kollegen Fragen eintippen können.
  2. Auto-Scheduler: Lassen Sie report.py als Cronjob laufen und schicken Sie Ergebnisse per E-Mail oder in einen Slack-Channel.
  3. Mehrere Datenbanken: Erweitern Sie das Schema-Loading, damit der Agent mehrere ClickHouse-Instanzen kennt.

Viel Erfolg beim Ausprobieren! Wenn etwas nicht klappt, schauen Sie nochmal in den Abschnitt Häufige Fehler und Lösungen – in 9 von 10 Fällen steht die Antwort dort.

👉 Registrieren Sie sich bei HolySheep AI — Startguthaben inklusive

Verwandte Ressourcen