Cách nâng cao dự án dbt của bạn bằng các mô hình ngôn ngữ lớn

TL;DR

Bạn có thể tự động giải quyết các tác vụ Xử lý ngôn ngữ tự nhiên điển hình (phân loại, phân tích cảm tính, v.v.) cho dữ liệu văn bản của mình bằng LLM với mức giá rẻ nhất là 10 USD cho mỗi 1 triệu hàng (tùy thuộc vào nhiệm vụ và mô hình), duy trì trong môi trường dbt của bạn. Hướng dẫn, chi tiết và mã bên dưới

Nếu bạn đang sử dụng dbt làm lớp chuyển đổi, bạn có thể gặp tình huống muốn trích xuất thông tin có ý nghĩa từ dữ liệu văn bản phi cấu trúc. Những dữ liệu đó có thể bao gồm đánh giá của khách hàng, tiêu đề, mô tả, nguồn/phương tiện Google Analytics, v.v. Bạn có thể muốn phân loại chúng thành các nhóm hoặc tìm nạp cảm xúc và giọng điệu.

Các giải pháp tiềm năng sẽ là

• Áp dụng các mô hình học máy (hoặc gọi LLM) bên ngoài luồng dbt
• Xác định các phân loại đơn giản bên trong các mô hình dbt bằng cách sử dụng câu lệnh CASE WHEN
• Xác định trước các danh mục và tải chúng lên lớp cơ sở dữ liệu thô của bạn hoặc tận dụng chức năng hạt giống dbt

Khi các mô hình dbt Python đang phát triển, có một giải pháp nữa: bạn có thể giữ các tác vụ Xử lý ngôn ngữ tự nhiên này bên trong môi trường dbt của mình dưới dạng một trong các mô hình dbt.

Nếu điều đó có thể hữu ích cho bạn, hãy xem hướng dẫn từng bước bên dưới về cách sử dụng API OpenAI trong dự án dbt của bạn. Bạn có thể tái tạo mọi thứ từ hướng dẫn này trong môi trường của mình, lấy mã và mẫu dữ liệu từ kho lưu trữ GitHub (xem các liên kết ở cuối).

Thiết lập môi trường

Nếu bạn đã có dự án và dữ liệu dbt hoặc không muốn tái tạo kết quả, hãy chuyển đến (4) hoặc bỏ qua hoàn toàn phần này. Nếu không, bạn sẽ cần những thứ sau:

1. Thiết lập dự án dbt . Tài liệu chính thức

   1. Bạn có thể chỉ cần sao chép cái tôi đã chuẩn bị cho hướng dẫn này từ GitHub .

   2. Đừng quên tạo/cập nhật tệp profile.yml của bạn.

      
      

2. Thiết lập cơ sở dữ liệu . Tôi đã sử dụng Bông tuyết. Thật không may, không có phiên bản miễn phí nhưng họ cung cấp bản dùng thử miễn phí 30 ngày .

   1. Hiện tại, các mô hình dbt Python chỉ hoạt động với Snowflake, Databricks và BigQuery (không có PostgreSQL). Vì vậy, hướng dẫn này sẽ phù hợp với bất kỳ trường hợp nào trong số đó, mặc dù một số chi tiết có thể khác nhau.

      
      

3. Chuẩn bị dữ liệu nguồn

   1. Là một tập dữ liệu, tôi đã sử dụng siêu dữ liệu gói R được xuất bản trong kho lưu trữ TidyTuesday.

      1. Bạn có thể tải về từ đây . Chi tiết về tập dữ liệu cótại đây
      2. Ngoài ra, bạn có thể sử dụng phiên bản nhẹ từ kho lưu trữ của tôi tại đây

   2. Tải nó lên cơ sở dữ liệu của bạn.

   3. Cập nhật tệp source.yml trong dự án dbt để khớp với tên cơ sở dữ liệu và lược đồ của bạn.

      
      

4. Nhận khóa API OpenAI

   1. Làm theo hướng dẫn bắt đầu nhanh từ các tài liệu chính thức .

   2. Không: e nó không miễn phí, nhưng nó trả tiền khi bạn sử dụng. Vì vậy, với tập dữ liệu 10 hàng thử nghiệm, bạn sẽ không bị tính phí nhiều hơn $1 trong quá trình thử nghiệm của mình.

   3. Để cẩn thận hơn, hãy đặt giới hạn chi tiêu.

      
      

