ভ্যালিডেশন ও রিপেয়ার: jsonschema, বিজনেস রুল ও সীমিত রিট্রাই

import json, re
from jsonschema import Draft202012Validator
from promptlab import render

DATE = r"^\d{4}-\d{2}-\d{2}$"
SCHEMA = {"type": "object", "additionalProperties": False,
          "required": ["invoice_number", "issue_date", "due_date", "currency", "lines", "total"],
          "properties": {
              "invoice_number": {"type": "string", "pattern": r"^INV-\d{4}$"},
              "issue_date": {"type": "string", "pattern": DATE},
              "due_date": {"type": ["string", "null"], "pattern": DATE},
              "currency": {"enum": ["USD", "EUR", "BDT"]},
              "total": {"type": "number", "minimum": 0},
              "lines": {"type": "array", "minItems": 1, "items": {
                  "type": "object", "additionalProperties": False, "required": ["description", "quantity", "unit_price"],
                  "properties": {"description": {"type": "string", "minLength": 1},
                                 "quantity": {"type": "integer", "minimum": 1}, "unit_price": {"type": "number", "minimum": 0}}}}}}
VALIDATOR = Draft202012Validator(SCHEMA)

QUIRK = {5: "the invoice total disagrees with its own lines", 8: "the date 06/04/2026 can be read two ways"}
def make_invoice(i):
    lines = [{"description": f"Item {'ABC'[j]}", "quantity": 1 + (i + j) % 3, "unit_price": float(10 + 7 * ((3 * i + j) % 5))} for j in range(1 + i % 3)]
    total = float(sum(l["quantity"] * l["unit_price"] for l in lines)) + (5.0 if i == 5 else 0.0)
    gold = {"invoice_number": f"INV-{1001 + i}", "issue_date": f"2026-03-{10 + i:02d}",
            "due_date": None if i % 4 == 3 else f"2026-04-{10 + i:02d}", "currency": ["USD", "EUR", "BDT"][i % 3],
            "lines": lines, "total": total}
    shown = "06/04/2026" if i == 8 else gold["issue_date"]
    head = f"Invoice {gold['invoice_number']} | Date: {shown} | Due: {gold['due_date'] or 'on receipt'} | Currency: {gold['currency']}"
    body = [f"{l['quantity']} x {l['description']} @ {l['unit_price']:.2f}" for l in lines]
    return "\n".join([head, *body, f"Total: {total:.2f}"]), gold

SOURCES, GOLD = zip(*(make_invoice(i) for i in range(12)))
print(SOURCES[1]); print("---"); print(SOURCES[8])

Invoice INV-1002 | Date: 2026-03-11 | Due: 2026-04-11 | Currency: EUR
2 x Item A @ 31.00
3 x Item B @ 38.00
Total: 176.00
---
Invoice INV-1009 | Date: 06/04/2026 | Due: 2026-04-18 | Currency: BDT
3 x Item A @ 38.00
1 x Item B @ 10.00
2 x Item C @ 17.00
Total: 158.00

def parse_json(text):
    """Strict syntax check: returns (document, None) or (None, reason). Rejects NaN and duplicate keys."""
    def no_dupes(pairs):
        keys = [k for k, _ in pairs]
        if len(keys) != len(set(keys)): raise ValueError("duplicate key")
        return dict(pairs)
    def no_constant(name): raise ValueError(f"{name} is not valid JSON")
    try: return json.loads(text, object_pairs_hook=no_dupes, parse_constant=no_constant), None
    except json.JSONDecodeError as exc: return None, f"not JSON: {exc.msg} (line {exc.lineno}, column {exc.colno})"
    except ValueError as exc: return None, f"not JSON: {exc}"

for text in ['{"total": 300.0}', '```json\n{"total": 300.0}\n```', '{"total": 300.0,}', '{"total": NaN}', '{"total": 1, "total": 2}', '{"total": 30']:
    print(f"{text[:26]!r:30}", parse_json(text)[1] or "ok")
print(json.loads('{"total": NaN}'))     # the standard parser accepts NaN unless you forbid it

