概要

簡単にセットアップできる DataStax の Astra DB と、GUI で LLM アプリケーションを作れる Langflow を使って RAG を構築して API を作成します。

前提

DataStax のアカウントと OpenAI の API key が必要です。DataStax のアカウントはクレジットカード登録なしで無料で作成できます。冒頭の作成方法を参照してください。OpenAI の API key はあらかじめクレジットを購入し、API を利用できるようにしておく必要があります。

利用するデータ

今回は経済産業省のデジタル時代の人材政策に関する検討会が作成しているレポート「生成AI時代のDX推進に必要な人材・スキルの考え方2024」のPDFをデータとして RAG をやってみます。以下のページから、概要ではなくて全文版のPDFをダウンロードしておきます。
https://www.meti.go.jp/shingikai/mono_info_service/digital_jinzai/20240628_report.html

作業全体の流れ

Astra Serverless DB (Vector) をセットアップする。
Astra Serverless DB にベクトルデータを投入する。
2-1.　Langflow を使う場合
2-2.　Python + Langchain を使う場合
Langflow を使って RAG を設計して API を作成する

Astra Serverless DB (Vector) をセットアップ

アカウント作成

Astra DB を使うために DataStax のアカウントを作成しましょう。DataStax のページに移動して、右上の TRY FOR FREE からアカウント作成に進みます。Free Plan は無料で利用できます。利用できる容量や read/write の回数に制限がありますが、PoC には十分です。

データベースの作成

アカウントができたらAstra Portalに移動してログインします。ログインすると以下のような画面が表示されますので、右側にある Create Serverless Database をクリックしましょう。
Astra DB には、大きく分けて2つの DB があり、新しいのが Serverless Database、以前からあるものが Classic Database です。Serverless のほうは、その名前の通りでインフラの設定・管理をする必要がありません。有料の Pay As You Go プランの場合は、DB は負荷に応じて自動的にスケールし、ユーザは利用した分だけ料金を支払います。Classic の場合、あらかじめスペックを決めてDBをデプロイし、それ以降は起動時間に応じて料金を支払います。Serverless のほうはベクトルデータベースに特化した機能も提供しています。

作成画面は以下のようになっていて、RAG ではベクトルデータベースを使うので、Serverless (Vector) を選びます。Vector 形式を選択しても、ベクトル以外の情報（例えば、顧客の氏名や年齢といった情報) も入れることはできます。適当に名前をつけて Provider と　Region を選んで Create Database をクリックします。

コレクションの作成

数分すると DB が作成され、以下の画面が表示されます。この画面から Create Collection をして Collection を作成します。Collection とはベクトルデータ用のテーブルのことです。

Collection を作成する際、デフォルトで Vector-enabled collection が ON になっていると思います。そうすると、データを投入するときに自動的にEmbeddingしてベクトルデータを投入してくれたり、ここで指定した類似度尺度 (Similarity Metrics) に従ってデータを検索してくれます。自動で Embeddings してくれる機能を Astra Vectorizeと呼んでいて、デフォルトでは NVIDIA の Embeddings モデルを利用できます。それ以外にも OpenAI や Hugging Face なども API key などを用意すれば利用できます。自分でベクトル化して投入する場合は Bring my own を指定します。

あとで日本語のドキュメントを扱おうと思うので、ここでは OpenAI のモデルを指定します。指定するためには、Add embedding provider integration を選択します。選択したら OpenAI や　Hugging Face などをプルダウンメニューから選べるようになります。

OpenAIを選ぶと以下のような画面で名前とAPI Keyを登録する画面が現れます。ここで登録しておくことによって、あとでいつでも OpenAI の API を利用することができます。指定が終わったら Add integration を押しましょう。

Embedding generation method として NVIDIA のかわりに OpenAI が選ばれ、先ほど登録した API key が指定された状態で、もとの Collection 作成画面に戻ります。下の方で Embedding model を選択し、最後の Create Collection をクリックすれば、データベースのセットアップは完了です。

Astra Serverless DB (Vector) にベクトルデータを投入する (Langflow編)

Langflow のインストールと起動