5. Thiết lập tích hợp truy cập bên ngoài trong Snowflake

   1. Điều này chỉ áp dụng nếu bạn sử dụng Snowflake.
   2. Nếu điều này không được thực hiện, các mô hình dbt Python không thể truy cập bất kỳ API nào trên internet (bao gồm cả API OpenAI).
   3. Thực hiện theo các hướng dẫn chính thức .
   4. Lưu trữ khóa API OpenAI trong tích hợp này.

Đưa ra danh sách các danh mục

Đầu tiên, nếu bạn đang giải quyết một nhiệm vụ phân loại, bạn cần các danh mục (hay còn gọi là lớp) để sử dụng trong lời nhắc LLM của mình. Về cơ bản, bạn sẽ nói: "Tôi có một danh sách các danh mục này, bạn có thể xác định văn bản này thuộc về danh mục nào không?"

Một số tùy chọn ở đây:

1. Tạo danh sách các danh mục được xác định trước theo cách thủ công

   1. Nó phù hợp nếu bạn cần các danh mục ổn định và có thể dự đoán được.

   2. Đừng quên thêm "Khác" vào đây để LLM sẽ có những lựa chọn này khi không chắc chắn.

   3. Yêu cầu LLM trong lời nhắc của bạn đề xuất tên danh mục bất cứ khi nào nó sử dụng danh mục "Khác".

   4. Tải danh sách được xác định trước lên lớp thô của cơ sở dữ liệu hoặc dưới dạng CSV trong dự án dbt của bạn (sử dụng dbt seed ).

      
      

2. Cung cấp mẫu dữ liệu của bạn cho LLM và yêu cầu nó đưa ra N danh mục.

   1. Cách tiếp cận tương tự như cách trước, nhưng chúng tôi đang nhận được trợ giúp về danh sách.

   2. Nếu bạn sử dụng GPT, tốt hơn nên sử dụng hạt giống ở đây để có khả năng tái tạo.

      
      

3. Không cần có các danh mục được xác định trước và để LLM thực hiện công việc khi đang di chuyển.

   1. Điều này có thể dẫn đến kết quả ít dự đoán hơn.

   2. Đồng thời, sẽ đủ tốt nếu bạn hài lòng với một mức độ ngẫu nhiên.

   3. Trong trường hợp sử dụng GPT, tốt hơn nên đặt nhiệt độ = 0 để tránh các kết quả khác nhau trong trường hợp bạn cần chạy lại.

      
      

Trong bài đăng trên blog này, tôi sẽ chọn tùy chọn thứ 3.

Tạo Mô hình Python dbt để gọi API OpenAI

Bây giờ, chúng ta hãy đi vào nội dung chính của bài đăng này và tạo một mô hình dbt sẽ lấy dữ liệu văn bản mới từ bảng ngược dòng, đưa dữ liệu đó vào API OpenAI và lưu danh mục vào bảng.

Như đã đề cập ở trên, tôi sẽ sử dụng tập dữ liệu gói R. R là ngôn ngữ lập trình rất phổ biến trong phân tích dữ liệu. Tập dữ liệu này chứa thông tin về các gói R từ dự án CRAN, chẳng hạn như phiên bản, giấy phép, tác giả, tiêu đề, mô tả, v.v. Chúng tôi quan tâm đến trường title vì chúng tôi sẽ tạo một danh mục cho mỗi gói dựa trên tiêu đề của nó.

