Identity and Onboarding

INPUT  photo / scan / PDF page
  |
  1. INTAKE          reject > 10MB, convert HEIC/PDF->image, cap resolution
  |
  2. QUALITY GATE    blur score, brightness, glare, resolution
  |                  -> FAIL FAST. Ask for a re-shoot. This single step
  |                     removes most downstream errors.
  |
  3. PREPROCESS      deskew, de-warp, crop to document edges,
  |                  denoise, contrast normalise
  |
  4. CLASSIFY        which document is this? (Aadhaar / PAN / DL / passport /
  |                  voter ID / utility bill / other)
  |
  5. LAYOUT + OCR    text detection -> recognition, OR a single VLM pass
  |
  6. FIELD EXTRACT   map raw text -> {name, dob, id_number, address, ...}
  |
  7. VALIDATE        checksums, format rules, cross-field consistency,
  |                  date sanity, name-vs-face consistency
  |
  8. CONFIDENCE      per-field score -> auto-accept / review / reject
  |
  9. VERIFY          issuing-database lookup via licensed provider
  |                  (this is the only step that proves the document is real)
  |
OUTPUT structured record + confidence + full audit trail

import cv2, numpy as np

def quality_gate(image_path, min_blur=100.0, min_side=800):
    """Reject bad captures before spending money on OCR.
    Tune thresholds on YOUR OWN sample of real rejected images."""
    img = cv2.imread(image_path)
    if img is None:
        return False, "unreadable_file"

    h, w = img.shape[:2]
    if min(h, w) < min_side:
        return False, "resolution_too_low"

    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)

    # Variance of Laplacian: low variance = few sharp edges = blurry
    blur = cv2.Laplacian(gray, cv2.CV_64F).var()
    if blur < min_blur:
        return False, "too_blurry"

    mean = gray.mean()
    if mean < 55:
        return False, "too_dark"
    if mean > 205:
        return False, "overexposed"

    # Glare: large saturated region usually sits over a field you need
    saturated = float((gray > 250).sum()) / gray.size
    if saturated > 0.06:
        return False, "glare_detected"

    return True, "ok"

# Returning a SPECIFIC reason matters. "Try again" gets you the same
# bad photo. "Too dark - move to a brighter spot" gets you a usable one.

import cv2, numpy as np

def deskew(gray):
    """Rotate so text lines are horizontal."""
    inv = cv2.bitwise_not(gray)
    thr = cv2.threshold(inv, 0, 255, cv2.THRESH_BINARY | cv2.THRESH_OTSU)[1]
    coords = np.column_stack(np.where(thr > 0))
    if len(coords) < 50:
        return gray
    angle = cv2.minAreaRect(coords)[-1]
    angle = -(90 + angle) if angle < -45 else -angle
    if abs(angle) < 0.4:          # don't rotate for nothing
        return gray
    h, w = gray.shape
    M = cv2.getRotationMatrix2D((w // 2, h // 2), angle, 1.0)
    return cv2.warpAffine(gray, M, (w, h),
                          flags=cv2.INTER_CUBIC,
                          borderMode=cv2.BORDER_REPLICATE)

def crop_to_document(img):
    """Find the largest 4-sided contour and crop to it.
    Removes the bedsheet, the table, the fingers."""
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    edged = cv2.Canny(cv2.GaussianBlur(gray, (5, 5), 0), 50, 200)
    cnts, _ = cv2.findContours(edged, cv2.RETR_LIST, cv2.CHAIN_APPROX_SIMPLE)
    for c in sorted(cnts, key=cv2.contourArea, reverse=True)[:5]:
        peri = cv2.arcLength(c, True)
        approx = cv2.approxPolyDP(c, 0.02 * peri, True)
        if len(approx) == 4 and cv2.contourArea(c) > 0.25 * img.size / 3:
            x, y, w, h = cv2.boundingRect(approx)
            return img[y:y + h, x:x + w]
    return img   # fall back to the original rather than returning nothing

def preprocess(path):
    img = crop_to_document(cv2.imread(path))
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    gray = deskew(gray)
    gray = cv2.fastNlMeansDenoising(gray, h=10)
    return cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8)).apply(gray)

import json, re

EXTRACTION_PROMPT = """You are extracting fields from an identity document.

Below is raw OCR text. Return ONLY a JSON object. No prose, no markdown.

Rules:
- Use ONLY text present in the OCR output. Never infer or complete a value.
- If a field is absent or unreadable, set it to null.
- Do not correct spellings. Return exactly what appears.
- dob must be ISO format YYYY-MM-DD, or null if the format is ambiguous.

Schema:
{"document_type": "aadhaar|pan|passport|driving_licence|voter_id|other",
 "name": string|null,
 "father_name": string|null,
 "dob": string|null,
 "gender": "M"|"F"|"O"|null,
 "id_number": string|null,
 "address": string|null,
 "issue_date": string|null,
 "expiry_date": string|null}

OCR TEXT:
---
{ocr_text}
---"""

