aeat.domain.transactions._llm module

LLM-backed transaction classifiers with a parametric prompt builder.

Defines the LLMClassifier protocol plus subprocess-based reference implementations for the three local LLM CLIs (build_claude_classifier(), build_antigravity_classifier(), build_codex_classifier()). The prompt is built PROGRAMMATICALLY from the available enum values so the LLM prompt stays in sync with aeat.domain.transactions.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 aeat.domain.transactions.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 aeat.domain.categories.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 parse_response(): a response that picks a value outside the allowed set raises LLMClassifierError, so a hallucinating model cannot corrupt the catalogue.

class LLMClassificationResponse(**data)[source]

Bases: BaseModel

One LLM-emitted classification result for a transaction.

Parameters:
classification: BusinessClassification
confidence: Decimal
reason: str
category: SpendingCategory | None
iva_category: IvaCategory | None
business_pct: Decimal | None
multiple_components: bool | None

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.

Type:

Evidence-read multiplicity judgement

class LLMSplitChild(**data)[source]

Bases: 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).

Parameters:
proportion: Decimal
category: SpendingCategory | None
iva_category: IvaCategory | None
evidence_citation: str
class LLMSplitResponse(**data)[source]

Bases: 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.

Parameters:
children: tuple[LLMSplitChild, ...]
reason: str
property recommends_split: bool

True when the model proposes more than one child (a genuine split).

class LLMClassifier(*args, **kwargs)[source]

Bases: Protocol

Classify one transaction with an LLM-generated decision.

property decided_by: str

Return the classified_by identifier this classifier emits.

classify(transaction, *, evidence_text=None)[source]

Return one classification decision for transaction.

Parameters:

  • transaction (Transaction) – The transaction to classify.

  • evidence_text (str | None) – 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.

Return type:

LLMClassificationResponse

Returns:

A LLMClassificationResponse with the classification result.

class LLMSplitProposer(*args, **kwargs)[source]

Bases: Protocol

Propose an evidence-driven N-way split for one transaction.

property decided_by: str

Return the classified_by identifier this proposer emits.

propose_split(transaction, *, evidence_text=None)[source]

Return an N-way split proposal for transaction.

Parameters:

  • transaction (Transaction) – The transaction to split.

  • evidence_text (str | None) – Optional on-host-extracted attached-evidence text to inject into the prompt.

Return type:

LLMSplitResponse

Returns:

A validated LLMSplitResponse.

class ClassificationChoice(value, hint)[source]

Bases: object

One allowed BusinessClassification paired with an LLM-facing hint.

Parameters:
value: BusinessClassification
hint: str
class CategoryChoice(value, hint)[source]

Bases: object

One allowed SpendingCategory paired with an LLM-facing hint.

Parameters:
value: SpendingCategory
hint: str
class IvaCategoryChoice(value, hint)[source]

Bases: object

One allowed aeat.domain.iva.IvaCategory paired with an LLM-facing hint.

Parameters:
value: IvaCategory
hint: str
default_classification_choices()[source]

Return the default allowed-classifications tuple used by the prompt.

Return type:

tuple[ClassificationChoice, ...]

Returns:

Tuple of ClassificationChoice objects for each allowed category.

class PromptSpec(classifications=<factory>, categories=(), iva_categories=(), header="You are classifying a Spanish autónomo's bank transaction for tax purposes.")[source]

Bases: object

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 parse_response() regardless of how well-formed the JSON is.

Parameters:
classifications: tuple[ClassificationChoice, ...]
categories: tuple[CategoryChoice, ...]
iva_categories: tuple[IvaCategoryChoice, ...]
header: str
allowed_classifications()[source]

Return the set of BusinessClassification values the LLM is allowed to emit.

Return type:

frozenset[BusinessClassification]

allowed_categories()[source]

Return the set of category values the LLM is allowed to emit (empty = none).

Return type:

frozenset[SpendingCategory]

Returns:

Frozenset of SpendingCategory values the LLM may emit.

allowed_iva_categories()[source]

Return the set of aeat.domain.iva.IvaCategory values the LLM may emit (empty = none).

Return type:

frozenset[IvaCategory]

Returns:

Frozenset of aeat.domain.iva.IvaCategory values the LLM may select from; empty when the spec does not ask for an IVA category.

render(transaction, *, evidence_text=None, evidence_image_present=False)[source]

