So verbessern Sie Ihr dbt-Projekt mit großen Sprachmodellen

Kurz zusammengefasst

Sie können typische Natural Language Processing-Aufgaben (Klassifizierung, Sentimentanalyse usw.) für Ihre Textdaten automatisch mit LLM lösen, und das für nur 10 $ pro 1 Mio. Zeilen (je nach Aufgabe und Modell), während Sie in Ihrer dbt-Umgebung bleiben. Anweisungen, Details und Code finden Sie weiter unten.

Wenn Sie dbt als Transformationsebene verwenden, kann es vorkommen, dass Sie aus unstrukturierten Textdaten aussagekräftige Informationen extrahieren möchten. Zu diesen Daten können Kundenrezensionen, Titel, Beschreibungen, Google Analytics-Quellen/-Medien usw. gehören. Sie möchten sie möglicherweise in Gruppen kategorisieren oder Stimmungen und Töne abrufen.

Mögliche Lösungen wären

• Anwenden von Machine-Learning-Modellen (oder Aufrufen eines LLM) außerhalb des dbt-Flows
• Definieren Sie einfache Kategorisierungen innerhalb von dbt-Modellen mithilfe von CASE WHEN-Anweisungen
• Definieren Sie Kategorien im Voraus und laden Sie sie entweder in Ihre Rohdatenbankebene hoch oder nutzen Sie die dbt-Seed-Funktionalität

Da sich Python-dbt-Modelle weiterentwickeln, gibt es noch eine weitere Lösung: Sie können diese Aufgaben zur Verarbeitung natürlicher Sprache als eines der dbt-Modelle in Ihrer dbt-Umgebung behalten.

Wenn das für Sie hilfreich sein könnte, finden Sie unten eine Schritt-für-Schritt-Anleitung zur Verwendung der OpenAI-API in Ihrem dbt-Projekt. Sie können alles aus dieser Anleitung in Ihrer Umgebung reproduzieren, indem Sie den Code und die Datenbeispiele aus dem GitHub-Repository verwenden (siehe Links am Ende).

Umgebung einrichten

Wenn Sie bereits ein dbt-Projekt und Daten haben oder die Ergebnisse nicht reproduzieren möchten, springen Sie zu (4) oder überspringen Sie diesen Abschnitt vollständig. Andernfalls benötigen Sie Folgendes:

1. Richten Sie das dbt-Projekt ein . Offizielle Dokumente

   1. Sie können einfach das von mir für diese Anleitung vorbereitete Dokument von GitHub klonen.

   2. Vergessen Sie nicht, Ihre Datei „profiles.yml“ zu erstellen/aktualisieren.

      
      

2. Richten Sie die Datenbank ein . Ich habe Snowflake verwendet. Leider gibt es keine kostenlose Version, aber es gibt eine 30-tägige kostenlose Testversion .

   1. Derzeit funktionieren dbt-Python-Modelle nur mit Snowflake, Databricks und BigQuery (kein PostgreSQL). Daher sollte dieses Tutorial für alle davon funktionieren, obwohl einige Details abweichen können.

      
      

3. Vorbereiten der Quelldaten

   1. Als Datensatz habe ich Metadaten eines R-Pakets verwendet, die im TidyTuesday-Repository veröffentlicht wurden.

      1. Sie können es hier herunterladen. Details zum Datensatz finden Siehier
      2. Alternativ können Sie eine leichte Version aus meinem Repository hier verwenden

   2. Laden Sie es in Ihre Datenbank hoch.

   3. Aktualisieren Sie die Datei source.yml im dbt-Projekt, damit sie mit Ihren Datenbank- und Schemanamen übereinstimmt.

      
      

4. Holen Sie sich den OpenAI-API-Schlüssel

   1. Befolgen Sie die Schnellstartanweisungen aus den offiziellen Dokumenten .

   2. Nicht: Es ist nicht kostenlos, aber es ist ein Pay-as-you-go-Modell. Für den Testdatensatz mit 10 Zeilen werden Ihnen während Ihrer Experimente also nicht mehr als 1 US-Dollar berechnet.

   3. Um besonders vorsichtig zu sein, legen Sie ein Ausgabenlimit fest.

      
      

