Comment améliorer votre projet dbt avec de grands modèles linguistiques

TL;DR

Vous pouvez résoudre automatiquement les tâches typiques de traitement du langage naturel (classification, analyse des sentiments, etc.) pour vos données textuelles à l'aide de LLM pour seulement 10 $ par million de lignes (cela dépend de la tâche et du modèle), en restant dans votre environnement dbt. Les instructions, les détails et le code sont ci-dessous

Si vous utilisez dbt comme couche de transformation, vous pourriez vous trouver dans une situation où vous souhaiterez extraire des informations significatives à partir de données textuelles non structurées. Ces données peuvent inclure des avis clients, des titres, des descriptions, des sources/supports Google Analytics, etc. Vous souhaiterez peut-être les classer en groupes ou récupérer des sentiments et des tons.

Les solutions potentielles seraient

• Appliquer des modèles d'apprentissage automatique (ou appeler un LLM) en dehors du flux dbt
• Définir des catégorisations simples dans les modèles dbt à l'aide des instructions CASE WHEN
• prédéfinissez les catégories à l'avance et téléchargez-les dans votre couche de base de données brute ou exploitez la fonctionnalité de départ dbt

À mesure que les modèles Python dbt évoluent, il existe une solution supplémentaire : vous pouvez conserver ces tâches de traitement du langage naturel dans votre environnement dbt en tant qu'un des modèles dbt.

Si cela peut vous être utile, consultez ci-dessous un guide étape par étape sur la façon d'utiliser l'API OpenAI dans votre projet dbt. Vous pouvez reproduire tout ce qui est contenu dans ce guide dans votre environnement, en disposant de l'échantillon de code et de données du référentiel GitHub (voir les liens à la fin).

Configurer l'environnement

Si vous avez déjà un projet et des données dbt ou si vous ne souhaitez pas reproduire les résultats, passez à (4) ou ignorez complètement cette section. Sinon, vous aurez besoin des éléments suivants :

1. Mettre en place le projet dbt . Documents officiels

   1. Vous pouvez simplement cloner celui que j'ai préparé pour ce guide depuis GitHub .

   2. N'oubliez pas de créer/mettre à jour votre fichier profiles.yml.

      
      

2. Configurer la base de données . J'ai utilisé Flocon de neige. Malheureusement, il n'existe pas de version gratuite, mais ils proposent cependant un essai gratuit de 30 jours .

   1. Actuellement, les modèles Python dbt fonctionnent uniquement avec Snowflake, Databricks et BigQuery (pas de PostgreSQL). Ce didacticiel devrait donc fonctionner pour chacun d'entre eux, même si certains détails peuvent varier.

      
      

3. Préparer les données sources

   1. En tant qu'ensemble de données, j'ai utilisé les métadonnées d'un package R publiées dans le référentiel TidyTuesday.

      1. Vous pouvez le télécharger à partir d' ici . Les détails sur l'ensemble de données sontici
      2. Alternativement, vous pouvez utiliser une version allégée de mon référentiel ici

   2. Téléchargez-le dans votre base de données.

   3. Mettez à jour le fichier source.yml dans le projet dbt pour qu'il corresponde aux noms de votre base de données et de votre schéma.

      
      

4. Obtenez la clé API OpenAI

   1. Suivez les instructions de démarrage rapide des documents officiels .

   2. Non : ce n’est pas gratuit, mais c’est payant. Ainsi, avec l'ensemble de données test de 10 lignes, vous ne serez pas facturé plus de 1 $ lors de vos expériences.

   3. Pour être très prudent, fixez une limite de dépenses.

      
      

5. Configurer l'intégration de l'accès externe dans Snowflake

   1. Cela s'applique uniquement si vous utilisez Snowflake.
   2. Si cela n'est pas fait, les modèles dbt Python ne peuvent accéder à aucune API sur Internet (y compris l'API OpenAI).
   3. Suivez les instructions officielles .
   4. Stockez la clé API OpenAI dans cette intégration.

Établissez une liste de catégories

