RFI Schema Design

Request for Information (RFI) workflows are the upstream trigger for most downstream change management: a single field question can cascade into a schedule adjustment, a budget reallocation, and a contractually binding change order. The specific sub-problem this page solves is how a pipeline turns a free-form RFI — submitted as a PDF, an email thread, or a portal form — into a typed, machine-readable event that automation can route deterministically. When that event is missing or loosely structured, RFIs sit in inboxes past their response deadline, cost exposure stays invisible until the monthly draw, and the same inquiry gets logged twice under two different location strings. Inside a deterministic construction data architecture and taxonomy, the RFI schema is the contract that makes an inquiry routable: it pins down identity, classification, and quantified impact so the record can drive escalation and change order generation without a human re-keying it. This page details the ingestion-to-routing pipeline for that contract — the schema itself, the idempotent normalization that resolves field references to canonical scope, and the confidence-scored routing that decides whether an RFI auto-routes, waits for review, or is quarantined. It targets Python automation builders, project engineers, and estimators who need predictable RFI data under real-world input variance.

Prerequisites

This subsystem sits downstream of document extraction and upstream of the change order ledger and the approval router. Before implementing the patterns below, you need:

• Python 3.11+ with pydantic v2 for typed validation, plus the standard-library decimal, re, difflib, enum, uuid, and logging modules. No floating-point money touches an impact field; every monetary value is a Decimal.
• A canonical scope taxonomy to resolve against. RFI locations and disciplines must align with the same Work Breakdown Structure that drives WBS mapping strategies; the RFI carries what was asked and where, the WBS node carries where that scope sits in the project.
• A standardized cost vocabulary so quantified impacts post against real accounts. Impact fields reference the canonical keys defined by budget code standardization rather than ad-hoc cost labels.
• A task queue — Celery on a Redis or RabbitMQ broker — so malformed or low-confidence RFIs can be parked in a dead-letter queue and replayed instead of dropped. The escalation policy for parked and SLA-breached records is owned by fallback alert routing.
• An upstream extraction step that has produced raw field strings and a per-field confidence score. RFIs lifted from scanned drawings carry confidence metadata from the ingestion pipeline; the routing logic below depends on it.

The pipeline assumes inbound payloads have already cleared structural schema validation rules at the API gateway, so the work here is RFI-specific normalization, impact validation, and routing — not document parsing.

Architecture: states, inputs, and routing

An RFI is not a static document; it is a record that moves through a finite set of states, and every transition is a place where automation either advances the record or parks it. The schema has to support two orthogonal concerns at once: a lifecycle state machine (Draft → Submitted → In Review → Answered → Closed, with an Escalated branch) and an ingestion pipeline that normalizes and routes each inbound payload. Keeping these separate is what lets a high-impact RFI escalate without breaking its lifecycle invariants. The state machine below governs the legal transitions a single RFI may take.

The ingestion pipeline runs orthogonally to the lifecycle: a raw payload flows through classification normalization, WBS resolution, impact validation, and routing. Each stage has its own failure branch, and the routing decision uses the site-canonical confidence bands — a classification match of 0.92 or above auto-routes, 0.75–0.92 parses but flags the record for human review, and below 0.75 the record is quarantined to the dead-letter queue rather than committed against a guessed scope.

Step-by-step implementation

Step 1 — Define the RFI schema contract

The schema is a versioned, typed contract. Every payload declares a schema_version at the root so a field addition never silently breaks a downstream consumer, and identity fields (project_uuid, rfi_number, created_at) are immutable once minted. The discipline is a controlled vocabulary expressed as a Literal, not a free string, so cross-discipline reporting can aggregate without a fragile string comparison. Timestamps are timezone-aware per the ISO 8601 date and time standard; a naive timestamp is rejected at the boundary because it would corrupt SLA math across project sites in different zones.

import uuid
from datetime import datetime
from decimal import Decimal, InvalidOperation
from enum import Enum
from typing import Literal, Optional

from pydantic import BaseModel, ConfigDict, Field, field_validator

# Discipline is a controlled vocabulary, never a free string:
# ARCHitectural, STRuctural, MEP, CIVil, ELECtrical, PLuMBing.
Discipline = Literal["ARCH", "STR", "MEP", "CIV", "ELEC", "PLMB"]


class RFIStatus(str, Enum):
    DRAFT = "Draft"
    SUBMITTED = "Submitted"
    IN_REVIEW = "In Review"
    ANSWERED = "Answered"
    CLOSED = "Closed"
    ESCALATED = "Escalated"


