Аналитическая инженерия с поддержкой LLM: Как мы используем ИИ внутри нашего проекта dbt сегодня, без новых инструментов.

Облачные платформы данных открывают новые возможности; dbt помогает внедрить их в производство

Первый сдвиг парадигмы, который позволил dbt существовать и быть полезным, заключался в переходе баз данных в облако.

Внезапно стало возможным для большего числа людей выполнять более качественную работу с данными, так как огромные препятствия превратились в огромные возможности:

• Теперь мы могли динамически масштабировать вычисления по требованию, без необходимости обновления до более крупной локальной базы данных.
• Теперь мы могли хранить и запрашивать огромные наборы данных, такие как данные о кликах, без предварительной агрегации и трансформации.

Сегодня следующая волна инноваций происходит в области ИИ и LLM, и она приходит на облачные платформы данных, которые специалисты dbt уже используют каждый день. Например, Snowflake только что выпустили свои функции Cortex для доступа к инструментам с поддержкой LLM, настроенным для выполнения общих задач с вашими существующими наборами данных. Это открывает перед нами новые возможности:

• Теперь мы можем извлекать смысл из больших неструктурированных блоков текста, без утомительного создания сложных регулярных выражений.
• Теперь мы можем резюмировать или переводить контент без необходимости обращения к внешним API третьих сторон.
• Наиболее значимо, теперь мы можем встраивать возможности рассуждения в наши модели dbt, описывая, что мы хотим, чтобы произошло.

Аналитические инженеры всегда существовали на пересечении бизнес-контекста и данных - LLM в хранилище делают возможным встраивание большего бизнес-контекста и разблокировку большего объема данных, увеличивая нашу эффективность в обоих направлениях одновременно.

Анатомия рабочего процесса с поддержкой LLM

Мои коллеги и я провели эксперименты в прошлом году с использованием GPT-4 для улучшения семантического слоя, но это первый раз, когда стало возможным использовать ИИ непосредственно внутри нашего проекта dbt, без какого-либо дополнительного инструментария.

Когда мы искали первый случай использования с поддержкой ИИ в нашем аналитическом стеке, мы хотели найти что-то, что:

• Решает реальную бизнес-проблему для нас сегодня
• Использует уникальные возможности LLM
• Осознает их текущие неопределенности и ограничения
• Предвосхищает будущие улучшения доступных нам моделей, так что вещи, которые не работают сегодня, могут вскоре работать очень хорошо.

После того как мы выбрали наш случай использования, работа по аналитической инженерии по созданию и оркестровке новых моделей dbt показалась очень знакомой; на самом деле, это было точно так же, как и любая другая модель, которую я строил.

• Я все еще строил DAG слоями, с существующими моделями на стадии как основой и создавая новые модульные сегменты сверху
• Я все еще следовал тем же лучшим практикам и конвенциям по написанию, стилю и контролю версий моего кода
• Я все еще обеспечивал, чтобы мои модели вели себя так, как я ожидал, проходя процесс код-ревью и автоматического тестирования, прежде чем развернуть мои рабочие нагрузки LLM в производственной среде с оркестратором dbt Cloud.

Вкратце, тот же dbt, который я знаю и люблю, но дополненный новой мощью, которую предоставляет Cortex.

Разработка нашего первого аналитического рабочего процесса с поддержкой LLM в dbt Cloud

Когда я думал о проекте, который был бы возможен только в том случае, если бы мы могли разобраться в большом объеме неструктурированного текста, я довольно быстро понял, что это может помочь мне быть в курсе событий в сообществе dbt в Slack. Хотя мы проводим много времени в Slack, ежедневно происходит сотни обсуждений в десятках каналов, поэтому мы часто пропускаем важные или интересные разговоры.

Мы уже загружаем данные из Slack в Snowflake для базовой аналитики, но наличие агента триажа, который мог бы внимательно следить за Slack и сообщать нам о вещах, которые мы иначе могли бы пропустить, помогло бы команде по работе с разработчиками лучше понимать потребности разработчиков dbt.

Когда проект был завершен, он выглядел так:

Пример некоторых резюмированных обсуждений для обзора (слегка отредактировано для анонимности).

