【Google Cloud】RAG × Gemini で自分専用の音楽ディグエージェントを作ってみた

本記事は  AI・MLウィーク  1日目の記事です。
💻  告知記事  ▶▶ 本記事 ▶▶  2日目  📱

はじめに

こんにちは!入社2年目の清水です。
音楽ライブやフェスに行くのが趣味で、今年はすでに 20 本ほど予定が入っています ! 一方で、自分好みの音楽を見つける「音楽ディグ」は意外と難しいと感じています。
そこで本記事では、Google Cloud の AI（Gemini）を使って 「自分専用の音楽ディグエージェント」 を作ってみた取り組みを紹介します。音楽ディグに悩んでいる方や、Google Cloud の AI の活用例に興味のある方は、ぜひ参考にしてみてください !

「自分専用の音楽ディグエージェント」とは

本エージェントは、 「自分の過去の音楽体験（記憶）」 をGeminiに理解させ、 パーソナライズされた音楽コンシェルジュ として機能する仕組みです。 単なる検索や類似曲の列挙ではなく、「なぜその音楽が好きなのか」という好みの背景（テンポ／リズム／高揚感など）を踏まえて、 ”今の自分に刺さる可能性が高い”新しい出会い につなげることを目指しています !

2つのコア機能

1. 音楽の発掘
自分の楽曲嗜好メモ（好き・苦手の傾向など）を基に、YouTube にある膨大な音源から自分に合いそうな未知のアーティストや楽曲を見つけます。ただ似た曲を機械的に並べるのではなく、好みの理由まで考慮して候補を提示します。

2. 体験までをスムーズにつなぐ仕組み
発掘した楽曲は、自動で YouTube のプレイリストへ反映されます。「探す → 選ぶ → まとめる」といった手間を減らし、聴く体験に集中できるようにしました。

従来のレコメンド機能との違い

このエージェントは、従来の音楽レコメンドよりも “あなたが好きになる理由”を深く理解できる 点が特徴です。

1. 好みの背景まで読み取る
視聴履歴のような行動データだけではなく、あなたが書いた文章からなぜその曲が刺さったのかを読み取ります。 歌詞の雰囲気、ライブでの感情、サウンドの質感など、言葉にしたニュアンスがそのまま選曲に反映されます。

2. 理由がわかるから、ディグがもっと楽しい
曲を提案するだけでなく、なぜその曲があなたに合いそうなのか を説明してくれるので、 「そうそう、そこが好きなんだよ…！」という納得感と一緒に深掘りできます。

例えば「2000年代の曲でおすすめを教えて」と依頼した場合、従来のレコメンドだとその年代のヒット曲がずらっと並ぶだけになりがちです。でも、このエージェントは同じ2000年代の中でも “あなたの文脈に合いそうな曲だけ” を拾ってくれます !

実装を見送った機能

• 当初構想
  特定のジャンルやアーティストに関連するライブ・フェス情報を Web 上からエージェントが自動で巡回・収集・選別し、 数あるイベント情報の中から本当に関心度の高いものだけを抽出して届ける仕組みを想定していました。
• 見送り理由
  Gemini による Google 検索の挙動が安定せず、取得結果の 再現性・信頼性を担保できない状態 だったため、今回は実装を見送りました。 具体的には、最新情報を取得しようとした際に、実在しないアーティスト名や楽曲、ライブ情報を生成してしまうハルシネーションが確認されました。

そのため、ライブ情報も含めた包括的なレコメンドエージェントの実装はいったん見送り、まずは安定して価値を提供できる 「音楽の発掘 → プレイリストへの反映」 に集中する判断としました。

アーキテクチャ概要

このエージェントは、個人の音楽的な好み（非構造化データ） を起点に、YouTube のプレイリスト生成 という実際の音楽体験へとつなげる構成になっています。 中核となるのは、Google Cloud の AI スタックを活用した 「記憶 × 推論 × 実行」 のパイプライン です。

構成図

※Gemini や Vector Search は、Vertex AI上で動作しています。

主要コンポーネントと役割

このシステムは、大きく4つのレイヤーで構成されています。

1. データ蓄積レイヤー

