dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/infrastructure/ai_steps.py
"""AI steps (the AI-step contract): a step PROVIDES a capability (`ai.suggest`), with the ManualTask as its fallback and reviewer. The mode is
part of the signed BPMN: off | suggest (AI drafts, a person decides). All three steps are suggest-only (`safe_for_auto: false`).
In order: mode `off` -> nothing runs. A fixed BASELINE the brick computes itself, with no model (it is always shown, so the step is useful
with no provider wired). Then `propose()` asks the `ai.suggest` provider the composition wired; none wired is a first-class path: a ManualTask
in pur_task for a person. The provider never applies anything: a finding changes no bill and no bank detail until a person decides."""
from dataclasses import dataclass, field
from typing import Optional
from erp_platform.capabilities import CapabilitySeam, ManualTask
AI_STEPS = {
"ai_read_bill": {"capability": "ai.suggest", "mode": "suggest", "modes_allowed": ["off", "suggest"], "safe_for_auto": False,
"fallback": "the clerk types the bill", "benefit": "the clerk stops typing supplier invoices",
"inputs": ["the PDF (an object raw/purchasing/bills/<sha256>.pdf)", "the supplier's open orders", "the supplier's last bills"],
"eval": "a labelled set of supplier PDFs: field accuracy per field, and bills recorded with a wrong amount per 100"},
"ai_check_bill": {"capability": "ai.suggest", "mode": "suggest", "modes_allowed": ["off", "suggest"], "safe_for_auto": False,
"fallback": "the exact-key duplicate check only (not_duplicate)",
"benefit": "a duplicate under a slightly different number, or an amount far off this supplier's normal, is caught",
"inputs": ["the bill", "the supplier's bills of the last 12 months", "its orders"],
"eval": "seeded near-duplicates and clean bills: recall, false findings per 100 bills"},
"ai_flag_bank_change": {"capability": "ai.suggest", "mode": "suggest", "modes_allowed": ["off", "suggest"], "safe_for_auto": False,
"fallback": "the call-back gate alone", "benefit": "the 'we have changed bank' fraud is flagged to the approver",
"inputs": ["the change", "how it was requested", "the supplier's history", "the next run's amount for them"],
"eval": "known fraud patterns and genuine changes: recall of the fraud patterns"},
}
SEAMS = {k: CapabilitySeam("ai.suggest") for k in AI_STEPS}
def propose(step: str, key: str, payload: dict, mode: str):
if mode == "off":
return None
if mode not in AI_STEPS[step]["modes_allowed"]:
raise ValueError(f"{step}: mode {mode!r} is not allowed")
return SEAMS[step].call(key, payload)
@dataclass
class Answer:
step: str
mode: str
kind: str # off | done | manual | pending
suggestion: dict = field(default_factory=dict)
rationale: str = ""
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,
"model": self.model, "message": self.message}
def ask(step: str, mode: str, subject_ref: str, version: int, context: dict) -> Answer:
"""One run of an AI step for one version of one record. `context` holds JSON SCALARS only."""
got = propose(step, f"{step}|{subject_ref}|{version}", {"step": step, "subject_ref": subject_ref, "fields": sorted(context), "context": context, "instructions": None}, mode)
if got is None:
return Answer(step, mode, "off", message="this step is switched off in the settings")
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("model", ""), got.message)
return Answer(step, mode, "pending", message=got.message or got.status)
# ── the baselines the brick computes itself (no model) ────────────────────────────────────────────
def _norm(s: str) -> str:
return "".join(ch for ch in (s or "").lower() if ch.isalnum())
def likely_duplicates(bill: dict, others: list) -> list:
"""Same supplier, same total and a date within 10 days, under another number — or the same number once punctuation is ignored."""
out = []
for o in others:
if o["bill_id"] == bill["bill_id"]:
continue
same_no = _norm(o["supplier_invoice_no"]) == _norm(bill["supplier_invoice_no"])
near = o["total"] == bill["total"] and abs((o["invoice_date"] - bill["invoice_date"]).days) <= 10
if same_no or near:
out.append({"bill_id": o["bill_id"], "supplier_invoice_no": o["supplier_invoice_no"], "total": str(o["total"]),
"invoice_date": o["invoice_date"].isoformat(),
"why": "the same number apart from punctuation" if same_no else "the same total within 10 days, under another number"})
return out
def unusual_amount(bill_total, history: list) -> Optional[dict]:
"""An amount more than 5x this supplier's average of the last bills (with at least 3)."""
past = [h for h in history if h > 0]
if len(past) < 3:
return None
avg = sum(past) / len(past)
return {"average": str(round(avg, 2)), "why": "far above this supplier's usual bill"} if bill_total > avg * 5 else None
def bank_signals(asked_by: str, country_changed: bool, next_run_amount) -> list:
out = []
if (asked_by or "").lower() in ("email", "e-mail", "letter", "phone", "telephone"):
out.append({"signal": "requested_remotely", "why": f"the change was asked for by {asked_by}"})
if country_changed:
out.append({"signal": "new_country", "why": "the new account is in another country"})
if next_run_amount and next_run_amount > 0:
out.append({"signal": "before_large_run", "why": f"{next_run_amount} is due to this supplier in the next run"})
return out