Enrichment

Opt-in deep-parse enrichment: structured overlays from freeform vgmdb text.

AlbumEnrichment

Bases: VgmdbModel

Per-track credits extracted from an album's freeform notes, keyed by track number.

A separate overlay: it does not modify the M1 Album/Track models. is_empty is true when no track carries any credit (e.g. when no backend was configured).

Source code in src/vgmdb_client/enrich/models.py

class AlbumEnrichment(VgmdbModel):
    """Per-track credits extracted from an album's freeform notes, keyed by track number.

    A separate overlay: it does not modify the M1 ``Album``/``Track`` models. ``is_empty`` is true
    when no track carries any credit (e.g. when no backend was configured).
    """

    album_id: int
    track_credits: dict[int, list[Credit]] = Field(default_factory=dict)

    @property
    def is_empty(self) -> bool:
        return not any(self.track_credits.values())

EnrichmentBackend

Bases: Protocol

A source of enrichment for an album's freeform text.

Implementations (an LLM endpoint now, an embedded model later) are interchangeable.

Source code in src/vgmdb_client/enrich/backend.py

class EnrichmentBackend(Protocol):
    """A source of enrichment for an album's freeform text.

    Implementations (an LLM endpoint now, an embedded model later) are interchangeable.
    """

    def enrich(self, album: Album, raw_text: str) -> AlbumEnrichment:
        """Return the enrichment for ``album`` derived from its freeform ``raw_text``."""
        ...

enrich(album, raw_text)

Return the enrichment for album derived from its freeform raw_text.

Source code in src/vgmdb_client/enrich/backend.py

def enrich(self, album: Album, raw_text: str) -> AlbumEnrichment:
    """Return the enrichment for ``album`` derived from its freeform ``raw_text``."""
    ...

EnrichmentError

Bases: VgmdbClientError

Raised when a configured enrichment backend fails to produce a valid result.

Source code in src/vgmdb_client/enrich/errors.py

class EnrichmentError(VgmdbClientError):
    """Raised when a configured enrichment backend fails to produce a valid result."""

    def __init__(self, message: str = "Enrichment failed.") -> None:
        super().__init__(message)

OpenAICompatibleBackend

Enrichment via an OpenAI-compatible /chat/completions endpoint.

The prompt is customizable (system_prompt / user_template), the output can be requested as free JSON, a JSON schema, or a forced tool call (output_mode), every reply is validated against an internal response model, and an invalid reply is retried with a corrective message.

Source code in src/vgmdb_client/enrich/llm.py

class OpenAICompatibleBackend:
    """Enrichment via an OpenAI-compatible ``/chat/completions`` endpoint.

    The prompt is customizable (``system_prompt`` / ``user_template``), the output can be requested as
    free JSON, a JSON schema, or a forced tool call (``output_mode``), every reply is validated against
    an internal response model, and an invalid reply is retried with a corrective message.
    """

    def __init__(
        self,
        url: str,
        model: str = _DEFAULT_MODEL,
        api_key: str | None = None,
        timeout: float = _DEFAULT_TIMEOUT,
        *,
        output_mode: OutputMode = _DEFAULT_OUTPUT_MODE,
        system_prompt: str | None = None,
        user_template: str | None = None,
        max_retries: int = 1,
    ) -> None:
        self._url = url
        self._model = model
        self._api_key = api_key
        self._timeout = timeout
        self._output_mode = output_mode
        self._system_prompt = system_prompt if system_prompt is not None else _DEFAULT_SYSTEM_PROMPT
        self._user_template = user_template if user_template is not None else _DEFAULT_USER_TEMPLATE
        self._max_retries = max_retries

    def enrich(self, album: Album, raw_text: str) -> AlbumEnrichment:
        messages = self._build_messages(album, raw_text)
        last_error: Exception | None = None
        for _ in range(self._max_retries + 1):
            raw = self._call(messages)  # raises EnrichmentError on transport / malformed envelope
            try:
                response = _LlmResponse.model_validate_json(raw)
                return _build_enrichment(album.id, response)
            except (ValidationError, ValueError) as exc:
                last_error = exc
                messages = [*messages, _corrective_message(exc)]
        raise EnrichmentError(_MALFORMED) from last_error

    def _build_messages(self, album: Album, raw_text: str) -> list[dict[str, str]]:
        tracklist = "\n".join(
            f"{track.number}. {track.titles.default or ''}"
            for disc in album.discs
            for track in disc.tracks
            if track.number is not None
        )
        try:
            user = self._user_template.format(tracklist=tracklist, notes=raw_text)
        except (KeyError, IndexError, ValueError) as exc:
            raise EnrichmentError(_BAD_TEMPLATE) from exc
        return [{"role": "system", "content": self._system_prompt}, {"role": "user", "content": user}]

    def _mode_payload(self) -> dict[str, Any]:
        """The request fields that select the output mode."""
        if self._output_mode == "json_schema":
            schema = {"name": "enrichment", "schema": _RESPONSE_SCHEMA, "strict": False}
            return {"response_format": {"type": "json_schema", "json_schema": schema}}
        if self._output_mode == "tool":
            function = {"name": _TOOL_NAME, "description": "Return per-track credits.", "parameters": _RESPONSE_SCHEMA}
            return {
                "tools": [{"type": "function", "function": function}],
                "tool_choice": {"type": "function", "function": {"name": _TOOL_NAME}},
            }
        return {"response_format": {"type": "json_object"}}

    def _call(self, messages: list[dict[str, str]]) -> str:
        """POST the request and return the raw result string (JSON content, or tool-call arguments)."""
        payload: dict[str, Any] = {"model": self._model, "messages": messages, "temperature": 0}
        payload.update(self._mode_payload())
        headers = {"Content-Type": "application/json"}
        if self._api_key:
            headers["Authorization"] = f"Bearer {self._api_key}"
        try:
            response = httpx.post(self._url, json=payload, headers=headers, timeout=self._timeout)
            response.raise_for_status()
            message = response.json()["choices"][0]["message"]
            if self._output_mode == "tool":
                return str(message["tool_calls"][0]["function"]["arguments"])
            return _content_text(message.get("content"))
        except (httpx.HTTPError, KeyError, IndexError, TypeError) as exc:
            raise EnrichmentError(_REQUEST_FAILED) from exc