5. Einrichten der externen Zugriffsintegration in Snowflake

   1. Dies gilt nur, wenn Sie Snowflake verwenden.
   2. Wenn dies nicht erfolgt, können die dbt-Python-Modelle nicht auf APIs im Internet zugreifen (einschließlich der OpenAI-API).
   3. Befolgen Sie die offiziellen Anweisungen .
   4. Speichern Sie den OpenAI-API-Schlüssel in dieser Integration.

Erstellen Sie eine Liste mit Kategorien

Wenn Sie eine Klassifizierungsaufgabe lösen, benötigen Sie zunächst Kategorien (auch Klassen genannt), die Sie in Ihrer LLM-Eingabeaufforderung verwenden können. Im Grunde sagen Sie: „Ich habe eine Liste dieser Kategorien. Könnten Sie definieren, zu welcher dieser Text gehört?“

Einige Optionen hier:

1. Manuelles Erstellen einer Liste vordefinierter Kategorien

   1. Es ist geeignet, wenn Sie stabile und vorhersehbare Kategorien benötigen.

   2. Vergessen Sie nicht, hier „Andere“ hinzuzufügen, damit LLM im Zweifelsfall über diese Optionen verfügt.

   3. Bitten Sie LLM in Ihrer Eingabeaufforderung, einen Kategorienamen vorzuschlagen, wenn die Kategorie „Sonstige“ verwendet wird.

   4. Laden Sie eine vordefinierte Liste in die Rohebene der Datenbank oder als CSV in Ihr dbt-Projekt hoch (unter Verwendung von dbt seed ).

      
      

2. Geben Sie LLM eine Stichprobe Ihrer Daten und bitten Sie es, N Kategorien zu erstellen.

   1. Derselbe Ansatz wie der vorherige, aber wir bekommen Hilfe bei der Liste.

   2. Wenn Sie GPT verwenden, ist es aus Gründen der Reproduzierbarkeit besser, hier Seed zu verwenden.

      
      

3. Verzichten Sie auf vordefinierte Kategorien und überlassen Sie LLM die Arbeit unterwegs.

   1. Dies kann zu weniger vorhersehbaren Ergebnissen führen.

   2. Gleichzeitig ist es gut genug, wenn Sie mit einer gewissen Zufälligkeitsmarge einverstanden sind.

   3. Im GPT-Anwendungsfall ist es besser, die Temperatur auf 0 zu setzen, um bei einer erneuten Ausführung unterschiedliche Ergebnisse zu vermeiden.

      
      

In diesem Blogbeitrag werde ich mich für die dritte Option entscheiden.

Erstellen Sie ein dbt-Python-Modell zum Aufrufen der OpenAI-API

Kommen wir nun zum Kern dieses Beitrags und erstellen ein dbt-Modell, das neue Textdaten aus der Upstream-Tabelle übernimmt, sie in die OpenAI-API einspeist und die Kategorie in der Tabelle speichert.

Wie oben erwähnt, werde ich den Datensatz „R-Pakete“ verwenden. R ist eine sehr beliebte Programmiersprache in der Datenanalyse. Dieser Datensatz enthält Informationen zu R-Paketen aus dem CRAN-Projekt, wie Version, Lizenz, Autor, Titel, Beschreibung usw. Wir sind am title interessiert, da wir für jedes Paket basierend auf seinem Titel eine Kategorie erstellen werden.

1. Bereiten Sie die Basis für das Modell vor

   • Die dbt-Konfiguration kann über die Methode dbt.config(...) übergeben werden.

     
     

   • Es gibt zusätzliche Argumente in dbt.config, beispielsweise ist packages eine Paketanforderung.

     
     

   • Das dbt-Python-Modell kann auf Upstream-Modelle verweisen dbt.ref('...') oder dbt.source('...')

     
     

   • Es muss ein DataFrame zurückgegeben werden. Ihre Datenbank speichert es als Tabelle.

   
   

    import os import openai import pandas as pd COL_TO_CATEGORIZE = 'title' def model(dbt, session): import _snowflake dbt.config( packages=['pandas', 'openai'], ) df = dbt.ref('package').to_pandas() df.drop_duplicates(subset=[COL_TO_CATEGORIZE], inplace=True) return df

2. Mit OpenAI API verbinden

   • Wir müssen secrets und external_access_integrations an die dbt.config übergeben. Sie enthält die geheime Referenz, die in Ihrer Snowflake External Access Integration gespeichert ist.

     
     

   • Hinweis: Diese Funktion wurde erst vor einigen Tagen veröffentlicht und ist nur in der Beta-Version dbt 1.8.0-b3 verfügbar.

    dbt.config( packages=['pandas', 'openai'], secrets={'openai_key': 'openai_key', 'openai_org': 'openai_org'}, external_access_integrations=['openai_external_access_integration'], ) client = openai.OpenAI( api_key=_snowflake.get_generic_secret_string('openai_key'), organization=_snowflake.get_generic_secret_string('openai_org'), )

3. Machen Sie das dbt-Modell inkrementell und deaktivieren Sie vollständige Aktualisierungen.

   • Dieser Teil ist wichtig, um die Kosten der OpenAI-API niedrig zu halten.
   • Dadurch wird verhindert, dass derselbe Text mehrfach kategorisiert wird.
   • Andernfalls senden Sie bei jeder Ausführung von dbt run vollständige Daten an OpenAI, was mehrmals täglich vorkommen kann.
   • Wir fügen materialized='incremental' , incremental_strategy='append' , full_refresh = False zu dbt.config hinzu
   • Jetzt wird der vollständige Scan nur für den ersten dbt-Lauf durchgeführt und bei den späteren Läufen (unabhängig davon, ob inkrementell oder mit vollständiger Aktualisierung) wird nur Delta kategorisiert.
   • Wenn Sie besonders vorsichtig sein möchten, können Sie Ihre Daten ein wenig vorverarbeiten, um die Anzahl eindeutiger Einträge zu verringern. Vermeiden Sie jedoch eine zu umfassende Vorverarbeitung, da LLMs mit natürlicher Sprache besser funktionieren.

    dbt.config( materialized='incremental', incremental_strategy='append', full_refresh = False, packages=['pandas', 'openai'], secrets={'openai_key': 'openai_key', 'openai_org': 'openai_org'}, external_access_integrations=['openai_external_access_integration'], ) if dbt.is_incremental: pass

   
   

4. Inkrementalitätslogik hinzufügen

   • Beim inkrementellen Durchlauf (aufgrund unseres Setups bedeutet dies bei jedem Durchlauf außer dem ersten) müssen wir alle bereits kategorisierten Titel entfernen.
   • Wir können dies einfach tun, indem wir dbt.this verwenden. Ähnlich wie bei normalen inkrementellen Modellen.

    if dbt.is_incremental: categorized_query = f''' SELECT DISTINCT "{ COL_TO_CATEGORIZE }" AS primary_key FROM { dbt.this } WHERE "category" IS NOT NULL ''' categorized = [row.PRIMARY_KEY for row in session.sql(categorized_query).collect()] df = df.loc[~df[COL_TO_CATEGORIZE].isin(categorized), :]

5. Rufen Sie die OpenAI-API stapelweise auf

   • Um die Kosten zu senken, ist es besser, Daten stapelweise an die OpenAI API zu senden.
   • Die Systemaufforderung kann 5-mal größer sein als der Text, den wir klassifizieren müssen. Wenn wir die Systemaufforderung für jeden Titel separat senden, führt dies zu einem viel höheren Token-Verbrauch für sich wiederholende Dinge.
   • Die Batchgröße sollte jedoch nicht zu groß sein. Bei großen Batches liefert GPT weniger stabile Ergebnisse. Aus meinen Experimenten geht hervor, dass Batchgröße = 5 gut genug funktioniert.
   • Um sicherzustellen, dass die Antwort die entsprechende Größe nicht überschreitet, habe ich außerdem die max_tokens Beschränkung hinzugefügt.

    BATCH_SIZE = 5 n_rows = df.shape[0] categories = [None for idx in range(n_rows)] for idx in range(0, n_rows, BATCH_SIZE): df_sliced = df.iloc[idx:idx+BATCH_SIZE, :] user_prompt = f'```{ "|".join(df_sliced[COL_TO_CATEGORIZE].to_list()) }```' chat_completion = client.chat.completions.create( messages=[ {'role': 'system', 'content': SYSTEM_PROMPT}, {'role': 'user', 'content': user_prompt} ], model='gpt-3.5-turbo', temperature=0, max_tokens=10*BATCH_SIZE + 2*BATCH_SIZE, ) gpt_response = chat_completion.choices[0].message.content gpt_response = [category.strip() for category in gpt_response.split('|')] categories[idx:idx + len(gpt_response)] = gpt_response df['category'] = categories df.dropna(subset=['category'], inplace=True)

   
   

