AI は ETL の再発明に時間を無駄にするべきではありません

最近のAIの進歩はとても刺激的です。人々は、カスタマー サポート エクスペリエンスの向上やコードの作成と実行から、新しい音楽の作成、さらには医療画像技術の高速化に至るまで、あらゆる種類の斬新な方法でこれを使用しています。

しかしその過程で、憂慮すべき傾向が現れてきました。AI コミュニティはデータ移動 (別名 ETL) を再発明しているようです。それらをコネクタ、エクストラクター、統合、ドキュメント ローダーなどと呼ぶかどうかに関係なく、人々は同じコードを作成して、同じ API、ドキュメント形式、データベースからデータを抽出し、それらをベクトル DB または LLM のインデックスにロードしています。

問題は、堅牢な抽出パイプラインと読み込みパイプラインを最初から構築および維持するのは多大な労力を要することです。そして、その分野には先行技術が膨大にあるため、AI 分野のほぼすべてのエンジニアや企業にとって、それを再構築するのは膨大な時間の無駄です。約 1 時間ごとに最新ニュースが発表されるこの分野では、サイドクエストに取り組むことではなく、コア製品をユーザーにとって素晴らしいものにすることに主な焦点を置く必要があります。そして、ほとんどすべての人にとって、中核となる製品はデータの移動ではありません。それはあなたが醸造している AI を活用した魔法のソースです。

堅牢な抽出、変換、読み込み (ETL) パイプラインの構築に伴う課題については多くのことが書かれています ( 1 、 2 ) が、それを AI 内で文脈化してみましょう。

なぜ AI にはデータの移動が必要なのでしょうか?

公開データでトレーニングされた LLM は優れていますが、さらに優れているものがあるかご存知ですか?私たち、私たちの会社、そして私たちのユーザーに特有の質問に答えることができる AI。 ChatGPT が会社の Wiki 全体を学習し、すべての電子メール、Slack メッセージ、会議メモと議事録を読み、会社の分析環境にプラグインし、質問に答えるときにこれらのソースをすべて使用できるようになれば、私たちは皆それを喜ぶでしょう。または、AI を自社の製品 (たとえば、Notion AI など)に統合する場合、ユーザーを支援する際に、ユーザーに関するすべての情報をアプリの AI モデルに認識させたいと考えます。

データの移動はこれらすべての前提条件です。

モデルを微調整している場合でも、検索拡張生成 (RAG)を使用している場合でも、データが存在する場所からデータを抽出し、モデルが消化できる形式に変換して、AI アプリがアクセスできるデータストアに読み込む必要があります。あなたのユースケースに対応します。

上の図は、RAG を使用した場合の様子を示していますが、RAG を使用していない場合でも、基本的な手順はほとんど変わらないと想像できます。次のことを行うには、データを抽出、変換、ロードする必要があります (ETL)。あなたとあなたのユースケースに特有の非公開情報を認識する AI モデルを構築します。

データの移動はなぜ難しいのですか?

API またはデータベースからデータを抽出するための基本的な機能 MVP の構築は、常にではありませんが、通常はすぐに (1 週間未満) 通知すれば実行できます。本当に難しいのは、それを本番環境に対応させて、その状態を維持することです。抽出パイプラインを構築するときに思い浮かぶ標準的な課題をいくつか見てみましょう。

増分抽出と状態管理

意味のあるデータ量がある場合は、パイプラインがこれまでに見たことのないデータのみを抽出するように、増分抽出を実装する必要があります。これを行うには、各接続が抽出したデータを追跡するための永続化レイヤーが必要です。

一時的なエラー処理、バックオフ、障害時の再開、エアギャップ

常に上流のデータ ソースが使用されますが、明確な理由がない場合もあります。パイプラインはこれに対して回復力があり、適切なバックオフ ポリシーを使用して再試行する必要があります。障害がそれほど一時的ではない場合 (ただし、ユーザーのせいではない場合)、パイプラインは中断した場所を記憶し、アップストリームが修正されたら同じ場所から再開できる十分な回復力を備えている必要があります。また、場合によっては、アップストリームから発生する問題が非常に深刻であるため (API がレコードからいくつかの重要なフィールドを削除するなど)、何が起こっているかを調べて手動で何をすべきかを決定するまで、パイプライン全体を完全に一時停止したいことがあります。

構成エラーを特定して積極的に修正する

顧客のデータを抽出するデータ抽出パイプラインを構築している場合は、顧客に代わってデータを抽出するために顧客から提供されたすべての設定が正しいことを確認し、間違っている場合はすぐに行うための防御チェックを実装する必要があります。対処可能なエラー メッセージを提供します。ほとんどの API では、包括的なエラー テーブルが公開されておらず、たとえ公開されていても、API トークンなどに割り当てられた権限を確認するために使用できるエンドポイントがほとんど提供されないため、これを簡単に行うことができません。そのため、包括的なエラー テーブルのバランスを取る方法を見つける必要があります。ユーザーへの迅速なフィードバックでチェックします。