Render the prompt for transaction against this spec.

Parameters:

  • transaction (Transaction) – The transaction to classify.

  • evidence_text (str | None) – 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 (bool) – 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 type:

str

default_prompt_spec()[source]

Return the default PromptSpec: classification-only, four decision states.

Return type:

PromptSpec

prompt_spec_with_every_spending_category(*, classifications=None)[source]

Return a prompt spec that also asks the LLM to suggest a SpendingCategory.

Pulls authoritative Spanish display labels from aeat.domain.categories.resolve_category_profiles(2025) 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 aeat.domain.categories.SpendingCategory member is covered) fall back to the humanised enum value.

Parameters:

classifications (tuple[ClassificationChoice, ...] | None) – Optional override for the classification choices; defaults to default_classification_choices().

Return type:

PromptSpec

Returns:

A PromptSpec whose categories tuple covers every registered aeat.domain.categories.SpendingCategory.

default_iva_category_choices()[source]

Return the grounded IVA-category choices for the saturation prompt.

The allow-list is the closed aeat.domain.iva.IvaCategory enum (the registry aeat.domain.iva.IvaCatalogue is validated to carry a regulation for every member, so the enum and the catalogue set are identical). Each choice is hinted with a concise description from _IVA_CATEGORY_HINTS. The model SELECTS a category only; every regulated euro figure is derived downstream from the registry rate, never emitted by the model (2026-06-04-llm-ledger-classification-adr).

Return type:

tuple[IvaCategoryChoice, ...]

Returns:

One IvaCategoryChoice per aeat.domain.iva.IvaCategory, ordered by enum declaration.

prompt_spec_with_saturation_fields(*, classifications=None)[source]

Return a prompt spec for full saturation: spending + IVA category selection.

Extends 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 (2026-06-04-llm-ledger-classification-adr).

Parameters:

classifications (tuple[ClassificationChoice, ...] | None) – Optional override for the classification choices; defaults to default_classification_choices().

Return type:

PromptSpec

Returns:

A PromptSpec carrying both the spending-category and the IVA-category allow-lists.

parse_response(stdout, *, spec=None)[source]

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.

Parameters:

  • stdout (str) – Raw stdout captured from the LLM CLI.

  • spec (PromptSpec | None) – Prompt spec the response should conform to. When provided, reject classifications or categories outside the allow-list.

Return type:

LLMClassificationResponse

Returns:

A validated 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.

build_split_prompt(transaction, *, spec=None, evidence_text=None, evidence_image_present=False)[source]

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).

Return type:

str

Parameters:
parse_split_response(stdout, *, spec=None)[source]

Extract and validate an N-way split proposal from LLM stdout.

Finds the first balanced JSON object, validates it as LLMSplitResponse, and rejects any child whose category or iva_category falls outside the spec’s allow-list – the same hallucination guard parse_response() applies to a flat classification.

Parameters:

  • stdout (str) – Raw stdout captured from the LLM CLI.

  • spec (PromptSpec | None) – Prompt spec whose allow-lists each child must satisfy.

Return type:

LLMSplitResponse

Returns:

A validated LLMSplitResponse.

Raises:

LLMClassifierError – When no JSON object is present, the schema is violated, or a child selection is outside the allow-list.

class SubprocessLLMClassifier(name, command, model=None, timeout_seconds=120.0, spec=<factory>, prompt_via_argument=False)[source]

Bases: object

LLM classifier that shells out to a local CLI binary.

Pipes the prompt via stdin by default (more reliable than a positional argument for long multi-line prompts, especially on Windows where CreateProcess quoting can corrupt arguments).

Reads output from stdout. Transaction prompts and classifier responses are sensitive financial data, so this adapter deliberately avoids file-backed subprocess handoff.

Set prompt_via_argument=True for CLIs that reject stdin and require the prompt as the final positional argument.

Parameters:
name: str
command: tuple[str, ...]
model: str | None
timeout_seconds: float
spec: PromptSpec
prompt_via_argument: bool
property decided_by: str

Return the classified_by identifier including the model when set.

classify(transaction, *, evidence_text=None)[source]

Shell out to the LLM CLI, parse, validate, return.

Parameters:

  • transaction (Transaction) – The transaction to classify.

  • evidence_text (str | None) – Optional on-host-extracted attached-evidence text injected into the prompt (sent via stdin, never a file).

