diff --git "a/BuildwellTHREAD_Complete_Specification.md" "b/BuildwellTHREAD_Complete_Specification.md" new file mode 100644--- /dev/null +++ "b/BuildwellTHREAD_Complete_Specification.md" @@ -0,0 +1,3960 @@ +# **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**