RuleBasedBackend

Extract per-track credits from notes with conservative deterministic rules.

Source code in src/vgmdb_client/enrich/rules.py

class RuleBasedBackend:
    """Extract per-track credits from notes with conservative deterministic rules."""

    def enrich(self, album: Album, raw_text: str) -> AlbumEnrichment:
        track_credits: dict[int, list[Credit]] = {}
        current: set[int] | None = None
        for line in raw_text.splitlines():
            if not line.strip():
                current = None
                continue
            header = _header_tracks(line)
            if header is not None:
                current = header
                continue
            role = _role_line(line)
            if role is not None:
                self._emit(track_credits, current, role[0], role[1])
            else:
                current = None  # prose breaks the block context (don't leak a header across it)

        # Keep only credits for tracks the album actually has: drops bare-year/non-track headers and
        # bounds anything an over-wide range produced.
        valid = {track.number for disc in album.discs for track in disc.tracks if track.number is not None}
        ordered = {track: track_credits[track] for track in sorted(track_credits) if track in valid}
        return AlbumEnrichment(album_id=album.id, track_credits=ordered)

    @staticmethod
    def _emit(
        track_credits: dict[int, list[Credit]],
        current: set[int] | None,
        role_phrase: str,
        value: str,
    ) -> None:
        """Emit credits for a role line: inline ``(ranges)`` per name, else the current track set."""

        def add(tracks: set[int], names: list[str]) -> None:
            if not tracks or not names:
                return
            for track in tracks:
                track_credits.setdefault(track, []).append(_credit(role_phrase, names))

        ranged = [(names, _parse_track_set(rng)) for names, rng in _INLINE_GROUP.findall(value)]
        ranged = [(names, tracks) for names, tracks in ranged if tracks]
        if ranged:  # inline parenthetical: names attach to their own ranges
            for names, tracks in ranged:
                add(tracks, _split_names(names))
        elif current is not None:  # block: attribute to the current header's tracks
            add(current, _split_names(value))

backend_from_env()

Build a backend from LLM_URL/LLM_MODEL/LLM_API_KEY/LLM_OUTPUT_MODE, else None.

Source code in src/vgmdb_client/enrich/llm.py

def backend_from_env() -> OpenAICompatibleBackend | None:
    """Build a backend from ``LLM_URL``/``LLM_MODEL``/``LLM_API_KEY``/``LLM_OUTPUT_MODE``, else ``None``."""
    url = os.environ.get("LLM_URL")
    if not url:
        return None
    mode_raw = os.environ.get("LLM_OUTPUT_MODE", _DEFAULT_OUTPUT_MODE)
    mode = cast(OutputMode, mode_raw) if mode_raw in get_args(OutputMode) else _DEFAULT_OUTPUT_MODE
    return OpenAICompatibleBackend(
        url=url,
        model=os.environ.get("LLM_MODEL", _DEFAULT_MODEL),
        api_key=os.environ.get("LLM_API_KEY"),
        output_mode=mode,
    )

enrich_album(album, backend=None)

Enrich an album's per-track credits from its notes via backend.

With no backend this degrades gracefully to an empty :class:AlbumEnrichment (the M3 album is fully usable); a configured backend that fails raises :class:EnrichmentError.

Source code in src/vgmdb_client/enrich/__init__.py

def enrich_album(album: Album, backend: EnrichmentBackend | None = None) -> AlbumEnrichment:
    """Enrich an album's per-track credits from its notes via ``backend``.

    With no backend this degrades gracefully to an empty :class:`AlbumEnrichment` (the M3 album is
    fully usable); a configured backend that fails raises :class:`EnrichmentError`.
    """
    if backend is None:
        return AlbumEnrichment(album_id=album.id)
    return backend.enrich(album, album.notes or "")