Validating extracted RFI fields against custom JSON schemas

Automated ingestion pulls Request for Information (RFI) data out of PDFs, scanned submittals, and field-captured forms, but the extracted payload almost never arrives clean. Raw OCR routinely emits type mismatches, truncated strings, malformed dates, currency strings where a number belongs, and outright hallucinated fields that no template produced. This page covers exactly one slice of the pipeline: how to validate that extracted RFI dictionary against a custom JSON Schema — a language-agnostic contract that encodes construction-specific rules — so a structurally broken record is caught and quarantined before it can drift a cost ledger or misroute an approval. A generic “is this a string?” check is not enough here; the validation layer has to know that RFI-2026-00123 is a legal identifier and RFI #123 is not, that 03 30 00 is a real CSI MasterFormat section and concrete is not. This is the gate every record must clear before it enters the broader schema validation rules layer and, beyond it, the automated document ingestion and parsing pipeline.

Key rules and specification

Author the contract so each construction constraint is an explicit, testable rule rather than a loose string type. The table below is the minimum viable RFI contract; treat additionalProperties: false as non-negotiable, because that single rule is what rejects hallucinated OCR artifacts.

Two of these rules carry more weight than the rest. The cost_impact rule is where extraction noise does the most damage: OCR hands you "$12,500", a JSON number field expects 12500.0, and a naive pipeline either rejects a perfectly good RFI or — worse — coerces the string to 0 and quietly understates a change-order exposure. And additionalProperties: false is the only rule that defends against the model inventing a confidence_note or page_2_text field that no downstream consumer expects.

Production validation and routing

The snippet below keeps one canonical JSON Schema as the cross-service contract, validates the extracted dict against it with the jsonschema library (aggregating every violation, not just the first), and mirrors the same contract in a Pydantic v2 model that normalizes currency noise before re-asserting it. Routing follows the pipeline’s canonical confidence bands: at or above 0.92 an extraction auto-routes, 0.75–0.92 goes to human review, and anything below 0.75 is quarantined — but a schema violation overrides confidence entirely, because a structurally broken record is never safe to auto-route no matter how confident the OCR was.

from __future__ import annotations

import logging
from datetime import date
from typing import Any, Callable, Literal

from jsonschema import Draft202012Validator, FormatChecker
from pydantic import BaseModel, Field, field_validator

logger = logging.getLogger("construction.rfi_validation")

# --- 1. The canonical custom JSON Schema: the language-agnostic contract ---
RFI_JSON_SCHEMA: dict[str, Any] = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "type": "object",
    "additionalProperties": False,  # reject hallucinated OCR fields outright
    "required": [
        "rfi_number", "issue_date", "masterformat_section",
        "discipline", "description", "status",
    ],
    "properties": {
        "rfi_number": {"type": "string", "pattern": r"^RFI-\d{4}-\d{3,5}$"},
        "issue_date": {"type": "string", "format": "date"},
        "due_date": {"type": "string", "format": "date"},
        # MasterFormat section number, XX XX XX (e.g. "03 30 00" cast-in-place concrete)
        "masterformat_section": {"type": "string", "pattern": r"^\d{2} \d{2} \d{2}$"},
        "discipline": {"enum": ["ARCH", "STR", "MEP", "CIV", "ELEC", "PLMB"]},
        "description": {"type": "string", "minLength": 10, "maxLength": 4000},
        "status": {"enum": ["OPEN", "PENDING_RESPONSE", "ANSWERED", "CLOSED", "WITHDRAWN"]},
        "cost_impact": {"type": ["number", "null"], "minimum": -1_000_000, "maximum": 10_000_000},
        "schedule_impact_days": {"type": ["integer", "null"], "minimum": 0, "maximum": 365},
        "attachments": {"type": "array", "items": {"type": "string", "format": "uri"}, "maxItems": 25},
    },
}

# format_checker activates "format": "date" so 2026-02-30 is rejected, not waved through.
_VALIDATOR = Draft202012Validator(RFI_JSON_SCHEMA, format_checker=FormatChecker())


def validate_against_schema(payload: dict[str, Any]) -> list[dict[str, str]]:
    """Aggregate every schema violation as a structured, routable record."""
    errors: list[dict[str, str]] = []
    for err in sorted(_VALIDATOR.iter_errors(payload), key=lambda e: list(e.absolute_path)):
        field = ".".join(str(p) for p in err.absolute_path) or "$"
        errors.append({"field": field, "rule": err.validator, "detail": err.message})
    return errors


