Como aprimorar seu projeto dbt com grandes modelos de linguagem

DR

Você pode resolver automaticamente tarefas típicas de processamento de linguagem natural (classificação, análise de sentimento, etc.) para seus dados de texto usando LLM por um preço tão baixo quanto US$ 10 por 1 milhão de linhas (depende da tarefa e do modelo), permanecendo em seu ambiente dbt. Instruções, detalhes e código estão abaixo

Se você estiver usando dbt como camada de transformação, poderá ter uma situação em que deseja extrair informações significativas de dados de texto não estruturados. Esses dados podem incluir comentários de clientes, títulos, descrições, fontes/meios do Google Analytics, etc. Você pode querer categorizá-los em grupos ou buscar sentimentos e tons.

As possíveis soluções seriam

• Aplique modelos de aprendizado de máquina (ou chame um LLM) fora do fluxo dbt
• Defina categorizações simples dentro de modelos dbt usando instruções CASE WHEN
• Predefina categorias antecipadamente e carregue-as em sua camada de banco de dados bruto ou aproveite a funcionalidade de semente dbt

À medida que os modelos dbt do Python estão evoluindo, há mais uma solução: você pode manter essas tarefas de processamento de linguagem natural dentro do seu ambiente dbt como um dos modelos dbt.

Se isso for útil para você, veja abaixo um guia passo a passo sobre como usar a API OpenAI em seu projeto dbt. Você pode reproduzir tudo deste guia em seu ambiente, tendo o código e a amostra de dados do repositório GitHub (veja links no final).

Configurar ambiente

Se você já possui um projeto e dados dbt ou não deseja reproduzir os resultados, vá para (4) ou pule esta seção completamente. Caso contrário, você precisará do seguinte:

1. Configure o projeto dbt . Documentos oficiais

   1. Você pode simplesmente clonar aquele que preparei para este guia no GitHub .

   2. Não se esqueça de criar/atualizar seu arquivo profiles.yml.

      
      

2. Configure o banco de dados . Eu usei floco de neve. Infelizmente, não existe uma versão gratuita, mas eles oferecem um teste gratuito de 30 dias .

   1. Atualmente, os modelos dbt Python funcionam apenas com Snowflake, Databricks e BigQuery (sem PostgreSQL). Portanto, este tutorial deve funcionar para qualquer um deles, embora alguns detalhes possam variar

      
      

3. Preparar dados de origem

   1. Como conjunto de dados, usei metadados de um pacote R publicados no repositório TidyTuesday.

      1. Você pode baixá-lo aqui . Os detalhes sobre o conjunto de dados estãoaqui
      2. Alternativamente, você pode usar uma versão leve do meu repositório aqui

   2. Faça upload para seu banco de dados.

   3. Atualize o arquivo source.yml no projeto dbt para corresponder ao seu banco de dados e aos nomes do esquema.

      
      

4. Obtenha a chave da API OpenAI

   1. Siga as instruções de início rápido dos documentos oficiais .

   2. Não: não é gratuito, mas é pré-pago. Portanto, com o conjunto de dados de teste de 10 linhas, não será cobrado mais de US$ 1 durante seus experimentos.

   3. Para ser extremamente cuidadoso, estabeleça um limite de gastos.

      
      

5. Configure a integração de acesso externo no Snowflake

   1. Isso se aplica apenas se você usar o Snowflake.
   2. Se isso não for feito, os modelos dbt Python não poderão acessar nenhuma API na Internet (incluindo a API OpenAI).
   3. Siga as instruções oficiais .
   4. Armazene a chave da API OpenAI nesta integração.

Crie uma lista de categorias

Em primeiro lugar, se você estiver resolvendo uma tarefa de classificação, precisará de categorias (também conhecidas como classes) para usar em seu prompt LLM. Basicamente, você dirá: “Tenho uma lista dessas categorias, você poderia definir a qual delas pertence este texto?”

Algumas opções aqui:

1. Crie uma lista de categorias predefinidas manualmente

   1. É adequado se você precisar de categorias estáveis e previsíveis.

   2. Não se esqueça de adicionar “Outros” aqui, para que o LLM tenha essas opções quando for incerto.

   3. Peça ao LLM em seu prompt para sugerir um nome de categoria sempre que usar a categoria “Outros”.

   4. Faça upload de uma lista predefinida para a camada bruta do banco de dados ou como um CSV em seu projeto dbt (utilizando dbt seed ).

      
      