まず Langflow を手元にインストールしましょう。公式のインストール手順は Langflow 公式ページにあります。今回はローカルで起動するので、Install Langflow Locally のところを見てください。

Terminal を開いて pip を使って Langflow をインストールします。

python -m pip install langflow -U

インストールが完了したらLangflowを起動します。

python -m langflow run

Langflow の設定

起動して http://127.0.0.1:7860 から Langflow へアクセスすると以下の画面が表示されます。新しく Project を作るために New Project をクリックします。いくつかテンプレートを選べるので、Vector Store RAG を選びます。

以下のような上下二つに分かれたフローが表示されます。上のフローはベクトルデータベースのセットアップが完了してから、実際にチャットで問い合わせを送るのに利用します。下のフローはファイルを読み込んで、ファイルを細かく分割 (Chunking) して Embedding してベクトルDBに投入します。

まずは下のフローを利用します。基本的にはデフォルトのまま使うことができます。オレンジでデフォルトの処理の説明を記載し、追加入力すべきところはオレンジで囲みました。青字の部分はEmbeddingsの代替の方法として Astra Vectorize を利用できることを示しています。オレンジで囲んだ追加入力すべきところを箇条書きで説明します。

テキストのチャンク分割の separator は、デフォルトでは改行が指定されています。より意味のあるまとまりとするために、段落の区切り (\n\n) を指定します。
Embeddings モデルは OpenAI の text-embedding-3-small を指定します。OpenAI の API キーを指定します。もし Astra Vectorize に任せたい場合は、左のメニューからドラッグして、OpenAI の text-embedding-3-small を指定します。API Key name は Astra DB のセットアップのときに指定した API Key の名前を入力します。Astra DB 側は、Emebddings モデルを指定されればそこからベクトルデータを受け取り、Astra Vectorize 指定されたら Astra DB 側でベクトル化します。
Astra DB 側は、Collection Name はコレクションを作成するときに指定した名前（上の例では vectors)を指定します。Astra DB Application Token と API Endpoint は　Astra Portal のデータベースの画面から確認できます (下図) 。Application Tokens は Generate して利用します。

ベクトルデータの投入

それではデータを読み込んでベクトル化して Astra DB に投入しましょう。手順を図に書き込みましたが、PDFファイルを指定して一番右の再生ボタンをクリックするとそれまでに必要なすべてのステップが実行されます。もし途中結果としてチャンクの内容を見たい場合は Split Text のボックスの右下の Chunks をクリックすれば見ることができます。途中で Chunk サイズが大きいなどのエラーがあればエラーメッセージも表示されるので Chunk Sizeを調整します。追加のファイルがある場合はファイルを選びなおせば Astra DB に追加されていきます。もし Astra DB の内容をリセットして新規で書き込みたい場合は Astra DB のブロックをクリックして３点リーダ(...)を表示し、Advanced -> Pre Delete Collection を ON にすれば良いです。

ベクトルデータの確認

Astra Portal で作成したコレクションのページにいくとデータが登録されていることを確認できます。

Astra Serverless DB (Vector) にベクトルデータを投入する (Python+Langchain編)

Python と Langchain を使って pdf を読み込んでベクトル化して Astra DB にデータを追加します。ベクトル化には Astra Vectorize と OpenAI の統合を利用するので、Python 側では Embeddings のモデルを用意する必要はありません。

まずは必要な python のライブラリをインストールします。unstructured は PDF の読み込みに、それ以外は langchain と Astra DB の統合のライブラリを使って簡単にデータを投入します。

pip install "unstructured[pdf]" langchain_community astrapy langchain_astradb

あとは PDF ファイルと同じところに以下の python ファイルを作成して実行します。

from astrapy.info import CollectionVectorServiceOptions
from langchain_astradb import AstraDBVectorStore
from langchain_community.document_loaders import UnstructuredPDFLoader
from langchain_text_splitters import CharacterTextSplitter

'''
1. PDF ファイルを unstructured を使って読み込み
'''
loader = UnstructuredPDFLoader("./20240628_2.pdf")
pdf_data = loader.load()

