Cómo mejorar su proyecto dbt con modelos de lenguaje grandes

TL;DR

Puede resolver automáticamente tareas típicas de procesamiento del lenguaje natural (clasificación, análisis de sentimientos, etc.) para sus datos de texto utilizando LLM por tan solo $10 por 1 millón de filas (depende de la tarea y el modelo), permaneciendo en su entorno dbt. Las instrucciones, los detalles y el código se encuentran a continuación.

Si está utilizando dbt como capa de transformación, es posible que tenga una situación en la que desee extraer información significativa de datos de texto no estructurados. Dichos datos pueden incluir reseñas de clientes, títulos, descripciones, fuentes/medios de Google Analytics, etc. Es posible que desee clasificarlos en grupos o buscar opiniones y tonos.

Las posibles soluciones serían

• Aplique modelos de aprendizaje automático (o llame a un LLM) fuera del flujo de dbt
• Defina categorizaciones simples dentro de modelos dbt usando declaraciones CASE WHEN
• Predefina las categorías por adelantado y cárguelas en la capa de su base de datos sin procesar o aproveche la funcionalidad de semilla de dbt.

A medida que los modelos dbt de Python evolucionan, existe una solución más: puede mantener estas tareas de procesamiento del lenguaje natural dentro de su entorno dbt como uno de los modelos dbt.

Si esto puede resultarle útil, consulte a continuación una guía paso a paso sobre cómo utilizar la API OpenAI en su proyecto dbt. Puede reproducir todo lo contenido en esta guía en su entorno, teniendo el código y la muestra de datos del repositorio de GitHub (ver enlaces al final).

Configurar el entorno

Si ya tiene un proyecto y datos dbt o no desea reproducir los resultados, salte a (4) u omita esta sección por completo. De lo contrario, necesitará lo siguiente:

1. Configure el proyecto dbt . Documentos oficiales

   1. Puedes simplemente clonar el que preparé para esta guía desde GitHub .

   2. No olvide crear/actualizar su archivo perfiles.yml.

      
      

2. Configurar la base de datos . Usé copo de nieve. Desafortunadamente, no existe una versión gratuita, pero ofrecen una prueba gratuita de 30 días .

   1. Actualmente, los modelos dbt Python solo funcionan con Snowflake, Databricks y BigQuery (no PostgreSQL). Entonces, este tutorial debería funcionar para cualquiera de ellos, aunque algunos detalles pueden variar.

      
      

3. Preparar datos de origen

   1. Como conjunto de datos, utilicé metadatos de un paquete R publicado en el repositorio de TidyTuesday.

      1. Puedes descargarlo desde aquí . Los detalles sobre el conjunto de datos estánaquí.
      2. Alternativamente, puedes usar una versión ligera de mi repositorio aquí.

   2. Súbelo a tu base de datos.

   3. Actualice el archivo source.yml en el proyecto dbt para que coincida con los nombres de su base de datos y esquema.

      
      

4. Obtenga la clave API de OpenAI

   1. Siga las instrucciones de inicio rápido de los documentos oficiales .

   2. No: no es gratis, pero es de pago por uso. Por lo tanto, con el conjunto de datos de prueba de 10 filas, no se le cobrará más de $1 durante sus experimentos.

   3. Para tener mucho cuidado, establezca un límite de gasto.

      
      

5. Configurar la integración de acceso externo en Snowflake

   1. Esto se aplica sólo si usas Snowflake.
   2. Si no se hace esto, los modelos dbt Python no pueden acceder a ninguna API en Internet (incluida la API OpenAI).
   3. Siga las instrucciones oficiales .
   4. Almacene la clave API de OpenAI en esta integración.

Crea una lista de categorías

En primer lugar, si está resolviendo una tarea de clasificación, necesita categorías (también conocidas como clases) para usar en su mensaje de LLM. Básicamente dirás: "Tengo una lista de estas categorías, ¿podrías definir a cuál pertenece este texto?"

Algunas opciones aquí:

1. Crear una lista de categorías predefinidas manualmente

   1. Es adecuado si necesita categorías estables y predecibles.

   2. No olvide agregar "Otros" aquí, para que LLM tenga estas opciones cuando no esté seguro.

   3. Pídale a LLM en su mensaje que sugiera un nombre de categoría cada vez que utilice la categoría "Otros".

   4. Cargue una lista predefinida en la capa sin formato de la base de datos o como un CSV en su proyecto dbt (utilizando dbt seed ).

      
      