• 使用技術 : Cloud Storage
• 使用用途 : ユーザーの好みや過去の音楽体験（好きなアーティスト、刺さったポイント、気分の傾向など）をテキストデータとして蓄積します。 あえて厳密に構造化せず、「このテンション感が好き」「この空気感が刺さる」といった曖昧で人間的な表現をそのまま残すことを重視しています。

2. 記憶の検索（RAG）レイヤー

• 使用技術 : Vertex AI Vector Search
• 使用用途 : 蓄積したテキストデータをベクトル化し、現在のリクエストに近い 「過去の好みの文脈」 を高速に検索します。 これにより、最近ハマりやすい音楽の系統やテンションが上がりやすい曲調、無意識に好んでいる傾向といった言語化しづらい嗜好を、検索可能な形で扱えるようにしています。

3. 推論・統合（Brain）レイヤー

• 使用技術 : Vertex AI（Gemini 2.5 Pro）
• 使用用途 : このレイヤーが、エージェントの 「頭脳」 にあたります。 Vector Search から取得した過去の嗜好文脈（RAG）と現在のリクエスト内容（気分・用途 など）を統合し、「今の自分に合いそうな音楽は何か」 を総合的に判断します。単なる類似検索ではなく、「なぜこの曲を勧めるのか」 という理由づけを伴った推論を行う点が特徴です。

4. 実行（Execution）レイヤー

• 使用技術 : Cloud Run functions / YouTube Data API
• 使用用途 : 推論結果を実際の行動に落とし込むレイヤーです。Gemini が選定した楽曲をもとに、YouTube のプレイリストを自動生成します。 「考えるだけ」で終わらせず、すぐに聴ける状態まで自動でつなげることで、音楽体験として完結させます。

フェーズ1 : Google Cloud プロジェクトの準備

最初のフェーズでは、エージェント開発の土台となるGoogle Cloud の開発環境を整えます。

1. Google Cloud プロジェクトの作成

まずは、開発用のGoogle Cloudプロジェクトを作成します。 このプロジェクトが、以降すべてのリソース（AI・関数・ストレージ）の管理単位になります。 PoC用途であれば、特別な設定は不要で、デフォルト設定のまま作成して問題ありません。

2. 必要なAPIの有効化

次に、今回のエージェントで利用するAPIを有効化します。 Google Cloud Console の「APIとサービス」→「ライブラリ」 から、以下の6つを検索して有効化します。

API表示名 (Console)	サービス名 (CLI用)	用途
Cloud Storage API	storage.googleapis.com	RAG用の元データ（テキストファイル）を保存するために使用
Service Networking API	servicenetworking.googleapis.com	Vertex AIのインデックス作成時に使用
Vertex AI API	aiplatform.googleapis.com	Gemini（推論）やベクトル検索（RAG）の中核となるAPI
Cloud Functions API	cloudfunctions.googleapis.com	エージェントのロジックを実行するサーバーレス環境
Cloud Build API	cloudbuild.googleapis.com	Cloud Run functionsをデプロイする際、コンテナをビルドするために使用
YouTube Data API v3	youtube.googleapis.com	YouTube のプレイリスト作成・操作に使用

フェーズ2 : RAG（記憶）の構築

このフェーズは、本エージェントに「自分らしさ」を持たせるための最重要パートです。 ここで構築するRAG（Retrieval Augmented Generation）は、単なる検索用データではなく、「あなたの音楽の記憶」そのものをAIに与える役割を担います。本フェーズでは、Google Cloud の Vertex AI Vector Search を使い、人間の感覚に近い曖昧な好みを、検索・推論に使える形へと変換していきます。

1. 自分の嗜好データの準備

まずは、AI に読み込ませる嗜好データを用意します。 ここでのポイントは、音楽の好みや過去の体験を あえて自然な文章のままテキスト化する ことです。 「アップテンポが好き」「この系統は刺さりやすい」といった、人間が普段感じている感覚を、そのまま言葉にして残していきます。
作成したデータは、preferences.jsonl（JSON Lines形式）として作成し、Cloud Storage にアップロードします。

記述例

{"id": "Artist_Chevon_Uptempo_FastPaced_Lyrics", "content":"好きなアーティスト：Chevon。「ダンス・デカダンス」のようなアップテンポなリズムと畳み掛けるようなリリックの疾走感に惹かれる。" }

