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: . Configure el proyecto dbt Documentos oficiales Puedes simplemente clonar el que preparé para esta guía desde . GitHub No olvide crear/actualizar su archivo perfiles.yml. . Usé copo de nieve. Desafortunadamente, no existe una versión gratuita, pero ofrecen una . Configurar la base de datos prueba gratuita de 30 días 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. Preparar datos de origen Como conjunto de datos, utilicé metadatos de un paquete R publicado en el repositorio de TidyTuesday. Puedes descargarlo desde . Los detalles sobre el conjunto de datos están aquí aquí. Alternativamente, puedes usar una versión ligera de mi repositorio aquí. Súbelo a tu base de datos. Actualice el archivo en el proyecto dbt para que coincida con los nombres de su base de datos y esquema. source.yml Obtenga la clave API de OpenAI Siga las instrucciones de inicio rápido de . los documentos oficiales 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. Para tener mucho cuidado, establezca un límite de gasto. Configurar la integración de acceso externo en Snowflake Esto se aplica sólo si usas Snowflake. Si no se hace esto, los modelos dbt Python no pueden acceder a ninguna API en Internet (incluida la API OpenAI). Siga las . instrucciones oficiales 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í: Crear una lista de categorías predefinidas manualmente Es adecuado si necesita categorías estables y predecibles. No olvide agregar "Otros" aquí, para que LLM tenga estas opciones cuando no esté seguro. Pídale a LLM en su mensaje que sugiera un nombre de categoría cada vez que utilice la categoría "Otros". 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 Envíe una muestra de sus datos a LLM y pídale que genere N categorías. El mismo enfoque que el anterior, pero estamos recibiendo ayuda con la lista. Si usa GPT, es mejor usar semilla aquí para mayor reproducibilidad. Olvídese de las categorías predefinidas y deje que LLM haga el trabajo sobre la marcha. Esto podría conducir a resultados menos predecibles. Al mismo tiempo, es bastante bueno si estás bien con un margen de aleatoriedad. 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 , ya que vamos a crear una categoría para cada paquete en función de su título. title 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, es un requisito del paquete. packages El modelo dbt Python puede hacer referencia a modelos anteriores o dbt.ref('...') 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 Conéctese a la API de OpenAI Necesitamos pasar y al dbt.config. Contendrá la referencia secreta que está almacenada en su integración de acceso externo de Snowflake. secrets external_access_integrations 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'), ) 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 , lo que puede ocurrir varias veces al día. dbt run Estamos agregando , , , a dbt.config materialized='incremental' incremental_strategy='append' full_refresh = False 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 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 . Similar a los modelos incrementales normales. dbt.this 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), :] 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) 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 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 . 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, . Si desea evaluarlo manualmente, el lugar al que debe acudir es un tokenizador oficial de OpenAI . aquí TikTok 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: : . Precio: entrada: $10 / 1 millón de tokens; Salida: $30 / 1 millón de tokens. GPT-4 Turbo 4,7 dólares : Precio: entrada: $30 / 1 millón de tokens; Salida: $60 / 1 millón de tokens. GPT-4 12,6 dólares. : Precio: entrada: $0,5 / 1 millón de tokens; Salida: $1.5 / 1 millón de tokens. GPT-3.5 Turbo 0,2 dólares. 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 . Siéntete libre de jugar con él, descargar los datos y crear lo que desees encima. aquí Algunos detalles interesantes que encontré: La categoría principal es (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. Data Visualization Las dos categorías con mayor crecimiento en 2023 fueron y . Parece que R comenzó a usarse más como herramienta de procesamiento de datos. Data Import Data Processing El mayor crecimiento interanual entre las 30 categorías principales fue 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 :) Natural Language Processing Más ideas 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). 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. 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. 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' 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 en su solicitud para requerir salida JSON. Consulte los . response_format={ "type": "json_object" } 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. Hay una alternativa en Snowflake: usar la función . Consulte una de Joel Labes en el blog de dbt. cortex.complete() excelente publicación ¡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