'''
2. Langchain の CharacterTextSplitter でチャンキング
'''
chunk_overlap = 200
chunk_size = 1000
separator = "\n\n"
splitter = CharacterTextSplitter(
    chunk_overlap=chunk_overlap,
    chunk_size=chunk_size,
    separator=separator,
)
docs = splitter.split_documents(pdf_data)

'''
3. Astra Vectorize の設定
ASTRA_DB_APPLICATION_TOKEN と ASTRA_DB_API_ENDPOINT は
Astra Portal で確認する。
'''
ASTRA_DB_APPLICATION_TOKEN="AstraCS:hogehoge"
ASTRA_DB_API_ENDPOINT="https://hogehoge-us-east-2.apps.astra.datastax.com"

openai_vectorize_options = CollectionVectorServiceOptions(
    provider="openai",
    model_name="text-embedding-3-small",
    authentication={
        "providerKey": "openai-key",
    },
)

vector_store = AstraDBVectorStore(
    collection_name="python_vectors",
    api_endpoint=ASTRA_DB_API_ENDPOINT,
    token=ASTRA_DB_APPLICATION_TOKEN,
    collection_vector_service_options=openai_vectorize_options
)

'''
4. データをベクトル化して投入
'''
vector_store.add_documents(documents=docs, ids=[i for i in range(len(docs))])

Langflow で RAG を設計して API を作成

ユーザからのクエリに対して RAG を使ってレスポンスを返すフローを Langflow で設計します。もしベクトルデータの投入を Python から行っている場合は、上の Langflow の設定に従って、Vector Store RAGのテンプレートを開きます。上下二つのフローのうち上がRAGの実行フローになります。フローの各ステップをそのまま使うことができますが、以下の図に示すように、OpenAI の API Key を入れたり、Astra Portal を確認して Astra DB の設定を入力する必要があります。詳細な設定方法は上記の Langflow の設定を見てください。

もしベクトル DB へのデータ投入時に Astra Vectorize を使っている場合は、最初の OpenAI Embeddings も Astra Vectorize に変更します。本質的な違いはありませんが、Astra Vectorize の場合は、元のテキストデータを $vectorize というカラムに保持し、OpenAI Embeddings の場合は content というカラムに保持するので、カラム名の差異がエラーにつながります。

これで RAG のフローは完成です。上記の図の一番右側にある Chat Output のボックスにある再生ボタンをクリックしてエラーが出なければOKです。画面の下の方にPlayground や API というボタンがあり、チャットを試したり API を確認したりすることができます。

それでは Playground を試してみましょう。RAGが機能しているか確認するために、PDFに答えが書いてある質問「生成 AI がもたらす潜在的な経済効果はどの程度ですか」を聞いてみます。

PDFの中にも同じ記述があるので RAG で正しく答えられています！次は Langflow の画面上だけではなくて、API から同じ質問をしてみたいと思います。API は先ほどの Playground の横にある API をクリックするとサンプルが表示されますのでこれを利用しましょう。今回は Python で API を操作するサンプルを利用します。以下の python のコードをコピーして、試しに rag.pyとして実行してみます。

そのままだと日本語は文字化けしてしまうので、python コードの最後の方で json.dumps する際に ensure_ascii=Falseを足しておきます。

    #print(json.dumps(response, indent=2))
    print(json.dumps(response, indent=2,ensure_ascii=False))

それでは実行します。引数に先ほどと同じクエリを入れると同じ出力が返されましたね。Python から Langflow の API を使用することができました！

まとめ

Astra DB のセットアップ方法と Langflow または Langchain を使った RAG API の構築方法を説明しました。Astra DB のセットアップが楽なのと、Langflow が RAG のテンプレートを用意してくれるので、細かいことを気にしなくてすぐに RAG を構築できました。個人的には GUI よりもコードを書く派なのですが、Langchain, Astra DB, Astra Vectorize の統合で短い python のコードでデータ投入できたのは楽でした。データ投入は調整も必要になると思うので、ここは python で書いちゃって、チャットのフローは Langflow に任せてもいいかも。

次は Vercel AI SDK を使ってフロントエンドから Langflow の API を利用するのをやってみようと思います。