Document Classification with NLP

Overview

This skill implements NLP-based document classification and information extraction for construction projects. Automate document sorting, key term extraction, and content analysis.

Document Types:

• RFIs (Requests for Information)
• Submittals and shop drawings
• Change orders and variations
• Specifications and standards
• Contracts and agreements
• Safety reports and permits

Quick Start

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
import pandas as pd

# Sample training data
documents = [
    ("Please clarify the steel reinforcement spacing for the foundation slab", "RFI"),
    ("Attached shop drawing for HVAC ductwork layout", "Submittal"),
    ("Additional cost for unforeseen soil conditions", "Change Order"),
    ("Fire-rated wall assembly specification Section 09 21 16", "Specification"),
]

texts, labels = zip(*documents)

# Train classifier
classifier = Pipeline([
    ('tfidf', TfidfVectorizer(max_features=1000, ngram_range=(1, 2))),
    ('clf', MultinomialNB())
])

classifier.fit(texts, labels)

# Classify new document
new_doc = "Request to approve substitution of specified light fixtures"
prediction = classifier.predict([new_doc])[0]
print(f"Classification: {prediction}")  # Output: Submittal

Advanced Classification System

Document Classifier Class

import re
import pandas as pd
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.svm import LinearSVC
from sklearn.ensemble import RandomForestClassifier
from sklearn.pipeline import Pipeline
from sklearn.model_selection import cross_val_score
from sklearn.preprocessing import LabelEncoder
from typing import List, Dict, Tuple, Optional
import spacy
from dataclasses import dataclass

@dataclass
class ClassificationResult:
    document_id: str
    predicted_class: str
    confidence: float
    alternative_classes: List[Tuple[str, float]]
    extracted_entities: Dict[str, List[str]]
    keywords: List[str]

