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

2024年9月8日 10:59

はじめに

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週間以内に再生されたトラックを最大50曲取得します。
オーディオフィーチャーの取得:
- 各トラックのオーディオフィーチャー（音響的特徴）を取得します。
トラック情報の追加:
- 曲名とアーティスト名をオーディオフィーチャーデータに追加します。

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

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

新しいプレイリストの作成:
- 指定された名前と説明でプレイリストを作成します。
プレイリストへのトラック追加:
- 指定されたトラック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上で高度な音楽キュレーションと自動プレイリスト作成が可能になります。以下に、これらのツールの連携した使用方法と機能を説明します：

音楽分析と選曲（SpotifySearchTool）:
- ユーザーの最近再生した曲（過去1週間、最大50曲）を取得します。
- 各曲の詳細な音楽特性（アコースティック度、ダンス適性、エネルギー、テンポなど）を分析します。
- 曲名やアーティスト情報も含めた包括的なデータセットを生成します。
プレイリスト作成（SpotifyPlaylistTool）:
- 分析結果に基づいて選択された曲をプレイリストに追加します。
- 新しいプレイリストを作成し、名前と説明を設定します。
- 選択された曲をプレイリストに追加します。

連携の流れ: