LoginSignup
11
4

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

ZOZOAdvent Calendar 2024

Day 15
@ShionSakuraiin株式会社ZOZO

【RAG】Vertex AI Agent Builderを使ってお手軽にAIエージェントを構築する話

Last updated at Posted at 2024-12-15

はじめに

🎄 本記事は ZOZO Advent Calendar 2024 シリーズ8の15日目です

社内の情報をRAGで読み込み、faq-botとしてLLMを利用するという事例を最近よく見かけます。本記事では、Google CloudのVertex AI Agent Builderを利用して、全自動RAG環境の構築方法をご紹介します！✨

・ 想定読者　　　　：RAGの概要を理解していて、サービス化を検討している方
・ 読んで分かること：Vertex AI Agent Builderを使った自動RAG環境の構築方法

Vertex AI Agent Builderとは？

Vertex AI Agent Builderとは「自然言語またはコードファーストのアプローチを使用して、AIエージェントとアプリケーションを作成」するためのサービスです。

RAG環境やAIエージェントをGUIから簡単に構築でき、更新に関しても自動で行う設定が可能なので、保守や運用のコストを軽減することが可能です。

本記事ではAIエージェントを量産できる体制を整えるため、Python SDKを利用して環境構築を行います。

システムの全体像

今回ご紹介するシステムは大きく分けて以下の2つのフローで構築しています。

  1. RAG環境の構築：依頼されたタイミングで対応
  2. AIエージェントによる回答：1番完了後であればいつでも利用可能

システム構成.jpg

社内ドキュメントをConfluence以外で管理している方は、1番のフローを少し置き換えて構築してみてください。

RAG環境の構築方法

構築方法について以下の手順で説明しています。

  • Step1：依頼されたスペースの資料をConfluence APIで取得
  • Step2：取得したデータをCloud Storageに保存
  • Step3：Cloud StorageとVertex AI Agent Builderを連携

📝 Step1：依頼されたスペースの資料をConfluence APIで取得

依頼されたスペースに存在するドキュメントの一覧を取得します。

def request_confluence(url):
    # getメソッドでURLに対してGETリクエストを送る
    headers = {'Authorization': f'Bearer {os.environ["CONFLUENCE_TOKEN"]}' }
    params = {'body': "storage"}
    response = requests.get(url, headers=headers, params=params)
    
    # レスポンスをテキスト形式に変換
    result = response.text
    result_dict = json.loads(result)
    return result_dict

def get_chiled_page_id_list(result_dict):
    page_id_list = []
    
    # 子ページの一覧を取得
    for page in result_dict["results"]:
        page_id = page["id"]
        page_id_list.append(page_id)

    return page_id_list

# 社内のドキュメントに合わせて設定を変更してください
document_page = os.environ["document_page"]
api_url = f"https://{document_page}/wiki/rest/api/content?spaceKey={space_key}&type=page&limit=25&start=0"


id_list = []
while api_url:
    result_dict = request_confluence(api_url)
    id_list += get_chiled_page_id_list(result_dict)

    try:
        api_url = f"https://{document_page}/wiki" + result_dict["_links"]["next"]
    except KeyError:
        api_url = ""

Confluence APIではスペースに存在するページのリストを25個ずつ読み取ることができるため、再帰的な処理によってスペース全体の資料リストを作成します。

次に、作成したリスト内のドキュメントを読み取ります。

def body_read_from_url(url):
    # getメソッドでURLに対してGETリクエストを送る
    headers = {'Authorization': f'Bearer {os.environ["CONFLUENCE_TOKEN"]}' }
    params = {'body': "storage"}

    response = requests.get(url, headers=headers, params=params)
    # レスポンスをテキスト形式に変換
    result = response.text

    # BeautifulSoupオブジェクトを作成
    soup = BeautifulSoup(result, 'html.parser')
    return soup

レスポンスはhtml形式のドキュメントとなっているため、BeautifulSoupオブジェクトに変換し、対象のドキュメントを任意の形式に整形してください。

document_list.append([page_id, html_text])

