| # **BuildwellTHREAD: Complete System Specification** | |
| ## **Multi-Agent AI System for UK Building Regulations Compliance & Golden Thread Management** | |
| --- | |
| **Document Version:** 1.0 | |
| **Date:** January 2026 | |
| **Status:** Development Specification | |
| **Target Platform:** Isambard AI Supercomputer (Nvidia GH200 Grace-Hopper) | |
| **Project Lead:** London Belgravia Surveyors | |
| --- | |
| ## **EXECUTIVE SUMMARY** | |
| BuildwellTHREAD is a comprehensive AI-powered system designed to automate UK Building Regulations compliance checking for Gateway 2 and Gateway 3 submissions, with integrated Golden Thread documentation management. The system addresses the critical challenge of high rejection rates at Building Safety Regulator (BSR) gateways while streamlining the compliance verification process. | |
| ### **Key Objectives** | |
| 1. **Reduce Gateway Rejection Rates**: Target reduction from current 40-67% to below 20% | |
| 2. **Comprehensive Coverage**: All Building Regulations Parts A-T with specialized agents | |
| 3. **Golden Thread Integration**: Automated generation and maintenance per Building Safety Act requirements | |
| 4. **Professional Standards**: Compliance commentary aligned with RBI/Chartered Building Engineer code of conduct | |
| 5. **Processing Efficiency**: Automated analysis replacing weeks of manual review with hours of AI-assisted checking | |
| ### **System Capabilities** | |
| - Multi-agent architecture with 20+ specialized compliance agents | |
| - Multimodal document processing (text, drawings, specifications) | |
| - Spatial reasoning and measurement from architectural plans | |
| - Regulatory knowledge retrieval from Approved Documents and standards | |
| - Product verification against BBA and KIWA databases | |
| - Gateway 2/3 document organization and validation | |
| - Golden Thread documentation generation and management | |
| --- | |
| ## **1. SYSTEM FUNCTIONS** | |
| ### **1.1 Core Processing Functions** | |
| #### **Gateway 2 Submission Processing** | |
| - **Document Ingestion**: Accept multi-format submissions (PDF, DWG, specifications) | |
| - **Document Classification**: Automatically categorize by document type and Building Regulation relevance | |
| - **Completeness Validation**: Check against Gateway 2 submission requirements per Build UK guidance | |
| - **Compliance Analysis**: Execute multi-agent compliance checking across all applicable Parts A-T | |
| - **Report Generation**: Produce comprehensive compliance commentary with evidence trails | |
| - **Gap Identification**: Flag missing information or non-compliances with specific recommendations | |
| #### **Gateway 3 Support** | |
| - **Completion Certificate Readiness**: Assess documentation completeness for Gateway 3 submission | |
| - **As-Built Validation**: Compare as-built documentation against approved Gateway 2 submission | |
| - **Change Documentation**: Track and validate variations from original approval | |
| - **Final Compliance Verification**: Confirm all conditions satisfied and works completed as approved | |
| #### **Golden Thread Management** | |
| - **Structured Information Creation**: Generate Golden Thread documentation per Building Safety Act requirements | |
| - **Version Control**: Maintain change history throughout project lifecycle | |
| - **Stakeholder Access Management**: Provide appropriate access levels for different project participants | |
| - **BSR Portal Integration**: Format documentation for submission to Building Safety Regulator systems | |
| - **Lifecycle Maintenance**: Support ongoing updates during construction and post-occupation phases | |
| ### **1.2 Analysis Functions** | |
| #### **Spatial Analysis** | |
| - **Measurement Operations**: Calculate lengths, areas, volumes, clearances from submitted drawings | |
| - **Spatial Relationship Modeling**: Understand topological relationships between building elements | |
| - **Path Analysis**: Evaluate circulation routes, means of escape, accessible routes | |
| - **Visibility Analysis**: Assess sightlines, protected views, overlooking considerations | |
| - **Proximity Calculations**: Measure distances between elements for compliance checking | |
| #### **Compliance Verification** | |
| - **Rule-Based Checking**: Validate explicit requirements (dimensions, ratings, specifications) | |
| - **Performance-Based Assessment**: Analyze designs against performance objectives | |
| - **Cross-Part Validation**: Identify conflicts between different Building Regulation requirements | |
| - **Scenario Modeling**: Simulate occupancy, egress, acoustic performance, energy use | |
| - **Product Validation**: Verify specified products against approved standards and certifications | |
| #### **Documentation Functions** | |
| - **Evidence Collection**: Link findings to specific submission documents with page/drawing references | |
| - **Regulatory Citation**: Reference specific Approved Document sections, British Standards, guidance | |
| - **Recommendation Generation**: Provide actionable suggestions for addressing identified issues | |
| - **Confidence Scoring**: Assign confidence levels to automated determinations | |
| - **Escalation Flagging**: Identify cases requiring expert human review | |
| --- | |
| ## **2. DEVELOPMENT APPROACH** | |
| ### **2.1 Development Philosophy** | |
| #### **Comprehensive Coverage** | |
| - **All Parts A-T**: System addresses entire Building Regulations scope, not selective subset | |
| - **No Gaps**: Ensure no regulation part can be overlooked in analysis | |
| - **Interconnected Requirements**: Recognize cross-part dependencies and interactions | |
| - **Future-Proof**: Architecture supports addition of new requirements as regulations evolve | |
| #### **Realistic Assessment Over Optimism** | |
| - **Critical Evaluation**: Identify genuine challenges and limitations upfront | |
| - **Feasibility Focus**: Prioritize proven technologies and achievable goals | |
| - **Transparent Limitations**: Clearly communicate what system can and cannot determine | |
| - **Conservative Bias**: Err toward stricter interpretation when regulatory ambiguity exists | |
| #### **Detailed Specification Before Implementation** | |
| - **Thorough Planning**: Complete architectural design before coding begins | |
| - **Clear Interfaces**: Define data structures and APIs between components | |
| - **Test Cases**: Develop validation scenarios based on real-world submissions | |
| - **Iterative Refinement**: Plan for learning and improvement from initial deployments | |
| ### **2.2 Processing Methodology** | |
| #### **Pipeline Architecture** | |
| ``` | |
| Submission → Ingestion → Classification → Spatial Extraction → | |
| Storage → Retrieval → Multi-Agent Analysis → Validation → | |
| Report Generation → Golden Thread Update → Output | |
| ``` | |
| #### **Sequential Processing Stages** | |
| **Stage 1: Document Ingestion (5-10 minutes)** | |
| - Parse PDF submissions | |
| - Extract text content and metadata | |
| - Segment architectural drawings from specifications | |
| - Initial quality assessment | |
| - Generate document inventory | |
| **Stage 2: Spatial Processing (15-30 minutes)** | |
| - Run FloorplanTransformation on architectural drawings | |
| - Extract vector representations of building elements | |
| - Classify rooms, walls, doors, stairs, structural elements | |
| - Calculate dimensions and spatial relationships | |
| - Build graph database of building topology | |
| **Stage 3: Knowledge Retrieval (Concurrent)** | |
| - Query RAG system for relevant regulatory requirements | |
| - Retrieve applicable Approved Document sections | |
| - Identify similar precedent cases | |
| - Pull product certification databases | |
| - Assemble context for each specialist agent | |
| **Stage 4: Multi-Agent Analysis (20-40 minutes)** | |
| - Dispatch submission to relevant Part-specific agents | |
| - Execute parallel compliance checking across all Parts | |
| - Perform cross-part validation | |
| - Aggregate findings and resolve conflicts | |
| - Generate evidence-based determinations | |
| **Stage 5: Report Generation (5-10 minutes)** | |
| - Synthesize multi-agent findings | |
| - Structure compliance commentary per Gateway requirements | |
| - Generate Golden Thread documentation | |
| - Package outputs for delivery | |
| - Create audit trail | |
| **Total Processing Time: 45-90 minutes per submission** | |
| ### **2.3 Deployment Considerations** | |
| #### **Platform: Isambard AI Supercomputer** | |
| - **Hardware**: Nvidia GH200 Grace-Hopper superchips | |
| - **Operating System**: Linux-based (Ubuntu or similar) | |
| - **Advantages**: Massive parallel processing capability, high-bandwidth memory, ideal for large language models | |
| - **Challenges**: Windows software compatibility (FloorplanTransformation) | |
| #### **Windows Dependency Resolution** | |
| **Problem**: FloorplanTransformation tool is Windows-only | |
| **Solution Options**: | |
| 1. **Wine/Proton Compatibility Layer**: Run Windows executable on Linux via Wine emulator | |
| 2. **Separate Windows Preprocessing Server**: Dedicated Windows VM for FloorplanTransformation, feed results to Linux main system | |
| 3. **Alternative Tool Development**: Develop Linux-native replacement using computer vision models | |
| 4. **Hybrid Approach**: Use FloorplanTransformation for training data generation, deploy alternative for production | |
| #### **Geographic Calibration** | |
| - Ensure cost databases reflect UK construction pricing | |
| - Validate dimensional standards against UK practice (metric, UK-specific standards) | |
| - Align terminology with UK Building Regulations language | |
| - Reference UK-specific product certification bodies (BBA, KIWA, BSI) | |
| --- | |
| ## **3. SYSTEM ARCHITECTURE** | |
| ### **3.1 High-Level Architecture** | |
| ``` | |
| ┌─────────────────────────────────────────────────────────────────┐ | |
| │ SUBMISSION INTERFACE │ | |
| │ - Web portal / API for document upload │ | |
| │ - Authentication and project management │ | |
| │ - Progress tracking and notifications │ | |
| └─────────────────────────┬───────────────────────────────────────┘ | |
| │ | |
| ↓ | |
| ┌─────────────────────────────────────────────────────────────────┐ | |
| │ DOCUMENT INGESTION & PROCESSING LAYER │ | |
| │ │ | |
| │ ┌──────────────┐ ┌──────────────┐ ┌────────────────┐ │ | |
| │ │ PDF Parser │ │ OCR Engine │ │ Document │ │ | |
| │ │ (PyMuPDF) │ │ (if needed) │ │ Classifier │ │ | |
| │ └──────────────┘ └──────────────┘ └────────────────┘ │ | |
| │ ↓ ↓ ↓ │ | |
| │ ┌──────────────────────────────────────────────────────┐ │ | |
| │ │ FloorplanTransformation Engine │ │ | |
| │ │ (Windows compatibility via Wine/separate server) │ │ | |
| │ │ → Extracts spatial vector data from drawings │ │ | |
| │ └──────────────────────────────────────────────────────┘ │ | |
| │ ↓ │ | |
| │ ┌──────────────────────────────────────────────────────┐ │ | |
| │ │ Spatial Analysis & Graph Building │ │ | |
| │ │ → Identifies rooms, walls, doors, stairs │ │ | |
| │ │ → Calculates measurements and relationships │ │ | |
| │ │ → Constructs building topology graph │ │ | |
| │ └──────────────────────────────────────────────────────┘ │ | |
| │ │ | |
| └─────────────────────────┬───────────────────────────────────────┘ | |
| │ | |
| ↓ | |
| ┌─────────────────────────────────────────────────────────────────┐ | |
| │ STORAGE LAYER │ | |
| │ │ | |
| │ ┌────────────────┐ ┌────────────────┐ ┌──────────────┐ │ | |
| │ │ Vector Store │ │ Graph Database │ │ Relational DB│ │ | |
| │ │ (Milvus) │ │ (Neo4j) │ │ (PostgreSQL) │ │ | |
| │ │ │ │ │ │ │ │ | |
| │ │ • Document │ │ • Spatial │ │ • Metadata │ │ | |
| │ │ embeddings │ │ topology │ │ • Projects │ │ | |
| │ │ • Regulatory │ │ • Compliance │ │ • Users │ │ | |
| │ │ text │ │ chains │ │ • Products │ │ | |
| │ │ • Precedents │ │ • Golden Thread│ │ • History │ │ | |
| │ └────────────────┘ └────────────────┘ └──────────────┘ │ | |
| │ │ | |
| └─────────────────────────┬───────────────────────────────────────┘ | |
| │ | |
| ↓ | |
| ┌─────────────────────────────────────────────────────────────────┐ | |
| │ RAG SYSTEM (Multimodal) │ | |
| │ │ | |
| │ ┌────────────────────────────────────────────────────────┐ │ | |
| │ │ Text Retrieval System │ │ | |
| │ │ • Approved Documents A-T (chunked & embedded) │ │ | |
| │ │ • British Standards (BS, BS EN) │ │ | |
| │ │ • Guidance documents (BSR, NHBC, etc.) │ │ | |
| │ │ • Historical precedents and interpretations │ │ | |
| │ └────────────────────────────────────────────────────────┘ │ | |
| │ │ | |
| │ ┌────────────────────────────────────────────────────────┐ │ | |
| │ │ Visual Retrieval System │ │ | |
| │ │ • Approved Document diagrams (extracted & indexed) │ │ | |
| │ │ • Reference construction details │ │ | |
| │ │ • Visual similarity search for diagram comparison │ │ | |
| │ └────────────────────────────────────────────────────────┘ │ | |
| │ │ | |
| │ ┌────────────────────────────────────────────────────────┐ │ | |
| │ │ Product Database │ │ | |
| │ │ • BBA approved products │ │ | |
| │ │ • KIWA certifications │ │ | |
| │ │ • CE marking database │ │ | |
| │ │ • Manufacturer specifications │ │ | |
| │ └────────────────────────────────────────────────────────┘ │ | |
| │ │ | |
| └─────────────────────────┬───────────────────────────────────────┘ | |
| │ | |
| ↓ | |
| ┌─────────────────────────────────────────────────────────────────┐ | |
| │ MULTI-AGENT COMPLIANCE FRAMEWORK │ | |
| │ │ | |
| │ ┌────────────────────────────────────────────────────────┐ │ | |
| │ │ Master Orchestrator Agent │ │ | |
| │ │ • Submission analysis & routing │ │ | |
| │ │ • Agent coordination │ │ | |
| │ │ • Conflict resolution │ │ | |
| │ │ • Report synthesis │ │ | |
| │ └────────────────────────────────────────────────────────┘ │ | |
| │ ↓ │ | |
| │ ┌────────────────┼────────────────┐ │ | |
| │ ↓ ↓ ↓ │ | |
| │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ | |
| │ │ Part A │ │ Part B │ │ Part C │...│ Part T │ │ | |
| │ │ Agent │ │ Agent │ │ Agent │ │ Agent │ │ | |
| │ │ │ │ │ │ │ │ │ │ | |
| │ │ Structure│ │ Fire │ │ Site/ │ │ Broadband│ │ | |
| │ │ │ │ Safety │ │ Moisture │ │ │ │ | |
| │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ | |
| │ │ | |
| │ Each agent has: │ | |
| │ • Specialized knowledge base (RAG-retrieved) │ | |
| │ • Domain-specific tools │ | |
| │ • Compliance checking logic │ | |
| │ • Standardized output format │ | |
| │ │ | |
| └─────────────────────────┬───────────────────────────────────────┘ | |
| │ | |
| ↓ | |
| ┌─────────────────────────────────────────────────────────────────┐ | |
| │ TOOL LAYER │ | |
| │ │ | |
| │ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────┐ │ | |
| │ │ Measurement │ │ Spatial Reasoning│ │ Scenario │ │ | |
| │ │ Tools │ │ Tools │ │ Planning │ │ | |
| │ │ │ │ │ │ Tools │ │ | |
| │ │ • Distances │ │ • Path finding │ │ • Egress │ │ | |
| │ │ • Areas │ │ • Adjacency │ │ • Acoustic │ │ | |
| │ │ • Volumes │ │ • Containment │ │ • Energy │ │ | |
| │ │ • Clearances │ │ • Visibility │ │ • Occupancy │ │ | |
| │ └──────────────────┘ └──────────────────┘ └──────────────┘ │ | |
| │ │ | |
| │ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────┐ │ | |
| │ │ Comparison │ │ Product Checker │ │ Calculation │ │ | |
| │ │ Tools │ │ Tools │ │ Tools │ │ | |
| │ │ │ │ │ │ │ │ | |
| │ │ • Design vs │ │ • BBA database │ │ • U-values │ │ | |
| │ │ diagrams │ │ • KIWA database │ │ • SAP/SBEM │ │ | |
| │ │ • As-built vs │ │ • CE marking │ │ • Loads │ │ | |
| │ │ approved │ │ • Standards │ │ • Acoustics │ │ | |
| │ └──────────────────┘ └──────────────────┘ └──────────────┘ │ | |
| │ │ | |
| └─────────────────────────┬───────────────────────────────────────┘ | |
| │ | |
| ↓ | |
| ┌─────────────────────────────────────────────────────────────────┐ | |
| │ OUTPUT GENERATION LAYER │ | |
| │ │ | |
| │ ┌────────────────────────────────────────────────────────┐ │ | |
| │ │ Compliance Report Generator │ │ | |
| │ │ • Gateway 2/3 format compliance commentary │ │ | |
| │ │ • Evidence trails and regulatory citations │ │ | |
| │ │ • Confidence scores and escalation flags │ │ | |
| │ │ • Actionable recommendations │ │ | |
| │ └────────────────────────────────────────────────────────┘ │ | |
| │ │ | |
| │ ┌────────────────────────────────────────────────────────┐ │ | |
| │ │ Golden Thread Documentation Generator │ │ | |
| │ │ • Structured information per Building Safety Act │ │ | |
| │ │ • Version control and change tracking │ │ | |
| │ │ • Stakeholder-appropriate views │ │ | |
| │ │ • BSR portal formatted outputs │ │ | |
| │ └────────────────────────────────────────────────────────┘ │ | |
| │ │ | |
| │ ┌────────────────────────────────────────────────────────┐ │ | |
| │ │ Visualization & Evidence System │ │ | |
| │ │ • Issue highlighting on drawings │ │ | |
| │ │ • Interactive compliance dashboard │ │ | |
| │ │ • Evidence package assembly │ │ | |
| │ └────────────────────────────────────────────────────────┘ │ | |
| │ │ | |
| └─────────────────────────────────────────────────────────────────┘ | |
| ``` | |
| ### **3.2 Component Details** | |
| #### **3.2.1 Document Ingestion Layer** | |
| **Purpose**: Accept and process diverse submission documents into standardized formats | |
| **Components**: | |
| - **PDF Parser (PyMuPDF)**: Extract text, images, metadata from PDF submissions | |
| - **OCR Engine (Tesseract/PaddleOCR)**: Handle scanned or low-quality documents | |
| - **Document Classifier**: Categorize documents by type (architectural plans, specifications, calculations, reports) | |
| - **Page Segmentation**: Separate multi-page documents into logical sections | |
| - **Quality Assessment**: Flag low-quality or incomplete documents early | |
| **Outputs**: | |
| - Structured text content | |
| - Extracted images (drawings, photos, diagrams) | |
| - Document metadata (title, author, date, version) | |
| - Document classification labels | |
| - Quality scores | |
| #### **3.2.2 Spatial Processing Layer** | |
| **Purpose**: Convert raster architectural drawings into structured spatial representations | |
| **FloorplanTransformation Engine**: | |
| - **Input**: PDF/image format architectural plans | |
| - **Process**: Computer vision-based element detection and vectorization | |
| - **Output**: Vector representations of building elements with classifications | |
| **Linux Compatibility Solutions**: | |
| 1. **Wine Emulator**: Run Windows .exe via compatibility layer | |
| 2. **Dedicated Windows Server**: Separate preprocessing service | |
| 3. **API Wrapper**: Encapsulate Windows tool behind REST API | |
| 4. **Alternative CV Models**: Develop replacement using open-source tools | |
| **Spatial Analysis Pipeline**: | |
| ```python | |
| # Pseudo-code example | |
| def process_floorplan(pdf_path: str) -> SpatialData: | |
| # Extract plan images from PDF | |
| images = extract_plan_images(pdf_path) | |
| # Run FloorplanTransformation (via Wine or API) | |
| vector_data = floorplan_transformation(images) | |
| # Classify elements | |
| rooms = classify_rooms(vector_data) | |
| walls = classify_walls(vector_data) | |
| doors = classify_doors(vector_data) | |
| stairs = classify_stairs(vector_data) | |
| # Calculate measurements | |
| measurements = calculate_dimensions(vector_data) | |
| # Build topology | |
| topology = build_spatial_graph(rooms, walls, doors, stairs) | |
| return SpatialData( | |
| rooms=rooms, | |
| walls=walls, | |
| doors=doors, | |
| stairs=stairs, | |
| measurements=measurements, | |
| topology=topology | |
| ) | |
| ``` | |
| **Graph Database Population**: | |
| - Create nodes for each spatial element (rooms, doors, stairs, etc.) | |
| - Define relationships (connects, contains, adjacent_to, above, below) | |
| - Store geometric properties (dimensions, coordinates, areas) | |
| - Enable graph queries for compliance checking | |
| #### **3.2.3 Storage Layer** | |
| **Milvus Vector Database**: | |
| - **Purpose**: Store and retrieve document embeddings for semantic search | |
| - **Contents**: | |
| - Regulatory text embeddings (Approved Documents, Standards) | |
| - Submission document embeddings | |
| - Historical precedent embeddings | |
| - Visual embeddings (diagram similarity) | |
| - **Indexing**: HNSW (Hierarchical Navigable Small World) for fast approximate nearest neighbor search | |
| - **Partitioning**: Separate collections per Building Regulation Part for efficient retrieval | |
| **Neo4j Graph Database**: | |
| - **Purpose**: Model spatial relationships and compliance chains | |
| - **Node Types**: | |
| - Building elements (rooms, walls, doors, stairs) | |
| - Compliance requirements (regulations, sub-clauses) | |
| - Evidence items (document references, measurements) | |
| - Golden Thread information items | |
| - **Relationship Types**: | |
| - Spatial: connects, contains, adjacent_to, above, below | |
| - Compliance: requires, satisfies, conflicts_with | |
| - Evidence: references, supports, contradicts | |
| - Provenance: derived_from, modified_from, supersedes | |
| - **Query Patterns**: | |
| - Path finding (shortest egress route) | |
| - Subgraph extraction (all elements in fire compartment) | |
| - Dependency tracking (requirements affected by design change) | |
| **PostgreSQL Relational Database**: | |
| - **Purpose**: Store structured metadata and operational data | |
| - **Schema**: | |
| - Projects (ID, name, address, client, status) | |
| - Submissions (ID, project, gateway, date, status) | |
| - Users (authentication, roles, permissions) | |
| - Products (manufacturer, model, certifications) | |
| - Audit logs (actions, timestamps, users) | |
| - Processing history (submission, agent, runtime, outcome) | |
| #### **3.2.4 RAG System (Retrieval-Augmented Generation)** | |
| **Text Retrieval System**: | |
| **Indexing Strategy**: | |
| ``` | |
| Approved Documents → Chunked by regulation section → Embedded → Indexed in Milvus | |
| ↓ | |
| Each chunk includes: | |
| - Text content | |
| - Regulation reference (e.g., "Part B1, Section 2.5") | |
| - Document source | |
| - Effective date | |
| - Related diagrams/tables | |
| - Keywords | |
| ``` | |
| **Embedding Model**: | |
| - NV-Embed-v2 or similar high-performance model | |
| - 768-dimensional dense vectors | |
| - Trained on technical/regulatory text for domain relevance | |
| **Retrieval Process**: | |
| 1. **Query Construction**: Agent formulates query based on compliance check needed | |
| 2. **Embedding**: Query embedded using same model as corpus | |
| 3. **Vector Search**: Retrieve top-K most similar chunks from Milvus | |
| 4. **Reranking** (optional): Use cross-encoder model to refine results | |
| 5. **Context Assembly**: Combine retrieved chunks with query for LLM | |
| **Visual Retrieval System**: | |
| **Diagram Corpus**: | |
| - Extract all diagrams from Approved Documents | |
| - Annotate with metadata (Part, section, diagram number, caption) | |
| - Generate visual embeddings (CLIP or similar vision-language model) | |
| - Index in separate Milvus collection | |
| **Visual Similarity Search**: | |
| - Compare submitted designs against reference diagrams | |
| - Identify similar configurations or details | |
| - Flag deviations from approved construction methods | |
| **Product Database Integration**: | |
| - **BBA Certificates**: Regularly updated database of approved products | |
| - **KIWA Certifications**: European product approvals relevant to UK | |
| - **CE Marking Database**: Products complying with European standards | |
| - **Search Functionality**: Query by product name, manufacturer, application, performance criteria | |
| #### **3.2.5 Multi-Agent Framework** | |
| **Master Orchestrator Agent**: | |
| **Responsibilities**: | |
| 1. **Submission Analysis**: Determine which Building Regulation Parts apply | |
| 2. **Agent Dispatching**: Route submission to relevant specialist agents | |
| 3. **Parallel Coordination**: Manage concurrent agent execution | |
| 4. **Conflict Resolution**: Identify and resolve inter-agent conflicts | |
| 5. **Report Synthesis**: Aggregate findings into unified output | |
| **Process Flow**: | |
| ```python | |
| def process_submission(submission: Submission) -> ComplianceReport: | |
| # 1. Analyze submission | |
| relevant_parts = identify_applicable_parts(submission) | |
| # 2. Dispatch to agents (parallel execution) | |
| agent_results = {} | |
| with ThreadPoolExecutor() as executor: | |
| futures = { | |
| executor.submit(agent.check_compliance, submission): part | |
| for part, agent in specialist_agents.items() | |
| if part in relevant_parts | |
| } | |
| for future in as_completed(futures): | |
| part = futures[future] | |
| agent_results[part] = future.result() | |
| # 3. Validate cross-part consistency | |
| conflicts = detect_conflicts(agent_results) | |
| # 4. Synthesize report | |
| report = synthesize_report( | |
| agent_results=agent_results, | |
| conflicts=conflicts, | |
| submission=submission | |
| ) | |
| return report | |
| ``` | |
| **Specialist Agents (Parts A-T)**: | |
| **Agent Architecture** (consistent across all Parts): | |
| ```python | |
| class BuildingRegulationAgent: | |
| """ | |
| Base class for all Part-specific agents. | |
| """ | |
| def __init__( | |
| self, | |
| part_id: str, | |
| rag_system: RAGSystem, | |
| spatial_db: Neo4jClient, | |
| tools: ToolRegistry | |
| ): | |
| self.part_id = part_id | |
| self.rag = rag_system | |
| self.spatial_db = spatial_db | |
| self.tools = tools | |
| # Load Part-specific knowledge | |
| self.knowledge_base = self._load_knowledge() | |
| def check_compliance( | |
| self, | |
| submission: Submission | |
| ) -> List[ComplianceCheck]: | |
| """ | |
| Main entry point for compliance checking. | |
| Returns list of ComplianceCheck objects with: | |
| - Finding (pass/fail/unclear) | |
| - Evidence (specific references) | |
| - Reasoning (explanation) | |
| - Confidence score | |
| - Recommendations (if non-compliant) | |
| """ | |
| checks = [] | |
| # 1. Extract relevant information | |
| relevant_data = self._extract_relevant_data(submission) | |
| # 2. Retrieve regulatory requirements | |
| requirements = self._get_requirements(relevant_data) | |
| # 3. Execute compliance checks | |
| for requirement in requirements: | |
| check = self._evaluate_compliance( | |
| requirement=requirement, | |
| data=relevant_data | |
| ) | |
| checks.append(check) | |
| # 4. Cross-validate findings | |
| checks = self._cross_validate(checks) | |
| return checks | |
| def _load_knowledge(self) -> Dict: | |
| """ | |
| Load Part-specific knowledge from RAG system. | |
| """ | |
| # Retrieve relevant Approved Document sections | |
| # Retrieve applicable standards | |
| # Retrieve historical precedents | |
| pass | |
| def _get_requirements(self, data: Dict) -> List[Requirement]: | |
| """ | |
| Identify which specific requirements apply based on data. | |
| """ | |
| # Query RAG system for applicable requirements | |
| # Filter by building type, use, characteristics | |
| pass | |
| def _evaluate_compliance( | |
| self, | |
| requirement: Requirement, | |
| data: Dict | |
| ) -> ComplianceCheck: | |
| """ | |
| Evaluate specific requirement against submission data. | |
| """ | |
| # Use tools to measure/calculate as needed | |
| # Compare against requirement criteria | |
| # Generate evidence trail | |
| # Assign confidence score | |
| pass | |
| ``` | |
| **Agent Specializations**: | |
| Each Part agent inherits from base class and adds specialized capabilities: | |
| **Part A (Structure) Agent**: | |
| - Load calculation tools | |
| - Structural material verification | |
| - Foundation adequacy assessment | |
| - Disproportionate collapse checking (for HRBs) | |
| **Part B (Fire Safety) Agent**: | |
| - Egress path modeling | |
| - Travel distance calculation | |
| - Compartmentation validation | |
| - Fire resistance verification | |
| - Occupancy load calculation | |
| **Part C (Site Preparation) Agent**: | |
| - Contamination assessment | |
| - Radon risk evaluation | |
| - Ground gas analysis | |
| - Moisture protection checking | |
| **[Parts D-T]**: Each with domain-specific tools and knowledge | |
| **Agent Communication**: | |
| - Shared access to spatial graph database | |
| - Message passing for cross-dependencies | |
| - Conflict notification to orchestrator | |
| - Evidence sharing between related agents | |
| #### **3.2.6 Tool Layer** | |
| **Measurement Tools**: | |
| ```python | |
| class MeasurementTools: | |
| """ | |
| Spatial measurement capabilities. | |
| """ | |
| def measure_distance( | |
| self, | |
| point_a: Coordinate, | |
| point_b: Coordinate, | |
| path_type: str = 'straight' # or 'along_route' | |
| ) -> float: | |
| """ | |
| Calculate distance between points. | |
| """ | |
| if path_type == 'straight': | |
| return euclidean_distance(point_a, point_b) | |
| elif path_type == 'along_route': | |
| # Query graph DB for path | |
| path = self.spatial_db.shortest_path(point_a, point_b) | |
| return sum_path_length(path) | |
| def calculate_area( | |
| self, | |
| room_id: str | |
| ) -> float: | |
| """ | |
| Calculate room floor area from polygon vertices. | |
| """ | |
| vertices = self.spatial_db.get_room_vertices(room_id) | |
| return polygon_area(vertices) | |
| def measure_clearance( | |
| self, | |
| element_id: str, | |
| direction: str | |
| ) -> float: | |
| """ | |
| Measure clearance in specified direction. | |
| """ | |
| element_bounds = self.spatial_db.get_element_bounds(element_id) | |
| obstacles = self.spatial_db.find_obstacles( | |
| element_bounds, | |
| direction | |
| ) | |
| return calculate_clearance(element_bounds, obstacles, direction) | |
| ``` | |
| **Spatial Reasoning Tools**: | |
| ```python | |
| class SpatialReasoningTools: | |
| """ | |
| Higher-level spatial analysis. | |
| """ | |
| def find_egress_routes( | |
| self, | |
| start_room: str, | |
| building_id: str | |
| ) -> List[EgressRoute]: | |
| """ | |
| Find all egress routes from room to building exit. | |
| """ | |
| query = """ | |
| MATCH path = (start:Room {id: $start_room})-[:CONNECTS*]->(exit:Exit) | |
| WHERE exit.building_id = $building_id | |
| RETURN path | |
| ORDER BY length(path) | |
| """ | |
| paths = self.spatial_db.query(query, start_room=start_room, building_id=building_id) | |
| routes = [] | |
| for path in paths: | |
| route = EgressRoute( | |
| path=path, | |
| distance=calculate_path_distance(path), | |
| doors=extract_doors_from_path(path), | |
| stairs=extract_stairs_from_path(path) | |
| ) | |
| routes.append(route) | |
| return routes | |
| def check_accessibility( | |
| self, | |
| route: Route, | |
| requirements: AccessibilityRequirements | |
| ) -> AccessibilityAssessment: | |
| """ | |
| Evaluate if route meets accessibility requirements. | |
| """ | |
| issues = [] | |
| # Check widths | |
| for segment in route.segments: | |
| if segment.width < requirements.min_width: | |
| issues.append(f"Width {segment.width}mm < {requirements.min_width}mm at {segment.location}") | |
| # Check gradients | |
| for ramp in route.ramps: | |
| if ramp.gradient > requirements.max_gradient: | |
| issues.append(f"Gradient {ramp.gradient} > {requirements.max_gradient} at {ramp.location}") | |
| # Check door widths | |
| for door in route.doors: | |
| if door.clear_width < requirements.min_door_width: | |
| issues.append(f"Door width {door.clear_width}mm < {requirements.min_door_width}mm at {door.location}") | |
| return AccessibilityAssessment( | |
| compliant=(len(issues) == 0), | |
| issues=issues | |
| ) | |
| ``` | |
| **Scenario Planning Tools**: | |
| ```python | |
| class ScenarioPlanningTools: | |
| """ | |
| Simulation and modeling capabilities. | |
| """ | |
| def simulate_egress( | |
| self, | |
| building: Building, | |
| occupancy: int, | |
| fire_location: str | |
| ) -> EgressSimulationResult: | |
| """ | |
| Simulate building evacuation scenario. | |
| """ | |
| # Initialize population | |
| occupants = distribute_occupants(building, occupancy) | |
| # Define fire spread | |
| fire_model = create_fire_model(building, fire_location) | |
| # Run evacuation simulation | |
| time_steps = [] | |
| current_time = 0 | |
| while not all_evacuated(occupants): | |
| # Move occupants toward exits | |
| occupants = move_occupants(occupants, building, fire_model) | |
| # Update fire spread | |
| fire_model = update_fire_model(fire_model, current_time) | |
| # Check for casualties | |
| casualties = check_casualties(occupants, fire_model) | |
| # Record state | |
| time_steps.append(SimulationState( | |
| time=current_time, | |
| occupants=occupants, | |
| fire=fire_model, | |
| casualties=casualties | |
| )) | |
| current_time += SIMULATION_TIMESTEP | |
| if current_time > MAX_SIMULATION_TIME: | |
| break | |
| return EgressSimulationResult( | |
| evacuation_time=current_time, | |
| casualties=sum(ts.casualties for ts in time_steps), | |
| bottlenecks=identify_bottlenecks(time_steps), | |
| time_steps=time_steps | |
| ) | |
| def model_acoustic_performance( | |
| self, | |
| wall: Wall, | |
| source_room: Room, | |
| receiver_room: Room | |
| ) -> AcousticPerformance: | |
| """ | |
| Predict sound insulation performance. | |
| """ | |
| # Calculate direct transmission | |
| direct_loss = calculate_transmission_loss(wall.construction) | |
| # Calculate flanking paths | |
| flanking_loss = calculate_flanking_transmission( | |
| wall=wall, | |
| source_room=source_room, | |
| receiver_room=receiver_room | |
| ) | |
| # Combine paths | |
| total_loss = combine_transmission_paths(direct_loss, flanking_loss) | |
| return AcousticPerformance( | |
| sound_reduction_index=total_loss, | |
| meets_requirement=total_loss >= REQUIREMENT_DnTw | |
| ) | |
| ``` | |
| **Comparison Tools**: | |
| ```python | |
| class ComparisonTools: | |
| """ | |
| Compare designs against reference standards. | |
| """ | |
| def compare_against_approved_diagram( | |
| self, | |
| submitted_design: SpatialData, | |
| reference_diagram_id: str | |
| ) -> ComparisonResult: | |
| """ | |
| Compare submitted design against Approved Document diagram. | |
| """ | |
| # Retrieve reference diagram | |
| reference = self.rag.retrieve_diagram(reference_diagram_id) | |
| # Extract comparable features | |
| submitted_features = extract_features(submitted_design) | |
| reference_features = extract_features(reference) | |
| # Compare | |
| differences = [] | |
| for feature_name, submitted_value in submitted_features.items(): | |
| reference_value = reference_features.get(feature_name) | |
| if reference_value is None: | |
| differences.append(f"{feature_name}: Not in reference diagram") | |
| elif not values_match(submitted_value, reference_value, tolerance=0.05): | |
| differences.append( | |
| f"{feature_name}: Submitted {submitted_value} vs Reference {reference_value}" | |
| ) | |
| return ComparisonResult( | |
| matches=(len(differences) == 0), | |
| differences=differences, | |
| similarity_score=calculate_similarity(submitted_features, reference_features) | |
| ) | |
| def compare_as_built_to_approved( | |
| self, | |
| as_built: Submission, | |
| approved: Submission | |
| ) -> ChangeReport: | |
| """ | |
| Identify changes between approved and as-built submissions. | |
| """ | |
| changes = [] | |
| # Compare spatial data | |
| spatial_changes = self._compare_spatial_data( | |
| as_built.spatial_data, | |
| approved.spatial_data | |
| ) | |
| changes.extend(spatial_changes) | |
| # Compare specifications | |
| spec_changes = self._compare_specifications( | |
| as_built.specifications, | |
| approved.specifications | |
| ) | |
| changes.extend(spec_changes) | |
| # Categorize changes by significance | |
| critical_changes = [c for c in changes if c.affects_compliance] | |
| minor_changes = [c for c in changes if not c.affects_compliance] | |
| return ChangeReport( | |
| critical_changes=critical_changes, | |
| minor_changes=minor_changes, | |
| requires_revised_approval=(len(critical_changes) > 0) | |
| ) | |
| ``` | |
| **Product Verification Tools**: | |
| ```python | |
| class ProductVerificationTools: | |
| """ | |
| Check specified products against certification databases. | |
| """ | |
| def verify_bba_approval( | |
| self, | |
| product_name: str, | |
| manufacturer: str, | |
| application: str | |
| ) -> ApprovalStatus: | |
| """ | |
| Check if product has current BBA approval. | |
| """ | |
| # Query BBA database | |
| results = self.bba_db.search( | |
| product=product_name, | |
| manufacturer=manufacturer, | |
| application=application | |
| ) | |
| if not results: | |
| return ApprovalStatus( | |
| approved=False, | |
| message="Product not found in BBA database" | |
| ) | |
| # Check approval is current | |
| most_recent = max(results, key=lambda r: r.certificate_date) | |
| if most_recent.expired: | |
| return ApprovalStatus( | |
| approved=False, | |
| message=f"BBA approval expired on {most_recent.expiry_date}", | |
| certificate_number=most_recent.certificate_number | |
| ) | |
| return ApprovalStatus( | |
| approved=True, | |
| certificate_number=most_recent.certificate_number, | |
| certificate_date=most_recent.certificate_date, | |
| expiry_date=most_recent.expiry_date | |
| ) | |
| def check_ce_marking( | |
| self, | |
| product: Product, | |
| standard: str | |
| ) -> CEMarkingStatus: | |
| """ | |
| Verify product has CE marking to specified standard. | |
| """ | |
| # Query CE marking database | |
| marking = self.ce_db.get_marking( | |
| product_code=product.code, | |
| manufacturer_id=product.manufacturer_id | |
| ) | |
| if not marking: | |
| return CEMarkingStatus( | |
| marked=False, | |
| message="No CE marking found for product" | |
| ) | |
| if standard not in marking.standards: | |
| return CEMarkingStatus( | |
| marked=True, | |
| compliant=False, | |
| message=f"CE marked but not to required standard {standard}", | |
| standards=marking.standards | |
| ) | |
| return CEMarkingStatus( | |
| marked=True, | |
| compliant=True, | |
| standards=marking.standards, | |
| notified_body=marking.notified_body | |
| ) | |
| ``` | |
| #### **3.2.7 Output Generation Layer** | |
| **Compliance Report Generator**: | |
| ```python | |
| class ComplianceReportGenerator: | |
| """ | |
| Generate structured compliance reports for Gateway submissions. | |
| """ | |
| def generate_gateway2_report( | |
| self, | |
| submission: Submission, | |
| agent_results: Dict[str, List[ComplianceCheck]], | |
| conflicts: List[Conflict] | |
| ) -> GatewayReport: | |
| """ | |
| Create comprehensive Gateway 2 compliance report. | |
| """ | |
| report = GatewayReport( | |
| submission_id=submission.id, | |
| date=datetime.now(), | |
| building=submission.building_details | |
| ) | |
| # Executive summary | |
| report.executive_summary = self._generate_executive_summary( | |
| agent_results, | |
| conflicts | |
| ) | |
| # Part-by-part findings | |
| for part, checks in agent_results.items(): | |
| part_section = self._generate_part_section( | |
| part=part, | |
| checks=checks | |
| ) | |
| report.add_section(part_section) | |
| # Cross-part issues | |
| if conflicts: | |
| conflict_section = self._generate_conflict_section(conflicts) | |
| report.add_section(conflict_section) | |
| # Recommendations | |
| report.recommendations = self._prioritize_recommendations( | |
| agent_results, | |
| conflicts | |
| ) | |
| # Evidence package | |
| report.evidence = self._compile_evidence(agent_results) | |
| # Confidence assessment | |
| report.overall_confidence = self._calculate_overall_confidence( | |
| agent_results | |
| ) | |
| return report | |
| def _generate_executive_summary( | |
| self, | |
| agent_results: Dict, | |
| conflicts: List | |
| ) -> str: | |
| """ | |
| Create executive summary of key findings. | |
| """ | |
| total_checks = sum(len(checks) for checks in agent_results.values()) | |
| compliant = sum( | |
| 1 for checks in agent_results.values() | |
| for check in checks | |
| if check.status == "compliant" | |
| ) | |
| non_compliant = sum( | |
| 1 for checks in agent_results.values() | |
| for check in checks | |
| if check.status == "non_compliant" | |
| ) | |
| requires_clarification = total_checks - compliant - non_compliant | |
| summary = f""" | |
| EXECUTIVE SUMMARY | |
| Total compliance checks performed: {total_checks} | |
| - Compliant: {compliant} ({compliant/total_checks*100:.1f}%) | |
| - Non-compliant: {non_compliant} ({non_compliant/total_checks*100:.1f}%) | |
| - Requires clarification: {requires_clarification} ({requires_clarification/total_checks*100:.1f}%) | |
| Cross-part conflicts identified: {len(conflicts)} | |
| CRITICAL ISSUES REQUIRING IMMEDIATE ATTENTION: | |
| """ | |
| # Add critical issues | |
| critical_issues = [ | |
| check for checks in agent_results.values() | |
| for check in checks | |
| if check.status == "non_compliant" and check.priority == "critical" | |
| ] | |
| for i, issue in enumerate(critical_issues, 1): | |
| summary += f"\n{i}. {issue.regulation}: {issue.requirement}" | |
| return summary | |
| def _generate_part_section( | |
| self, | |
| part: str, | |
| checks: List[ComplianceCheck] | |
| ) -> ReportSection: | |
| """ | |
| Generate detailed section for specific Building Regulation Part. | |
| """ | |
| section = ReportSection( | |
| title=f"Part {part} - {PART_NAMES[part]}", | |
| checks=checks | |
| ) | |
| # Group checks by sub-regulation | |
| grouped = self._group_checks_by_regulation(checks) | |
| for regulation, regulation_checks in grouped.items(): | |
| subsection = f"\n\n### {regulation}\n\n" | |
| for check in regulation_checks: | |
| subsection += self._format_compliance_check(check) | |
| section.add_content(subsection) | |
| return section | |
| def _format_compliance_check( | |
| self, | |
| check: ComplianceCheck | |
| ) -> str: | |
| """ | |
| Format individual compliance check for report. | |
| """ | |
| # Status icon | |
| if check.status == "compliant": | |
| status_icon = "✓" | |
| elif check.status == "non_compliant": | |
| status_icon = "✗" | |
| else: | |
| status_icon = "?" | |
| # Build formatted text | |
| text = f""" | |
| {status_icon} **{check.requirement}** | |
| Status: {check.status.upper()} | |
| Confidence: {check.confidence:.0%} | |
| Priority: {check.priority.upper()} | |
| **Evidence:** | |
| {self._format_evidence_list(check.evidence)} | |
| **Reasoning:** | |
| {check.reasoning} | |
| """ | |
| if check.recommendations: | |
| text += f""" | |
| **Recommendations:** | |
| {self._format_recommendations(check.recommendations)} | |
| """ | |
| if check.regulatory_references: | |
| text += f""" | |
| **Regulatory References:** | |
| {self._format_references(check.regulatory_references)} | |
| """ | |
| return text | |
| ``` | |
| **Golden Thread Documentation Generator**: | |
| ```python | |
| class GoldenThreadGenerator: | |
| """ | |
| Generate and maintain Golden Thread documentation. | |
| """ | |
| def initialize_golden_thread( | |
| self, | |
| project: Project, | |
| gateway2_submission: Submission | |
| ) -> GoldenThread: | |
| """ | |
| Create initial Golden Thread structure for project. | |
| """ | |
| golden_thread = GoldenThread( | |
| project_id=project.id, | |
| created_date=datetime.now(), | |
| current_version="1.0" | |
| ) | |
| # Building Information | |
| golden_thread.building_info = self._extract_building_info( | |
| gateway2_submission | |
| ) | |
| # Design Information | |
| golden_thread.design_info = self._extract_design_info( | |
| gateway2_submission | |
| ) | |
| # Safety Case | |
| golden_thread.safety_case = self._generate_safety_case( | |
| gateway2_submission | |
| ) | |
| # Compliance Record | |
| golden_thread.compliance_record = self._create_compliance_record( | |
| gateway2_submission | |
| ) | |
| # Document Register | |
| golden_thread.document_register = self._create_document_register( | |
| gateway2_submission | |
| ) | |
| return golden_thread | |
| def update_golden_thread( | |
| self, | |
| golden_thread: GoldenThread, | |
| change: DesignChange, | |
| authorization: Authorization | |
| ) -> GoldenThread: | |
| """ | |
| Update Golden Thread with design change. | |
| """ | |
| # Create new version | |
| new_version = golden_thread.increment_version() | |
| # Record change | |
| new_version.add_change_record( | |
| date=datetime.now(), | |
| change_description=change.description, | |
| affected_elements=change.affected_elements, | |
| justification=change.justification, | |
| authorized_by=authorization.person, | |
| authorization_date=authorization.date | |
| ) | |
| # Update affected sections | |
| if change.affects_building_info: | |
| new_version.building_info = self._update_building_info( | |
| golden_thread.building_info, | |
| change | |
| ) | |
| if change.affects_design: | |
| new_version.design_info = self._update_design_info( | |
| golden_thread.design_info, | |
| change | |
| ) | |
| if change.affects_safety_case: | |
| new_version.safety_case = self._update_safety_case( | |
| golden_thread.safety_case, | |
| change | |
| ) | |
| # Re-validate compliance | |
| new_version.compliance_record = self._revalidate_compliance( | |
| new_version, | |
| change | |
| ) | |
| # Archive previous version | |
| self._archive_version(golden_thread) | |
| return new_version | |
| def generate_bsr_submission_package( | |
| self, | |
| golden_thread: GoldenThread, | |
| gateway: str # "2" or "3" | |
| ) -> BSRSubmissionPackage: | |
| """ | |
| Format Golden Thread for BSR portal submission. | |
| """ | |
| package = BSRSubmissionPackage( | |
| gateway=gateway, | |
| project_id=golden_thread.project_id, | |
| submission_date=datetime.now() | |
| ) | |
| # Include required documents per gateway | |
| if gateway == "2": | |
| package.add_document("building_information", golden_thread.building_info) | |
| package.add_document("design_information", golden_thread.design_info) | |
| package.add_document("safety_case", golden_thread.safety_case) | |
| package.add_document("compliance_record", golden_thread.compliance_record) | |
| package.add_document("fire_strategy", golden_thread.fire_strategy) | |
| elif gateway == "3": | |
| package.add_document("completion_certificate_application", ...) | |
| package.add_document("as_built_information", golden_thread.as_built_info) | |
| package.add_document("change_log", golden_thread.change_log) | |
| package.add_document("final_compliance_verification", ...) | |
| package.add_document("operation_and_maintenance_manual", ...) | |
| # Format per BSR requirements | |
| package.format_for_bsr() | |
| return package | |
| ``` | |
| --- | |
| ## **4. AGENT DESIGN SPECIFICATIONS** | |
| ### **4.1 Standard Agent Design Pattern** | |
| All Building Regulation Part agents follow this consistent pattern: | |
| ```python | |
| from dataclasses import dataclass | |
| from typing import List, Dict, Optional | |
| from enum import Enum | |
| @dataclass | |
| class ComplianceCheck: | |
| """ | |
| Standard structure for compliance findings. | |
| """ | |
| regulation: str # e.g., "Part B1 - Means of Escape" | |
| requirement: str # What the regulation requires | |
| status: str # "compliant", "non_compliant", "insufficient_info" | |
| evidence: List[str] # Specific evidence from submission | |
| reasoning: str # Explanation of determination | |
| confidence: float # 0.0 to 1.0 | |
| priority: str # "critical", "high", "medium", "low" | |
| recommendations: List[str] # Actions if non-compliant | |
| regulatory_references: List[str] # Approved Doc sections, standards | |
| related_checks: List[str] # Links to related findings | |
| class BuildingRegulationPartAgent: | |
| """ | |
| Template for all Part-specific agents. | |
| """ | |
| def __init__( | |
| self, | |
| part_id: str, | |
| rag_system: RAGSystem, | |
| spatial_analyzer: SpatialAnalyzer, | |
| tool_registry: ToolRegistry, | |
| llm_client: LLMClient | |
| ): | |
| self.part_id = part_id | |
| self.rag = rag_system | |
| self.spatial = spatial_analyzer | |
| self.tools = tool_registry | |
| self.llm = llm_client | |
| # Load Part-specific knowledge | |
| self.knowledge_base = self._load_regulatory_knowledge() | |
| # Register Part-specific tools | |
| self.specialized_tools = self._register_specialized_tools() | |
| def check_compliance( | |
| self, | |
| submission: Submission, | |
| spatial_data: SpatialData, | |
| specifications: Dict | |
| ) -> List[ComplianceCheck]: | |
| """ | |
| Main compliance checking entry point. | |
| Args: | |
| submission: Full Gateway submission | |
| spatial_data: Processed building spatial information | |
| specifications: Extracted technical specifications | |
| Returns: | |
| List of ComplianceCheck objects for all applicable requirements | |
| """ | |
| checks = [] | |
| # 1. Identify applicable requirements | |
| requirements = self._identify_applicable_requirements( | |
| submission, | |
| spatial_data, | |
| specifications | |
| ) | |
| # 2. Execute checks for each requirement | |
| for requirement in requirements: | |
| check = self._evaluate_requirement( | |
| requirement=requirement, | |
| submission=submission, | |
| spatial_data=spatial_data, | |
| specifications=specifications | |
| ) | |
| checks.append(check) | |
| # 3. Cross-validate findings | |
| checks = self._cross_validate_checks(checks) | |
| # 4. Prioritize issues | |
| checks = self._prioritize_checks(checks) | |
| return checks | |
| def _load_regulatory_knowledge(self) -> Dict: | |
| """ | |
| Load Part-specific regulatory knowledge from RAG system. | |
| Returns: | |
| Dictionary containing: | |
| - Approved Document sections | |
| - Relevant British Standards | |
| - Guidance documents | |
| - Historical precedents | |
| - Reference diagrams | |
| """ | |
| # Query RAG for Part-specific content | |
| queries = self._generate_knowledge_queries() | |
| knowledge = { | |
| 'approved_doc': [], | |
| 'standards': [], | |
| 'guidance': [], | |
| 'precedents': [], | |
| 'diagrams': [] | |
| } | |
| for query in queries: | |
| # Text retrieval | |
| text_results = self.rag.retrieve_text( | |
| query=query, | |
| collection=f"part_{self.part_id.lower()}_text", | |
| top_k=20 | |
| ) | |
| knowledge['approved_doc'].extend(text_results) | |
| # Diagram retrieval | |
| diagram_results = self.rag.retrieve_diagrams( | |
| query=query, | |
| collection=f"part_{self.part_id.lower()}_diagrams", | |
| top_k=5 | |
| ) | |
| knowledge['diagrams'].extend(diagram_results) | |
| return knowledge | |
| def _identify_applicable_requirements( | |
| self, | |
| submission: Submission, | |
| spatial_data: SpatialData, | |
| specifications: Dict | |
| ) -> List[Requirement]: | |
| """ | |
| Determine which specific requirements apply to this building. | |
| Requirements vary based on: | |
| - Building type (dwelling, commercial, HRB, etc.) | |
| - Building use | |
| - Building characteristics (height, occupancy, etc.) | |
| - Design approach (prescriptive vs performance) | |
| """ | |
| # Use LLM with RAG context to determine applicability | |
| building_characteristics = self._extract_characteristics( | |
| submission, | |
| spatial_data, | |
| specifications | |
| ) | |
| prompt = f""" | |
| Based on the following building characteristics, identify which | |
| requirements from Part {self.part_id} apply: | |
| Building Characteristics: | |
| {building_characteristics} | |
| Regulatory Context: | |
| {self.knowledge_base['approved_doc']} | |
| List the specific requirements that apply, with justification. | |
| """ | |
| response = self.llm.generate( | |
| prompt=prompt, | |
| temperature=0.1 # Low temperature for consistency | |
| ) | |
| requirements = self._parse_requirements(response) | |
| return requirements | |
| def _evaluate_requirement( | |
| self, | |
| requirement: Requirement, | |
| submission: Submission, | |
| spatial_data: SpatialData, | |
| specifications: Dict | |
| ) -> ComplianceCheck: | |
| """ | |
| Evaluate a specific requirement against submission. | |
| Process: | |
| 1. Extract relevant data using tools | |
| 2. Compare against requirement criteria | |
| 3. Generate evidence trail | |
| 4. Determine compliance status | |
| 5. Calculate confidence | |
| 6. Generate recommendations if non-compliant | |
| """ | |
| # This is implemented by each specific agent | |
| # with domain-specific logic | |
| raise NotImplementedError | |
| def _cross_validate_checks( | |
| self, | |
| checks: List[ComplianceCheck] | |
| ) -> List[ComplianceCheck]: | |
| """ | |
| Cross-validate findings for consistency. | |
| Ensures: | |
| - No contradictory findings | |
| - Related findings are consistent | |
| - Evidence supports conclusions | |
| """ | |
| # Check for contradictions | |
| contradictions = self._find_contradictions(checks) | |
| if contradictions: | |
| # Resolve contradictions (e.g., re-evaluate with more data) | |
| checks = self._resolve_contradictions(checks, contradictions) | |
| # Validate evidence chains | |
| for check in checks: | |
| check = self._validate_evidence(check) | |
| return checks | |
| def _prioritize_checks( | |
| self, | |
| checks: List[ComplianceCheck] | |
| ) -> List[ComplianceCheck]: | |
| """ | |
| Assign priority levels to findings. | |
| Priority based on: | |
| - Safety implications | |
| - Likelihood of BSR rejection | |
| - Difficulty of remediation | |
| - Cost implications | |
| """ | |
| for check in checks: | |
| if check.status != "compliant": | |
| check.priority = self._calculate_priority(check) | |
| # Sort by priority | |
| checks.sort(key=lambda c: PRIORITY_ORDER[c.priority]) | |
| return checks | |
| ``` | |
| ### **4.2 Specific Agent Examples** | |
| #### **Part A Agent: Structure** | |
| ```python | |
| class PartAAgent(BuildingRegulationPartAgent): | |
| """ | |
| Part A: Structure - Checks structural adequacy and safety. | |
| Key areas: | |
| - Loading (dead loads, imposed loads, wind loads, snow loads) | |
| - Structural integrity and stability | |
| - Disproportionate collapse (for HRBs) | |
| - Ground movement | |
| - Foundations | |
| """ | |
| def __init__(self, *args, **kwargs): | |
| super().__init__(part_id="A", *args, **kwargs) | |
| # Part A specific tools | |
| self.load_calculator = LoadCalculator() | |
| self.structural_analyzer = StructuralAnalyzer() | |
| def _evaluate_requirement( | |
| self, | |
| requirement: Requirement, | |
| submission: Submission, | |
| spatial_data: SpatialData, | |
| specifications: Dict | |
| ) -> ComplianceCheck: | |
| """ | |
| Evaluate structural requirement. | |
| """ | |
| if requirement.type == "loading": | |
| return self._check_loading(requirement, spatial_data, specifications) | |
| elif requirement.type == "disproportionate_collapse": | |
| return self._check_disproportionate_collapse(requirement, spatial_data, specifications) | |
| elif requirement.type == "foundations": | |
| return self._check_foundations(requirement, specifications) | |
| else: | |
| return self._generic_structural_check(requirement, spatial_data, specifications) | |
| def _check_loading( | |
| self, | |
| requirement: Requirement, | |
| spatial_data: SpatialData, | |
| specifications: Dict | |
| ) -> ComplianceCheck: | |
| """ | |
| Check loading assumptions are appropriate. | |
| Requirements: | |
| - Dead loads: weight of structure and fixed equipment | |
| - Imposed loads: occupancy loads per BS EN 1991-1-1 | |
| - Wind loads: per BS EN 1991-1-4 | |
| - Snow loads: per BS EN 1991-1-3 | |
| """ | |
| # Extract loading specifications | |
| loading_spec = specifications.get('structural_calculations', {}).get('loading', {}) | |
| # Validate against standards | |
| issues = [] | |
| # Check imposed loads | |
| for room in spatial_data.rooms: | |
| room_use = self._determine_room_use(room) | |
| required_imposed_load = IMPOSED_LOADS_BS_EN_1991[room_use] | |
| specified_load = loading_spec.get(room.id, {}).get('imposed_load') | |
| if specified_load is None: | |
| issues.append(f"No imposed load specified for {room.name}") | |
| elif specified_load < required_imposed_load: | |
| issues.append( | |
| f"{room.name}: Imposed load {specified_load} kN/m² " | |
| f"< required {required_imposed_load} kN/m²" | |
| ) | |
| if issues: | |
| return ComplianceCheck( | |
| regulation="Part A - Loading", | |
| requirement="Loading assumptions must comply with BS EN 1991", | |
| status="non_compliant" if any("specified" in i for i in issues) else "insufficient_info", | |
| evidence=issues, | |
| reasoning="Loading specifications do not meet BS EN 1991 requirements for all areas.", | |
| confidence=0.90, | |
| priority="high", | |
| recommendations=[ | |
| "Review imposed load specifications per BS EN 1991-1-1 Table 6.2", | |
| "Ensure all room types have appropriate loading specified", | |
| "Provide structural engineer's justification for any reduced loads" | |
| ], | |
| regulatory_references=["BS EN 1991-1-1", "Approved Document A Section 2"] | |
| ) | |
| else: | |
| return ComplianceCheck( | |
| regulation="Part A - Loading", | |
| requirement="Loading assumptions comply with standards", | |
| status="compliant", | |
| evidence=["Loading specifications reviewed", "All loads meet or exceed BS EN 1991 requirements"], | |
| reasoning="Structural loading specifications are adequate per BS EN 1991 series.", | |
| confidence=0.85, | |
| priority="medium", | |
| recommendations=[], | |
| regulatory_references=["BS EN 1991-1-1"] | |
| ) | |
| def _check_disproportionate_collapse( | |
| self, | |
| requirement: Requirement, | |
| spatial_data: SpatialData, | |
| specifications: Dict | |
| ) -> ComplianceCheck: | |
| """ | |
| Check disproportionate collapse provisions for HRBs. | |
| Requirement (Part A3): | |
| Buildings must be designed so that localized failure doesn't cause | |
| disproportionate collapse. | |
| For HRBs (Class 2B): Specific provisions required: | |
| - Effective horizontal ties | |
| - Effective vertical ties | |
| - Key element design | |
| - Or systematic risk assessment | |
| """ | |
| building_class = self._determine_building_class(spatial_data, specifications) | |
| if building_class != "2B": | |
| # Not HRB, standard provisions sufficient | |
| return ComplianceCheck( | |
| regulation="Part A3 - Disproportionate Collapse", | |
| requirement="Building class determined, Class 2B provisions not required", | |
| status="compliant", | |
| evidence=[f"Building class: {building_class}", "Class 2B requirements not applicable"], | |
| reasoning="Building does not fall into Class 2B (HRB) category requiring enhanced disproportionate collapse provisions.", | |
| confidence=0.95, | |
| priority="low", | |
| recommendations=[], | |
| regulatory_references=["Approved Document A Section 5"] | |
| ) | |
| # Class 2B building - check for provisions | |
| structural_strategy = specifications.get('structural_calculations', {}).get('disproportionate_collapse', '') | |
| has_ties = 'ties' in structural_strategy.lower() or 'tying' in structural_strategy.lower() | |
| has_key_elements = 'key element' in structural_strategy.lower() | |
| has_risk_assessment = 'risk assessment' in structural_strategy.lower() | |
| if not (has_ties or has_key_elements or has_risk_assessment): | |
| return ComplianceCheck( | |
| regulation="Part A3 - Disproportionate Collapse (Class 2B)", | |
| requirement="HRBs must have disproportionate collapse provisions", | |
| status="insufficient_info", | |
| evidence=[ | |
| "Building is Class 2B (HRB)", | |
| "No disproportionate collapse strategy evident" | |
| ], | |
| reasoning="Class 2B buildings require specific provisions per Approved Document A Section 5. No strategy documented.", | |
| confidence=0.95, | |
| priority="critical", | |
| recommendations=[ | |
| "Provide structural engineer's disproportionate collapse strategy", | |
| "Options:", | |
| " 1. Effective horizontal and vertical ties (typical approach)", | |
| " 2. Key element design (elements to withstand 34 kN/m²)", | |
| " 3. Systematic risk assessment per BS EN 1991-1-7", | |
| "Submit calculations demonstrating chosen approach" | |
| ], | |
| regulatory_references=["Approved Document A Section 5", "BS EN 1992-1-1 Section 9"] | |
| ) | |
| # Strategy present - validate it | |
| return ComplianceCheck( | |
| regulation="Part A3 - Disproportionate Collapse (Class 2B)", | |
| requirement="Disproportionate collapse strategy for HRB", | |
| status="compliant", | |
| evidence=[ | |
| "Disproportionate collapse strategy provided", | |
| "Ties" if has_ties else "", | |
| "Key elements" if has_key_elements else "", | |
| "Risk assessment" if has_risk_assessment else "" | |
| ], | |
| reasoning="Disproportionate collapse provisions documented for Class 2B building.", | |
| confidence=0.80, | |
| priority="high", | |
| recommendations=[ | |
| "Verify structural calculations support stated strategy", | |
| "Ensure design reviewed by Building Safety Regulator" | |
| ], | |
| regulatory_references=["Approved Document A Section 5"] | |
| ) | |
| ``` | |
| #### **Part B Agent: Fire Safety** | |
| ```python | |
| class PartBAgent(BuildingRegulationPartAgent): | |
| """ | |
| Part B: Fire Safety - Most complex and critical agent. | |
| Sections: | |
| - B1: Means of escape | |
| - B2: Internal fire spread (linings) | |
| - B3: Internal fire spread (structure) | |
| - B4: External fire spread | |
| - B5: Access and facilities for fire service | |
| This is often the primary cause of Gateway rejections. | |
| """ | |
| def __init__(self, *args, **kwargs): | |
| super().__init__(part_id="B", *args, **kwargs) | |
| # Part B specific tools | |
| self.egress_simulator = EgressSimulator() | |
| self.compartmentation_checker = CompartmentationChecker() | |
| self.fire_resistance_validator = FireResistanceValidator() | |
| def _evaluate_requirement( | |
| self, | |
| requirement: Requirement, | |
| submission: Submission, | |
| spatial_data: SpatialData, | |
| specifications: Dict | |
| ) -> ComplianceCheck: | |
| """ | |
| Evaluate fire safety requirement. | |
| """ | |
| if requirement.section == "B1": | |
| return self._check_means_of_escape(requirement, spatial_data, specifications) | |
| elif requirement.section == "B2": | |
| return self._check_internal_linings(requirement, specifications) | |
| elif requirement.section == "B3": | |
| return self._check_fire_resistance(requirement, spatial_data, specifications) | |
| elif requirement.section == "B4": | |
| return self._check_external_fire_spread(requirement, spatial_data, specifications) | |
| elif requirement.section == "B5": | |
| return self._check_fire_service_access(requirement, spatial_data, specifications) | |
| def _check_means_of_escape( | |
| self, | |
| requirement: Requirement, | |
| spatial_data: SpatialData, | |
| specifications: Dict | |
| ) -> ComplianceCheck: | |
| """ | |
| Check means of escape provisions (B1). | |
| Key requirements: | |
| - Travel distances to exits | |
| - Number of escape routes | |
| - Width of escape routes | |
| - Protected stairways | |
| - Emergency lighting | |
| - Signage | |
| For HRBs: Enhanced requirements including 2.5 minute evacuation time. | |
| """ | |
| is_hrb = specifications.get('building_type') == 'HRB' | |
| # Calculate travel distances | |
| travel_distance_issues = [] | |
| for room in spatial_data.rooms: | |
| if room.is_habitable: | |
| routes = self.spatial.find_egress_routes( | |
| start_room=room.id, | |
| building_id=spatial_data.building_id | |
| ) | |
| shortest_route = min(routes, key=lambda r: r.distance) | |
| # Determine maximum permitted travel distance | |
| if is_hrb: | |
| max_distance = 9.0 if len(routes) < 2 else 18.0 # meters | |
| else: | |
| max_distance = self._get_max_travel_distance( | |
| room.use, | |
| len(routes) | |
| ) | |
| if shortest_route.distance > max_distance: | |
| travel_distance_issues.append({ | |
| 'room': room.name, | |
| 'distance': shortest_route.distance, | |
| 'max_allowed': max_distance, | |
| 'excess': shortest_route.distance - max_distance | |
| }) | |
| if travel_distance_issues: | |
| return ComplianceCheck( | |
| regulation="Part B1 - Means of Escape (Travel Distance)", | |
| requirement=f"Travel distance to nearest exit: {'HRB: 9m (one direction) or 18m (two directions)' if is_hrb else 'Per Approved Document B Table 3.1'}", | |
| status="non_compliant", | |
| evidence=[ | |
| f"{len(travel_distance_issues)} room(s) exceed travel distance limits", | |
| f"Worst case: {max(travel_distance_issues, key=lambda x: x['excess'])['room']} " | |
| f"({max(travel_distance_issues, key=lambda x: x['excess'])['distance']:.1f}m)" | |
| ], | |
| reasoning="Travel distances exceed limits per Approved Document B. This is a critical life safety issue.", | |
| confidence=0.95, | |
| priority="critical", | |
| recommendations=[ | |
| "Reduce travel distances by:", | |
| " - Adding additional exits/stairways", | |
| " - Reconfiguring floor layout", | |
| " - Repositioning protected escape routes", | |
| f"Rooms requiring attention: {', '.join(i['room'] for i in travel_distance_issues[:5])}" | |
| ], | |
| regulatory_references=["Approved Document B Volume 2 Table 3.1", "BS 9999"] | |
| ) | |
| # Check alternative means of escape | |
| alternative_escape_issues = [] | |
| for room in spatial_data.rooms: | |
| if room.requires_alternative_escape: # Determined by room characteristics | |
| routes = self.spatial.find_egress_routes(room.id, spatial_data.building_id) | |
| if len(routes) < 2: | |
| alternative_escape_issues.append({ | |
| 'room': room.name, | |
| 'routes_found': len(routes), | |
| 'routes_required': 2 | |
| }) | |
| if alternative_escape_issues: | |
| return ComplianceCheck( | |
| regulation="Part B1 - Alternative Means of Escape", | |
| requirement="Rooms requiring alternative escape must have two independent routes", | |
| status="non_compliant", | |
| evidence=[ | |
| f"{len(alternative_escape_issues)} room(s) lack alternative escape routes", | |
| "Rooms affected: " + ", ".join(i['room'] for i in alternative_escape_issues[:10]) | |
| ], | |
| reasoning="Insufficient alternative means of escape. Some rooms only have single escape route when two required.", | |
| confidence=0.90, | |
| priority="critical", | |
| recommendations=[ | |
| "Provide second independent escape route for affected rooms", | |
| "Routes must be truly independent (not converge before final exit)", | |
| "Consider additional external escape stair if necessary" | |
| ], | |
| regulatory_references=["Approved Document B Section 2"] | |
| ) | |
| # If we reach here, basic egress provisions appear satisfactory | |
| return ComplianceCheck( | |
| regulation="Part B1 - Means of Escape", | |
| requirement="Adequate means of escape provisions", | |
| status="compliant", | |
| evidence=[ | |
| f"Travel distances checked for {len(spatial_data.rooms)} rooms", | |
| "All travel distances within limits", | |
| "Alternative escape routes provided where required" | |
| ], | |
| reasoning="Means of escape provisions meet Approved Document B requirements.", | |
| confidence=0.85, | |
| priority="high", | |
| recommendations=[ | |
| "Verify emergency lighting and signage specifications", | |
| "Ensure protected routes maintained during construction" | |
| ], | |
| regulatory_references=["Approved Document B Section 2"] | |
| ) | |
| ``` | |
| ### **4.3 RAG Strategy Per Agent** | |
| Each agent has tailored RAG retrieval strategy: | |
| #### **Part A (Structure) RAG Strategy** | |
| ```python | |
| class PartARAGStrategy: | |
| """ | |
| RAG strategy optimized for structural requirements. | |
| """ | |
| def generate_queries(self, building_characteristics: Dict) -> List[str]: | |
| """ | |
| Generate queries specific to structural checking. | |
| """ | |
| queries = [ | |
| # Loading queries | |
| f"imposed loads for {building_characteristics['use']} buildings BS EN 1991", | |
| f"wind loads {building_characteristics['location']} UK", | |
| f"snow loads {building_characteristics['location']} UK", | |
| # Structural integrity | |
| f"structural stability requirements {building_characteristics['height']}m building", | |
| f"disproportionate collapse class 2B {building_characteristics['storeys']} storey", | |
| # Foundations | |
| f"foundation design {building_characteristics['ground_conditions']}", | |
| f"foundation depth frost heave UK", | |
| # Fire resistance (structural) | |
| f"fire resistance requirements {building_characteristics['use']} {building_characteristics['height']}m", | |
| f"structural fire design {building_characteristics['structural_material']}" | |
| ] | |
| return queries | |
| def retrieve_context( | |
| self, | |
| query: str, | |
| rag_system: RAGSystem | |
| ) -> Dict: | |
| """ | |
| Retrieve context with emphasis on standards and calculations. | |
| """ | |
| # Text retrieval with focus on BS EN standards | |
| text_results = rag_system.retrieve_text( | |
| query=query, | |
| collection="part_a_text", | |
| top_k=15, | |
| filters={'document_type': ['approved_document', 'bs_en_standard']} | |
| ) | |
| # Retrieve calculation examples | |
| precedent_results = rag_system.retrieve_text( | |
| query=f"{query} calculation example", | |
| collection="structural_precedents", | |
| top_k=5 | |
| ) | |
| # Retrieve diagrams | |
| diagram_results = rag_system.retrieve_diagrams( | |
| query=query, | |
| collection="part_a_diagrams", | |
| top_k=3 | |
| ) | |
| return { | |
| 'regulatory_text': text_results, | |
| 'precedents': precedent_results, | |
| 'diagrams': diagram_results | |
| } | |
| ``` | |
| #### **Part B (Fire Safety) RAG Strategy** | |
| ```python | |
| class PartBRAGStrategy: | |
| """ | |
| RAG strategy for fire safety - most complex due to broad scope. | |
| """ | |
| def generate_queries(self, building_characteristics: Dict) -> List[str]: | |
| """ | |
| Generate comprehensive fire safety queries. | |
| """ | |
| queries = [] | |
| # Means of escape queries | |
| queries.extend([ | |
| f"travel distance requirements {building_characteristics['use']}", | |
| f"alternative means of escape {building_characteristics['occupancy']} occupants", | |
| f"protected stairway requirements {building_characteristics['storeys']} storeys", | |
| f"evacuation time requirements HRB" if building_characteristics['is_hrb'] else "", | |
| ]) | |
| # Compartmentation queries | |
| queries.extend([ | |
| f"compartmentation requirements {building_characteristics['use']}", | |
| f"fire resistance periods {building_characteristics['height']}m building", | |
| f"cavity barriers requirements", | |
| ]) | |
| # External fire spread (post-Grenfell critical) | |
| if building_characteristics['is_hrb']: | |
| queries.extend([ | |
| f"external wall systems HRB requirements post-Grenfell", | |
| f"cladding fire performance requirements", | |
| f"combustible materials restrictions HRB" | |
| ]) | |
| # Fire service access | |
| queries.extend([ | |
| f"fire fighting shaft requirements {building_characteristics['height']}m", | |
| f"fire service vehicle access requirements", | |
| f"dry rising mains requirements" | |
| ]) | |
| return [q for q in queries if q] # Remove empty strings | |
| def retrieve_context( | |
| self, | |
| query: str, | |
| rag_system: RAGSystem, | |
| is_hrb: bool | |
| ) -> Dict: | |
| """ | |
| Retrieve fire safety context with HRB emphasis if applicable. | |
| """ | |
| # Base retrieval | |
| text_results = rag_system.retrieve_text( | |
| query=query, | |
| collection="part_b_text", | |
| top_k=20 # More results due to complexity | |
| ) | |
| # HRB-specific guidance if applicable | |
| if is_hrb: | |
| hrb_results = rag_system.retrieve_text( | |
| query=f"{query} HRB higher risk building", | |
| collection="hrb_guidance", | |
| top_k=10 | |
| ) | |
| else: | |
| hrb_results = [] | |
| # Rejection precedents (learn from common failures) | |
| rejection_results = rag_system.retrieve_text( | |
| query=f"{query} BSR rejection reasons", | |
| collection="bsr_rejections", | |
| top_k=5 | |
| ) | |
| # Diagrams (especially for egress layouts) | |
| diagram_results = rag_system.retrieve_diagrams( | |
| query=query, | |
| collection="part_b_diagrams", | |
| top_k=5 | |
| ) | |
| return { | |
| 'regulatory_text': text_results, | |
| 'hrb_guidance': hrb_results, | |
| 'rejection_precedents': rejection_results, | |
| 'diagrams': diagram_results | |
| } | |
| ``` | |
| --- | |
| ## **5. REQUIRED OUTCOMES** | |
| ### **5.1 Primary Deliverables** | |
| #### **Gateway 2 Compliance Package** | |
| **Contents**: | |
| 1. **Executive Summary** | |
| - Overall compliance status | |
| - Critical issues requiring immediate attention | |
| - Summary statistics (compliant vs non-compliant checks) | |
| - Estimated likelihood of BSR approval | |
| 2. **Part-by-Part Compliance Commentary** | |
| - Detailed findings for each applicable Building Regulation Part (A-T) | |
| - Evidence-based determinations with regulatory citations | |
| - Confidence scores for each finding | |
| - NOT design advice - commentary only per RBI/Chartered Building Engineer code | |
| 3. **Cross-Part Analysis** | |
| - Identification of conflicts between different regulation parts | |
| - Recommendations for resolving conflicts | |
| - Dependencies and interactions highlighted | |
| 4. **Gap Analysis** | |
| - Missing information that prevents full assessment | |
| - Additional documentation required | |
| - Clarifications needed from design team | |
| 5. **Prioritized Issue List** | |
| - Critical issues (life safety, structural integrity, likely BSR rejection) | |
| - High priority issues (compliance required, moderate remediation difficulty) | |
| - Medium priority issues (compliance preferred, easier remediation) | |
| - Low priority observations (enhancement suggestions, best practice) | |
| 6. **Evidence Package** | |
| - Document references supporting each finding | |
| - Drawing annotations highlighting issues | |
| - Calculation summaries where applicable | |
| - Product verification results | |
| 7. **Recommendations** | |
| - Specific, actionable steps to address non-compliances | |
| - Alternative approaches where applicable | |
| - Suggested consultations with specialists | |
| - Timeline implications of changes | |
| **Format**: | |
| - PDF report for human review | |
| - Structured JSON/XML for system integration | |
| - Interactive dashboard for project team access | |
| - Annotated drawings with issue highlighting | |
| **Example Report Structure**: | |
| ``` | |
| BUILDWELLTHREAD COMPLIANCE REPORT | |
| Gateway 2 Submission Assessment | |
| Project: [Project Name] | |
| Address: [Address] | |
| Submission Date: [Date] | |
| Report Date: [Date] | |
| Reference: [Unique ID] | |
| EXECUTIVE SUMMARY | |
| ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| Overall Assessment: NON-COMPLIANT [requires revisions] | |
| Compliance Summary: | |
| ✓ Compliant: 156 checks (64%) | |
| ✗ Non-Compliant: 45 checks (18%) | |
| ? Requires Clarification: 44 checks (18%) | |
| Total Checks Performed: 245 | |
| CRITICAL ISSUES (IMMEDIATE ATTENTION REQUIRED): | |
| 1. Part B1 - Travel Distance: 8 rooms exceed maximum (see Section 2.1) | |
| 2. Part B3 - Fire Resistance: Compartmentation incomplete (see Section 2.3) | |
| 3. Part A3 - Disproportionate Collapse: Strategy not provided for HRB (see Section 1.3) | |
| LIKELIHOOD OF BSR APPROVAL: LOW (without revisions) | |
| Estimated remediation time: 4-6 weeks | |
| ... | |
| PART A: STRUCTURE | |
| ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| A1: LOADING | |
| ✓ COMPLIANT: Imposed load specifications | |
| Status: Compliant | |
| Confidence: 85% | |
| Evidence: | |
| - Imposed loads specified per BS EN 1991-1-1 Table 6.2 | |
| - All room types have appropriate loading | |
| - Structural calculations signed by Chartered Engineer | |
| ✗ NON-COMPLIANT: Wind load calculations | |
| Status: Non-Compliant | |
| Priority: HIGH | |
| Confidence: 90% | |
| Evidence: | |
| - Wind load calculations use Eurocode but not UK National Annex | |
| - Basic wind speed not adjusted for UK location | |
| - No consideration of terrain category | |
| Reasoning: | |
| Wind load calculations must use UK National Annex to BS EN 1991-1-4. | |
| Current calculations use generic Eurocode values which are not appropriate | |
| for UK sites. This affects sizing of structural elements and connections. | |
| Recommendations: | |
| 1. Recalculate wind loads using BS EN 1991-1-4 with UK National Annex | |
| 2. Use UK wind map for site-specific basic wind speed | |
| 3. Include terrain category and orography factors for site | |
| 4. Verify structural adequacy with revised loads | |
| Regulatory References: | |
| - BS EN 1991-1-4:2005+A1:2010 | |
| - UK National Annex to BS EN 1991-1-4 | |
| - Approved Document A Section 2.1 | |
| ... | |
| ``` | |
| #### **Gateway 3 Support Package** | |
| **Contents**: | |
| 1. **Completion Certificate Readiness Assessment** | |
| - Verification all Gateway 2 conditions satisfied | |
| - Outstanding items list | |
| - Completion certificate application status | |
| 2. **As-Built Validation** | |
| - Comparison of as-built to approved drawings | |
| - Material substitutions verification | |
| - Performance testing results review | |
| 3. **Change Log** | |
| - All variations from approved Gateway 2 submission | |
| - Justification for changes | |
| - Re-validation of compliance where affected | |
| - Authorization trail | |
| 4. **Final Compliance Verification** | |
| - Confirmation all Building Regulations requirements met | |
| - Testing and commissioning results verification | |
| - Certification package completeness check | |
| 5. **Operation and Maintenance Information** | |
| - Building manual adequacy assessment | |
| - Safety information compilation | |
| - Handover documentation review | |
| **Outcome**: Supporting documentation for Gateway 3 submission with confidence that completion certificate will be granted. | |
| #### **Golden Thread Documentation** | |
| **Structure** (per Building Safety Act requirements): | |
| ``` | |
| Golden Thread Information Set | |
| ├── Building Information | |
| │ ├── Location and address | |
| │ ├── Building characteristics (height, storeys, occupancy) | |
| │ ├── Use and occupancy information | |
| │ ├── Principal contractor details | |
| │ └── Principal designer details | |
| │ | |
| ├── Design Information | |
| │ ├── Architectural plans (by floor and section) | |
| │ ├── Structural design | |
| │ ├── Fire safety strategy | |
| │ ├── Building services design | |
| │ ├── Materials specifications | |
| │ └── Product certifications | |
| │ | |
| ├── Safety Case | |
| │ ├── Fire safety strategy | |
| │ ├── Structural fire protection | |
| │ ├── Means of escape provisions | |
| │ ├── Fire service access and facilities | |
| │ ├── Risk assessments | |
| │ └── Safety management plan | |
| │ | |
| ├── Compliance Record | |
| │ ├── Gateway 2 approval documentation | |
| │ ├── Gateway 3 completion certificate | |
| │ ├── Building Regulations compliance certificates | |
| │ ├── Fire safety compliance | |
| │ ├── Structural approval | |
| │ └── Other regulatory approvals | |
| │ | |
| ├── Change Control | |
| │ ├── Design change register | |
| │ ├── Change authorization records | |
| │ ├── Impact assessments | |
| │ ├── Re-validation of compliance | |
| │ └── As-built vs approved comparison | |
| │ | |
| ├── Construction Phase Information | |
| │ ├── Progress updates | |
| │ ├── Inspection reports | |
| │ ├── Testing and commissioning | |
| │ ├── Site safety records | |
| │ └── Handover information | |
| │ | |
| └── Operation and Maintenance | |
| ├── Building user manual | |
| ├── Safety information for residents | |
| ├── Maintenance schedules | |
| ├── Emergency procedures | |
| └── Contact information | |
| ``` | |
| **Characteristics**: | |
| - **Structured**: Consistent format following Building Safety Act guidance | |
| - **Versioned**: Complete change history with dates and authorization | |
| - **Accessible**: Different views for different stakeholders (residents, facilities management, BSR) | |
| - **Persistent**: Maintained throughout building lifecycle | |
| - **Interoperable**: Compatible with BSR digital platform | |
| ### **5.2 Performance Targets** | |
| #### **Accuracy Targets** | |
| **Overall Goal**: >90% agreement with expert human reviewers | |
| **By Priority Level**: | |
| - Critical safety issues: >95% detection rate (minimize false negatives) | |
| - High priority compliance: >90% accuracy | |
| - Medium priority: >85% accuracy | |
| - Low priority observations: >80% accuracy | |
| **False Positive Tolerance**: <15% (balance between catching issues and not overwhelming with false flags) | |
| #### **Processing Targets** | |
| - **Time**: 45-90 minutes per submission (end-to-end) | |
| - **Throughput**: Process multiple submissions in parallel | |
| - **Availability**: 99% uptime (critical for workflow integration) | |
| - **Scalability**: Support 10-50 concurrent submissions | |
| #### **Quality Targets** | |
| - **Completeness**: 100% of applicable Building Regulations Parts checked | |
| - **Evidence Quality**: Every finding supported by specific document references | |
| - **Clarity**: Recommendations actionable without additional interpretation | |
| - **Consistency**: Same submission produces same results across multiple runs | |
| #### **Business Targets** | |
| - **BSR Rejection Rate Reduction**: Target <20% (from current 40-67%) | |
| - **Time Savings**: Reduce manual review time by 70-80% | |
| - **Cost Reduction**: Lower compliance checking costs by 50%+ | |
| - **Quality Improvement**: Catch issues earlier in design process | |
| ### **5.3 Success Metrics** | |
| #### **Technical Success Indicators** | |
| 1. **Validation Against Human Experts** | |
| - Blind comparison: AI findings vs expert review | |
| - Agreement rate measured | |
| - Types of discrepancies analyzed | |
| - Continuous improvement based on discrepancies | |
| 2. **BSR Gateway Performance** | |
| - Track rejection rates of AI-reviewed submissions | |
| - Analyze rejection reasons (were they preventable?) | |
| - Compare to industry baseline | |
| - Measure improvement over time | |
| 3. **User Feedback** | |
| - Usefulness ratings from design teams | |
| - Accuracy perception from reviewers | |
| - Time savings reported | |
| - Adoption rate within organization | |
| #### **Operational Success Indicators** | |
| 1. **System Reliability** | |
| - Uptime percentage | |
| - Mean time between failures | |
| - Average processing time | |
| - Error rate by component | |
| 2. **Golden Thread Utility** | |
| - Completeness of generated documentation | |
| - Acceptance by BSR | |
| - Ease of maintenance and updates | |
| - Stakeholder satisfaction | |
| 3. **Business Impact** | |
| - Cost per submission | |
| - Time to completion | |
| - Number of design iterations reduced | |
| - Overall project timeline improvement | |
| --- | |
| ## **6. ERROR HANDLING AND QUALITY ASSURANCE** | |
| ### **6.1 Error Categories and Responses** | |
| #### **Input Processing Errors** | |
| **Corrupt or Unreadable PDFs**: | |
| ```python | |
| class CorruptPDFHandler: | |
| """ | |
| Handle corrupted or problematic PDF inputs. | |
| """ | |
| def handle(self, pdf_path: str) -> Result: | |
| """ | |
| Attempt recovery or graceful degradation. | |
| """ | |
| try: | |
| # Attempt standard parsing | |
| document = parse_pdf(pdf_path) | |
| return Success(document) | |
| except CorruptPDFError as e: | |
| # Attempt recovery | |
| try: | |
| repaired = repair_pdf(pdf_path) | |
| document = parse_pdf(repaired) | |
| return Success( | |
| document, | |
| warnings=["PDF was repaired - verify content integrity"] | |
| ) | |
| except Exception: | |
| # Cannot recover | |
| return Failure( | |
| error="PDF is corrupt and cannot be recovered", | |
| action="manual_review_required", | |
| user_message="This PDF cannot be processed. Please re-export and resubmit.", | |
| file=pdf_path | |
| ) | |
| ``` | |
| **Poor Quality Drawings**: | |
| ```python | |
| class DrawingQualityAssessor: | |
| """ | |
| Assess and handle low-quality drawings. | |
| """ | |
| def assess_quality(self, image: Image) -> QualityScore: | |
| """ | |
| Score drawing quality on multiple dimensions. | |
| """ | |
| scores = { | |
| 'resolution': self._check_resolution(image), | |
| 'contrast': self._check_contrast(image), | |
| 'clarity': self._check_clarity(image), | |
| 'completeness': self._check_completeness(image) | |
| } | |
| overall = np.mean(list(scores.values())) | |
| return QualityScore( | |
| overall=overall, | |
| dimensions=scores | |
| ) | |
| def handle(self, image: Image, quality: QualityScore) -> Result: | |
| """ | |
| Decide how to proceed based on quality. | |
| """ | |
| if quality.overall > 0.8: | |
| # High quality - proceed normally | |
| return Success(image) | |
| elif quality.overall > 0.5: | |
| # Acceptable but suboptimal | |
| # Proceed with warnings and reduced confidence | |
| return Success( | |
| image, | |
| warnings=[ | |
| f"Drawing quality suboptimal (score: {quality.overall:.2f})", | |
| "Some automated measurements may be less reliable", | |
| "Manual verification recommended for critical dimensions" | |
| ], | |
| confidence_penalty=0.15 | |
| ) | |
| else: | |
| # Unacceptable quality | |
| # Request better quality submission | |
| return Failure( | |
| error="Drawing quality insufficient for automated analysis", | |
| quality_score=quality.overall, | |
| issues=[dim for dim, score in quality.dimensions.items() if score < 0.5], | |
| user_message="Please provide higher quality drawings (minimum 300 DPI, clear lines)", | |
| action="resubmission_required" | |
| ) | |
| ``` | |
| **Missing or Incomplete Submissions**: | |
| ```python | |
| class CompletenessChecker: | |
| """ | |
| Validate submission completeness before processing. | |
| """ | |
| REQUIRED_DOCUMENTS = { | |
| 'gateway_2': [ | |
| 'site_plan', | |
| 'floor_plans', | |
| 'elevations', | |
| 'sections', | |
| 'specifications', | |
| 'structural_calculations', | |
| 'fire_strategy' | |
| ] | |
| } | |
| def check(self, submission: Submission) -> CompletenessResult: | |
| """ | |
| Check if submission is complete. | |
| """ | |
| required = self.REQUIRED_DOCUMENTS['gateway_2'] | |
| provided = submission.document_types | |
| missing = [doc for doc in required if doc not in provided] | |
| if missing: | |
| return CompletenessResult( | |
| complete=False, | |
| missing_documents=missing, | |
| can_proceed=len(missing) < 3, # Can partially process if only minor gaps | |
| user_message=self._format_missing_docs_message(missing) | |
| ) | |
| return CompletenessResult( | |
| complete=True, | |
| missing_documents=[], | |
| can_proceed=True | |
| ) | |
| def _format_missing_docs_message(self, missing: List[str]) -> str: | |
| """ | |
| Create user-friendly message about missing documents. | |
| """ | |
| return f""" | |
| The following required documents are missing from your submission: | |
| {self._format_list(missing)} | |
| Please provide these documents to enable complete compliance checking. | |
| Note: We can perform partial analysis with the documents provided, | |
| but some compliance checks cannot be completed without the missing items. | |
| """ | |
| ``` | |
| #### **Spatial Processing Errors** | |
| **FloorplanTransformation Failures**: | |
| ```python | |
| class FloorplanTransformationErrorHandler: | |
| """ | |
| Handle failures in spatial processing pipeline. | |
| """ | |
| def handle( | |
| self, | |
| pdf_path: str, | |
| error: Exception | |
| ) -> Result: | |
| """ | |
| Attempt fallback strategies when FloorplanTransformation fails. | |
| """ | |
| # Log the error | |
| logger.error(f"FloorplanTransformation failed for {pdf_path}: {error}") | |
| # Attempt fallback strategy 1: Alternative CV model | |
| try: | |
| spatial_data = self.alternative_cv_model.process(pdf_path) | |
| return Success( | |
| spatial_data, | |
| warnings=[ | |
| "Used alternative processing method", | |
| "Spatial measurements may be less accurate", | |
| "Manual verification recommended" | |
| ], | |
| confidence_penalty=0.20 | |
| ) | |
| except Exception as e2: | |
| logger.error(f"Alternative CV model also failed: {e2}") | |
| # Attempt fallback strategy 2: Manual annotation interface | |
| try: | |
| # Create manual annotation task | |
| task_id = self.manual_annotation_queue.create_task( | |
| pdf_path=pdf_path, | |
| priority="high", | |
| reason="Automated processing failed" | |
| ) | |
| return Pending( | |
| message="Automated processing failed - manual annotation required", | |
| task_id=task_id, | |
| estimated_time="30-60 minutes", | |
| user_message="We're having trouble automatically processing these plans. " | |
| "Our team will manually review and we'll notify you when complete." | |
| ) | |
| except Exception as e3: | |
| logger.error(f"Could not create manual task: {e3}") | |
| # Complete failure - cannot proceed | |
| return Failure( | |
| error="Unable to process floorplans", | |
| action="expert_review_required", | |
| user_message="These plans could not be processed automatically. " | |
| "Please contact support for manual review options." | |
| ) | |
| ``` | |
| **Ambiguous Element Detection**: | |
| ```python | |
| class AmbiguityResolver: | |
| """ | |
| Handle uncertain spatial element identification. | |
| """ | |
| def resolve( | |
| self, | |
| element: SpatialElement, | |
| confidence: float | |
| ) -> Resolution: | |
| """ | |
| Decide how to handle ambiguous elements. | |
| """ | |
| if confidence > 0.9: | |
| # High confidence - accept | |
| return Accept(element) | |
| elif confidence > 0.6: | |
| # Medium confidence - flag and proceed with caveats | |
| return AcceptWithCaveats( | |
| element=element, | |
| confidence=confidence, | |
| warnings=[ | |
| f"Element '{element.id}' identification uncertain (confidence: {confidence:.0%})", | |
| f"Type: {element.type} (alternative: {element.alternative_types})", | |
| "Recommend manual verification" | |
| ] | |
| ) | |
| else: | |
| # Low confidence - request clarification | |
| return RequestClarification( | |
| element=element, | |
| confidence=confidence, | |
| question=f"Please clarify: Is element at {element.location} a {element.type}?", | |
| alternatives=element.alternative_types, | |
| impact="Affects: " + ", ".join(element.dependent_checks) | |
| ) | |
| ``` | |
| #### **RAG System Errors** | |
| **No Relevant Regulations Retrieved**: | |
| ```python | |
| class RetrievalFailureHandler: | |
| """ | |
| Handle cases where RAG system returns no relevant results. | |
| """ | |
| def handle( | |
| self, | |
| query: str, | |
| collection: str | |
| ) -> Result: | |
| """ | |
| Respond to failed retrieval. | |
| """ | |
| # Log the issue | |
| logger.warning(f"No results for query: {query} in collection: {collection}") | |
| # Attempt query expansion | |
| expanded_queries = self._expand_query(query) | |
| for expanded in expanded_queries: | |
| results = self.rag.retrieve(expanded, collection) | |
| if results: | |
| return Success( | |
| results, | |
| warnings=["Used expanded query for retrieval"] | |
| ) | |
| # Still no results - this might be a novel situation | |
| return Failure( | |
| error="No relevant regulatory guidance found", | |
| query=query, | |
| action="flag_for_expert", | |
| user_message="This appears to be a novel or unusual situation. " | |
| "Expert review recommended.", | |
| escalation_priority="medium" | |
| ) | |
| def _expand_query(self, query: str) -> List[str]: | |
| """ | |
| Generate alternative queries. | |
| """ | |
| # Use LLM to generate alternative phrasings | |
| prompt = f""" | |
| The following query returned no results: | |
| "{query}" | |
| Generate 3 alternative queries that might retrieve relevant information | |
| from UK Building Regulations documents. | |
| """ | |
| alternatives = self.llm.generate(prompt) | |
| return parse_alternatives(alternatives) | |
| ``` | |
| **Contradictory Guidance**: | |
| ```python | |
| class ContradictionResolver: | |
| """ | |
| Handle cases where retrieved regulations appear contradictory. | |
| """ | |
| def resolve( | |
| self, | |
| regulations: List[Regulation] | |
| ) -> Resolution: | |
| """ | |
| Identify and resolve contradictions. | |
| """ | |
| # Check for obvious contradictions | |
| contradictions = self._find_contradictions(regulations) | |
| if not contradictions: | |
| return NoContradiction(regulations) | |
| # Attempt resolution by checking: | |
| # 1. Document hierarchy (later supersedes earlier) | |
| # 2. Specificity (specific overrides general) | |
| # 3. Context applicability | |
| resolved = [] | |
| for contradiction in contradictions: | |
| resolution = self._resolve_single_contradiction(contradiction) | |
| resolved.append(resolution) | |
| if all(r.confident for r in resolved): | |
| return ResolvedContradiction( | |
| regulations=resolved, | |
| explanations=[r.explanation for r in resolved] | |
| ) | |
| # Cannot confidently resolve - escalate | |
| return UnresolvableContradiction( | |
| regulations=regulations, | |
| contradictions=contradictions, | |
| action="escalate_to_expert", | |
| user_message="We've identified potentially conflicting requirements. " | |
| "Expert interpretation needed." | |
| ) | |
| ``` | |
| #### **Agent-Level Errors** | |
| **Confidence Below Threshold**: | |
| ```python | |
| class ConfidenceManager: | |
| """ | |
| Manage low-confidence agent determinations. | |
| """ | |
| CONFIDENCE_THRESHOLDS = { | |
| 'critical_safety': 0.90, | |
| 'high_priority': 0.80, | |
| 'medium_priority': 0.70, | |
| 'low_priority': 0.60 | |
| } | |
| def handle( | |
| self, | |
| check: ComplianceCheck, | |
| threshold_category: str | |
| ) -> HandlingDecision: | |
| """ | |
| Decide how to handle low-confidence findings. | |
| """ | |
| required_confidence = self.CONFIDENCE_THRESHOLDS[threshold_category] | |
| if check.confidence >= required_confidence: | |
| # Acceptable confidence | |
| return Accept(check) | |
| elif check.confidence >= required_confidence - 0.10: | |
| # Slightly below threshold - accept with strong caveats | |
| return AcceptWithReview( | |
| check=check, | |
| warnings=[ | |
| f"Confidence ({check.confidence:.0%}) below preferred threshold", | |
| "Recommend expert verification" | |
| ], | |
| review_priority="high" if threshold_category == "critical_safety" else "medium" | |
| ) | |
| else: | |
| # Significantly below threshold - cannot make determination | |
| return InsufficientConfidence( | |
| check=check, | |
| status="requires_expert_review", | |
| reason=f"Confidence {check.confidence:.0%} insufficient for {threshold_category} finding", | |
| required_action="expert_review_before_proceeding" | |
| ) | |
| ``` | |
| **Inter-Agent Conflicts**: | |
| ```python | |
| class ConflictResolver: | |
| """ | |
| Resolve conflicts between different agent findings. | |
| """ | |
| def detect_conflicts( | |
| self, | |
| agent_results: Dict[str, List[ComplianceCheck]] | |
| ) -> List[Conflict]: | |
| """ | |
| Identify conflicts between agent findings. | |
| """ | |
| conflicts = [] | |
| # Check for direct contradictions | |
| # (e.g., Part L wants more insulation, Part O wants less to avoid overheating) | |
| for part_a, checks_a in agent_results.items(): | |
| for part_b, checks_b in agent_results.items(): | |
| if part_a >= part_b: | |
| continue # Avoid duplicate comparisons | |
| part_conflicts = self._find_conflicts_between_parts( | |
| part_a, checks_a, | |
| part_b, checks_b | |
| ) | |
| conflicts.extend(part_conflicts) | |
| return conflicts | |
| def resolve(self, conflict: Conflict) -> Resolution: | |
| """ | |
| Attempt to resolve identified conflict. | |
| """ | |
| # Check if there's established precedence | |
| precedence = self._check_precedence(conflict.part_a, conflict.part_b) | |
| if precedence: | |
| return PrecedenceResolution( | |
| winner=precedence.primary_part, | |
| rationale=precedence.explanation, | |
| recommendation=f"Prioritize {precedence.primary_part} requirement. " | |
| f"Seek alternative solutions for {precedence.secondary_part}." | |
| ) | |
| # Check for alternative solutions that satisfy both | |
| alternatives = self._find_alternative_solutions(conflict) | |
| if alternatives: | |
| return AlternativeSolution( | |
| alternatives=alternatives, | |
| recommendation="Consider alternative approaches that satisfy both requirements" | |
| ) | |
| # Cannot resolve - escalate | |
| return UnresolvedConflict( | |
| conflict=conflict, | |
| action="expert_judgment_required", | |
| recommendation="This conflict requires design team discussion and expert judgment" | |
| ) | |
| ``` | |
| #### **Database Errors** | |
| **Vector Database Unavailable**: | |
| ```python | |
| class VectorDBFailureHandler: | |
| """ | |
| Handle Milvus vector database failures. | |
| """ | |
| def handle(self, error: Exception) -> Result: | |
| """ | |
| Respond to vector DB unavailability. | |
| """ | |
| logger.critical(f"Vector database unavailable: {error}") | |
| # Attempt reconnection | |
| if self._reconnect_attempt(): | |
| return Recovered() | |
| # Use cached results if available | |
| if self.cache.has_recent_data(): | |
| return DegradedMode( | |
| message="Using cached regulatory data", | |
| limitations=[ | |
| "Cannot retrieve newest regulations", | |
| "Precedent search unavailable", | |
| "Some contextual information may be missing" | |
| ], | |
| max_degraded_time="30 minutes" | |
| ) | |
| # Cannot proceed | |
| return Failure( | |
| error="Regulatory knowledge base unavailable", | |
| action="system_maintenance_required", | |
| user_message="System temporarily unavailable. Please try again shortly.", | |
| estimated_recovery="15-30 minutes" | |
| ) | |
| ``` | |
| **Graph Database Inconsistencies**: | |
| ```python | |
| class GraphDBValidation: | |
| """ | |
| Detect and handle Neo4j graph inconsistencies. | |
| """ | |
| def validate_spatial_graph( | |
| self, | |
| spatial_data: SpatialData | |
| ) -> ValidationResult: | |
| """ | |
| Check for logical inconsistencies in spatial graph. | |
| """ | |
| issues = [] | |
| # Check for orphaned nodes | |
| orphans = self._find_orphaned_nodes(spatial_data) | |
| if orphans: | |
| issues.append( | |
| Issue( | |
| type="orphaned_elements", | |
| elements=orphans, | |
| severity="medium", | |
| message=f"{len(orphans)} elements not connected to building topology" | |
| ) | |
| ) | |
| # Check for impossible relationships | |
| impossible = self._find_impossible_relationships(spatial_data) | |
| if impossible: | |
| issues.append( | |
| Issue( | |
| type="impossible_topology", | |
| relationships=impossible, | |
| severity="high", | |
| message=f"{len(impossible)} impossible spatial relationships detected" | |
| ) | |
| ) | |
| # Check for duplicate elements | |
| duplicates = self._find_duplicates(spatial_data) | |
| if duplicates: | |
| issues.append( | |
| Issue( | |
| type="duplicate_elements", | |
| elements=duplicates, | |
| severity="low", | |
| message=f"{len(duplicates)} duplicate elements found" | |
| ) | |
| ) | |
| if issues: | |
| return ValidationFailure( | |
| issues=issues, | |
| action="repair_or_notify", | |
| can_proceed=all(i.severity != "high" for i in issues) | |
| ) | |
| return ValidationSuccess() | |
| def repair(self, issues: List[Issue]) -> RepairResult: | |
| """ | |
| Attempt automatic repair of graph inconsistencies. | |
| """ | |
| repaired = [] | |
| failed = [] | |
| for issue in issues: | |
| try: | |
| if issue.type == "orphaned_elements": | |
| self._repair_orphans(issue.elements) | |
| repaired.append(issue) | |
| elif issue.type == "duplicate_elements": | |
| self._remove_duplicates(issue.elements) | |
| repaired.append(issue) | |
| else: | |
| # Cannot auto-repair | |
| failed.append(issue) | |
| except Exception as e: | |
| logger.error(f"Repair failed for {issue.type}: {e}") | |
| failed.append(issue) | |
| return RepairResult( | |
| repaired=repaired, | |
| failed=failed, | |
| success=len(failed) == 0 | |
| ) | |
| ``` | |
| ### **6.2 Quality Assurance Mechanisms** | |
| #### **Multi-Level Validation** | |
| ```python | |
| class QualityAssurance: | |
| """ | |
| Multi-level quality assurance for compliance determinations. | |
| """ | |
| def validate_finding( | |
| self, | |
| check: ComplianceCheck | |
| ) -> QAResult: | |
| """ | |
| Apply multiple validation checks to finding. | |
| """ | |
| validations = [] | |
| # 1. Evidence consistency check | |
| evidence_valid = self._validate_evidence(check) | |
| validations.append(('evidence', evidence_valid)) | |
| # 2. Reasoning coherence check | |
| reasoning_valid = self._validate_reasoning(check) | |
| validations.append(('reasoning', reasoning_valid)) | |
| # 3. Regulatory citation accuracy | |
| citations_valid = self._validate_citations(check) | |
| validations.append(('citations', citations_valid)) | |
| # 4. Confidence calibration | |
| confidence_calibrated = self._calibrate_confidence(check) | |
| validations.append(('confidence', confidence_calibrated)) | |
| # 5. Cross-check with similar cases | |
| precedent_consistent = self._check_precedent_consistency(check) | |
| validations.append(('precedent', precedent_consistent)) | |
| # Aggregate results | |
| all_valid = all(valid for _, valid in validations) | |
| if all_valid: | |
| return QAPass(check) | |
| else: | |
| failed = [name for name, valid in validations if not valid] | |
| return QAFail( | |
| check=check, | |
| failed_validations=failed, | |
| action="review_and_revise" | |
| ) | |
| def _validate_evidence(self, check: ComplianceCheck) -> bool: | |
| """ | |
| Check that evidence actually supports conclusion. | |
| """ | |
| # Use LLM to assess if evidence supports conclusion | |
| prompt = f""" | |
| Does the following evidence support the conclusion? | |
| Conclusion: {check.status} - {check.requirement} | |
| Evidence: | |
| {chr(10).join(check.evidence)} | |
| Answer: yes/no with brief explanation | |
| """ | |
| response = self.llm.generate(prompt, temperature=0.1) | |
| return 'yes' in response.lower() | |
| def _calibrate_confidence(self, check: ComplianceCheck) -> bool: | |
| """ | |
| Ensure confidence score is appropriate. | |
| """ | |
| # Factors that should reduce confidence: | |
| reducing_factors = [] | |
| if len(check.evidence) < 2: | |
| reducing_factors.append("insufficient_evidence") | |
| if check.status == "insufficient_info": | |
| # Confidence should be high for "insufficient info" findings | |
| # (high confidence that we don't have enough information) | |
| if check.confidence < 0.80: | |
| reducing_factors.append("confidence_too_low_for_insufficient_info") | |
| if check.regulatory_references == []: | |
| reducing_factors.append("no_regulatory_references") | |
| if reducing_factors and check.confidence > 0.70: | |
| # Confidence should be reduced | |
| check.confidence = max(0.60, check.confidence - 0.15) | |
| check.metadata['confidence_adjusted'] = True | |
| check.metadata['adjustment_reason'] = reducing_factors | |
| return True # Calibration applied | |
| ``` | |
| #### **Audit Trail Generation** | |
| ```python | |
| class AuditTrail: | |
| """ | |
| Comprehensive audit trail for all system decisions. | |
| """ | |
| def log_decision( | |
| self, | |
| decision_type: str, | |
| input_data: Dict, | |
| output: Any, | |
| agent: str, | |
| reasoning: str, | |
| confidence: float, | |
| timestamp: datetime | |
| ): | |
| """ | |
| Log a compliance decision with full context. | |
| """ | |
| entry = AuditEntry( | |
| id=generate_uuid(), | |
| timestamp=timestamp, | |
| decision_type=decision_type, | |
| agent=agent, | |
| input_hash=hash_data(input_data), | |
| input_summary=summarize(input_data), | |
| output=output, | |
| reasoning=reasoning, | |
| confidence=confidence, | |
| model_version=self.model_version, | |
| rag_version=self.rag_version, | |
| system_version=self.system_version | |
| ) | |
| # Store in database | |
| self.db.store(entry) | |
| # Create human-readable log | |
| self.log_file.write(self._format_entry(entry)) | |
| def _format_entry(self, entry: AuditEntry) -> str: | |
| """ | |
| Format audit entry for human review. | |
| """ | |
| return f""" | |
| ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| AUDIT ENTRY: {entry.id} | |
| ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| Timestamp: {entry.timestamp} | |
| Agent: {entry.agent} | |
| Decision Type: {entry.decision_type} | |
| Input Summary: | |
| {entry.input_summary} | |
| Output: | |
| {entry.output} | |
| Reasoning: | |
| {entry.reasoning} | |
| Confidence: {entry.confidence:.0%} | |
| System Versions: | |
| - Model: {entry.model_version} | |
| - RAG: {entry.rag_version} | |
| - System: {entry.system_version} | |
| ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| """ | |
| def generate_audit_report( | |
| self, | |
| submission_id: str | |
| ) -> AuditReport: | |
| """ | |
| Generate comprehensive audit report for submission. | |
| """ | |
| entries = self.db.get_entries(submission_id=submission_id) | |
| report = AuditReport( | |
| submission_id=submission_id, | |
| total_decisions=len(entries), | |
| agents_involved=list(set(e.agent for e in entries)), | |
| decision_timeline=self._create_timeline(entries), | |
| confidence_distribution=self._analyze_confidence(entries), | |
| model_versions_used=list(set(e.model_version for e in entries)), | |
| full_trail=entries | |
| ) | |
| return report | |
| ``` | |
| #### **Continuous Learning and Improvement** | |
| ```python | |
| class FeedbackLoop: | |
| """ | |
| Collect feedback and improve system over time. | |
| """ | |
| def record_outcome( | |
| self, | |
| submission_id: str, | |
| ai_assessment: ComplianceReport, | |
| actual_outcome: Outcome | |
| ): | |
| """ | |
| Record actual BSR decision for learning. | |
| """ | |
| # Compare AI assessment with actual outcome | |
| comparison = self._compare_assessment_to_outcome( | |
| ai_assessment, | |
| actual_outcome | |
| ) | |
| # Store for analysis | |
| self.db.store_outcome( | |
| submission_id=submission_id, | |
| ai_assessment=ai_assessment, | |
| actual_outcome=actual_outcome, | |
| comparison=comparison | |
| ) | |
| # Update performance metrics | |
| self._update_metrics(comparison) | |
| # Flag cases for review | |
| if comparison.significant_discrepancy: | |
| self._flag_for_review(submission_id, comparison) | |
| def analyze_performance(self) -> PerformanceReport: | |
| """ | |
| Analyze system performance over time. | |
| """ | |
| outcomes = self.db.get_all_outcomes() | |
| metrics = { | |
| 'overall_accuracy': self._calculate_accuracy(outcomes), | |
| 'accuracy_by_part': self._calculate_accuracy_by_part(outcomes), | |
| 'false_positive_rate': self._calculate_false_positive_rate(outcomes), | |
| 'false_negative_rate': self._calculate_false_negative_rate(outcomes), | |
| 'confidence_calibration': self._assess_confidence_calibration(outcomes) | |
| } | |
| # Identify patterns in errors | |
| error_patterns = self._identify_error_patterns(outcomes) | |
| # Generate recommendations | |
| recommendations = self._generate_improvement_recommendations( | |
| metrics, | |
| error_patterns | |
| ) | |
| return PerformanceReport( | |
| metrics=metrics, | |
| error_patterns=error_patterns, | |
| recommendations=recommendations | |
| ) | |
| def trigger_retraining(self, criteria: RetrainingCriteria) -> bool: | |
| """ | |
| Determine if model retraining is needed. | |
| """ | |
| performance = self.analyze_performance() | |
| should_retrain = ( | |
| performance.accuracy < criteria.min_accuracy or | |
| performance.false_negative_rate > criteria.max_false_negative or | |
| performance.significant_error_patterns | |
| ) | |
| if should_retrain: | |
| logger.info("Retraining triggered based on performance analysis") | |
| self._initiate_retraining(performance) | |
| return should_retrain | |
| ``` | |
| --- | |
| ## **7. PROFESSIONAL STANDARDS COMPLIANCE** | |
| ### **7.1 RBI/Chartered Building Engineer Code Alignment** | |
| **Key Principle**: System provides compliance **commentary**, not design **advice** | |
| #### **Distinguishing Commentary from Advice** | |
| **Commentary (Permitted)**: | |
| - "The submitted design does not meet Part B1 travel distance requirements (45m max, 60m provided)" | |
| - "Approved Document B Section 2.5 requires alternative means of escape for this occupancy" | |
| - "The specification lacks fire resistance ratings for the separating wall" | |
| - "Similar configurations have been rejected at Gateway 2 for inadequate compartmentation" | |
| **Design Advice (Not Provided)**: | |
| - ✗ "You should relocate the stairway to the northwest corner" | |
| - ✗ "We recommend using 2-hour fire-rated construction" | |
| - ✗ "The optimal approach is to reduce the corridor length" | |
| - ✗ "Install an additional exit here [specific location]" | |
| **Implementation in System**: | |
| ```python | |
| class ProfessionalStandardsValidator: | |
| """ | |
| Ensure all outputs comply with professional standards. | |
| """ | |
| PROHIBITED_PHRASES = [ | |
| "you should", | |
| "we recommend", | |
| "you must", | |
| "the best approach", | |
| "optimal design", | |
| "ideal solution" | |
| ] | |
| def validate_output( | |
| self, | |
| compliance_check: ComplianceCheck | |
| ) -> ValidationResult: | |
| """ | |
| Ensure output is commentary, not advice. | |
| """ | |
| # Check reasoning and recommendations | |
| text_to_check = ( | |
| compliance_check.reasoning + " " + | |
| " ".join(compliance_check.recommendations) | |
| ).lower() | |
| # Flag prohibited phrases | |
| violations = [ | |
| phrase for phrase in self.PROHIBITED_PHRASES | |
| if phrase in text_to_check | |
| ] | |
| if violations: | |
| return ValidationFailure( | |
| violations=violations, | |
| suggestion="Rephrase to describe requirements rather than prescribe solutions" | |
| ) | |
| # Check for imperative language | |
| if self._contains_imperatives(text_to_check): | |
| return ValidationFailure( | |
| issue="imperative_language", | |
| suggestion="Use 'requirements state' rather than 'you must'" | |
| ) | |
| # Ensure regulatory basis cited | |
| if not compliance_check.regulatory_references: | |
| return ValidationFailure( | |
| issue="no_regulatory_basis", | |
| suggestion="Include regulatory references for all findings" | |
| ) | |
| return ValidationSuccess() | |
| def reframe_as_commentary( | |
| self, | |
| advice_text: str | |
| ) -> str: | |
| """ | |
| Convert advice-style text to commentary-style. | |
| """ | |
| # Use LLM to reframe | |
| prompt = f""" | |
| Reframe the following as regulatory commentary rather than design advice. | |
| Focus on: | |
| - What the regulations require | |
| - Where the submission falls short | |
| - What information is needed | |
| Avoid: | |
| - Prescriptive recommendations | |
| - Specific design solutions | |
| - Imperative language | |
| Original: {advice_text} | |
| Reframed commentary: | |
| """ | |
| reframed = self.llm.generate(prompt, temperature=0.1) | |
| return reframed | |
| ``` | |
| #### **Limitation Disclaimers** | |
| All outputs include clear disclaimers: | |
| ``` | |
| IMPORTANT DISCLAIMER: | |
| This compliance commentary is provided as a guidance tool to assist in | |
| preparing Building Regulations submissions. It does NOT constitute: | |
| - Design advice or recommendations | |
| - Professional engineering or architectural services | |
| - Final determination of compliance (only Building Safety Regulator can make final determinations) | |
| - A substitute for qualified professional review | |
| All findings should be reviewed by appropriately qualified professionals | |
| (Chartered Engineers, Registered Building Inspectors, etc.) before relying | |
| on them for submission or construction purposes. | |
| This system identifies potential compliance issues based on submitted | |
| documentation, but cannot account for all project-specific factors or | |
| replace expert judgment. | |
| The user retains full responsibility for ensuring submissions meet all | |
| applicable regulations and standards. | |
| ``` | |
| ### **7.2 Liability and Professional Responsibility** | |
| **System Positioning**: | |
| - **Decision Support Tool**: Augments human expertise, doesn't replace it | |
| - **Early Warning System**: Identifies potential issues for expert review | |
| - **Documentation Aid**: Helps organize and structure compliance evidence | |
| **User Responsibility**: | |
| - **Verification**: All AI findings must be verified by qualified professionals | |
| - **Interpretation**: Regulations require professional judgment in application | |
| - **Final Decisions**: Human experts make all final compliance determinations | |
| - **Documentation**: Users responsible for accuracy of submitted information | |
| **Recommended Workflow**: | |
| ``` | |
| 1. Submit documents to BuildwellTHREAD | |
| 2. Receive AI compliance commentary | |
| 3. Review findings with qualified building engineer/RBI | |
| 4. Conduct additional analysis as needed | |
| 5. Make professional determination | |
| 6. Prepare final Gateway submission | |
| 7. Maintain audit trail of all review steps | |
| ``` | |
| --- | |
| ## **8. DEPLOYMENT AND OPERATIONS** | |
| ### **8.1 Deployment on Isambard AI** | |
| #### **Resource Allocation** | |
| ```yaml | |
| Isambard AI Configuration: | |
| Platform: Nvidia GH200 Grace-Hopper Superchips | |
| Operating System: Ubuntu 22.04 LTS | |
| Container Platform: Docker or Singularity | |
| Compute Resources (Example Allocation): | |
| Development/Training: | |
| - GPUs: 8x GH200 (480GB total GPU memory) | |
| - CPUs: 64 ARM cores | |
| - RAM: 1TB system memory | |
| - Storage: 10TB NVMe | |
| Production Inference: | |
| - GPUs: 4x GH200 (240GB total GPU memory) | |
| - CPUs: 32 ARM cores | |
| - RAM: 512GB system memory | |
| - Storage: 5TB NVMe | |
| Network: | |
| - High-speed interconnect between nodes | |
| - External connectivity for document uploads/downloads | |
| - Firewall rules for security | |
| ``` | |
| #### **Software Stack** | |
| ```dockerfile | |
| # Base Image | |
| FROM nvidia/cuda:12.2.0-runtime-ubuntu22.04 | |
| # System dependencies | |
| RUN apt-get update && apt-get install -y \ | |
| python3.11 \ | |
| python3-pip \ | |
| git \ | |
| wget \ | |
| wine64 \ # For FloorplanTransformation | |
| && rm -rf /var/lib/apt/lists/* | |
| # Python dependencies | |
| COPY requirements.txt /tmp/ | |
| RUN pip3 install -r /tmp/requirements.txt | |
| # Key packages | |
| # - transformers (for LLMs) | |
| # - pymilvus (vector DB client) | |
| # - neo4j (graph DB client) | |
| # - fastapi (API server) | |
| # - celery (task queue) | |
| # - pytorch (deep learning framework) | |
| # Application code | |
| COPY buildwellthread /app/buildwellthread | |
| WORKDIR /app | |
| # FloorplanTransformation setup | |
| COPY FloorplanTransformation /app/floorplan_transform | |
| ENV WINEPREFIX=/app/.wine | |
| # Entrypoint | |
| CMD ["uvicorn", "buildwellthread.api:app", "--host", "0.0.0.0", "--port", "8000"] | |
| ``` | |
| ### **8.2 Operational Procedures** | |
| #### **Submission Processing Workflow** | |
| ```python | |
| # Orchestration with task queue | |
| from celery import Celery, chain | |
| app = Celery('buildwellthread') | |
| @app.task | |
| def ingest_documents(submission_id: str): | |
| """ | |
| Task 1: Ingest and parse uploaded documents. | |
| """ | |
| submission = load_submission(submission_id) | |
| parsed = document_ingestion.process(submission.files) | |
| save_parsed_documents(submission_id, parsed) | |
| return submission_id | |
| @app.task | |
| def extract_spatial_data(submission_id: str): | |
| """ | |
| Task 2: Process floorplans into spatial data. | |
| """ | |
| parsed = load_parsed_documents(submission_id) | |
| spatial_data = spatial_processor.process(parsed.drawings) | |
| save_spatial_data(submission_id, spatial_data) | |
| return submission_id | |
| @app.task | |
| def build_knowledge_context(submission_id: str): | |
| """ | |
| Task 3: Retrieve relevant regulations from RAG. | |
| """ | |
| submission = load_submission(submission_id) | |
| spatial_data = load_spatial_data(submission_id) | |
| context = rag_system.build_context(submission, spatial_data) | |
| save_context(submission_id, context) | |
| return submission_id | |
| @app.task | |
| def run_compliance_agents(submission_id: str): | |
| """ | |
| Task 4: Execute multi-agent compliance checking. | |
| """ | |
| submission = load_submission(submission_id) | |
| spatial_data = load_spatial_data(submission_id) | |
| context = load_context(submission_id) | |
| agent_results = orchestrator.process(submission, spatial_data, context) | |
| save_agent_results(submission_id, agent_results) | |
| return submission_id | |
| @app.task | |
| def generate_outputs(submission_id: str): | |
| """ | |
| Task 5: Generate reports and Golden Thread docs. | |
| """ | |
| agent_results = load_agent_results(submission_id) | |
| compliance_report = report_generator.generate(agent_results) | |
| golden_thread = golden_thread_generator.generate(agent_results) | |
| save_outputs(submission_id, compliance_report, golden_thread) | |
| # Notify user | |
| notify_user(submission_id, "complete") | |
| return submission_id | |
| # Chain tasks together | |
| def process_submission(submission_id: str): | |
| """ | |
| Orchestrate full processing pipeline. | |
| """ | |
| workflow = chain( | |
| ingest_documents.s(submission_id), | |
| extract_spatial_data.s(), | |
| build_knowledge_context.s(), | |
| run_compliance_agents.s(), | |
| generate_outputs.s() | |
| ) | |
| result = workflow.apply_async() | |
| return result | |
| ``` | |
| #### **Monitoring and Alerting** | |
| ```python | |
| class SystemMonitor: | |
| """ | |
| Monitor system health and performance. | |
| """ | |
| def collect_metrics(self): | |
| """ | |
| Collect system metrics for monitoring. | |
| """ | |
| return { | |
| 'active_submissions': self.get_active_submission_count(), | |
| 'queue_depth': self.get_queue_depth(), | |
| 'processing_time_p50': self.get_processing_percentile(0.50), | |
| 'processing_time_p95': self.get_processing_percentile(0.95), | |
| 'error_rate': self.get_error_rate(), | |
| 'gpu_utilization': self.get_gpu_utilization(), | |
| 'memory_usage': self.get_memory_usage(), | |
| 'storage_usage': self.get_storage_usage(), | |
| 'database_latency': self.get_database_latency(), | |
| 'confidence_distribution': self.get_confidence_distribution() | |
| } | |
| def check_health(self) -> HealthStatus: | |
| """ | |
| Perform health check on all components. | |
| """ | |
| checks = { | |
| 'api': self.ping_api(), | |
| 'task_queue': self.check_celery(), | |
| 'vector_db': self.ping_milvus(), | |
| 'graph_db': self.ping_neo4j(), | |
| 'relational_db': self.ping_postgres(), | |
| 'storage': self.check_storage_availability(), | |
| 'model_server': self.check_model_availability() | |
| } | |
| all_healthy = all(checks.values()) | |
| return HealthStatus( | |
| healthy=all_healthy, | |
| component_status=checks, | |
| timestamp=datetime.now() | |
| ) | |
| def alert_if_needed(self, metrics: Dict): | |
| """ | |
| Send alerts for concerning metrics. | |
| """ | |
| # High error rate | |
| if metrics['error_rate'] > 0.10: # 10% errors | |
| self.send_alert( | |
| severity='HIGH', | |
| message=f"Error rate elevated: {metrics['error_rate']:.1%}" | |
| ) | |
| # Long processing times | |
| if metrics['processing_time_p95'] > 7200: # 2 hours | |
| self.send_alert( | |
| severity='MEDIUM', | |
| message=f"P95 processing time: {metrics['processing_time_p95']/60:.0f} minutes" | |
| ) | |
| # Queue backlog | |
| if metrics['queue_depth'] > 100: | |
| self.send_alert( | |
| severity='MEDIUM', | |
| message=f"Queue depth: {metrics['queue_depth']} submissions" | |
| ) | |
| # Storage space | |
| if metrics['storage_usage'] > 0.85: # 85% full | |
| self.send_alert( | |
| severity='HIGH', | |
| message=f"Storage {metrics['storage_usage']:.0%} full" | |
| ) | |
| ``` | |
| --- | |
| ## **9. CONCLUSION AND NEXT STEPS** | |
| ### **9.1 Summary** | |
| BuildwellTHREAD represents a comprehensive solution for automating UK Building Regulations compliance checking and Golden Thread management. The system addresses critical challenges in the current Gateway submission process: | |
| **Problems Solved**: | |
| - High Gateway 2/3 rejection rates (40-67% → target <20%) | |
| - Lengthy manual review processes (days → hours) | |
| - Inconsistent compliance checking across projects | |
| - Difficulty maintaining Golden Thread documentation | |
| - Limited early detection of compliance issues | |
| **Core Capabilities**: | |
| - Multi-agent architecture with specialized compliance checking per Building Regulation Part | |
| - Multimodal document processing (text, drawings, specifications) | |
| - Intelligent spatial reasoning and measurement | |
| - Regulatory knowledge retrieval via RAG system | |
| - Automated Golden Thread generation and maintenance | |
| - Professional standards-compliant commentary (not design advice) | |
| **Technical Foundation**: | |
| - Deployment on Isambard AI supercomputer (Nvidia GH200) | |
| - Sophisticated error handling and quality assurance | |
| - Comprehensive audit trails for all decisions | |
| - Continuous learning and improvement mechanisms | |
| ### **9.2 Development Roadmap** | |
| **Phase 1: Foundation (Months 1-3)** | |
| - Set up infrastructure on Isambard AI | |
| - Develop document ingestion pipeline | |
| - Implement FloorplanTransformation integration | |
| - Build core RAG system | |
| - Establish databases (Milvus, Neo4j, PostgreSQL) | |
| **Phase 2: Core Agents (Months 4-6)** | |
| - Implement Part A (Structure) agent | |
| - Implement Part B (Fire Safety) agent | |
| - Implement Part M (Access) agent | |
| - Develop orchestrator framework | |
| - Build tool library (measurement, spatial reasoning) | |
| **Phase 3: Comprehensive Coverage (Months 7-9)** | |
| - Implement remaining Part agents (C-L, N-T) | |
| - Develop cross-part validation | |
| - Build Golden Thread generator | |
| - Create reporting system | |
| **Phase 4: Testing and Validation (Months 10-12)** | |
| - Validation against expert human reviews | |
| - Testing with real Gateway submissions | |
| - Performance optimization | |
| - User interface development | |
| - Documentation and training materials | |
| **Phase 5: Deployment and Operations (Month 12+)** | |
| - Production deployment | |
| - User onboarding | |
| - Continuous monitoring and improvement | |
| - Regular model updates based on outcomes | |
| ### **9.3 Success Factors** | |
| **Critical for Success**: | |
| 1. **High-Quality Training Data**: Access to real Gateway submissions and BSR decisions | |
| 2. **Expert Collaboration**: Ongoing input from building control professionals | |
| 3. **Iterative Refinement**: Continuous improvement based on real-world performance | |
| 4. **Professional Standards**: Strict adherence to RBI/Chartered Engineer codes | |
| 5. **Computational Resources**: Reliable access to Isambard AI or equivalent infrastructure | |
| **Risk Mitigation**: | |
| 1. **Technical Risks**: Fallback strategies for all critical dependencies | |
| 2. **Regulatory Risks**: Clear positioning as decision support, not replacement for professionals | |
| 3. **Operational Risks**: Comprehensive monitoring and alert systems | |
| 4. **Quality Risks**: Multi-level validation and audit trails | |
| ### **9.4 Expected Impact** | |
| **For Design Teams**: | |
| - Faster identification of compliance issues (days → hours) | |
| - Clearer understanding of regulatory requirements | |
| - Higher quality submissions with fewer iterations | |
| - Reduced risk of costly late-stage redesigns | |
| **For Building Control Bodies**: | |
| - More consistent compliance checking | |
| - Reduced review burden for complete submissions | |
| - Better-documented decision trails | |
| - Improved communication with applicants | |
| **For BSR**: | |
| - Higher quality Gateway submissions | |
| - Reduced rejection rates | |
| - Better-maintained Golden Thread documentation | |
| - More efficient review process | |
| **For Industry**: | |
| - Accelerated project timelines | |
| - Reduced compliance-related costs | |
| - Improved building safety outcomes | |
| - Better regulatory adherence | |
| --- | |
| ## **DOCUMENT CONTROL** | |
| **Version**: 1.0 | |
| **Date**: January 2026 | |
| **Author**: BuildwellTHREAD Development Team | |
| **Status**: Development Specification | |
| **Distribution**: | |
| - Isambard AI Development Team | |
| - London Belgravia Surveyors | |
| - Building Safety Regulator (for consultation) | |
| - RBI/Chartered Building Engineers (for professional standards review) | |
| **Revision History**: | |
| - v1.0 (Jan 2026): Initial comprehensive specification | |
| **Contact**: | |
| For questions or clarifications, please contact the BuildwellTHREAD project team. | |
| --- | |
| ## **APPENDICES** | |
| ### **Appendix A: Building Regulation Parts Reference** | |
| Quick reference for all UK Building Regulations Parts covered: | |
| - **Part A**: Structure | |
| - **Part B**: Fire Safety | |
| - **Part C**: Site Preparation and Resistance to Contaminants and Moisture | |
| - **Part D**: Toxic Substances | |
| - **Part E**: Resistance to the Passage of Sound | |
| - **Part F**: Ventilation | |
| - **Part G**: Sanitation, Hot Water Safety and Water Efficiency | |
| - **Part H**: Drainage and Waste Disposal | |
| - **Part J**: Combustion Appliances and Fuel Storage | |
| - **Part K**: Protection from Falling, Collision and Impact | |
| - **Part L**: Conservation of Fuel and Power | |
| - **Part M**: Access to and Use of Buildings | |
| - **Part N**: Glazing (Safety in Relation to Impact, Opening and Cleaning) | |
| - **Part O**: Overheating | |
| - **Part P**: Electrical Safety (Dwellings) | |
| - **Part Q**: Security (Dwellings) | |
| - **Part R**: Physical Infrastructure for High-Speed Electronic Communications Networks | |
| - **Part S**: Infrastructure for Charging of Electric Vehicles | |
| - **Part T**: Infrastructure for Broadband | |
| ### **Appendix B: Glossary** | |
| **BSR**: Building Safety Regulator | |
| **HRB**: Higher-Risk Building (≥18m or ≥7 storeys) | |
| **Gateway 2**: BSR gateway for final design approval before construction | |
| **Gateway 3**: BSR gateway for completion certificate | |
| **Golden Thread**: Structured set of information about building throughout lifecycle | |
| **RAG**: Retrieval-Augmented Generation | |
| **RBI**: Registered Building Inspector | |
| **BBA**: British Board of Agrément | |
| **KIWA**: International certification body | |
| ### **Appendix C: Technical Specifications** | |
| **Model Requirements**: | |
| - Base LLM: 7B-72B parameters (depends on deployment resources) | |
| - Embedding Model: NV-Embed-v2 or equivalent (768-dimensional) | |
| - Vision Model: CLIP or similar (if needed for diagram comparison) | |
| **Database Requirements**: | |
| - Vector DB: Milvus 2.x+ with HNSW indexing | |
| - Graph DB: Neo4j 5.x+ with GDS (Graph Data Science) library | |
| - Relational DB: PostgreSQL 14+ with full-text search capabilities | |
| **Storage Requirements**: | |
| - Regulatory corpus: ~50GB (Approved Documents, Standards, Guidance) | |
| - Submission documents: ~100GB per 1000 submissions | |
| - Processed spatial data: ~50GB per 1000 submissions | |
| - System logs and audit trails: ~10GB per month | |
| **Compute Requirements**: | |
| - Development: 8x high-end GPUs with 480GB+ total memory | |
| - Production: 4x GPUs with 240GB+ total memory for concurrent processing | |
| - CPU: 32-64 cores for parallel processing tasks | |
| --- | |
| **END OF DOCUMENT** | |