class CostImpact(BaseModel):
    model_config = ConfigDict(strict=True)
    currency: Literal["USD", "CAD", "EUR"]
    amount: Decimal = Field(ge=0, decimal_places=2)
    # Canonical budget code from the standardized cost vocabulary.
    budget_code: str = Field(pattern=r"^\d{6}[A-Z]{3}\d{3}$")

    @field_validator("amount", mode="before")
    @classmethod
    def coerce_amount(cls, v: object) -> Decimal:
        # Coerce ints/strings to Decimal at the boundary; never let a float
        # subtraction drift a cost rollup three systems downstream.
        try:
            return Decimal(str(v))
        except InvalidOperation as exc:
            raise ValueError(f"Invalid cost amount: {v!r}") from exc


class ScheduleImpact(BaseModel):
    model_config = ConfigDict(strict=True)
    days_delta: int
    affected_activity_ids: list[str] = Field(min_length=1)


class ClassificationBlock(BaseModel):
    model_config = ConfigDict(strict=True)
    discipline: Discipline
    trade: str = Field(pattern=r"^[A-Z0-9_-]+$")
    location_raw: str
    # Canonical WBS element: PROJ-NNN-DIV-NN, resolved during ingestion.
    wbs_resolved: Optional[str] = Field(default=None, pattern=r"^[A-Z]{3,5}-\d{3}-\d{2}-\d{2}$")


class RFIPayload(BaseModel):
    model_config = ConfigDict(strict=True)
    schema_version: str = Field(pattern=r"^v\d+\.\d+$")
    project_uuid: uuid.UUID
    rfi_number: str = Field(pattern=r"^RFI-\d{4}-\d{3}$")
    created_at: datetime
    status: RFIStatus
    classification: ClassificationBlock
    cost_impact: Optional[CostImpact] = None
    schedule_impact: Optional[ScheduleImpact] = None
    impact_confidence: Literal["Preliminary", "Verified", "Contractually_Bound"] = "Preliminary"

    @field_validator("created_at")
    @classmethod
    def require_timezone(cls, v: datetime) -> datetime:
        # A naive timestamp would break SLA windows across project time zones.
        if v.tzinfo is None:
            raise ValueError("created_at must be timezone-aware (ISO 8601)")
        return v

Constraining each field with a regex or Literal means a malformed payload is rejected at construction time with a precise error path, rather than corrupting an escalation decision later. Keeping cost_impact and schedule_impact optional lets a field engineer submit a preliminary inquiry quickly, while the routing layer (Step 4) enforces that impacts are present before the record may advance to Answered.

Step 2 — Normalize classification deterministically

Field-generated RFIs arrive with wild terminology variance: "Elec.", "electrical", and "E" all mean the same discipline, and location strings carry trailing whitespace, inconsistent casing, and ad-hoc delimiters. Normalization must be a pure, idempotent transformation — the same raw input always yields the same cleaned tokens, and running it twice changes nothing. Idempotency is what makes pipeline retries safe: a redelivered message during a broker hiccup must not produce a divergent classification.

import re

_DELIM = re.compile(r"[^A-Z0-9]+")

# Map common field variants onto the canonical discipline vocabulary.
_DISCIPLINE_ALIASES: dict[str, str] = {
    "ARCHITECTURAL": "ARCH", "ARCH": "ARCH", "A": "ARCH",
    "STRUCTURAL": "STR", "STRUCT": "STR", "S": "STR",
    "MECHANICAL": "MEP", "ELECTRICAL_MECH": "MEP", "MEP": "MEP",
    "CIVIL": "CIV", "CIV": "CIV", "C": "CIV",
    "ELECTRICAL": "ELEC", "ELEC": "ELEC", "E": "ELEC",
    "PLUMBING": "PLMB", "PLUMB": "PLMB", "P": "PLMB",
}


def normalize_discipline(raw: str) -> str:
    """Resolve a free-form discipline string to the canonical code.

    Pure and idempotent: the canonical codes map to themselves, so a
    second pass is a no-op. Unknown disciplines raise rather than guess.
    """
    key = _DELIM.sub("_", raw.strip().upper()).strip("_")
    if key in _DISCIPLINE_ALIASES:
        return _DISCIPLINE_ALIASES[key]
    raise ValueError(f"Unknown discipline: {raw!r}")


def clean_location(raw: str) -> str:
    """Collapse delimiters and casing so location lookups are stable."""
    return _DELIM.sub("_", raw.strip().upper()).strip("_")

Isolating normalization from business logic keeps it testable on its own and guarantees it can never accidentally depend on ledger or queue state.

Step 3 — Resolve location to a canonical WBS node by confidence

