LangChain、Airbyte ソースをドキュメントローダーとして導入

エディターズノート：本記事は Airbyte チームとの共同執筆です。Airbyte により、LangChain のドキュメントローダーとしてさらに多くのデータソースを簡単に接続できるようになりました。

今や、LangChain ベースのアプリケーション内で直接、Gong、Hubspot、Salesforce、Shopify、Stripe、Typeform、Zendesk Support といった Airbyte ソースを利用することが可能になりました。これらはすべて ドキュメントローダー として実装されています。

例えば、ユーザーの Stripe 請求書をロードする場合は、AirbyteStripeLoader を使用します。インストールは非常に簡単で、LangChain がローカル環境にインストールされていれば、必要なソースのみを追加インストールすれば準備完了です：

pip install airbyte-source-stripe

その後、ローダーをインポートし、読み込みたいストリームと設定を渡すだけで済みます。

from langchain.document_loaders.airbyte import AirbyteStripeLoader
config = {
  "client_secret": "<secret key>",
  "account_id": "<account id>",
  "start_date": "<date from which to start retrieving records from in ISO format, e.g. 2020-10-20T00:00:00Z>"
}
loader = AirbyteStripeLoader(config=config, stream_name="invoices")
documents = loader.load()
# use documents in vector store or otherwise

なぜこれが重要なのか

これは、Airbyte が提供する 300 以上のデータソースを LangChain のドキュメントローダーとして利用可能にする第一歩です。

Airbyte を使えば、あらゆるデータソースからデータを抽出し、ウェアハウスやベクトルデータベースへ転送して LLM のユースケースを動かすことができます。このようなデータパイプラインの構築方法については、チュートリアルをご覧ください。

通常は Airbyte Cloud を利用するか、ローカルで Airbyte インスタンスを立ち上げて接続を設定し、スケジュール実行または API トリガーによってデータを定期的に更新します。

しかし、これから始めたい場合や、すべてをローカル環境で動かしたいだけの場合は、フル機能の Airbyte インスタンス（UI、スケジューリングサービス、スケーラビリティなど）を用意するのはやりすぎかもしれません。

今回のリリースにより、Python ベースのソースを LangChain で直接実行できるようになりました。Airbyte インスタンスを立ち上げたり、Airbyte Cloud へ API 呼び出しを行ったりする必要はありません。

ホスト型と組み込み型の Airbyte を使い分ける

裏側で動作するコードは共通であるため、Airbyte で構築されたローダーはすべて、Airbyte サービス内の対応するソースと互換性があります。つまり、組み込み型の読み取りパイプラインを、セルフホスト型の Airbyte インストールや Airbyte Cloud インスタンスへ簡単に移行できます。設定オブジェクトの構造もレコード形式も 100% 互換です。

Airbyte のホスト型サービスで同期を実行する場合のメリットは以下の通りです。

• 実行中のパイプラインを UI で管理できる
• 同期失敗時のアラート通知が受け取れる
• スケジュールに基づいて容易にパイプラインを実行できる

一方、LangChain のローダーを使用して同期を実行する場合は、以下のような利点があります。

• 追加のサービス運用によるオーバーヘッドが発生しない
• タイミングやパイプラインの実行を完全に制御できる

Airbyte レコードを LangChain ドキュメントへマッピングする

デフォルトでは、ローダーの一部として各レコードが Document オブジェクトに変換され、レコード内のすべてのフィールドはメタデータとして扱われます。ドキュメントのテキスト部分は空文字列のままです。

この動作をカスタマイズし、データの特性に応じてレコードからテキスト部分を構築したい場合は、record handler を指定できます。

def handle_record(record, id):
    return Document(page_content=record.data["title"], metadata=record.data)
loader = AirbyteGongLoader(config=config, record_handler=handle_record, stream_name="calls")

増分ロード

Python アプリケーションが事実上 Airbyte プラットフォームとして機能するため、同期の実行方法を完全に制御できます。例えば、ストリームが対応していれば、ローダーの last_state プロパティにアクセスすることで 増分同期 の恩恵を受けることも可能です。これにより、前回ロード以降に変更されたドキュメントのみを読み込み、既存のベクトルデータベースを効果的に更新できます。

import airbyte_cdk.models.airbyte_protocol import AirbyteMessage
with open('stripe_sync_checkpoint.json', 'wb') as file:
    file.write(loader.last_state.json())

// ... later
with open('stripe_sync_checkpoint.json', 'r') as file:
    current_state = AirbyteStateMessage.parse_raw(file.read())
incremental_loader = AirbyteStripeLoader(config=config, stream_name="invoices", state=current_state)
new_docs = incremental_loader.load()

カスタムソース

現時点で pip パッケージとして利用可能な Airbyte ソースは以下の通りです（今後さらに追加予定）：

Gong、Hubspot、Salesforce、Shopify、Stripe、Typeform、Zendesk Support などの主要なデータソースは、それぞれ専用のパッケージをインストールするだけで利用可能です。

• Gong pip install airbyte-source-gong
• Hubspot pip install airbyte-source-hubspot
• Salesforce pip install airbyte-source-salesforce
• Shopify pip install airbyte-source-shopify
• Stripe pip install airbyte-source-stripe
• Typeform pip install airbyte-source-typeform
• Zendesk Support pip install airbyte-source-zendesk-support

一方で、独自の Airbyte ソースを実装している場合は、Airbyte CDK の Source インターフェースと連携する AirbyteCDKLoader ベースクラスを利用することで、それらも統合することが可能です。

from langchain.document_loaders.airbyte import AirbyteCDKLoader
from my_source.source import MyCustomSource # plug in your own source here
config = {
   # your custom configuration
}
loader = AirbyteCDKLoader(source_class=MyCustomSource, config=config, stream_name="my-stream")

また、主要な Airbyte リポジトリから直接 Git を経由してソースをインストールすることもできます。例えば GitHub ソースを取得するには、以下のように実行するだけです。

以下のコマンドを実行して、GitHub ソースをインストールします。

pip install "source_github@git+https://github.com/airbytehq/airbyte.git@master#subdirectory=airbyte-integrations/connectors/source-github"

これで、AirbyteCDKLoader としてソースを利用できるようになります。

from source_github.source import SourceGithub
issues_loader = AirbyteCDKLoader(source_class=SourceGithub, config=config, stream_name="issues")

独自のソースを作成する方法については、コネクタ開発のドキュメントをご覧ください。作成は簡単で、必要に応じてローカルの埋め込み型ローダーからホストされた Airbyte インスタンスへシームレスに切り替えることができます。

ご質問はありますか？

Airbyte を活用して LLM ベースのアプリケーションへのデータ転送を検討されている場合は、アンケートにご協力ください。皆様のフィードバックをもとに、最も重要な機能の開発優先度を決定いたします。

ご質問がある場合や、既存のソースを同様にローダーとして公開することに関心がある場合は、コミュニティ Slack チャンネルまたは LangChain Discord サーバー内の Airbyte チャンネルまでお気軽にお問い合わせください。