From 55ec34999f32e167836678807fa08441d0ebb774 Mon Sep 17 00:00:00 2001 From: Lukas Date: Mon, 11 May 2026 16:36:17 -0400 Subject: [PATCH 1/2] Add music recommender domain demo --- README.md | 34 +++++- arch__MusicRecommender.py | 17 +++ music_domain.py | 244 +++++++++++++++++++++++++++++++++++++ music_recommender.py | 167 +++++++++++++++++++++++++ tests/test_music_domain.py | 51 ++++++++ 5 files changed, 510 insertions(+), 3 deletions(-) create mode 100644 arch__MusicRecommender.py create mode 100644 music_domain.py create mode 100644 music_recommender.py create mode 100644 tests/test_music_domain.py diff --git a/README.md b/README.md index 26cb9cc..407e909 100644 --- a/README.md +++ b/README.md @@ -23,14 +23,43 @@ If you plan to run the app in a conda or virtual environment, make sure to set u pip install git+https://github.com/aolabsai/ao_arch git+https://github.com/aolabsai/ao_core ``` -3. Run the application with the following command: +3. Run the YouTube recommender application with the following command: ```bash - streamlit run recommender.py + streamlit run main.py ``` 4. Once running, the app will be accessible at `localhost:8501`. +### Music Domain Demo + +This fork adds a second recommender domain that applies the same AO Labs +real-time training loop to music discovery. Instead of fetching YouTube videos, +the demo recommends tracks from a local sample catalog using binary features for +genre, tempo, energy, release recency, vocal/instrumental format, and listening +context. + +Run the music recommender with: + +```bash +streamlit run music_recommender.py +``` + +The music demo can run in two modes: + +- With `ao_core` and `ao_arch` installed, it creates a live AO Agent from + `arch__MusicRecommender.py` and trains it from the "Recommend more" and + "Stop recommending" feedback buttons. +- Without those AO packages installed, it uses a deterministic local fallback so + the domain mapping, interface, and tests can still be exercised. + +The music-domain mapping lives in `music_domain.py`. Its unit tests can be run +with: + +```bash +python -m unittest discover -s tests +``` + ### Docker Installation @@ -61,4 +90,3 @@ The recommender system works by loading a set of random video links. Once the us Fork the repository, make your changes, and submit a pull request for review. - diff --git a/arch__MusicRecommender.py b/arch__MusicRecommender.py new file mode 100644 index 0000000..d393718 --- /dev/null +++ b/arch__MusicRecommender.py @@ -0,0 +1,17 @@ +# -*- coding: utf-8 -*- +""" +AO architecture for the music-domain recommender demo. +""" + +import ao_arch as ar + + +description = "Music Recommender System" + +# genre + tempo + energy + recency + vocal/instrumental + listening context +arch_i = [3, 2, 2, 1, 1, 2] +arch_z = [10] +arch_c = [] +connector_function = "full_conn" + +arch = ar.Arch(arch_i, arch_z, arch_c, connector_function, description) diff --git a/music_domain.py b/music_domain.py new file mode 100644 index 0000000..9188935 --- /dev/null +++ b/music_domain.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Iterable, List, Sequence, Tuple + + +GENRE_BUCKETS = { + "pop": 0, + "hip-hop": 1, + "electronic": 2, + "rock": 3, + "jazz": 4, + "r&b": 5, + "folk": 6, + "classical": 7, +} + +LISTENING_CONTEXTS = { + "surprise me": [0, 0], + "focus": [0, 1], + "workout": [1, 0], + "wind down": [1, 1], +} + + +@dataclass(frozen=True) +class TrackCandidate: + title: str + artist: str + genres: Tuple[str, ...] + tempo_bpm: int + energy: int + release_year: int + vocal: bool + summary: str + + +MUSIC_CATALOG: Tuple[TrackCandidate, ...] = ( + TrackCandidate( + "City Lights After Midnight", + "North Pier", + ("electronic", "pop"), + 118, + 8, + 2023, + False, + "Bright synth pulses for late-night momentum without lyrics.", + ), + TrackCandidate( + "Paper Planets", + "Mara Vale", + ("folk", "pop"), + 92, + 4, + 2021, + True, + "Warm acoustic storytelling with soft percussion and close vocals.", + ), + TrackCandidate( + "Baseline Sketches", + "The Static Notes", + ("jazz",), + 136, + 6, + 2018, + False, + "A nimble trio piece built around upright bass and brushed drums.", + ), + TrackCandidate( + "Highway Glass", + "Signal Choir", + ("rock",), + 154, + 9, + 2020, + True, + "Guitar-driven road-song energy with a big final chorus.", + ), + TrackCandidate( + "Soft Reset", + "Luma Drift", + ("r&b", "electronic"), + 86, + 3, + 2024, + True, + "Low-tempo vocals and ambient pads for decompressing.", + ), + TrackCandidate( + "Assembly Line Sunrise", + "Metro Form", + ("hip-hop",), + 98, + 7, + 2022, + True, + "Percussive verses with a steady bounce and clean hook.", + ), + TrackCandidate( + "Window Seat Etude", + "Iris Calder", + ("classical",), + 72, + 2, + 2017, + False, + "Solo piano with a restrained melodic arc for quiet focus.", + ), + TrackCandidate( + "Neon Shortcut", + "Byte Harbor", + ("electronic", "hip-hop"), + 128, + 8, + 2025, + False, + "A short instrumental loop with punchy drums and arcade textures.", + ), + TrackCandidate( + "Hollow Gold", + "June Relay", + ("pop", "r&b"), + 104, + 5, + 2019, + True, + "Mid-tempo vocal pop with polished harmonies and a mellow groove.", + ), + TrackCandidate( + "Field Recording No. 3", + "Orchard Static", + ("folk", "classical"), + 78, + 2, + 2020, + False, + "A sparse chamber-folk instrumental with natural room tone.", + ), + TrackCandidate( + "Switchback", + "Redline Theory", + ("rock", "electronic"), + 142, + 9, + 2024, + True, + "Aggressive drums, distorted bass, and a fast vocal hook.", + ), + TrackCandidate( + "Blue Hour Receipt", + "Kei Santos", + ("jazz", "r&b"), + 88, + 4, + 2022, + True, + "Smoky keys and relaxed vocals for a late-evening queue.", + ), +) + + +def to_bits(value: int, width: int) -> List[int]: + if value < 0: + raise ValueError("value must be non-negative") + if value >= 2**width: + raise ValueError("value does not fit in requested bit width") + return [int(bit) for bit in format(value, f"0{width}b")] + + +def normalize_context(context: str) -> str: + normalized = context.strip().lower() + if normalized not in LISTENING_CONTEXTS: + return "surprise me" + return normalized + + +def primary_genre(track: TrackCandidate) -> str: + for genre in track.genres: + if genre in GENRE_BUCKETS: + return genre + return "pop" + + +def tempo_bucket(tempo_bpm: int) -> int: + if tempo_bpm < 90: + return 0 + if tempo_bpm < 115: + return 1 + if tempo_bpm < 140: + return 2 + return 3 + + +def energy_bucket(energy: int) -> int: + if energy < 4: + return 0 + if energy < 7: + return 1 + if energy < 9: + return 2 + return 3 + + +def recency_bucket(release_year: int) -> int: + return 1 if release_year >= 2021 else 0 + + +def vocal_bucket(vocal: bool) -> int: + return 1 if vocal else 0 + + +def encode_track(track: TrackCandidate, context: str) -> List[int]: + return ( + to_bits(GENRE_BUCKETS[primary_genre(track)], 3) + + to_bits(tempo_bucket(track.tempo_bpm), 2) + + to_bits(energy_bucket(track.energy), 2) + + [recency_bucket(track.release_year)] + + [vocal_bucket(track.vocal)] + + LISTENING_CONTEXTS[normalize_context(context)] + ) + + +def recommendation_percentage(agent_output: Sequence[int]) -> int: + if not agent_output: + return 0 + positive = sum(1 for value in agent_output if value == 1) + return round((positive / len(agent_output)) * 100) + + +def pick_next_track(seen_titles: Iterable[str], index_seed: int = 0) -> TrackCandidate: + seen = set(seen_titles) + unseen = [track for track in MUSIC_CATALOG if track.title not in seen] + candidates = unseen or list(MUSIC_CATALOG) + return candidates[index_seed % len(candidates)] + + +def format_track_card(track: TrackCandidate) -> str: + genres = ", ".join(track.genres) + vocal_text = "vocal" if track.vocal else "instrumental" + return ( + f"**{track.title}** by {track.artist} ({track.release_year})\n\n" + f"{genres} | {track.tempo_bpm} BPM | energy {track.energy}/10 | {vocal_text}\n\n" + f"{track.summary}" + ) diff --git a/music_recommender.py b/music_recommender.py new file mode 100644 index 0000000..9be1add --- /dev/null +++ b/music_recommender.py @@ -0,0 +1,167 @@ +from __future__ import annotations + +import numpy as np +import pandas as pd +import streamlit as st + +from music_domain import ( + MUSIC_CATALOG, + encode_track, + format_track_card, + pick_next_track, + recommendation_percentage, +) + +try: + import ao_core as ao + from arch__MusicRecommender import arch +except ImportError: + ao = None + arch = None + + +CONTEXTS = ("Surprise me", "Focus", "Workout", "Wind down") + + +def ensure_state() -> None: + st.session_state.setdefault("seen_music_titles", []) + st.session_state.setdefault("music_index_seed", 0) + st.session_state.setdefault("music_feedback_history", []) + st.session_state.setdefault("current_track_bits", []) + st.session_state.setdefault("current_track", pick_next_track([], 0)) + + if "music_agent" not in st.session_state and ao is not None: + st.session_state.music_agent = ao.Agent(arch, notes="Music Domain Agent") + for _ in range(4): + st.session_state.music_agent.reset_state() + st.session_state.music_agent.reset_state(training=True) + + +def fallback_response(binary_input: list[int]) -> list[int]: + score = sum(binary_input) + return [1 if (score + idx) % 4 in (1, 2, 3) else 0 for idx in range(10)] + + +def agent_response(binary_input: list[int]) -> list[int]: + if ao is None: + return fallback_response(binary_input) + + st.session_state.music_agent.reset_state() + response = None + for _ in range(5): + response = st.session_state.music_agent.next_state( + INPUT=binary_input, + print_result=False, + ) + return response + + +def train_agent(user_response: str) -> None: + if ao is None: + return + + label_value = 1 if user_response == "Recommend more" else 0 + label = np.full( + st.session_state.music_agent.arch.Z__flat.shape, + label_value, + dtype=np.int8, + ) + repetitions = 6 if label_value else 10 + + for _ in range(repetitions): + st.session_state.music_agent.reset_state() + st.session_state.music_agent.next_state( + INPUT=st.session_state.current_track_bits, + LABEL=label, + print_result=False, + unsequenced=True, + ) + + +def advance_track() -> None: + st.session_state.seen_music_titles.append(st.session_state.current_track.title) + st.session_state.music_index_seed += 1 + st.session_state.current_track = pick_next_track( + st.session_state.seen_music_titles, + st.session_state.music_index_seed, + ) + + +st.set_page_config( + page_title="Music Recommender Demo by AO Labs", + page_icon="misc/ao_favicon.png", + layout="wide", + initial_sidebar_state="expanded", +) + +ensure_state() + +with st.sidebar: + st.write("## Music Dataset") + st.write(f"{len(MUSIC_CATALOG)} sample tracks") + if ao is None: + st.warning( + "AO packages are not installed, so the app is using a deterministic local response. " + "Install ao_core and ao_arch to train a live AO Agent." + ) + elif st.button("Reset Music Agent"): + del st.session_state.music_agent + ensure_state() + st.success("Music Agent reset") + +st.title("Real-Time Personal Music Recommender") +st.write("A music-domain demo for AO Labs' continuously trainable recommender.") + +context = st.selectbox("Set your current listening context:", CONTEXTS) +track = st.session_state.current_track +binary_input = encode_track(track, context) +st.session_state.current_track_bits = binary_input + +response = agent_response(binary_input) +recommendation = recommendation_percentage(response) + +left, right = st.columns([0.6, 0.4], gap="large") + +with left: + st.markdown(format_track_card(track)) + st.write(f"Agent recommendation: {recommendation}%") + st.progress(recommendation) + st.caption(f"Agent input bits: {binary_input}") + +with right: + if st.button("Recommend more", type="primary"): + train_agent("Recommend more") + st.session_state.music_feedback_history.append( + [track.title, track.artist, context, recommendation, "Recommend more"] + ) + advance_track() + st.rerun() + + if st.button("Stop recommending"): + train_agent("Stop recommending") + st.session_state.music_feedback_history.append( + [track.title, track.artist, context, recommendation, "Stop recommending"] + ) + advance_track() + st.rerun() + + if st.button("Skip"): + st.session_state.music_feedback_history.append( + [track.title, track.artist, context, recommendation, "Skipped"] + ) + advance_track() + st.rerun() + +st.write("---") +st.write("### Training History") + +if st.session_state.music_feedback_history: + st.dataframe( + pd.DataFrame( + st.session_state.music_feedback_history, + columns=["Title", "Artist", "Context", "Recommendation", "Feedback"], + ), + use_container_width=True, + ) +else: + st.info("Rate a track to start training this music-domain agent.") diff --git a/tests/test_music_domain.py b/tests/test_music_domain.py new file mode 100644 index 0000000..1348c2a --- /dev/null +++ b/tests/test_music_domain.py @@ -0,0 +1,51 @@ +import unittest + +from music_domain import ( + MUSIC_CATALOG, + encode_track, + format_track_card, + pick_next_track, + recommendation_percentage, +) + + +class MusicDomainTests(unittest.TestCase): + def test_catalog_titles_are_unique(self): + titles = [track.title for track in MUSIC_CATALOG] + self.assertEqual(len(titles), len(set(titles))) + + def test_encode_track_returns_eleven_binary_features(self): + bits = encode_track(MUSIC_CATALOG[0], "Focus") + self.assertEqual(len(bits), 11) + self.assertTrue(all(bit in (0, 1) for bit in bits)) + + def test_encode_track_is_deterministic(self): + track = MUSIC_CATALOG[3] + self.assertEqual( + encode_track(track, "Workout"), + encode_track(track, "Workout"), + ) + + def test_unknown_context_uses_default_bucket(self): + track = MUSIC_CATALOG[1] + self.assertEqual(encode_track(track, "not a context")[-2:], [0, 0]) + + def test_recommendation_percentage(self): + self.assertEqual(recommendation_percentage([1, 1, 0, 0]), 50) + self.assertEqual(recommendation_percentage([]), 0) + + def test_pick_next_track_skips_seen_titles(self): + first = MUSIC_CATALOG[0] + picked = pick_next_track([first.title], 0) + self.assertNotEqual(picked.title, first.title) + + def test_format_track_card_contains_key_fields(self): + track = MUSIC_CATALOG[0] + card = format_track_card(track) + self.assertIn(track.title, card) + self.assertIn(track.artist, card) + self.assertIn(str(track.tempo_bpm), card) + + +if __name__ == "__main__": + unittest.main() From 439bd885dca64924635b32fcc677526eda096a64 Mon Sep 17 00:00:00 2001 From: Lukas Date: Tue, 12 May 2026 08:39:57 -0400 Subject: [PATCH 2/2] Allow custom music catalog uploads --- README.md | 10 +++- music_domain.py | 97 ++++++++++++++++++++++++++++++++++++-- music_recommender.py | 44 ++++++++++++++++- tests/test_music_domain.py | 23 +++++++++ 4 files changed, 168 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 407e909..fe054a5 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,15 @@ with: python -m unittest discover -s tests ``` +The sidebar can also load a custom CSV catalog, which makes the demo reusable +for a real music dataset instead of only the bundled sample tracks. The CSV +must include: + +```csv +title,artist,genres,tempo_bpm,energy,release_year,vocal,summary +City Lights,Example Artist,electronic;pop,118,8,2024,no,Bright synth pulse +``` + ### Docker Installation @@ -89,4 +98,3 @@ The recommender system works by loading a set of random video links. Once the us Fork the repository, make your changes, and submit a pull request for review. - diff --git a/music_domain.py b/music_domain.py index 9188935..4dd9a4a 100644 --- a/music_domain.py +++ b/music_domain.py @@ -1,6 +1,8 @@ from __future__ import annotations +import csv from dataclasses import dataclass +from io import StringIO from typing import Iterable, List, Sequence, Tuple @@ -35,6 +37,18 @@ class TrackCandidate: summary: str +CSV_REQUIRED_COLUMNS = ( + "title", + "artist", + "genres", + "tempo_bpm", + "energy", + "release_year", + "vocal", + "summary", +) + + MUSIC_CATALOG: Tuple[TrackCandidate, ...] = ( TrackCandidate( "City Lights After Midnight", @@ -159,6 +173,77 @@ class TrackCandidate: ) +def _parse_bool(value: str) -> bool: + normalized = value.strip().lower() + if normalized in {"1", "true", "yes", "y", "vocal"}: + return True + if normalized in {"0", "false", "no", "n", "instrumental"}: + return False + raise ValueError(f"invalid vocal value: {value!r}") + + +def _parse_genres(value: str) -> Tuple[str, ...]: + delimiter = ";" if ";" in value else "|" if "|" in value else "," + genres = tuple( + genre.strip().lower() + for genre in value.split(delimiter) + if genre.strip() + ) + if not genres: + raise ValueError("genres must include at least one genre") + return genres + + +def load_catalog_from_csv(csv_text: str) -> Tuple[TrackCandidate, ...]: + reader = csv.DictReader(StringIO(csv_text.strip())) + if not reader.fieldnames: + raise ValueError("CSV must include a header row") + + normalized_fields = {field.strip().lower() for field in reader.fieldnames} + missing = [column for column in CSV_REQUIRED_COLUMNS if column not in normalized_fields] + if missing: + raise ValueError(f"CSV is missing required columns: {', '.join(missing)}") + + tracks: list[TrackCandidate] = [] + for row_number, row in enumerate(reader, start=2): + normalized_row = { + (key or "").strip().lower(): (value or "").strip() + for key, value in row.items() + } + try: + title = normalized_row["title"] + artist = normalized_row["artist"] + if not title or not artist: + raise ValueError("title and artist are required") + + tempo_bpm = int(normalized_row["tempo_bpm"]) + energy = int(normalized_row["energy"]) + release_year = int(normalized_row["release_year"]) + if tempo_bpm <= 0: + raise ValueError("tempo_bpm must be positive") + if energy < 0 or energy > 10: + raise ValueError("energy must be between 0 and 10") + + tracks.append( + TrackCandidate( + title=title, + artist=artist, + genres=_parse_genres(normalized_row["genres"]), + tempo_bpm=tempo_bpm, + energy=energy, + release_year=release_year, + vocal=_parse_bool(normalized_row["vocal"]), + summary=normalized_row["summary"], + ) + ) + except ValueError as exc: + raise ValueError(f"row {row_number}: {exc}") from exc + + if not tracks: + raise ValueError("CSV must include at least one track") + return tuple(tracks) + + def to_bits(value: int, width: int) -> List[int]: if value < 0: raise ValueError("value must be non-negative") @@ -227,10 +312,16 @@ def recommendation_percentage(agent_output: Sequence[int]) -> int: return round((positive / len(agent_output)) * 100) -def pick_next_track(seen_titles: Iterable[str], index_seed: int = 0) -> TrackCandidate: +def pick_next_track( + seen_titles: Iterable[str], + index_seed: int = 0, + catalog: Sequence[TrackCandidate] = MUSIC_CATALOG, +) -> TrackCandidate: + if not catalog: + raise ValueError("catalog must include at least one track") seen = set(seen_titles) - unseen = [track for track in MUSIC_CATALOG if track.title not in seen] - candidates = unseen or list(MUSIC_CATALOG) + unseen = [track for track in catalog if track.title not in seen] + candidates = unseen or list(catalog) return candidates[index_seed % len(candidates)] diff --git a/music_recommender.py b/music_recommender.py index 9be1add..cadc363 100644 --- a/music_recommender.py +++ b/music_recommender.py @@ -1,5 +1,7 @@ from __future__ import annotations +import hashlib + import numpy as np import pandas as pd import streamlit as st @@ -8,6 +10,7 @@ MUSIC_CATALOG, encode_track, format_track_card, + load_catalog_from_csv, pick_next_track, recommendation_percentage, ) @@ -24,11 +27,15 @@ def ensure_state() -> None: + st.session_state.setdefault("music_catalog", MUSIC_CATALOG) st.session_state.setdefault("seen_music_titles", []) st.session_state.setdefault("music_index_seed", 0) st.session_state.setdefault("music_feedback_history", []) st.session_state.setdefault("current_track_bits", []) - st.session_state.setdefault("current_track", pick_next_track([], 0)) + st.session_state.setdefault( + "current_track", + pick_next_track([], 0, st.session_state.music_catalog), + ) if "music_agent" not in st.session_state and ao is not None: st.session_state.music_agent = ao.Agent(arch, notes="Music Domain Agent") @@ -84,6 +91,7 @@ def advance_track() -> None: st.session_state.current_track = pick_next_track( st.session_state.seen_music_titles, st.session_state.music_index_seed, + st.session_state.music_catalog, ) @@ -98,7 +106,39 @@ def advance_track() -> None: with st.sidebar: st.write("## Music Dataset") - st.write(f"{len(MUSIC_CATALOG)} sample tracks") + st.write(f"{len(st.session_state.music_catalog)} tracks loaded") + + uploaded_catalog = st.file_uploader( + "Use a custom music CSV", + type=["csv"], + help=( + "Required columns: title, artist, genres, tempo_bpm, energy, " + "release_year, vocal, summary. Separate multiple genres with semicolons." + ), + ) + if uploaded_catalog is not None: + uploaded_bytes = uploaded_catalog.getvalue() + catalog_hash = hashlib.sha256(uploaded_bytes).hexdigest() + if catalog_hash != st.session_state.get("music_catalog_hash"): + try: + st.session_state.music_catalog = load_catalog_from_csv( + uploaded_bytes.decode("utf-8-sig") + ) + st.session_state.music_catalog_hash = catalog_hash + st.session_state.seen_music_titles = [] + st.session_state.music_index_seed = 0 + st.session_state.music_feedback_history = [] + st.session_state.current_track = pick_next_track( + [], + 0, + st.session_state.music_catalog, + ) + st.success(f"Loaded {len(st.session_state.music_catalog)} tracks") + except UnicodeDecodeError: + st.error("CSV must be UTF-8 encoded.") + except ValueError as exc: + st.error(str(exc)) + if ao is None: st.warning( "AO packages are not installed, so the app is using a deterministic local response. " diff --git a/tests/test_music_domain.py b/tests/test_music_domain.py index 1348c2a..328f644 100644 --- a/tests/test_music_domain.py +++ b/tests/test_music_domain.py @@ -4,6 +4,7 @@ MUSIC_CATALOG, encode_track, format_track_card, + load_catalog_from_csv, pick_next_track, recommendation_percentage, ) @@ -39,6 +40,28 @@ def test_pick_next_track_skips_seen_titles(self): picked = pick_next_track([first.title], 0) self.assertNotEqual(picked.title, first.title) + def test_pick_next_track_accepts_custom_catalog(self): + csv_text = """title,artist,genres,tempo_bpm,energy,release_year,vocal,summary +Test Track,Example Artist,pop;rock,101,5,2026,yes,Demo summary +Second Track,Example Artist,jazz,88,3,2025,no,Another summary +""" + catalog = load_catalog_from_csv(csv_text) + picked = pick_next_track([catalog[0].title], 0, catalog) + self.assertEqual(picked.title, catalog[1].title) + + def test_load_catalog_from_csv_validates_required_columns(self): + with self.assertRaises(ValueError): + load_catalog_from_csv("title,artist\nMissing,Columns\n") + + def test_load_catalog_from_csv_parses_binary_features(self): + csv_text = """title,artist,genres,tempo_bpm,energy,release_year,vocal,summary +CSV Track,CSV Artist,electronic;pop,128,8,2024,instrumental,Upload-ready record +""" + catalog = load_catalog_from_csv(csv_text) + self.assertEqual(catalog[0].genres, ("electronic", "pop")) + self.assertFalse(catalog[0].vocal) + self.assertEqual(len(encode_track(catalog[0], "Focus")), 11) + def test_format_track_card_contains_key_fields(self): track = MUSIC_CATALOG[0] card = format_track_card(track)