はじめに
GMO NIKKOのY-Kです。
GCPのVertex AIのベクトル検索を利用してRAGを作ったので軽くまとめてみようと思います。
RAGの応答精度向上のための手法は枚挙に暇がないので、今回は精度を度外視して一番シンプルなRAGモデルの骨格を作成していきます。
GCPのVertex AIのベクトル検索のクイックスタートを参考にしていきます。
https://cloud.google.com/vertex-ai/docs/vector-search/quickstart?hl=ja
データ準備
まずはRAGが参照するデータを準備します。
GMOの企業理念や経営ノウハウなどを集約した、GMO スピリットベンチャー宣言を元データにしようと思います。
https://www.gmo.jp/brand/sv/
以下のような構造のGCSバケットを作成して、上記URL先の内容をtxtにしたものをoriginへアップロードします。
|
1 2 3 4 5 |
{バケット名} ┣ batch_root : ベクトル化したデータ保存する場所 ┗ documents ┣ chunk : originのデータを分割して保存する場所 ┗ origin : RAGに読み込ませたいデータを保存する場所 |
batch_rootに関してはクイックスタートの名前に合わせているだけなので、わかりやすい名前に変更しても問題ないです。
データ整形
colab上で作業していきます。
まずは認証
|
1 2 3 4 5 |
# 認証 from google.colab import auth auth.authenticate_user() PROJECT_ID = "{GCPのプロジェクトIDを入れてください}" |
必要ライブラリのインストール
|
1 2 3 4 |
!pip install --upgrade nltk !pip install langchain langchain_community unstructured !pip install openai !pip install -qU langchain-openai |
importと作業用ディレクトリの作成
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
from google.cloud import storage import nltk nltk.download('punkt') BUCKET_NAME = "{GCSのバケット名}" GCS_DOC_ORIGIN = "documents/original/" GCS_DOC_CHUNK = "documents/chunk/" GCS_BATCH_ROOT = "batch_root/" import os os.makedirs(BUCKET_NAME + "/" + GCS_DOC_ORIGIN, exist_ok=True) os.makedirs(BUCKET_NAME + "/" + GCS_DOC_CHUNK, exist_ok=True) os.makedirs(BUCKET_NAME + "/" + GCS_BATCH_ROOT, exist_ok=True) |
GCSからoriginのファイルをcolabのローカルにダウンロードします
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
def list_blob(prefix): storage_client = storage.Client(project=PROJECT_ID) bucket = storage_client.bucket(BUCKET_NAME) blobs = bucket.list_blobs(prefix=prefix) return blobs def download_blob(prefix): blobs = list_blob(prefix) for blob in blobs: if blob.name != prefix: blob.download_to_filename(BUCKET_NAME + "/" + prefix + blob.name.split("/")[-1]) download_blob(GCS_DOC_ORIGIN) |
読み込ませる元データをチャンクに分割します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import DirectoryLoader # ディレクトリの読み込み loader = DirectoryLoader(BUCKET_NAME + "/" + GCS_DOC_ORIGIN) # テキストをチャンクに分割 docs = loader.load_and_split( text_splitter=RecursiveCharacterTextSplitter( chunk_size=200,# 分割したチャンクごとの文字数 chunk_overlap=20 # チャンク間で被らせる文字数 ) ) for idx, doc in enumerate(docs): with open(BUCKET_NAME + "/" + GCS_DOC_CHUNK + doc.metadata['source'].split("/")[-1].replace(".", f"_{idx}."), "w") as f: f.write(doc.page_content) |
チャンクに分割できているのか軽く見てみます。
|
1 2 3 |
for doc in docs[0:3]: print(doc.page_content) print("========================================") |
|
1 2 3 4 5 6 7 8 |
世に「ベンチャー」の定義は多く存在しますが、我々の言う「ベンチャー」とは「旧来の伝統的企業に対抗して、革新的商品・サービスを提供することでお客様に『笑顔』『感動』を提供し、多くの人に尊敬され応援される、『ファン』の多い会社を作る・・・そんな『志』＝『夢』を実現させるために、革新的なスピード・着眼点・手段・頭脳によって、リスクを恐れず突き進む者の集団」を指します。 ======================================== 我々は、スピリットベンチャー宣言を憲法や誓書のように共有・徹底し、これをベースに一人ひとりが個性を発揮します。 1995年の創業以来、我々が培ったマインドをここに宣言します。 ======================================== 我々は「ベンチャー」である。技術力や経営手法がクローズアップされる企業は数多くありますが、真の「ベンチャー」の優位性はスピリットの共有・徹底にあります。テクニックやスキル、マニュアルを偏重する旧来の経営手法においては、理論的に経営資源を扱ってきました。人も同様、基本的には性悪説に基づき利益と恐怖による統制が原則と考えられてきたのです。 ======================================== |
いい感じにできていそうです。
|
1 2 3 4 5 6 7 8 |
def upload_from_filename(source_file_name, destination_blob_name): storage_client = storage.Client(project=PROJECT_ID) bucket = storage_client.bucket(BUCKET_NAME) blob_gcs = bucket.blob(destination_blob_name) blob_gcs.upload_from_filename(source_file_name) for file_path in os.listdir(BUCKET_NAME + "/" + GCS_DOC_CHUNK): upload_from_filename(BUCKET_NAME + "/" + GCS_DOC_CHUNK + file_path, GCS_DOC_CHUNK + file_path) |
GCSに一旦あげます。
ベクトル化
chunkにあげたファイルをベクトル化して、vertexAIのインデックスにアップロードできる形式に加工します。
ベクトル化にはOpenAIのAPIを利用します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
from langchain_openai import OpenAIEmbeddings import json os.environ["OPENAI_API_KEY"] = "OpenAIのAPIキー" embeddings = OpenAIEmbeddings( model="text-embedding-3-large", dimensions=1024 ) for file_path in os.listdir(BUCKET_NAME + "/" + GCS_DOC_CHUNK): with open(BUCKET_NAME + "/" + GCS_DOC_CHUNK + file_path, "r") as f: text = f.read() vector = embeddings.embed_query(text) json.dump({"id": file_path.replace(".txt", ""), "embedding": vector}, open(BUCKET_NAME + "/" + GCS_BATCH_ROOT + file_path.replace(".txt", ".json"), "w")) |
今回は{ID, ベクトル}のjson(jsonl)形式で作成しましたが、定義できるファイル形式やフィールドは他にもあります。
詳細が気になる方はVertex AIのドキュメントの以下のページをご確認ください。
https://cloud.google.com/vertex-ai/docs/vector-search/setup/format-structure?hl=ja
|
1 2 |
for file_path in os.listdir(BUCKET_NAME + "/" + GCS_DOC_CHUNK): upload_from_filename(BUCKET_NAME + "/" + GCS_DOC_CHUNK + file_path, GCS_DOC_CHUNK + file_path) |
GCSにアップロードします。
GCPのVertex AIのベクトル検索を利用する
作成したベクトルデータをアップロードしてベクトルDBを作成します。
GCPのGUIベースで解説していきます。
Vertex AIのベクトル検索のページを開き、右上の新しいインデックスを作成を選択します
表示名はrag_test_indexとし、ベクトルデータを保存したGCSのフォルダを指定します。
次元数はベクトルの次元数と合わせましょう(今回は1024)
近似近傍数は20にし、あとはデフォルトで問題ないです。
アップロードには1時間ほどかかります・・・・
アップロードの時間潰しに今何をやっているかを軽く説明します。
GCP上でベクトル検索を行うモデルを作成するには以下の工程が必要になります。
1：検索対象となるデータの準備
IDとembeddingの要素を持った特定の形式でアップロード
{ID: 0, “embedding”: [1,2,3,4,…]}{}
2：インデックスの作成 (今ここ)
ベクトルデータをVertex AI上で扱えるようにしたデータ形式に加工
3：インデックスエンドポイントの作成
予測データを投げて結果を受け取るためのエンドポイントの作成
4：インデックスをインデックスエンドポイントにデプロイ
予測データを投げた時にインデックスでベクトル的に近いものを検索して返す
おそらく、ここまで読んでもまだインデックスの作成は完了していないと思います。
待っている間に、インデックスエンドポイントの作成も行います。
こちらは数分で完了すると思います。
インデックスの作成が完了したら次はインデックスエンドポイントへデプロイします。
終わっていない場合は次の「RAGの実装」を先にやるのが良いと思います。
作成したインデックスのデプロイをクリックし、エンドポイントへデプロイします。
デプロイにも1時間ほどかかるので、ベクトル検索について資料を読むか、次の「RAGの実装」を進めるでも良いと思います。
デプロイが完了すると、インデックスの詳細からデプロイされたインデックスをクリックすることができるようになります。
それをクリックすると、デプロイしたインデックスに対してリクエストを送るコードが表示されます。
<FEATURE_VECTOR>を検索したいベクトルにする必要があるのでコードを整形します。
ついでに neighbor_count=3 にして3件だけ返すようにします。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 |
from google.cloud import aiplatform_v1 # Set variables for the current deployed index. API_ENDPOINT="{伏字}" INDEX_ENDPOINT="{伏字}" DEPLOYED_INDEX_ID="{伏字}" # Configure Vector Search client client_options = { "api_endpoint": API_ENDPOINT } vector_search_client = aiplatform_v1.MatchServiceClient( client_options=client_options, ) ##### #以下を追加 ##### input = "真の「ベンチャー」の優位性はなんですか？" vector = embeddings.embed_query(text) # Build FindNeighborsRequest object datapoint = aiplatform_v1.IndexDatapoint( feature_vector=vector ) query = aiplatform_v1.FindNeighborsRequest.Query( datapoint=datapoint, # The number of nearest neighbors to be retrieved neighbor_count=3 # ここも少なくしても良いと思います。 ) request = aiplatform_v1.FindNeighborsRequest( index_endpoint=INDEX_ENDPOINT, deployed_index_id=DEPLOYED_INDEX_ID, # Request can have multiple queries queries=[query], return_full_datapoint=False, ) # Execute the request response = vector_search_client.find_neighbors(request) # Handle the response print(response[0]) |
入力文に近い順にデータが返されます。
ベクトルデータのIDをチャンクのファイル名にしているので、返却されたIDを利用してGCSのファイルを読みにいきます。
|
1 2 3 4 5 6 7 8 9 |
chunk_id = response.nearest_neighbors[0].neighbors[0].datapoint.datapoint_id + ".txt" def read_gcs(chunk_id): storage_client = storage.Client(project=PROJECT_ID) bucket = storage_client.get_bucket(BUCKET_NAME) blob = storage.Blob(GCS_DOC_CHUNK + chunk_id, bucket) return blob.download_as_text() read_gcs(chunk_id) |
これで、入力文と類似する内容をベクトル検索することができるようになりました。
RAGの実装
参照するデータの準備もできたので、いよいよRAGの実装に入ります。
|
1 2 |
from langchain_openai import ChatOpenAI chat = ChatOpenAI(model_name="gpt-4o") |
まずはChatGPTを利用できるようにします。
|
1 2 |
chat.invoke("こんにちは").content # こんにちは！今日はどのようなお手伝いができますか？ |
返答が返ってきていればOKです。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
from langchain_openai import ChatOpenAI from langchain.prompts.chat import ( ChatPromptTemplate, SystemMessagePromptTemplate, AIMessagePromptTemplate, HumanMessagePromptTemplate, ) # systemメッセージプロンプトテンプレートの準備 template = '''あなたはGMOのスピリットベンチャー宣言のフレーズを回答するアシスタントです。次の引用文を元に回答してください " {input_sv} "''' system_message_prompt = SystemMessagePromptTemplate.from_template(template) # humanメッセージプロンプトテンプレートの準備 human_template="{text}" human_message_prompt = HumanMessagePromptTemplate.from_template(human_template) # chatプロンプトテンプレートの準備 chat_prompt = ChatPromptTemplate.from_messages([system_message_prompt, human_message_prompt]) |
プロンプトを作成します。
動くかテストしてみます。
今回は入力文を与えてベクトル検索をした結果が返ってきたと仮定してテストします。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
# 入力文 input = "真の「ベンチャー」の優位性はなんですか？" # ベクトル検索 # chunk = func(input) chunk = "我々は「ベンチャー」である。技術力や経営手法がクローズアップされる企業は数多くありますが、真の「ベンチャー」の優位性はスピリットの共有・徹底にあります。テクニックやスキル、マニュアルを偏重する旧来の経営手法においては、理論的に経営資源を扱ってきました。人も同様、基本的には性悪説に基づき利益と恐怖による統制が原則と考えられてきたのです。" # 応答生成 chat(chat_prompt.format_prompt( input_sv = chunk, text=input ).to_messages()).content # 応答：真の「ベンチャー」の優位性は、スピリットの共有・徹底にあります。 |
正しく動いてそうです。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 |
def search_vector(vector): # Configure Vector Search client client_options = { "api_endpoint": API_ENDPOINT } vector_search_client = aiplatform_v1.MatchServiceClient( client_options=client_options, ) # Build FindNeighborsRequest object datapoint = aiplatform_v1.IndexDatapoint( feature_vector=vector ) query = aiplatform_v1.FindNeighborsRequest.Query( datapoint=datapoint, # The number of nearest neighbors to be retrieved neighbor_count=1 ) request = aiplatform_v1.FindNeighborsRequest( index_endpoint=INDEX_ENDPOINT, deployed_index_id=DEPLOYED_INDEX_ID, # Request can have multiple queries queries=[query], return_full_datapoint=False, ) # Execute the request response = vector_search_client.find_neighbors(request) return response.nearest_neighbors[0].neighbors[0].datapoint.datapoint_id def rag(text): # 入力文ベクトル化 vector = embeddings.embed_query(text) # ベクトル検索 chunk_id = search_vector(vector) chunk = read_gcs(chunk_id + ".txt") print(f"チャンクID:{chunk_id}\nチャンク:\n{chunk}\n") # 応答生成 print("応答 :\n" + chat(chat_prompt.format_prompt(input_sv = chunk,text=text).to_messages()).content) |
エンドポイントへインデックスのデプロイが完了しているのであれば、先ほど仮定していた部分を補完しつつ実装していきます。
応答を確認してみます。
|
1 2 3 4 5 6 7 8 9 |
rag("真の「ベンチャー」の優位性はなんですか？") # 実行結果 チャンクID:sv_2 チャンク: 我々は「ベンチャー」である。技術力や経営手法がクローズアップされる企業は数多くありますが、真の「ベンチャー」の優位性はスピリットの共有・徹底にあります。テクニックやスキル、マニュアルを偏重する旧来の経営手法においては、理論的に経営資源を扱ってきました。人も同様、基本的には性悪説に基づき利益と恐怖による統制が原則と考えられてきたのです。 応答 : 真の「ベンチャー」の優位性は、スピリットの共有・徹底にあります。 |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 |
rag("目標とはなんですか") # 実行結果 チャンクID:sv_19 チャンク: 【目標面】 目標とは (1)55カ年計画に基づく売上・利益と数値化された行動 2051年に売上10兆円・利益1兆円を達成 (2)スピリットベンチャー宣言の共有・徹底 (3)結果としての「笑顔」「感動」―で構成される。 目標を立て、目標に向かって走る"人財"で組織する。 経営はすべて逆算思考。目的・ゴールから考えよう。 応答 : 目標とは、GMOのスピリットベンチャー宣言において、以下のように定義されています： 1. 55カ年計画に基づく売上・利益と数値化された行動：具体的には、2051年に売上10兆円・利益1兆円を達成すること。 2. スピリットベンチャー宣言の共有・徹底。 3. 結果としての「笑顔」「感動」を得ること。 これらの目標を立て、それに向かって走る「人財」によって組織されることが強調されています。また、経営はすべて逆算思考で行われ、目的やゴールから考えることが重要とされています。 |
割と良い精度で出ました。よかったです。
まとめ
今回はOpenAIのAPIとGCPのVertex AIのベクトル検索を利用してRAGを作成しました。
精度は度外視でシンプルなものを作成したので、これを元に肉付けをしていくとより良いものができると思います。(特にチャンクの分割やベクトル化の部分)
より良いものができるとは思いますが、、、
・ベクトル検索を使う必要があるのか(tf-idfなどの古典的アルゴリズムで検索するのはダメなのか？)
・参照する情報をプロンプトに全部含めて回答するだけでも良い回答が得られる
などなど、そもそもRAGである必要がない場合もあるので、実際の要望と今回の実装&今後の改修の手間を比較して決めていくのが重要かと思います。
ここまで読んでいただきありがとうございました。