認証とシークレット管理

API のシンプルさは、単純なベアラー トークン認証から、セッション トークンや使い捨てトークン OAuth の「創造的な」実装まで多岐にわたります。認証を実行するロジックを実装するだけでなく、1 時間に 1 回更新されるシークレットを管理し、複数の同時ワーカー間でシークレットの更新を調整する必要があります。

抽出とロードの速度、同時実行性、レート制限の最適化

同時ワーカーについて言えば、抽出の高スループットを実現するために同時実行性を実装する必要があるでしょう。これは小さなデータセットでは問題にならないかもしれませんが、大規模なデータセットでは絶対に重要です。 API は公式のレート制限を公開していますが、IP がブラックリストに登録されたり、永久にレート制限されたりすることなく、API によって提供されるレート制限を最大化するには、最適な並列処理パラメータを経験的に見つける必要があります。

アップストリーム API の変更への適応

API は常に変更され、文書化されていない新しい動作や癖が生じます。多くのベンダーは、四半期ごとに新しい API バージョンを公開しています。これらすべての更新が業務にどのような影響を与えるかを常に監視し、すべてを最新の状態に保つためにエンジニアリング時間を費やす必要があります。新しいエンドポイントは常に登場し、一部のエンドポイントはその動作を変更します (常に警告が得られるわけではありません)。

スケジューリング、モニタリング、ロギング、可観測性

特定の API からデータを抽出するコード以外にも、すべてのデータ抽出ツールで活用されるいくつかの水平機能を構築する必要がある可能性があります。スケジュールを設定するだけでなく、スケジュールが機能しない場合や、他の問題が発生して調査が必要になった場合に備えて、ログを記録したり監視したりすることも必要になります。また、昨日、今日、先週などに抽出されたレコードの数や、それらがどの API エンドポイントまたはデータベース テーブルから抽出されたのかなど、ある程度の可観測性も必要になる可能性があります。

データのブロックまたはハッシュ

データをどこから取得するかによっては、列をダウンストリームに送信する前に列をブロックまたはハッシュするためのプライバシー機能が必要になる場合があります。

明確にしておきますが、いくつかのファイルを一度だけ移動したい場合には、上記のことは当てはまりません。

ただし、データの移動が必要な製品を構築する場合には当てはまります。遅かれ早かれ、これらの懸念事項のほとんどに対処する必要があります。そして、それらのどれも克服できないロケット科学ではありませんが、それらを組み合わせると、すぐに 1 つまたは複数のフルタイムの仕事を増やすことができ、より多くのデータソースから取得すればするほど、その傾向はさらに増します。

そして、それがまさにデータ抽出とパイプラインの維持の難しさです。そのコストの大部分は、これらのパイプラインを機能的かつ堅牢に保つために必要な継続的な追加投資から発生します。ほとんどの AI エンジニアにとって、それはユーザーに最も価値をもたらす仕事ではありません。彼らの時間は他の場所で過ごすのが最適です。

では、AI エンジニアはここでデータを移動するには何をしなければならないのでしょうか?

データの抽出と読み込みパイプラインが必要になった場合は、独自のソリューションを自動的に構築するのではなく、すでに利用可能なソリューションを試してください。すべてではないにしても、多くの懸念を解決できる可能性があります。そうでない場合は、最後の手段として独自のものを構築してください。

また、既存のプラットフォームが必要なものをすべてサポートしていない場合でも、移植可能で拡張可能なフレームワークを使用すれば、ほとんどの作業を実現できるはずです。このようにすると、すべてを最初から構築するのではなく、プラットフォームの既製の機能を使用して目標の 90% を達成し、最後の 10% だけを構築して保守することができます。最も一般的な例はロングテール統合です。必要な API との統合がプラットフォームに同梱されていない場合、優れたプラットフォームを使用すると、その統合を構築するためのコードやノーコード ソリューションを簡単に作成できます。プラットフォームが提供するすべての便利な機能を引き続き利用できます。コネクタを Python パッケージとしてインポートし、コードから好きなようにトリガーするという柔軟性が必要な場合でも、Airbyte コネクタや Singer コネクタなどの多くのオープンソース EL ツールの 1 つを使用できます。

はっきり言っておきますが、データ移動は完全に解決されたわけではありません。既存のソリューションがまったく不十分であり、新しいソリューションを構築する必要がある状況があります。しかし、これは AI エンジニアリング人口の過半数ではありません。ほとんどの人は、Jira、Confluence、Slack、Notion、Gmail、Salesforce などとの同じ統合を何度も再構築する必要はありません。すでに実戦テストが行われ、誰でも使用できるように公開されているソリューションを使用して、ユーザーが実際に関心を持っている価値を追加していきましょう。