class ConstructionDocumentClassifier:
    """Classify and analyze construction documents"""

    # Document type patterns
    DOCUMENT_PATTERNS = {
        'RFI': [
            r'request\s+for\s+information',
            r'clarification\s+(needed|required|requested)',
            r'please\s+(clarify|confirm|advise)',
            r'question\s+(regarding|about)',
            r'rfi\s*#?\d*'
        ],
        'Submittal': [
            r'submittal',
            r'shop\s+drawing',
            r'product\s+data',
            r'sample\s+submission',
            r'approval\s+request',
            r'material\s+submission'
        ],
        'Change Order': [
            r'change\s+order',
            r'variation\s+order',
            r'cost\s+(increase|adjustment|addition)',
            r'scope\s+change',
            r'additional\s+work',
            r'unforeseen\s+conditions'
        ],
        'Specification': [
            r'section\s+\d{2}\s+\d{2}\s+\d{2}',
            r'specification',
            r'performance\s+requirement',
            r'material\s+standard',
            r'quality\s+standard'
        ],
        'Safety Report': [
            r'incident\s+report',
            r'safety\s+(inspection|violation|observation)',
            r'hazard\s+(identification|assessment)',
            r'near\s+miss',
            r'osha',
            r'jha|jsa'
        ],
        'Contract': [
            r'contract\s+agreement',
            r'terms\s+and\s+conditions',
            r'scope\s+of\s+work',
            r'payment\s+terms',
            r'warranty\s+provision'
        ]
    }

    def __init__(self, use_spacy: bool = True):
        self.classifier = None
        self.vectorizer = None
        self.label_encoder = LabelEncoder()

        if use_spacy:
            try:
                self.nlp = spacy.load("en_core_web_sm")
            except:
                self.nlp = None
        else:
            self.nlp = None

    def train(self, documents: List[str], labels: List[str]) -> Dict:
        """Train the document classifier"""
        # Encode labels
        y = self.label_encoder.fit_transform(labels)

        # Create pipeline
        self.classifier = Pipeline([
            ('tfidf', TfidfVectorizer(
                max_features=5000,
                ngram_range=(1, 3),
                stop_words='english',
                sublinear_tf=True
            )),
            ('clf', LinearSVC(C=1.0, class_weight='balanced'))
        ])

        # Train
        self.classifier.fit(documents, y)

        # Cross-validation
        scores = cross_val_score(self.classifier, documents, y, cv=5)

        return {
            'accuracy_mean': scores.mean(),
            'accuracy_std': scores.std(),
            'classes': list(self.label_encoder.classes_)
        }

    def classify(self, document: str) -> ClassificationResult:
        """Classify a single document"""
        if self.classifier is None:
            # Use rule-based classification if no model trained
            return self._rule_based_classify(document)

        # Get prediction
        prediction = self.classifier.predict([document])[0]
        predicted_class = self.label_encoder.inverse_transform([prediction])[0]

        # Get confidence scores
        decision_scores = self.classifier.decision_function([document])[0]
        probs = self._softmax(decision_scores)

        alternatives = [
            (self.label_encoder.inverse_transform([i])[0], float(probs[i]))
            for i in np.argsort(probs)[::-1][1:4]
        ]

        # Extract entities and keywords
        entities = self._extract_entities(document)
        keywords = self._extract_keywords(document)

        return ClassificationResult(
            document_id="",
            predicted_class=predicted_class,
            confidence=float(probs[prediction]),
            alternative_classes=alternatives,
            extracted_entities=entities,
            keywords=keywords
        )

    def _rule_based_classify(self, document: str) -> ClassificationResult:
        """Rule-based classification using patterns"""
        doc_lower = document.lower()
        scores = {}

        for doc_type, patterns in self.DOCUMENT_PATTERNS.items():
            score = sum(
                1 for pattern in patterns
                if re.search(pattern, doc_lower)
            )
            scores[doc_type] = score

        if max(scores.values()) == 0:
            predicted = 'Other'
            confidence = 0.5
        else:
            predicted = max(scores, key=scores.get)
            confidence = scores[predicted] / len(self.DOCUMENT_PATTERNS[predicted])

        return ClassificationResult(
            document_id="",
            predicted_class=predicted,
            confidence=confidence,
            alternative_classes=[],
            extracted_entities=self._extract_entities(document),
            keywords=self._extract_keywords(document)
        )

    def _extract_entities(self, document: str) -> Dict[str, List[str]]:
        """Extract named entities from document"""
        entities = {
            'dates': [],
            'organizations': [],
            'people': [],
            'monetary': [],
            'references': []
        }

        # Date patterns
        date_pattern = r'\d{1,2}[/-]\d{1,2}[/-]\d{2,4}'
        entities['dates'] = re.findall(date_pattern, document)

        # Money patterns
        money_pattern = r'\$[\d,]+(?:\.\d{2})?'
        entities['monetary'] = re.findall(money_pattern, document)

        # Reference numbers
        ref_pattern = r'(?:RFI|CO|SI|PR)[-#]?\s*\d+'
        entities['references'] = re.findall(ref_pattern, document, re.IGNORECASE)

        # Use spaCy for NER if available
        if self.nlp:
            doc = self.nlp(document)
            for ent in doc.ents:
                if ent.label_ == 'ORG':
                    entities['organizations'].append(ent.text)
                elif ent.label_ == 'PERSON':
                    entities['people'].append(ent.text)

        return entities

    def _extract_keywords(self, document: str, top_n: int = 10) -> List[str]:
        """Extract key terms from document"""
        # Construction-specific terms
        construction_terms = [
            'concrete', 'steel', 'reinforcement', 'foundation', 'structural',
            'hvac', 'plumbing', 'electrical', 'mechanical', 'architectural',
            'specification', 'drawing', 'detail', 'schedule', 'submittals',
            'rfi', 'change order', 'delay', 'inspection', 'approval'
        ]

        doc_lower = document.lower()
        found_terms = [term for term in construction_terms if term in doc_lower]

        return found_terms[:top_n]

    def _softmax(self, x: np.ndarray) -> np.ndarray:
        """Convert decision scores to probabilities"""
        exp_x = np.exp(x - np.max(x))
        return exp_x / exp_x.sum()

    def batch_classify(self, documents: List[str]) -> pd.DataFrame:
        """Classify multiple documents"""
        results = [self.classify(doc) for doc in documents]

        return pd.DataFrame([{
            'Predicted_Class': r.predicted_class,
            'Confidence': r.confidence,
            'Keywords': ', '.join(r.keywords),
            'Dates_Found': ', '.join(r.extracted_entities['dates']),
            'References_Found': ', '.join(r.extracted_entities['references'])
        } for r in results])

Information Extraction

Key Information Extractor