6. Es ist Zeit, über eine Eingabeaufforderung für LLM zu sprechen. Das ist, was ich bekommen habe:

Sie erhalten eine Liste mit Titeln der CRAN R-Pakete in ```-Klammern. Die Titel werden durch das Zeichen "|" getrennt. Überlegen Sie sich für jeden Titel eine Kategorie. Es werden nur Kategorienamen zurückgegeben, die durch das Zeichen "|" getrennt sind.

• Bringen Sie die Anweisungen direkt auf den Punkt.
• Verwenden Sie die ```-Technik, um SQL-Injections zu vermeiden.
• Machen Sie sich das Ergebnisformat klar. In meinem Fall habe ich "|" als Trennzeichen für Ein- und Ausgaben angefordert.

7. Endgültiger DBT-Modellcode

    import os import openai import pandas as pd SYSTEM_PROMPT = '''You will be provided a list of CRAN R package titles in ``` brackets. Titles will be separated by "|" sign. Come up with a category for each title. Return only category names separated by "|" sign. ''' COL_TO_CATEGORIZE = 'title' BATCH_SIZE = 5 def model(dbt, session): import _snowflake dbt.config( materialized='incremental', incremental_strategy='append', full_refresh = False, packages=['pandas', 'openai'], secrets={'openai_key': 'openai_key', 'openai_org': 'openai_org'}, external_access_integrations=['openai_external_access_integration'], ) client = openai.OpenAI( api_key=_snowflake.get_generic_secret_string('openai_key'), organization=_snowflake.get_generic_secret_string('openai_org'), ) df = dbt.ref('package').to_pandas() df.drop_duplicates(subset=[COL_TO_CATEGORIZE], inplace=True) if dbt.is_incremental: categorized_query = f''' SELECT DISTINCT "{ COL_TO_CATEGORIZE }" AS primary_key FROM { dbt.this } WHERE "category" IS NOT NULL ''' categorized = [row.PRIMARY_KEY for row in session.sql(categorized_query).collect()] df = df.loc[~df[COL_TO_CATEGORIZE].isin(categorized), :] n_rows = df.shape[0] categories = [None for idx in range(n_rows)] for idx in range(0, n_rows, BATCH_SIZE): df_sliced = df.iloc[idx:idx+BATCH_SIZE, :] user_prompt = f'```{ "|".join(df_sliced[COL_TO_CATEGORIZE].to_list()) }```' chat_completion = client.chat.completions.create( messages=[ {'role': 'system', 'content': SYSTEM_PROMPT}, {'role': 'user', 'content': user_prompt} ], model='gpt-3.5-turbo', temperature=0, max_tokens=10*BATCH_SIZE + 2*BATCH_SIZE, ) gpt_response = chat_completion.choices[0].message.content gpt_response = [category.strip() for category in gpt_response.split('|')] categories[idx:idx + len(gpt_response)] = gpt_response df['category'] = categories df.dropna(subset=['category'], inplace=True) return df

Kostenschätzungen

Die Preise für die OpenAI API sind hier aufgeführt. Die Gebühren werden für die Anzahl der angeforderten und zurückgegebenen Token berechnet. Das Token ist eine Instanz, die mit einer Anzahl von Zeichen in Ihrer Anfrage korreliert. Es gibt Open-Source-Pakete, um eine Anzahl von Token für einen bestimmten Text auszuwerten. Zum Beispiel Tiktoken . Wenn Sie es manuell auswerten möchten, finden Sie hier einen offiziellen OpenAI-Tokenizer.

In unserem Datensatz gibt es ca. 18.000 Titel. Das entspricht ungefähr 320.000 Eingabetoken (180.000 Titel und 140.000 Systemaufforderungen, wenn wir eine Batchgröße von 5 verwenden) und 50.000 Ausgabetoken. Je nach Modell betragen die Kosten für den vollständigen Scan:

1. GPT-4 Turbo : 4,7 $ . Preis: Eingabe: 10 $/1 Mio. Token; Ausgabe: 30 $/1 Mio. Token.
2. GPT-4 : 12,6 $. Preisgestaltung: Eingabe: 30 $/1 Mio. Token; Ausgabe: 60 $/1 Mio. Token.
3. GPT-3.5 Turbo : 0,2 $. Preis: Eingabe: 0,5 $ / 1 Mio. Token; Ausgabe: 1,5 $ / 1 Mio. Token.

Ergebnisse

Das dbt-Modell funktionierte einwandfrei. Ich konnte alle 18.000 Pakete lückenlos kategorisieren. Das Modell erwies sich als kosteneffizient und schützte vor mehrfachen dbt-Läufen.

Ich habe das Ergebnis-Dashboard hier auf Tableau Public veröffentlicht. Sie können gerne damit experimentieren, die Daten herunterladen und darauf aufbauend alles erstellen, was Sie möchten.

Einige interessante Details, die ich gefunden habe:

• Die Top-1-Kategorie ist Data Visualization (1.190 Pakete oder 6 %). Ich denke, das beweist die Popularität von R als Visualisierungstool, insbesondere mit Paketen wie Shiny, Plotly und anderen.

• Die beiden am stärksten wachsenden Kategorien im Jahr 2023 waren Data Import und Data Processing . Es klingt, als würde R zunehmend als Datenverarbeitungstool verwendet.

• Das größte Wachstum im Vergleich zum Vorjahr unter den Top-30-Kategorien verzeichnete 2019 Natural Language Processing . Zwei Jahre nach dem berühmten Paper „Attention Is All You Need“ und ein halbes Jahr nach der Veröffentlichung von GPT-1 :)

Weitere Ideen

1. Wir können einen alternativen Ansatz verwenden – die GPT-Einbettungen .

   • Es ist viel billiger.

   • Allerdings ist dieser Teil mit mehr technischem Aufwand verbunden, da Sie den Klassifizierungsteil selbst durchführen sollten (bleiben Sie dran, ich werde diese Option in einem der nächsten Beiträge untersuchen).

     
     

2. Natürlich ist es sinnvoll, diesen Teil aus dbt zu entfernen und ihn in Cloud-Funktionen oder die von Ihnen verwendete Infrastruktur zu verschieben. Wenn Sie ihn gleichzeitig unter dbt behalten möchten, hilft Ihnen dieser Beitrag weiter.

   
   

3. Vermeiden Sie das Hinzufügen jeglicher Logik zum Modell. Es sollte eine Aufgabe erfüllen – LLM aufrufen und das Ergebnis speichern. So vermeiden Sie, es erneut ausführen zu müssen.

   
   

4. Die Wahrscheinlichkeit ist hoch, dass Sie in Ihrem dbt-Projekt viele Umgebungen verwenden. Sie müssen darauf achten und vermeiden, dieses Modell bei jedem Pull Request in jeder Entwicklerumgebung immer wieder auszuführen.

   • Dazu können Sie eine Logik mit if dbt.config.get("target_name") == 'dev' einbinden.

     
     

5. Antworten mit einem Trennzeichen können instabil sein.

   • Beispielsweise kann GPT weniger Elemente zurückgeben als erwartet und es wird schwierig sein, ursprüngliche Titel der Kategorienliste zuzuordnen.

   • Um dies zu umgehen, fügen Sie Ihrer Anfrage response_format={ "type": "json_object" } hinzu, um eine JSON-Ausgabe anzufordern. Weitere Informationen finden Sie in den offiziellen Dokumenten .

   • Mit der JSON-Ausgabe können Sie in der Eingabeaufforderung eine Antwort im Format {"title": "category"} anfordern und diese dann Ihren Anfangswerten zuordnen.

   • Beachten Sie, dass es teurer wird, da die Antwortgröße zunimmt.

   • Seltsamerweise sank die Qualität der Klassifizierung dramatisch, als ich für GPT 3.5 Turbo auf JSON umstieg.

     
     

6. In Snowflake gibt es eine Alternative – die Verwendung der Funktion cortex.complete() . Sehen Sie sich einen tollen Beitrag von Joel Labes im dbt-Blog an.

   
   

Das ist es! Sagen Sie mir, was Sie denken.

Links

Vollständiger Code auf GitHub: Link

Öffentliches Tableau-Dashboard: Link

TidyTuesday R-Datensatz:Link