# --- 2. Typed gateway model: normalize OCR noise, then re-assert the contract ---
class ExtractedRFI(BaseModel):
    """Pydantic v2 mirror of RFI_JSON_SCHEMA for type-safe downstream consumption."""

    model_config = {"extra": "forbid"}  # same intent as additionalProperties: false

    rfi_number: str = Field(pattern=r"^RFI-\d{4}-\d{3,5}$")
    issue_date: date
    due_date: date | None = None
    masterformat_section: str = Field(pattern=r"^\d{2} \d{2} \d{2}$")
    discipline: Literal["ARCH", "STR", "MEP", "CIV", "ELEC", "PLMB"]
    description: str = Field(min_length=10, max_length=4000)
    status: Literal["OPEN", "PENDING_RESPONSE", "ANSWERED", "CLOSED", "WITHDRAWN"]
    cost_impact: float | None = Field(default=None, ge=-1_000_000, le=10_000_000)
    schedule_impact_days: int | None = Field(default=None, ge=0, le=365)

    @field_validator("cost_impact", mode="before")
    @classmethod
    def strip_currency(cls, v: Any) -> Any:
        """OCR emits '$12,500'; coerce to float so cost rollups stay numeric, not string-zeroed."""
        if isinstance(v, str):
            cleaned = v.strip().replace(",", "").replace("$", "")
            return None if cleaned == "" else float(cleaned)
        return v


def route_extracted_rfi(
    payload: dict[str, Any],
    ocr_confidence: float,
    publish: Callable[[str], None],
) -> str:
    """Validate an extracted RFI, then route by schema result and OCR confidence."""
    # Structural validity gates everything: a broken record never auto-routes.
    schema_errors = validate_against_schema(payload)
    if schema_errors:
        logger.warning("RFI %s failed schema: %s", payload.get("rfi_number"), schema_errors)
        return "quarantine"

    if ocr_confidence >= 0.92:
        clean = ExtractedRFI.model_validate(payload)
        publish(clean.model_dump_json())  # only type-safe records reach the ledger
        logger.info("Auto-routed %s", clean.rfi_number)
        return "auto_route"
    if ocr_confidence >= 0.75:
        return "human_review"
    return "quarantine"

Common mistakes and gotchas

• Validating string currency against a number field. Leaving "$12,500" to hit a number rule throws a type error and bounces a valid RFI, while a “helpful” float(x or 0) fallback turns an unparseable amount into 0 and understates the change-order exposure that estimators are tracking. Normalize the currency string to a float in a field_validator(mode="before") before any bounds check runs, and let a genuinely empty value become null, not zero.
• Stopping at the first error. Calling .validate() raises on the first violation, so a coordinator fixes the rfi_number, resubmits, and only then discovers the bad status — three round trips for one document. Use iter_errors() to aggregate the full violation set into one structured payload the error handling protocols layer can act on in a single pass.
• Forgetting additionalProperties: false. Without it, a hallucinated OCR key such as page_2_footer passes validation and persists into the tracking database, where it confuses every consumer that did a SELECT *. Closing the schema is the cheapest defense against extraction noise, and it pairs with format: date to reject the impossible calendar dates OCR loves to invent from smudged stamps.

Integration pointer

This validation gate sits immediately after extraction and before routing. Upstream, the bytes it inspects come from the field extraction techniques stage running on documents that already survived OCR preprocessing for construction docs; the ocr_confidence score that drives routing is produced there. Downstream, only schema-clean records get published, and the masterformat_section validated here is the key that WBS mapping strategies uses to allocate the RFI to a cost code. The RFI contract itself is the implementation of the field conventions defined in RFI schema design — keep the JSON Schema and the API payload structure in lockstep so the gateway and the ingestion layer never disagree about what a valid RFI looks like.

Frequently asked questions

Related

• Schema Validation Rules — the validation layer this gate belongs to
• Best practices for structuring RFI JSON payloads for APIs — the payload contract this schema mirrors
• Field Extraction Techniques — the upstream stage that produces the dict validated here
• How to map CSI MasterFormat to custom WBS codes in Python — what the validated MasterFormat section feeds into
• Implementing retry logic for failed API document pulls — the network gate records pass before they reach this schema gate

← Back to Schema Validation Rules