'{"total": 300.0}'             ok
'```json\n{"total": 300.0}\n`' not JSON: Expecting value (line 1, column 1)
'{"total": 300.0,}'            not JSON: Expecting property name enclosed in double quotes (line 1, column 17)
'{"total": NaN}'               not JSON: NaN is not valid JSON
'{"total": 1, "total": 2}'     not JSON: duplicate key
'{"total": 30'                 not JSON: Expecting ',' delimiter (line 1, column 13)
{'total': nan}

def diagnose(doc):
    """Schema layer: sort jsonschema errors into missing, extra and incorrect fields."""
    found = {"missing": [], "extra": [], "incorrect": []}
    for e in VALIDATOR.iter_errors(doc):
        where = "/".join(map(str, e.path)) or "(top)"
        if e.validator == "required":
            found["missing"] += [f"{where}/{m}" for m in re.findall(r"'(\w+)' is a required", e.message)]
        elif e.validator == "additionalProperties":
            found["extra"] += [f"{where}/{x}" for x in re.findall(r"'(\w+)'", e.message.split("(")[1])]
        else:
            found["incorrect"].append(f"{where}: {e.instance!r} fails {e.validator}")
    return {k: sorted(v) for k, v in found.items() if v}

ok_doc = json.loads(json.dumps(GOLD[1]))
print(diagnose(ok_doc) or "no problems")
bad = {**ok_doc, "currency": "eur", "lines": [{"description": "Item A", "quantity": "2", "unit_price": 10.0, "sku": "A1"}]}
del bad["total"]
print(diagnose(bad))
print(Draft202012Validator({"type": "string", "format": "date"}).is_valid("14 March"), "<- 'format' is not enforced by default")

no problems
{'missing': ['(top)/total'], 'extra': ['lines/0/sku'], 'incorrect': ["currency: 'eur' fails enum", "lines/0/quantity: '2' fails type"]}
True <- 'format' is not enforced by default

TODAY = "2026-12-31"      # a fixed "today" keeps the example repeatable; in real code pass the current date in

def rules(doc):
    """Business rules: facts the schema cannot express. Run only on a document that already fits the schema."""
    found = []
    expected = sum(l["quantity"] * l["unit_price"] for l in doc["lines"])
    if abs(expected - doc["total"]) > 0.005: found.append(f"total {doc['total']} but the lines add up to {expected}")
    if doc["due_date"] and doc["due_date"] < doc["issue_date"]: found.append("due_date is before issue_date")
    if doc["issue_date"] > TODAY: found.append("issue_date is in the future")
    return found

print(rules(GOLD[1]), rules(GOLD[5]))
print(rules({**GOLD[1], "due_date": "2026-03-01", "issue_date": "2027-01-05"}))

[] ['total 100.0 but the lines add up to 95.0']
['due_date is before issue_date', 'issue_date is in the future']

EXTRACT = """Extract the invoice as ONE JSON object and nothing else (no Markdown).
Fields: invoice_number, issue_date (YYYY-MM-DD), due_date (or null), currency (USD, EUR, BDT),
lines (a list of objects with description, quantity, unit_price), total. Use null if missing; never guess.
### INVOICE
{source}"""
REPAIR = EXTRACT + "\n### PROBLEMS WITH YOUR LAST ANSWER\n{problems}\nFix only these problems, using the invoice text above."

def read_source(source):
    """What every toy 'reads' from the text. Note the silent guess: a slash date is read as day/month/year."""
    head, *rest = source.splitlines()
    f = dict(p.strip().split(": ", 1) for p in head.split("|")[1:])
    d = re.fullmatch(r"(\d\d)/(\d\d)/(\d{4})", f["Date"])
    lines = [dict(zip(("quantity", "description", "unit_price"), (int(m[0]), m[1], float(m[2]))))
             for m in (re.fullmatch(r"(\d+) x (.+) @ ([\d.]+)", l).groups() for l in rest[:-1])]
    return {"invoice_number": head.split()[1], "issue_date": f"{d[3]}-{d[2]}-{d[1]}" if d else f["Date"],
            "due_date": None if f["Due"] == "on receipt" else f["Due"], "currency": f["Currency"],
            "lines": lines, "total": float(rest[-1].split(": ")[1])}