2. Envíe una muestra de sus datos a LLM y pídale que genere N categorías.

   1. El mismo enfoque que el anterior, pero estamos recibiendo ayuda con la lista.

   2. Si usa GPT, es mejor usar semilla aquí para mayor reproducibilidad.

      
      

3. Olvídese de las categorías predefinidas y deje que LLM haga el trabajo sobre la marcha.

   1. Esto podría conducir a resultados menos predecibles.

   2. Al mismo tiempo, es bastante bueno si estás bien con un margen de aleatoriedad.

   3. En el caso de uso de GPT, es mejor poner temperatura = 0 para evitar resultados diferentes en caso de que necesite volver a ejecutar.

      
      

En esta publicación de blog, optaré por la tercera opción.

Cree un modelo dbt Python para llamar a la API OpenAI

Ahora, vayamos al meollo de esta publicación y creemos un modelo dbt que tomará nuevos datos de texto de la tabla ascendente, los alimentará a la API de OpenAI y guardará la categoría en la tabla.

Como se mencionó anteriormente, usaré el conjunto de datos de paquetes R. R es un lenguaje de programación muy popular en el análisis de datos. Este conjunto de datos contiene información sobre los paquetes R del proyecto CRAN, como versión, licencia, autor, título, descripción, etc. Nos interesa el campo title , ya que vamos a crear una categoría para cada paquete en función de su título.

1. Prepara la base para el modelo.

   • La configuración dbt se puede pasar mediante el método dbt.config(...) .

     
     

   • Hay argumentos adicionales en dbt.config, por ejemplo, packages es un requisito del paquete.

     
     

   • El modelo dbt Python puede hacer referencia a modelos anteriores dbt.ref('...') o dbt.source('...')

     
     

   • Debe devolver un DataFrame. Su base de datos la guardará como una tabla.

   
   

    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. Conéctese a la API de OpenAI

   • Necesitamos pasar secrets y external_access_integrations al dbt.config. Contendrá la referencia secreta que está almacenada en su integración de acceso externo de Snowflake.

     
     

   • Nota: esta función se lanzó hace solo unos días y solo está disponible en la versión beta 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. Haga que el modelo dbt sea incremental y desactive las actualizaciones completas.

   • Esta parte es esencial para mantener bajos los costos de la API OpenAI.
   • Evitará que categorice el mismo texto varias veces.
   • De lo contrario, enviará datos completos a OpenAI cada vez que ejecute dbt run , lo que puede ocurrir varias veces al día.
   • Estamos agregando materialized='incremental' , incremental_strategy='append' , full_refresh = False , a dbt.config
   • Ahora, el análisis completo será solo para la primera ejecución de dbt, y para las ejecuciones posteriores (sin importar la actualización incremental o completa), categorizará solo delta.
   • Si desea ser más consciente, puede preprocesar un poco sus datos para reducir la cantidad de entradas únicas, pero evite preprocesar demasiado, ya que los LLM funcionan mejor con lenguaje natural.

    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. Agregar lógica de incrementalidad

   • En la ejecución incremental (debido a nuestra configuración, significa en cualquier ejecución excepto la primera), debemos eliminar todos los títulos ya categorizados.
   • Podemos hacerlo simplemente usando dbt.this . Similar a los modelos incrementales normales.

    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. Llame a la API de OpenAI en lotes

   • Para reducir los costos, es mejor enviar datos a la API de OpenAI en lotes.
   • El mensaje del sistema puede ser 5 veces más grande que el texto que necesitamos clasificar. Si enviamos el mensaje del sistema por separado para cada título, se producirá un uso mucho mayor del token para cosas repetitivas.
   • Sin embargo, el lote no debería ser grande. Con lotes grandes, GPT comienza a producir resultados menos estables. Según mis experimentos, el tamaño del lote = 5 funciona bastante bien.
   • Además, para garantizar que la respuesta no exceda el tamaño relevante, agregué la restricción 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. Es hora de hablar sobre un mensaje para LLM. Eso es lo que tengo:

Se le proporcionará una lista de títulos de paquetes CRAN R entre corchetes ```. Los títulos estarán separados por "|" firmar. Crea una categoría para cada título. Devuelve sólo nombres de categorías separados por "|" firmar.

• Mantenga las instrucciones directas al grano.
• Utilice la técnica ``` para evitar inyecciones SQL.
• Sea claro en el formato del resultado. En mi caso, pedí "|" como separador tanto de entradas como de salidas