A cleaned location string still has to resolve to a real scope node before any financial workflow can run. The pipeline first attempts an exact match against the master WBS map, then falls back to fuzzy matching, and the site-canonical confidence bands decide the record’s fate. This is the same routing vocabulary used across every subsystem, so an engineer reading a held RFI sees the same thresholds they see on a held budget record. Resolving scope correctly here is what keeps earned-value rollups and field extraction techniques downstream from inheriting a misfiled location.

from dataclasses import dataclass
from difflib import SequenceMatcher

# Site-canonical routing bands, applied here to WBS-match confidence.
AUTO_ROUTE = 0.92
HUMAN_REVIEW = 0.75

RoutingState = Literal["AUTO_ROUTE", "HUMAN_REVIEW", "QUARANTINE"]


@dataclass(frozen=True)
class WBSResolution:
    wbs_node: Optional[str]
    confidence: float
    routing_state: RoutingState


def resolve_wbs(location: str, master_wbs_map: dict[str, str]) -> WBSResolution:
    """Resolve a cleaned location string to a canonical WBS node by band."""
    if location in master_wbs_map:
        return WBSResolution(master_wbs_map[location], 1.0, "AUTO_ROUTE")

    best_node, best_score = None, 0.0
    for candidate, node in master_wbs_map.items():
        score = SequenceMatcher(None, location, candidate).ratio()
        if score > best_score:
            best_node, best_score = node, score

    if best_score >= AUTO_ROUTE:
        return WBSResolution(best_node, best_score, "AUTO_ROUTE")
    if best_score >= HUMAN_REVIEW:
        return WBSResolution(best_node, best_score, "HUMAN_REVIEW")
    return WBSResolution(None, best_score, "QUARANTINE")

Step 4 — Validate, then route by impact and SLA

The final stage assembles the validated RFIPayload, then makes the routing decision. Two gates matter: impacts must be present and quantified before an RFI may advance to Answered, and the magnitude of those impacts determines whether the record auto-approves, enters the approval queue, or escalates. Threshold breaches and SLA overruns hand off to the fallback alert router rather than blocking in the pipeline.

import logging

logger = logging.getLogger("rfi_schema")

# Auto-approve clarifications under these thresholds; escalate above them.
COST_ESCALATION = Decimal("25000.00")
SCHEDULE_ESCALATION_DAYS = 5


def ingest_rfi(payload: dict, master_wbs_map: dict[str, str]) -> tuple[RFIPayload, RoutingState]:
    """Normalize, resolve scope, validate, and return the record + routing.

    Raises ValueError on any unrecoverable state so the caller can park the
    record in the dead-letter queue instead of committing a guessed scope.
    """
    cls = payload["classification"]
    cls["discipline"] = normalize_discipline(cls["discipline"])
    resolution = resolve_wbs(clean_location(cls["location_raw"]), master_wbs_map)

    if resolution.routing_state == "QUARANTINE":
        raise ValueError(f"Unresolvable location: {cls['location_raw']!r}")
    cls["wbs_resolved"] = resolution.wbs_node

    record = RFIPayload.model_validate(payload)  # raises ValidationError on bad data

    # Accountability gate: no transition to Answered without quantified impact.
    if record.status is RFIStatus.ANSWERED and record.cost_impact is None:
        raise ValueError("Answered RFI must carry a quantified cost impact")

    routing = route_rfi(record)
    if resolution.routing_state == "HUMAN_REVIEW":
        routing = "HUMAN_REVIEW"
        logger.warning(
            "RFI %s WBS matched at %.2f; flag for review", record.rfi_number, resolution.confidence
        )
    return record, routing


def route_rfi(record: RFIPayload) -> RoutingState:
    """Decide routing from quantified impact magnitude."""
    cost = record.cost_impact.amount if record.cost_impact else Decimal("0")
    days = record.schedule_impact.days_delta if record.schedule_impact else 0
    if cost >= COST_ESCALATION or days >= SCHEDULE_ESCALATION_DAYS:
        return "HUMAN_REVIEW"  # high impact: route to approval / escalation queue
    return "AUTO_ROUTE"        # low-impact clarification: auto-approve


if __name__ == "__main__":
    wbs_map = {"LEVEL_3_EAST_CORE": "TWR-103-09-02"}
    sample = {
        "schema_version": "v1.0",
        "project_uuid": str(uuid.uuid4()),
        "rfi_number": "RFI-2026-014",
        "created_at": "2026-06-27T14:30:00-05:00",
        "status": "Submitted",
        "classification": {
            "discipline": "electrical",
            "trade": "EC-DIV26",
            "location_raw": " Level 3 East Core ",
        },
        "cost_impact": {"currency": "USD", "amount": "31500.00", "budget_code": "260000SUB014"},
        "impact_confidence": "Verified",
    }
    rfi, state = ingest_rfi(sample, wbs_map)
    print(f"{rfi.rfi_number} -> {rfi.classification.wbs_resolved} :: {state}")

Schema and configuration reference

Treat these field constraints as part of the contract; downstream routing, cost rollups, and ERP mappings depend on the exact patterns.

Routing and threshold keys, used identically wherever RFI routing runs:

Verification and testing

Prove that each branch is deterministic and that bad input produces a structured outcome, never a silent default.

import pytest

WBS_MAP = {"LEVEL_3_EAST_CORE": "TWR-103-09-02"}


def base_payload(**overrides) -> dict:
    payload = {
        "schema_version": "v1.0",
        "project_uuid": str(uuid.uuid4()),
        "rfi_number": "RFI-2026-014",
        "created_at": "2026-06-27T14:30:00-05:00",
        "status": "Submitted",
        "classification": {"discipline": "ELEC", "trade": "EC-DIV26", "location_raw": "Level 3 East Core"},
    }
    payload.update(overrides)
    return payload


def test_discipline_normalization_is_idempotent():
    assert normalize_discipline("electrical") == "ELEC"
    assert normalize_discipline(normalize_discipline("electrical")) == "ELEC"


def test_unresolvable_location_quarantines():
    bad = base_payload(classification={"discipline": "ELEC", "trade": "EC", "location_raw": "ZZZ Nowhere"})
    with pytest.raises(ValueError):
        ingest_rfi(bad, WBS_MAP)


def test_high_cost_routes_to_review():
    p = base_payload(cost_impact={"currency": "USD", "amount": "31500.00", "budget_code": "260000SUB014"})
    _, state = ingest_rfi(p, WBS_MAP)
    assert state == "HUMAN_REVIEW"


def test_naive_timestamp_rejected():
    p = base_payload(created_at="2026-06-27T14:30:00")  # no offset
    with pytest.raises(Exception):
        ingest_rfi(p, WBS_MAP)


def test_answered_requires_cost_impact():
    p = base_payload(status="Answered")
    with pytest.raises(ValueError):
        ingest_rfi(p, WBS_MAP)

Run the suite with python -m pytest tests/test_rfi_schema.py -v. A green run confirms that classification is idempotent, that WBS resolution routes by confidence band, that naive timestamps are rejected, and that the accountability gate holds.

Troubleshooting

Naive timestamps break SLA escalation. An RFI submitted from a portal that emits local time without an offset validates everywhere except the require_timezone check, or worse, slips through a looser schema and computes a negative time-to-breach. Root cause: the timestamp has no zone. Fix: reject naive datetimes at the boundary, and normalize all created_at values to UTC in the extraction layer while preserving the original offset in the audit trail.

Fuzzy WBS matches file the RFI against the wrong scope. Two adjacent locations — Level 3 East Core and Level 3 East Corridor — score above 0.92, and an RFI auto-routes to the wrong WBS node, dragging its cost impact into the wrong rollup. Root cause: the auto-route band is too permissive for a dense location map. Fix: require an exact match for auto-route on short, similar tokens, and push every fuzzy match into the human-review band until the location map is disambiguated.

European decimal formats quarantine valid impacts. A subcontractor submits a cost impact of 1.250,00 and the Decimal coercion fails, parking a legitimate RFI. Root cause: the validator assumes US ,/. conventions. Fix: normalize numeric strings — strip thousands separators and standardize the decimal point — in extraction before they reach the schema, and keep the raw string in the audit trail.

Unknown discipline strings drop RFIs silently. A new trade abbreviation appears in the field and normalize_discipline raises, but the caller swallows the exception, so the RFI never lands anywhere. Root cause: an unmapped alias plus a silent except. Fix: route unknown disciplines to the dead-letter queue with the raw token attached so the alias map can be extended, and never bare-except around normalization.

Duplicate RFIs after a broker retry. A redelivered message re-ingests the same RFI and creates a second ledger event. Root cause: ingestion is idempotent but the commit is not. Fix: key the commit on rfi_number plus project_uuid so a replay updates the existing record in place rather than inserting a duplicate.

Frequently Asked Questions

Related

• Best practices for structuring RFI JSON payloads for APIs
• WBS Mapping Strategies
• Budget Code Standardization
• Schema Validation Rules
• Fallback Alert Routing

← Back to Construction Data Architecture & Taxonomy