До одного раза в день мы получаем сообщение в нашем внутреннем Slack с ссылками на несколько интересных обсуждений для каждой области фокуса и кратким резюме обсуждения на данный момент. Оттуда мы можем углубиться, погрузившись в обсуждение самостоятельно, где бы оно ни происходило. Разрабатывая это, я нашел несколько обсуждений, которые не нашел бы никаким другим способом (что само по себе было проблемой, так как моя модель фильтрует обсуждения, как только в них участвует сотрудник dbt Labs, поэтому я постоянно терял все свои тестовые данные).

У вас, вероятно, нет точно такого же случая использования, как у меня, но вы можете представить себе широкий набор случаев использования аналитической инженерии с поддержкой LLM:

• SaaS-компания может извлекать информацию из звонков по продажам или заявок в службу поддержки, чтобы получить представление о разговорах
• Разработчик мобильных приложений может анализировать отзывы в магазине приложений для анализа настроений
• Вычисляя векторные встраивания для текста, дедупликация схожих, но не идентичных текстов становится более осуществимой.

Вот фрагмент кода, использующего функцию cortex.complete() - обратите внимание, что все это выглядит как обычный SQL, потому что это и есть SQL!

        select trim(
            snowflake.cortex.complete(
                'llama2-70b-chat',
                concat(
                    'Write a short, two sentence summary of this Slack thread. Focus on issues raised. Be brief. <thread>',
                    text_to_summarize,
                    '</thread>. The users involved are: <users>',
                    participant_metadata.participant_users::text,
                    '</users>'
                )
            )
        ) as thread_summary,

Советы по созданию моделей dbt с поддержкой LLM

• Всегда стройте инкрементально. Любой, кто взаимодействовал с инструментом с поддержкой LLM, знает, что может потребоваться некоторое время, чтобы получить результаты от запроса, и что результаты могут варьироваться от одного вызова к другому. По причинам скорости, стоимости и согласованности я реализовал обе модели инкрементально, даже несмотря на то, что по количеству строк таблицы крошечные. Я также добавил конфигурацию full_refresh: false, чтобы защититься от других полных обновлений, которые мы запускаем для захвата поздно поступающих фактов.
• Остерегайтесь ограничений на количество токенов. Запросы, содержащие слишком много токенов, обрезаются, что может привести к неожиданным результатам, если точка отсечения находится в середине сообщения. В будущем я бы сначала попробовал использовать модель llama-70b (предел ~4k токенов), а для неудачных строк сделал бы второй проход, используя модель mistral-7b (предел ~32k токенов). Как и многие аспекты рабочих процессов с поддержкой LLM, мы ожидаем, что ограничения на длину токенов значительно увеличатся в ближайшее время.
• Оркестрируйте защитно, пока что. Из-за вышеуказанных соображений я запускаю эти шаги в их собственной задаче dbt Cloud, запускаемой после успешного завершения нашей основной задачи проекта. Я не хочу, чтобы команда данных была напугана неудачным производственным запуском из-за моих экспериментов. Мы используем YAML-селекторы для определения того, что запускается в нашей задаче по умолчанию; я создал новый селектор для этих моделей и затем добавил этот селектор в список исключений задачи по умолчанию. Как только это станет более стабильным, я интегрирую это в нашу обычную задачу.
• Итеративно улучшайте свой запрос. Так же, как вы постепенно улучшаете SQL-запрос, вам нужно часто корректировать свой запрос в процессе разработки, чтобы убедиться, что вы получаете ожидаемые результаты. В общем, я начинал с самой короткой команды, которая, по моему мнению, могла бы сработать, и корректировал ее на основе получаемых результатов. Один из немного разочаровывающих аспектов инженерии запросов: я могу потратить полдня на работу над проблемой, и в конце концов у меня будет только одна строка кода, чтобы добавить в коммит.
• Помните, что ваши результаты недетерминированы. Для кого-то, кто любит говорить об идемпотентности, иметь модель, результаты которой варьируются в зависимости от "настроения" некоторых камней, которые мы обманули, заставив мечтать, немного странно и требует немного более защитного кодирования, чем вы могли бы привыкнуть. Например, один из запросов, которые я использую, ориентирован на классификацию (определение области продукта обсуждения), и обычно результатом является просто название этого продукта. Но иногда он возвращает небольшое объяснение своего мышления, поэтому мне нужно явно извлекать это значение из ответа, а не бездумно принимать все, что я получаю. Определение допустимых вариантов в переменной Jinja помогло сохранить их в синхронизации: я могу передать их в запрос, а затем повторно использовать тот же список при извлечении правильного ответа.

