diff --git a/toolkits/spotify/arcade_spotify/tools/constants.py b/toolkits/spotify/arcade_spotify/tools/constants.py new file mode 100644 index 00000000..ff8bf38b --- /dev/null +++ b/toolkits/spotify/arcade_spotify/tools/constants.py @@ -0,0 +1,14 @@ +SPOTIFY_BASE_URL = "https://api.spotify.com/v1" + +ENDPOINTS = { + "player_get_playback_state": "/me/player", + "player_get_currently_playing": "/me/player/currently-playing", + "player_modify_playback": "/me/player/play", + "player_pause_playback": "/me/player/pause", + "player_skip_to_next": "/me/player/next", + "player_skip_to_previous": "/me/player/previous", + "player_seek_to_position": "/me/player/seek", + "tracks_get_track": "/tracks/{track_id}", + "tracks_get_recommendations": "/recommendations", + "tracks_get_audio_features": "/audio-features", +} diff --git a/toolkits/spotify/arcade_spotify/tools/models.py b/toolkits/spotify/arcade_spotify/tools/models.py new file mode 100644 index 00000000..9f625f32 --- /dev/null +++ b/toolkits/spotify/arcade_spotify/tools/models.py @@ -0,0 +1,29 @@ +from dataclasses import asdict, dataclass, field +from typing import Optional + + +@dataclass +class PlaybackState: + is_playing: Optional[bool] = None + progress_ms: Optional[int] = ( + None # Progress into the currently playing track or episode in milliseconds + ) + device_name: Optional[str] = None + device_id: Optional[str] = None + currently_playing_type: Optional[str] = None + album_id: Optional[str] = None + album_name: Optional[str] = None + album_artists: list[str] = field(default_factory=list) + album_spotify_url: Optional[str] = None + track_id: Optional[str] = None + track_name: Optional[str] = None + track_artists: list[str] = field(default_factory=list) + show_id: Optional[str] = None + show_spotify_url: Optional[str] = None + episode_name: Optional[str] = None + episode_spotify_url: Optional[str] = None + message: Optional[str] = None + + def to_dict(self) -> dict: + """Convert the PlaybackState instance to a dictionary, excluding None values.""" + return {k: v for k, v in asdict(self).items() if v is not None and v != []} diff --git a/toolkits/spotify/arcade_spotify/tools/player.py b/toolkits/spotify/arcade_spotify/tools/player.py index 427c9859..8fdb6f7e 100644 --- a/toolkits/spotify/arcade_spotify/tools/player.py +++ b/toolkits/spotify/arcade_spotify/tools/player.py @@ -1,162 +1,223 @@ from typing import Annotated, Optional -import httpx - from arcade.sdk import ToolContext, tool from arcade.sdk.auth import Spotify -from arcade.sdk.errors import ToolExecutionError - -SPOTIFY_BASE_URL = "https://api.spotify.com/v1" +from arcade.sdk.errors import RetryableToolError +from arcade_spotify.tools.utils import ( + convert_to_playback_state, + get_url, + handle_404_playback_state, + send_spotify_request, +) -async def _send_spotify_request( +# NOTE: This tool only works for Spotify Premium users +@tool(requires_auth=Spotify(scopes=["user-read-playback-state", "user-modify-playback-state"])) +async def adjust_playback_position( context: ToolContext, - method: str, - endpoint: str, - params: dict | None = None, - json_data: dict | None = None, -) -> httpx.Response: + absolute_position_ms: Annotated[ + Optional[int], "The absolute position in milliseconds to seek to" + ] = None, + relative_position_ms: Annotated[ + Optional[int], + "The relative position from the current playback position in milliseconds to seek to", + ] = None, +) -> Annotated[dict, "The updated playback state"]: + """Adjust the playback position within the currently playing track. + + Knowledge of the current playback state is NOT needed to use this tool as it handles + clamping the position to valid start/end boundaries to prevent overshooting or negative values. + + This tool allows you to seek to a specific position within the currently playing track. + You can either provide an absolute position in milliseconds or a relative position from + the current playback position in milliseconds. + + Note: + Either absolute_position_ms or relative_position_ms must be provided, but not both. """ - Send an asynchronous request to the Spotify API. + if (absolute_position_ms is None) == (relative_position_ms is None): + raise RetryableToolError( + "Either absolute_position_ms or relative_position_ms must be provided, but not both", + additional_prompt_content="Provide a value for either absolute_position_ms or relative_position_ms, but not both.", + retry_after_ms=500, + ) - Args: - context: The tool context containing the authorization token. - method: The HTTP method (GET, POST, PUT, DELETE, etc.). - endpoint: The API endpoint path (e.g., "/me/player/play"). - params: Query parameters to include in the request. - json_data: JSON data to include in the request body. + if relative_position_ms is not None: + playback_state = await get_playback_state(context) + if playback_state.get("device_id") is None: + playback_state["message"] = "No track to adjust position" + return playback_state - Returns: - The response object from the API request. + absolute_position_ms = playback_state["progress_ms"] + relative_position_ms - Raises: - ToolExecutionError: If the request fails for any reason. - """ - url = f"{SPOTIFY_BASE_URL}{endpoint}" - headers = {"Authorization": f"Bearer {context.authorization.token}"} + absolute_position_ms = max(0, absolute_position_ms) - async with httpx.AsyncClient() as client: - try: - response = await client.request( - method, url, headers=headers, params=params, json=json_data - ) - response.raise_for_status() - except httpx.RequestError as e: - raise ToolExecutionError(f"Failed to send request to Spotify API: {e}") + url = get_url("player_seek_to_position") + params = {"position_ms": absolute_position_ms} - return response + response = await send_spotify_request(context, "PUT", url, params=params) + + playback_state = handle_404_playback_state(response, "No track to adjust position", False) + if playback_state: + return playback_state + + response.raise_for_status() + + playback_state = await get_playback_state(context) + return playback_state -def _handle_spotify_api_error(response: httpx.Response): - """ - Handle errors from the Spotify API by mapping common status codes to ToolExecutionErrors. +# NOTE: This tool only works for Spotify Premium users +@tool(requires_auth=Spotify(scopes=["user-read-playback-state", "user-modify-playback-state"])) +async def skip_to_previous_track( + context: ToolContext, +) -> Annotated[dict, "The updated playback state"]: + """Skip to the previous track in the user's queue, if any""" + url = get_url("player_skip_to_previous") - Args: - response: The response object from the API request. + response = await send_spotify_request(context, "POST", url) - Raises: - ToolExecutionError: If the response contains an error status code. - """ - status_code_map = { - 401: ToolExecutionError("Unauthorized: Invalid or expired token"), - 403: ToolExecutionError("Forbidden: User does not have Spotify Premium"), - 429: ToolExecutionError("Too Many Requests: Rate limit exceeded"), + playback_state = handle_404_playback_state(response, "No track to go back to", False) + if playback_state: + return playback_state + + response.raise_for_status() + + playback_state = await get_playback_state(context) + + return playback_state + + +# NOTE: This tool only works for Spotify Premium users +@tool(requires_auth=Spotify(scopes=["user-read-playback-state", "user-modify-playback-state"])) +async def skip_to_next_track( + context: ToolContext, +) -> Annotated[dict, "The updated playback state"]: + """Skip to the next track in the user's queue, if any""" + url = get_url("player_skip_to_next") + + response = await send_spotify_request(context, "POST", url) + + playback_state = handle_404_playback_state(response, "No track to skip", False) + if playback_state: + return playback_state + + response.raise_for_status() + + playback_state = await get_playback_state(context) + + return playback_state + + +# NOTE: This tool only works for Spotify Premium users +@tool(requires_auth=Spotify(scopes=["user-read-playback-state", "user-modify-playback-state"])) +async def pause_playback( + context: ToolContext, +) -> Annotated[dict, "The updated playback state"]: + """Pause the currently playing track, if any""" + playback_state = await get_playback_state(context) + + # There is no current state, therefore nothing to pause + if playback_state.get("device_id") is None: + playback_state["message"] = "No track to pause" + return playback_state + # Track is already paused + if playback_state.get("is_playing") is False: + playback_state["message"] = "Track is already paused" + return playback_state + + url = get_url("player_pause_playback") + + response = await send_spotify_request(context, "PUT", url) + response.raise_for_status() + + playback_state["is_playing"] = False + return playback_state + + +# NOTE: This tool only works for Spotify Premium users +@tool( + requires_auth=Spotify( + scopes=["user-read-playback-state", "user-modify-playback-state"], + ) +) +async def resume_playback( + context: ToolContext, +) -> Annotated[dict, "The updated playback state"]: + """Resume the currently playing track, if any""" + playback_state = await get_playback_state(context) + + # There is no current state, therefore nothing to resume + if playback_state.get("device_id") is None: + playback_state["message"] = "No track to resume" + return playback_state + # Track is already playing + if playback_state.get("is_playing") is True: + playback_state["message"] = "Track is already playing" + return playback_state + + url = get_url("player_modify_playback") + + response = await send_spotify_request(context, "PUT", url) + response.raise_for_status() + + playback_state["is_playing"] = True + return playback_state + + +# NOTE: This tool only works for Spotify Premium users +@tool( + requires_auth=Spotify( + scopes=["user-read-playback-state", "user-modify-playback-state"], + ) +) +async def start_tracks_playback_by_id( + context: ToolContext, + track_ids: Annotated[ + list[str], + "A list of Spotify track (song) IDs to play. Order of execution is not guarenteed.", + ], + position_ms: Annotated[ + Optional[int], + "The position in milliseconds to start the first track from", + ] = 0, +) -> Annotated[dict, "The updated playback state"]: + """Start playback of a list of tracks (songs)""" + url = get_url("player_modify_playback") + body = { + "uris": [f"spotify:track:{track_id}" for track_id in track_ids], + "position_ms": position_ms, } - if response.status_code in status_code_map: - raise status_code_map[response.status_code] - elif response.status_code >= 400: - raise ToolExecutionError(f"Error: {response.status_code} - {response.text}") + response = await send_spotify_request(context, "PUT", url, json_data=body) + response.raise_for_status() + + playback_state = await get_playback_state(context) + return playback_state -@tool( - requires_auth=Spotify( - scopes=["user-modify-playback-state"], - ) -) -async def pause( - context: ToolContext, - device_id: Annotated[ - Optional[str], - "The id of the device this command is targeting. If omitted, the active device is targeted.", - ] = None, -) -> Annotated[str, "Success string confirming the pause"]: - """Pause the current track""" - endpoint = "/me/player/pause" - params = {"device_id": device_id} if device_id else {} - - response = await _send_spotify_request(context, "PUT", endpoint, params=params) - if response.status_code >= 200 and response.status_code < 300: - return "Playback paused" - else: - _handle_spotify_api_error(response) - - -@tool( - requires_auth=Spotify( - scopes=["user-modify-playback-state"], - ) -) -async def resume( - context: ToolContext, - device_id: Annotated[ - Optional[str], - "The id of the device this command is targeting. If omitted, the active device is targeted.", - ] = None, -) -> Annotated[str, "Success string confirming the playback resume"]: - """Resume the current track, if any""" - endpoint = "/me/player/play" - params = {"device_id": device_id} if device_id else {} - - response = await _send_spotify_request(context, "PUT", endpoint, params=params) - if response.status_code >= 200 and response.status_code < 300: - return "Playback resumed" - else: - _handle_spotify_api_error(response) - - -@tool( - requires_auth=Spotify( - scopes=["user-read-playback-state"], - ) -) +@tool(requires_auth=Spotify(scopes=["user-read-playback-state"])) async def get_playback_state( context: ToolContext, ) -> Annotated[dict, "Information about the user's current playback state"]: - """Get information about the user's current playback state, including track or episode, progress, and active device.""" - endpoint = "/me/player" + """ + Get information about the user's current playback state, including track or episode, and active device. + This tool does not perform any actions. Use other tools to control playback. + """ + url = get_url("player_get_playback_state") + response = await send_spotify_request(context, "GET", url) + response.raise_for_status() + data = {"is_playing": False} if response.status_code == 204 else response.json() + return convert_to_playback_state(data).to_dict() - response = await _send_spotify_request(context, "GET", endpoint) - if response.status_code == 204: - return {"status": "Playback not available or active"} - elif response.status_code == 200: - data = response.json() - # TODO: Return a more structured model - result = { - "device_name": data.get("device", {}).get("name"), - "currently_playing_type": data.get("currently_playing_type"), - } - - if data.get("currently_playing_type") == "track": - item = data.get("item", {}) - album = item.get("album", {}) - result.update({ - "album_name": album.get("name"), - "album_artists": [artist.get("name") for artist in album.get("artists", [])], - "album_spotify_url": album.get("external_urls", {}).get("spotify"), - "track_name": item.get("name"), - "track_artists": [artist.get("name") for artist in item.get("artists", [])], - }) - elif data.get("currently_playing_type") == "episode": - item = data.get("item", {}) - show = item.get("show", {}) - result.update({ - "show_name": show.get("name"), - "show_spotify_url": show.get("external_urls", {}).get("spotify"), - "episode_name": item.get("name"), - "episode_spotify_url": item.get("external_urls", {}).get("spotify"), - }) - return result - else: - _handle_spotify_api_error(response) +@tool(requires_auth=Spotify(scopes=["user-read-currently-playing"])) +async def get_currently_playing( + context: ToolContext, +) -> Annotated[dict, "Information about the user's currently playing track"]: + """Get information about the user's currently playing track""" + url = get_url("player_get_currently_playing") + response = await send_spotify_request(context, "GET", url) + response.raise_for_status() + data = {"is_playing": False} if response.status_code == 204 else response.json() + return convert_to_playback_state(data).to_dict() diff --git a/toolkits/spotify/arcade_spotify/tools/tracks.py b/toolkits/spotify/arcade_spotify/tools/tracks.py new file mode 100644 index 00000000..6fd84056 --- /dev/null +++ b/toolkits/spotify/arcade_spotify/tools/tracks.py @@ -0,0 +1,136 @@ +from typing import Annotated, Optional + +from arcade.sdk import ToolContext, tool +from arcade.sdk.auth import Spotify +from arcade_spotify.tools.utils import ( + get_url, + send_spotify_request, +) + + +@tool(requires_auth=Spotify()) +async def get_track_from_id( + context: ToolContext, + track_id: Annotated[str, "The Spotify ID of the track"], +) -> Annotated[dict, "Information about the track"]: + """Get information about a track""" + url = get_url("tracks_get_track", track_id=track_id) + + response = await send_spotify_request(context, "GET", url) + response.raise_for_status() + return response.json() + + +@tool(requires_auth=Spotify()) +async def get_recommendations( + context: ToolContext, + seed_artists: Annotated[ + list[str], "A list of Spotify artist IDs to seed the recommendations with" + ], + seed_genres: Annotated[ + list[str], "A list of Spotify genre IDs to seed the recommendations with" + ], + seed_tracks: Annotated[ + list[str], "A list of Spotify track IDs to seed the recommendations with" + ], + limit: Annotated[int, "The maximum number of recommended tracks to return"] = 5, + target_acousticness: Annotated[ + Optional[float], + "The target acousticness of the recommended tracks (between 0 and 1)", + ] = None, + target_danceability: Annotated[ + Optional[float], + "The target danceability of the recommended tracks (between 0 and 1)", + ] = None, + target_duration_ms: Annotated[ + Optional[int], + "The target duration of the recommended tracks in milliseconds", + ] = None, + target_energy: Annotated[ + Optional[float], + "The target energy of the recommended tracks (between 0 and 1)", + ] = None, + target_instrumentalness: Annotated[ + Optional[float], + "The target instrumentalness of the recommended tracks (between 0 and 1)", + ] = None, + target_key: Annotated[ + Optional[int], + "The target key of the recommended tracks (0-11)", + ] = None, + target_liveness: Annotated[ + Optional[float], + "The target liveness of the recommended tracks (between 0 and 1)", + ] = None, + target_loudness: Annotated[ + Optional[float], + "The target loudness of the recommended tracks (in decibels)", + ] = None, + target_mode: Annotated[ + Optional[int], + "The target mode of the recommended tracks (0 or 1)", + ] = None, + target_popularity: Annotated[ + Optional[int], + "The target popularity of the recommended tracks (0-100)", + ] = None, + target_speechiness: Annotated[ + Optional[float], + "The target speechiness of the recommended tracks (between 0 and 1)", + ] = None, + target_tempo: Annotated[ + Optional[float], + "The target tempo of the recommended tracks (in beats per minute)", + ] = None, + target_time_signature: Annotated[ + Optional[int], + "The target time signature of the recommended tracks", + ] = None, + target_valence: Annotated[ + Optional[float], + "The target valence of the recommended tracks (between 0 and 1)", + ] = None, +) -> Annotated[dict, "A list of recommended tracks"]: + """Get track (song) recommendations based on seed artists, genres, and tracks + If a provided target value is outside of the expected range, it will clamp to the nearest valid value. + """ + url = get_url("tracks_get_recommendations") + params = { + "seed_artists": seed_artists, + "seed_genres": seed_genres, + "seed_tracks": seed_tracks, + "limit": limit, + "target_acousticness": target_acousticness, + "target_danceability": target_danceability, + "target_duration_ms": target_duration_ms, + "target_energy": target_energy, + "target_instrumentalness": target_instrumentalness, + "target_key": target_key, + "target_liveness": target_liveness, + "target_loudness": target_loudness, + "target_mode": target_mode, + "target_popularity": target_popularity, + "target_speechiness": target_speechiness, + "target_tempo": target_tempo, + "target_time_signature": target_time_signature, + "target_valence": target_valence, + } + params = {k: v for k, v in params.items() if v is not None} + + response = await send_spotify_request(context, "GET", url, params=params) + response.raise_for_status() + return response.json() + + +@tool(requires_auth=Spotify()) +async def get_tracks_audio_features( + context: ToolContext, + track_ids: Annotated[list[str], "A list of Spotify track (song) IDs"], +) -> Annotated[dict, "A list of audio features for the tracks"]: + """Get audio features for a list of tracks (songs)""" + url = get_url("tracks_get_audio_features") + params = {"ids": ",".join(track_ids)} + + response = await send_spotify_request(context, "GET", url, params=params) + response.raise_for_status() + return response.json() diff --git a/toolkits/spotify/arcade_spotify/tools/utils.py b/toolkits/spotify/arcade_spotify/tools/utils.py new file mode 100644 index 00000000..a0dd9906 --- /dev/null +++ b/toolkits/spotify/arcade_spotify/tools/utils.py @@ -0,0 +1,97 @@ +import httpx + +from arcade.core.schema import ToolContext +from arcade_spotify.tools.constants import ENDPOINTS, SPOTIFY_BASE_URL +from arcade_spotify.tools.models import PlaybackState + + +async def send_spotify_request( + context: ToolContext, + method: str, + url: str, + params: dict | None = None, + json_data: dict | None = None, +) -> httpx.Response: + """ + Send an asynchronous request to the Spotify API. + + Args: + context: The tool context containing the authorization token. + method: The HTTP method (GET, POST, PUT, DELETE, etc.). + url: The full URL for the API endpoint. + params: Query parameters to include in the request. + json_data: JSON data to include in the request body. + + Returns: + The response object from the API request. + + Raises: + ToolExecutionError: If the request fails for any reason. + """ + headers = {"Authorization": f"Bearer {context.authorization.token}"} + + async with httpx.AsyncClient() as client: + response = await client.request(method, url, headers=headers, params=params, json=json_data) + + return response + + +def handle_404_playback_state(response, message, is_playing: bool) -> dict | None: + if response.status_code == 404: + return convert_to_playback_state({ + "is_playing": is_playing, + "message": message, + }).to_dict() + + +def get_url(endpoint: str, **kwargs) -> str: + """ + Get the full Spotify URL for a given endpoint. + + :param endpoint: The endpoint key from ENDPOINTS + :param kwargs: The parameters to format the URL with + :return: The full URL + """ + return f"{SPOTIFY_BASE_URL}{ENDPOINTS[endpoint].format(**kwargs)}" + + +def convert_to_playback_state(data: dict) -> PlaybackState: + """ + Convert the Spotify API endpoint "/me/player" response data to a PlaybackState object. + + Args: + data: The response data from the Spotify API endpoint "/me/player". + + Returns: + An instance of PlaybackState populated with the data. + """ + playback_state = PlaybackState( + device_name=data.get("device", {}).get("name"), + device_id=data.get("device", {}).get("id"), + currently_playing_type=data.get("currently_playing_type"), + is_playing=data.get("is_playing"), + progress_ms=data.get("progress_ms"), + message=data.get("message"), + ) + + if data.get("currently_playing_type") == "track": + item = data.get("item", {}) + album = item.get("album", {}) + playback_state.album_name = album.get("name") + playback_state.album_id = album.get("id") + playback_state.album_artists = [artist.get("name") for artist in album.get("artists", [])] + playback_state.album_spotify_url = album.get("external_urls", {}).get("spotify") + playback_state.track_name = item.get("name") + playback_state.track_id = item.get("id") + playback_state.track_artists = [artist.get("name") for artist in item.get("artists", [])] + elif data.get("currently_playing_type") == "episode": + item = data.get("item", {}) + show = item.get("show", {}) + playback_state.show_name = show.get("name") + playback_state.show_id = show.get("id") + playback_state.show_spotify_url = show.get("external_urls", {}).get("spotify") + playback_state.episode_name = item.get("name") + playback_state.episode_id = item.get("id") + playback_state.episode_spotify_url = item.get("external_urls", {}).get("spotify") + + return playback_state diff --git a/toolkits/spotify/evals/eval_player.py b/toolkits/spotify/evals/eval_player.py new file mode 100644 index 00000000..8314d940 --- /dev/null +++ b/toolkits/spotify/evals/eval_player.py @@ -0,0 +1,135 @@ +from arcade_spotify.tools.player import ( + adjust_playback_position, + get_currently_playing, + get_playback_state, + pause_playback, + resume_playback, + skip_to_next_track, + skip_to_previous_track, + start_tracks_playback_by_id, +) + +from arcade.sdk import ToolCatalog +from arcade.sdk.eval import ( + EvalRubric, + EvalSuite, + tool_eval, +) +from arcade.sdk.eval.critic import NumericCritic + +# Evaluation rubric +rubric = EvalRubric( + fail_threshold=0.9, + warn_threshold=0.95, +) + +catalog = ToolCatalog() +catalog.add_tool(adjust_playback_position, "Spotify") +catalog.add_tool(skip_to_next_track, "Spotify") +catalog.add_tool(skip_to_previous_track, "Spotify") +catalog.add_tool(pause_playback, "Spotify") +catalog.add_tool(resume_playback, "Spotify") +catalog.add_tool(start_tracks_playback_by_id, "Spotify") +catalog.add_tool(get_playback_state, "Spotify") +catalog.add_tool(get_currently_playing, "Spotify") + + +@tool_eval() +def spotify_eval_suite() -> EvalSuite: + """Create an evaluation suite for Spotify "player" tools.""" + suite = EvalSuite( + name="Spotify Tools Evaluation", + system_message="You are an AI assistant that can manage Spotify using the provided tools.", + catalog=catalog, + rubric=rubric, + ) + + suite.add_case( + name="Adjust playback position", + user_message="can you skip to the 10th second of the song", + expected_tool_calls=[ + ( + adjust_playback_position, + { + "absolute_position_ms": 10000, + }, + ) + ], + critics=[ + NumericCritic( + critic_field="absolute_position_ms", weight=1.0, value_range=(9000, 11000) + ), + ], + ) + + suite.add_case( + name="Adjust playback position relative to current position", + user_message="go back 10 seconds", + expected_tool_calls=[ + ( + adjust_playback_position, + { + "relative_position_ms": -10000, + }, + ) + ], + critics=[ + NumericCritic( + critic_field="relative_position_ms", + weight=1.0, + value_range=(-11000, -9000), + ), + ], + ) + + suite.add_case( + name="Skip to previous track", + user_message="oops i didn't mean to skip that song, go back", + expected_tool_calls=[(skip_to_previous_track, {})], + ) + + suite.add_case( + name="Skip to next track", + user_message="skip this song and also the next one", + expected_tool_calls=[(skip_to_next_track, {}), (skip_to_next_track, {})], + ) + + suite.add_case( + name="Pause playback", + user_message="wait im getting a text, stop playing it please", + expected_tool_calls=[(pause_playback, {})], + ) + + suite.add_case( + name="Resume playback", + user_message="ok i'm back, you can press play again", + expected_tool_calls=[(resume_playback, {})], + ) + + suite.add_case( + name="Start playback of a list of tracks", + user_message="Play these two 03gaqN3aWm9TQxuHay0G8R, 03gaqN3aWm9TQxuHay0G8R. But start at the 10th second of the first track", + expected_tool_calls=[ + ( + start_tracks_playback_by_id, + { + "track_ids": ["03gaqN3aWm9TQxuHay0G8R", "03gaqN3aWm9TQxuHay0G8R"], + "position_ms": 10000, + }, + ) + ], + ) + + suite.add_case( + name="Get playback state", + user_message="what's the name of this song and who plays it?", + expected_tool_calls=[(get_currently_playing, {})], + ) + + suite.add_case( + name="Get playback state", + user_message="what device is playing music rn?", + expected_tool_calls=[(get_playback_state, {})], + ) + + return suite diff --git a/toolkits/spotify/evals/eval_tracks.py b/toolkits/spotify/evals/eval_tracks.py new file mode 100644 index 00000000..87e7558a --- /dev/null +++ b/toolkits/spotify/evals/eval_tracks.py @@ -0,0 +1,201 @@ +from arcade_spotify.tools.tracks import ( + get_recommendations, + get_track_from_id, + get_tracks_audio_features, +) + +from arcade.sdk import ToolCatalog +from arcade.sdk.eval import ( + BinaryCritic, + EvalRubric, + EvalSuite, + tool_eval, +) + +# Evaluation rubric +rubric = EvalRubric( + fail_threshold=0.9, + warn_threshold=0.95, +) + +catalog = ToolCatalog() +catalog.add_tool(get_track_from_id, "Spotify") +catalog.add_tool(get_recommendations, "Spotify") +catalog.add_tool(get_tracks_audio_features, "Spotify") + +# This history is a conversation where the user asks for the currently playing song, and then asks for information about it. +history = [ + {"role": "user", "content": "what song is playing rn"}, + { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": "call_fqZHT1DNcil1pbIv6UbOWVq9", + "type": "function", + "function": {"name": "Spotify_GetCurrentlyPlaying", "arguments": "{}"}, + } + ], + }, + { + "role": "tool", + "content": '{"album_artists":["Gajaka"],"album_id":"7b5h7MJ70KC92FrwSvRaQi","album_name":"Aerial","album_spotify_url":"https://open.spotify.com/album/7b5h7MJ70KC92FrwSvRaQi","currently_playing_type":"track","is_playing":true,"progress_ms":96962,"track_artists":["Gajaka"],"track_id":"03gaqN3aWm9TQxuHay0G8R","track_name":"Aerial"}', + "tool_call_id": "call_fqZHT1DNcil1pbIv6UbOWVq9", + "name": "Spotify_GetCurrentlyPlaying", + }, + { + "role": "assistant", + "content": 'The song currently playing is "Aerial" by Gajaka. You can listen to it on Spotify [here](https://open.spotify.com/album/7b5h7MJ70KC92FrwSvRaQi).', + }, +] + +# This history is a conversation where the user asks for the currently playing song, and then asks to skip to the next song, and then asks for information about the new song. +history_2 = [ + {"role": "user", "content": "what song is playing right now?"}, + { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": "call_u2nS7hd7ETdbjElHW4mrFphd", + "type": "function", + "function": {"name": "Spotify_GetCurrentlyPlaying", "arguments": "{}"}, + } + ], + }, + { + "role": "tool", + "content": '{"album_artists":["The Dawning"],"album_id":"2dmrP48hhdCIoxeYJFMSVO","album_name":"summer breeze","album_spotify_url":"https://open.spotify.com/album/2dmrP48hhdCIoxeYJFMSVO","currently_playing_type":"track","is_playing":true,"progress_ms":144092,"track_artists":["The Dawning"],"track_id":"14CQdqVFygEls6szbfwuTL","track_name":"summer breeze"}', + "tool_call_id": "call_u2nS7hd7ETdbjElHW4mrFphd", + "name": "Spotify_GetCurrentlyPlaying", + }, + { + "role": "assistant", + "content": 'The song "summer breeze" by The Dawning is currently playing. You can check it out on [Spotify](https://open.spotify.com/album/2dmrP48hhdCIoxeYJFMSVO).', + }, + {"role": "user", "content": "please skip this song"}, + { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": "call_htECRssaXZ4jtNvjdUY2r3ml", + "type": "function", + "function": {"name": "Spotify_SkipToNextTrack", "arguments": "{}"}, + } + ], + }, + { + "role": "tool", + "content": '{"album_artists":["Philó"],"album_id":"525aF5PKAQXzbmP7oAb0gC","album_name":"Skylar","album_spotify_url":"https://open.spotify.com/album/525aF5PKAQXzbmP7oAb0gC","currently_playing_type":"track","device_id":"84b6aa89387cfe3e132465001647027bfd1b44f3","device_name":"iPhone","is_playing":true,"progress_ms":233,"track_artists":["Philó"],"track_id":"5r9fTbn6YZVxHKaU4Rze3t","track_name":"Tempestade"}', + "tool_call_id": "call_htECRssaXZ4jtNvjdUY2r3ml", + "name": "Spotify_SkipToNextTrack", + }, + { + "role": "assistant", + "content": 'I\'ve skipped to the next song. Now playing: "Tempestade" by Philó. You can listen to it on [Spotify](https://open.spotify.com/album/525aF5PKAQXzbmP7oAb0gC).', + }, + {"role": "user", "content": "now what song is playing?"}, + {"role": "assistant", "content": 'The song now playing is "Tempestade" by Philó.'}, +] + +# This history is a conversation where the user asks for what is currently playing, and then asks for more information about the track, then asks for the audio features for the track. +history_3 = [ + {"role": "user", "content": "get info about 14CQdqVFygEls6szbfwuTL"}, + { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": "call_bFqQIQY89z82BXJqSXhlkiro", + "type": "function", + "function": { + "name": "Spotify_GetTrackFromId", + "arguments": '{"track_id":"14CQdqVFygEls6szbfwuTL"}', + }, + } + ], + }, + { + "role": "tool", + "content": '{"album":{"album_type":"single","artists":[{"external_urls":{"spotify":"https://open.spotify.com/artist/4ev14fn325vhWuykve3QtA"},"href":"https://api.spotify.com/v1/artists/4ev14fn325vhWuykve3QtA","id":"4ev14fn325vhWuykve3QtA","name":"The Dawning","type":"artist","uri":"spotify:artist:4ev14fn325vhWuykve3QtA"}],"available_markets":["AR","AU","AT","BE","BO","BR","BG","CA","CL","CO","CR","CY","CZ","DK","DO","DE","EC","EE","SV","FI","FR","GR","GT","HN","HK","HU","IS","IE","IT","LV","LT","LU","MY","MT","MX","NL","NZ","NI","NO","PA","PY","PE","PH","PL","PT","SG","SK","ES","SE","CH","TW","TR","UY","US","GB","AD","LI","MC","ID","JP","TH","VN","RO","IL","ZA","SA","AE","BH","QA","OM","KW","EG","MA","DZ","TN","LB","JO","PS","IN","BY","KZ","MD","UA","AL","BA","HR","ME","MK","RS","SI","KR","BD","PK","LK","GH","KE","NG","TZ","UG","AG","AM","BS","BB","BZ","BT","BW","BF","CV","CW","DM","FJ","GM","GE","GD","GW","GY","HT","JM","KI","LS","LR","MW","MV","ML","MH","FM","NA","NR","NE","PW","PG","PR","WS","SM","ST","SN","SC","SL","SB","KN","LC","VC","SR","TL","TO","TT","TV","VU","AZ","BN","BI","KH","CM","TD","KM","GQ","SZ","GA","GN","KG","LA","MO","MR","MN","NP","RW","TG","UZ","ZW","BJ","MG","MU","MZ","AO","CI","DJ","ZM","CD","CG","IQ","LY","TJ","VE","ET","XK"],"external_urls":{"spotify":"https://open.spotify.com/album/2dmrP48hhdCIoxeYJFMSVO"},"href":"https://api.spotify.com/v1/albums/2dmrP48hhdCIoxeYJFMSVO","id":"2dmrP48hhdCIoxeYJFMSVO","images":[{"height":640,"url":"https://i.scdn.co/image/ab67616d0000b2730046525154a4201332796065","width":640},{"height":300,"url":"https://i.scdn.co/image/ab67616d00001e020046525154a4201332796065","width":300},{"height":64,"url":"https://i.scdn.co/image/ab67616d000048510046525154a4201332796065","width":64}],"name":"summer breeze","release_date":"2024-08-02","release_date_precision":"day","total_tracks":1,"type":"album","uri":"spotify:album:2dmrP48hhdCIoxeYJFMSVO"},"artists":[{"external_urls":{"spotify":"https://open.spotify.com/artist/4ev14fn325vhWuykve3QtA"},"href":"https://api.spotify.com/v1/artists/4ev14fn325vhWuykve3QtA","id":"4ev14fn325vhWuykve3QtA","name":"The Dawning","type":"artist","uri":"spotify:artist:4ev14fn325vhWuykve3QtA"}],"available_markets":["AR","AU","AT","BE","BO","BR","BG","CA","CL","CO","CR","CY","CZ","DK","DO","DE","EC","EE","SV","FI","FR","GR","GT","HN","HK","HU","IS","IE","IT","LV","LT","LU","MY","MT","MX","NL","NZ","NI","NO","PA","PY","PE","PH","PL","PT","SG","SK","ES","SE","CH","TW","TR","UY","US","GB","AD","LI","MC","ID","JP","TH","VN","RO","IL","ZA","SA","AE","BH","QA","OM","KW","EG","MA","DZ","TN","LB","JO","PS","IN","BY","KZ","MD","UA","AL","BA","HR","ME","MK","RS","SI","KR","BD","PK","LK","GH","KE","NG","TZ","UG","AG","AM","BS","BB","BZ","BT","BW","BF","CV","CW","DM","FJ","GM","GE","GD","GW","GY","HT","JM","KI","LS","LR","MW","MV","ML","MH","FM","NA","NR","NE","PW","PG","PR","WS","SM","ST","SN","SC","SL","SB","KN","LC","VC","SR","TL","TO","TT","TV","VU","AZ","BN","BI","KH","CM","TD","KM","GQ","SZ","GA","GN","KG","LA","MO","MR","MN","NP","RW","TG","UZ","ZW","BJ","MG","MU","MZ","AO","CI","DJ","ZM","CD","CG","IQ","LY","TJ","VE","ET","XK"],"disc_number":1,"duration_ms":165811,"explicit":false,"external_ids":{"isrc":"SE69N2406255"},"external_urls":{"spotify":"https://open.spotify.com/track/14CQdqVFygEls6szbfwuTL"},"href":"https://api.spotify.com/v1/tracks/14CQdqVFygEls6szbfwuTL","id":"14CQdqVFygEls6szbfwuTL","is_local":false,"name":"summer breeze","popularity":60,"preview_url":"https://p.scdn.co/mp3-preview/9b2ac80563d092f9b1d442f09635ccf646ccb3e0?cid=26913f34d26f4c16a15d5a93e309a1dc","track_number":1,"type":"track","uri":"spotify:track:14CQdqVFygEls6szbfwuTL"}', + "tool_call_id": "call_bFqQIQY89z82BXJqSXhlkiro", + "name": "Spotify_GetTrackFromId", + }, + { + "role": "assistant", + "content": 'The track "Summer Breeze" by The Dawning is available on Spotify. Here are some details about the track:\n\n- **Artist:** [The Dawning](https://open.spotify.com/artist/4ev14fn325vhWuykve3QtA)\n- **Album:** [Summer Breeze](https://open.spotify.com/album/2dmrP48hhdCIoxeYJFMSVO) (Single)\n- **Release Date:** August 2, 2024\n- **Duration:** 2 minutes and 45 seconds\n- **Popularity:** 60\n- **ISRC:** SE69N2406255\n- **Explicit Content:** No\n\nThe track is available in numerous markets, including the US, UK, Canada, Australia, and many more. You can listen to a preview of the track [here](https://p.scdn.co/mp3-preview/9b2ac80563d092f9b1d442f09635ccf646ccb3e0?cid=26913f34d26f4c16a15d5a93e309a1dc).', + }, +] + + +@tool_eval() +def spotify_eval_suite() -> EvalSuite: + """Create an evaluation suite for Spotify "track" tools.""" + suite = EvalSuite( + name="Spotify Tools Evaluation", + system_message="You are an AI assistant that can manage Spotify using the provided tools.", + catalog=catalog, + rubric=rubric, + ) + + suite.add_case( + name="Get information about a track", + user_message="can you get more info about that song", + expected_tool_calls=[ + ( + get_track_from_id, + { + "track_id": "03gaqN3aWm9TQxuHay0G8R", + }, + ) + ], + critics=[ + BinaryCritic(critic_field="track_id", weight=1.0), + ], + additional_messages=history, + ) + + suite.add_case( + name="Get audio features for a track", + user_message="what bpm is that song?", + expected_tool_calls=[ + ( + get_tracks_audio_features, + { + "track_ids": [ + "14CQdqVFygEls6szbfwuTL", + "5r9fTbn6YZVxHKaU4Rze3t", + ] + }, + ) + ], + critics=[ + BinaryCritic(critic_field="track_id", weight=1.0), + ], + additional_messages=history_2, + ) + + suite.add_case( + name="Get recommendations based on a specific track", + user_message="Give me 6 recommendations based on this song.", + expected_tool_calls=[ + ( + get_recommendations, + { + "seed_artists": ["4ev14fn325vhWuykve3QtA"], + "seed_genres": [], # None provided in the history + "seed_tracks": ["14CQdqVFygEls6szbfwuTL"], + "limit": 6, + }, + ) + ], + critics=[ + BinaryCritic(critic_field="seed_artists", weight=0.3), + BinaryCritic(critic_field="seed_genres", weight=0.3), + BinaryCritic(critic_field="seed_tracks", weight=0.3), + ], + additional_messages=history_3, + ) + + return suite