1. Chuẩn bị nền cho mô hình

   • Cấu hình dbt có thể được truyền qua phương thức dbt.config(...) .

     
     

   • Có các đối số bổ sung trong dbt.config, ví dụ: packages là một yêu cầu về gói.

     
     

   • Mô hình dbt Python có thể tham chiếu các mô hình ngược dòng dbt.ref('...') hoặc dbt.source('...')

     
     

   • Nó phải trả về DataFrame. Cơ sở dữ liệu của bạn sẽ lưu nó dưới dạng bảng.

   
   

    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. Kết nối với API OpenAI

   • Chúng ta cần chuyển secrets và external_access_integrations cho dbt.config. Nó sẽ chứa tham chiếu bí mật được lưu trữ trong Tích hợp truy cập bên ngoài Snowflake của bạn.

     
     

   • Lưu ý: tính năng này mới được phát hành vài ngày trước và chỉ có trong phiên bả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. Làm cho mô hình dbt tăng dần và tắt tính năng làm mới hoàn toàn.

   • Phần này rất cần thiết để giữ cho chi phí API OpenAI ở mức thấp.
   • Nó sẽ ngăn nó phân loại cùng một văn bản nhiều lần.
   • Nếu không, bạn sẽ gửi toàn bộ dữ liệu tới OpenAI mỗi khi bạn thực thi dbt run , có thể vài lần một ngày.
   • Chúng tôi đang thêm materialized='incremental' , incremental_strategy='append' , full_refresh = False , vào dbt.config
   • Bây giờ, quá trình quét toàn bộ sẽ chỉ dành cho lần chạy dbt đầu tiên và đối với những lần chạy sau (bất kể tăng dần hay làm mới toàn bộ), nó sẽ chỉ phân loại delta.
   • Nếu muốn lưu ý hơn, bạn có thể xử lý trước dữ liệu của mình một chút để giảm số lượng mục nhập duy nhất, nhưng tránh xử lý trước quá nhiều vì LLM hoạt động tốt hơn với ngôn ngữ tự nhiên.

    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. Thêm logic tăng dần

   • Trong lần chạy tăng dần (do thiết lập của chúng tôi, điều đó có nghĩa là trong bất kỳ lần chạy nào ngoại trừ lần chạy đầu tiên), chúng tôi cần xóa tất cả các tiêu đề đã được phân loại.
   • Chúng ta có thể làm điều đó bằng cách sử dụng dbt.this . Tương tự như các mô hình gia tăng thông thường.

    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. Gọi API OpenAI theo đợt

   • Để giảm chi phí, tốt hơn hết bạn nên gửi dữ liệu tới API OpenAI theo đợt.
   • Lời nhắc hệ thống có thể lớn hơn 5 lần so với văn bản chúng ta cần phân loại. Nếu chúng tôi gửi lời nhắc hệ thống riêng cho từng tiêu đề, điều đó sẽ dẫn đến mức sử dụng mã thông báo cao hơn nhiều cho những thứ lặp đi lặp lại.
   • Tuy nhiên, lô không nên lớn. Với các lô lớn, GPT bắt đầu tạo ra kết quả kém ổn định hơn. Từ thử nghiệm của tôi, kích thước lô = 5 hoạt động đủ tốt.
   • Ngoài ra, để đảm bảo phản hồi không vượt quá kích thước phù hợp, tôi đã thêm ràng buộc 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. Đã đến lúc nói về lời nhắc cho LLM. Đó là những gì tôi nhận được:

Bạn sẽ được cung cấp danh sách tiêu đề gói CRAN R trong ngoặc ```. Các tiêu đề sẽ được phân tách bằng dấu "|" dấu hiệu. Hãy đưa ra một danh mục cho mỗi tiêu đề. Chỉ trả lại tên danh mục được phân tách bằng "|" dấu hiệu.

• Giữ hướng dẫn đi thẳng vào vấn đề.
• Sử dụng kỹ thuật ``` để tránh việc tiêm SQL.
• Hãy rõ ràng về định dạng kết quả. Trong trường hợp của tôi, tôi đã yêu cầu "|" như một dấu phân cách cho cả đầu vào và đầu ra

7. Mã mô hình dbt cuối cùng

    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

Ước tính chi phí

Giá API OpenAI được liệt kê ở đây . Họ tính phí theo số lượng token được yêu cầu và trả lại. Mã thông báo là một phiên bản tương quan với một số ký tự trong yêu cầu của bạn. Có các gói nguồn mở để đánh giá một số mã thông báo cho một văn bản nhất định. Ví dụ: Tiktoken . Nếu bạn muốn đánh giá nó theo cách thủ công, nơi cần đến là mã thông báo OpenAI chính thức tại đây .

Trong tập dữ liệu của chúng tôi, có ~ 18 nghìn đầu sách. Đại khái, nó tương đương với 320K mã thông báo đầu vào (180K tiêu đề và 140K lời nhắc hệ thống nếu chúng tôi sử dụng kích thước lô = 5) và 50K mã thông báo đầu ra. Tùy thuộc vào kiểu máy, chi phí cho việc quét toàn bộ sẽ là:

