30-Tage-DSPy-Challenge – Tag 27 & 29: Abschlussprojekt – Entwicklung eines SQL-Query-Generators

Die Tage 27 bis 29 der 30-Tage-DSPy-Challenge widmen ich mich der praktischen Anwendung der erlernten Konzepte in einem Abschlussprojekt. In diesem Beitrag werde ich die Umsetzung des SQL-Query-Generators umsetzen. Ziel des Systems ist es, natürlichsprachliche Fragen unter Berücksichtigung eines definierten Datenbankschemas automatisch in syntaktisch korrekte SQL-Abfragen zu übersetzen.

Konfiguration der Umgebung

Zu Beginn wird die Laufzeitumgebung eingerichtet. Für dieses Projekt kommt ein lokales Sprachmodell (gemma-3-4b-it) zum Einsatz, welches über eine lokale API angesprochen wird. Das Modell läuft in einem llama.cpp Server und kann über localhost Port 8080 mit der OpenAI API angesprochen werden.

import dspy
from datasets import load_dataset

# Konfiguration des lokalen LLms
local_llm = dspy.LM(
    "openai/unsloth/gemma-3-4b-it-GGUF:Q4_K_M", 
    api_base="http://localhost:8080/v1", 
    api_key="no_key_needed",
    temperature=0.1,
    cache=False,
)

dspy.configure(lm=local_llm)

Definition der Signatur

Die Kernkomponente der semantischen Verarbeitung bildet die Signatur TextToSQL. Sie definiert die Schnittstelle zwischen der unstrukturierten Eingabe und der strukturierten Ausgabe.

• Eingabefelder
  • context
    Enthält das Datenbankschema (Tabellen, Spalten, Datentypen und Beziehungen). Dies ist notwendig, damit das Modell weiß, worauf sich die Frage bezieht.
  • question
    Die natürlichsprachliche Anfrage des Nutzers.
• Ausgabefeld
  • sql_query
    Die generierte SQL-Anweisung.

class TextToSQL(dspy.Signature):
    """
    Übersetzt eine natürlichsprachliche Frage in eine SQL-Abfrage basierend auf dem gegebenen Schema.
    """
    context = dspy.InputField(desc="Das Schema der Datenbank (Tabellen, Spalten, Typen).")
    question = dspy.InputField(desc="Die Frage, die mittels SQL beantwortet werden soll.")
    sql_query = dspy.OutputField(desc="Die generierte SQL-Abfrage.")

Implementierung des Moduls

Das Modul SQLGenerator kapselt die Verarbeitungslogik. Anstelle von dspy.Predict wird hier dspy.ChainOfThought verwendet. Die „Chain of Thought“-Strategie (Gedankenkette) ist bei der SQL-Generierung von Vorteil, da sie das Modell dazu anleitet, die logischen Schritte (z. B. Auswahl der Tabelle → Identifikation der Spalten → Formulierung der Bedingungen) explizit zu durchlaufen, bevor der finale SQL-Code generiert wird.

class SQLGenerator(dspy.Module):
    def __init__(self):
        super().__init__()
        # ChainOfThought fördert die logische Herleitung der Query
        self.generate = dspy.ChainOfThought(TextToSQL)

    def forward(self, question, context):
        return self.generate(question=question, context=context)

Bereitstellung von Kontext und Trainingsdaten

Für die Few-Shot-Optimierung werden ein Datenbankschema und Beispiele benötigt. Das Schema definiert zwei Tabellen (users und orders) inklusive Fremdschlüsselbeziehungen. Die train_examples dienen als Demonstrationen, wie Fragen basierend auf diesem Schema in SQL übersetzt werden sollen.

Es werden drei Fälle abgedeckt:

1. Eine einfache SELECT-Abfrage mit WHERE-Bedingung.
2. Eine Aggregatfunktion (SUM).
3. Ein JOIN über zwei Tabellen.

