NLP Intent Classification for CMMS Work Order Routing

NLP intent classification is the semantic-routing stage of the Work Order Ingestion & Parsing Pipelines domain — the component that takes a normalized, free-text maintenance request and resolves it into a discrete intent label that drives trade assignment, priority, and SLA fields on the outgoing work order.

By the time text reaches this stage it has already been captured, decoded, and stripped of transport artifacts upstream; this component does not poll mailboxes or parse documents. It reads a clean request string, predicts an intent such as HVAC_REPAIR, ELECTRICAL_EMERGENCY, or PREVENTIVE_FILTER_CHANGE, attaches a confidence score, and either resolves the routing fields automatically or hands a low-confidence request to human triage. Treating routing as a classification problem — rather than brittle keyword matching — is what lets one pipeline absorb the wildly inconsistent phrasing technicians and tenants actually submit (“AC dripping in 204”, “breaker keeps tripping panel B”, “quarterly filter swap due”) without a hand-written rule per phrase. This guide implements that stage end to end: prerequisites, the input/output data contract, a step-by-step Python build, a configuration reference, validation checks, and the failure modes you will hit in production.

Prerequisites

This component runs as a stateless, in-process classifier that downstream batching calls per request. Before you deploy it, confirm the following are in place.

• Python 3.11+ with spacy>=3.7 for the text-classification model, pydantic>=2.6 for routing-matrix validation, and httpx>=0.27 for the asynchronous CMMS dispatch call. No mail or document libraries belong in this stage — text arrives already extracted.
• A trained intent model on disk or installed as a package. The code below loads a fine-tuned pipeline named cmms_intent_router_v2 whose textcat component is configured for exclusive (single-label) classification. Building and versioning that model is covered in training spaCy models for maintenance intent routing.
• CMMS REST API v1 with write access to the work-order resource (POST /api/v1/work-orders) and read access to the trade roster. The classifier never writes back to the asset registry; it only emits a payload the dispatch layer posts.
• Environment variables: CMMS_BASE_URL, CMMS_API_TOKEN, INTENT_MODEL_NAME (default cmms_intent_router_v2), and INTENT_CONFIDENCE_THRESHOLD (default 0.75). The token must carry the workorders:write scope; a token missing it fails closed at startup rather than silently dropping every dispatch.

Architecture and Data Contract

The component sits between text normalization and crew dispatch. It consumes a single normalized request string plus the asset it concerns, and emits one routing-resolved WorkOrderPayload — never the reverse. Four boundaries keep the stage honest and stop upstream phrasing noise from leaking into the dispatch layer:

• Input boundary: the classifier accepts clean text only. Markup, signatures, and quoted reply chains are removed before this point, whether the request originated from email intake configuration or from a document run through PDF parsing with Python. The classifier trusts the string and never re-decodes transport bytes.
• Inference boundary: the model returns a probability distribution across the fixed intent vocabulary. The stage picks the top label and its confidence; nothing here mutates state.
• Decision boundary: confidence is compared against the configured threshold. Above it, routing resolves automatically; below it, the request is flagged for human triage instead of being force-routed to a guessed trade.
• Output boundary: a resolved intent maps through a version-controlled routing matrix into the canonical WorkOrderPayload, including the SLA fields. The classification scope terminates the moment that payload is constructed; dispatch and retry belong to the downstream stage.

The contract across this stage is explicit. The input is a normalized str plus an asset_id; the output is a WorkOrderPayload whose SLA fields are populated from the matched intent. Modeling the routing matrix entries as a validated structure means an unmapped intent or a malformed priority is rejected at the boundary instead of producing a work order no crew can action.

Step-by-Step Implementation

1. Define the canonical work order payload

Routing produces the same WorkOrderPayload used everywhere on this site, so the SLA fields stay identical across the pipeline. The SLA fields — priority, requested_completion, and escalation_tier — are exactly what the routing matrix populates from the matched intent. Field-level schema rules live in work order schema standards.

from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Dict, List, Optional


