dodil/erp-brick-manufacturing

DODIL ERP brick: manufacturing

src/brick_manufacturing/infrastructure/ai_steps.py

"""The AI step of this brick (module.yaml ai_steps.explain_variance): an assistant BESIDE the process, never a step in the signed lifecycle.

An AI step is a consumer of the platform capability `ai.suggest` (capabilities.yaml), with the process itself as its fallback. The mode is the
owner's (mfg_settings key ai_explain_variance_mode: off | suggest, default suggest; `auto` is not allowed - safe_for_auto is false). With no
provider wired, a provider that is down, or an answer nobody can use, the cost accountant simply keeps the variance split: the step never
blocks closing an order and never changes one. The provider never writes anything; every finding is a row in mfg_ai_finding the accountant
accepts or dismisses."""
import json

from erp_platform.capabilities import CapabilityRefused, CapabilitySeam, InvalidCapabilityInput, ManualTask, Result

STEP = "explain_variance"
INSTRUCTIONS = ("List up to three likely causes of this production order's variance, most likely first. Each cause is {cause, evidence}: the evidence "
                "must come from the context (a lower yield on one input lot, a longer operation than the routing planned, an over-issue not "
                "reversed, an order that differs from the same product's last orders). You only suggest: the cost accountant decides.")

SEAM = CapabilitySeam("ai.suggest")


def _flat(context: dict) -> dict:
    """ai.suggest's context is a map of JSON scalars: a list or a map goes in as its JSON text (the provider reads it as it would a table)."""
    return {k: (v if v is None or isinstance(v, (str, int, float, bool)) else json.dumps(v, sort_keys=True)) for k, v in context.items()}


def explain(order_id: str, digest: str, context: dict):
    """(model, suggestion_map, rationale, confidence, sources) from the provider, or (None, reason): the fallback is the variance split alone."""
    try:
        got = SEAM.call(f"{STEP}|mfg_production_order:{order_id}|{digest[:24]}",
                        {"step": STEP, "subject_ref": f"mfg_production_order:{order_id}", "fields": ["causes"], "context": _flat(context),
                         "instructions": INSTRUCTIONS})
    except (CapabilityRefused, InvalidCapabilityInput) as e:
        return None, f"the explanation service refused the question: {e}"
    if isinstance(got, ManualTask):
        return None, "no explanation service is wired in this installation"
    assert isinstance(got, Result)
    if got.status != "done" or not got.output:
        return None, f"the explanation service {got.status}" + (f": {got.message}" if got.message else "")
    out = got.output
    return (out.get("model") or got.provider or "", out.get("suggestion") or {}, out.get("rationale") or "", int(out.get("confidence") or 0),
            list(out.get("sources") or [])), ""