Premièrement, si vous résolvez une tâche de classification, vous avez besoin de catégories (c'est-à-dire de classes) à utiliser dans votre invite LLM. En gros, vous direz : « J'ai une liste de ces catégories, pourriez-vous définir à laquelle appartient ce texte ?

Quelques options ici :

1. Créer manuellement une liste de catégories prédéfinies

   1. Cela convient si vous avez besoin de catégories stables et prévisibles.

   2. N'oubliez pas d'ajouter les "Autres" ici, afin que LLM disposera de ces options en cas d'incertitude.

   3. Demandez à LLM dans votre invite de suggérer un nom de catégorie chaque fois qu'il utilise la catégorie « Autres ».

   4. Téléchargez une liste prédéfinie sur la couche brute de la base de données ou sous forme de CSV dans votre projet dbt (en utilisant dbt seed ).

      
      

2. Introduisez un échantillon de vos données dans LLM et demandez-lui de proposer N catégories.

   1. Même démarche que la précédente, mais on nous aide pour la liste.

   2. Si vous utilisez GPT, il est préférable d'utiliser seed ici pour des raisons de reproductibilité.

      
      

3. Évitez les catégories prédéfinies et laissez LLM faire le travail en déplacement.

   1. Cela pourrait conduire à des résultats moins prévisibles.

   2. En même temps, c’est suffisant si vous vous débrouillez bien avec une marge de hasard.

   3. Dans le cas d'utilisation de GPT, il est préférable de mettre température = 0 pour éviter des résultats différents au cas où vous auriez besoin de réexécuter.

      
      

Dans cet article de blog, j'opterai pour la 3ème option.

Créer un modèle Python dbt pour appeler l'API OpenAI

Passons maintenant à l'essentiel de cet article et créons un modèle dbt qui prendra de nouvelles données texte de la table en amont, les transmettra à l'API OpenAI et enregistrera la catégorie dans la table.

Comme mentionné ci-dessus, je vais utiliser l'ensemble de données des packages R. R est un langage de programmation très populaire dans l'analyse de données. Cet ensemble de données contient des informations sur les packages R du projet CRAN, telles que la version, la licence, l'auteur, le titre, la description, etc. Nous sommes intéressés par le champ title , car nous allons créer une catégorie pour chaque package en fonction de son titre.

1. Préparer la base du modèle

   • La configuration dbt peut être transmise via la méthode dbt.config(...) .

     
     

   • Il existe des arguments supplémentaires dans dbt.config, par exemple, packages est une exigence de package.

     
     

   • Le modèle Python dbt peut référencer des modèles en amont dbt.ref('...') ou dbt.source('...')

     
     

   • Il doit renvoyer un DataFrame. Votre base de données l'enregistrera sous forme de table.

   
   

    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. Connectez-vous à l'API OpenAI

   • Nous devons transmettre secrets et external_access_integrations au dbt.config. Il contiendra la référence secrète stockée dans votre intégration d'accès externe Snowflake.

     
     

   • Remarque : cette fonctionnalité a été publiée il y a seulement quelques jours et n'est disponible que dans la version bêta dbt 1.8.0-b3.

    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. Rendez le modèle dbt incrémentiel et désactivez les actualisations complètes.

   • Cette partie est essentielle pour maintenir les coûts de l'API OpenAI à un faible niveau.
   • Cela l'empêchera de catégoriser plusieurs fois le même texte.
   • Sinon, vous enverrez des données complètes à OpenAI chaque fois que vous exécuterez dbt run , ce qui peut être plusieurs fois par jour.
   • Nous ajoutons materialized='incremental' , incremental_strategy='append' , full_refresh = False , à dbt.config
   • Désormais, l'analyse complète ne concernera que la première exécution de dbt, et pour les exécutions ultérieures (peu importe l'actualisation incrémentielle ou complète), elle catégorisera uniquement le delta.
   • Si vous souhaitez être très attentif, vous pouvez prétraiter un peu vos données pour réduire le nombre d'entrées uniques, mais évitez de trop prétraiter car les LLM fonctionnent mieux avec le langage naturel.

    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. Ajouter une logique d'incrémentalité

   • Lors de l'exécution incrémentielle (en raison de notre configuration, cela signifie sur n'importe quelle exécution sauf la première), nous devons supprimer tous les titres déjà catégorisés.
   • Nous pouvons le faire en utilisant simplement dbt.this . Semblable aux modèles incrémentiels normaux.

    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. Appeler l'API OpenAI par lots

   • Pour réduire les coûts, il est préférable d'envoyer les données à l'API OpenAI par lots.
   • L'invite système peut être 5 fois plus grande que le texte que nous devons classer. Si nous envoyons l'invite système séparément pour chaque titre, cela entraînera une utilisation beaucoup plus élevée des jetons pour les choses répétitives.
   • Le lot ne devrait cependant pas être gros. Avec de gros lots, GPT commence à produire des résultats moins stables. D'après mes expériences, la taille du lot = 5 fonctionne assez bien.
   • De plus, pour garantir que la réponse ne dépasse pas la taille appropriée, j'ai ajouté la contrainte max_tokens .

    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. Il est temps de parler d'une invite pour LLM. Voilà ce que j'ai obtenu :

Vous recevrez une liste des titres de packages CRAN R entre parenthèses ```. Les titres seront séparés par "|" signe. Proposez une catégorie pour chaque titre. Renvoie uniquement les noms de catégories séparés par "|" signe.

• Gardez les instructions directement au point.
• Utilisez la technique ``` pour éviter les injections SQL.
• Soyez clair sur le format du résultat. Dans mon cas, j'ai demandé "|" comme séparateur pour les entrées et les sorties

