dodil/erp-brick-fixed-assets

DODIL ERP brick: fixed assets

src/brick_fixed_assets/infrastructure/ai_steps.py

"""The AI step of this brick (module.yaml ai_steps): `ai_suggest_asset_class` — the class, the key and life per area, and
the capital allowance pool, suggested from the description, supplier and amount. Suggest only (`safe_for_auto: false`):
a person accepts, corrects or dismisses each field, and every decision is a row in fa_ai_finding.

An AI step is a PROVIDER of the capability `ai.suggest`, with a ManualTask as its fallback. No provider wired is a
first-class path, not an error. A fixed BASELINE the brick computes itself, with no model, is always shown: the class
defaults. Tests run with no provider, so they need no model call.
"""
from dataclasses import dataclass, field
from typing import Optional

from erp_platform.capabilities import CapabilitySeam, ManualTask

AI_STEPS = {
    "ai_suggest_asset_class": {
        "capability": "ai.suggest",
        "mode": "suggest",
        "modes_allowed": ["off", "suggest"],
        "safe_for_auto": False,
        "fallback": "the class defaults",
        "benefit": "the accountant stops looking up which class, life and tax pool a new asset belongs to",
        "inputs": ["the description", "the supplier", "the amount", "the bill line's expense category",
                   "the class definitions", "the last 50 assets with their classes"],
        "eval": "a labelled set of 200 asset descriptions; class and pool accuracy, and cars put in the main pool per 100 "
                "(the costly error)",
    },
}
SEAMS = {"ai_suggest_asset_class": CapabilitySeam("ai.suggest")}


@dataclass
class Answer:
    """What a step produced: `kind` is off | done (a provider answered) | manual (a person does it) | pending."""
    step: str
    mode: str
    kind: str
    suggestion: dict = field(default_factory=dict)
    rationale: str = ""
    confidence: Optional[int] = None
    model: str = ""
    message: str = ""
    task: Optional[ManualTask] = None

    def as_dict(self) -> dict:
        return {"step": self.step, "mode": self.mode, "kind": self.kind, "suggestion": self.suggestion,
                "rationale": self.rationale, "confidence": self.confidence, "model": self.model, "message": self.message}


def ask(step: str, mode: str, subject_ref: str, version: int, fields: list, context: dict) -> Answer:
    """One run of an AI step for one version of one asset. `context` holds JSON SCALARS only."""
    spec = AI_STEPS[step]
    if mode == "off":
        return Answer(step, mode, "off", message="this step is switched off in the settings")
    if mode not in spec["modes_allowed"]:
        raise ValueError(f"{step}: mode {mode!r} is not allowed ({', '.join(spec['modes_allowed'])})")
    key = f"{step}|{subject_ref}|{version}"             # a retried step asks once per version of the asset
    got = SEAMS[step].call(key, {"step": step, "subject_ref": subject_ref, "fields": fields, "context": context,
                                 "instructions": None})
    if isinstance(got, ManualTask):
        return Answer(step, mode, "manual", message=got.reason, task=got)
    if got.status == "done":
        out = got.output or {}
        return Answer(step, mode, "done", out.get("suggestion") or {}, out.get("rationale", ""), out.get("confidence"),
                      out.get("model", ""), got.message)
    return Answer(step, mode, "pending", message=got.message or got.status)