BNY's Agents Read Payment Fields For Meaning, BNP Paribas Cut A Workflow From Ten Steps To Six: Building A Payment-Repair Agent That Learns Exceptions From Humans, In Code

from dataclasses import dataclass, field
import re

IBAN_RE = re.compile(r"^[A-Z]{2}[0-9]{2}[A-Z0-9]{11,30}$")
BIC_RE = re.compile(r"^[A-Z]{6}[A-Z0-9]{2}([A-Z0-9]{3})?$")
PURPOSE_CODES = {"SALA", "SUPP", "TRAD", "INTC", "DIVI", "LOAN", "TAXS"}   # from the external code list

@dataclass
class Finding:
    path: str; code: str; severity: str; detail: str = ""

@dataclass
class ValidationResult:
    findings: list = field(default_factory=list)
    @property
    def blocking(self): return [f for f in self.findings if f.severity == "error"]

def iban_checksum_ok(iban: str) -> bool:
    s = iban[4:] + iban[:4]
    digits = "".join(str(int(c, 36)) for c in s)
    return int(digits) % 97 == 1

def validate_pacs008(tx: dict) -> ValidationResult:
    r = ValidationResult()
    cdtr_acct = tx.get("CdtrAcct", {}).get("Id", {}).get("IBAN")
    if not cdtr_acct:
        r.findings.append(Finding("CdtrAcct.Id.IBAN", "MISSING", "error"))
    elif not IBAN_RE.match(cdtr_acct) or not iban_checksum_ok(cdtr_acct):
        r.findings.append(Finding("CdtrAcct.Id.IBAN", "INVALID_IBAN", "error", cdtr_acct))
    bic = tx.get("CdtrAgt", {}).get("FinInstnId", {}).get("BICFI")
    if bic and not BIC_RE.match(bic):
        r.findings.append(Finding("CdtrAgt.FinInstnId.BICFI", "INVALID_BIC", "error", bic))
    purpose = tx.get("Purp", {}).get("Cd")
    if purpose and purpose not in PURPOSE_CODES:
        r.findings.append(Finding("Purp.Cd", "UNKNOWN_PURPOSE", "warning", purpose))
    if not tx.get("Cdtr", {}).get("Nm"):
        r.findings.append(Finding("Cdtr.Nm", "MISSING_NAME", "error"))
    addr = tx.get("Cdtr", {}).get("PstlAdr", {})
    if addr and not addr.get("Ctry"):
        r.findings.append(Finding("Cdtr.PstlAdr.Ctry", "MISSING_COUNTRY", "warning"))
    return r

import json
from dataclasses import dataclass

@dataclass(frozen=True)
class RepairProposal:
    path: str; current: str; proposed: str; evidence: list; confidence: float; rationale: str

PROMPT = (
    "You are a payments operations analyst. A payment failed a check. Using ONLY the reference data "
    "and exception rules provided, propose the single most likely intended value for the failing field. "
    "Cite the reference-data ids you relied on. If the evidence is insufficient, return proposed=null. "
    "Respond as JSON with fields: proposed, evidence (list of ids), confidence (0-1), rationale."
)

def propose_repair(tx: dict, finding, refdata: dict, rules: list, llm) -> RepairProposal:
    ctx = {"payment": tx, "finding": finding.__dict__, "reference_data": refdata, "exception_rules": rules}
    out = json.loads(llm(PROMPT, json.dumps(ctx)))
    return RepairProposal(
        path=finding.path, current=finding.detail, proposed=out.get("proposed"),
        evidence=out.get("evidence", []), confidence=float(out.get("confidence", 0)),
        rationale=out.get("rationale", ""),
    )

# Example of the reference data the model sees for an INVALID_BIC finding:
# {"bic_directory": [{"id": "bicdir:MIDLGB22", "bic": "MIDLGB22", "status": "active", "replaces": "MIDLGB2L"}],
#  "beneficiary_history": [{"id": "hist:8812", "iban": "GB29...", "last_bic": "MIDLGB22", "count": 14}]}

from decimal import Decimal

AUTO_APPLY = {
    # path -> (min confidence, max amount, required evidence kinds)
    "CdtrAgt.FinInstnId.BICFI": (0.95, Decimal("250000"), {"bicdir"}),
    "Cdtr.PstlAdr.Ctry":        (0.90, Decimal("1000000"), {"hist", "bicdir"}),
    "Purp.Cd":                  (0.90, Decimal("1000000"), {"hist"}),
}
NEVER_AUTO = {"CdtrAcct.Id.IBAN", "Cdtr.Nm", "IntrBkSttlmAmt"}   # account, name, amount: always a person

def decide(proposal, amount: Decimal, sanctions_clear: bool) -> str:
    if proposal.proposed is None or not sanctions_clear or proposal.path in NEVER_AUTO:
        return "queue"
    rule = AUTO_APPLY.get(proposal.path)
    if not rule:
        return "queue"
    min_conf, max_amt, kinds = rule
    evidence_kinds = {e.split(":")[0] for e in proposal.evidence}
    if proposal.confidence >= min_conf and amount <= max_amt and evidence_kinds & kinds:
        return "apply"
    return "queue"

from collections import defaultdict
from dataclasses import dataclass

@dataclass(frozen=True)
class Correction:
    finding_code: str; path: str; context_key: str; resolution: str; analyst: str; proposal_accepted: bool

class ExceptionTeacher:
    def __init__(self, promote_after=3):
        self.candidates = defaultdict(list)
        self.rules = []                      # promoted, reviewed rules fed to propose_repair()
        self.promote_after = promote_after

    def record(self, c: Correction) -> None:
        key = (c.finding_code, c.path, c.context_key)
        self.candidates[key].append(c)
        same = [x for x in self.candidates[key] if x.resolution == c.resolution]
        if len(same) >= self.promote_after and len({x.analyst for x in same}) >= 2:
            self.rules.append({
                "when": {"finding": c.finding_code, "path": c.path, "context": c.context_key},
                "then": c.resolution,
                "evidence": f"{len(same)} corrections by {len({x.analyst for x in same})} analysts",
                "status": "pending_review",           # a supervisor approves before it is used
            })

# context_key examples: "sender:ACME-LTD|cdtr:GB29..." or "bic_replaced:MIDLGB2L->MIDLGB22"
# Each promoted rule is a sentence the model reads next time, with the human evidence behind it.