見出し画像

AI音楽キュレーション：LangGraph Studio x Spotify APIで作るAI DJ Assistant🎧

AstroPomeAI

はじめに

AIの進歩により、音楽キュレーションの世界が大きく変わろうとしています。ストリーミングサービスの普及で膨大な音楽にアクセスできるようになった一方、最適な曲を見つけることはより困難になっています。

この課題に対し、AIを活用した音楽推薦システムを考えました。AIは単なる補助ツールではなく、プレイリスト作成者や音楽キュレーターの創造性を拡張する新たなパートナーとなる可能性があるかもしれません。

本記事では、最新のAI技術であるLangGraphと、SpotifyのAPIを組み合わせた音楽キュレーションAIアシスタントの開発プロセスを詳しく紹介します。LangGraphの登場とLangChainの更新に伴い、以前作成したSpotify APIエージェントを再構築し、LangGraph Studioを活用してユニークなプレイリスト作成システムを構築しました。

以前作成したSpotify APIエージェントの記事


エージェントの再構築：

LangChainの最新バージョンに対応するため、Spotify APIと連携するエージェントを再設計しました。昨年開発したカスタムツールはLangChain v0.1で作成していましたが今回は、v0.2へ更新しました。

SpotifySearchTool

Spotify APIを使用して最近再生されたトラックの音楽特性（オーディオフィーチャー）を取得する機能を提供しています。主な機能は以下の通りです：

  1. 最近再生されたトラックの取得:

    • 過去1週間以内に再生されたトラックを最大50曲取得します。

  2. オーディオフィーチャーの取得:

    • 各トラックのオーディオフィーチャー（音響的特徴）を取得します。

  3. トラック情報の追加:

    • 曲名とアーティスト名をオーディオフィーチャーデータに追加します。

取得される主なオーディオフィーチャーには以下が含まれます：

  • acousticness（アコースティック度）

  • danceability（ダンス適性）

  • energy（エネルギー）

  • instrumentalness（楽器の度合い）

  • liveness（ライブ感）

  • loudness（音量）

  • speechiness（スピーチ度）

  • tempo（テンポ）

  • valence（明るさ）

from typing import Optional, Type
from langchain.pydantic_v1 import BaseModel, Field
from langchain_core.callbacks import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)
from langchain_core.tools import BaseTool
import spotipy
import json
from datetime import datetime, timedelta