# Definition des Datenbank-Kontexts
db_schema = """
Tabelle: users
- id (INTEGER)
- name (VARCHAR)
- signup_date (DATE)
- country (VARCHAR)

Tabelle: orders
- id (INTEGER)
- user_id (INTEGER, Foreign Key zu users.id)
- amount (DECIMAL)
- order_date (DATE)
"""

# Trainingsbeispiele erstellen
train_examples = [
    dspy.Example(
        context=db_schema,
        question="Zeige alle Nutzer aus Deutschland.",
        sql_query="SELECT * FROM users WHERE country = 'Deutschland';"
    ).with_inputs('context', 'question'),

    dspy.Example(
        context=db_schema,
        question="Wie hoch ist der Gesamtumsatz aller Bestellungen?",
        sql_query="SELECT SUM(amount) FROM orders;"
    ).with_inputs('context', 'question'),

    dspy.Example(
        context=db_schema,
        question="Liste die Namen der Nutzer auf, die mehr als 100 Euro ausgegeben haben.",
        sql_query="SELECT T1.name FROM users AS T1 JOIN orders AS T2 ON T1.id = T2.user_id WHERE T2.amount > 100;"
    ).with_inputs('context', 'question')
]

Optimierung des Programms

Um die Leistung des Modells zu maximieren, wird ein BootstrapFewShot-Optimizer eingesetzt. Dieser wählt automatisch die effektivsten Beispiele (Demonstrationen) aus den Trainingsdaten aus und integriert sie in den Prompt.

Für die Evaluierung wird eine benutzerdefinierte Metrik validate_sql definiert. In diesem Szenario prüfe ich mittels einfachem String-Vergleich, ob die generierte Query exakt mit der erwarteten Query übereinstimmt. In einer Produktionsumgebung könnte hier die Ausführung der Query gegen eine Testdatenbank (Execution Accuracy) präziser überprüft werden.

from dspy.teleprompt import BootstrapFewShot

# Eigene Metrik definieren: Vergleicht die generierte SQL-Query mit dem Beispiel
def validate_sql(example, pred, trace=None):
    # Einfacher String-Vergleich (bereinigt um Whitespace)
    return example.sql_query.strip() == pred.sql_query.strip()

# Initialisierung des Optimizers mit der benutzerdefinierten Metrik
optimizer = BootstrapFewShot(metric=validate_sql)

# Kompilierung des Programms
compiled_sql_generator = optimizer.compile(SQLGenerator(), trainset=train_examples)

Der Kompilierungsprozess durchläuft die Beispiele und optimiert den Prompt für die definierte Metrik.

Test und Ergebnis

Abschließend wird das kompilierte Modell mit einer neuen, im Training nicht gesehenen Frage getestet.

# Test des kompilierten Modells
test_question = "Wann hat sich der Nutzer mit der ID 42 angemeldet?"

prediction = compiled_sql_generator(question=test_question, context=db_schema)

print(f"Frage: {test_question}")
# Zugriff auf den Rationale (Gedankengang), falls vorhanden
print(f"Gedankengang: {getattr(prediction, 'rationale', 'Keine Erklärung generiert')}")
print(f"SQL-Query: {prediction.sql_query}")

Ergebnis:

Frage: Wann hat sich der Nutzer mit der ID 42 angemeldet?
Gedankengang: Keine Erklärung generiert
SQL-Query: SELECT signup_date FROM users WHERE id = 42;

Das System generiert eine korrekte SQL-Abfrage: SELECT signup_date FROM users WHERE id = 42;. Es hat die Entitäten („Nutzer“), die Zielspalte („Wann“ -> signup_date) und die Bedingung („ID 42“) korrekt aus dem Schema und der Frage abgeleitet.

Analyse der Historie