2. Alimente uma amostra de seus dados ao LLM e peça-lhe que crie N categorias.

   1. A mesma abordagem da anterior, mas estamos recebendo ajuda com a lista.

   2. Se você usa GPT, é melhor usar sementes aqui para reprodutibilidade.

      
      

3. Vá sem categorias predefinidas e deixe o LLM fazer o trabalho em qualquer lugar.

   1. Isso pode levar a resultados menos previsíveis.

   2. Ao mesmo tempo, é bom o suficiente se você concordar com uma margem de aleatoriedade.

   3. No caso de uso do GPT, é melhor colocar temperatura = 0 para evitar resultados diferentes caso seja necessário executar novamente.

      
      

Nesta postagem do blog, irei com a 3ª opção.

Crie um modelo dbt Python para chamar a API OpenAI

Agora, vamos ao cerne desta postagem e criar um modelo dbt que pegará novos dados de texto da tabela upstream, alimentará a API OpenAI e salvará a categoria na tabela.

Conforme mencionado acima, usarei o conjunto de dados de pacotes R. R é uma linguagem de programação altamente popular na análise de dados. Este conjunto de dados contém informações sobre os pacotes R do projeto CRAN, como versão, licença, autor, título, descrição, etc. Estamos interessados no campo title , pois vamos criar uma categoria para cada pacote com base em seu título.

1. Prepare a base para o modelo

   • A configuração do dbt pode ser passada através do método dbt.config(...) .

     
     

   • Existem argumentos adicionais em dbt.config, por exemplo, packages é um requisito de pacote.

     
     

   • O modelo dbt Python pode fazer referência aos modelos upstream dbt.ref('...') ou dbt.source('...')

     
     

   • Deve retornar um DataFrame. Seu banco de dados irá salvá-lo como uma tabela.

   
   

    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. Conecte-se à API OpenAI

   • Precisamos passar secrets e external_access_integrations para o dbt.config. Ele conterá a referência secreta armazenada em sua integração de acesso externo Snowflake.

     
     

   • Observação: este recurso foi lançado há apenas alguns dias e está disponível apenas na versão 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. Torne o modelo dbt incremental e desative as atualizações completas.

   • Esta parte é essencial para manter baixos os custos da API OpenAI.
   • Isso impedirá que o mesmo texto seja categorizado várias vezes.
   • Caso contrário, você enviará dados completos para o OpenAI sempre que executar dbt run , o que pode ocorrer várias vezes ao dia.
   • Estamos adicionando materialized='incremental' , incremental_strategy='append' , full_refresh = False , a dbt.config
   • Agora, a verificação completa será apenas para a primeira execução do dbt e, para as execuções posteriores (não importa a atualização incremental ou completa), ela categorizará apenas o delta.
   • Se quiser ser mais cuidadoso, você pode pré-processar um pouco seus dados para reduzir o número de entradas exclusivas, mas evite pré-processar demais, pois os LLMs funcionam melhor com linguagem 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. Adicionar lógica de incrementalidade

   • Na execução incremental (devido à nossa configuração, significa em qualquer execução, exceto a primeira), precisamos remover todos os títulos já categorizados.
   • Podemos fazer isso simplesmente usando dbt.this . Semelhante aos modelos incrementais normais.

    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. Chame a API OpenAI em lotes

   • Para reduzir custos, é melhor enviar dados para a API OpenAI em lotes.
   • O prompt do sistema pode ser 5 vezes maior que o texto que precisamos classificar. Se enviarmos o prompt do sistema separadamente para cada título, isso levará a um uso de token muito maior para coisas repetitivas.
   • O lote não deve ser grande, no entanto. Com lotes grandes, o GPT começa a produzir resultados menos estáveis. Pelas minhas experiências, tamanho do lote = 5 funciona bem o suficiente.
   • Além disso, para garantir que a resposta não exceda o tamanho relevante, adicionei a restrição 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. É hora de falar sobre uma solicitação para LLM. Isso é o que eu consegui:

Você receberá uma lista de títulos de pacotes CRAN R entre colchetes ```. Os títulos serão separados por "|" sinal. Crie uma categoria para cada título. Retorne apenas nomes de categorias separados por "|" sinal.

• Mantenha as instruções diretas ao ponto.
• Use a técnica ``` para evitar injeções de SQL.
• Seja claro no formato do resultado. No meu caso, pedi "|" como um separador para entradas e saídas