-- сокращенный список сегментов для удобочитаемости
{% set segments = ['Warehouse configuration', 'dbt Cloud IDE', 'dbt Core', 'SQL', 'dbt Orchestration', 'dbt Explorer', 'Unknown'] %}

    select trim(
        snowflake.cortex.complete(
            'llama2-70b-chat',
            concat(
                'Identify the dbt product segment that this message relates to, out of [{{ segments | join ("|") }}]. Your response should be only the segment with no explanation. <message>',
                text,
                '</message>'
            )
        )
    ) as product_segment_raw,

    -- повторное использование переменной segments Jinja здесь
    coalesce(regexp_substr(product_segment_raw, '{{ segments | join ("|") }}'), 'Unknown') as product_segment

Поделитесь своим опытом

Если вы делаете что-то подобное в своей работе или побочном проекте, я был бы рад услышать об этом в разделе комментариев на Discourse или в канале machine-learning-general в Slack.

Приложение: Пример полной модели

Вот полная модель, которую я запускаю для создания общих сообщений, которые публикуются в Slack, построенная на основе построчного резюме в более ранней модели:

{{
    config(
        materialized='incremental',
        unique_key='unique_key',
        full_refresh=false
    )
-}}


{# 
    Этот словарь partition_by предназначен для упрощения использования столбцов, которые используются в разных частях запроса. 
    SQL используется в компонентах partition by агрегатов оконных функций, а имена столбцов используются (в сочетании с SQL) для выбора соответствующих столбцов в финальной модели.
    Их можно было бы прописать вручную, но это создает много мест для обновления при изменении, например, с дневного на недельное усечение. 

    Стороннее замечание: я все еще не в восторге от этого подхода и был бы рад услышать об альтернативах!
#}
{%- set partition_by = [
    {'column': 'summary_period', 'sql': 'date_trunc(day, sent_at)'},
    {'column': 'product_segment', 'sql': 'lower(product_segment)'},
    {'column': 'is_further_attention_needed', 'sql': 'is_further_attention_needed'},
] -%}

{% set partition_by_sqls = [] -%}
{% set partition_by_columns = [] -%}

{% for p in partition_by -%}
    {% do partition_by_sqls.append(p.sql) -%}
    {% do partition_by_columns.append(p.column) -%}
{% endfor -%}


with

summaries as (

    select * from {{ ref('fct_slack_thread_llm_summaries') }}
    where not has_townie_participant

),

aggregated as (
    select distinct
        {# Использование столбцов, определенных выше #}
        {% for p in partition_by -%}
            {{ p.sql }} as {{ p.column }},
        {% endfor -%}

        -- Это создает JSON-массив, где каждый элемент - это одна ветка + ее постоянная ссылка. 
        -- Каждый массив разбивается по столбцам partition_by, определенным выше, так что есть
        -- одно резюме на период времени и продукт и т.д.
        array_agg(
            object_construct(
                'permalink', thread_permalink,
                'thread', thread_summary
            )
        ) over (partition by {{ partition_by_sqls | join(', ') }}) as agg_threads,
        count(*) over (partition by {{ partition_by_sqls | join(', ') }}) as num_records,
        
        -- Столбцы partition являются зерном таблицы и могут быть использованы для создания
        -- уникального ключа для инкрементальных целей
        {{ dbt_utils.generate_surrogate_key(partition_by_columns) }} as unique_key
    from summaries
    {% if is_incremental() %}
        where unique_key not in (select this.unique_key from {{ this }} as this) 
    {% endif %}

),

summarised as (

    select
        *,
        trim(snowflake.cortex.complete(
            'llama2-70b-chat',
            concat(
                'In a few bullets, describe the key takeaways from these threads. For each object in the array, summarise the `thread` field, then provide the Slack permalink URL from the `permalink` field for that element in markdown format at the end of each summary. Do not repeat my request back to me in your response.',
                agg_threads::text
            )
        )) as overall_summary
    from aggregated

),

final as (
    select 
        * exclude overall_summary,
        -- LLM любит говорить что-то вроде "Конечно, вот ваше резюме:" несмотря на мои усилия. Поэтому это удаляет эту строку
        regexp_replace(
            overall_summary, '(^Sure.+:\n*)', ''
        ) as overall_summary

    from summarised
)

select * from final