1. GPT-4 Turbo : 4,7 USD . Giá cả: đầu vào: $10 / 1 triệu token; đầu ra: $30 / 1 triệu token.
2. GPT-4 : 12,6 USD. Giá cả: đầu vào: $30 / 1 triệu token; đầu ra: $60 / 1 triệu token.
3. GPT-3.5 Turbo : 0,2 USD. Giá cả: đầu vào: 0,5 USD / 1 triệu token; đầu ra: 1,5 USD / 1 triệu token.

Kết quả

Mô hình dbt hoạt động như một cơ duyên. Tôi đã phân loại thành công tất cả các gói 18K mà không có bất kỳ khoảng trống nào. Mô hình này tỏ ra hiệu quả về mặt chi phí và được bảo vệ trước nhiều lần chạy dbt.

Tôi đã xuất bản bảng điều khiển kết quả lên Tableau Public tại đây . Hãy thoải mái chơi với nó, tải xuống dữ liệu và tạo bất cứ thứ gì bạn muốn trên đó.

Một số chi tiết thú vị tôi tìm thấy:

• Danh mục top 1 là Data Visualization (1.190 gói, tương đương 6%). Tôi đoán điều này chứng tỏ sự phổ biến của R như một công cụ trực quan hóa, đặc biệt là với các gói như Shiny, Plotly và các gói khác.

• Hai danh mục phát triển hàng đầu vào năm 2023 là Data Import và Data Processing . Có vẻ như R bắt đầu được sử dụng nhiều hơn như một công cụ xử lý dữ liệu.

• Mức tăng trưởng hàng năm lớn nhất trong số 30 hạng mục hàng đầu là Natural Language Processing vào năm 2019. Hai năm sau bài báo nổi tiếng "Chú ý là tất cả những gì bạn cần" và nửa năm sau khi phát hành GPT-1 :)

Ý tưởng khác

1. Chúng tôi có thể sử dụng một cách tiếp cận khác - nhúng GPT .

   • Nó rẻ hơn nhiều.

   • Nhưng nặng về kỹ thuật hơn vì bạn nên tự mình thực hiện phần phân loại (hãy chú ý theo dõi vì tôi sẽ khám phá tùy chọn này trong một trong các bài đăng tiếp theo).

     
     

2. Chắc chắn, việc loại bỏ phần này khỏi dbt và đẩy nó lên các chức năng đám mây hoặc bất kỳ cơ sở hạ tầng nào bạn sử dụng là điều hợp lý. Đồng thời, nếu bạn muốn giữ nó dưới dbt - bài đăng này sẽ giúp bạn hiểu rõ hơn.

   
   

3. Tránh thêm bất kỳ logic nào vào mô hình. Nó sẽ thực hiện một công việc - gọi LLM và lưu kết quả. Điều này sẽ giúp bạn tránh phải chạy lại nó.

   
   

4. Rất có thể bạn đang sử dụng nhiều môi trường trong dự án dbt của mình. Bạn cần lưu ý và tránh chạy đi chạy lại mô hình này trong từng môi trường nhà phát triển trên mỗi Yêu cầu Kéo.

   • Để thực hiện việc này, bạn có thể kết hợp logic với if dbt.config.get("target_name") == 'dev'

     
     

5. Phản hồi bằng dấu phân cách có thể không ổn định.

   • Ví dụ: GPT có thể trả về ít phần tử hơn bạn mong đợi và sẽ khó ánh xạ các tiêu đề ban đầu vào danh sách danh mục.

   • Để khắc phục điều này, hãy thêm response_format={ "type": "json_object" } vào yêu cầu của bạn để yêu cầu đầu ra JSON. Xem tài liệu chính thức .

   • Với đầu ra JSON, bạn có thể yêu cầu cung cấp câu trả lời ở định dạng {"title": "category"}, sau đó ánh xạ câu trả lời đó tới các giá trị ban đầu của bạn.

   • Lưu ý rằng nó sẽ đắt hơn vì nó sẽ tăng kích thước phản hồi.

   • Thật kỳ lạ, chất lượng phân loại giảm đáng kể khi tôi chuyển sang JSON cho GPT 3.5 Turbo.

     
     

6. Có một giải pháp thay thế trong Snowflake - sử dụng hàm Cortex.complete() . Hãy xem một bài đăng tuyệt vời của Joel Labes trên blog dbt.

   
   

Đó là nó! Cho tôi biết bạn nghĩ gì.

Liên kết

Mã đầy đủ trên GitHub: link

Bảng điều khiển công khai của Tableau: liên kết

Tập dữ liệu TidyTuesday R:liên kết