Die mit Hilfe des Befehelhs local_llm.inspect_history(n=10) erzeugte Log-Ausgabe (Siehe day27.ipynb) gewährt einen detaillierten Einblick in den Optimierungsprozess des BootstrapFewShot-Moduls und die anschließende Inferenzphase des DSPy-Programms. Der Prozess beginnt mit der Generierung von logischen Herleitungen, dem sogenannten Chain-of-Thought, für die bereitgestellten Trainingsbeispiele. Hierbei wird deutlich, wie der Optimizer versucht, das Sprachmodell dazu zu bringen, nicht nur die SQL-Lösung zu erraten, sondern den Weg dahin explizit zu formulieren.

Im ersten aufgezeichneten Schritt bearbeitet das Modell die Anfrage nach Nutzern aus Deutschland. Das Modell erkennt korrekt die Notwendigkeit einer Filterung über die Spalte country, trifft jedoch die semantische Entscheidung, den String-Literal „Deutschland“ ins Englische zu „Germany“ zu übersetzen. Da die Validierungsmetrik auf einem strikten String-Vergleich mit dem Trainingsbeispiel basiert, welches „Deutschland“ erwartet, wird dieser an sich logisch korrekte Gedankengang vom Optimizer verworfen. Ein ähnliches Phänomen zeigt sich beim dritten Beispiel, in dem nach Nutzern mit Ausgaben über 100 Euro gefragt wird. Das Modell interpretiert die Frage komplexer als die Trainingsdaten und erstellt eine Abfrage mit einer Aggregation und Gruppierung, um die Gesamtsumme pro Nutzer zu prüfen. Das Trainingsbeispiel hingegen erwartete eine einfachere Filterung auf Einzelbestellungen. Aufgrund dieser Diskrepanz zwischen der Modellinterpretation und der fixen Zielvorgabe schlägt auch hier die Validierung fehl, weshalb die generierte Erklärung nicht in den finalen Prompt übernommen wird.

Erfolgreich verläuft hingegen der zweite Versuch, bei dem der Gesamtumsatz aller Bestellungen ermittelt werden soll. Hier stimmen die Interpretation des Modells und das erwartete SQL-Statement exakt überein. Das Modell identifiziert die Aggregatfunktion SUM auf der Spalte amount als korrekten Weg und generiert das passende SQL. Da die Ausgabe den Validierungstest besteht, speichert der Optimizer diesen spezifischen Gedankengang als erfolgreiches Demonstrationsbeispiel für die spätere Nutzung ab.

Der letzte Abschnitt der Historie zeigt die Anwendung des nun kompilierten Programms auf die Testfrage bezüglich des Nutzers mit der ID 42. Der hier verwendete Prompt zeigt das Ergebnis der vorangegangenen Optimierung. Während die Beispiele zu „Deutschland“ und den „100 Euro“ ohne expliziten Gedankengang als reine Input-Output-Paare präsentiert werden, enthält das Beispiel zum „Gesamtumsatz“ die zuvor validierte Herleitung. Interessanterweise genügt dieses eine vollständige Chain-of-Thought-Beispiel, um das Modell dazu zu animieren, auch für die neue, unbekannte Testfrage eine eigene Herleitung zu generieren. Das Modell analysiert korrekt, dass eine Filterung nach der ID in der Tabelle users notwendig ist, formuliert diesen Schritt verbal und erzeugt anschließend die syntaktisch korrekte SQL-Abfrage. Dies belegt, dass der Optimizer trotz der fehlgeschlagenen Validierungen in den komplexeren Fällen ein funktionierendes System erstellt hat, das die gewünschte „Erst denken, dann antworten“-Struktur erfolgreich auf neue Eingaben überträgt.

Zusammenfassung

Das Projekt demonstriert die Fähigkeit von DSPy, komplexe Übersetzungsaufgaben durch strukturierte Signaturen und Optimierungsalgorithmen zu lösen. Durch die Verwendung von ChainOfThought und BootstrapFewShot konnte ein robustes System erstellt werden, das Kontextwissen (Datenbankschema) effektiv nutzt, um syntaktisch korrekte SQL-Befehle aus natürlicher Sprache zu generieren.