From e8d20158e3c8dee654097f68cb34dbe6a9043d49 Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Mon, 10 Aug 2026 13:17:33 +0200 Subject: [PATCH] refactor: rebuild orchestration around a track selection --- backend/app/pipeline/orchestrator.py | 188 +++++++++++++++------------ docs/logboek.md | 32 +++-- 2 files changed, 124 insertions(+), 96 deletions(-) diff --git a/backend/app/pipeline/orchestrator.py b/backend/app/pipeline/orchestrator.py index bfc1cba..f1f07cb 100644 --- a/backend/app/pipeline/orchestrator.py +++ b/backend/app/pipeline/orchestrator.py @@ -2,9 +2,8 @@ import asyncio import time -from collections.abc import AsyncGenerator +from collections.abc import AsyncGenerator, Iterator from contextlib import aclosing -from dataclasses import dataclass import structlog @@ -35,14 +34,58 @@ RERANK_FALLBACK_MESSAGE = "Ranking output was invalid, so grounded results are s RERANK_FALLBACK_JUSTIFICATION = "Selected as a grounded match for your request." -@dataclass -class _RankingState: - selected_ids: set[str] - selected_tracks: list[Track] +class _TrackSelection: + """Rank tracks from one grounded pool, enforcing bound and uniqueness.""" + + def __init__(self, pool: tuple[Track, ...], limit: int) -> None: + """Bind the only tracks that may ever be selected.""" + self.pool = pool + self.limit = limit + self.selected: list[Track] = [] + self._tracks_by_id = {track.id: track for track in pool} + self._selected_ids: set[str] = set() @property - def track_count(self) -> int: - return len(self.selected_tracks) + def is_full(self) -> bool: + """Return whether the selection reached its limit.""" + return len(self.selected) >= self.limit + + @property + def remaining_count(self) -> int: + """Return how many further selections are allowed.""" + return self.limit - len(self.selected) + + def describe_selected_ids(self) -> str: + """Render the ids already emitted, for a correction instruction.""" + return ", ".join(sorted(self._selected_ids)) or "none" + + def select(self, track_id: str, justification: str) -> PipelineTrackEvent: + """Accept one recommender selection or reject it as invalid output.""" + if self.is_full: + raise RecommenderOutputError("Rerank returned too many track ids") + track = self._tracks_by_id.get(track_id) + if track is None: + raise RecommenderOutputError("Rerank selected an out-of-pool track id") + if track_id in self._selected_ids: + raise RecommenderOutputError("Rerank selected a duplicate track id") + return self._emit(track, justification) + + def fill_from_pool(self, justification: str) -> Iterator[PipelineTrackEvent]: + """Complete the selection in pool order after a failed rerank.""" + for track in self.pool: + if self.is_full: + return + if track.id not in self._selected_ids: + yield self._emit(track, justification) + + def _emit(self, track: Track, justification: str) -> PipelineTrackEvent: + self._selected_ids.add(track.id) + self.selected.append(track) + return PipelineTrackEvent( + rank=len(self.selected), + track=track, + justification=justification, + ) class RecommendationPipeline: @@ -86,19 +129,7 @@ class RecommendationPipeline: candidate_count=len(intent.candidates), ) - pool = self.last_pools.get(session_id) if intent.is_refinement else None - if pool is None: - result = await self.grounder.ground( - catalog, - intent.candidates, - taste.known_track_ids, - intent.familiarity, - self.settings.rerank_count + self.settings.rerank_pool_buffer, - ) - pool = result.tracks - if pool: - self.last_pools[session_id] = pool - + pool = await self._grounded_pool(session_id, catalog, intent, taste) if len(pool) < self.settings.grounding_floor: yield PipelineErrorEvent( code="insufficient_grounding", @@ -106,103 +137,96 @@ class RecommendationPipeline: ) return - state = _RankingState(selected_ids=set(), selected_tracks=[]) - async for event in self._stream_ranking(intent, pool, taste.text, history, state): + selection = _TrackSelection(pool, self.settings.rerank_count) + async for event in self._ranked_events(intent, taste.text, history, selection): yield event - new_track_count = sum( - track.id not in taste.known_track_ids for track in state.selected_tracks - ) - structlog.get_logger().info( - "recommendations_complete", - track_count=state.track_count, - new_track_count=new_track_count, - ) + _log_completion(selection, taste) yield PipelineDoneEvent( - track_count=state.track_count, + track_count=len(selection.selected), total_ms=round((time.monotonic() - started_at) * 1000), ) - async def _stream_ranking( + async def _grounded_pool( + self, + session_id: str, + catalog: MusicCatalog, + intent: Intent, + taste: CompressedTasteProfile, + ) -> tuple[Track, ...]: + """Reuse the session's pool on refinement, otherwise ground anew.""" + if intent.is_refinement: + cached_pool = self.last_pools.get(session_id) + if cached_pool: + return cached_pool + result = await self.grounder.ground( + catalog, + intent.candidates, + taste.known_track_ids, + intent.familiarity, + self.settings.rerank_count + self.settings.rerank_pool_buffer, + ) + if result.tracks: + self.last_pools[session_id] = result.tracks + return result.tracks + + async def _ranked_events( self, intent: Intent, - pool: tuple[Track, ...], taste_summary: str, history: tuple[ConversationTurn, ...], - state: _RankingState, + selection: _TrackSelection, ) -> AsyncGenerator[PipelineTrackEvent | PipelineWarningEvent]: + """Stream one rerank, retry once on invalid output, then fall back.""" correction: str | None = None - for _attempt_number in range(2): - remaining_count = self.settings.rerank_count - state.track_count - if remaining_count <= 0: + for _ in range(2): + if selection.is_full: return try: - async for event in self._validated_rerank( - intent, - pool, - taste_summary, - history, - remaining_count, - correction, - state, + async for event in self._rerank_once( + intent, taste_summary, history, selection, correction ): yield event return except RecommenderOutputError as error: - emitted_ids = ", ".join(sorted(state.selected_ids)) or "none" correction = ( - f"Validation failed: {error}. Already emitted track ids: {emitted_ids}." + f"Validation failed: {error}." + f" Already emitted track ids: {selection.describe_selected_ids()}." ) yield PipelineWarningEvent(code=RERANK_FALLBACK_CODE, message=RERANK_FALLBACK_MESSAGE) - for track in pool: - if state.track_count >= self.settings.rerank_count: - break - if track.id in state.selected_ids: - continue - state.selected_ids.add(track.id) - state.selected_tracks.append(track) - yield PipelineTrackEvent( - rank=state.track_count, - track=track, - justification=RERANK_FALLBACK_JUSTIFICATION, - ) + for event in selection.fill_from_pool(RERANK_FALLBACK_JUSTIFICATION): + yield event - async def _validated_rerank( + async def _rerank_once( self, intent: Intent, - pool: tuple[Track, ...], taste_summary: str, history: tuple[ConversationTurn, ...], - selection_count: int, + selection: _TrackSelection, correction: str | None, - state: _RankingState, ) -> AsyncGenerator[PipelineTrackEvent]: - tracks_by_id = {track.id: track for track in pool} stream = self.recommender.stream_rerank( intent, - pool, + selection.pool, taste_summary, history, - selection_count, + selection.remaining_count, correction, ) async with aclosing(stream) as selections: - async for selection in selections: - if state.track_count >= self.settings.rerank_count: - raise RecommenderOutputError("Rerank returned too many track ids") - track = tracks_by_id.get(selection.track_id) - if track is None: - raise RecommenderOutputError("Rerank selected an out-of-pool track id") - if track.id in state.selected_ids: - raise RecommenderOutputError("Rerank selected a duplicate track id") - state.selected_ids.add(track.id) - state.selected_tracks.append(track) - yield PipelineTrackEvent( - rank=state.track_count, - track=track, - justification=selection.justification, - ) + async for item in selections: + yield selection.select(item.track_id, item.justification) + + +def _log_completion(selection: _TrackSelection, taste: CompressedTasteProfile) -> None: + """Log how many recommendations were served and how many are new.""" + new_track_count = sum(track.id not in taste.known_track_ids for track in selection.selected) + structlog.get_logger().info( + "recommendations_complete", + track_count=len(selection.selected), + new_track_count=new_track_count, + ) class _TasteProfileCache: diff --git a/docs/logboek.md b/docs/logboek.md index fc8bcfb..5814a02 100644 --- a/docs/logboek.md +++ b/docs/logboek.md @@ -29,7 +29,7 @@ Waarom: Wat ik heb laten vallen of uitgesteld: -- Domainmodellen, protocollen, prompts, pipeline-instellingen en het +- Domainmodellen, protocols, prompts, pipeline-instellingen en het frontend-typecontract bewust nog niet neergezet. Die ontstaan in de stap waar ze horen. Ik probeer op die manier bewust vroeg drift en dode code te voorkomen. - Geen apart beslisdocument. De motivering staat in de README en hier. @@ -40,19 +40,19 @@ Wat ik heb laten vallen of uitgesteld: Wat ik deed: - Login via Spotify met PKCE en cookie-sessies: tokens blijven server-side, de browser krijgt alleen een opaque HttpOnly cookie. -- Dunne async client op de Spotify Web API met per-endpoint retrybeleid: +- Dunne async client op de Spotify Web API met per-endpoint retry policy: leesacties herhalen maximaal 1 keer en alleen binnen een grens (Retry-After), schrijfacties op playlists nooit. - Token-refresh is single-flight: parallelle requests delen 1 refresh in plaats van er allemaal zelf een te starten. - Mapping van Spotify-JSON naar eigen modellen op 1 plek; kapotte items vallen weg in plaats van dat ze de app breken. -- Getypeerde fouten en 12 transport- en routetests op een mock transport. +- Getypeerde errors en 12 transport- en routetests op een mock transport. - Na een eerste review de login-flow uit de routes getrokken naar een eigen - module (routes zijn nu dunne doorgeefluiken) en het retrybeleid herschreven + module (routes zijn nu dunne passthroughs) en de retry policy herschreven naar een lineaire keten van losse regels in plaats van een loop met flags; - foutdetails uit de Spotify-body worden meegenomen in de getypeerde fouten - en quota-uitputting wordt apart herkend en nooit opnieuw geprobeerd. + foutdetails uit de Spotify-body worden meegenomen in de getypeerde errors + en quota exhaustion (QUOTA_EXCEEDED) wordt apart herkend en nooit opnieuw geprobeerd. - Daarna het API-contract vastgelegd: request-schema en de gestreamde events (metadata / track / warning / error / done), gespiegeld in TypeScript. @@ -60,7 +60,7 @@ Wat ik deed: Waarom: - Schrijfacties blind herhalen kan dubbele playlist-items opleveren; dat - risico sluit ik structureel uit in het retrybeleid. + risico sluit ik structureel uit in de retry policy. - Het contract eerst bevriezen maakt parallel werken aan frontend en pipeline mogelijk zonder elkaar te breken. @@ -80,15 +80,19 @@ Wat ik deed: Spotify-smaakprofiel naar prompttekst plus een set bekende track-ids, prompts als data in een eigen module, en alle instelbare waarden gesectioneerd in de config met per waarde het waarom. -- Twee LLM-aanroepen achter een eigen interface: aanroep 1 interpreteert de +- Twee LLM-calls achter een eigen interface: call 1 interpreteert de vraag (stemming, activiteit, taal, bekendheid) en stelt 30-40 echte - nummers voor als gestructureerde output; aanroep 2 herordent uitsluitend + nummers voor als gestructureerde output; call 2 herordent uitsluitend geverifieerde nummers en streamt per nummer een eerlijke onderbouwing. Elke output wordt gevalideerd, met hooguit 1 herstelpoging. -- Grounding: begrensde parallelle zoekslag met vroege stop, een deadline, +- Grounding: begrensde parallelle search fan-out met early stop, een deadline, een naam-naar-id cache en twee aparte metrieken: niet gevonden versus - wel gevonden maar afgekeurd door de controle. Een track-id dat niet in + wel gevonden maar afgekeurd door de match-check. Een track-id dat niet in de geverifieerde pool zit kan nooit bij de gebruiker terechtkomen. +- Na review de orkestratie herschreven: de stream-functie leest nu als de + pipeline-stappen zelf, en de selectie-logica (alleen pool-ids, geen + duplicaten, begrensd aantal, ranking) zit in 1 kleine klasse die zowel + het normale pad als de fallback bedient. Waarom: @@ -96,13 +100,13 @@ Waarom: deterministische muren: alles wat hij ziet is echte data, alles wat de gebruiker ziet is geverifieerd op Spotify. Een verzonnen nummer valt stilletjes af en verschijnt nooit. -- Zoeken geeft maximaal 10 resultaten per aanroep, dus resolutie is per +- Zoeken geeft maximaal 10 resultaten per call, dus resolutie is per definitie een fan-out; liever een kandidaat laten vallen dan het verkeerde nummer aanbevelen. Wat ik heb laten vallen of uitgesteld: -- Verfijnvragen doen geen nieuwe zoekslag: turn 2 herordent de bestaande - geverifieerde pool. Sneller en consistent, maar een verfijning haalt +- Refinements doen geen nieuwe search-ronde: turn 2 herordent de bestaande + geverifieerde pool. Sneller en consistent, maar een refinement haalt geen nieuwe nummers op. Dit is een bewuste afweging, mocht er tijd over zijn is dit 1 van de uitbreidingen die ik op zou kunnen pakken.