"""LLM parser abstraction with strict Transaction schema fallback.""" from __future__ import annotations import json from collections.abc import Callable from typing import Any, Protocol from voiceledger.parser.base import Parser from voiceledger.parser.rules import RuleParser from voiceledger.parser.schema import Transaction SYSTEM_PROMPT = """You extract bookkeeping transactions for VoiceLedger. Return only strict JSON matching this schema: { "transaction_type": "sale|expense|inventory_purchase|customer_credit|customer_payment|unknown", "item": "string or null", "quantity": "number or null", "unit_price": "number or null", "amount": "number or null", "customer": "string or null", "payment_status": "paid|unpaid|credit|unknown", "notes": "original user text", "confidence": "number from 0 to 1" } Do not include markdown, prose, comments, or extra keys.""" class HuggingFaceInferenceClient(Protocol): """Minimal Hugging Face Inference API compatible client interface.""" def text_generation(self, prompt: str, **kwargs: Any) -> str: """Generate text from a prompt.""" class LLMParser(Parser): """Parse transaction text with an LLM and fall back to rule parsing on failure.""" def __init__( self, client: HuggingFaceInferenceClient | Callable[..., Any], model: str | None = None, fallback_parser: Parser | None = None, ) -> None: """Create an LLM-backed parser. The client can be a Hugging Face Inference API style object with a `text_generation()` method or a callable that accepts `prompt` and generation keyword arguments. """ self.client = client self.model = model self.fallback_parser = fallback_parser or RuleParser() def parse(self, text: str) -> Transaction: """Parse text with strict JSON generation and schema validation.""" try: response = self._generate(text) payload = _extract_json_object(response) transaction = Transaction.model_validate(payload) return transaction.model_copy(update={"notes": transaction.notes or text.strip()}) except Exception: return self.fallback_parser.parse(text) def _generate(self, text: str) -> str: """Call the configured Hugging Face compatible generation client.""" prompt = _build_prompt(text) generation_kwargs: dict[str, Any] = { "max_new_tokens": 256, "temperature": 0.0, "return_full_text": False, } if self.model is not None: generation_kwargs["model"] = self.model if hasattr(self.client, "text_generation"): response = self.client.text_generation(prompt, **generation_kwargs) else: response = self.client(prompt=prompt, **generation_kwargs) return _coerce_generation_text(response) def _build_prompt(text: str) -> str: """Build a strict JSON extraction prompt.""" return f"{SYSTEM_PROMPT}\n\nUser text: {text.strip()}\nJSON:" def _extract_json_object(response: str) -> dict[str, Any]: """Extract and parse the first JSON object from generated text.""" start = response.find("{") end = response.rfind("}") if start == -1 or end == -1 or end < start: raise ValueError("LLM response did not contain a JSON object.") payload = json.loads(response[start : end + 1]) if not isinstance(payload, dict): raise ValueError("LLM response JSON must be an object.") return payload def _coerce_generation_text(response: Any) -> str: """Normalize common Hugging Face generation response shapes to text.""" if isinstance(response, str): return response if isinstance(response, list) and response: first = response[0] if isinstance(first, dict): return str(first.get("generated_text", "")) if isinstance(response, dict): return str(response.get("generated_text", response.get("text", ""))) raise ValueError("Unsupported generation response shape.")