128 lines
4.6 KiB
Python
128 lines
4.6 KiB
Python
"""Anthropic recommender adapter backed by recorded response chunks."""
|
|
|
|
import asyncio
|
|
import json
|
|
from collections.abc import AsyncGenerator
|
|
from contextvars import ContextVar
|
|
|
|
from app.adapters.anthropic.llm import (
|
|
IntentOutput,
|
|
RecommendationObjectParser,
|
|
RerankOutput,
|
|
to_intent,
|
|
)
|
|
from app.adapters.demo.cassette import DemoCassette, load_cassette
|
|
from app.adapters.demo.scenario import select_replay_scenario
|
|
from app.config import Settings
|
|
from app.domain.models import (
|
|
ConversationTurn,
|
|
Intent,
|
|
PreviousRecommendation,
|
|
RerankSelection,
|
|
Track,
|
|
)
|
|
from app.ports.protocols import RecommenderOutputError
|
|
|
|
|
|
class DemoRecommender:
|
|
"""Replay recorded intent and rerank output through live validation."""
|
|
|
|
def __init__(self, settings: Settings) -> None:
|
|
"""Bind replay pacing and task-local scenario state."""
|
|
self.settings = settings
|
|
self._cassette: ContextVar[DemoCassette | None] = ContextVar(
|
|
"demo_recommender_cassette",
|
|
default=None,
|
|
)
|
|
|
|
async def create_intent(
|
|
self,
|
|
query: str,
|
|
history: tuple[ConversationTurn, ...],
|
|
previous_recommendations: tuple[PreviousRecommendation, ...],
|
|
taste_summary: str,
|
|
candidate_count: int,
|
|
) -> Intent:
|
|
"""Parse the selected scenario's recorded structured intent."""
|
|
match = select_replay_scenario(query, bool(previous_recommendations))
|
|
cassette = load_cassette(match.key)
|
|
self._cassette.set(cassette)
|
|
return parse_recorded_intent(cassette.intent_response_body)
|
|
|
|
async def stream_rerank(
|
|
self,
|
|
intent: Intent,
|
|
grounded_tracks: tuple[Track, ...],
|
|
taste_summary: str,
|
|
history: tuple[ConversationTurn, ...],
|
|
selection_count: int,
|
|
correction: str | None = None,
|
|
) -> AsyncGenerator[RerankSelection]:
|
|
"""Replay recorded SSE chunks through the live incremental parser."""
|
|
cassette = self._cassette.get()
|
|
if cassette is None:
|
|
raise RecommenderOutputError("Demo rerank has no selected scenario")
|
|
parser = RecommendationObjectParser()
|
|
async for text_delta in _text_deltas(
|
|
cassette.rerank_response_chunks,
|
|
self.settings.demo_chunk_delay_seconds,
|
|
):
|
|
for selection in parser.feed(text_delta):
|
|
yield RerankSelection(selection.track_id, selection.justification)
|
|
try:
|
|
validated = RerankOutput.model_validate_json(parser.complete_text)
|
|
except ValueError as error:
|
|
raise RecommenderOutputError("Recorded rerank response is invalid") from error
|
|
if len(validated.recommendations) > selection_count:
|
|
raise RecommenderOutputError("Recorded rerank returned too many selections")
|
|
|
|
|
|
def parse_recorded_intent(response_body: bytes) -> Intent:
|
|
"""Parse an Anthropic message body through the live intent output model."""
|
|
try:
|
|
response = json.loads(response_body)
|
|
content = response["content"]
|
|
text = content[0]["text"]
|
|
if not isinstance(text, str):
|
|
raise TypeError
|
|
return to_intent(IntentOutput.model_validate_json(text))
|
|
except (IndexError, KeyError, TypeError, ValueError) as error:
|
|
raise RecommenderOutputError("Recorded intent response is invalid") from error
|
|
|
|
|
|
async def _text_deltas(
|
|
chunks: tuple[bytes, ...],
|
|
delay_seconds: float,
|
|
) -> AsyncGenerator[str]:
|
|
buffer = b""
|
|
for chunk_index, chunk in enumerate(chunks):
|
|
if chunk_index and delay_seconds > 0:
|
|
await asyncio.sleep(delay_seconds)
|
|
buffer += chunk
|
|
buffer = buffer.replace(b"\r\n", b"\n")
|
|
while b"\n\n" in buffer:
|
|
event, buffer = buffer.split(b"\n\n", 1)
|
|
text_delta = _event_text_delta(event)
|
|
if text_delta is not None:
|
|
yield text_delta
|
|
if buffer:
|
|
text_delta = _event_text_delta(buffer)
|
|
if text_delta is not None:
|
|
yield text_delta
|
|
|
|
|
|
def _event_text_delta(event: bytes) -> str | None:
|
|
data_lines = [line[5:].strip() for line in event.splitlines() if line.startswith(b"data:")]
|
|
if not data_lines:
|
|
return None
|
|
try:
|
|
payload = json.loads(b"\n".join(data_lines))
|
|
except (json.JSONDecodeError, UnicodeDecodeError):
|
|
return None
|
|
if not isinstance(payload, dict) or payload.get("type") != "content_block_delta":
|
|
return None
|
|
delta = payload.get("delta")
|
|
if not isinstance(delta, dict) or delta.get("type") != "text_delta":
|
|
return None
|
|
text = delta.get("text")
|
|
return text if isinstance(text, str) else None
|