# Survey Report Generator — Complete System Redesign v2 ## Cursor Implementation Prompt — Template-Agnostic Edition --- ## PARADIGM SHIFT FROM v1 The v1 system hardcoded RICS section codes, condition rating colours, section ordering, and structural elements as universal facts. It would break entirely for any other firm's template and added document elements (ratings, badges, footers) that the actual template may not contain. **v2 inverts the model entirely:** > The uploaded template is the sole authority for every structural decision. > The system discovers — never assumes — sections, ordering, rating systems, > placeholder syntax, and formatting conventions from whatever document the > firm uploads. If the template does not contain a structural element, > the system does not produce it. The LLM has two permitted actions only: 1. Fill observed facts into retrieved template paragraphs (the mapping step) 2. Decide whether an observation matches a retrieved paragraph (the grounding step) The LLM is NEVER permitted to: - Generate report prose from scratch - Add structural elements not present in the template (ratings, scores, labels, badges) - Invent defects, measurements, materials, or recommendations not in the notes - Reorder, rename, or reformat sections beyond what the template defines --- ## OVERVIEW & MISSION You are building a professional survey report generation system for any property surveying firm, regardless of which template format or professional body they follow. The system does the following and only the following: 1. **Schema Discovery**: On first upload, read the firm's master template and extract a precise schema — sections, ordering, rating systems (if any), placeholder syntax, and structural hierarchy. Store this as the tenant's authoritative schema. 2. **PII Scrubbing**: Strip all personally identifiable information from raw surveyor notes before any data leaves the local system or reaches any AI call. 3. **Notes Parsing**: Using the discovered schema as the only section taxonomy, assign each observation to the correct section. 4. **RAG Retrieval**: Retrieve the most semantically relevant paragraph(s) from the firm's own uploaded template library for each section. 5. **Mapping**: The LLM fills the surveyor's facts into the retrieved template paragraph — nothing more, nothing less. Any structural elements in the output (headings, rating fields, etc.) must originate from the template paragraph itself, not from assumptions. 6. **Grounding Check**: A second LLM call verifies every claim in the output traces to either the surveyor's observations or the retrieved paragraph. 7. **Assembly**: Build the final DOCX using the template's own structure and ordering, not a hardcoded section list. --- ## ARCHITECTURE ### Tech Stack - **Backend**: Python (FastAPI) - **Vector Store**: FAISS (local, per-tenant isolation) - **Embeddings**: `text-embedding-3-small` (OpenAI) — swappable via config - **LLM**: Anthropic Claude claude-sonnet-4-20250514 via `anthropic` Python SDK - **Document Output**: `python-docx` - **PII Scrubber**: Custom regex + spaCy transformer NER pipeline - **Schema Store**: JSON per tenant (discovered from uploaded template) - **Auth**: JWT, per-tenant data isolation ### File Structure ``` survey-report-system/ ├── backend/ │ ├── main.py # FastAPI app entrypoint │ ├── config.py # Settings (env vars, thresholds, feature flags) │ ├── requirements.txt │ │ │ ├── api/ │ │ ├── routes/ │ │ │ ├── upload.py # Template ingestion + schema discovery trigger │ │ │ ├── report.py # Report generation endpoints │ │ │ ├── schema.py # Schema inspection / update endpoints │ │ │ └── auth.py # JWT auth │ │ │ ├── core/ │ │ ├── pii_scrubber.py # PII detection + redaction (two-pass) │ │ ├── template_discoverer.py # NEW: reads uploaded template → TemplateSchema │ │ ├── notes_parser.py # Dynamic: driven by TemplateSchema, not hardcoded │ │ ├── rag_store.py # Schema-agnostic FAISS (section_id is dynamic) │ │ ├── section_mapper.py # Orchestrator: scrub → parse → retrieve → map → check │ │ ├── report_assembler.py # Template-driven DOCX builder │ │ └── grounding_checker.py # Post-generation audit │ │ │ ├── prompts/ │ │ ├── discovery_prompt.py # NEW: LLM-based schema extraction │ │ ├── mapping_prompt.py # Template-driven mapping (no assumed structure) │ │ └── grounding_prompt.py # Grounding auditor │ │ │ ├── models/ │ │ ├── schema.py # TemplateSchema, SectionDefinition, RatingSystem │ │ ├── section.py # SectionNote (schema-agnostic) │ │ └── report.py # Full report model │ │ │ └── utils/ │ ├── doc_extractor.py # Extract text + heading structure from DOCX/PDF │ ├── docx_builder.py # python-docx helpers │ └── tenant_store.py # Per-tenant FAISS + schema + metadata store │ ├── data/ │ └── tenants/ │ └── {tenant_id}/ │ ├── schema.json # Discovered template schema │ ├── index.faiss # Embeddings │ └── metadata.json # Chunk metadata │ └── docs/ └── LIVE_AI_PROMPTS.md # Auto-generated prompt inventory ``` --- ## MODULE 1: PII SCRUBBER (`core/pii_scrubber.py`) Runs **before any data leaves the local system and before any LLM call**. Two passes: regex (deterministic, fast) then spaCy transformer NER (catches names and locations the regex misses). The scrubber is domain-agnostic — it does not assume RICS or any specific report format. ### What Must Be Redacted | Category | Examples | Replacement Token | |---|---|---| | Full street addresses | "1a Woodland Hill, Lambeth" | `[REDACTED_ADDRESS]` | | Postcodes / ZIP codes | "SE19 1PB", "10001" | `[REDACTED_POSTCODE]` | | Person names | Any proper name of an individual | `[REDACTED_NAME]` | | Company / firm names | Any named organisation | `[REDACTED_COMPANY]` | | Phone numbers | UK, US, international formats | `[REDACTED_PHONE]` | | Email addresses | Any `x@y.z` format | `[REDACTED_EMAIL]` | | Reference / case numbers | Alphanumeric IDs, report numbers | `[REDACTED_REF]` | | Specific individual dates | "1st December 2025" | `[REDACTED_DATE]` | | Currency amounts | "£450,000", "$1.2M" | `[REDACTED_AMOUNT]` | | URLs containing identifiers | Any URL with names or IDs | `[REDACTED_URL]` | ### Implementation ```python # core/pii_scrubber.py import re import spacy from typing import Tuple, List, Dict nlp = spacy.load("en_core_web_trf") # Transformer NER — highest accuracy # Regex patterns — order matters; most specific first REGEX_PATTERNS = [ # UK postcodes (full: "SE19 1PB") — place before generic alphanumeric ID pattern (r'\b[A-Z]{1,2}[0-9][0-9A-Z]?\s*[0-9][A-Z]{2}\b', '[REDACTED_POSTCODE]'), # US ZIP codes ("10001", "90210-1234") (r'\b\d{5}(?:-\d{4})?\b', '[REDACTED_POSTCODE]'), # Email (r'\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b', '[REDACTED_EMAIL]'), # UK phone numbers (07..., +44..., 01..., 02...) (r'\b(?:(?:\+44|0)[\d\s\-\(\)]{9,13})\b', '[REDACTED_PHONE]'), # US phone numbers (r'\b(?:\+1[\s\-]?)?\(?\d{3}\)?[\s\-]?\d{3}[\s\-]?\d{4}\b', '[REDACTED_PHONE]'), # Currency amounts (£, $, €, preceded by digit) (r'[£$€]\s*[\d,]+(?:\.\d{1,2})?(?:\s*(?:million|m|k|bn))?\b', '[REDACTED_AMOUNT]'), # Reference / report / case numbers (alphanumeric with separator, e.g. NCS-100147, REF/2024/001) (r'\b[A-Z]{2,}[\-\/][A-Z0-9][\-\/A-Z0-9]{2,}\b', '[REDACTED_REF]'), # Pure numeric IDs of 6+ digits (RICS member numbers, report IDs, etc.) (r'\b\d{6,}\b', '[REDACTED_REF]'), # URLs (r'https?://[^\s]+', '[REDACTED_URL]'), # Written-out dates tied to individuals ("1st December 2025", "January 15th 2024") ( r'\b\d{1,2}(?:st|nd|rd|th)?\s+(?:January|February|March|April|May|June|' r'July|August|September|October|November|December)\s+\d{4}\b', '[REDACTED_DATE]' ), # Numeric dates (15/12/2025, 15-12-2025, 2025/12/15) (r'\b\d{1,2}[\/\-]\d{1,2}[\/\-]\d{2,4}\b', '[REDACTED_DATE]'), (r'\b\d{4}[\/\-]\d{2}[\/\-]\d{2}\b', '[REDACTED_DATE]'), ] # spaCy entity types that map to PII NER_ENTITY_MAP = { 'PERSON': '[REDACTED_NAME]', 'ORG': '[REDACTED_COMPANY]', 'GPE': '[REDACTED_LOCATION]', # Geopolitical entity 'LOC': '[REDACTED_LOCATION]', # Non-GPE location 'FAC': '[REDACTED_LOCATION]', # Buildings/facilities } def scrub(text: str) -> Tuple[str, List[Dict]]: """ Two-pass PII scrubbing. Pass 1 — Regex: deterministic, fast, catches structured patterns. Pass 2 — spaCy NER: catches names, company names, locations regex misses. Returns: scrubbed_text: PII replaced with tokens redaction_log: List of {type, count} dicts for audit. Never stores original values. """ redaction_log = [] # Pass 1: Regex (most specific patterns first) for pattern, token in REGEX_PATTERNS: matches = re.findall(pattern, text, re.IGNORECASE) if matches: redaction_log.append({'type': token, 'count': len(matches)}) text = re.sub(pattern, token, text, flags=re.IGNORECASE) # Pass 2: spaCy NER (process on regex-scrubbed text to reduce confusion) doc = nlp(text) # Reverse order to preserve character offsets when replacing for ent in reversed(doc.ents): if ent.label_ in NER_ENTITY_MAP: replacement = NER_ENTITY_MAP[ent.label_] redaction_log.append({'type': replacement, 'label': ent.label_}) text = text[:ent.start_char] + replacement + text[ent.end_char:] return text, redaction_log def scrub_rag_chunk(chunk: str) -> str: """ Used during ingest of reference (past-report) documents. Strips all PII so no property's specific data can leak into another report. Master template paragraphs (boilerplate with no real property data) are exempt from this — they are ingested as-is because they are structural templates, not property-specific records. """ scrubbed, _ = scrub(chunk) return scrubbed def assert_no_pii(text: str) -> bool: """ Final belt-and-braces check on any text before it leaves the system. Returns True if no PII patterns remain, False otherwise. Used as a gate in the report assembler before DOCX export. """ for pattern, _ in REGEX_PATTERNS: if re.search(pattern, text, re.IGNORECASE): return False # Run a lightweight NER check doc = nlp(text) for ent in doc.ents: if ent.label_ in NER_ENTITY_MAP: return False return True ``` **Critical rules:** - `scrub()` runs on surveyor notes BEFORE any embedding or LLM call - `scrub_rag_chunk()` runs on reference-tier documents at ingest time, never on master templates - `assert_no_pii()` runs on every mapped paragraph AFTER the LLM generates it, before it enters the DOCX --- ## RAG DOCUMENT INGESTION — PII HANDLING POLICY This section defines how sensitive information is treated when documents are uploaded to become part of the RAG store. It is a distinct concern from the scrubbing of surveyor notes: the risk profile is different and the rules are not the same for both document tiers. ### The Risk This Policy Prevents The RAG store is the memory of the system. When the mapping LLM retrieves a paragraph to fill with observations, it receives that paragraph's text directly in its context. If a retrieved paragraph contains a real client's name, a real property address, or a specific defect description tied to a specific past survey, the LLM may reproduce that information in the new report — either verbatim or by treating it as a known fact about the current property. This is cross-property data leakage: private information from one client's survey contaminating another client's report. It is a data protection violation regardless of whether the LLM does it accidentally or deliberately. The PII handling policy for ingested documents exists entirely to prevent this failure mode. ### The Two-Tier Document Model Every document uploaded to the RAG store is assigned to one of two tiers. The tier determines whether PII scrubbing is applied and how strictly. --- #### TIER_MASTER — Firm's Approved Template Library **What these documents are:** The firm's own master paragraph library: generic, property-agnostic prose written to work for any survey of the relevant type. No real client, no real address, no specific measured defects. These are structural templates, not records of past work. **PII scrubbing at ingest: NOT APPLIED.** The reason is that these documents contain no property-specific data to strip, and scrubbing them would risk corrupting the placeholder syntax and generic professional language the mapping step depends on. Running a spaCy NER pass over a paragraph that contains the word "the owner should" or "the surveyor observed" would risk redacting generic nouns that happen to resemble names. **The guard that replaces scrubbing: ingest-time PII verification.** Before a TIER_MASTER document is chunked and embedded, the upload endpoint runs `assert_no_pii()` across the full document text. If any PII pattern is detected, the ingest is rejected with a clear error message: ``` HTTP 400: Master template document appears to contain personally identifiable information (detected: [REDACTED_POSTCODE] pattern). Master templates must be property-agnostic boilerplate. Remove all real addresses, names, and reference numbers before uploading, or re-upload this document as a reference document instead. ``` The ingest does not proceed until the document passes this check. This is not a warning — it is a hard gate. The firm must either clean the document or reclassify it as a reference document (which will then be scrubbed on ingest). **Metadata flag:** All TIER_MASTER chunks are stored with `is_scrubbed: false`. This accurately reflects that no scrubbing was applied, and it is safe precisely because the ingest-time verification confirmed no PII was present. --- #### TIER_REFERENCE — Past Completed Reports **What these documents are:** Completed survey reports from previous jobs, uploaded to provide the RAG store with examples of professional terminology, sentence rhythm, and the firm's writing style. These documents contain real client data: real names, real addresses, real inspection dates, real reference numbers, real amounts, and real property-specific observations. **PII scrubbing at ingest: MANDATORY AND UNCONDITIONAL.** `scrub_rag_chunk()` is called on every paragraph chunk before it is embedded or stored. This is the same two-pass scrubber used on surveyor notes (regex first, then spaCy NER), applied at chunk level rather than document level so that the embedding captures the technical content after redaction. **What survives scrubbing in a reference document:** | Survives | Removed | |---|---| | Technical defect vocabulary ("spalling brickwork", "failed pointing") | All names (client, surveyor, firm) | | Professional sentence structure and tone | All addresses and postcodes | | Generic recommendations ("further investigation is recommended") | All reference and report numbers | | Building material descriptions ("render", "slate", "uPVC") | All currency amounts | | Condition language ("moderate deterioration was noted") | All inspection dates | | Section headings and structural phrases | All phone numbers and emails | The embedding of a scrubbed reference chunk therefore captures style and vocabulary — which is the legitimate purpose of the reference tier — while containing nothing that could identify the original property or client. **Metadata flag:** All TIER_REFERENCE chunks are stored with `is_scrubbed: true`. This flag is set unconditionally at ingest — there is no code path that stores a TIER_REFERENCE chunk without first calling `scrub_rag_chunk()`. --- ### The Search-Level Safety Backstop Even if the ingest logic were to fail — due to a bug, a race condition, or a direct database manipulation — the search function provides a second line of defence. In `SurveyRagStore.search()`, any result where `tier == TIER_REFERENCE` and `is_scrubbed == False` is excluded from the returned list before it reaches the mapping LLM. The condition is checked on every search call: ```python # In SurveyRagStore.search() — enforced on every call if meta['tier'] == TIER_REFERENCE and not meta.get('is_scrubbed', False): continue # Never returned to the LLM ``` This means a TIER_REFERENCE chunk that somehow entered the store without scrubbing will never appear in a retrieval result. It remains in the index (so the ingest appears to succeed) but is permanently invisible to the mapping step. --- ### The Audit Trail Every scrubbing action on every ingested document is logged. The log records: - Document filename and tier - Timestamp of ingest - Number of chunks processed - For each chunk: a list of `{type, count}` redaction entries (e.g., `{"type": "[REDACTED_NAME]", "count": 3}`) - Whether the chunk passed `is_scrubbed` verification The log never records the original values of redacted items — only their type and count. This provides compliance evidence that scrubbing occurred without creating a secondary record of the PII itself. --- ### Implementation in `scrub_rag_chunk()` (already shown in MODULE 1) ```python def scrub_rag_chunk(chunk: str) -> str: """ Applied to every paragraph chunk from a TIER_REFERENCE document before embedding or storage. Uses the same two-pass pipeline as note scrubbing. Called by: SurveyRagStore.ingest_document() when tier == TIER_REFERENCE. Never called on TIER_MASTER documents. The ingest_document() method is the only place this function is called. There is no other code path that stores a TIER_REFERENCE chunk. """ scrubbed, _ = scrub(chunk) return scrubbed ``` The strict call-site rule — only `ingest_document()` calls `scrub_rag_chunk()`, and `ingest_document()` always calls it for TIER_REFERENCE — means there is no ambiguity about when scrubbing applies. A code reviewer can verify the policy by searching the codebase for calls to `scrub_rag_chunk()`: there must be exactly one, inside `ingest_document()`, guarded by `if tier == TIER_REFERENCE`. --- ### Summary Table | Stage | Document type | Scrubbing applied | Guard mechanism | |---|---|---|---| | Notes from surveyor | N/A (not a document) | `scrub()` — always | Hard gate before any LLM call | | Master template ingest | TIER_MASTER | None | `assert_no_pii()` rejects ingest if PII found | | Reference report ingest | TIER_REFERENCE | `scrub_rag_chunk()` — every chunk | `is_scrubbed` flag; search excludes unverified chunks | | LLM output paragraph | N/A (generated text) | `assert_no_pii()` then `scrub()` if needed | Raises ValueError if PII found post-generation | --- ## MODULE 2: TEMPLATE SCHEMA DISCOVERY (`core/template_discoverer.py`) This is the foundational new module in v2. It runs exactly once per firm — when they upload their master template document. It extracts a complete machine-readable schema that every other module uses as its sole structural authority. ### What the Schema Captures ```python # models/schema.py from pydantic import BaseModel from typing import List, Optional, Dict, Any class RatingValue(BaseModel): value: str # e.g. "1", "2", "3", "A", "NI" meaning: Optional[str] = None # e.g. "no immediate action", "urgent repair" class RatingSystem(BaseModel): detected: bool = False type: Optional[str] = None # "numeric", "letter", "text", "boolean" — None if absent values: List[RatingValue] = [] # Ordered list of possible values format_template: Optional[str] = None # e.g. "Condition Rating [VALUE]" — None if absent # How the rating appears in the template paragraph (for LLM to replicate) inline_example: Optional[str] = None # Verbatim example from template class SectionDefinition(BaseModel): id: str # Stable identifier (auto-generated if none in template) label: str # Human label exactly as it appears in the template order: int # 1-based position in the template parent_id: Optional[str] = None # For sub-sections; None if flat structure has_rating_field: bool = False # Whether this section contains a rating in the template rating_inline_format: Optional[str] = None # e.g. "Condition Rating [X]" — verbatim keywords: List[str] = [] # Auto-extracted for notes matching placeholder_hints: List[str] = [] # Placeholder strings found in this section's template text class PlaceholderSyntax(BaseModel): detected_formats: List[str] = [] # All placeholder formats found, e.g. ["[PLACEHOLDER]"] primary_format: Optional[str] = None # Most common format class TemplateSchema(BaseModel): tenant_id: str source_filename: str extracted_at: str # ISO timestamp report_type: Optional[str] = None # Detected report type, e.g. "RICS Home Survey Level 3" sections: List[SectionDefinition] = [] section_hierarchy: str = "flat" # "flat", "two-level", "three-level" rating_system: RatingSystem = RatingSystem() placeholder_syntax: PlaceholderSyntax = PlaceholderSyntax() raw_section_texts: Dict[str, str] = {} # section_id → full template text for that section # Any custom metadata the discovery LLM extracted additional_metadata: Dict[str, Any] = {} ``` ### Discovery Prompt (`prompts/discovery_prompt.py`) ```python # prompts/discovery_prompt.py DISCOVERY_SYSTEM = """ You are a document structure analyst. You receive the extracted text and heading structure of a professional report template. Your task is to return a precise machine-readable JSON schema describing the template's structure. You must extract exactly what is present in the document. You must never add, invent, or assume structural elements that are not explicitly present. Specifically: - If the template has a rating or condition system, describe it precisely using the exact values and format from the document. If there is no rating system, set "detected": false and do not describe one. - If sections have sub-sections, reflect that hierarchy. If the structure is flat, say so. - Extract keywords for each section from its label and opening sentences only. Do not invent keywords from general domain knowledge. - For placeholders: identify the exact syntax used (e.g. "[...]", "{...}", "___"). If no placeholders exist, leave the list empty. Return ONLY a valid JSON object. No preamble, no markdown fences, no explanation. The JSON must match this schema exactly: { "report_type": string or null, "section_hierarchy": "flat" | "two-level" | "three-level", "rating_system": { "detected": boolean, "type": "numeric" | "letter" | "text" | "boolean" | null, "values": [{"value": string, "meaning": string or null}], "format_template": string or null, "inline_example": string or null }, "placeholder_syntax": { "detected_formats": [string], "primary_format": string or null }, "sections": [ { "id": string, "label": string, "order": integer, "parent_id": string or null, "has_rating_field": boolean, "rating_inline_format": string or null, "keywords": [string], "placeholder_hints": [string] } ], "additional_metadata": {} } """ DISCOVERY_USER_TEMPLATE = """ DOCUMENT FILENAME: {filename} EXTRACTED HEADING STRUCTURE: {heading_outline} EXTRACTED FULL TEXT (first 8000 characters): {document_text_excerpt} Analyse this template document and return the JSON schema. """ ``` ### Template Discoverer Implementation ```python # core/template_discoverer.py import json import anthropic from datetime import datetime, timezone from utils.doc_extractor import extract_structure from models.schema import TemplateSchema, SectionDefinition, RatingSystem, PlaceholderSyntax from prompts.discovery_prompt import DISCOVERY_SYSTEM, DISCOVERY_USER_TEMPLATE client = anthropic.Anthropic() def discover_schema( doc_bytes: bytes, filename: str, tenant_id: str, content_type: str ) -> TemplateSchema: """ Reads an uploaded master template document and returns a TemplateSchema. Steps: 1. Extract text + heading outline from the document 2. Send to LLM with strict JSON-only instructions 3. Parse and validate the returned JSON 4. Augment with raw section texts for the RAG store 5. Persist schema.json for this tenant """ # Step 1: Extract structure from document bytes structure = extract_structure(doc_bytes, content_type) # structure = { # "heading_outline": "...", # Indented list of headings # "full_text": "...", # Plain text of entire document # "section_texts": { # Heading label → body text for that section # "Chimney Stacks": "...", # ... # } # } # Step 2: LLM schema extraction user_message = DISCOVERY_USER_TEMPLATE.format( filename=filename, heading_outline=structure["heading_outline"], document_text_excerpt=structure["full_text"][:8000] # Respect context limits ) response = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=4000, system=DISCOVERY_SYSTEM, messages=[{"role": "user", "content": user_message}] ) raw_json = response.content[0].text.strip() # Strip any accidental markdown fences if raw_json.startswith("```"): raw_json = raw_json.split("```")[1] if raw_json.startswith("json"): raw_json = raw_json[4:] raw_json = raw_json.strip() # Step 3: Parse and validate try: data = json.loads(raw_json) except json.JSONDecodeError as e: raise ValueError( f"Schema discovery LLM returned invalid JSON: {e}\n" f"Raw response: {raw_json[:500]}" ) # Step 4: Build TemplateSchema from parsed data sections = [] for i, s in enumerate(data.get("sections", [])): sections.append(SectionDefinition( id=s.get("id", f"S{i+1:03d}"), label=s.get("label", f"Section {i+1}"), order=s.get("order", i + 1), parent_id=s.get("parent_id"), has_rating_field=s.get("has_rating_field", False), rating_inline_format=s.get("rating_inline_format"), keywords=s.get("keywords", []), placeholder_hints=s.get("placeholder_hints", []) )) rating_data = data.get("rating_system", {}) rating_system = RatingSystem( detected=rating_data.get("detected", False), type=rating_data.get("type"), values=rating_data.get("values", []), format_template=rating_data.get("format_template"), inline_example=rating_data.get("inline_example") ) ph_data = data.get("placeholder_syntax", {}) placeholder_syntax = PlaceholderSyntax( detected_formats=ph_data.get("detected_formats", []), primary_format=ph_data.get("primary_format") ) schema = TemplateSchema( tenant_id=tenant_id, source_filename=filename, extracted_at=datetime.now(timezone.utc).isoformat(), report_type=data.get("report_type"), sections=sections, section_hierarchy=data.get("section_hierarchy", "flat"), rating_system=rating_system, placeholder_syntax=placeholder_syntax, raw_section_texts=structure.get("section_texts", {}), additional_metadata=data.get("additional_metadata", {}) ) # Step 5: Persist _persist_schema(schema, tenant_id) return schema def load_schema(tenant_id: str) -> TemplateSchema: """Load a previously discovered schema from disk.""" from pathlib import Path schema_path = Path(f"data/tenants/{tenant_id}/schema.json") if not schema_path.exists(): raise FileNotFoundError( f"No template schema found for tenant {tenant_id}. " f"Upload a master template document first." ) with open(schema_path, "r") as f: return TemplateSchema.model_validate_json(f.read()) def _persist_schema(schema: TemplateSchema, tenant_id: str): from pathlib import Path schema_dir = Path(f"data/tenants/{tenant_id}") schema_dir.mkdir(parents=True, exist_ok=True) with open(schema_dir / "schema.json", "w") as f: f.write(schema.model_dump_json(indent=2)) ``` ### Doc Extractor Utility (`utils/doc_extractor.py`) ```python # utils/doc_extractor.py from typing import Dict from docx import Document import pypdf def extract_structure(doc_bytes: bytes, content_type: str) -> Dict: """ Extracts a standardised structure dict from a DOCX or PDF. Returns: { "heading_outline": str, # Indented heading tree "full_text": str, # All body text concatenated "section_texts": dict[str, str] # heading label → body text beneath it } """ if 'pdf' in content_type: return _extract_from_pdf(doc_bytes) else: return _extract_from_docx(doc_bytes) def _extract_from_docx(doc_bytes: bytes) -> Dict: import io doc = Document(io.BytesIO(doc_bytes)) headings = [] section_texts = {} full_text_parts = [] current_heading = None current_body_parts = [] HEADING_STYLES = { 'Heading 1', 'Heading 2', 'Heading 3', 'Heading 4', 'Heading 5', 'Heading 6' } for para in doc.paragraphs: text = para.text.strip() if not text: continue style_name = para.style.name if para.style else "" if style_name in HEADING_STYLES: # Flush previous section if current_heading is not None: body = "\n".join(current_body_parts).strip() section_texts[current_heading] = body current_body_parts = [] # Determine indent level from heading number level = int(style_name[-1]) if style_name[-1].isdigit() else 1 indent = " " * (level - 1) headings.append(f"{indent}{text}") current_heading = text else: current_body_parts.append(text) full_text_parts.append(text) # Flush last section if current_heading is not None: section_texts[current_heading] = "\n".join(current_body_parts).strip() return { "heading_outline": "\n".join(headings), "full_text": "\n\n".join(full_text_parts), "section_texts": section_texts } def _extract_from_pdf(doc_bytes: bytes) -> Dict: import io reader = pypdf.PdfReader(io.BytesIO(doc_bytes)) full_text_parts = [] for page in reader.pages: page_text = page.extract_text() if page_text: full_text_parts.append(page_text) full_text = "\n\n".join(full_text_parts) # For PDFs, heading detection is heuristic # Lines that are short (< 80 chars) and followed by a blank line are likely headings lines = full_text.split("\n") headings = [] section_texts = {} current_heading = None current_body = [] for i, line in enumerate(lines): line = line.strip() if not line: continue next_line = lines[i + 1].strip() if i + 1 < len(lines) else "" is_likely_heading = ( len(line) < 80 and (not next_line or len(next_line) > 60) and not line.endswith(".") ) if is_likely_heading: if current_heading: section_texts[current_heading] = "\n".join(current_body).strip() current_body = [] headings.append(line) current_heading = line else: current_body.append(line) if current_heading: section_texts[current_heading] = "\n".join(current_body).strip() return { "heading_outline": "\n".join(headings), "full_text": full_text, "section_texts": section_texts } ``` --- ## MODULE 3: DYNAMIC NOTES PARSER (`core/notes_parser.py`) In v1 this module contained a hardcoded `RICS_SECTIONS` dictionary and a hardcoded `KEYWORD_SECTION_MAP`. In v2, both are built dynamically from the discovered schema at runtime. There are no hardcoded section codes anywhere. ```python # core/notes_parser.py import re from typing import List, Optional, Dict from pydantic import BaseModel from models.schema import TemplateSchema, SectionDefinition class SectionNote(BaseModel): section_id: str # From TemplateSchema.sections[i].id (dynamic) section_label: str # From TemplateSchema.sections[i].label raw_observations: List[str] # Bullet-form facts extracted from the notes rating_value: Optional[str] = None # Only populated if schema.rating_system.detected = True shorthand_expanded: Optional[str] = None # After optional expansion pass def build_keyword_map(schema: TemplateSchema) -> Dict[str, str]: """ Build a keyword → section_id map dynamically from the discovered schema. Sources for keywords per section: 1. schema.sections[i].keywords (discovered by LLM from heading + first sentences) 2. Words in the section label itself (lowercased, meaningful words only) Returns: {keyword: section_id} """ STOP_WORDS = { 'and', 'or', 'the', 'of', 'in', 'a', 'an', 'to', 'for', 'with', 'other', 'general', 'section', 'part', 'item' } keyword_map = {} for section in schema.sections: # From discovered keywords for kw in section.keywords: keyword_map[kw.lower()] = section.id # From label words label_words = re.findall(r'\b[a-zA-Z]{3,}\b', section.label) for word in label_words: w = word.lower() if w not in STOP_WORDS: keyword_map[w] = section.id return keyword_map def _detect_section_from_text( text: str, schema: TemplateSchema, keyword_map: Dict[str, str] ) -> Optional[str]: """ Attempt to identify which schema section a block of text belongs to. Three strategies, in order: 1. Explicit section ID match (e.g. "D1:", "Section 3:", "3.2") 2. Explicit section label match (case-insensitive) 3. Keyword match from keyword_map Returns section_id or None if no match found. """ text_lower = text.lower() # Strategy 1: Explicit section ID — look for pattern at start of line for section in schema.sections: pattern = r'(?i)^\s*' + re.escape(section.id) + r'\s*[:\.\-]' if re.search(pattern, text, re.MULTILINE): return section.id # Strategy 2: Explicit section label for section in schema.sections: if section.label.lower() in text_lower: return section.id # Strategy 3: Keyword match (first match wins — ordered by section order) words_in_text = set(re.findall(r'\b[a-zA-Z]{3,}\b', text_lower)) for section in sorted(schema.sections, key=lambda s: s.order): section_keywords = set(kw.lower() for kw in section.keywords) label_words = set( w.lower() for w in re.findall(r'\b[a-zA-Z]{3,}\b', section.label) ) all_kws = section_keywords | label_words if all_kws & words_in_text: return section.id return None def _extract_rating_value(text: str, schema: TemplateSchema) -> Optional[str]: """ If the schema has a rating system, attempt to find a rating value in this text. Returns the matched value string or None. Only runs if schema.rating_system.detected is True. Uses the known values from the schema — never invents new values. """ if not schema.rating_system.detected: return None for rv in schema.rating_system.values: # Look for the value as a standalone token (not embedded in a larger number) pattern = r'(? List[SectionNote]: """ Converts scrubbed surveyor notes into a list of SectionNote objects. The section taxonomy is entirely determined by the schema — no hardcoded codes. Strategy: 1. Split the notes into blocks (double newline, or explicit section markers) 2. For each block, detect which schema section it belongs to 3. Extract any rating value (only if schema.rating_system.detected) 4. Split the block into individual observations (one per bullet / sentence) 5. Collect unassigned text into an UNASSIGNED bucket for the user to review Returns a list of SectionNote, one per detected section (merged if same section appears multiple times in the notes). """ keyword_map = build_keyword_map(schema) # Split into blocks blocks = re.split(r'\n{2,}', scrubbed_notes.strip()) # Accumulate observations per section section_accumulator: Dict[str, List[str]] = {} section_ratings: Dict[str, Optional[str]] = {} unassigned_blocks: List[str] = [] for block in blocks: block = block.strip() if not block: continue section_id = _detect_section_from_text(block, schema, keyword_map) if section_id is None: unassigned_blocks.append(block) continue # Extract individual observations from the block # Split on: newlines, bullet characters, semicolons at sentence boundaries raw_lines = re.split(r'\n|(?<=\.)\s+(?=[A-Z])', block) observations = [] for line in raw_lines: line = line.strip().lstrip('•-*·▸▹◆►').strip() if line and len(line) > 5: observations.append(line) if section_id not in section_accumulator: section_accumulator[section_id] = [] section_accumulator[section_id].extend(observations) # Rating detection (only if schema has rating system) if section_id not in section_ratings: section_ratings[section_id] = _extract_rating_value(block, schema) # Build SectionNote objects result = [] schema_sections_by_id = {s.id: s for s in schema.sections} for section_id, observations in section_accumulator.items(): section_def = schema_sections_by_id.get(section_id) if not section_def: continue result.append(SectionNote( section_id=section_id, section_label=section_def.label, raw_observations=observations, rating_value=section_ratings.get(section_id) )) # Sort by schema order result.sort( key=lambda sn: schema_sections_by_id.get(sn.section_id, SectionDefinition( id="", label="", order=9999 )).order ) # Attach unassigned blocks as a special section so nothing is silently lost if unassigned_blocks: result.append(SectionNote( section_id="UNASSIGNED", section_label="Unassigned Observations", raw_observations=unassigned_blocks, rating_value=None )) return result ``` --- ## MODULE 4: RAG STORE (`core/rag_store.py`) The RAG store in v2 is schema-agnostic. The `section_code` field from v1 is replaced by a `section_id` that comes from the schema, not from a hardcoded dictionary. Both document tiers (master template paragraphs, reference past reports) remain — the distinction between them is PII handling, not structural assumption. ```python # core/rag_store.py import faiss import numpy as np import json import os from pathlib import Path from typing import List, Dict, Optional from openai import OpenAI embed_client = OpenAI() # For text-embedding-3-small TIER_MASTER = 'master' # Firm's standard template paragraphs — NOT PII-scrubbed TIER_REFERENCE = 'reference' # Past completed reports — PII-scrubbed on ingest class SurveyRagStore: """ Per-tenant FAISS vector store. Schema-agnostic: section_id is whatever string the schema defines. Metadata per chunk: chunk_id — unique identifier section_id — schema section id (e.g. "D1", "S003", "Roof Coverings") section_label — human-readable label paragraph_text — the text of this chunk source_doc — original filename tier — TIER_MASTER or TIER_REFERENCE is_scrubbed — bool: was PII scrubbing applied at ingest? """ EMBED_DIM = 1536 # text-embedding-3-small output dimension def __init__(self, tenant_id: str): self.tenant_id = tenant_id self.store_path = Path(f'data/tenants/{tenant_id}') self.store_path.mkdir(parents=True, exist_ok=True) index_path = self.store_path / 'index.faiss' meta_path = self.store_path / 'metadata.json' if index_path.exists(): self.index = faiss.read_index(str(index_path)) with open(meta_path, 'r') as f: self.metadata: List[Dict] = json.load(f) else: # Inner product on normalised vectors = cosine similarity self.index = faiss.IndexFlatIP(self.EMBED_DIM) self.metadata = [] def ingest_document( self, doc_text: str, source_filename: str, tier: str, section_id_hint_map: Optional[Dict[str, str]] = None # Maps heading text → section_id from schema, so chunks get correct IDs ) -> int: """ Chunk the document by paragraph boundary, embed, and store. Returns the number of chunks ingested. For TIER_MASTER: preserve original text exactly. For TIER_REFERENCE: PII-scrub each chunk before embedding. """ from core.pii_scrubber import scrub_rag_chunk paragraphs = _split_into_paragraphs(doc_text) ingested = 0 for i, para in enumerate(paragraphs): is_scrubbed = False if tier == TIER_REFERENCE: para = scrub_rag_chunk(para) is_scrubbed = True # Detect section_id: use hint map first, then keyword/label search section_id = 'UNKNOWN' section_label = '' if section_id_hint_map: for heading, sid in section_id_hint_map.items(): if heading.lower() in para.lower()[:120]: section_id = sid break embedding = _embed(para) norm = np.linalg.norm(embedding) if norm > 0: embedding = embedding / norm self.index.add(np.array([embedding], dtype='float32')) self.metadata.append({ 'chunk_id': f'{source_filename}_chunk_{i:04d}', 'section_id': section_id, 'section_label': section_label, 'paragraph_text': para, 'source_doc': source_filename, 'tier': tier, 'is_scrubbed': is_scrubbed, }) ingested += 1 self._persist() return ingested def search( self, query: str, section_id: str, top_k: int = 5, tier_preference: str = TIER_MASTER ) -> List[Dict]: """ Retrieve top_k most relevant paragraphs for the given section. Priority ranking: 1. Same section_id AND tier_preference → highest rank 2. Same section_id, any tier → second rank 3. Semantically similar, any section → fallback IMPORTANT: Chunks from TIER_REFERENCE that were NOT scrubbed (is_scrubbed=False) are excluded from results as a safety measure to prevent PII leakage. """ if self.index.ntotal == 0: return [] query_emb = _embed(query) norm = np.linalg.norm(query_emb) if norm > 0: query_emb = query_emb / norm pool_k = min(top_k * 15, self.index.ntotal) scores, indices = self.index.search( np.array([query_emb], dtype='float32'), pool_k ) results = [] for score, idx in zip(scores[0], indices[0]): if idx == -1: continue meta = self.metadata[idx].copy() meta['score'] = float(score) # Safety: exclude reference-tier chunks that weren't scrubbed if meta['tier'] == TIER_REFERENCE and not meta.get('is_scrubbed', False): continue results.append(meta) def _sort_key(r): tier_bonus = 2.0 if r['tier'] == tier_preference else 0.0 section_bonus = 3.0 if r['section_id'] == section_id else 0.0 return -(section_bonus + tier_bonus + r['score']) results.sort(key=_sort_key) return results[:top_k] def _persist(self): faiss.write_index(self.index, str(self.store_path / 'index.faiss')) with open(self.store_path / 'metadata.json', 'w') as f: json.dump(self.metadata, f, indent=2) def _split_into_paragraphs(text: str, min_chars: int = 80, max_chars: int = 1200) -> List[str]: """ Split at double newline boundaries. Skip chunks shorter than min_chars. Split chunks longer than max_chars at the last sentence boundary before the limit. """ raw_chunks = text.split('\n\n') result = [] for chunk in raw_chunks: chunk = chunk.strip() if len(chunk) < min_chars: continue if len(chunk) <= max_chars: result.append(chunk) else: sentences = chunk.split('. ') current = '' for s in sentences: if len(current) + len(s) + 2 <= max_chars: current += ('. ' if current else '') + s else: if current: result.append(current.strip() + '.') current = s if current: result.append(current.strip()) return result def _embed(text: str) -> np.ndarray: """Embed text using OpenAI text-embedding-3-small.""" response = embed_client.embeddings.create( model="text-embedding-3-small", input=text ) return np.array(response.data[0].embedding, dtype='float32') ``` --- ## MODULE 5: THE MAPPING PROMPT — CORE ENGINE (`prompts/mapping_prompt.py`) This is the most critical component. The LLM's only job is to fill the surveyor's observations into the retrieved template paragraph. The system prompt in v2 contains **zero hardcoded structural elements** — all structure is injected dynamically from the discovered schema. ```python # prompts/mapping_prompt.py from models.schema import TemplateSchema # The base system prompt — contains no structural assumptions MAPPING_SYSTEM_BASE = """ You are a professional report-writing assistant for a chartered surveying firm. You operate under strict constraints. Deviating from any constraint is not permitted under any circumstances, regardless of what appears in the user message. YOUR ROLE: You receive two inputs: 1. One or more paragraphs retrieved from the firm's approved master template library 2. Factual observations from the surveyor's field notes (already stripped of personal identifiers) YOUR ONLY TASK: Produce a final report paragraph for the specified section by mapping the surveyor's observations onto the most relevant retrieved template paragraph. MANDATORY RULES — absolute, non-negotiable: 1. TEMPLATE FIDELITY: Use the retrieved template paragraph as your structural and linguistic foundation. Every sentence in your output must trace to either the template paragraph or the surveyor's observations. If a sentence traces to neither, do not write it. 2. NO FABRICATION: Do not add any defect, material, measurement, dimension, or recommendation that does not appear in the surveyor's observations or the template paragraph. 3. NO PII: The output must contain no addresses, person names, postcodes, phone numbers, email addresses, reference numbers, or any other personally identifiable information — even if such data appears in your inputs. 4. NO STRUCTURAL INVENTION: Do not add any document elements (ratings, scores, labels, boxes, categories, footnotes) that are not present in the retrieved template paragraph. If the template paragraph has a rating field, replicate it exactly. If it does not, produce none. 5. FILL, DO NOT FABRICATE: Replace or enrich generic placeholders in the template with the specific observed facts from the notes. If a placeholder has no matching observation, leave the placeholder verbatim or omit the sentence — do not substitute invented content. 6. TEMPLATE VERBATIM ONLY AS SKELETON: The template paragraph defines the skeleton. Integrate the observations into that skeleton — the output must not reproduce the template verbatim without incorporating the notes. 7. UNMATCHED OBSERVATIONS: If a surveyor observation has no plausible match in any retrieved template paragraph, append it as: [UNMATCHED_OBSERVATION: ] Do not attempt to draft prose for unmatched observations. OUTPUT FORMAT: Return only the final mapped paragraph text followed by any [UNMATCHED_OBSERVATION] tags. No headings, no section labels, no preamble, no explanation. """ def build_mapping_system_prompt(schema: TemplateSchema) -> str: """ Appends schema-specific structural rules to the base system prompt. These rules are derived from the discovered schema — never hardcoded. If the schema has a rating system, the LLM is told its exact format. If it does not, no rating instructions are added at all. """ additions = [] if schema.rating_system.detected: # Build the rating instruction from the schema's own values and format values_desc = ", ".join( f"{rv.value}" + (f" ({rv.meaning})" if rv.meaning else "") for rv in schema.rating_system.values ) format_template = schema.rating_system.format_template or "[VALUE]" example = schema.rating_system.inline_example or "" additions.append(f""" RATING SYSTEM: This template uses a rating system. The permissible values are: {values_desc}. The format as it appears in template paragraphs is: {format_template} {f'Example from the template: "{example}"' if example else ''} When the retrieved template paragraph contains a rating field, fill it with the value indicated in the surveyor's notes. If the notes do not specify a value, use the value that best reflects the described condition — but only choose from the permitted values listed above. Never invent a rating value not in this list. """) else: additions.append(""" RATING SYSTEM: This template does not use a rating system. Do not add any rating, score, condition label, or classification to the output under any circumstances. """) if schema.placeholder_syntax.primary_format: additions.append(f""" PLACEHOLDER SYNTAX: This template uses the following format for placeholders: {schema.placeholder_syntax.primary_format} When you see this format in the retrieved paragraph, replace it with the appropriate observed fact from the surveyor's notes. If no observation matches a placeholder, leave it verbatim. """) return MAPPING_SYSTEM_BASE + "\n".join(additions) def build_mapping_messages( section_id: str, section_label: str, retrieved_paragraphs: list, scrubbed_observations: list, schema: TemplateSchema, rating_value: str = None ) -> list: """ Builds the messages list for the Claude API call. All context about the rating system comes from the schema — nothing is hardcoded. """ # Format retrieved paragraphs rag_block = "" for i, para in enumerate(retrieved_paragraphs[:3], 1): rag_block += ( f"[RETRIEVED TEMPLATE PARAGRAPH {i} — " f"Source: {para['tier'].upper()}, " f"Section: {para['section_id']}]\n" f"{para['paragraph_text'].strip()}\n\n" ) # Format observations obs_block = "\n".join(f"• {obs}" for obs in scrubbed_observations) # Rating instruction — only if schema has rating system AND a value was detected rating_instruction = "" if schema.rating_system.detected and rating_value: rating_instruction = ( f"\nThe surveyor's notes indicate a rating value of: {rating_value}\n" f"Reflect this in the output using the template's rating format.\n" ) user_message = f"""SECTION: {section_id} — {section_label} {rating_instruction} --- RETRIEVED TEMPLATE PARAGRAPHS --- {rag_block} --- SURVEYOR'S FIELD OBSERVATIONS (PII removed) --- {obs_block} --- TASK --- Map the field observations onto the most relevant retrieved template paragraph above. Produce the final report paragraph for section '{section_label}'. Do not add any element not present in the template paragraphs above. """ return [{"role": "user", "content": user_message}] ``` --- ## MODULE 6: SECTION MAPPER — ORCHESTRATOR (`core/section_mapper.py`) Ties every module together. Every structural decision flows from the schema. ```python # core/section_mapper.py import anthropic from core.pii_scrubber import scrub, assert_no_pii from core.notes_parser import parse_notes_to_sections from core.rag_store import SurveyRagStore, TIER_MASTER from core.grounding_checker import check_grounding from core.template_discoverer import load_schema from prompts.mapping_prompt import build_mapping_system_prompt, build_mapping_messages from models.schema import TemplateSchema client = anthropic.Anthropic() def generate_report( raw_notes: str, tenant_id: str, property_metadata: dict, # {property_type, tenure} — NO address, NO names schema: TemplateSchema = None # If None, loaded from disk ) -> dict: """ Master pipeline. Returns a dict keyed by section_id, each containing: - status: 'OK' | 'NO_RAG_MATCH' | 'GROUNDING_REVIEW' | 'UNASSIGNED' - paragraph: the final mapped text - rating_value: (only if schema has rating system and one was detected) - rag_sources: chunk IDs used - grounding_passed: bool - unmatched_observations: list of observations with no matching template Pipeline: 1. Load schema (or use passed schema) 2. PII-scrub raw notes 3. Parse notes into sections (using schema taxonomy) 4. For each section: retrieve template paragraphs → map → ground-check → PII-check 5. Return all sections """ # Step 1: Load schema if schema is None: schema = load_schema(tenant_id) # Step 2: PII scrub — MUST happen before any LLM call scrubbed_notes, redaction_log = scrub(raw_notes) # Step 3: Parse into schema-driven sections sections = parse_notes_to_sections(scrubbed_notes, schema) # Step 4: Build schema-aware mapping system prompt (done once, reused per section) mapping_system = build_mapping_system_prompt(schema) # Step 5: Load RAG store rag_store = SurveyRagStore(tenant_id) results = {} for section in sections: # Handle UNASSIGNED bucket — surface for manual review, do not attempt to map if section.section_id == 'UNASSIGNED': results['UNASSIGNED'] = { 'status': 'UNASSIGNED', 'paragraph': '[These observations could not be assigned to any template section. Manual review required.]', 'unmatched_observations': section.raw_observations, 'grounding_passed': False, } continue # Build semantic search query from section label + top observations query = f"{section.section_label}: " + " ".join(section.raw_observations[:3]) # Retrieve template paragraphs (master tier preferred) retrieved = rag_store.search( query=query, section_id=section.section_id, top_k=5, tier_preference=TIER_MASTER ) if not retrieved: results[section.section_id] = { 'status': 'NO_RAG_MATCH', 'paragraph': ( f'[No template paragraph found for section "{section.section_label}". ' f'Manual entry required.]' ), 'unmatched_observations': section.raw_observations, 'grounding_passed': False, } continue # Build mapping messages and call LLM messages = build_mapping_messages( section_id=section.section_id, section_label=section.section_label, retrieved_paragraphs=retrieved, scrubbed_observations=section.raw_observations, schema=schema, rating_value=section.rating_value ) response = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=1000, system=mapping_system, messages=messages, ) mapped_text = response.content[0].text.strip() # Extract UNMATCHED_OBSERVATION tags before grounding check import re unmatched_tags = re.findall( r'\[UNMATCHED_OBSERVATION:\s*(.*?)\]', mapped_text, re.DOTALL ) mapped_text_clean = re.sub( r'\[UNMATCHED_OBSERVATION:.*?\]', '', mapped_text, flags=re.DOTALL ).strip() # Grounding check grounding_result = check_grounding( mapped_paragraph=mapped_text_clean, source_observations=section.raw_observations, source_rag_paragraphs=[p['paragraph_text'] for p in retrieved] ) if not grounding_result['passed']: mapped_text_clean = grounding_result['cleaned_text'] # Final PII belt-and-braces check if not assert_no_pii(mapped_text_clean): mapped_text_clean, _ = scrub(mapped_text_clean) results[section.section_id] = { 'status': 'OK' if grounding_result['passed'] else 'GROUNDING_REVIEW', 'paragraph': mapped_text_clean, 'section_label': section.section_label, 'rating_value': section.rating_value if schema.rating_system.detected else None, 'rag_sources': [p['chunk_id'] for p in retrieved[:3]], 'grounding_passed': grounding_result['passed'], 'grounding_violations': grounding_result.get('violations', []), 'unmatched_observations': unmatched_tags, } return results ``` --- ## MODULE 7: GROUNDING CHECKER (`core/grounding_checker.py`) Unchanged in logic from v1 — but the language in the auditor prompt is made domain-agnostic (no RICS-specific references). ```python # core/grounding_checker.py import re import json import anthropic client = anthropic.Anthropic() GROUNDING_SYSTEM = """ You are a quality auditor for professional survey reports. Your task is to verify that every factual claim in a generated paragraph is supported by at least one of two allowed sources: Source A: The surveyor's original field observations Source B: The retrieved template paragraph(s) Flag any content that: - States a specific measurement, defect, material, or product name not present in either source - Makes a recommendation not supported by either source - Contains any proper noun (person name, company, address) that slipped through scrubbing - Introduces a structural element (label, rating, category) not present in the template source Return ONLY a valid JSON object. No preamble, no explanation. { "passed": true | false, "violations": ["...list of specific invented phrases or sentences..."], "cleaned_text": "...the paragraph with violations replaced by [REVIEW_REQUIRED: ]..." } """ def check_grounding( mapped_paragraph: str, source_observations: list, source_rag_paragraphs: list ) -> dict: """ Two-stage grounding audit: Stage 1 — Regex fast pass: catches any PII patterns that slipped through Stage 2 — LLM semantic audit: catches invented content that regex cannot detect """ # Stage 1: Fast regex PII check pii_patterns = [ r'\b[A-Z]{1,2}[0-9][0-9A-Z]?\s*[0-9][A-Z]{2}\b', # UK postcodes r'\b\d{5}(?:-\d{4})?\b', # US ZIP codes r'\b\d{6,}\b', # Long numeric IDs r'\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b', # Emails ] regex_violations = [] cleaned = mapped_paragraph for pattern in pii_patterns: matches = re.findall(pattern, cleaned) if matches: regex_violations.extend(matches) cleaned = re.sub(pattern, '[REVIEW_REQUIRED: PII detected]', cleaned) # Stage 2: LLM semantic audit obs_text = '\n'.join(f' • {o}' for o in source_observations) rag_text = '\n\n---\n\n'.join(source_rag_paragraphs[:2]) audit_message = f"""GENERATED PARAGRAPH TO AUDIT: {mapped_paragraph} ALLOWED SOURCE A — SURVEYOR FIELD OBSERVATIONS: {obs_text} ALLOWED SOURCE B — RETRIEVED TEMPLATE PARAGRAPHS: {rag_text} Audit the generated paragraph. Every factual claim must trace to Source A or Source B. Return the JSON audit result.""" response = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=600, system=GROUNDING_SYSTEM, messages=[{"role": "user", "content": audit_message}] ) raw = response.content[0].text.strip() if raw.startswith("```"): raw = raw.split("```")[1].lstrip("json").strip() try: result = json.loads(raw) except json.JSONDecodeError: # Parse failure: treat as passed if no regex violations result = { 'passed': len(regex_violations) == 0, 'violations': regex_violations, 'cleaned_text': cleaned } if regex_violations: result['passed'] = False result['violations'] = regex_violations + result.get('violations', []) result['cleaned_text'] = cleaned # Use regex-cleaned version return result ``` --- ## MODULE 8: REPORT ASSEMBLER (`core/report_assembler.py`) In v2, the assembler has no hardcoded section order, no hardcoded condition colours, and no hardcoded structural elements. Everything is driven by the discovered schema. ```python # core/report_assembler.py import io from docx import Document from docx.shared import Pt, RGBColor from docx.oxml.ns import qn from models.schema import TemplateSchema # Only if the schema has a rating system with numeric values 1-3 do we apply # standard traffic-light colours. Even these are not applied if the schema # does not define colour conventions. DEFAULT_RATING_COLORS = { '3': RGBColor(0xBF, 0x1E, 0x2E), # Red '2': RGBColor(0xF5, 0xA6, 0x23), # Amber '1': RGBColor(0x2E, 0x7D, 0x32), # Green 'NI': RGBColor(0x90, 0x90, 0x90), # Grey 'R': RGBColor(0x1A, 0x23, 0x7E), # Dark blue } def build_docx( mapped_sections: dict, # {section_id: {paragraph, rating_value, status, ...}} report_metadata: dict, # {property_type, tenure} — NO addresses or names schema: TemplateSchema, # The discovered schema — sole structural authority template_docx_path: str = None # Path to branded DOCX template if provided ) -> bytes: """ Assembles the final report DOCX. Returns bytes for download. Section order is taken from schema.sections ordered by section.order. No section codes, headings, or structural elements are added unless they originate from the schema or the mapped paragraph text. """ doc = Document(template_docx_path) if template_docx_path else Document() if not template_docx_path: # Only apply minimal generic styles if no branded template is provided _apply_minimal_styles(doc) # Cover page — property metadata only (no addresses from AI pipeline) _add_cover_page(doc, report_metadata, schema) doc.add_page_break() # Sections — in the exact order defined by the schema ordered_sections = sorted(schema.sections, key=lambda s: s.order) for section_def in ordered_sections: sid = section_def.id if sid not in mapped_sections: continue # Section not in notes — omit entirely section_data = mapped_sections[sid] _add_section(doc, section_def, section_data, schema) # Surface any unassigned observations at the end for manual review if 'UNASSIGNED' in mapped_sections: _add_unassigned_block(doc, mapped_sections['UNASSIGNED']) buffer = io.BytesIO() doc.save(buffer) return buffer.getvalue() def _add_section(doc, section_def, section_data, schema: TemplateSchema): """Add a single section to the document.""" status = section_data.get('status', 'OK') paragraph_text = section_data.get('paragraph', '') rating_value = section_data.get('rating_value') # Section heading — use the label from the schema, not any assumed code heading_text = f"{section_def.id} {section_def.label}" if section_def.id else section_def.label heading = doc.add_heading(heading_text, level=3) # Rating annotation — ONLY if the schema defines a rating system # AND this specific section has a rating field in the template # AND a rating value was detected in the notes if ( schema.rating_system.detected and section_def.has_rating_field and rating_value is not None ): rating_run = heading.add_run(f" {rating_value}") rating_run.bold = True # Apply colour if it maps to a known colour — otherwise leave default color = DEFAULT_RATING_COLORS.get(rating_value) if color: rating_run.font.color.rgb = color # Paragraph body if status == 'NO_RAG_MATCH': p = doc.add_paragraph() p.add_run( f'[No template paragraph matched for "{section_def.label}". ' f'Manual entry required.]' ).italic = True elif status == 'GROUNDING_REVIEW': p = doc.add_paragraph(paragraph_text) review_p = doc.add_paragraph() review_p.add_run( '[Grounding review required — some content could not be verified against sources.]' ).italic = True else: doc.add_paragraph(paragraph_text) # Unmatched observations — surface clearly for manual drafting for unmatched in section_data.get('unmatched_observations', []): if unmatched.strip(): up = doc.add_paragraph() up.add_run( f'[Note requiring manual drafting: {unmatched}]' ).italic = True def _add_unassigned_block(doc, unassigned_data): """Adds a clearly labelled block for observations that couldn't be section-matched.""" doc.add_page_break() doc.add_heading('Unassigned Observations — Manual Review Required', level=2) p = doc.add_paragraph( 'The following observations from the surveyor\'s notes could not be ' 'automatically assigned to any template section. Please review and draft manually.' ) for obs in unassigned_data.get('unmatched_observations', []): doc.add_paragraph(f'• {obs}', style='List Bullet') def _add_cover_page(doc, metadata, schema: TemplateSchema): """ Minimal cover page using only non-PII metadata. Address and client name must be added manually by the user after export. """ doc.add_heading('Survey Report', level=1) if schema.report_type: doc.add_paragraph(schema.report_type) if metadata.get('property_type'): doc.add_paragraph(f"Property type: {metadata['property_type']}") if metadata.get('tenure'): doc.add_paragraph(f"Tenure: {metadata['tenure']}") doc.add_paragraph( '[Address, client name, inspection date and surveyor details ' 'to be completed by the surveyor before issue.]' ).italic = True def _apply_minimal_styles(doc): """Apply minimal generic styles when no branded template is provided.""" from docx.shared import Pt style = doc.styles['Normal'] font = style.font font.name = 'Calibri' font.size = Pt(10) ``` --- ## MODULE 9: API ENDPOINTS ### Template Upload + Schema Discovery (`api/routes/upload.py`) ```python # api/routes/upload.py from fastapi import APIRouter, UploadFile, File, Depends, HTTPException, Query from core.template_discoverer import discover_schema, load_schema from core.rag_store import SurveyRagStore, TIER_MASTER, TIER_REFERENCE from core.pii_scrubber import scrub_rag_chunk from utils.doc_extractor import extract_structure router = APIRouter(prefix='/api/upload', tags=['upload']) @router.post('/template') async def upload_master_template( file: UploadFile = File(...), tenant_id: str = Depends(get_current_tenant) ): """ Upload the firm's master template document. This triggers schema discovery and ingests the template into the MASTER tier RAG store. Must be done before any report generation. """ content = await file.read() content_type = file.content_type or 'application/octet-stream' # Step 1: Discover schema from the template schema = discover_schema( doc_bytes=content, filename=file.filename, tenant_id=tenant_id, content_type=content_type ) # Step 2: Ingest template paragraphs into MASTER tier RAG store structure = extract_structure(content, content_type) # Build section_id hint map from schema (label → id) for chunk attribution hint_map = {s.label: s.id for s in schema.sections} # Also add discovered raw section texts as individual ingestion units full_text = structure['full_text'] rag_store = SurveyRagStore(tenant_id) chunks_ingested = rag_store.ingest_document( doc_text=full_text, source_filename=file.filename, tier=TIER_MASTER, section_id_hint_map=hint_map ) return { 'status': 'template_ingested', 'filename': file.filename, 'sections_discovered': len(schema.sections), 'has_rating_system': schema.rating_system.detected, 'rating_type': schema.rating_system.type, 'report_type': schema.report_type, 'chunks_ingested': chunks_ingested, 'tier': TIER_MASTER, } @router.post('/reference') async def upload_reference_document( file: UploadFile = File(...), tenant_id: str = Depends(get_current_tenant) ): """ Upload a past completed report as a reference document. This is PII-scrubbed on ingest and stored in REFERENCE tier for terminology/style guidance. Reference documents are NEVER used as a source of facts — only style examples. """ content = await file.read() content_type = file.content_type or 'application/octet-stream' structure = extract_structure(content, content_type) full_text = structure['full_text'] # PII-scrub entire document before embedding scrubbed_text = scrub_rag_chunk(full_text) # Load schema to get hint map for section attribution try: schema = load_schema(tenant_id) hint_map = {s.label: s.id for s in schema.sections} except FileNotFoundError: hint_map = {} rag_store = SurveyRagStore(tenant_id) chunks_ingested = rag_store.ingest_document( doc_text=scrubbed_text, source_filename=file.filename, tier=TIER_REFERENCE, section_id_hint_map=hint_map ) return { 'status': 'reference_ingested', 'filename': file.filename, 'chunks_ingested': chunks_ingested, 'tier': TIER_REFERENCE, 'pii_scrubbed': True } @router.get('/schema') async def get_schema(tenant_id: str = Depends(get_current_tenant)): """Return the currently discovered schema for inspection.""" try: schema = load_schema(tenant_id) return schema.model_dump() except FileNotFoundError: raise HTTPException( status_code=404, detail='No template schema found. Upload a master template first.' ) ``` ### Report Generation (`api/routes/report.py`) ```python # api/routes/report.py from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import Response from pydantic import BaseModel, field_validator from core.section_mapper import generate_report from core.report_assembler import build_docx from core.template_discoverer import load_schema from core.pii_scrubber import scrub router = APIRouter(prefix='/api/report', tags=['report']) class GenerateReportRequest(BaseModel): raw_notes: str property_type: str # e.g. "semi-detached house", "flat" — generic descriptor tenure: str # e.g. "freehold", "leasehold" # CRITICAL: No client name, no address, no personal data here. # Address and client details are added manually by the surveyor post-export. @field_validator('raw_notes') @classmethod def notes_not_empty(cls, v): if not v.strip(): raise ValueError('raw_notes cannot be empty') return v @router.post('/generate') async def generate( req: GenerateReportRequest, tenant_id: str = Depends(get_current_tenant) ): """ Full pipeline: surveyor notes in → DOCX out. PII scrubbing is the first operation inside generate_report(). No raw notes ever reach an LLM. """ try: schema = load_schema(tenant_id) except FileNotFoundError: raise HTTPException( status_code=400, detail='No template has been uploaded for this account. ' 'Upload a master template document first.' ) mapped_sections = generate_report( raw_notes=req.raw_notes, tenant_id=tenant_id, property_metadata={ 'property_type': req.property_type, 'tenure': req.tenure, }, schema=schema ) docx_bytes = build_docx( mapped_sections=mapped_sections, report_metadata={ 'property_type': req.property_type, 'tenure': req.tenure, }, schema=schema ) return Response( content=docx_bytes, media_type=( 'application/vnd.openxmlformats-officedocument' '.wordprocessingml.document' ), headers={'Content-Disposition': 'attachment; filename="survey_report_draft.docx"'} ) @router.post('/preview') async def preview_mapped_sections( req: GenerateReportRequest, tenant_id: str = Depends(get_current_tenant) ): """ Returns JSON of mapped sections without building a DOCX. Useful for the UI to show section-by-section preview before download. """ try: schema = load_schema(tenant_id) except FileNotFoundError: raise HTTPException(status_code=400, detail='No template found.') mapped_sections = generate_report( raw_notes=req.raw_notes, tenant_id=tenant_id, property_metadata={ 'property_type': req.property_type, 'tenure': req.tenure, }, schema=schema ) return { 'schema_report_type': schema.report_type, 'sections_mapped': len([s for s in mapped_sections.values() if s['status'] == 'OK']), 'sections_needing_review': len([ s for s in mapped_sections.values() if s['status'] in ('NO_RAG_MATCH', 'GROUNDING_REVIEW', 'UNASSIGNED') ]), 'sections': mapped_sections } ``` --- ## MODULE 10: NOTES EXPANDER — OPTIONAL (`prompts/notes_expander.py`) The expander converts extreme shorthand into full professional observations BEFORE the mapping step. In v2, the shorthand dictionary is schema-informed — it does not assume any rating system exists, and adds rating-system shorthand only if the schema has one. ```python # prompts/notes_expander.py from models.schema import TemplateSchema EXPANDER_SYSTEM_BASE = """ You are a professional report-writing assistant for a property surveying firm. You receive shorthand field notes and expand them into full professional observations. RULES: - Expand only what is clearly implied by the shorthand. Never add defects, materials, or measurements not mentioned in the original. - If a shorthand code is ambiguous, preserve it verbatim with: [AMBIGUOUS: ] - Never add observations that are not present in the input. - Output a clean bulleted list — one observation per bullet. - Use clear, professional language consistent with a formal property survey. COMMON SHORTHAND (domain-agnostic building terms): - det / dett → deterioration noted - NI → not inspected - rep reqd → repair required - v. limited → very limited access / visibility - n/a → not applicable / not present - gen → generally / general condition - DPC → damp proof course - MC → moisture content reading - UV / uPVC → unplasticised polyvinyl chloride (plastic) material - conc → concrete - s/s → stainless steel - pt / part → part of / partial - approx → approximately - ext → external / exterior - int → internal / interior - org → original - prev → previous / previously """ def build_expander_system_prompt(schema: TemplateSchema) -> str: """ Appends rating-system-specific shorthand ONLY if the schema defines one. Never adds rating shorthand for schemas without a rating system. """ additions = EXPANDER_SYSTEM_BASE if schema.rating_system.detected and schema.rating_system.values: values_shorthand = "\n".join( f"- {rv.value} → rating value: {rv.value}" + (f" ({rv.meaning})" if rv.meaning else "") for rv in schema.rating_system.values ) format_hint = schema.rating_system.format_template or "[VALUE]" additions += f""" RATING SYSTEM SHORTHAND: This template uses the following rating values. Expand shorthand rating references using these exact values: {values_shorthand} When a rating value appears in the notes, expand it as: {format_hint} with the appropriate value substituted for [VALUE]. """ else: additions += """ RATING SYSTEM: This template does not use a rating system. Do not expand any shorthand as a rating, condition score, or similar classification. """ return additions ``` --- ## CONFIGURATION (`config.py`) ```python # config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): # API keys anthropic_api_key: str openai_api_key: str # For embeddings # Models embedding_model: str = "text-embedding-3-small" mapping_model: str = "claude-sonnet-4-20250514" grounding_model: str = "claude-sonnet-4-20250514" discovery_model: str = "claude-sonnet-4-20250514" # RAG settings rag_top_k: int = 5 paragraph_min_chars: int = 80 paragraph_max_chars: int = 1200 # PII scrubbing spacy_model: str = "en_core_web_trf" # Output max_tokens_mapping: int = 1000 max_tokens_grounding: int = 600 max_tokens_discovery: int = 4000 # Feature flags notes_expansion_enabled: bool = True grounding_check_enabled: bool = True # Grounding alert threshold — if more than this fraction of sections fail, # the response will flag the entire report for manual review grounding_alert_threshold: float = 0.2 class Config: env_file = ".env" settings = Settings() ``` --- ## REQUIREMENTS (`requirements.txt`) ``` fastapi>=0.110.0 uvicorn>=0.27.0 anthropic>=0.25.0 openai>=1.12.0 faiss-cpu>=1.8.0 numpy>=1.26.0 spacy>=3.7.0 python-docx>=1.1.0 pydantic>=2.6.0 pydantic-settings>=2.2.0 python-multipart>=0.0.9 python-jose>=3.3.0 passlib>=1.7.4 pypdf>=4.0.0 ``` --- ## COMPLETE DATA FLOW ``` TENANT ONBOARDING (one-time per firm) ───────────────────────────────────── Upload master template (DOCX or PDF) │ ▼ [TEMPLATE DISCOVERER] Extract heading structure + full text │ ▼ [DISCOVERY LLM CALL] Claude extracts TemplateSchema JSON: sections, ordering, rating system (if any), placeholder syntax, keywords │ ▼ schema.json → saved per tenant │ ▼ [RAG INGEST — MASTER TIER] Template paragraphs chunked + embedded into FAISS (NOT PII-scrubbed — it's boilerplate) Optionally: upload past completed reports │ ▼ [RAG INGEST — REFERENCE TIER] Past reports PII-scrubbed → embedded into FAISS REPORT GENERATION (per survey) ───────────────────────────────────── Surveyor submits raw field notes │ ▼ [1] PII SCRUBBER ──────────────── redaction_log (audit only, no originals stored) │ scrubbed_notes ▼ [2] NOTES PARSER (schema-driven) │ List[SectionNote] — section_ids from schema, not hardcoded ▼ [3] Optional: NOTES EXPANDER (schema-informed shorthand expansion) │ enriched observations ▼ [4] For each SectionNote: │ ├─ RAG SEARCH → FAISS (section_id filter, MASTER tier preferred) │ │ retrieved template paragraphs │ ▼ ├─ MAPPING LLM CALL (schema-aware system prompt) │ │ mapped paragraph text │ ▼ ├─ GROUNDING CHECK (LLM audit) │ │ violations flagged / cleaned │ ▼ └─ FINAL PII CHECK (assert_no_pii regex) │ clean mapped paragraph ▼ [5] REPORT ASSEMBLER Schema section ordering → python-docx Schema rating system → conditional rating display No hardcoded section list, no hardcoded colours │ ▼ DOCX DOWNLOAD (Address, client name, inspection date added manually by surveyor before issue) ``` --- ## SECURITY REQUIREMENTS 1. **All LLM calls use PII-scrubbed text exclusively** — raw notes never reach any API endpoint 2. **Tenant isolation** — each tenant's FAISS index and schema live in a private directory; cross-tenant access is architecturally impossible 3. **Scrubbing audit log** — every redaction recorded with type and count only (never the original value) 4. **Output PII gate** — `assert_no_pii()` runs on every mapped paragraph before it enters the DOCX 5. **Reference tier safety gate** — chunks with `is_scrubbed=False` are excluded from RAG search results 6. **Master template PII guard** — at ingest time, the upload endpoint checks that the uploaded master template is structural boilerplate (no real addresses) by running a lightweight regex scan and warning the user if potential PII is found 7. **Request-level PII middleware** — FastAPI middleware intercepts the `raw_notes` field in every request and runs a lightweight postcode + email regex scan as a first-line check before any processing 8. **API key management** — Anthropic and OpenAI API keys in environment variables only, never logged --- ## TESTING PLAN ### Unit Tests - `test_pii_scrubber.py` - Regex coverage: postcodes, emails, phones, currency, reference numbers, dates - spaCy NER: person names, company names, locations - `assert_no_pii()` on clean text → True; on text with PII → False - `test_template_discoverer.py` - Upload a DOCX with known sections → verify schema has correct section count and order - Upload a template WITH a rating system → verify `rating_system.detected=True` and values correct - Upload a template WITHOUT a rating system → verify `rating_system.detected=False` - Upload a PDF → verify heading extraction produces a usable outline - `test_notes_parser.py` - Notes with explicit section IDs → correct assignment - Notes with only keywords → correct keyword-based assignment - Notes with no recognisable section → lands in UNASSIGNED - Notes with rating values (only tested with schema that has rating system) - `test_rag_store.py` - Ingest MASTER tier document → chunks stored with `is_scrubbed=False` - Ingest REFERENCE tier document → chunks stored with `is_scrubbed=True` - Search with `section_id` filter → MASTER tier sections returned preferentially - Reference chunks with `is_scrubbed=False` never appear in search results - `test_mapping_prompt.py` - Schema WITH rating system → system prompt contains rating instructions - Schema WITHOUT rating system → system prompt contains explicit no-rating instruction - Build messages with rating value present → rating instruction in user message - Build messages with rating value absent → no rating instruction ### Integration Tests - **End-to-end with rating template**: Notes with rating indicators → DOCX contains rating annotations only on sections where the schema says `has_rating_field=True` - **End-to-end without rating template**: Notes passed to a schema with no rating system → DOCX contains zero rating annotations regardless of what shorthand appears in the notes - **New firm onboarding**: Upload a completely different firm's template → schema discovered correctly → report generated using that template's section structure exclusively - **Grounding adversarial**: Inject a defect into notes that has zero matching template paragraph → `UNMATCHED_OBSERVATION` tag appears in output, not invented prose - **PII adversarial**: Embed a postcode and person name in notes → neither appears in DOCX output --- ## CRITICAL IMPLEMENTATION NOTES FOR CURSOR 1. **Build `core/pii_scrubber.py` and `test_pii_scrubber.py` first** with 100% test coverage before touching any other module. This is the security foundation. 2. **Build `core/template_discoverer.py` second** and verify it with at least two different template formats before building the notes parser. The schema must be stable before anything depends on it. 3. **The notes parser has zero hardcoded section codes.** If you find yourself typing a section code like `D1` or `E3` anywhere outside a test fixture, stop — you are hardcoding. The taxonomy comes from the schema exclusively. 4. **The mapping system prompt is built at runtime** by `build_mapping_system_prompt(schema)`. There must be no static SYSTEM_PROMPT string that contains rating rules or section codes. The string is always constructed from the schema. 5. **Rating logic is entirely conditional on `schema.rating_system.detected`.** Search your codebase for "Condition Rating", "CR1", "CR2", "CR3" — if any of these appear as constants or default strings outside test fixtures, they must be removed and replaced with schema-derived values. 6. **The report assembler uses `schema.sections` ordered by `section.order`** for its output order. The `SECTION_ORDER` list from v1 does not exist in v2. 7. **Coloured rating annotations in the DOCX** are applied only when ALL of the following are true: - `schema.rating_system.detected is True` - `section_def.has_rating_field is True` for that specific section - `section_data['rating_value'] is not None` If any of these is false, no colour annotation is added. 8. **The DOCX assembler does NOT add an AI Transparency footer** by default. This was in v1 but is not part of any template — it is an assumption. If the firm wants it, add a config flag `ai_transparency_footer_enabled: bool = False` and only render it when explicitly enabled. 9. **`assert_no_pii()` must gate every paragraph before it enters `build_docx()`.** The assembler should raise a `ValueError` if a paragraph fails the PII check. This forces the caller to re-run `scrub()` and log the incident. 10. **FAISS health check at startup**: `main.py` must include a startup event that checks each tenant's FAISS index exists and has at least one vector. If a generation request arrives and the index is empty, return a clear 400 error: `"Template has not been ingested. Upload a master template first."` 11. **Schema versioning**: If the firm uploads a new master template, the schema and FAISS index are both replaced atomically. Store the previous schema as `schema_prev.json` before overwriting, in case a rollback is needed. 12. **The discovery LLM call uses `max_tokens=4000`** to handle large templates with many sections. If the template has more than 8000 characters, send only the first 8000 of body text but the complete heading outline — headings are compact and carry all the structural information needed. ```