整形したデータは、page idとドキュメントのペアとしてリストに格納します。

💾 Step2：取得したデータをCloud Storageに保存

作成したドキュメントのリストをCloud Storageに格納していきます。

# GCSに新しいフォルダを作成してJSONLを保存
bucket_name = "hoge"
folder_name = "huga"

client = storage.Client()
bucket = client.get_bucket(bucket_name)

file_list = []

# 新しいフォルダを作成してファイルをアップロード
for document in document_list:
    page_id = document[0]
    file_path = f"{folder_name}/{page_id}.html"
    file_list.append(file_path)
    blob = bucket.blob(file_path)
    blob.upload_from_string(document[1].encode('utf-8'), content_type='text/html; charset=utf-8')

回答を生成するだけであればpage idの情報は不要なのですが、参考にした資料情報も将来的に使う可能性があるため、保存するファイル名にpage idを使用しています。

🔗 Step3：Cloud StorageとVertex AI Agent Builderを連携

Vertex AI Agent BuilderにおいてVector DBは、「データストア」と呼ばれており、以下のコードではデータストアの作成を行います。

from google.api_core.client_options import ClientOptions
from google.cloud import discoveryengine
from google.cloud import storage
import json

def create_data_store(data_store_id):
    project_id = "your-project"
    location = "global"
    
    client_options = (
        ClientOptions(api_endpoint=f"{location}-discoveryengine.googleapis.com")
        if location != "global"
        else None
    )

    # クライアントの作成
    client = discoveryengine.DataStoreServiceClient(client_options=client_options)

    # コレクション名
    parent = client.collection_path(
        project=project_id,
        location=location,
        collection="default_collection",
    )

    data_store = discoveryengine.DataStore(
        display_name=data_store_id,
        industry_vertical=discoveryengine.IndustryVertical.GENERIC,
        solution_types=[discoveryengine.SolutionType.SOLUTION_TYPE_SEARCH],
        content_config=discoveryengine.DataStore.ContentConfig.CONTENT_REQUIRED,
    )

    request = discoveryengine.CreateDataStoreRequest(
        parent=parent,
        data_store_id=data_store_id,
        data_store=data_store,
    )

    # リクエストの作成
    operation = client.create_data_store(request=request)

    print(f"Waiting for operation to complete: {operation.operation.name}")
    response = operation.result()

    print(response)

    return operation.operation.name

DBにデータを格納するためにはテーブルが必要というイメージが分かりやすいと思います。

次に、Cloud Storageに保存したデータをデータストアへ格納します。

from google.api_core.client_options import ClientOptions
from google.cloud import discoveryengine

def import_documents_gcs_sample(file_list, data_store_id):

    project_id = "your project"
    location = "global"

    client_options = (
        ClientOptions(api_endpoint=f"{location}-discoveryengine.googleapis.com")
        if location != "global"
        else None
    )

    client = discoveryengine.DocumentServiceClient(client_options=client_options)

    # e.g. projects/{project}/locations/{location}/dataStores/{data_store_id}/branches/{branch}
    collection = "default_collection"
    branch = "0"
    parent_value = f"projects/{project_id}/locations/{location}/collections/{collection}/dataStores/{data_store_id}/branches/{branch}"

    
    for file in file_list:
        gcs_uri = f"gs://{bucket_name}/{file}"
        
        request = discoveryengine.ImportDocumentsRequest(
            parent=parent_value,
            gcs_source=discoveryengine.GcsSource(
                input_uris=[gcs_uri],
                data_schema="content",
            ),
            reconciliation_mode=discoveryengine.ImportDocumentsRequest.ReconciliationMode.INCREMENTAL,
        )

        # リクエストの作成
        operation = client.import_documents(request=request)

        print(f"Waiting for operation to complete: {operation.operation.name}")
        response = operation.result()

        # メタデータの取得
        metadata = discoveryengine.ImportDocumentsMetadata(operation.metadata)

        print(response)
        print(metadata)

    return

これでVector DBが完成しました。

最後に、作成したデータストアに接続するためのアプリ化を行います。

from typing import List