class ConstructionInfoExtractor:
    """Extract key information from construction documents"""

    def __init__(self):
        self.patterns = {
            'rfi_number': r'RFI\s*[-#]?\s*(\d+)',
            'submittal_number': r'(?:Submittal|SI)\s*[-#]?\s*(\d+)',
            'change_order_number': r'(?:Change Order|CO|PCO)\s*[-#]?\s*(\d+)',
            'spec_section': r'Section\s*(\d{2}\s*\d{2}\s*\d{2})',
            'cost_amount': r'\$\s*([\d,]+(?:\.\d{2})?)',
            'duration_days': r'(\d+)\s*(?:calendar\s+)?days?',
            'drawing_reference': r'(?:Drawing|Dwg|DWG)\s*[-#]?\s*([A-Z\d-]+)',
            'date': r'(\d{1,2}[/-]\d{1,2}[/-]\d{2,4})',
            'contractor_name': r'(?:Contractor|Subcontractor):\s*([^\n]+)',
            'project_name': r'Project:\s*([^\n]+)',
            'priority': r'(?:Priority|Urgency):\s*(Critical|High|Medium|Low)'
        }

    def extract_all(self, document: str) -> Dict:
        """Extract all available information"""
        results = {}

        for field, pattern in self.patterns.items():
            matches = re.findall(pattern, document, re.IGNORECASE)
            results[field] = matches if matches else None

        # Post-process
        if results.get('cost_amount'):
            results['cost_amount'] = [
                float(amt.replace(',', ''))
                for amt in results['cost_amount']
            ]

        return results

    def extract_rfi_details(self, document: str) -> Dict:
        """Extract RFI-specific information"""
        return {
            'rfi_number': self._find_first(document, self.patterns['rfi_number']),
            'date_submitted': self._find_first(document, self.patterns['date']),
            'spec_section': self._find_first(document, self.patterns['spec_section']),
            'drawing_ref': self._find_first(document, self.patterns['drawing_reference']),
            'question': self._extract_question(document),
            'priority': self._find_first(document, self.patterns['priority'])
        }

    def extract_change_order_details(self, document: str) -> Dict:
        """Extract change order specific information"""
        costs = re.findall(self.patterns['cost_amount'], document)
        total_cost = sum(float(c.replace(',', '')) for c in costs) if costs else None

        return {
            'co_number': self._find_first(document, self.patterns['change_order_number']),
            'date': self._find_first(document, self.patterns['date']),
            'cost_impact': total_cost,
            'duration_impact': self._find_first(document, self.patterns['duration_days']),
            'reason': self._extract_reason(document),
            'contractor': self._find_first(document, self.patterns['contractor_name'])
        }

    def _find_first(self, document: str, pattern: str) -> Optional[str]:
        match = re.search(pattern, document, re.IGNORECASE)
        return match.group(1) if match else None

    def _extract_question(self, document: str) -> Optional[str]:
        """Extract the question from an RFI"""
        # Look for question markers
        patterns = [
            r'Question:\s*(.+?)(?:\n\n|$)',
            r'(?:Please\s+)?(?:clarify|confirm|advise)(.+?)(?:\.|$)',
        ]
        for pattern in patterns:
            match = re.search(pattern, document, re.IGNORECASE | re.DOTALL)
            if match:
                return match.group(1).strip()[:500]
        return None

    def _extract_reason(self, document: str) -> Optional[str]:
        """Extract reason for change order"""
        patterns = [
            r'Reason:\s*(.+?)(?:\n\n|$)',
            r'(?:Due to|Because of)\s*(.+?)(?:\.|$)',
        ]
        for pattern in patterns:
            match = re.search(pattern, document, re.IGNORECASE | re.DOTALL)
            if match:
                return match.group(1).strip()[:500]
        return None

Processing Pipeline

def process_document_batch(documents: List[str], output_path: str):
    """Process and classify a batch of documents"""
    classifier = ConstructionDocumentClassifier()
    extractor = ConstructionInfoExtractor()

    results = []

    for i, doc in enumerate(documents):
        # Classify
        classification = classifier.classify(doc)

        # Extract info based on type
        if classification.predicted_class == 'RFI':
            extracted = extractor.extract_rfi_details(doc)
        elif classification.predicted_class == 'Change Order':
            extracted = extractor.extract_change_order_details(doc)
        else:
            extracted = extractor.extract_all(doc)

        results.append({
            'Document_ID': i + 1,
            'Classification': classification.predicted_class,
            'Confidence': classification.confidence,
            'Keywords': ', '.join(classification.keywords),
            **extracted
        })

    df = pd.DataFrame(results)
    df.to_excel(output_path, index=False)

    return df

Quick Reference

Document Type	Key Patterns	Extracted Info
RFI	"request for information", "clarify"	Number, spec section, question
Submittal	"shop drawing", "approval request"	Number, product, spec section
Change Order	"change order", "additional cost"	Number, cost, duration impact
Specification	"Section XX XX XX"	Section number, requirements
Safety Report	"incident", "hazard"	Date, type, severity

Resources

• spaCy: https://spacy.io
• Scikit-learn: https://scikit-learn.org
• DDC Website: https://datadrivenconstruction.io

Next Steps

• See vector-search for semantic document search
• See llm-data-automation for advanced extraction
• See pdf-to-structured for PDF processing