Return type:

LLMClassificationResponse

Returns:

A LLMClassificationResponse with the parsed classification result.

propose_split(transaction, *, evidence_text=None)[source]

Shell out with a split-proposal prompt, parse and validate the split.

Parameters:

  • transaction (Transaction) – The transaction to split.

  • evidence_text (str | None) – Optional on-host-extracted attached-evidence text injected into the prompt (sent via stdin, never a file).

Return type:

LLMSplitResponse

Returns:

A validated LLMSplitResponse.

build_claude_classifier(*, alias=None, model=None, spec=None, minimum_tier=ModelTier.MEDIUM)[source]

Build a classifier that shells out to claude --bare -p.

Parameters:
Return type:

SubprocessLLMClassifier

Returns:

A SubprocessLLMClassifier configured for the claude CLI.

build_antigravity_classifier(*, alias=None, model=None, spec=None, minimum_tier=ModelTier.MEDIUM)[source]

Build a classifier that shells out to agy --prompt <prompt>.

Antigravity (Google’s agentic CLI agy) is the supported successor to the retired standalone gemini CLI. Its --print / -p / --prompt mode runs a single prompt non-interactively; the prompt is the VALUE of that flag, so it is passed as the final positional argument (prompt_via_argument=True). Unlike the old gemini CLI (a Node wrapper whose command line overflowed a ~8 KB limit on the larger saturation prompt), agy is a native binary invoked through the subprocess argument list, so it carries the full platform command-line budget. --model selects a model when one is pinned; otherwise agy uses its own current default. Its stdout may carry start-up noise; the JSON iterator in parse_response() tolerates it.

Parameters:

  • alias (str | None) – Capability-tier alias (antigravity-default). Enforces minimum_tier.

  • model (str | None) – Explicit provider-specific model override.

  • spec (PromptSpec | None) – Prompt spec override.

  • minimum_tier (ModelTier) – Refuses aliases below this tier.

Return type:

SubprocessLLMClassifier

Returns:

A SubprocessLLMClassifier configured for the agy CLI with prompt_via_argument=True.

build_codex_classifier(*, alias=None, model=None, spec=None, minimum_tier=ModelTier.MEDIUM)[source]

Build a classifier that shells out to codex exec.

Uses --ephemeral + --skip-git-repo-check so the invocation does not require a git repo and does not persist sessions. The The subprocess adapter parses JSON candidates from stdout so no transaction data is written to a temporary file.

Parameters:

  • alias (str | None) – Capability-tier alias (codex-default / codex-o3). Enforces minimum_tier.

  • model (str | None) – Explicit provider-specific model override.

  • spec (PromptSpec | None) – Prompt spec override.

  • minimum_tier (ModelTier) – Refuses aliases below this tier.

Return type:

SubprocessLLMClassifier

Returns:

A SubprocessLLMClassifier configured for the codex CLI.

resolve_classifier(provider, *, alias=None, model=None, spec=None, minimum_tier=ModelTier.MEDIUM)[source]

Return a classifier for the given provider name.

Parameters:
Return type:

LLMClassifier

Returns:

A concrete implementation of LLMClassifier.

Raises:

LLMClassifierError – If provider is not a registered builder, or if tier resolution fails.

resolve_split_proposer(provider, *, spec=None)[source]

Resolve the production split proposer for provider.

Resolves the provider’s classifier (via resolve_classifier()) and narrows it to an LLMSplitProposer. The subprocess classifiers are the only production proposers; a registered classifier that does not also propose splits is refused instructively.

Parameters:

  • provider (str) – One of "claude", "antigravity", "codex".

  • spec (PromptSpec | None) – Optional prompt spec override (the category + IVA-category saturation spec is the natural choice so split children carry the same allow-list-guarded selections).

Return type:

LLMSplitProposer

Returns:

An LLMSplitProposer for provider.

Raises:

LLMClassifierError – If provider resolves to a classifier that does not support evidence-driven splitting.

register_classifier(name, builder)[source]

Register a classifier builder under name.

Intended for tests that want to inject a concrete in-process classifier (no mocks) and for third-party extensions that want to add a new provider without monkey-patching the module.

Return type:

None

Parameters:
unregister_classifier(name)[source]

Remove a builder previously added via register_classifier().

Return type:

None

Parameters:

name (str)