7. Code final du modèle de dette

    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

Estimations des coûts

Les tarifs de l'API OpenAI sont répertoriés ici . Ils facturent le nombre de jetons demandés et retournés. Le jeton est une instance corrélée à un certain nombre de caractères dans votre requête. Il existe des packages open source permettant d'évaluer un certain nombre de jetons pour un texte donné. Par exemple, TikTok . Si vous souhaitez l'évaluer manuellement, l'endroit où aller est un tokenizer officiel OpenAI ici .

Dans notre ensemble de données, il y a environ 18 000 titres. En gros, cela équivaut à 320 000 jetons d'entrée (180 000 titres et 140 000 invites système si nous utilisons une taille de lot = 5) et à 50 000 jetons de sortie. Selon le modèle, les coûts pour l'analyse complète seront :

1. GPT-4 Turbo : 4,7 $ . Tarification : entrée : 10 $ / 1 million de jetons ; sortie : 30 $ / 1 million de jetons.
2. GPT-4 : 12,6 $. Tarification : entrée : 30 $ / 1 million de jetons ; sortie : 60 $ / 1 million de jetons.
3. GPT-3.5 Turbo : 0,2 $. Tarification : entrée : 0,5 $ / 1 million de jetons ; sortie : 1,5 $ / 1 million de jetons.

Résultats

Le modèle DBT a fonctionné à merveille. J'ai réussi à catégoriser tous les packages 18K sans aucune lacune. Le modèle s'est avéré rentable et protégé contre de multiples passages à la dette.

J'ai publié le tableau de bord des résultats sur Tableau Public ici . N'hésitez pas à jouer avec, à télécharger les données et à créer ce que vous désirez dessus.

Quelques détails intéressants que j'ai trouvés :

• La première catégorie est Data Visualization (1 190 packages, soit 6 %). Je suppose que cela prouve la popularité de R en tant qu'outil de visualisation, en particulier avec des packages comme Shiny, Plotly et autres.

• Les deux principales catégories en croissance en 2023 étaient Data Import et Data Processing . On dirait que R a commencé à être davantage utilisé comme outil de traitement de données.

• La plus forte croissance d'une année sur l'autre parmi les 30 premières catégories a été Natural Language Processing en 2019. Deux ans après le célèbre article « Attention Is All You Need » et six mois après la sortie de GPT-1 :)

Autres idées

1. Nous pouvons utiliser une approche alternative : les intégrations GPT .

   • C'est beaucoup moins cher.

   • Mais plus lourd en ingénierie car vous devez effectuer la partie classification par vous-même (restez à l'écoute, car je vais explorer cette option dans l'un des prochains articles).

     
     

2. Certes, il est logique de supprimer cette partie de dbt et de la transférer vers les fonctions cloud ou toute autre infrastructure que vous utilisez. En même temps, si vous souhaitez le conserver sous la dette, cet article vous couvre.

   
   

3. Évitez d'ajouter une logique au modèle. Il ne devrait faire qu'un seul travail : appeler LLM et enregistrer le résultat. Cela vous aidera à éviter de le réexécuter.

   
   

4. Il y a de fortes chances que vous utilisiez de nombreux environnements dans votre projet dbt. Vous devez être attentif et éviter d'exécuter ce modèle encore et encore dans chaque environnement de développeur à chaque Pull Request.

   • Pour ce faire, vous pouvez incorporer une logique avec if dbt.config.get("target_name") == 'dev'

     
     

5. La réponse avec un délimiteur peut être instable.

   • Par exemple, GPT peut renvoyer moins d’éléments que prévu, et il sera difficile de mapper les titres initiaux à la liste des catégories.

   • Pour surmonter ce problème, ajoutez response_format={ "type": "json_object" } dans votre demande pour exiger une sortie JSON. Voir la documentation officielle .

   • Avec la sortie JSON, vous pouvez demander à l'invite de fournir une réponse au format {"title": "category"}, puis la mapper à vos valeurs initiales.

   • Notez que cela coûtera plus cher, car cela augmentera la taille de la réponse.

   • Curieusement, la qualité de la classification a considérablement chuté lorsque je suis passé à JSON pour GPT 3.5 Turbo.

     
     

6. Il existe une alternative dans Snowflake : en utilisant la fonction cortex.complete() . Découvrez un excellent article de Joel Labes sur le blog dbt.

   
   

C'est ça! Laissez-moi savoir ce que vous pensez.

Liens

Code complet sur GitHub : lien

Tableau de bord Tableau Public : lien

Ensemble de données TidyTuesday R :lien