def answer(stop, text=""):
    return {"stop": stop, "text": text}      # API-shaped result; the stop field is named differently by each provider

def parts(prompt):
    """Invoice number, what a careful reader would extract, and whether this is a repair request."""
    doc = read_source(prompt.split("### INVOICE\n", 1)[1].split("\n### PROBLEMS")[0])
    return int(doc["invoice_number"][4:]) - 1001, doc, "### PROBLEMS" in prompt

def model_a(prompt):
    """Stand-in 'strict' model: clean JSON, except one reply cut off by a token limit."""
    n, doc, repair = parts(prompt)
    return answer("max_tokens", json.dumps(doc)[:60]) if n == 7 and not repair else answer("end", json.dumps(doc))

def model_b(prompt):
    """Stand-in 'wrapper' model: fences on even invoices, a string quantity on every third, one refusal."""
    n, doc, repair = parts(prompt)
    if not repair:
        if n == 9: return answer("refusal")
        if n % 3 == 0: doc["lines"][0]["quantity"] = str(doc["lines"][0]["quantity"])
        if n % 2 == 0: return answer("end", "```json\n" + json.dumps(doc) + "\n```")
    return answer("end", json.dumps(doc))

def model_c(prompt):
    """Stand-in 'sloppy values' model: lower-case currency, a prose date, an invented total; clean when told what is wrong."""
    n, doc, repair = parts(prompt)
    if not repair:
        if n % 3 == 1: doc["currency"] = doc["currency"].lower()
        if n % 4 == 0: doc["issue_date"] = "14 Mar 2026"
        if n % 5 == 2: doc["total"] += 1.0
    return answer("end", json.dumps(doc))

def model_stuck(prompt):
    return answer("end", '{"total": "unknown"}')              # never valid, whatever you tell it

Your JSON was invalid. Try again.

[the original task and the invoice text, unchanged]

### PROBLEMS WITH YOUR LAST ANSWER
not JSON: Expecting property name enclosed in double quotes (line 1, column 17)
Fix only these problems, using the invoice text above. Output one JSON object and nothing else.

def ambiguous_source(source):
    """A rule on the SOURCE, before any model call: dd/mm vs mm/dd cannot be known from '06/04/2026'."""
    m = re.search(r"\b(\d\d)/(\d\d)/\d{4}\b", source)
    return "ambiguous date " + m.group(0) if m and int(m.group(1)) <= 12 and int(m.group(2)) <= 12 else None

def grounded(doc, source):
    return f"{doc['total']:.2f}" in source           # the model copied the total that the source really shows

def process(source, model, max_repairs=2):
    if why := ambiguous_source(source):               # do not ask a model to guess; do not repair
        return {"status": "needs_review", "why": why, "calls": 0}
    prompt, calls = render(EXTRACT, source=source), 0
    while True:
        result, calls = model(prompt), calls + 1
        if result["stop"] != "end":                    # check the stop status BEFORE parsing
            return {"status": "failed", "why": "stop=" + result["stop"], "calls": calls}
        doc, error = parse_json(result["text"])
        if error: problems = [error]
        elif schema := diagnose(doc): problems = [f"{k}: {', '.join(v)}" for k, v in schema.items()]
        elif not (problems := rules(doc)):
            return {"status": "ok" if calls == 1 else "repaired", "doc": doc, "calls": calls}
        elif grounded(doc, source):                    # the source itself is inconsistent: repairing would be a guess
            return {"status": "needs_review", "why": "; ".join(problems), "calls": calls}
        if calls > max_repairs:
            return {"status": "failed", "why": problems[0].split(":")[0], "calls": calls}
        prompt = render(REPAIR, source=source, problems="\n".join(problems))

for i in (0, 4, 5, 8, 7, 9):
    for name, model in (("a", model_a), ("b", model_b)):
        r = process(SOURCES[i], model); print(f"INV-{1001 + i} model {name}:", r["status"], r.get("why", ""), "| calls", r["calls"])
r = process(SOURCES[0], model_stuck)
print("stuck model:", r["status"], "|", r["why"], "| calls", r["calls"])

