"""LLM classifier contracts with a parametric prompt builder.
Defines the :class:`LLMClassifier` and :class:`LLMSplitProposer` protocols and
the allow-list-guarded response parsing every classifier answer passes through.
Concrete transports live outside this module: the on-host readers in the gated
inference package, and a subprocess harness on the test side. The subprocess
CLI implementations that once lived here were the off-host route, and they were
deleted with it.
This module keeps the part that is a SAFETY contract rather than a transport --
the prompt spec and the parse that confine a model to selecting
``classification`` / ``category`` / ``iva_category`` and never emitting a
regulated number. That belongs in the domain, not behind an optional install.
The prompt is built
PROGRAMMATICALLY from the available enum values so the LLM prompt
stays in sync with :class:`cadrumo.domain.transactions.enums.BusinessClassification`:
adding a new value automatically requires a developer to decide
whether it belongs in the default LLM choice set.
The prompt spec is parametrized:
- ``classifications``: which :class:`cadrumo.domain.transactions.enums.BusinessClassification`
values the LLM may pick. Defaults to the four *decision* states
(``BUSINESS`` / ``PERSONAL`` / ``MIXED`` / ``PROCESSED_UNCLASSIFIED``).
Pipeline-state values (``NOT_YET_PROCESSED``, ``SKIPPED_BY_RULE``,
``FAILED_VALIDATION``) are excluded because they are not LLM
decisions -- they are internal pipeline bookkeeping.
- ``categories``: optional :class:`cadrumo.domain.categories.spending_category.SpendingCategory`
values the LLM may additionally attach. Empty by default
(classification-only). When populated, the response includes a
``category`` field.
Every decision the LLM emits is validated against the spec's
allow-list via :func:`parse_response`: a response that picks a value
outside the allowed set raises :class:`LLMClassifierError`, so a
hallucinating model cannot corrupt the catalogue.
"""
from __future__ import annotations
import re
from collections.abc import Iterable
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import TYPE_CHECKING, Protocol, runtime_checkable
from pydantic import BaseModel, Field, field_validator
from ...core.errors.hierarchy import pydantic_validation_boundary
from ...core.i18n.render import tr as _tr
from ...core.models import STRICT_FROZEN_CONFIG as _STRICT_FROZEN
from ...core.unit_proportion import UNIT_PROPORTION_MAX, UNIT_PROPORTION_MIN, is_unit_proportion
from ..calculations.registry.iva_category_catalogue import resolve_iva_category_catalogue
from ..categories.registry import resolve_category_profiles
from ..categories.spending_category import SpendingCategory
from ..categories.spending_category_catalogue import spending_category_tokens
from ..iva.schema import IvaCategory
from .enums import BusinessClassification
from .errors import LLMClassifierError, TransactionValidationError
from .model_tier import MINIMUM_CLASSIFICATION_TIER, ModelProfile, ModelTier
from .models import Transaction
if TYPE_CHECKING:
from ..calculations.registry.authority import PinnedAuthorityOperation
_REASON_MAX_LENGTH = 2048
# ── response model ────────────────────────────────────────────────
[docs]
class LLMClassificationResponse(BaseModel):
"""One LLM-emitted classification result for a transaction."""
model_config = _STRICT_FROZEN
classification: BusinessClassification
confidence: Decimal
reason: str = Field(min_length=1, max_length=_REASON_MAX_LENGTH)
category: SpendingCategory | None = None
iva_category: IvaCategory | None = None
business_pct: Decimal | None = None
multiple_components: bool | None = None
"""Evidence-read multiplicity judgement: True when the attached invoice carries
multiple distinct rate/category lines that warrant a split into independently
filable base/IVA children. ``None`` when no evidence was read (the model cannot
judge multiplicity from the bank row alone). A boolean judgement, not an
allow-list value, so hallucination containment is unaffected; it only drives a
non-blocking split *recommendation*, never a write."""
@field_validator("confidence")
@classmethod
@pydantic_validation_boundary
def _check_confidence_range(cls, value: Decimal) -> Decimal:
"""Restrict confidence to the inclusive 0..1 range."""
if not is_unit_proportion(value):
raise TransactionValidationError("confidence must be within the inclusive 0..1 range")
return value
@field_validator("business_pct")
@classmethod
@pydantic_validation_boundary
def _check_business_pct_range(cls, value: Decimal | None) -> Decimal | None:
"""Restrict the proposed MIXED business percentage to the inclusive 0..1 range.
The model only *proposes* the split direction; the percentage is a
non-regulated hint the operator confirms. ``None`` is the common case
(BUSINESS / PERSONAL suggestions carry no percentage).
"""
if value is not None and not is_unit_proportion(value):
raise TransactionValidationError("business_pct must be within the inclusive 0..1 range")
return value
@field_validator("reason")
@classmethod
@pydantic_validation_boundary
def _strip_reason(cls, value: str) -> str:
"""Trim whitespace and reject empty reasons."""
trimmed = value.strip()
if not trimmed:
raise TransactionValidationError("reason must not be empty")
return trimmed
[docs]
class LLMSplitChild(BaseModel):
"""One proposed child of an evidence-driven transaction split.
The model proposes a *proportion* of the parent (a fraction, like
``business_pct``) plus the selections for that child; the application derives
each child's euro amount from the parent gross and the regulated tax substrate
from the registry. The model never emits a euro amount or a regulated number
(``llm-selects-system-derives-tax-numbers``).
"""
model_config = _STRICT_FROZEN
proportion: Decimal
category: SpendingCategory | None = None
iva_category: IvaCategory | None = None
evidence_citation: str = Field(default="", max_length=_REASON_MAX_LENGTH)
@field_validator("proportion")
@classmethod
@pydantic_validation_boundary
def _check_proportion(cls, value: Decimal) -> Decimal:
"""Restrict each child proportion to the half-open (0, 1] range."""
if not (UNIT_PROPORTION_MIN < value <= UNIT_PROPORTION_MAX):
raise TransactionValidationError("each split child proportion must be within (0, 1]")
return value
[docs]
class LLMSplitResponse(BaseModel):
"""An evidence-driven split proposal: children whose proportions sum to one.
A proposal of **one** child (proportion ``1.0``) is the "no split warranted"
verdict — the model read the invoice and judged it a single line/rate. The
application surfaces that verdict for review and never applies a degenerate
one-way split. Two or more children is a genuine split recommendation.
"""
model_config = _STRICT_FROZEN
children: tuple[LLMSplitChild, ...]
reason: str = Field(min_length=1, max_length=_REASON_MAX_LENGTH)
@property
def recommends_split(self) -> bool:
"""True when the model proposes more than one child (a genuine split)."""
return len(self.children) > 1
@field_validator("children")
@classmethod
@pydantic_validation_boundary
def _check_children(cls, value: tuple[LLMSplitChild, ...]) -> tuple[LLMSplitChild, ...]:
"""Require at least one child whose proportions sum to ~1.0.
One child is the no-split verdict; two or more is a split. Either way the
proportions must sum to approximately 1.0 (a single child must therefore
carry proportion 1.0).
"""
if not value:
raise TransactionValidationError("a split proposal must carry at least one child")
total = sum((child.proportion for child in value), Decimal("0"))
if abs(total - UNIT_PROPORTION_MAX) > Decimal("0.01"):
raise TransactionValidationError("split child proportions must sum to approximately 1.0")
return value
@field_validator("reason")
@classmethod
@pydantic_validation_boundary
def _strip_reason(cls, value: str) -> str:
"""Trim whitespace and reject empty reasons."""
trimmed = value.strip()
if not trimmed:
raise TransactionValidationError("reason must not be empty")
return trimmed
# ── protocol ──────────────────────────────────────────────────────
[docs]
class LLMClassifier(Protocol):
"""Classify one transaction with an LLM-generated decision."""
@property
def decided_by(self) -> str:
"""Return the ``classified_by`` identifier this classifier emits."""
...
[docs]
def classify(self, transaction: Transaction, *, evidence_text: str | None = None) -> LLMClassificationResponse:
"""Return one classification decision for ``transaction``.
Args:
transaction: The transaction to classify.
evidence_text: Optional on-host-extracted attached-evidence text to
inject into the prompt. Gating of whether evidence may reach a given
(on-host vs cloud) classifier is the caller's responsibility.
Returns:
A :class:`LLMClassificationResponse` with the classification result.
"""
...
[docs]
@runtime_checkable
class LLMSplitProposer(Protocol):
"""Propose an evidence-driven N-way split for one transaction."""
@property
def decided_by(self) -> str:
"""Return the ``classified_by`` identifier this proposer emits."""
...
[docs]
def propose_split(self, transaction: Transaction, *, evidence_text: str | None = None) -> LLMSplitResponse:
"""Return an N-way split proposal for ``transaction``.
Args:
transaction: The transaction to split.
evidence_text: Optional on-host-extracted attached-evidence text to
inject into the prompt.
Returns:
A validated :class:`LLMSplitResponse`.
"""
...
# ── parametric prompt builder ─────────────────────────────────────
[docs]
@dataclass(frozen=True, slots=True)
class ClassificationChoice:
"""One allowed :class:`BusinessClassification` paired with an LLM-facing hint."""
value: BusinessClassification
hint: str
[docs]
@dataclass(frozen=True, slots=True)
class CategoryChoice:
"""One allowed :class:`SpendingCategory` paired with an LLM-facing hint."""
value: SpendingCategory
hint: str
[docs]
@dataclass(frozen=True, slots=True)
class IvaCategoryChoice:
"""One allowed :class:`cadrumo.domain.iva.schema.IvaCategory` paired with an LLM-facing hint."""
value: IvaCategory
hint: str
# Descriptive hints for the four LLM-addressable classification states. Kept
# as module constants so the descriptions live next to their values and can
# be overridden by callers that build a custom PromptSpec.
_DEFAULT_CLASSIFICATION_HINTS: dict[BusinessClassification, str] = {
BusinessClassification.BUSINESS: "certain business expense or income",
BusinessClassification.PERSONAL: "certain personal expense or income",
BusinessClassification.MIXED: "partially business, partially personal",
BusinessClassification.PROCESSED_UNCLASSIFIED: ("you looked at it carefully but cannot decide either way"),
}
[docs]
def default_classification_choices() -> tuple[ClassificationChoice, ...]:
"""Return the default allowed-classifications tuple used by the prompt.
Returns:
Tuple of :class:`ClassificationChoice` objects for each allowed category.
"""
return tuple(ClassificationChoice(value=value, hint=hint) for value, hint in _DEFAULT_CLASSIFICATION_HINTS.items())
[docs]
@dataclass(frozen=True)
class PromptSpec:
"""Parametrized classification prompt spec.
The prompt and the response allow-list are derived from the same
tuple of choices so they cannot drift. A response whose
classification is not in ``allowed_classifications()`` is rejected
by :func:`parse_response` regardless of how well-formed the JSON is.
"""
classifications: tuple[ClassificationChoice, ...] = field(
default_factory=default_classification_choices,
)
categories: tuple[CategoryChoice, ...] = ()
iva_categories: tuple[IvaCategoryChoice, ...] = ()
header: str = "You are classifying a Spanish autónomo's bank transaction for tax purposes."
[docs]
def allowed_classifications(self) -> frozenset[BusinessClassification]:
"""Return the set of :class:`BusinessClassification` values the LLM is allowed to emit."""
return frozenset(choice.value for choice in self.classifications)
[docs]
def allowed_categories(self) -> frozenset[SpendingCategory]:
"""Return the set of category values the LLM is allowed to emit (empty = none).
Returns:
Frozenset of :class:`SpendingCategory` values the LLM may emit.
"""
return frozenset(choice.value for choice in self.categories)
[docs]
def allowed_iva_categories(self) -> frozenset[IvaCategory]:
"""Return the set of :class:`cadrumo.domain.iva.schema.IvaCategory` values the LLM may emit (empty = none).
Returns:
Frozenset of :class:`cadrumo.domain.iva.schema.IvaCategory` values the LLM may
select from; empty when the spec does not ask for an IVA category.
"""
return frozenset(choice.value for choice in self.iva_categories)
[docs]
def render(
self,
transaction: Transaction,
*,
evidence_text: str | None = None,
evidence_image_present: bool = False,
) -> str:
"""Render the prompt for ``transaction`` against this spec.
Args:
transaction: The transaction to classify.
evidence_text: Optional on-host-extracted text of an attached evidence
document (e.g. a purchase invoice). When present it is injected into
the prompt for the model to read; the model uses it only to select
the classification/category/iva_category and must never copy a euro
figure from it (the regulated numbers stay registry-derived).
evidence_image_present: Set when the evidence is attached as an image
(the on-host vision-read path) instead of inlined text; the prompt
then points the model at the attached image.
"""
return _render_prompt(
self,
transaction,
evidence_text=evidence_text,
evidence_image_present=evidence_image_present,
)
[docs]
def default_prompt_spec() -> PromptSpec:
"""Return the default :class:`PromptSpec`: classification-only, four decision states."""
return PromptSpec()
[docs]
def prompt_spec_with_every_spending_category(
*,
year: int,
operation: PinnedAuthorityOperation,
classifications: tuple[ClassificationChoice, ...] | None = None,
) -> PromptSpec:
"""Return a prompt spec that also asks the LLM to suggest a SpendingCategory.
Pulls authoritative Spanish display labels from the category profile
registry for ``year`` rather than
inventing ad-hoc hints from the enum value -- the LLM picks
categories far more accurately against the real AEAT terminology
than against mangled snake_case. Categories with no registered
profile (none today; every
:class:`cadrumo.domain.categories.spending_category.SpendingCategory` member is covered)
fall back to the humanised enum value.
Args:
year: Filing year whose category profiles supply the hints. Passed
explicitly because the proportionality kind a category carries is
year-versioned regulatory data.
operation: The generation-pinned authority operation used for both the
dated category vocabulary and its profiles.
classifications: Optional override for the classification
choices; defaults to :func:`default_classification_choices`.
Returns:
A :class:`PromptSpec` whose ``categories`` tuple covers every
registered :class:`cadrumo.domain.categories.spending_category.SpendingCategory`.
"""
category_choices = tuple(
CategoryChoice(value=value, hint=_category_hint(value, year=year, operation=operation))
for value in spending_category_tokens(effective_date=date(year, 12, 31), authority=operation)
)
return PromptSpec(
classifications=classifications or default_classification_choices(),
categories=category_choices,
)
[docs]
def default_iva_category_choices(*, operation: PinnedAuthorityOperation) -> tuple[IvaCategoryChoice, ...]:
"""Return the grounded IVA-category choices for the saturation prompt.
The allow-list and concise hints are projected from the dated IVA category
facts registry. The model SELECTS a category only; every
regulated euro figure is derived downstream from the registry rate, never
emitted by the model.
Returns:
One :class:`IvaCategoryChoice` per registry-declared IVA category, in
authored order.
Args:
operation: The generation-pinned authority operation used to resolve the
IVA category catalogue.
"""
catalogue = resolve_iva_category_catalogue(authority=operation)
return tuple(
IvaCategoryChoice(value=definition.token, hint=catalogue.hint(definition.token))
for definition in catalogue.definitions
)
[docs]
def prompt_spec_with_saturation_fields(
*,
year: int,
operation: PinnedAuthorityOperation,
classifications: tuple[ClassificationChoice, ...] | None = None,
) -> PromptSpec:
"""Return a prompt spec for full saturation: spending + IVA category selection.
Extends :func:`prompt_spec_with_every_spending_category` with the
registry-grounded IVA-category allow-list (and invites a proposed MIXED
``business_pct``) so one reviewed suggestion can carry the rich tax
metadata. The model selects categories only; the regulated rate, taxable
base, and IVA amount are derived downstream from the registry, never
emitted by the model.
Args:
year: Filing year whose category profiles supply the hints. Passed
explicitly because the proportionality kind a category carries is
year-versioned regulatory data.
operation: The generation-pinned authority operation used for the
spending and IVA category projections.
classifications: Optional override for the classification choices;
defaults to :func:`default_classification_choices`.
Returns:
A :class:`PromptSpec` carrying both the spending-category and the
IVA-category allow-lists.
"""
category_choices = tuple(
CategoryChoice(value=value, hint=_category_hint(value, year=year, operation=operation))
for value in spending_category_tokens(effective_date=date(year, 12, 31), authority=operation)
)
return PromptSpec(
classifications=classifications or default_classification_choices(),
categories=category_choices,
iva_categories=default_iva_category_choices(operation=operation),
)
def _category_hint(value: SpendingCategory, *, year: int, operation: PinnedAuthorityOperation) -> str:
"""Return the best available hint string for a SpendingCategory in ``year``.
Resolves the display label and ``notes`` translation keys to Spanish at
read time, and pairs them with the proportionality kind the category
profile registry declares for ``year`` -- gives the
LLM the authoritative AEAT terminology AND the deductibility
context (e.g. ``full_deductible``, ``usage_ratio_home_area``) that
disambiguates home-office from premises rent or drives MIXED vs
BUSINESS decisions. Falls back to the humanised enum value when a
category has no registered profile.
Resolution belongs here rather than in the profile registry, whose
loader is cached: resolving there would bake one operator's locale
into the shared profile and serve it to the next operator.
"""
profile = resolve_category_profiles(year, operation=operation).get(value)
if profile is None:
return value.value.replace("_", " ")
# Pinned to Spanish regardless of operator locale: the classifier reasons
# over Spanish AEAT invoices, so authoritative AEAT terminology must reach
# the prompt even when the operator has selected en/ca/hu.
spanish_label = _tr(profile.display_label, locale="es")
rule = profile.proportionality
notes = _tr(rule.notes, locale="es") if rule.notes else ""
notes_preview = notes.strip().splitlines()[0][:120] if notes else ""
segments = [spanish_label, f"[{rule.kind.value}]"]
if notes_preview:
segments.append(notes_preview)
return " — ".join(segments)
def _render_choices(lines: Iterable[tuple[str, str]]) -> str:
"""Render ``(value, hint)`` pairs as aligned bullet rows."""
rows = list(lines)
if not rows:
return ""
width = max(len(value) for value, _hint in rows)
return "\n".join(f" {value:<{width}} — {hint}" for value, hint in rows)
def _evidence_section(evidence_text: str) -> list[str]:
"""Render the attached-evidence block (selection-only; never emit its numbers)."""
return [
"Attached evidence document for this transaction (read it carefully; it is "
"authoritative for what was purchased). Use it ONLY to choose the "
"classification, category, and iva_category. Do NOT copy or output any euro "
"amount, rate, taxable base, or IVA figure from it -- those are computed "
"elsewhere from the registry.",
"--- begin evidence ---",
evidence_text,
"--- end evidence ---",
"",
]
def _vision_evidence_section() -> list[str]:
"""Render the attached-image evidence instruction (selection-only; never emit numbers).
Used when the evidence is a scanned or image invoice read on-host by a local
vision model: the document is attached as an image rather than inlined as
text, so the prompt points the model at the attached image instead of
embedding extracted text.
"""
return [
"An invoice or receipt image is attached to this message. Read it carefully; "
"it is authoritative for what was purchased. Use it ONLY to choose the "
"classification, category, and iva_category. Do NOT copy or output any euro "
"amount, rate, taxable base, or IVA figure from it -- those are computed "
"elsewhere from the registry.",
"",
]
def _evidence_block(evidence_text: str | None, evidence_image_present: bool) -> list[str]:
"""Select the evidence instruction: inlined text, attached image, or none."""
if evidence_text:
return _evidence_section(evidence_text)
if evidence_image_present:
return _vision_evidence_section()
return []
def _append_category_prompt(
sections: list[str],
schema_fields: list[str],
spec: PromptSpec,
) -> None:
"""Append the optional spending-category selection and schema field."""
if not spec.categories:
return
category_block = _render_choices((choice.value.value, choice.hint) for choice in spec.categories)
sections.extend(
[
"",
"When classification is BUSINESS or MIXED, also pick exactly one SpendingCategory:",
category_block,
],
)
schema_fields.append('"category": "<one SpendingCategory or null>"')
def _append_iva_prompt(
sections: list[str],
schema_fields: list[str],
spec: PromptSpec,
) -> None:
"""Append the optional IVA-category selection and schema fields."""
if not spec.iva_categories:
return
iva_block = _render_choices((choice.value.value, choice.hint) for choice in spec.iva_categories)
sections.extend(
[
"",
"Also pick exactly one iva_category — the IVA situation that fits this transaction. "
"Pick the category only; do NOT compute or output any rate, base, or IVA amount.",
iva_block,
],
)
schema_fields.append('"iva_category": "<one IvaCategory or null>"')
schema_fields.append('"business_pct": <0.0-1.0 when MIXED, else null>')
def _append_evidence_prompt(
sections: list[str],
schema_fields: list[str],
*,
evidence_present: bool,
) -> None:
"""Append the optional evidence multiplicity instruction and field."""
if not evidence_present:
return
sections.extend(
[
"",
"Also judge whether the attached invoice carries MULTIPLE distinct lines at "
"different IVA rates or expense categories that should be split into separate "
"entries (so each line's deductible IVA and base-rate expense file independently). "
"Set multiple_components true only when two or more distinct rate/category lines are "
"present; set it false for a single-line, single-rate invoice.",
],
)
schema_fields.append('"multiple_components": <true|false>')
def _render_prompt_footer(spec: PromptSpec, schema_fields: list[str]) -> list[str]:
"""Return the schema/example footer for a rendered classification prompt."""
schema_line = "{" + ", ".join(schema_fields) + "}"
example_confidence = "0.85"
example_reason = "restaurante meal with a named client strongly suggests business meal"
example = f'{{"classification": "BUSINESS", "confidence": {example_confidence}, "reason": "{example_reason}"'
if spec.categories:
example += ', "category": "manutencion_dietas_nacional"'
if spec.iva_categories:
example += ', "iva_category": "domestic_general", "business_pct": null'
example += "}"
return [
"",
"Respond ONLY with a single JSON object. No prose before or after. No markdown fences.",
f"Schema: {schema_line}",
f"Example response format: {example}",
]
def _render_prompt(
spec: PromptSpec,
transaction: Transaction,
*,
evidence_text: str | None = None,
evidence_image_present: bool = False,
) -> str:
"""Build the full prompt string for one transaction against a spec.
When ``evidence_text`` is given it is inlined for the model to read. When
``evidence_image_present`` is set (and no text), the prompt instead points the
model at an attached invoice image (the on-host vision-read path).
"""
raw = transaction.raw
effective_date = raw.value_date or raw.booked_date
classification_block = _render_choices((choice.value.value, choice.hint) for choice in spec.classifications)
sections = [
spec.header,
"",
"You have all the information you need below. Do NOT ask clarifying questions. "
"Do NOT offer to help further. Pick the most likely classification from the "
"closed list and answer in ONE line of JSON, immediately, nothing else.",
"",
"Transaction:",
f" Date: {effective_date.isoformat()}",
f" Amount: {raw.amount} {raw.currency}",
f" Counterparty: {raw.counterparty or '(unknown)'}",
f" Description: {raw.description}",
"",
*_evidence_block(evidence_text, evidence_image_present),
"Classify it as exactly one of these BusinessClassification values:",
classification_block,
]
schema_fields = [
'"classification": "<one value>"',
'"confidence": <0.0-1.0>',
'"reason": "<one sentence>"',
]
_append_category_prompt(sections, schema_fields, spec)
_append_iva_prompt(sections, schema_fields, spec)
evidence_present = bool(evidence_text) or evidence_image_present
_append_evidence_prompt(sections, schema_fields, evidence_present=evidence_present)
sections.extend(_render_prompt_footer(spec, schema_fields))
return "\n".join(sections)
# ── response parsing ──────────────────────────────────────────────
_JSON_OBJECT_RE = re.compile(r"\{[^{}]*\}")
def _classification_failure(
response: LLMClassificationResponse,
payload: str,
allowed_classifications: frozenset[BusinessClassification],
) -> str | None:
"""Return the allow-list failure for the classification field, if any."""
if response.classification not in allowed_classifications:
return f"disallowed classification {response.classification.value!r} (payload {payload[:100]!r})"
return None
def _category_failure(
response: LLMClassificationResponse,
payload: str,
allowed_categories: frozenset[SpendingCategory],
) -> str | None:
"""Return the allow-list failure for an optional spending category, if any."""
if response.category is None:
return None
if not allowed_categories:
return f"unexpected category {response.category.value!r} (payload {payload[:100]!r})"
if response.category not in allowed_categories:
return f"disallowed category {response.category.value!r} (payload {payload[:100]!r})"
return None
def _iva_category_failure(
response: LLMClassificationResponse,
payload: str,
allowed_iva_categories: frozenset[IvaCategory],
) -> str | None:
"""Return the allow-list failure for an optional IVA category, if any."""
if response.iva_category is None:
return None
if not allowed_iva_categories:
return f"unexpected iva_category {response.iva_category.value!r} (payload {payload[:100]!r})"
if response.iva_category not in allowed_iva_categories:
return f"disallowed iva_category {response.iva_category.value!r} (payload {payload[:100]!r})"
return None
def _classification_response_failure(
response: LLMClassificationResponse,
payload: str,
*,
allowed_classifications: frozenset[BusinessClassification],
allowed_categories: frozenset[SpendingCategory],
allowed_iva_categories: frozenset[IvaCategory],
) -> str | None:
"""Return the first allow-list failure in the parser's field order."""
failure = _classification_failure(response, payload, allowed_classifications)
if failure is not None:
return failure
failure = _category_failure(response, payload, allowed_categories)
if failure is not None:
return failure
return _iva_category_failure(response, payload, allowed_iva_categories)
def _parse_classification_candidate(
payload: str,
*,
allowed_classifications: frozenset[BusinessClassification],
allowed_categories: frozenset[SpendingCategory],
allowed_iva_categories: frozenset[IvaCategory],
) -> LLMClassificationResponse | str:
"""Validate one candidate and return either its response or its failure text."""
try:
response = LLMClassificationResponse.model_validate_json(payload)
except ValueError as exc:
return f"schema: {str(exc)[:160]} (payload {payload[:100]!r})"
failure = _classification_response_failure(
response,
payload,
allowed_classifications=allowed_classifications,
allowed_categories=allowed_categories,
allowed_iva_categories=allowed_iva_categories,
)
return response if failure is None else failure
[docs]
def parse_response(
stdout: str,
*,
spec: PromptSpec | None = None,
) -> LLMClassificationResponse:
"""Extract a valid JSON object from LLM stdout, validate, enforce spec.
Iterates every JSON-object candidate in the output and returns the
first one that validates against the schema AND passes the spec's
allow-list. Guards against an LLM that echoes a prompt-injected
JSON block before emitting its real answer: a malformed or
disallowed first candidate no longer poisons the result.
Args:
stdout: Raw stdout captured from the LLM CLI.
spec: Prompt spec the response should conform to. When provided,
reject classifications or categories outside the allow-list.
Returns:
A validated :class:`LLMClassificationResponse`.
Raises:
LLMClassifierError: If no candidate JSON object exists, or if
none of the candidates passes both schema validation and
the spec's allow-list.
"""
resolved_spec = spec or default_prompt_spec()
allowed_classifications = resolved_spec.allowed_classifications()
allowed_categories = resolved_spec.allowed_categories()
allowed_iva_categories = resolved_spec.allowed_iva_categories()
failures: list[str] = []
any_candidate_seen = False
for match in _JSON_OBJECT_RE.finditer(stdout):
any_candidate_seen = True
payload = match.group(0)
candidate = _parse_classification_candidate(
payload,
allowed_classifications=allowed_classifications,
allowed_categories=allowed_categories,
allowed_iva_categories=allowed_iva_categories,
)
if isinstance(candidate, LLMClassificationResponse):
return candidate
failures.append(candidate)
if not any_candidate_seen:
raise LLMClassifierError(f"no JSON object in LLM output: {stdout[:400]!r}")
raise LLMClassifierError(
f"no JSON candidate matched schema + spec; tried {len(failures)}: " + "; ".join(failures[:3]),
)
[docs]
def build_split_prompt(
transaction: Transaction,
*,
spec: PromptSpec | None = None,
evidence_text: str | None = None,
evidence_image_present: bool = False,
) -> str:
"""Build a prompt asking the model to propose an evidence-driven N-way split.
The model reads the attached invoice and proposes per-child *proportions* plus
selections; it must never emit a euro amount (the application derives the
amounts from the parent gross and the tax substrate from the registry). The
invoice is supplied as inlined text (``evidence_text``) or, on the on-host
vision-read path, as an attached image (``evidence_image_present``).
"""
resolved_spec = spec or default_prompt_spec()
raw = transaction.raw
effective_date = (raw.value_date or raw.booked_date).isoformat()
sections = [
"You are dividing one Spanish autonomo bank transaction into the lines of its attached invoice.",
"",
"Transaction:",
f" Date: {effective_date}",
f" Amount: {raw.amount} {raw.currency}",
f" Counterparty: {raw.counterparty or '(unknown)'}",
f" Description: {raw.description}",
"",
*_evidence_block(evidence_text, evidence_image_present),
"Propose how to divide this transaction into children, one per distinct line or category on "
"the invoice. If the invoice is a SINGLE line at a single IVA rate (no split warranted), "
"return EXACTLY ONE child with proportion 1.0. If it carries two or more distinct lines or "
"IVA rates, return one child per line. For each child give a proportion (a fraction of the "
"total; all proportions MUST sum to 1.0), a spending category, an iva_category, and a short "
"evidence_citation naming the line. Do NOT output any euro amount, rate, base, or IVA figure.",
]
if resolved_spec.categories:
category_block = _render_choices((choice.value.value, choice.hint) for choice in resolved_spec.categories)
sections.extend(["", "Pick each child's spending category from:", category_block])
if resolved_spec.iva_categories:
iva_block = _render_choices((choice.value.value, choice.hint) for choice in resolved_spec.iva_categories)
sections.extend(["", "Pick each child's iva_category from:", iva_block])
sections.extend(
[
"",
"Respond ONLY with a single JSON object. No prose before or after. No markdown fences.",
'Schema: {"reason": "<one sentence>", "children": [{"proportion": <0..1>, '
'"category": "<SpendingCategory or null>", "iva_category": "<IvaCategory or null>", '
'"evidence_citation": "<short>"}, ...]}',
]
)
return "\n".join(sections)
def _advance_json_string_state(char: str, escaped: bool) -> tuple[bool, bool]:
"""Consume one character while scanning a JSON string."""
if escaped:
return True, False
if char == "\\":
return True, True
return char != '"', False
def _advance_json_structure_state(char: str, depth: int) -> tuple[int, bool, bool]:
"""Consume one non-string character and report a completed object."""
if char == '"':
return depth, True, False
if char == "{":
return depth + 1, False, False
if char == "}":
return depth - 1, False, depth == 1
return depth, False, False
def _extract_json_object(text: str) -> str | None:
"""Return the first balanced top-level JSON object substring, or ``None``.
Unlike the flat :data:`_JSON_OBJECT_RE`, this walks brace depth (ignoring
braces inside strings) so a nested object -- such as a split proposal whose
``children`` is an array of objects -- is captured whole.
"""
start = text.find("{")
if start < 0:
return None
depth = 0
in_string = False
escaped = False
for index in range(start, len(text)):
char = text[index]
if in_string:
in_string, escaped = _advance_json_string_state(char, escaped)
continue
depth, in_string, complete = _advance_json_structure_state(char, depth)
if complete:
return text[start : index + 1]
return None
def _split_child_failure(
child: LLMSplitChild,
allowed_categories: frozenset[SpendingCategory],
allowed_iva_categories: frozenset[IvaCategory],
) -> str | None:
"""Return the first split-child allow-list failure, if any."""
if child.category is not None and (not allowed_categories or child.category not in allowed_categories):
return f"disallowed split child category {child.category.value!r}"
if child.iva_category is not None and (
not allowed_iva_categories or child.iva_category not in allowed_iva_categories
):
return f"disallowed split child iva_category {child.iva_category.value!r}"
return None
def _parse_split_payload(payload: str) -> LLMSplitResponse:
"""Validate the schema portion of one split-response JSON payload."""
try:
return LLMSplitResponse.model_validate_json(payload)
except ValueError as exc:
raise LLMClassifierError(f"split response failed schema validation: {str(exc)[:200]}") from exc
[docs]
def parse_split_response(stdout: str, *, spec: PromptSpec | None = None) -> LLMSplitResponse:
"""Extract and validate an N-way split proposal from LLM stdout.
Finds the first balanced JSON object, validates it as
:class:`LLMSplitResponse`, and rejects any child whose ``category`` or
``iva_category`` falls outside the spec's allow-list -- the same hallucination
guard :func:`parse_response` applies to a flat classification.
Args:
stdout: Raw stdout captured from the LLM CLI.
spec: Prompt spec whose allow-lists each child must satisfy.
Returns:
A validated :class:`LLMSplitResponse`.
Raises:
LLMClassifierError: When no JSON object is present, the schema is
violated, or a child selection is outside the allow-list.
"""
resolved_spec = spec or default_prompt_spec()
allowed_categories = resolved_spec.allowed_categories()
allowed_iva_categories = resolved_spec.allowed_iva_categories()
payload = _extract_json_object(stdout)
if payload is None:
raise LLMClassifierError(f"no JSON object in LLM split output: {stdout[:400]!r}")
response = _parse_split_payload(payload)
for child in response.children:
failure = _split_child_failure(child, allowed_categories, allowed_iva_categories)
if failure is not None:
raise LLMClassifierError(failure)
return response
__all__ = [
"MINIMUM_CLASSIFICATION_TIER",
"CategoryChoice",
"ClassificationChoice",
"IvaCategoryChoice",
"LLMClassificationResponse",
"LLMClassifier",
"LLMClassifierError",
"ModelProfile",
"ModelTier",
"PromptSpec",
"default_classification_choices",
"default_iva_category_choices",
"default_prompt_spec",
"parse_response",
"prompt_spec_with_every_spending_category",
"prompt_spec_with_saturation_fields",
]