"""Strict Pydantic models for the outbound LLM adapter.
This module is the canonical owner of request, response, provider, cache,
usage, prompt, and validation record types. Consumers import these records
directly from this defining module; the sibling package initializer is inert.
"""
from __future__ import annotations
import base64
import re
from collections.abc import Callable
from datetime import date, datetime
from decimal import Decimal
from functools import wraps
from pydantic import BaseModel, ConfigDict, Field, field_validator
from ....core.config_support import LLMProvider
from ....core.errors.hierarchy import pydantic_validation_boundary
from ....core.hashing import sha256_hex
from ....core.identity.digest import ContentDigest
from ....core.image_media_type import ImageMediaType
from ....core.models import STRICT_FROZEN_CONFIG
from ....core.operator_action_enums import ActionEvidenceProvenance
from .consent import EvidenceConsentToken
from .errors import LLMValidationError
from .preconditions import LLMPreconditionCondition, llm_no_recovery_verdict
_PROMPT_ID_PATTERN = re.compile(r"^[a-z0-9]+(?:[-_][a-z0-9]+)*$")
def _pydantic_llm_validator[ValidationResultT](
function: Callable[..., ValidationResultT],
) -> Callable[..., ValidationResultT]:
"""Translate the registered LLM refusal at Pydantic's callback edge."""
@wraps(function)
def wrapped(*args: object, **kwargs: object) -> ValidationResultT:
try:
return function(*args, **kwargs)
except LLMValidationError as exc:
raise ValueError(str(exc)) from exc
return wrapped
[docs]
class LLMRequest(BaseModel):
"""User-facing completion request accepted by :class:`~llm.LLMClient`.
Provider and model override fields select
:class:`~llm.LLMProvider` values for one call, and
``images`` carries transient
:class:`~llm.MultimodalImageInput` payloads for
local vision flows.
``evidence_derived`` defaults to ``False`` because most requests carry no
taxpayer content at all -- a column-role mapping reads spreadsheet headers,
a corpus measurement reads a public synthetic document. The fail-closed
default lives one level up, at the evidence READERS, which mark every
request they build unless the caller names a public corpus: an unmarked
request is therefore a deliberate statement about the content, not an
omission. ``consent_token`` enters neither the cache key nor any persisted
record; it exists only between its minting site and the dispatch point.
"""
model_config = STRICT_FROZEN_CONFIG
prompt: str = Field(description="Prompt content passed to the provider.")
system: str | None = Field(default=None, description="Optional system instruction.")
max_tokens: int | None = Field(default=None, ge=1, description="Maximum output tokens to request.")
temperature: float | None = Field(default=None, ge=0.0, le=1.0, description="Sampling temperature.")
language: str | None = Field(default=None, description="ISO 639-1 language code for the requested output.")
cache_key: str | None = Field(default=None, description="Optional cache grouping key.")
provider_override: LLMProvider | None = Field(default=None, description="Override the configured provider.")
model_override: str | None = Field(default=None, min_length=1, description="Override the configured model.")
images: tuple[MultimodalImageInput, ...] = Field(
default=(),
description="On-host-prepared multimodal image inputs (empty for a text-only request).",
)
evidence_derived: bool = Field(
default=False,
description=(
"Whether this request's content derives from a taxpayer's evidence document. "
"Marked requests may not be dispatched off-host without a consent token."
),
)
consent_token: EvidenceConsentToken | None = Field(
default=None,
exclude=True,
description="Per-invocation off-host consent proof; never persisted and never cached.",
)
[docs]
@field_validator("prompt")
@classmethod
@_pydantic_llm_validator
@pydantic_validation_boundary
def validate_prompt(cls, value: str) -> str:
"""Ensure prompts are not empty or whitespace-only.
Raises:
:exc:`~llm.LLMValidationError`: When the
prompt is blank after trimming.
"""
normalized = value.strip()
if not normalized:
raise LLMValidationError(
context={"request_prompt_nonempty": False},
precondition_verdict=llm_no_recovery_verdict(
LLMPreconditionCondition.REQUEST_PROMPT_NONEMPTY,
facts={"request_prompt_nonempty": False},
provenance=ActionEvidenceProvenance.APPLICATION_STATE,
),
)
return normalized
[docs]
@field_validator("system")
@classmethod
@pydantic_validation_boundary
def validate_system(cls, value: str | None) -> str | None:
"""Normalize empty system prompts to ``None``."""
if value is None:
return None
normalized = value.strip()
return normalized or None
[docs]
@field_validator("language")
@classmethod
@pydantic_validation_boundary
def validate_language(cls, value: str | None) -> str | None:
"""Validate optional ISO 639-1 language codes."""
if value is None:
return None
return value
[docs]
class LLMResponse(BaseModel):
"""Completion response returned by :meth:`~llm.LLMClient.complete`.
Responses are persisted inside
:class:`~llm.CachedEntry` records and converted into
:class:`~llm.UsageRecord` values for cost tracking.
"""
model_config = STRICT_FROZEN_CONFIG
text: str = Field(description="Generated text returned by the provider.")
provider: LLMProvider = Field(description="Provider that produced the response.")
model: str = Field(description="Resolved model identifier.")
input_tokens: int = Field(ge=0, description="Prompt-side token count.")
output_tokens: int = Field(ge=0, description="Completion-side token count.")
cost_estimate_usd: Decimal | None = Field(
description=(
"Estimated call cost in USD, or None when no pricing entry covers this "
"provider and model. None is UNPRICED and Decimal('0') is FREE; a surface "
"that renders them alike reports an absence as a positive answer."
),
)
cache_hit: bool = Field(description="Whether the response came from the local cache.")
created_at: datetime = Field(description="Creation timestamp in UTC.")
request_id: str = Field(description="Stable hash of the public request.")
[docs]
class PromptDefinition(BaseModel):
"""Prompt metadata stored by :class:`~llm.PromptRegistry`."""
# Diverges from STRICT_FROZEN_CONFIG deliberately: a prompt definition carries a
# compiled template object pydantic cannot describe, so arbitrary types are permitted.
model_config = ConfigDict(strict=True, frozen=True, arbitrary_types_allowed=True)
id: str = Field(description="Stable kebab-case prompt identifier.")
version: int = Field(ge=1, description="Prompt version.")
template: str = Field(description="Renderable template text.")
expected_output_schema: type[BaseModel] | None = Field(
default=None,
description="Optional structured response schema for downstream validation.",
)
description: str = Field(description="Human-readable prompt purpose.")
[docs]
@field_validator("id")
@classmethod
@_pydantic_llm_validator
@pydantic_validation_boundary
def validate_id(cls, value: str) -> str:
"""Ensure prompt identifiers are kebab-case."""
if not _PROMPT_ID_PATTERN.fullmatch(value):
raise LLMValidationError(
context={"prompt_definition_id_valid": False},
precondition_verdict=llm_no_recovery_verdict(
LLMPreconditionCondition.PROMPT_DEFINITION_ID_VALID,
facts={"prompt_definition_id_valid": False},
provenance=ActionEvidenceProvenance.APPLICATION_STATE,
),
)
return value
[docs]
class PromptRegistry(BaseModel):
"""Registry of versioned :class:`~llm.PromptDefinition` values."""
# Diverges from STRICT_FROZEN_CONFIG deliberately: this registry exposes a
# ``register`` mutator, so declaring it frozen would advertise an immutability
# the class does not have.
model_config = STRICT_FROZEN_CONFIG
definitions: dict[str, PromptDefinition] = Field(
default_factory=dict,
description="Mapping of ``prompt-id:vN`` keys to prompt definitions.",
)
[docs]
def register(self, definition: PromptDefinition) -> None:
"""Add or replace a prompt definition."""
self.definitions[self._composite_key(definition.id, definition.version)] = definition
[docs]
def get(self, prompt_id: str, version: int | None = None) -> PromptDefinition:
"""Return a :class:`~llm.PromptDefinition` by id and optional version."""
if version is not None:
return self.definitions[self._composite_key(prompt_id, version)]
candidates = [item for item in self.definitions.values() if item.id == prompt_id]
if not candidates:
raise KeyError(prompt_id)
return max(candidates, key=lambda item: item.version)
[docs]
@classmethod
def seeded(cls) -> PromptRegistry:
"""Return a default :class:`~llm.PromptRegistry`."""
registry = cls()
registry.register(
PromptDefinition(
id="translation_v1",
version=1,
template=(
"Translate the following AEAT-related text from {source_lang} to {target_lang}. "
"Preserve legal meaning, tax terminology, and numbers exactly.\n\n"
"Context:\n{context}\n\n"
"Text:\n{text}"
),
expected_output_schema=None,
description="Faithful legal- and tax-aware translation prompt.",
),
)
registry.register(
PromptDefinition(
id="casilla_extract_v1",
version=1,
template=("Extract structured casilla information from the supplied source text.\n\nSource:\n{text}"),
expected_output_schema=None,
description="Placeholder seed for casilla extraction workflows.",
),
)
registry.register(
PromptDefinition(
id="manual_rule_extract_v1",
version=1,
template=("Extract structured tax rules from the supplied manual excerpt.\n\nManual excerpt:\n{text}"),
expected_output_schema=None,
description="Placeholder seed for manual rule extraction workflows.",
),
)
return registry
@staticmethod
def _composite_key(prompt_id: str, version: int) -> str:
"""Build the storage key for a versioned prompt."""
return f"{prompt_id}:v{version}"
[docs]
class CachedEntry(BaseModel):
"""Encrypted cache record persisted by :class:`~adapters.persistence.llm.cache.LLMCache`."""
model_config = STRICT_FROZEN_CONFIG
provider: LLMProvider = Field(description="Provider used for the original call.")
model: str = Field(description="Resolved provider model.")
prompt_hash: str = Field(description="Hash of rendered prompt text.")
args_hash: str = Field(description="Hash of request-side arguments.")
response: LLMResponse = Field(description="Original response payload.")
created_at: datetime = Field(description="Timestamp when the cache entry was written.")
[docs]
class UsageRecord(BaseModel):
"""Append-only usage record persisted by :class:`~adapters.persistence.llm.usage.UsageRecorder`."""
model_config = STRICT_FROZEN_CONFIG
prompt_id: str = Field(description="Prompt id associated with the call.")
caller: str = Field(description="Logical caller that initiated the request.")
text: str = Field(description="Generated text returned by the provider.")
provider: LLMProvider = Field(description="Provider used for the call.")
model: str = Field(description="Resolved model identifier.")
input_tokens: int = Field(ge=0, description="Prompt-side token count.")
output_tokens: int = Field(ge=0, description="Completion-side token count.")
cost_estimate_usd: Decimal | None = Field(
description="Estimated call cost in USD, or None when the model carries no pricing entry.",
)
cache_hit: bool = Field(description="Whether the response came from cache.")
created_at: datetime = Field(description="Timestamp when the record was written.")
request_id: str = Field(description="Stable request hash.")
[docs]
class CacheKey(BaseModel):
"""Derived cache key used by :class:`~adapters.persistence.llm.cache.LLMCache`."""
model_config = STRICT_FROZEN_CONFIG
provider: LLMProvider = Field(description="Provider namespace for the cache entry.")
model: str = Field(description="Resolved model namespace for the cache entry.")
prompt_hash: str = Field(description="Hash of rendered prompt text.")
args_hash: str = Field(description="Hash of request arguments.")
[docs]
class CacheStats(BaseModel):
"""Basic :class:`~adapters.persistence.llm.cache.LLMCache` statistics for CLI reporting."""
model_config = STRICT_FROZEN_CONFIG
entries: int = Field(ge=0, description="Number of cached files.")
total_bytes: int = Field(ge=0, description="Total size of cache files in bytes.")
[docs]
class UsageSummary(BaseModel):
"""Aggregated :class:`~adapters.persistence.llm.usage.UsageRecorder` statistics."""
model_config = STRICT_FROZEN_CONFIG
entries: int = Field(ge=0, description="Number of usage records included.")
total_input_tokens: int = Field(ge=0, description="Sum of input tokens.")
total_output_tokens: int = Field(ge=0, description="Sum of output tokens.")
total_cost_estimate_usd: Decimal | None = Field(
description=(
"Sum of estimated cost in USD, or None when ANY included record is unpriced. "
"A total that silently skipped unpriced rows would understate the bill while "
"looking complete, which is the same defect one layer along."
),
)
unpriced_entries: int = Field(ge=0, default=0, description="Records carrying no cost estimate.")
since: date | None = Field(default=None, description="Inclusive lower date bound.")
until: date | None = Field(default=None, description="Inclusive upper date bound.")