def structure_fields(ocr_text, llm_call):
    raw = llm_call(EXTRACTION_PROMPT.replace("{ocr_text}", ocr_text))
    raw = re.sub(r"^```(?:json)?|```$", "", raw.strip(), flags=re.M).strip()
    try:
        fields = json.loads(raw)
    except json.JSONDecodeError:
        return None, "llm_returned_invalid_json"

    # CRITICAL: every returned value must actually appear in the OCR text.
    # This is the guard against a fluent, confident hallucination.
    haystack = re.sub(r"[^A-Z0-9]", "", ocr_text.upper())
    for key in ("name", "id_number", "father_name"):
        val = fields.get(key)
        if not val:
            continue
        needle = re.sub(r"[^A-Z0-9]", "", str(val).upper())
        if needle and needle not in haystack:
            fields[key] = None
            fields.setdefault("_rejected", []).append(key)

    return fields, None

# Verhoeff algorithm - dihedral group D5. Catches all single-digit
# errors and all adjacent transpositions, which is exactly the OCR
# failure profile.

_D = [[0,1,2,3,4,5,6,7,8,9],[1,2,3,4,0,6,7,8,9,5],[2,3,4,0,1,7,8,9,5,6],
      [3,4,0,1,2,8,9,5,6,7],[4,0,1,2,3,9,5,6,7,8],[5,9,8,7,6,0,4,3,2,1],
      [6,5,9,8,7,1,0,4,3,2],[7,6,5,9,8,2,1,0,4,3],[8,7,6,5,9,3,2,1,0,4],
      [9,8,7,6,5,4,3,2,1,0]]
_P = [[0,1,2,3,4,5,6,7,8,9],[1,5,7,6,2,8,3,0,9,4],[5,8,0,3,7,9,6,1,4,2],
      [8,9,1,6,0,4,3,5,2,7],[9,4,5,3,1,2,6,8,7,0],[4,2,8,6,5,7,3,9,0,1],
      [2,7,9,3,8,0,6,4,1,5],[7,0,4,6,9,1,3,2,5,8]]

def verhoeff_valid(number: str) -> bool:
    digits = [int(d) for d in str(number) if d.isdigit()]
    if len(digits) != 12:
        return False
    c = 0
    for i, d in enumerate(reversed(digits)):
        c = _D[c][_P[i % 8][d]]
    return c == 0

def aadhaar_valid(number: str) -> bool:
    n = "".join(ch for ch in str(number) if ch.isdigit())
    if len(n) != 12:
        return False
    if n[0] in "01":            # Aadhaar never starts 0 or 1
        return False
    if len(set(n)) == 1:        # 111111111111 etc. - test data
        return False
    return verhoeff_valid(n)

import re
def pan_valid(pan: str) -> bool:
    """AAAAA9999A. 4th char encodes holder type, 5th is surname initial."""
    p = str(pan).strip().upper()
    if not re.fullmatch(r"[A-Z]{5}[0-9]{4}[A-Z]", p):
        return False
    return p[3] in "ABCFGHLJPTK"   # P = individual, C = company, etc.

def gstin_valid(g: str) -> bool:
    return bool(re.fullmatch(
        r"[0-3][0-9][A-Z]{5}[0-9]{4}[A-Z][1-9A-Z]Z[0-9A-Z]", str(g).strip().upper()))

from datetime import date

def cross_validate(f: dict) -> list:
    """Returns a list of problems. Empty list = internally consistent.
    None of these prove the document is genuine - they prove it is
    not internally contradictory."""
    issues = []

    dob = f.get("dob")
    if dob:
        try:
            y, m, d = (int(x) for x in dob.split("-"))
            born = date(y, m, d)
            age = (date.today() - born).days / 365.25
            if age < 0:      issues.append("dob_in_future")
            elif age < 18:   issues.append("applicant_is_minor")
            elif age > 110:  issues.append("dob_implausible")
        except Exception:
            issues.append("dob_unparseable")

    if f.get("expiry_date") and f.get("issue_date"):
        if f["expiry_date"] <= f["issue_date"]:
            issues.append("expiry_before_issue")

    name = (f.get("name") or "").strip()
    if name:
        if len(name) < 3:                       issues.append("name_too_short")
        if any(ch.isdigit() for ch in name):    issues.append("digits_in_name")
        if name == name.lower():                issues.append("name_case_suspicious")

    dt = f.get("document_type")
    num = f.get("id_number")
    if dt == "aadhaar" and num and not aadhaar_valid(num):
        issues.append("aadhaar_checksum_failed")
    if dt == "pan" and num and not pan_valid(num):
        issues.append("pan_format_failed")

    return issues