7. Código final do 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

Estimativas de custos

Os preços da API OpenAI estão listados aqui . Eles cobram pelo número de tokens solicitados e devolvidos. O token é uma instância correlacionada a vários caracteres da sua solicitação. Existem pacotes de código aberto para avaliar vários tokens para um determinado texto. Por exemplo, Tiktoken . Se você quiser avaliá-lo manualmente, o lugar certo é um tokenizer oficial da OpenAI aqui .

Em nosso conjunto de dados, existem cerca de 18 mil títulos. Aproximadamente, é igual a 320 mil tokens de entrada (180 mil títulos e 140 mil prompt do sistema se usarmos tamanho de lote = 5) e 50 mil tokens de saída. Dependendo do modelo, os custos da verificação completa serão:

1. GPT-4 Turbo : $ 4,7 . Preço: entrada: tokens de US$ 10/1 milhão; saída: tokens de US$ 30/1 milhão.
2. GPT-4 : $ 12,6. Preço: entrada: tokens de US$ 30/1 milhão; saída: tokens de US$ 60/1 milhão.
3. GPT-3.5 Turbo : $ 0,2. Preço: entrada: tokens de US$ 0,5/1 milhão; saída: tokens de US$ 1,5/1 milhão.

Resultados

O modelo dbt funcionou perfeitamente. Categorizei com sucesso todos os pacotes de 18K sem lacunas. O modelo provou ser econômico e protegido contra múltiplas execuções de dbt.

Publiquei o painel de resultados no Tableau Public aqui . Sinta-se à vontade para brincar com ele, baixar os dados e criar o que desejar.

Alguns detalhes interessantes que encontrei:

• A primeira categoria é Data Visualization (1.190 pacotes ou 6%). Acho que isso prova a popularidade do R como ferramenta de visualização, especialmente com pacotes como Shiny, Plotly e outros.

• As duas categorias com maior crescimento em 2023 foram Data Import e Data Processing . Parece que o R começou a ser usado mais como ferramenta de processamento de dados.

• O maior crescimento ano a ano entre as 30 principais categorias foi Natural Language Processing em 2019. Dois anos após o famoso artigo "Atenção é tudo que você precisa" e meio ano após o lançamento do GPT-1 :)

Mais ideias

1. Podemos usar uma abordagem alternativa – os embeddings GPT .

   • É muito mais barato.

   • Mas mais pesado de engenharia porque você mesmo deve fazer a parte de classificação (fique ligado, pois explorarei essa opção em um dos próximos posts).

     
     

2. Certamente, faz sentido remover esta parte do dbt e enviá-la para funções de nuvem ou qualquer infra-estrutura que você usar. Ao mesmo tempo, se você quiser mantê-lo sob controle, esta postagem oferece cobertura.

   
   

3. Evite adicionar qualquer lógica ao modelo. Deve fazer um trabalho – ligar para o LLM e salvar o resultado. Isso o ajudará a evitar executá-lo novamente.

   
   

4. Há grandes chances de você estar usando muitos ambientes em seu projeto dbt. Você precisa estar atento e evitar executar esse modelo repetidamente em cada ambiente de desenvolvedor em cada solicitação pull.

   • Para fazer isso, você pode incorporar lógica com if dbt.config.get("target_name") == 'dev'

     
     

5. A resposta com um delimitador pode ser instável.

   • Por exemplo, o GPT pode retornar menos elementos do que o esperado e será difícil mapear os títulos iniciais para a lista de categorias.

   • Para superar isso, adicione response_format={ "type": "json_object" } em sua solicitação para exigir saída JSON. Veja os documentos oficiais .

   • Com a saída JSON, você pode solicitar imediatamente uma resposta no formato {"title": "category"} e, em seguida, mapeá-la para seus valores iniciais.

   • Observe que será mais caro, pois aumentará o tamanho da resposta.

   • Estranhamente, a qualidade da classificação caiu drasticamente quando mudei para JSON para GPT 3.5 Turbo.

     
     

6. Existe uma alternativa no Snowflake - usando a função cortex.complete() . Confira uma ótima postagem de Joel Labes no blog do dbt.

   
   

É isso! Diz-me o que pensas.

Ligações

Código completo no GitHub: link

Painel público do Tableau: link

Conjunto de dados TidyTuesday R:link