from google.api_core.client_options import ClientOptions
from google.cloud import discoveryengine_v1 as discoveryengine

def create_engine_sample(engine_id, data_store_ids):
    project_id = "your-project"
    location = "global"

    client_options = (
        ClientOptions(api_endpoint=f"{location}-discoveryengine.googleapis.com")
        if location != "global"
        else None
    )

    client = discoveryengine.EngineServiceClient(client_options=client_options)

    parent = client.collection_path(
        project=project_id,
        location=location,
        collection="default_collection",
    )

    engine = discoveryengine.Engine(
        display_name="Test Engine",
        # Options: GENERIC, MEDIA, HEALTHCARE_FHIR
        industry_vertical=discoveryengine.IndustryVertical.GENERIC,
        # Options: SOLUTION_TYPE_RECOMMENDATION, SOLUTION_TYPE_SEARCH, SOLUTION_TYPE_CHAT, SOLUTION_TYPE_GENERATIVE_CHAT
        solution_type=discoveryengine.SolutionType.SOLUTION_TYPE_SEARCH,
        # For search apps only
        search_engine_config=discoveryengine.Engine.SearchEngineConfig(
            search_tier=discoveryengine.SearchTier.SEARCH_TIER_ENTERPRISE,
            search_add_ons=[discoveryengine.SearchAddOn.SEARCH_ADD_ON_LLM],
        ),
        data_store_ids=data_store_ids,
    )

    request = discoveryengine.CreateEngineRequest(
        parent=parent,
        engine=engine,
        engine_id=engine_id,
    )

    # リクエストの作成
    operation = client.create_engine(request=request)

    print(f"Waiting for operation to complete: {operation.operation.name}")
    response = operation.result()

    # メタデータの取得
    metadata = discoveryengine.CreateEngineMetadata(operation.metadata)

    print(response)
    print(metadata)

    return operation.operation.name

作成したアプリのIDは次の工程で利用するので控えておきましょう。

AIエージェントによる回答システムの構築

AIエージェントは以下の手順で構築します。

  • Step1：Slack Botの作成
  • Step2：Vertex AI Agent Builderを用いた回答の生成

Step1は通常のSlack Botと同様の手順になるので、本記事では省略します。

🔗 Step2：Vertex AI Agent Builderを用いた回答の生成

Geminiで回答を生成する際に、先ほどアプリ化したデータストアを参照することが可能です。

from vertexai import generative_models
import vertexai.preview.generative_models as generative_models
from vertexai.generative_models import GenerativeModel, Tool

def ai_agent(user_message):

    # AIエージェントの設定を行なってください
    system_prompt = f"""あなたは優秀なAIエージェントです。ユーザーからの質問に分かりやすく回答してください。"""

    # 先ほど作成したデータストアアプリのIDを入力してください
    datastore_id = "projects/your-project/locations/global/collections/default_collection/dataStores/hoge"

    tools = [
        Tool.from_retrieval(
            retrieval=generative_models.grounding.Retrieval(
                source=generative_models.grounding.VertexAISearch(datastore=datastore_id),
                disable_attribution=False,
            )
        ),
    ]

    model = GenerativeModel(
        "gemini-1.5-flash-002",
        system_instruction=[
            system_prompt,
        ],
        tools=tools,
    )

    response = model.generate_content(
    user_message,
    generation_config={
        "max_output_tokens": 4000,
        "temperature": 0,
        "top_p": 1,
        "top_k": 32
    },
    stream=False
    )

    try:
        print(response.candidates[0].content.parts[0].text)
        answer = response.candidates[0].content.parts[0].text
    except:
        print(response)
        answer = "もう一度わかりやすく質問してください！"
    return answer

datastore_idで先ほど作成したアプリのIDを指定することで、AIエージェントの完成です。

おわりに

本記事では、AIエージェントを量産するための簡単な環境構築方法についてご紹介しました。今回作成したAIエージェントをSlackアプリに組み込むことで、簡単にfaq-botを作成することができるので、皆さんの環境でも是非挑戦してみてください！🎉

11
4
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
11
4

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?