class Priority(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    STANDARD = "standard"
    PLANNED = "planned"


@dataclass
class WorkOrderPayload:
    """Canonical CMMS work order — SLA fields are mandatory site-wide."""
    work_order_id: str
    asset_id: str
    part_skus: List[str]
    required_quantities: Dict[str, int]
    priority: Priority = Priority.STANDARD
    requested_completion: Optional[datetime] = None
    escalation_tier: int = 0
    status: str = "open"
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

2. Define the classifier result contract

The classifier returns a small, explicit result rather than a raw spaCy Doc, so the decision layer reads only what it needs: the label, the confidence, and a precomputed flag for whether the request must go to triage.

from dataclasses import dataclass


@dataclass
class IntentResult:
    """Output of the inference step — drives the routing decision."""
    label: str
    confidence: float
    requires_human_review: bool

3. Normalize the incoming request text

Even “clean” upstream text carries casing inconsistencies, collapsed line breaks, and stray reference numbers that add no maintenance signal. A deterministic normalizer keeps the model’s attention on the words that actually distinguish a leak from a filter swap, which directly raises confidence margins.

import re

_WHITESPACE = re.compile(r"\s+")
_TICKET_REF = re.compile(r"\b(?:ref|ticket|wo)[-#:\s]*\d+\b", re.IGNORECASE)


def normalize_request(subject: str, body: str) -> str:
    """Concatenate subject and body into a single clean classifier input."""
    combined = f"{subject} ||| {body}"
    combined = _TICKET_REF.sub(" ", combined)        # drop internal reference IDs
    combined = combined.replace("|||", " ")          # delimiter was for joining only
    combined = _WHITESPACE.sub(" ", combined)        # collapse runs of whitespace
    return combined.strip().lower()

4. Run inference and apply the confidence gate

The model loads once at module scope — reloading a spaCy pipeline per request is the single most common throughput killer in this stage. A forward pass yields doc.cats, the probability distribution; the stage takes the argmax and compares it against the configured threshold to set the triage flag.

import os
import spacy

_nlp = None  # module-level cache; load the model exactly once per process.


def get_nlp():
    global _nlp
    if _nlp is None:
        _nlp = spacy.load(os.environ.get("INTENT_MODEL_NAME", "cmms_intent_router_v2"))
    return _nlp


def classify_intent(text: str, threshold: float = 0.75) -> IntentResult:
    """Predict a single intent label and flag low-confidence requests for triage."""
    doc = get_nlp()(text)
    scores = doc.cats  # {"HVAC_REPAIR": 0.91, "PLUMBING_LEAK": 0.04, ...}
    if not scores:
        return IntentResult(label="UNKNOWN", confidence=0.0, requires_human_review=True)
    best_label = max(scores, key=scores.get)
    confidence = float(scores[best_label])
    return IntentResult(
        label=best_label,
        confidence=confidence,
        requires_human_review=confidence < threshold,
    )

5. Resolve the routing matrix into SLA fields

The matched intent maps to a routing matrix that defines work type, the SLA priority, the response window, and the certified trade group. Keep this matrix in a version-controlled configuration file (YAML or JSON) so facilities teams adjust dispatch rules without a redeploy. Validating each entry with pydantic means a typo in a priority tier is caught at load time, not at dispatch. The trade group resolves against the same asset taxonomy defined in asset hierarchy design.

from datetime import datetime, timedelta, timezone
from typing import Dict
from pydantic import BaseModel, Field


class RouteRule(BaseModel):
    """One row of the routing matrix; validated at load time."""
    work_type: str
    priority: Priority
    response_minutes: int = Field(..., gt=0)
    trade_group: str
    escalation_tier: int = Field(default=0, ge=0, le=3)


ROUTING_MATRIX: Dict[str, RouteRule] = {
    "ELECTRICAL_EMERGENCY": RouteRule(
        work_type="EMERGENCY", priority=Priority.CRITICAL,
        response_minutes=30, trade_group="ELECTRICAL", escalation_tier=3,
    ),
    "PLUMBING_LEAK": RouteRule(
        work_type="CORRECTIVE", priority=Priority.HIGH,
        response_minutes=120, trade_group="PLUMBING", escalation_tier=1,
    ),
    "HVAC_REPAIR": RouteRule(
        work_type="CORRECTIVE", priority=Priority.STANDARD,
        response_minutes=480, trade_group="HVAC", escalation_tier=0,
    ),
    "PREVENTIVE_FILTER_CHANGE": RouteRule(
        work_type="PREVENTIVE", priority=Priority.PLANNED,
        response_minutes=10_080, trade_group="HVAC", escalation_tier=0,
    ),
}


def resolve_route(intent: IntentResult) -> RouteRule:
    """Look up the SLA route for a confident intent; raise on an unmapped label."""
    rule = ROUTING_MATRIX.get(intent.label)
    if rule is None:
        raise KeyError(f"Unmapped intent: {intent.label!r}")
    return rule

6. Build the work order payload and dispatch to the CMMS

The final step assembles a WorkOrderPayload from the resolved route, stamping the SLA fields the matrix supplied, and posts it. A request flagged for triage short-circuits here — it returns a triage envelope instead of a dispatch, so a guessed trade never reaches a crew. An idempotency key prevents a network retry from creating a duplicate work order.

import uuid
import httpx
from datetime import datetime, timezone
from typing import Any, Dict


def build_payload(intent: IntentResult, asset_id: str) -> WorkOrderPayload:
    """Turn a confident intent into a routable, SLA-stamped work order."""
    rule = resolve_route(intent)
    now = datetime.now(timezone.utc)
    return WorkOrderPayload(
        work_order_id=f"WO-{uuid.uuid4().hex[:12]}",
        asset_id=asset_id,
        part_skus=[],
        required_quantities={},
        priority=rule.priority,
        requested_completion=now + timedelta(minutes=rule.response_minutes),
        escalation_tier=rule.escalation_tier,
    )


async def route_request(
    text: str, asset_id: str, base_url: str, api_token: str, threshold: float = 0.75
) -> Dict[str, Any]:
    """Classify, gate on confidence, and dispatch — or return a triage envelope."""
    intent = classify_intent(text, threshold=threshold)
    if intent.requires_human_review:
        return {"status": "QUEUED_FOR_TRIAGE", "intent": intent.label,
                "confidence": intent.confidence}

    payload = build_payload(intent, asset_id)
    body = {
        "work_order_id": payload.work_order_id,
        "asset_id": payload.asset_id,
        "work_type": resolve_route(intent).work_type,
        "priority": payload.priority.value,
        "requested_completion": payload.requested_completion.isoformat(),
        "escalation_tier": payload.escalation_tier,
        "confidence_score": intent.confidence,
        "idempotency_key": str(uuid.uuid4()),
    }
    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.post(
            f"{base_url}/api/v1/work-orders",
            json=body,
            headers={"Authorization": f"Bearer {api_token}"},
        )
        resp.raise_for_status()
        return {"status": "AUTO_DISPATCHED", **resp.json()}

Routing one request at a time is correct for a low-volume mailbox, but a busy facility should classify inside the worker pool described in async batch processing so a burst of inbound requests never blocks the intake thread. Parts the matched intent implies — a filter, a seal, a breaker — are reserved separately through parts availability checks before the crew is dispatched.

Configuration Reference

Keep every tunable in a version-controlled configuration registry, not in the classifier source. The defaults below are conservative starting points for a mixed corrective/preventive workload.

Validation and Testing

Routing must be reproducible and confidence-aware, so the highest-value tests assert two things: a high-confidence prediction resolves to the right SLA fields, and a below-threshold prediction is diverted to triage rather than dispatched. Stubbing the classifier output keeps the test independent of model weights so it stays deterministic.

from datetime import datetime, timezone


def test_low_confidence_request_is_triaged():
    intent = IntentResult(label="HVAC_REPAIR", confidence=0.41,
                          requires_human_review=True)
    assert intent.requires_human_review is True
    # A triaged request must never resolve a route or stamp SLA fields.


def test_confident_intent_resolves_sla_fields():
    intent = IntentResult(label="ELECTRICAL_EMERGENCY", confidence=0.93,
                          requires_human_review=False)
    rule = resolve_route(intent)
    payload = build_payload(intent, asset_id="ELE-Panel-B")

    assert rule.priority is Priority.CRITICAL
    assert payload.escalation_tier == 3
    # 30-minute response window lands within the next hour.
    assert payload.requested_completion > datetime.now(timezone.utc)
    assert payload.work_order_id.startswith("WO-")

On a successful run the dispatch step returns {"status": "AUTO_DISPATCHED", ...} and the CMMS responds 201 Created with the work order it persisted; a triaged request returns {"status": "QUEUED_FOR_TRIAGE", ...} and never reaches the API. Assert against both branches in integration tests, and log the confidence score on every dispatch so you can chart the score distribution over time — a creeping leftward shift is the earliest signal of model drift.

Failure Modes and Troubleshooting

Expand each scenario for the root cause, the diagnostic signal, and the fix. The checklist items render as interactive checkboxes — work through them in order.

Everything routes to one trade or to triage

• Confirm the loaded model is the one you trained — print get_nlp().pipe_names and the model name; a stale or default pipeline returns a flat distribution that argmax resolves to the first label.
• Check that exclusive_classes was true during training; a multi-label model spreads probability so no score clears the threshold and every request is triaged.
• Verify the normalizer is not stripping the signal words — an over-aggressive regex that removes “leak” or “tripping” collapses distinct intents into one.

Confidence scores are suspiciously high on wrong predictions

• Inspect the training class balance; a label that dominated the training set produces overconfident predictions on out-of-distribution text. Retrain with balanced sampling per training spaCy models for maintenance intent routing.
• Confirm max_text_chars truncation is not cutting the body before the meaningful clause, leaving only a boilerplate greeting the model has overfit to.
• Compare the live confidence distribution against the validation baseline; a divergence means drift, and the threshold should hold requests in triage until a retrain ships.

KeyError: Unmapped intent at dispatch

• The model emits a label the routing matrix has no row for — a new intent was added to the training data but not to ROUTING_MATRIX. Add the route or map the label to an existing trade.
• Check the matrix loaded from the configuration file actually parsed; a pydantic ValidationError on a bad priority value can leave the matrix partially populated.
• Keep the model’s intent vocabulary and the matrix keys under the same review so they can never drift apart.

Throughput collapses under load

• Confirm the model is cached at module scope and not reloaded per call — a spacy.load inside classify_intent reloads weights on every request.
• Move classification into the worker pool from async batch processing so inference never blocks the intake thread.
• Use nlp.pipe() for batched inference when scoring many requests at once instead of calling the model per string.

Frequently Asked Questions

Why use a trained classifier instead of keyword rules?

Keyword rules break the moment a tenant writes “AC won’t cool” instead of “HVAC repair,” and the rule set grows unmaintainable as exceptions accumulate. A trained textcat model generalizes across phrasing, returns a calibrated confidence you can gate on, and degrades gracefully to triage when it is unsure — none of which a static keyword table offers. Rules are still useful as a guardrail, but the primary routing decision belongs to the classifier.

How should I set the confidence threshold?

Start at 0.75 and tune against the cost of a misroute versus the cost of triage volume. Emergency-heavy queues, where sending an electrical fire to the plumbing crew is unacceptable, justify a higher threshold and more triage; high-volume planned maintenance can tolerate a lower one. Chart the live confidence distribution and pick the threshold where precision on the auto-dispatched slice meets your SLA tolerance.

What happens to a request the model is unsure about?

It is diverted to a human triage queue rather than force-routed to a guessed trade. The requires_human_review flag short-circuits route_request before any dispatch, so a low-confidence prediction never produces a work order. Reviewing those triaged requests also yields the highest-value labeled examples for the next training round.

How do I detect and respond to model drift?

Log the confidence score and predicted label on every request, then compare the rolling distribution against the baseline captured at training time. A steady leftward shift in confidence, or a rising triage rate, signals that incoming language has moved away from the training data. Hold routing decisions in triage and retrain on the freshly labeled triage backlog rather than lowering the threshold to mask the drift.

Related

Feed this stage from email intake configuration, pull text out of attachments with PDF parsing with Python, scale inference inside async batch processing, build the model itself in training spaCy models for maintenance intent routing, and align the emitted fields with work order schema standards.

Part of: Work Order Ingestion & Parsing Pipelines.