このように、アーティスト名や印象に残った楽曲、「なぜ好きなのか」という理由をセットで記述することで、 後続の RAG 検索時に 「単なる似た曲」ではなく、「好みの背景まで含めた文脈」 を引き出せるようになります !

2．テキストを数値（ベクトル）に変換する

次に行うのが、ベクトル化（Embedding）です。 AI は文章をそのままの形では「意味の近さ」を基準に扱えないため、テキストを数値の集合（ベクトル）に変換し、意味を数値として比較できるようにします。 この処理によって、「このバンドが好き」「この楽曲の雰囲気が良かった」といった曖昧で感覚的な表現も、意味的な距離として検索・比較できるようになります。 本エージェントでは、Vertex AI Embeddings API を使用してベクトル化を行います。

以下のスクリプトをCloud Shellで実行すると、Cloud Storage 上の preferences.jsonl を一括でベクトル化し、結果を outputs フォルダに出力します。

実行スクリプトの詳細はこちら

#run_batch_embedding.py
import json
from google.cloud import storage
import vertexai
from vertexai.language_models import TextEmbeddingInput, TextEmbeddingModel

# ==========================================
# 設定エリア (ここをご自身の環境に合わせて変更してください)
# ==========================================
PROJECT_ID = "your-project-id"        # プロジェクトID
LOCATION = "your-region"          # リージョン (東京)
MODEL_NAME = "text-embedding-004"     # 使用するモデル名

# 入力ファイル (GCSパス)
INPUT_URI = "gs://yourbucket/preferences.jsonl"

# 出力ファイル (GCSパス) ※必ず末尾にファイル名まで書いてください
OUTPUT_URI = "gs://yourbucket/outputs/final_index_data.jsonl"
# ==========================================

def parse_gcs_uri(uri):
    """gs://bucket/path/to/file を bucket_name と blob_name に分解する"""
    if not uri.startswith("gs://"):
        raise ValueError(f"GCS URIは gs:// で始まる必要があります: {uri}")
    parts = uri.replace("gs://", "").split("/", 1)
    if len(parts) < 2:
        raise ValueError(f"URIにファイルパスが含まれていません: {uri}")
    return parts[0], parts[1]

def generate_embeddings_gcs_to_gcs():
    print("--- 処理を開始します ---")
    
    # 1. Vertex AI の初期化
    print(f"Vertex AI 初期化: Project={PROJECT_ID}, Location={LOCATION}")
    vertexai.init(project=PROJECT_ID, location=LOCATION)
    
    # モデルのロード
    model = TextEmbeddingModel.from_pretrained(MODEL_NAME)
    
    # 2. GCSクライアントの初期化
    storage_client = storage.Client(project=PROJECT_ID)

    # 入力と出力のBlob（ファイルオブジェクト）を準備
    in_bucket_name, in_blob_name = parse_gcs_uri(INPUT_URI)
    out_bucket_name, out_blob_name = parse_gcs_uri(OUTPUT_URI)

    input_blob = storage_client.bucket(in_bucket_name).blob(in_blob_name)
    output_blob = storage_client.bucket(out_bucket_name).blob(out_blob_name)

    print(f"読み込み元: {INPUT_URI}")
    print(f"書き込み先: {OUTPUT_URI}")

    # 3. GCSファイルをストリーミングで読み書き
    # input_blob を読み込みモード("r")、output_blob を書き込みモード("w")で開く
    try:
        with input_blob.open("r", encoding="utf-8") as fin, \
             output_blob.open("w", encoding="utf-8") as fout:
            
            count = 0
            for line in fin:
                if not line.strip():
                    continue

                try:
                    data = json.loads(line)
                    
                    # コンテンツとIDの取得 (キー名の揺れに対応)
                    text_content = data.get("content") or data.get("text")
                    record_id = data.get("id")
                    title = data.get("title", "") # タイトルがあれば取得

                    if not text_content or not record_id:
                        print(f"スキップ (必須データ不足): {line.strip()}")
                        continue

                    # Vertex AI 用の入力データ作成
                    input_item = TextEmbeddingInput(
                        text=text_content,
                        task_type="RETRIEVAL_DOCUMENT", # 検索対象ドキュメントとして設定
                        title=title
                    )

                    # APIコール (埋め込み生成)
                    embeddings = model.get_embeddings([input_item])
                    vector = embeddings[0].values

                    # Vector Search インデックス用フォーマットで作成
                    output_record = {
                        "id": str(record_id),
                        "embedding": vector
                    }

                    # GCSへ書き込み
                    fout.write(json.dumps(output_record) + "\n")
                    
                    count += 1
                    if count % 10 == 0:
                        print(f"{count} 件処理しました...")

                except Exception as e:
                    print(f"エラー発生 (行: {line.strip()[:30]}...): {e}")

        print(f"--- 完了！ 合計 {count} 件のデータを処理しました ---")
        print(f"出力ファイル: {OUTPUT_URI}")

    except Exception as e:
        print(f"ファイルオープンまたはアクセス中にエラーが発生しました: {e}")
        print("ヒント: 出力先のフォルダが存在しない場合は、GCSでは自動的に作成されるので問題ありませんが、権限などを確認してください。")