7. Código final del modelo dbt

    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

Estimaciones de costos

Los precios de la API de OpenAI se enumeran aquí . Cobran por la cantidad de tokens solicitados y devueltos. El token es una instancia correlacionada con una cantidad de caracteres en su solicitud. Existen paquetes de código abierto para evaluar una cantidad de tokens para un texto determinado. Por ejemplo, TikTok . Si desea evaluarlo manualmente, el lugar al que debe acudir es un tokenizador oficial de OpenAI aquí .

En nuestro conjunto de datos, hay ~18.000 títulos. Aproximadamente, equivale a 320 000 tokens de entrada (180 000 títulos y 140 000 mensajes del sistema si usamos un tamaño de lote = 5) y 50 000 tokens de salida. Dependiendo del modelo, los costes del escaneo completo serán:

1. GPT-4 Turbo : 4,7 dólares . Precio: entrada: $10 / 1 millón de tokens; Salida: $30 / 1 millón de tokens.
2. GPT-4 : 12,6 dólares. Precio: entrada: $30 / 1 millón de tokens; Salida: $60 / 1 millón de tokens.
3. GPT-3.5 Turbo : 0,2 dólares. Precio: entrada: $0,5 / 1 millón de tokens; Salida: $1.5 / 1 millón de tokens.

Resultados

El modelo dbt funcionó a las mil maravillas. Clasifiqué con éxito todos los paquetes de 18K sin espacios. El modelo demostró ser rentable y estar protegido contra múltiples ejecuciones de DBT.

Publiqué el panel de resultados en Tableau Public aquí . Siéntete libre de jugar con él, descargar los datos y crear lo que desees encima.

Algunos detalles interesantes que encontré:

• La categoría principal es Data Visualization (1190 paquetes, o 6%). Supongo que esto demuestra la popularidad de R como herramienta de visualización, especialmente con paquetes como Shiny, Plotly y otros.

• Las dos categorías con mayor crecimiento en 2023 fueron Data Import y Data Processing . Parece que R comenzó a usarse más como herramienta de procesamiento de datos.

• El mayor crecimiento interanual entre las 30 categorías principales fue Natural Language Processing en 2019. Dos años después del famoso artículo "La atención es todo lo que necesitas" y medio año después del lanzamiento de GPT-1 :)

Más ideas

1. Podemos utilizar un enfoque alternativo: las incorporaciones de GPT .

   • Es mucho más barato.

   • Pero requiere más ingeniería porque debes hacer la parte de clasificación por tu cuenta (estén atentos, ya que exploraré esta opción en una de las próximas publicaciones).

     
     

2. Ciertamente, tiene sentido eliminar esta parte de dbt y enviarla a las funciones de la nube o cualquier infraestructura que utilice. Al mismo tiempo, si desea mantenerlo bajo dbt, esta publicación lo cubre.

   
   

3. Evite agregar lógica al modelo. Debería hacer un trabajo: llamar a LLM y guardar el resultado. Esto le ayudará a evitar volver a ejecutarlo.

   
   

4. Es muy probable que esté utilizando muchos entornos en su proyecto dbt. Debe tener cuidado y evitar ejecutar este modelo una y otra vez en cada entorno de desarrollador en cada solicitud de extracción.

   • Para hacer esto, puede incorporar lógica con if dbt.config.get("target_name") == 'dev'

     
     

5. La respuesta con un delimitador puede ser inestable.

   • Por ejemplo, GPT puede devolver menos elementos de los esperados y será difícil asignar títulos iniciales a la lista de categorías.

   • Para superar esto, agregue response_format={ "type": "json_object" } en su solicitud para requerir salida JSON. Consulte los documentos oficiales .

   • Con la salida JSON, puede solicitar en el mensaje que proporcione una respuesta en el formato {"title": "category"} y luego asignarla a sus valores iniciales.

   • Tenga en cuenta que será más caro, ya que aumentará el tamaño de la respuesta.

   • Por extraño que parezca, la calidad de la clasificación disminuyó drásticamente cuando cambié a JSON para GPT 3.5 Turbo.

     
     

6. Hay una alternativa en Snowflake: usar la función cortex.complete() . Consulte una excelente publicación de Joel Labes en el blog de dbt.

   
   

¡Eso es todo! Déjame saber lo que piensas.

Enlaces

Código completo en GitHub: enlace

Panel público de Tableau: enlace

Conjunto de datos TidyTuesday R:enlace