You are a senior engineer who has built KYC document pipelines in
production for regulated lenders.

My situation:
- Documents I must handle: [e.g. Aadhaar, PAN, driving licence, utility bills]
- Languages / scripts: [e.g. English, Hindi, Tamil]
- Expected volume: [documents per month]
- Capture channel: [mobile app camera / web upload / agent-assisted]
- Data residency constraint: [e.g. data must remain in India]
- Team size and ML experience: [describe honestly]

Give me:

1. A stage-by-stage pipeline design, saying at each stage whether to
   build, buy, or use open source - and why for MY constraints.
2. A specific OCR engine recommendation with the reason, including
   what it will fail on.
3. The validation rules I must implement for these document types,
   including any checksum or format algorithms.
4. Confidence thresholds to start with, and how to tune them from
   real review-queue data.
5. Estimated cost per 1,000 documents for the buy option versus the
   self-hosted option, stating your assumptions.
6. The three things most likely to go wrong in month one.

Be specific about trade-offs. Where you are uncertain about current
pricing or vendor capability, say so rather than guessing.

from dataclasses import dataclass, field

@dataclass
class Decision:
    action: str                    # auto_accept | review | reject
    reasons: list = field(default_factory=list)
    priority: str = "normal"       # review queue ordering

# Thresholds are per FIELD, not per document. An address read at 0.72
# is survivable. An ID number read at 0.72 is not.
FIELD_MIN = {
    "id_number": 0.95,   # checksum-protected, so demand high confidence
    "name":      0.90,   # drives sanctions screening downstream
    "dob":       0.90,   # drives age eligibility
    "gender":    0.75,
    "address":   0.70,   # long, noisy, lower stakes
}

def route(fields, confidences, validation_issues, face_score, liveness_pass):
    d = Decision(action="auto_accept")

    # Hard rejects first - no point queueing these for a human
    if not liveness_pass:
        return Decision("reject", ["liveness_failed"], "high")
    for blocking in ("aadhaar_checksum_failed", "pan_format_failed",
                     "expiry_before_issue", "dob_in_future"):
        if blocking in validation_issues:
            return Decision("reject", [blocking])

    # Soft signals -> human review
    for f_name, floor in FIELD_MIN.items():
        if fields.get(f_name) is None:
            d.action = "review"; d.reasons.append(f"missing:{f_name}")
        elif confidences.get(f_name, 0.0) < floor:
            d.action = "review"; d.reasons.append(f"low_conf:{f_name}")

    if fields.get("_rejected"):        # LLM grounding check fired
        d.action = "review"
        d.reasons.append("ungrounded_llm_field")
        d.priority = "high"

    # Face match: a band, not a line
    if face_score < 0.60:
        return Decision("reject", ["face_mismatch"], "high")
    elif face_score < 0.80:
        d.action = "review"; d.reasons.append("face_borderline")

    if "applicant_is_minor" in validation_issues:
        return Decision("reject", ["under_18"])

    return d

{
  "attempt_id": "onb_01HXY...",
  "customer_ref": "cust_9f3a...",
  "timestamp": "2026-05-22T09:14:22Z",

  "capture": {
    "channel": "mobile_ios",
    "sdk_version": "3.4.1",
    "device_attestation": "passed",
    "quality_gate": {"result": "pass", "blur_score": 184.2, "retries": 1}
  },

  "extraction": {
    "document_type": "aadhaar",
    "engine": "paddleocr-2.9 + llm-structuring",
    "model_versions": {"ocr": "PP-OCRv4", "structuring": "<model-id>"},
    "fields": {"name": "...", "dob": "...", "id_number_masked": "XXXXXXXX1234"},
    "confidences": {"name": 0.97, "dob": 0.94, "id_number": 0.99},
    "grounding_check": "passed"
  },

  "validation": {"checksum": "pass", "cross_field_issues": []},

  "biometric": {
    "face_match_score": 0.91,
    "liveness": {"type": "passive", "result": "pass", "vendor": "...",
                 "vendor_ref": "lv_88121"}
  },

  "verification": {
    "source": "uidai_via_<licensed_provider>",
    "provider_ref": "txn_55219",
    "result": "match",
    "consent_ref": "cons_772a", "consent_timestamp": "2026-05-22T09:13:58Z"
  },

  "decision": {
    "action": "review",
    "reasons": ["face_borderline"],
    "policy_version": "kyc-policy-v7",
    "human_reviewer": "rev_114",
    "human_action": "approved",
    "human_rationale": "lighting artefact; document photo 9 years old",
    "final_timestamp": "2026-05-22T11:02:10Z"
  }
}