if __name__ == "__main__":
    generate_embeddings_gcs_to_gcs()

変換後データの例

{"id": "Artist_Chevon_Uptempo_FastPaced_Lyrics", "embedding": [-0.0046165334060788155, 0.01297974493354559, -0.01594969443976879, 0.0024088171776384115, -0.005221575498580933, -0.027011863887310028, -0.00156546535436064, 0.020216722041368484, ...

ベクトル化後のデータには、テキストの内容に対応する数値ベクトルが付与されます。このベクトル同士の距離を使うことで、「意味的に近い好み」 を高速に検索できるようになります。

3．インデックスの作成とデプロイ

このフェーズでは、ベクトル化したデータを「実際に検索できる状態」 にします。 ここで作成するインデックスは、いわばあなたの音楽の記憶を格納した検索辞書のような存在です。

3-1. インデックスの作成

まずは、Vertex AI 上にベクトル検索用のインデックスを作成します。

• Vertex AI コンソールから 「ベクトル検索」 を選択
• データソースとして、Cloud Storage上の outputsフォルダを指定
  ※インデックス作成時は、outputsフォルダにあるファイルの拡張子を .json に変更する
• 主な設定値は以下の通り
  • ディメンション : 768（Text Embedding APIで生成されるベクトルの次元数）
  • 近似近傍数：20
  • シャードのサイズ：小
  • そのほかの項目はデフォルト値のまま
• 設定内容を確認し、インデックスを作成

このインデックスにより、「今のリクエストに意味的に近い過去の嗜好」を高速に検索できるようになります。

3-2. エンドポイントへのデプロイ

作成したインデックスは、そのままでは外部から利用できません。 そこで、インデックスをエンドポイントにデプロイします。エンドポイントは任意の表示名で他はデフォルトで作成してください。このデプロイ処理には、30 分〜 1 時間程度かかることがあります。

これにより、Cloud Run functionsやGemini（推論フェーズ）といった外部コンポーネントから、 「今のリクエストに近い過去の好み」 をリアルタイムで検索できるようになります !

フェーズ3 : 音楽ディグエージェントの実装

ここからは、これまでに用意した「記憶（RAG）」と「推論（Gemini）」をつなぎ、実際に動くエージェント を実装していきます。 本フェーズでは、Google Cloud のサーバーレス環境 Cloud Run functions 上に、Gemini 2.5 Pro を頭脳とした音楽ディグエージェントを構築します。

1. YouTube操作用の認証設定

YouTube で「プレイリストを作成する」には、通常OAuth 2.0 によるユーザー認証が必要です。 ただし今回は、開発のしやすさと誰でもすぐに試せる手軽さを優先し、API キーのみで操作する構成を採用しました。

1-1. APIキーの作成

• Google Cloud Console のメニューから [API とサービス] > [認証情報] を開く
• 上部の [+ 認証情報を作成] をクリックし、[API キー] を選択
• 生成された文字列（例: AIzaSy...）を控えておく

1-2. キーの制限（セキュリティ推奨設定）

セキュリティ対策として、API キーの使用範囲を制限します。

• 作成したAPIキーをクリックし、編集画面を開く
• [API の制限] セクションで [キーを制限] を選択
• ドロップダウンから [YouTube Data API v3] のみを選択して保存

これにより、このキーは YouTube 操作専用として利用されます。

2. 開発環境の定義 (requirements.txt)

Cloud Run functions で使用するライブラリを定義します。

functions-framework==3.*
google-cloud-aiplatform==1.70.0
google-api-python-client>=2.100.0
google-auth>=2.20.0

3. エージェントの「脳」を作る (main.py)

ここがエージェントの中核です。この関数は、以下の流れを 1 リクエストで実行します。

• ユーザーのリクエストを受け取る
• Vector Search で過去の嗜好を検索（RAG）
• Gemini に文脈を渡して選曲を実行
• YouTube でプレイリストを生成

実行スクリプトの詳細はこちら

import os
import json
import traceback
import functions_framework
import vertexai
import datetime
from vertexai.language_models import TextEmbeddingModel
from google.cloud import aiplatform
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from vertexai.generative_models import GenerativeModel, Tool, GenerationConfig

# --- 設定 ---
MODEL_ID = "gemini-2.5-pro"
PROJECT_ID = os.environ.get("PROJECT_ID")
REGION = "us-central1" # Vertex AIのリソースがあるリージョンに合わせて変更してください
INDEX_ENDPOINT_ID = os.environ.get("INDEX_ENDPOINT_ID")
DEPLOYED_INDEX_ID = os.environ.get("DEPLOYED_INDEX_ID")
YT_API_KEY = os.environ.get("YT_API_KEY")

@functions_framework.http
def entry_point(request):
    try:
        if PROJECT_ID:
            vertexai.init(project=PROJECT_ID, location=REGION)
    except Exception:
        pass

    try:
        request_json = request.get_json(silent=True) or {}
        user_query = request_json.get('query') or "おすすめの曲"
        print(f"User Query: {user_query}")
        
        # --- 1. RAG (記憶の検索) ---
        past_prefs = ""
        if INDEX_ENDPOINT_ID and DEPLOYED_INDEX_ID:
            try:
                print("Starting Vector Search...")
                past_prefs = search_past_preferences(user_query)
                print(f"RAG Context Found: {len(past_prefs)} chars")
            except Exception as e:
                print(f"RAG Error (Skipping): {e}")
        else:
            print("RAG Skipped: Env vars not set.")

        # --- 2. Brain (Gemini with Grounding) ---
        # 過去の好み(past_prefs)を渡す
        ai_result = generate_dig_plan(user_query, past_prefs)
        
        # --- 3. Action (YouTube with Quota Safety) ---
        playlist_url = None
        system_note = None
        
        if ai_result.get("songs"):
            try:
                playlist_url = create_anonymous_playlist_url(ai_result["songs"])
            except Exception as e:
                print(f"YouTube Action Error: {e}")
                system_note = "※YouTubeの作成上限に達したため、プレイリストURLの発行をスキップしました。"

        response_data = {
            'message': 'Success',
            'comment': ai_result.get("comment"),
            'event_info': ai_result.get("event_info"),
            'playlist_url': playlist_url,
            'songs': ai_result.get("songs"),
            'rag_used': bool(past_prefs), # RAGが使われたか確認用フラグ
            'system_note': system_note
        }
        return json.dumps(response_data, ensure_ascii=False), 200, {'Content-Type': 'application/json'}

    except Exception as e:
        print(f"Critical Error: {e}")
        traceback.print_exc()
        return json.dumps({'error': str(e)}), 500, {'Content-Type': 'application/json'}


def search_past_preferences(query):
    """
    Vector Searchを実行して過去の好みを検索する
    """
    # テキストをベクトル化
    embedding_model = TextEmbeddingModel.from_pretrained("text-embedding-004")
    embeddings = embedding_model.get_embeddings([query])
    vector = embeddings[0].values

    # エンドポイント名整形
    ep_name = INDEX_ENDPOINT_ID
    if not ep_name.startswith("projects/"):
        ep_name = f"projects/{PROJECT_ID}/locations/{REGION}/indexEndpoints/{INDEX_ENDPOINT_ID}"

    # 検索実行
    endpoint = aiplatform.MatchingEngineIndexEndpoint(index_endpoint_name=ep_name)
    response = endpoint.find_neighbors(
        deployed_index_id=DEPLOYED_INDEX_ID,
        queries=[vector],
        num_neighbors=5
    )

    results = []
    for neighbors in response:
        for n in neighbors:
            # IDのアンダースコアをスペースに戻して読みやすくする
            text_content = n.id.replace('_', ' ')
            results.append(f"- {text_content}")
    
    return "\n".join(results) if results else ""


def generate_dig_plan(user_query, past_context):
    """
    Gemini実行部：RAGのコンテキストを受け取るように修正
    """
    tools = []
    
    # 検索ツール (辞書型定義で安定化)
    try:
        from google.cloud.aiplatform_v1beta1.types import Tool as GapicTool
        raw_tool = GapicTool(google_search={}) 
        search_tool = Tool._from_gapic(raw_tool)
        tools = [search_tool]
    except Exception:
        pass # 検索ツールエラー時はツールなしで続行

    # 日時設定
    now = datetime.datetime.now()
    today_str = now.strftime("%Y年%m月%d日")
    current_year = now.year

    # プロンプト：過去の好み(Context)を強く意識させる
    prompt = f"""
    【システム現在日時】
    現在は「{today_str}」です。
    
    【ユーザー情報 (重要！)】
    これはユーザーの過去の好みや記憶（RAG検索結果）です。
    この文脈に沿った提案を最優先してください。
    Context:
    {past_context}
    
    【検索ルール】
    WEB検索を使って「{current_year}年」の最新情報を探してください。
    
    User Request: {user_query}
    
    Task: 
    Based on the Context (user's taste) and latest Web Info, recommend 10 songs.
    If Context is empty, recommend general hits.
    
    Output JSON only.
    """

    try:
        model = GenerativeModel(MODEL_ID, tools=tools)
        response_schema = {
            "type": "object",
            "properties": {
                "comment": {"type": "string"},
                "event_info": {"type": "string"},
                "playlist_title": {"type": "string"},
                "songs": {"type": "array", "items": {"type": "object", "properties": {"artist": {"type": "string"}, "song": {"type": "string"}}, "required": ["artist", "song"]}}
            },
            "required": ["comment", "songs"]
        }

        response = model.generate_content(
            prompt,
            generation_config=GenerationConfig(
                response_mime_type="application/json",
                response_schema=response_schema,
                temperature=1.0
            )
        )
        return json.loads(response.text)
        
    except Exception as e:
        print(f"Gemini Error: {e}")
        return {"comment": "Error", "songs": []}

def create_anonymous_playlist_url(songs):
    if not YT_API_KEY: return None
    
    youtube = build('youtube', 'v3', developerKey=YT_API_KEY, cache_discovery=False)
    video_ids = []
    
    for s in songs:
        try:
            q = f"{s.get('artist')} {s.get('song')}"
            res = youtube.search().list(q=q, part="id", maxResults=1, type="video").execute()
            items = res.get('items', [])
            if items: video_ids.append(items[0]['id']['videoId'])
        except HttpError as e:
            # 403 Quota Exceeded なら即座に中断して、そこまでのリストを返す
            if e.resp.status == 403 and "quotaExceeded" in str(e):
                print("YouTube Quota limit reached.")
                break 
            continue
    
    if not video_ids: return None
    return f"https://www.youtube.com/watch_videos?video_ids={','.join(video_ids)}"

4. デプロイ

実装した関数を Cloud Run functions にデプロイします。Cloud Shell上でデプロイ用のフォルダを作成し、requirements.txtとmain.pyをアップロードしてデプロイしてください。

gcloud functions deploy music-agent \
  --gen2 \
  --runtime=python312 \
  --region=asia-northeast1 \
  --source=. \
  --entry-point=entry_point \
  --trigger-http \

5. 環境変数の設定

API キーやインデックス情報といった機密情報は、コードに直接書かず Cloud Run functions の環境変数として設定します。

• Cloud Run functions の一覧から対象の関数（例：music-agent）を選択
• 画面上部の [新しいリビジョンの編集とデプロイ] ボタンをクリック
• [コンテナ] > [変数とシークレット] を開く
• 以下の環境変数を追加

名前	値
PROJECT_ID	プロジェクトID
INDEX_ENDPOINT_ID	Vector Search のエンドポイントID
DEPLOYED_INDEX_ID	デプロイ済みインデックスID
YT_API_KEY	YouTube Data API キー

これで、HTTP リクエストひとつで「検索 → 思考 → 選曲」まで行う API が完成しました !

実際に使ってみる

ここまでで、音楽ディグエージェントの実装とデプロイが完了しました。 このセクションでは、実際に HTTP リクエストを送って、エージェントがどのように振る舞うかを確認します。

1. リクエストの送信

デプロイが完了すると、Cloud Run functions にはHTTP エンドポイントが発行されます。 今回は、curl を使ってシンプルにリクエストを送信します。

curl -X POST https://YOUR_FUNCTION_URL \
  -H "Content-Type: application/json" \
  -d '{"query": "私におすすめの楽曲を教えて"}'

2. レスポンスの内容

成功すると、以下のような JSON レスポンスが返ってきます（一部抜粋）。

{
  "message": "Success",
  "comment": "アップテンポで疾走感のある楽曲を中心に選びました。",
  "songs": [
    {"artist": "NEE", "song": "不革命前夜"},
    {"artist": "PEOPLE 1", "song": "DOGLAND"}
  ],
  "playlist_url": null,
  "rag_used": true
}

主な項目は以下の通りです。

• comment : Gemini が生成した、選曲の意図やコメント
• songs : レコメンドされた楽曲リスト（アーティスト名＋曲名）
• playlist_url : YouTube 上で再生可能なプレイリスト URL（作成できた場合）
• rag_used : 過去の嗜好データ（RAG）が使われたかどうかのフラグ

3. RAG の効果を確認する

エージェントは 過去に登録した嗜好データを参照したうえで推論を行っています。 一方で、RAG が取得できなかった場合は、次のように一般的なおすすめ曲を返すフォールバックが動作します。

{
  "message": "Success",
  "comment": "あなたの好みが分からなかったため、定番曲を中心に選びました。",
  "songs": [
    {"artist": "YOASOBI", "song": "夜に駆ける"},
    {"artist": "Official髭男dism", "song": "Pretender"}
  ]
}

4. YouTube Data APIのクウォータ

本エージェントではプレイリスト生成に YouTube Data API を使用していますが、APIのクォータ（利用割当）状況によってレスポンスの内容が変化します。

4-1. クウォータに余裕がある場合（通常時）

YouTube のプレイリストURLが生成されます。次のようにレスポンス内のURLをクリックすることで、選曲されたリストへ直接遷移し、そのまま再生が可能です。

{
  "message": "Success",
  "playlist_url": "https://www.youtube.com/watch_videos?video_ids=hN5MBlGv2Ac,UM9XNpgrqVk,ony539T074w,1FliVTcX8bQ,VPZK72W4Xxw",
  "songs": [
    {"artist": "Official髭男dism", "song": "Subtitle"},
    {"artist": "Vaundy", "song": "怪獣の花唄"}
  ]
}

4-2. クォータ切れの場合

利用状況によりAPI制限にかかった場合は、プレイリスト生成をスキップします。 この時 playlist_url は null となり、選曲された楽曲情報のテキストのみが返却されます。

このように、1リクエストで 「検索 → 推論 → 体験（プレイリスト）」までシームレスに完結する のが本エージェントの特徴です。

まとめ

本記事ではGoogle Cloud の AI（Gemini）を活用して、自分の音楽の好みを「記憶」として持ち、実際の音楽体験までつなげる音楽ディグエージェント を試作した取り組みを紹介しました。単なる「似ている曲の推薦」ではなく、「なぜその曲をすすめるのか」を言語化して返してくれる点は、従来のレコメンドとは違った“ディグ体験”として面白さを感じました !
一方で、当初構想していた Web 上の最新情報（リリース／ライブ情報）の自動取得については、Gemini による検索結果の安定性や再現性の観点から、今回は実装を見送りました。この点は、実際に手を動かしたからこそ見えてきた課題でもあり、「AI エージェントで何を任せるべきか」 を考える良い気づきになりました。
今後も、実際に使いながら改善していく中で「自分専用の音楽ディグエージェント」を少しずつ育てていきたいと思います !

執筆者：清水さくら アプリケーションエンジニア／2025 Japan All AWS Certifications Engineers