class SpotifySearchTool(BaseTool):
    name = "SpotifySearchTool"
    description = (
        "A tool that fetches audio features of the most recently saved tracks from Spotify. "
        "This tool does not require any arguments.\n\n"
        """Description of Return Parameters:
           acousticness: Acoustic confidence. Ex: 0.00242 (0-1)
           danceability: Dance suitability. Ex: 0.585
           duration_ms: Duration in ms. Ex: 237040
           energy: Intensity measure. Ex: 0.842
           id: Spotify track ID. Ex: "2takcwOaAZWiXQijPHIx7B"
           instrumentalness: Vocal prediction. Ex: 0.00686
           key: Track key. Ex: 9 (-1-11)
           liveness: Audience presence. Ex: 0.0866
           loudness: Loudness in dB. Ex: -5.883
           mode: Track modality. Ex: 0
           speechiness: Spoken word presence. Ex: 0.0556
           tempo: Tempo in BPM. Ex: 118.211
           time_signature: Time signature. Ex: 4 (3-7)
           type: Object type. Allowed: "audio_features"
           valence: Musical positiveness. Ex: 0.428 (0-1)
        """
    )
    args_schema: Type[BaseModel] = BaseModel  # No arguments required
    spotify_token: str = Field(..., description="Access token for Spotify")

    def __init__(self, spotify_token: str, *args, **kwargs):
        if not spotify_token:
            raise ValueError("Please set Spotify access token")
        super().__init__(spotify_token=spotify_token, *args, **kwargs)

    def _run(
        self,
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        sp = spotipy.Spotify(auth=self.spotify_token)

        one_week_ago_date = (datetime.now() - timedelta(weeks=1)).strftime('%Y-%m-%d')
        result = sp.current_user_recently_played(limit=50, after=one_week_ago_date)

        tracks = [item['track']['id'] for item in result['items']]
        audio_features_list = [sp.audio_features(track)[0] for track in tracks]

        for i, item in enumerate(result['items']):
            track_info = item['track']
            audio_features_list[i]['song_name'] = track_info['name']
            audio_features_list[i]['artists'] = ', '.join([artist['name'] for artist in track_info['artists']])

        for features in audio_features_list:
            features.pop('uri', None)
            features.pop('track_href', None)
            features.pop('analysis_url', None)

        return json.dumps(audio_features_list)

    async def _arun(
        self,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> str:
        """Use the SpotifySearchTool asynchronously."""
        return self._run()

SpotifyPlaylistTool

Spotify APIを使用して新しいプレイリストを作成し、指定されたトラックをそのプレイリストに追加する機能を提供しています。主な機能は以下の通りです：

  1. 新しいプレイリストの作成:

    • 指定された名前と説明でプレイリストを作成します。

  2. プレイリストへのトラック追加:

    • 指定されたトラックID（複数可）をプレイリストに追加します。

from typing import Optional, Type, List
from langchain.pydantic_v1 import BaseModel, Field
from langchain_core.callbacks import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)
from langchain_core.tools import BaseTool
import spotipy

class SpotifyPlaylistInput(BaseModel):
    track_ids: List[str] = Field(description="List of Spotify track IDs to add to the playlist")
    playlist_name: str = Field(description="Name of the new playlist to be created")
    playlist_description: str = Field(description="Description for the new playlist")

class SpotifyPlaylistTool(BaseTool):
    name = "SpotifyPlaylistTool"
    description = (
        "A tool that creates a new playlist and adds tracks to it on Spotify. "
        "This tool requires a list of track IDs, a playlist name, and a playlist description."
    )
    args_schema: Type[BaseModel] = SpotifyPlaylistInput
    spotify_token: str = Field(..., description="Access token for Spotify")
    user_id: str = Field(..., description="User ID for Spotify")

    def __init__(self, spotify_token: str, user_id: str, *args, **kwargs):
        if not spotify_token:
            raise ValueError("Please set Spotify access token")
        if not user_id:
            raise ValueError("Please set Spotify user ID")
        super().__init__(spotify_token=spotify_token, user_id=user_id, *args, **kwargs)

    def _run(
        self,
        track_ids: List[str],
        playlist_name: str,
        playlist_description: str,
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        sp = spotipy.Spotify(auth=self.spotify_token)

        # Create a new playlist
        user_playlist = sp.user_playlist_create(self.user_id, playlist_name, public=False, collaborative=False, description=playlist_description)

        # Add tracks to the playlist
        sp.playlist_add_items(user_playlist['id'], items=track_ids, position=None)

        return f"Playlist '{playlist_name}' created with {len(track_ids)} tracks."

    async def _arun(
        self,
        track_ids: List[str],
        playlist_name: str,
        playlist_description: str,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> str:
        """Use the SpotifyPlaylistTool asynchronously."""
        return self._run(track_ids, playlist_name, playlist_description)

これら2つのツール、SpotifySearchToolとSpotifyPlaylistToolを組み合わせることで、Spotify上で高度な音楽キュレーションと自動プレイリスト作成が可能になります。以下に、これらのツールの連携した使用方法と機能を説明します：

  1. 音楽分析と選曲（SpotifySearchTool）:

    • ユーザーの最近再生した曲（過去1週間、最大50曲）を取得します。

    • 各曲の詳細な音楽特性（アコースティック度、ダンス適性、エネルギー、テンポなど）を分析します。

    • 曲名やアーティスト情報も含めた包括的なデータセットを生成します。

  2. プレイリスト作成（SpotifyPlaylistTool）:

    • 分析結果に基づいて選択された曲をプレイリストに追加します。

    • 新しいプレイリストを作成し、名前と説明を設定します。

    • 選択された曲をプレイリストに追加します。

連携の流れ:

  1. SpotifySearchToolを使用して、ユーザーの最近の音楽嗜好を分析します。

  2. 分析結果を基に、特定の条件（例：高いダンス適性、特定のテンポ範囲など）に合う曲を選択します。

  3. 選択された曲のIDをSpotifyPlaylistToolに渡し、新しいプレイリストを作成します。

使用例:

  • ムード別プレイリスト: valenceやenergyの値に基づいて、明るい曲や落ち着いた曲のプレイリストを作成。

  • ワークアウトプレイリスト: 高いenergyとdanceabilityを持つ曲を選んでプレイリストを作成。

  • ジャンルミックス: 様々な音楽特性を持つ曲をバランスよく選んでエクレクティックなプレイリストを作成。

LangGraphとLangGraph Studioの活用：

今回のプロジェクトでは、LangGraphを使用してエージェントのワークフローを構築し、LangGraph Studioを通じてそのワークフローを視覚化および管理しました。これにより、複雑なタスクを直感的に設計し、効率的に実行することができました。

ユーザーが"DJ用のダウンテンポでグルーヴィーな曲のプレイリスト作成"を要求すると、AIアシスタントがSpotifySearchToolを使用して曲の特徴を検索してプレイリストを作成します。

楽曲と合わせて音楽特性（ダンサビリティ、エネルギー、テンポなど）が表示されています。AIが何を根拠に楽曲を選んだか確認できます。

次に、AIアシスタントに選んだ曲を元にSpotifyPlaylistToolを使ってSpotifyのプレイリストを作成してもらいました。

SpotifySearchToolはユーザーの再生履歴を50件取得しその中から選曲しているためアーティストに偏りがあります。まずは自分好みの音楽をSpotify上で沢山聴く必要があります🎧。

いいなと思ったら応援しよう！