INV-1001 model a: ok  | calls 1
INV-1001 model b: repaired  | calls 2
INV-1005 model a: ok  | calls 1
INV-1005 model b: repaired  | calls 2
INV-1006 model a: needs_review total 100.0 but the lines add up to 95.0 | calls 1
INV-1006 model b: needs_review total 100.0 but the lines add up to 95.0 | calls 1
INV-1009 model a: needs_review ambiguous date 06/04/2026 | calls 0
INV-1009 model b: needs_review ambiguous date 06/04/2026 | calls 0
INV-1008 model a: failed stop=max_tokens | calls 1
INV-1008 model b: ok  | calls 1
INV-1010 model a: ok  | calls 1
INV-1010 model b: failed stop=refusal | calls 1
stuck model: failed | missing | calls 3

from collections import Counter
CLEAN = [i for i in range(12) if i not in QUIRK]             # the 10 invoices without a built-in source problem

def layers(model):
    """First-attempt compliance, layer by layer (no repair): stop, syntax, schema, rules, exact match."""
    n = Counter()
    for i in CLEAN:
        result = model(render(EXTRACT, source=SOURCES[i]))
        n["stop"] += result["stop"] == "end"
        doc, error = parse_json(result["text"]) if result["stop"] == "end" else (None, "x")
        n["syntax"] += error is None
        fits = error is None and not diagnose(doc)
        n["schema"] += fits
        n["rules"] += fits and not rules(doc)
        n["exact"] += doc == GOLD[i]
    return n

print("model     stop syntax schema rules exact   (first attempt, 10 invoices)")
for name, model in (("model_a", model_a), ("model_b", model_b), ("model_c", model_c)):
    n = layers(model)
    print(f"{name:9}", *(f"{n[k]:>{len(k) + 1}}" for k in ("stop", "syntax", "schema", "rules", "exact")))
    final = Counter(process(SOURCES[i], model)["status"] for i in range(12))
    print("          with bounded repair:", dict(sorted(final.items())))

model     stop syntax schema rules exact   (first attempt, 10 invoices)
model_a       9       9       9      9      9
          with bounded repair: {'failed': 1, 'needs_review': 2, 'ok': 9}
model_b       9       4       3      3      3
          with bounded repair: {'failed': 1, 'needs_review': 2, 'ok': 3, 'repaired': 6}
model_c      10      10       5      4      4
          with bounded repair: {'needs_review': 2, 'ok': 4, 'repaired': 6}

from datetime import date

def real_date(text):
    try: date.fromisoformat(text); return True
    except ValueError: return False

doc = {**GOLD[1], "issue_date": "2026-02-31"}
print("schema:", diagnose(doc) or "no problems", "| real date:", real_date(doc["issue_date"]))

schema: no problems | real date: False

def model_flaky(prompt):                         # stand-in: broken first answer, valid once it is told what is wrong
    return answer("end", '{"total": 1,}') if "### PROBLEMS" not in prompt else model_a(prompt)

for limit in (2, 0):
    r = process(SOURCES[1], model_flaky, max_repairs=limit)
    print(f"max_repairs={limit}:", r["status"], "| calls", r["calls"], r.get("why", ""))

max_repairs=2: repaired | calls 2 
max_repairs=0: failed | calls 1 not JSON

rows = [(f"INV-{1001 + i}", process(SOURCES[i], model_c)) for i in range(12)]
for number, r in rows:
    print(number, f"{r['status']:13}", r["calls"], r.get("why", ""))
accepted = [r["doc"] for _, r in rows if "doc" in r]
print("accepted:", len(accepted), "| all pass schema and rules:", all(not diagnose(d) and not rules(d) for d in accepted),
      "| equal to the intended record:", sum(d == GOLD[int(d["invoice_number"][4:]) - 1001] for d in accepted))

INV-1001 repaired      2 
INV-1002 repaired      2 
INV-1003 repaired      2 
INV-1004 ok            1 
INV-1005 repaired      2 
INV-1006 needs_review  1 total 100.0 but the lines add up to 95.0
INV-1007 ok            1 
INV-1008 repaired      2 
INV-1009 needs_review  0 ambiguous date 06/04/2026
INV-1010 ok            1 
INV-1011 repaired      2 
INV-1012 ok            1 
accepted: 10 | all pass schema and rules: True | equal to the intended record: 10