Level: Intermediate
Objective: Master automated information extraction from documents

Document Intelligence Overview
Document Intelligence Studio
Prebuilt Models – Complete Guide
Custom Models
Feedback Loop and Retraining
On-Premises Deployment with Docker Container
API and Python SDK – Implementation
Document Intelligence Solution Architecture
Industrial Use Cases
Governance, Security and Compliance
Summary and Key Points
Glossary

1. Document Intelligence Overview

1.1 What is Document Intelligence?

Azure AI Document Intelligence (formerly Azure Form Recognizer) is an Azure AI service that enables building automated document processing software. It extracts:

Text: Advanced OCR (printed + handwritten)
Key-value pairs: “Total: $45.99”, “Date: 2024-01-15”
Tables: Complete table structures
Entities: Names, dates, amounts, addresses…
Semantic fields: “MerchantName”, “InvoiceDate” (prebuilt models)

flowchart LR
    DOC["📄 Raw Document\n(PDF, image, scan)"] --> DI["Azure AI\nDocument Intelligence"]
    
    DI --> OCR["🔤 OCR Layer\nRaw extracted text"]
    DI --> LAYOUT["📐 Layout Layer\nStructure: tables,\nlines, columns"]
    DI --> SEMANTIC["🧠 Semantic Layer\nField meaning\n(Prebuilt or custom model)"]
    
    SEMANTIC --> OUTPUT["📊 Structured Data\n{\n  merchant: 'Taco House',\n  total: 42.50,\n  date: '2024-01-15'\n}"]
    
    OUTPUT --> APP["Business\nApplication"]
    OUTPUT --> DB["Database"]
    OUTPUT --> WORKFLOW["Automated\nWorkflow"]

1.2 Document Intelligence vs Simple OCR

Capability	Simple OCR	Document Intelligence
Extract text	✅	✅
Locate text	✅ (bounding boxes)	✅ (bounding boxes + polygons)
Understand structure	❌	✅ (tables, forms)
Understand semantics	❌	✅ (“This field = total amount”)
Specialized models	❌	✅ (Receipts, Invoices, ID…)
Custom models	❌	✅ (Your own forms)
Key-value pair extraction	❌	✅
Table extraction	❌	✅ (Complete structure)

1.3 Typical Use Cases

mindmap
  root((Document Intelligence))
    Finance
      Automate invoice entry
      Extract bank statements
      Process expense reports
    HR
      Analyze resumes automatically
      Process hiring forms
      Verify identity documents
    Healthcare
      Extract prescriptions
      Process patient forms
      Analyze medical reports
    Logistics
      Process delivery notes
      Extract shipping labels
      Analyze manifests
    Legal
      Extract contract clauses
      Process notarial deeds
      Analyze legal files
    Commerce
      Digitize product catalogs
      Process purchase orders
      Extract receipt data

2. Document Intelligence Studio

URL: https://documentintelligence.ai.azure.com/studio

Prerequisites:

Active Azure subscription
Azure AI Document Intelligence resource created
Login with Azure credentials

Document Intelligence Studio is the web interface for exploring, testing and training models.

Features:

Test prebuilt models with your documents
Create and manage custom model projects
Annotate (label) documents
Train models
Analyze model performance

2.2 Studio Interface

Main sections:
├── Document analysis
│   ├── Read          → Generic OCR (printed + handwritten text)
│   ├── Layout        → Tables, checkboxes, structure
│   └── General document → Generic key-value pairs
├── Prebuilt models
│   ├── Invoice       → Commercial invoices
│   ├── Receipt       → Cash receipts
│   ├── ID document   → Identity cards, passports
│   ├── Business card → Business cards
│   ├── Tax (W2)      → US tax forms
│   ├── Contract      → Contracts (preview)
│   └── Health...     → Health (preview)
└── Custom models
    ├── Custom extraction → Your own fields
    ├── Custom classifier → Classify document types
    └── Composed model  → Combine multiple models

2.3 Using the Studio to Test a Prebuilt Model

Demo: Analyze a Receipt

1. Open Document Intelligence Studio
2. Select: Prebuilt models → Receipt
3. Configure the resource:
   - Choose your subscription
   - Choose your DI resource
4. Upload a receipt (drag & drop or URL)
5. Click "Run analysis"
6. Observe results:
   - Left panel: Original document with overlays
   - Right panel: Extracted fields with scores
   
Example results:
   MerchantName: "Clint's Taco House"  (98%)
   TransactionDate: "2024-01-15"       (99%)
   SubTotal: "$38.50"                   (97%)
   Tax: "$4.00"                         (95%)
   Total: "$42.50"                      (99%)
   Items:
     - 2x Tacos Al Pastor: $16.00
     - 1x Burrito Supreme: $12.00
     - 3x Soda: $7.50
     - 1x Guac: $3.00

3. Prebuilt Models – Complete Guide

3.1 All Prebuilt Models Table

Model	Documents	Main Fields	Use Case
prebuilt-read	Any text document	Text + language + rotation	Raw text extraction
prebuilt-layout	Any structured document	Tables, checkboxes, columns	Understanding structure
prebuilt-document	Generic forms	Key-value pairs, tables, entities	Non-standardized documents
prebuilt-receipt	Cash receipts	MerchantName, Total, Items, Date	Expense report management
prebuilt-invoice	Commercial invoices	InvoiceId, VendorName, DueDate, TotalAmount	Accounts payable
prebuilt-idDocument	ID cards, Passports	FirstName, LastName, DOB, DocumentNumber	Identity verification
prebuilt-businessCard	Business cards	ContactNames, JobTitles, Emails, Phones	CRM contact management
prebuilt-tax.us.w2	US W-2 form	Employer, Employee, Wages, Taxes	US taxation
prebuilt-healthInsuranceCard.us	US health insurance card	MemberId, PlanName, GroupNumber	US healthcare
prebuilt-contract	Contracts (preview)	Parties, Dates, Clauses	Legal analysis

3.2 Python Implementation – All Prebuilt Models

# Complete analysis with Document Intelligence Prebuilt models
from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.ai.documentintelligence.models import AnalyzeDocumentRequest
from azure.core.credentials import AzureKeyCredential
import os
import json
from pathlib import Path

# Initialization
endpoint = os.environ["AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT"]
key = os.environ["AZURE_DOCUMENT_INTELLIGENCE_KEY"]

di_client = DocumentIntelligenceClient(
    endpoint=endpoint,
    credential=AzureKeyCredential(key)
)

def analyze_document(
    source: str,
    model: str = "prebuilt-document",
    is_url: bool = False,
    pages: str = None
) -> dict:
    """
    Analyzes a document with the specified model.
    
    Args:
        source: Local path or URL of the document
        model: ID of the model to use
        is_url: True if source is a URL
        pages: Page numbers to analyze (e.g., "1-3" or "1, 3")
    
    Returns:
        Complete analysis result
    """
    if is_url:
        request = AnalyzeDocumentRequest(url_source=source)
        poller = di_client.begin_analyze_document(
            model_id=model,
            analyze_request=request,
            pages=pages
        )
    else:
        ext = Path(source).suffix.lower()
        content_types = {
            ".pdf": "application/pdf",
            ".jpg": "image/jpeg",
            ".jpeg": "image/jpeg",
            ".png": "image/png",
            ".tiff": "image/tiff",
            ".bmp": "image/bmp"
        }
        content_type = content_types.get(ext, "application/octet-stream")
        
        with open(source, "rb") as f:
            poller = di_client.begin_analyze_document(
                model_id=model,
                analyze_request=f.read(),
                content_type=content_type,
                pages=pages
            )
    
    return poller.result()

# === READ MODEL: Text Extraction ===
def extract_full_text(source: str, is_url: bool = False) -> dict:
    """Extracts all text from a document using the Read model."""
    result = analyze_document(source, "prebuilt-read", is_url)
    
    text_by_page = {}
    for page in result.pages:
        lines = []
        for line in page.lines:
            lines.append({
                "text": line.content,
                "polygon": line.polygon,
                "span": line.spans[0].offset if line.spans else None
            })
        text_by_page[f"page_{page.page_number}"] = {
            "lines": lines,
            "angle": page.angle,
            "width": page.width,
            "height": page.height,
            "unit": page.unit
        }
    
    return {
        "page_count": len(result.pages),
        "languages": [lang.locale for lang in (result.languages or [])],
        "pages": text_by_page,
        "full_text": " ".join([
            line.content
            for page in result.pages
            for line in page.lines
        ])
    }

# === LAYOUT MODEL: Document Structure ===
def extract_document_structure(source: str, is_url: bool = False) -> dict:
    """Extracts the complete structure (tables, checkboxes, columns)."""
    result = analyze_document(source, "prebuilt-layout", is_url)
    
    tables = []
    for table_idx, table in enumerate(result.tables or []):
        # Create a grid
        grid = [[None] * table.column_count for _ in range(table.row_count)]
        
        for cell in table.cells:
            grid[cell.row_index][cell.column_index] = {
                "content": cell.content,
                "type": cell.kind,  # "columnHeader", "content", etc.
                "rowspan": cell.row_span,
                "colspan": cell.column_span
            }
        
        tables.append({
            "id": table_idx + 1,
            "rows": table.row_count,
            "columns": table.column_count,
            "grid": grid,
            "region": table.bounding_regions[0] if table.bounding_regions else None
        })
    
    # Extract checkboxes
    checkboxes = []
    for page in result.pages:
        for selection in page.selection_marks or []:
            checkboxes.append({
                "state": selection.state,  # "selected" or "unselected"
                "confidence": selection.confidence,
                "polygon": selection.polygon
            })
    
    return {
        "tables": tables,
        "checkboxes": checkboxes,
        "table_count": len(tables),
        "checkbox_count": len(checkboxes)
    }

# === RECEIPT MODEL: Analyze Receipts ===
def analyze_receipt_detail(source: str, is_url: bool = False) -> dict:
    """Extracts all information from a cash receipt."""
    result = analyze_document(source, "prebuilt-receipt", is_url)
    
    receipts = []
    for doc in result.documents:
        receipt = {
            "type": doc.doc_type,
            "confidence": round(doc.confidence, 4),
            "fields": {}
        }
        
        # Receipt fields
        important_fields = [
            "MerchantName", "MerchantAddress", "MerchantPhoneNumber",
            "TransactionDate", "TransactionTime",
            "SubTotal", "TotalTax", "Total", "Tip", "TotalPrice",
            "Currency"
        ]
        
        for field_name in important_fields:
            if field_name in doc.fields:
                field = doc.fields[field_name]
                if field:
                    receipt["fields"][field_name] = {
                        "value": field.value or field.content,
                        "confidence": round(field.confidence or 0, 4),
                        "type": field.type
                    }
        
        # Individual items
        if "Items" in doc.fields and doc.fields["Items"]:
            items = []
            for item_field in (doc.fields["Items"].value or []):
                item = {}
                for item_name, item_value in (item_field.value or {}).items():
                    if item_value and (item_value.value or item_value.content):
                        item[item_name] = item_value.value or item_value.content
                items.append(item)
            receipt["items"] = items
        
        receipts.append(receipt)
    
    return {"receipts": receipts, "count": len(receipts)}

# === INVOICE MODEL: Analyze Invoices ===
def analyze_invoice_detail(source: str, is_url: bool = False) -> dict:
    """Extracts all information from a commercial invoice."""
    result = analyze_document(source, "prebuilt-invoice", is_url)
    
    invoices = []
    for doc in result.documents:
        invoice = {"confidence": round(doc.confidence, 4), "fields": {}}
        
        invoice_fields = [
            "InvoiceId", "InvoiceDate", "DueDate", "PurchaseOrder",
            "VendorName", "VendorAddress", "VendorAddressRecipient",
            "CustomerName", "CustomerAddress", "CustomerAddressRecipient",
            "SubTotal", "TotalTax", "FreightAmount", "TotalAmount",
            "AmountDue", "ServiceStartDate", "ServiceEndDate"
        ]
        
        for field_name in invoice_fields:
            if field_name in doc.fields:
                field = doc.fields[field_name]
                if field and (field.value or field.content):
                    invoice["fields"][field_name] = str(field.value or field.content)
        
        # Invoice line items
        if "Items" in doc.fields and doc.fields["Items"]:
            line_items = []
            for item in (doc.fields["Items"].value or []):
                line = {}
                for name, value in (item.value or {}).items():
                    if value and (value.value or value.content):
                        line[name] = str(value.value or value.content)
                line_items.append(line)
            invoice["line_items"] = line_items
        
        invoices.append(invoice)
    
    return {"invoices": invoices}

# === ID DOCUMENT MODEL ===
def analyze_identity_document(source: str, is_url: bool = False) -> dict:
    """Extracts information from an identity card or passport."""
    result = analyze_document(source, "prebuilt-idDocument", is_url)
    
    documents = []
    for doc in result.documents:
        id_doc = {
            "type": doc.doc_type,  # "idDocument.passport", "idDocument.driverLicense", etc.
            "confidence": round(doc.confidence, 4),
            "fields": {}
        }
        
        id_fields = [
            "FirstName", "LastName", "MiddleName",
            "DocumentNumber", "DateOfBirth", "DateOfExpiration",
            "Sex", "Nationality", "CountryRegion",
            "Address", "PlaceOfBirth",
            "MachineReadableZone"
        ]
        
        for field_name in id_fields:
            if field_name in doc.fields:
                field = doc.fields[field_name]
                if field and (field.value or field.content):
                    id_doc["fields"][field_name] = {
                        "value": str(field.value or field.content),
                        "confidence": round(field.confidence or 0, 4)
                    }
        
        documents.append(id_doc)
    
    return {"identity_documents": documents}

# === Demonstrations ===
print("=== Testing Prebuilt Models ===\n")

# Test Receipt model
print("1. Analyze a receipt:")
receipt_result = analyze_receipt_detail("restaurant_receipt.jpg")
for receipt in receipt_result["receipts"]:
    fields = receipt["fields"]
    print(f"   Merchant: {fields.get('MerchantName', {}).get('value', 'N/A')}")
    print(f"   Total: {fields.get('Total', {}).get('value', 'N/A')}")
    print(f"   Date: {fields.get('TransactionDate', {}).get('value', 'N/A')}")
    if "items" in receipt:
        print(f"   Items: {len(receipt['items'])} lines")

print("\n2. Analyze an invoice:")
invoice_result = analyze_invoice_detail("supplier_invoice.pdf")
for invoice in invoice_result["invoices"]:
    fields = invoice["fields"]
    print(f"   Vendor: {fields.get('VendorName', 'N/A')}")
    print(f"   Number: {fields.get('InvoiceId', 'N/A')}")
    print(f"   Total: {fields.get('TotalAmount', 'N/A')}")
    print(f"   Due Date: {fields.get('DueDate', 'N/A')}")

4. Custom Models

4.1 Why Create a Custom Model?

Prebuilt models are excellent for standardized document types. But your company probably has specific forms:

Internal purchase orders
Field visit reports
Quality forms
Specialized industry documents

flowchart TD
    Q{"Is the document\nstandardized?"}
    Q -->|Yes - Receipt, Invoice, ID| PREBUILT["✅ Prebuilt Model\n(Immediate, no training)"]
    Q -->|No - Specific form| CUSTOM["⚙️ Custom Model\n(Training required)"]
    
    CUSTOM --> C1["Custom Extraction\n(Extract specific fields)"]
    CUSTOM --> C2["Custom Classifier\n(Identify document type)"]
    CUSTOM --> C3["Composed Model\n(Combine multiple models)"]

4.2 Types of Custom Models

Type	Description	Usage	Minimum
Custom Extraction	Extract specific fields	Your internal forms	5 documents
Custom Classifier	Classify document type	Automatic mail sorting	5 doc/class
Composed Model	Combine multiple extractors	Form portfolio	N/A
Neural (preferred)	More accurate neural model	Variable layouts	5+ documents
Template	Template-based	Highly standardized forms	5+ documents

4.3 Preparing Training Data

File structure for training:

model_training/
├── document_01.pdf         # Raw document
├── document_01.labels.json # Field annotations
├── document_01.ocr.json    # Pre-calculated OCR from the service
├── document_02.pdf
├── document_02.labels.json
├── document_02.ocr.json
├── ...
└── fields.json             # Definition of fields to extract

Example labels.json file:

{
  "document": "document_01.pdf",
  "labels": [
    {
      "label": "InvoiceNumber",
      "value": [
        {
          "page": 1,
          "text": "INV-2024-001",
          "boundingBoxes": [[0.1, 0.05, 0.4, 0.05, 0.4, 0.08, 0.1, 0.08]]
        }
      ]
    },
    {
      "label": "TotalAmount",
      "value": [
        {
          "page": 1,
          "text": "1500.00",
          "boundingBoxes": [[0.7, 0.85, 0.9, 0.85, 0.9, 0.88, 0.7, 0.88]]
        }
      ]
    }
  ]
}

fields.json — field schema:

{
  "fields": {
    "EmployeeId": {
      "fieldKey": "EmployeeId",
      "fieldType": "string",
      "fieldFormat": "not-specified"
    },
    "FullName": {
      "fieldKey": "FullName",
      "fieldType": "string",
      "fieldFormat": "not-specified"
    },
    "HireDate": {
      "fieldKey": "HireDate",
      "fieldType": "date",
      "fieldFormat": "not-specified"
    },
    "Salary": {
      "fieldKey": "Salary",
      "fieldType": "number",
      "fieldFormat": "currency"
    }
  }
}

4.4 Training a Custom Model via the SDK

# Train a custom model with Azure AI Document Intelligence
from azure.ai.documentintelligence import DocumentIntelligenceAdministrationClient
from azure.ai.documentintelligence.models import (
    BuildDocumentModelRequest,
    AzureBlobContentSource,
    DocumentBuildMode
)
from azure.core.credentials import AzureKeyCredential
import os
import time
import json

admin_client = DocumentIntelligenceAdministrationClient(
    endpoint=os.environ["AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT"],
    credential=AzureKeyCredential(os.environ["AZURE_DOCUMENT_INTELLIGENCE_KEY"])
)

def train_custom_model(
    storage_account_url: str,
    container_name: str,
    model_id: str,
    description: str = "",
    mode: str = "neural"
) -> str:
    """
    Trains a custom Document Intelligence model.
    
    Args:
        storage_account_url: Azure storage account URL
        container_name: Container name with training data
        model_id: Unique ID for the model
        description: Model description
        mode: "neural" (recommended) or "template"
    
    Returns:
        Trained model ID
    """
    blob_source = AzureBlobContentSource(
        container_url=f"{storage_account_url}/{container_name}"
    )
    
    build_request = BuildDocumentModelRequest(
        model_id=model_id,
        description=description,
        build_mode=DocumentBuildMode.NEURAL if mode == "neural" else DocumentBuildMode.TEMPLATE,
        azure_blob_source=blob_source,
        tags={
            "environment": "production",
            "team": "document-processing",
            "version": "1.0"
        }
    )
    
    print(f"Starting training for model '{model_id}'...")
    print(f"Mode: {mode}")
    print(f"Source: {storage_account_url}/{container_name}")
    
    poller = admin_client.begin_build_document_model(build_request)
    
    # Polling with progress display
    start_time = time.time()
    while not poller.done():
        elapsed = (time.time() - start_time) / 60
        print(f"  Training in progress... {elapsed:.1f} min")
        time.sleep(10)
    
    model = poller.result()
    
    print(f"\n✅ Model trained successfully!")
    print(f"   ID: {model.model_id}")
    print(f"   Date: {model.created_date_time}")
    print(f"   Fields: {list(model.doc_types.values())[0].field_schema.keys() if model.doc_types else 'N/A'}")
    
    return model.model_id

def use_custom_model(
    source: str,
    model_id: str,
    is_url: bool = False
) -> dict:
    """
    Uses a custom model to analyze a document.
    
    Args:
        source: Document path or URL
        model_id: Trained custom model ID
    
    Returns:
        Extracted fields with confidence scores
    """
    if is_url:
        request = AnalyzeDocumentRequest(url_source=source)
        poller = di_client.begin_analyze_document(
            model_id=model_id,
            analyze_request=request
        )
    else:
        with open(source, "rb") as f:
            poller = di_client.begin_analyze_document(
                model_id=model_id,
                analyze_request=f.read(),
                content_type="application/pdf"
            )
    
    result = poller.result()
    
    extracted_fields = {}
    for doc in result.documents:
        for field_name, field in doc.fields.items():
            if field and (field.value or field.content):
                extracted_fields[field_name] = {
                    "value": str(field.value or field.content),
                    "confidence": round(field.confidence or 0, 4),
                    "type": field.type
                }
    
    return {
        "model_id": model_id,
        "document_count": len(result.documents),
        "fields": extracted_fields
    }

def list_models() -> list[dict]:
    """Lists all available models (prebuilt + custom)."""
    models = []
    
    for model in admin_client.list_document_models():
        models.append({
            "id": model.model_id,
            "description": model.description or "N/A",
            "created_date": str(model.created_date_time)[:10],
            "type": "prebuilt" if model.model_id.startswith("prebuilt") else "custom"
        })
    
    return sorted(models, key=lambda m: (m["type"], m["id"]))

# Train a model for purchase order forms
model_id = train_custom_model(
    storage_account_url="https://myaccount.blob.core.windows.net",
    container_name="training-data-purchase-orders",
    model_id="purchase-order-model-v1",
    description="Model to extract data from internal purchase order forms",
    mode="neural"
)

# Test the model
print("\n=== Testing the custom model ===")
results = use_custom_model(
    source="new_purchase_order.pdf",
    model_id=model_id
)

print(f"Extracted fields ({len(results['fields'])} total):")
for field, info in results["fields"].items():
    stars = "★" * int(info["confidence"] * 5)
    print(f"  {field:25}: {info['value']:20} ({info['confidence']:.0%}) {stars}")

# Display all models
print("\n=== Available Models ===")
for model in list_models():
    print(f"  [{model['type']:8}] {model['id']} - {model['description'][:50]}")

5. Feedback Loop and Retraining

5.1 Why is Feedback Important?

A custom model can make mistakes, especially on atypical documents. The feedback loop enables continuously improving accuracy.

flowchart TD
    DOC["📄 New document"] --> MODEL["🧠 Model v1.0\n(Automatic extraction)"]
    MODEL --> PRED["Predictions\n{InvoiceNumber: 'INV-001', Confidence: 0.62}"]
    PRED --> REVIEW["👤 Human review\n(Verify low-confidence\nextractions)"]
    REVIEW -->|"Correction"| CORRECTION["📝 Error correction\n(Modify incorrect values)"]
    CORRECTION --> RELABEL["🏷️ Re-annotation\n(Add corrected document\nto training set)"]
    RELABEL --> RETRAIN["🔄 Retrain model\n(v1.0 → v1.1)"]
    RETRAIN --> MODEL_V2["🧠 Model v1.1\n(More accurate)"]
    MODEL_V2 --> DOC

5.2 Feedback Loop Implementation

# Feedback and retraining system
import json
import os
import time
from datetime import datetime
from pathlib import Path
from azure.ai.documentintelligence import (
    DocumentIntelligenceClient,
    DocumentIntelligenceAdministrationClient
)
from azure.core.credentials import AzureKeyCredential

class DocumentIntelligenceFeedbackSystem:
    """
    Feedback system for improving a Document Intelligence model.
    
    Workflow:
    1. Analyze the document with the current model
    2. Identify low-confidence extractions
    3. Request human correction
    4. Store corrections
    5. Retrain the model with the new data
    """
    
    LOW_CONFIDENCE_THRESHOLD = 0.85  # Below this = human review
    
    def __init__(self, model_id: str, feedback_folder: str = "./feedback"):
        self.model_id = model_id
        self.feedback_folder = Path(feedback_folder)
        self.feedback_folder.mkdir(exist_ok=True)
        
        self.di_client = DocumentIntelligenceClient(
            endpoint=os.environ["AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT"],
            credential=AzureKeyCredential(os.environ["AZURE_DOCUMENT_INTELLIGENCE_KEY"])
        )
        
        self.admin_client = DocumentIntelligenceAdministrationClient(
            endpoint=os.environ["AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT"],
            credential=AzureKeyCredential(os.environ["AZURE_DOCUMENT_INTELLIGENCE_KEY"])
        )
    
    def analyze_and_identify_uncertainties(
        self, document_path: str
    ) -> dict:
        """
        Analyzes a document and identifies fields requiring review.
        
        Returns:
            Dict with confident fields + fields to review
        """
        with open(document_path, "rb") as f:
            poller = self.di_client.begin_analyze_document(
                model_id=self.model_id,
                analyze_request=f.read(),
                content_type="application/pdf"
            )
        
        result = poller.result()
        
        confident_fields = {}
        review_fields = {}
        
        if result.documents:
            for field_name, field in result.documents[0].fields.items():
                if field:
                    confidence = field.confidence or 0
                    value = str(field.value or field.content or "")
                    
                    info = {
                        "value": value,
                        "confidence": round(confidence, 4),
                        "corrected_value": None
                    }
                    
                    if confidence >= self.LOW_CONFIDENCE_THRESHOLD:
                        confident_fields[field_name] = info
                    else:
                        review_fields[field_name] = info
        
        return {
            "document": document_path,
            "model_version": self.model_id,
            "timestamp": datetime.now().isoformat(),
            "confident_fields": confident_fields,
            "fields_to_review": review_fields,
            "review_required": len(review_fields) > 0
        }
    
    def apply_human_corrections(
        self,
        analysis: dict,
        corrections: dict
    ) -> dict:
        """
        Applies human corrections to low-confidence fields.
        
        Args:
            analysis: Analysis result
            corrections: Dict {field_name: correct_value}
        
        Returns:
            Updated analysis with corrections
        """
        for field_name, correct_value in corrections.items():
            if field_name in analysis["fields_to_review"]:
                analysis["fields_to_review"][field_name]["corrected_value"] = correct_value
                print(f"  ✅ Corrected: {field_name} = '{correct_value}'")
            elif field_name in analysis["confident_fields"]:
                # Correction of a previously confident field
                analysis["confident_fields"][field_name]["corrected_value"] = correct_value
                print(f"  ✅ Corrected (high confidence): {field_name} = '{correct_value}'")
        
        return analysis
    
    def save_feedback(self, corrected_analysis: dict) -> str:
        """
        Saves feedback for future retraining.
        
        Returns:
            Path to the feedback file
        """
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"feedback_{timestamp}.json"
        feedback_path = self.feedback_folder / filename
        
        with open(feedback_path, "w", encoding="utf-8") as f:
            json.dump(corrected_analysis, f, ensure_ascii=False, indent=2)
        
        print(f"✅ Feedback saved: {feedback_path}")
        return str(feedback_path)
    
    def generate_feedback_statistics(self) -> dict:
        """Analyzes feedback files to identify recurring issues."""
        stats = {
            "total_documents": 0,
            "total_corrections": 0,
            "most_corrected_fields": {},
            "error_rate_by_field": {}
        }
        
        feedbacks = list(self.feedback_folder.glob("feedback_*.json"))
        stats["total_documents"] = len(feedbacks)
        
        for feedback_path in feedbacks:
            with open(feedback_path) as f:
                fb = json.load(f)
            
            # Count corrections
            for fields in [fb.get("confident_fields", {}), fb.get("fields_to_review", {})]:
                for field_name, info in fields.items():
                    if info.get("corrected_value"):
                        stats["total_corrections"] += 1
                        stats["most_corrected_fields"][field_name] = (
                            stats["most_corrected_fields"].get(field_name, 0) + 1
                        )
        
        # Calculate error rates
        if stats["total_documents"] > 0:
            for field, correction_count in stats["most_corrected_fields"].items():
                stats["error_rate_by_field"][field] = round(
                    correction_count / stats["total_documents"], 3
                )
        
        return stats
    
    def prepare_retraining_data(
        self,
        output_folder: str,
        feedback_threshold: int = 10
    ) -> bool:
        """
        Prepares retraining data from feedback.
        
        Args:
            output_folder: Folder to save new training data
            feedback_threshold: Minimum number of feedbacks to trigger retraining
        
        Returns:
            True if data prepared, False if not enough feedback
        """
        feedbacks = list(self.feedback_folder.glob("feedback_*.json"))
        
        if len(feedbacks) < feedback_threshold:
            print(f"⚠️ Not enough feedback: {len(feedbacks)}/{feedback_threshold}")
            return False
        
        # Transform feedback into Document Intelligence training-formatted data
        print(f"✅ {len(feedbacks)} feedbacks → Training data prepared")
        print(f"   Folder: {output_folder}")
        
        return True

# Feedback loop demonstration
system = DocumentIntelligenceFeedbackSystem(
    model_id="purchase-order-model-v1",
    feedback_folder="./feedback_data"
)

# 1. Analyze a new document
print("=== Analysis of a new purchase order ===")
analysis = system.analyze_and_identify_uncertainties("po_2024_0042.pdf")

print(f"\nConfident fields ({len(analysis['confident_fields'])} fields):")
for field, info in analysis["confident_fields"].items():
    print(f"  ✅ {field}: '{info['value']}' ({info['confidence']:.0%})")

print(f"\nFields to review ({len(analysis['fields_to_review'])} fields):")
for field, info in analysis["fields_to_review"].items():
    print(f"  ⚠️ {field}: '{info['value']}' ({info['confidence']:.0%}) ← CHECK")

# 2. Human corrections
if analysis["review_required"]:
    print("\n=== Human Corrections ===")
    corrections = {
        "OrderNumber": "PO-2024-0042",   # Corrected from "PO2024-0042"
        "SubTotal": "1850.00"             # Corrected from "1 850,00"
    }
    
    corrected_analysis = system.apply_human_corrections(analysis, corrections)
    feedback_file = system.save_feedback(corrected_analysis)

# 3. Statistics after 10 feedbacks
stats = system.generate_feedback_statistics()
print("\n=== Feedback Statistics ===")
print(f"Documents processed: {stats['total_documents']}")
print(f"Total corrections: {stats['total_corrections']}")
print("\nMost corrected fields:")
for field, count in sorted(stats["most_corrected_fields"].items(), 
                            key=lambda x: x[1], reverse=True):
    print(f"  {field}: {count} corrections ({stats['error_rate_by_field'].get(field, 0):.0%})")

6. On-Premises Deployment with Docker Container

6.1 Why Use a Container?

Document Intelligence containers allow running the service within your own infrastructure:

GDPR compliance: Data does not leave your network
Reduced latency: Local processing, no round-trip to Azure
Offline availability: Works even without internet
Cost: Can be more economical for large volumes

6.2 Available Containers

Container	Description	Min RAM	Recommended RAM
read	OCR only	8 GB	16 GB
layout	Structure analysis	8 GB	24 GB
general-document	Key-value pairs	8 GB	24 GB
invoice	Invoice model	8 GB	24 GB
receipt	Receipt model	8 GB	24 GB
id-document	ID card model	8 GB	24 GB

6.3 Deploying the Read Container

# Start the Document Intelligence container (Read model)
docker run \
  --rm \
  -it \
  -p 5000:5000 \
  --memory=8g \
  --cpus=4 \
  --name document-intelligence-read \
  mcr.microsoft.com/azure-cognitive-services/form-recognizer/read:latest \
  Eula=accept \
  Billing=https://your-resource.cognitiveservices.azure.com/ \
  ApiKey=YOUR_API_KEY

# Verify the container is active
curl http://localhost:5000/ready
# Expected response: "Healthy"

# View the container API documentation
# Open in a browser: http://localhost:5000/swagger

6.4 Using the Local Container via Python

# Use Document Intelligence on the local container
from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.core.credentials import AzureKeyCredential

# Connect to the LOCAL container (not Azure)
local_client = DocumentIntelligenceClient(
    endpoint="http://localhost:5000",  # Local container port
    credential=AzureKeyCredential("apikey")  # Any non-empty value
)

# Analyze a document with the local container
with open("local_document.pdf", "rb") as f:
    poller = local_client.begin_analyze_document(
        model_id="prebuilt-read",
        analyze_request=f.read(),
        content_type="application/pdf"
    )

result = poller.result()

print("=== Local Container Result ===")
for page in result.pages:
    print(f"Page {page.page_number}:")
    for line in page.lines:
        print(f"  {line.content}")

6.5 Docker Compose for Multi-Container Deployment

# docker-compose.yml - Multi-model Document Intelligence deployment
version: '3.8'

services:
  di-read:
    image: mcr.microsoft.com/azure-cognitive-services/form-recognizer/read:latest
    container_name: di-read
    ports:
      - "5000:5000"
    environment:
      - Eula=accept
      - Billing=${AZURE_DI_ENDPOINT}
      - ApiKey=${AZURE_DI_KEY}
    deploy:
      resources:
        limits:
          memory: 8G
          cpus: '2'
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:5000/ready"]
      interval: 30s
      timeout: 10s
      retries: 3
  
  di-layout:
    image: mcr.microsoft.com/azure-cognitive-services/form-recognizer/layout:latest
    container_name: di-layout
    ports:
      - "5001:5000"
    environment:
      - Eula=accept
      - Billing=${AZURE_DI_ENDPOINT}
      - ApiKey=${AZURE_DI_KEY}
    deploy:
      resources:
        limits:
          memory: 16G
          cpus: '4'
  
  di-invoice:
    image: mcr.microsoft.com/azure-cognitive-services/form-recognizer/invoice:latest
    container_name: di-invoice
    ports:
      - "5002:5000"
    environment:
      - Eula=accept
      - Billing=${AZURE_DI_ENDPOINT}
      - ApiKey=${AZURE_DI_KEY}
    deploy:
      resources:
        limits:
          memory: 16G
          cpus: '4'
  
  # Proxy to route to the correct container by model
  nginx-proxy:
    image: nginx:latest
    container_name: di-proxy
    ports:
      - "8080:80"
    volumes:
      - ./nginx.conf:/etc/nginx/conf.d/default.conf
    depends_on:
      - di-read
      - di-layout
      - di-invoice

# nginx.conf - Routing to DI containers
upstream di-read-backend { server di-read:5000; }
upstream di-layout-backend { server di-layout:5000; }
upstream di-invoice-backend { server di-invoice:5000; }

server {
    listen 80;
    
    location /read/ {
        proxy_pass http://di-read-backend/;
    }
    
    location /layout/ {
        proxy_pass http://di-layout-backend/;
    }
    
    location /invoice/ {
        proxy_pass http://di-invoice-backend/;
    }
}

7. API and Python SDK – Implementation

7.1 Complete Configuration

# Document Intelligence configuration and utilities
import os
import json
import base64
import requests
import time
from pathlib import Path
from typing import Optional, Union
from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.ai.documentintelligence.models import AnalyzeDocumentRequest
from azure.core.credentials import AzureKeyCredential
from azure.identity import DefaultAzureCredential

class DocumentIntelligenceClientWrapper:
    """Document Intelligence client with error handling and retry."""
    
    def __init__(self, use_managed_identity: bool = False):
        endpoint = os.environ["AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT"]
        
        if use_managed_identity:
            credential = DefaultAzureCredential()
        else:
            credential = AzureKeyCredential(
                os.environ["AZURE_DOCUMENT_INTELLIGENCE_KEY"]
            )
        
        self.client = DocumentIntelligenceClient(
            endpoint=endpoint,
            credential=credential
        )
    
    def analyze(
        self,
        source: Union[str, bytes],
        model: str,
        is_url: bool = False,
        pages: Optional[str] = None,
        features: Optional[list] = None
    ) -> dict:
        """
        Analyzes a document with automatic retry.
        
        Args:
            source: Path, URL or bytes of the document
            model: ID of the model to use
            is_url: True if source is a URL
            pages: Pages to analyze ("1-3" or "1, 3")
            features: Additional features ["keyValuePairs", "tables"]
        """
        from azure.core.exceptions import HttpResponseError
        
        max_retries = 3
        for attempt in range(max_retries):
            try:
                if is_url:
                    request = AnalyzeDocumentRequest(url_source=source)
                    poller = self.client.begin_analyze_document(
                        model_id=model,
                        analyze_request=request,
                        pages=pages
                    )
                elif isinstance(source, bytes):
                    poller = self.client.begin_analyze_document(
                        model_id=model,
                        analyze_request=source,
                        content_type="application/pdf",
                        pages=pages
                    )
                else:
                    ext = Path(source).suffix.lower()
                    content_types = {
                        ".pdf": "application/pdf",
                        ".jpg": "image/jpeg",
                        ".jpeg": "image/jpeg",
                        ".png": "image/png",
                        ".tiff": "image/tiff"
                    }
                    with open(source, "rb") as f:
                        poller = self.client.begin_analyze_document(
                            model_id=model,
                            analyze_request=f.read(),
                            content_type=content_types.get(ext, "application/octet-stream"),
                            pages=pages
                        )
                
                return poller.result()
            
            except HttpResponseError as e:
                if e.status_code == 429 and attempt < max_retries - 1:
                    wait_time = 2 ** attempt * 5
                    print(f"  ⏳ Rate limit, waiting {wait_time}s...")
                    time.sleep(wait_time)
                else:
                    raise
        
        raise Exception(f"Failed after {max_retries} attempts")
    
    def extract_normalized_fields(self, result, wanted_fields: list) -> dict:
        """
        Extracts fields from a DI result and returns a normalized dict.
        """
        extraction = {}
        
        if not result.documents:
            return extraction
        
        for doc in result.documents:
            for field_name in wanted_fields:
                if field_name in doc.fields:
                    field = doc.fields[field_name]
                    if field:
                        extraction[field_name] = {
                            "value": str(field.value or field.content or ""),
                            "confidence": round(field.confidence or 0, 4),
                            "type": field.type
                        }
        
        return extraction

# Advanced usage
client = DocumentIntelligenceClientWrapper()

# Analyze multiple pages of a document
result_pages = client.analyze(
    source="annual_report.pdf",
    model="prebuilt-document",
    pages="1-5"
)

# Extract specific fields
wanted_fields = ["VendorName", "InvoiceId", "TotalAmount", "DueDate"]
extraction = client.extract_normalized_fields(
    client.analyze("invoice.pdf", "prebuilt-invoice"),
    wanted_fields
)

print("Extracted fields:")
for field, info in extraction.items():
    confidence_emoji = "✅" if info["confidence"] > 0.9 else "⚠️" if info["confidence"] > 0.7 else "❌"
    print(f"  {confidence_emoji} {field:20}: {info['value']:20} ({info['confidence']:.0%})")

Two-phase asynchronous call pattern (REST API):

import requests
import json

# Analyze a receipt via REST
response = requests.post(
    "https://{resource}.cognitiveservices.azure.com/formrecognizer/documentModels/prebuilt-receipt:analyze?api-version=2023-07-31",
    headers={
        "Ocp-Apim-Subscription-Key": "{your-key}",
        "Content-Type": "application/pdf"
    },
    data=open("receipt.pdf", "rb")
)

# Retrieve the operation ID
operation_id = response.headers["Operation-Location"]
print(f"Operation: {operation_id}")

# Poll for result
while True:
    status_response = requests.get(operation_id,
        headers={"Ocp-Apim-Subscription-Key": "{your-key}"})
    
    status = status_response.json().get("status")
    if status == "succeeded":
        result = status_response.json()["analyzeResult"]
        break
    elif status == "failed":
        raise Exception("Analysis failed")
    time.sleep(2)

# Access extracted tables
for table in result.get("tables", []):
    print(f"Table: {table['rowCount']} rows x {table['columnCount']} columns")
    for cell in table["cells"]:
        print(f"  [{cell['rowIndex']},{cell['columnIndex']}] = {cell['content']}")

8. Document Intelligence Solution Architecture

8.1 Production Document Processing Architecture

flowchart TB
    subgraph "Ingestion"
        EMAIL["📧 Email Scanner\n(Exchange, Gmail)"]
        PORTAL["🌐 Upload Portal\n(Web/Mobile)"]
        SCANNER["🖨️ Network Scanner\n(TWAIN, WIA)"]
        STORAGE["📁 Azure Blob Storage\n(Drop zone)"]
    end
    
    subgraph "DI Processing"
        FUNC["⚡ Azure Function\n(Blob Trigger)"]
        CLASSIFIER["🏷️ Classifier\n(What type of doc?)"]
        
        PREBUILT["📋 Prebuilt Models\n(Receipt, Invoice, ID...)"]
        CUSTOM["⚙️ Custom Models\n(Specific forms)"]
        
        FUNC --> CLASSIFIER
        CLASSIFIER -->|"Known type"| PREBUILT
        CLASSIFIER -->|"Custom type"| CUSTOM
    end
    
    subgraph "Post-processing"
        VALIDATE["✅ Validation\n(Business rules)"]
        FEEDBACK_Q["📊 Feedback queue\n(Low confidence)"]
        EXPORT["📤 Export"]
    end
    
    subgraph "Destinations"
        ERP["ERP / SAP"]
        CRM["CRM / Salesforce"]
        DB_COSMOS["Azure Cosmos DB"]
        POWERBI["Power BI"]
    end
    
    EMAIL --> STORAGE
    PORTAL --> STORAGE
    SCANNER --> STORAGE
    
    STORAGE --> FUNC
    
    PREBUILT --> VALIDATE
    CUSTOM --> VALIDATE
    
    VALIDATE -->|"Confidence OK"| EXPORT
    VALIDATE -->|"Low confidence"| FEEDBACK_Q
    
    EXPORT --> ERP
    EXPORT --> CRM
    EXPORT --> DB_COSMOS
    DB_COSMOS --> POWERBI
    
    FEEDBACK_Q --> REVIEW["👤 Human review"]
    REVIEW --> RETRAIN["🔄 Retraining"]

8.2 Full Automation with Azure Functions

# Azure Function for automatic document processing
import azure.functions as func
import logging
import json
import os
from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.core.credentials import AzureKeyCredential
from azure.storage.blob import BlobServiceClient
from azure.servicebus import ServiceBusClient, ServiceBusMessage

app = func.FunctionApp()

@app.blob_trigger(
    arg_name="blob",
    path="incoming-documents/{name}",
    connection="STORAGE_CONNECTION"
)
def process_uploaded_document(blob: func.InputStream):
    """
    Triggered automatically when a document is uploaded to the container.
    
    Workflow:
    1. Detect document type
    2. Extract data with the appropriate model
    3. Validate data
    4. Route to the appropriate system
    """
    filename = blob.name
    logging.info(f"📄 Processing: {filename}")
    
    doc_data = blob.read()
    
    # 1. Initialize the DI client
    di_client = DocumentIntelligenceClient(
        endpoint=os.environ["AZURE_DI_ENDPOINT"],
        credential=AzureKeyCredential(os.environ["AZURE_DI_KEY"])
    )
    
    # 2. Classify the document
    doc_type = classify_document(di_client, doc_data, filename)
    logging.info(f"Detected type: {doc_type}")
    
    # 3. Extract based on type
    if doc_type == "invoice":
        data = extract_invoice(di_client, doc_data)
        route_to_erp(filename, data)
    
    elif doc_type == "receipt":
        data = extract_receipt(di_client, doc_data)
        route_to_expense_management(filename, data)
    
    elif doc_type == "identity":
        data = extract_identity(di_client, doc_data)
        route_to_hr(filename, data)
    
    else:
        logging.warning(f"Unknown type: {doc_type} → Manual review queue")
        route_to_manual_review(filename, doc_data, doc_type)
    
    logging.info(f"✅ {filename} processed successfully")

def classify_document(di_client, doc_data: bytes, filename: str) -> str:
    """Classifies the document type based on its content."""
    
    # Use the Read model to read the text
    poller = di_client.begin_analyze_document(
        model_id="prebuilt-read",
        analyze_request=doc_data,
        content_type="application/pdf"
    )
    result = poller.result()
    
    text = " ".join([
        line.content
        for page in result.pages
        for line in page.lines
    ]).lower()
    
    # Keyword-based classification
    if any(m in text for m in ["invoice", "bill", "payment due", "amount due"]):
        return "invoice"
    elif any(m in text for m in ["receipt", "total", "subtotal", "thank you for"]):
        return "receipt"
    elif any(m in text for m in ["passport", "national id", "drivers license"]):
        return "identity"
    elif any(m in text for m in ["contract", "agreement", "parties agree"]):
        return "contract"
    else:
        return "unknown"

def extract_invoice(di_client, doc_data: bytes) -> dict:
    """Extracts data from an invoice."""
    poller = di_client.begin_analyze_document(
        model_id="prebuilt-invoice",
        analyze_request=doc_data,
        content_type="application/pdf"
    )
    result = poller.result()
    
    data = {}
    if result.documents:
        for field, value in result.documents[0].fields.items():
            if value and (value.value or value.content):
                data[field] = {
                    "value": str(value.value or value.content),
                    "confidence": round(value.confidence or 0, 4)
                }
    
    return data

def route_to_erp(filename: str, data: dict) -> None:
    """Sends invoice data to the ERP."""
    servicebus_client = ServiceBusClient.from_connection_string(
        os.environ["SERVICE_BUS_CONNECTION"]
    )
    
    message_data = {
        "source": filename,
        "type": "invoice",
        "data": data,
        "timestamp": str(func.utcnow())
    }
    
    with servicebus_client:
        sender = servicebus_client.get_queue_sender("erp-invoices")
        message = ServiceBusMessage(json.dumps(message_data, ensure_ascii=False))
        sender.send_messages(message)
    
    logging.info(f"✅ Invoice sent to ERP: {data.get('InvoiceId', {}).get('value', 'N/A')}")

9. Industrial Use Cases

9.1 Expense Report Processing

# Complete expense report processing solution
from dataclasses import dataclass
from typing import Optional
from datetime import date
import re

@dataclass
class ExpenseReport:
    """Represents an automatically extracted expense report."""
    employee_id: str
    merchant: Optional[str]
    transaction_date: Optional[date]
    total_amount: Optional[float]
    currency: str = "USD"
    category: Optional[str] = None
    items: list = None
    overall_confidence: float = 0.0
    requires_validation: bool = False
    
    def __post_init__(self):
        if self.items is None:
            self.items = []

def process_expense_report(
    receipt_path: str,
    employee_id: str,
    di_client: DocumentIntelligenceClient
) -> ExpenseReport:
    """
    Processes a receipt and automatically creates an expense report.
    """
    with open(receipt_path, "rb") as f:
        poller = di_client.begin_analyze_document(
            model_id="prebuilt-receipt",
            analyze_request=f.read(),
            content_type="image/jpeg"
        )
    
    result = poller.result()
    
    report = ExpenseReport(employee_id=employee_id)
    
    if not result.documents:
        report.requires_validation = True
        return report
    
    doc = result.documents[0]
    confidences = []
    
    # Merchant
    if "MerchantName" in doc.fields and doc.fields["MerchantName"]:
        field = doc.fields["MerchantName"]
        report.merchant = str(field.value or field.content or "")
        confidences.append(field.confidence or 0)
    
    # Date
    if "TransactionDate" in doc.fields and doc.fields["TransactionDate"]:
        field = doc.fields["TransactionDate"]
        if field.value:
            report.transaction_date = field.value
        confidences.append(field.confidence or 0)
    
    # Total
    if "Total" in doc.fields and doc.fields["Total"]:
        field = doc.fields["Total"]
        value_str = str(field.value or field.content or "0")
        # Clean the amount (remove $, USD, etc.)
        amount_num = re.sub(r'[^\d.,]', '', value_str)
        amount_num = amount_num.replace(',', '.')
        try:
            report.total_amount = float(amount_num)
        except ValueError:
            pass
        confidences.append(field.confidence or 0)
    
    # Items
    if "Items" in doc.fields and doc.fields["Items"]:
        for item_field in (doc.fields["Items"].value or []):
            item = {}
            for name, val in (item_field.value or {}).items():
                if val:
                    item[name] = str(val.value or val.content or "")
            if item:
                report.items.append(item)
    
    # Overall confidence
    report.overall_confidence = sum(confidences) / len(confidences) if confidences else 0
    
    # Requires validation if low confidence or missing amount
    report.requires_validation = (
        report.overall_confidence < 0.85 or
        report.total_amount is None or
        report.transaction_date is None
    )
    
    # Categorize based on merchant
    if report.merchant:
        merchant_lower = report.merchant.lower()
        if any(m in merchant_lower for m in ["restaurant", "cafe", "taco", "bistro", "pizza"]):
            report.category = "Business Meals"
        elif any(m in merchant_lower for m in ["hotel", "inn", "marriott", "hilton", "airbnb"]):
            report.category = "Accommodation"
        elif any(m in merchant_lower for m in ["taxi", "uber", "lyft", "shell", "esso"]):
            report.category = "Transportation"
        elif any(m in merchant_lower for m in ["office", "staples", "best buy"]):
            report.category = "Office Supplies"
        else:
            report.category = "Other"
    
    return report

# Test
print("=== Expense Report Processing ===\n")

test_receipts = [
    ("restaurant_receipt.jpg", "EMP001"),
    ("hotel_receipt.jpg", "EMP002"),
    ("taxi_receipt.jpg", "EMP001"),
]

reports = []
for path, employee in test_receipts:
    report = process_expense_report(path, employee, di_client)
    reports.append(report)
    
    status = "⚠️ VALIDATION REQUIRED" if report.requires_validation else "✅ AUTO-APPROVED"
    print(f"[{employee}] {path}")
    print(f"  Merchant: {report.merchant}")
    print(f"  Date: {report.transaction_date}")
    print(f"  Total: {report.total_amount} {report.currency}")
    print(f"  Category: {report.category}")
    print(f"  Confidence: {report.overall_confidence:.0%}  → {status}")
    print()

# Summary by employee
print("=== Summary by Employee ===")
for employee_id in set(r.employee_id for r in reports):
    emp_reports = [r for r in reports if r.employee_id == employee_id]
    total = sum(r.total_amount for r in emp_reports if r.total_amount)
    print(f"{employee_id}: {len(emp_reports)} receipts, Total: ${total:.2f}")

10. Governance, Security and Compliance

# Security best practices for Document Intelligence
from azure.ai.documentintelligence.models import AnalyzeDocumentRequest
from azure.identity import DefaultAzureCredential
import hashlib

# ✅ Recommended: Managed Identity (no API key in code)
secure_client = DocumentIntelligenceClient(
    endpoint=os.environ["AZURE_DI_ENDPOINT"],
    credential=DefaultAzureCredential()
)

# ✅ Anonymize before sending sensitive documents
def anonymize_before_analysis(doc_bytes: bytes, contains_pii: bool = True) -> bytes:
    """
    If the document contains PII data, use Azure AI Language's
    PII detection BEFORE Document Intelligence.
    
    For testing with real data, consider anonymizing
    social security numbers, birth dates, etc.
    """
    if not contains_pii:
        return doc_bytes
    
    # In a real scenario: use Azure AI Language for PII detection
    # then replace with [REDACTED] before analyzing with Document Intelligence
    print("⚠️ Document contains PII - verify GDPR compliance")
    return doc_bytes

# ✅ Access logging (Audit Trail)
def analyze_with_audit(
    source: str,
    model: str,
    user: str,
    reason: str = "Automated processing"
) -> dict:
    """
    Analyzes a document with complete logging for compliance.
    """
    import logging
    from datetime import datetime
    
    # Log the access
    audit_log = {
        "timestamp": datetime.utcnow().isoformat(),
        "user": user,
        "file": os.path.basename(source),
        "file_hash": hashlib.sha256(open(source, "rb").read()).hexdigest()[:16],
        "model": model,
        "reason": reason
    }
    
    logging.info(f"AUDIT: {json.dumps(audit_log)}")
    
    # Analyze
    result = analyze_document(source, model)
    
    # Log completion
    audit_log["status"] = "completed"
    audit_log["fields_extracted"] = len(result.documents[0].fields) if result.documents else 0
    logging.info(f"AUDIT COMPLETED: {json.dumps(audit_log)}")
    
    return result

11. Summary and Key Points

11.1 Decision Architecture

flowchart TD
    DOC_TYPE{"Document type?"}
    
    DOC_TYPE -->|"Receipt"| RECEIPT["prebuilt-receipt\n→ MerchantName, Total, Items"]
    DOC_TYPE -->|"Invoice"| INVOICE["prebuilt-invoice\n→ InvoiceId, Amount, DueDate"]
    DOC_TYPE -->|"ID Card/Passport"| ID["prebuilt-idDocument\n→ FirstName, DOB, DocNumber"]
    DOC_TYPE -->|"Business Card"| BC["prebuilt-businessCard\n→ Name, Email, Phone"]
    DOC_TYPE -->|"Any text"| READ["prebuilt-read\n→ Full OCR"]
    DOC_TYPE -->|"Document structure"| LAYOUT["prebuilt-layout\n→ Tables, columns"]
    DOC_TYPE -->|"Custom form"| CUSTOM_MODEL["Custom model\n(5+ annotated examples)"]
    DOC_TYPE -->|"Multiple types"| COMPOSED["Composed model\n(Combine extractors)"]

11.2 Summary Table

Element	Description
Document Intelligence	Azure service for structured document extraction
OCR	Recognition of printed + handwritten text
Prebuilt	Microsoft models for standard types (receipts, invoices…)
Custom	Your own models for your specific forms
Layout	Structure extraction (tables, checkboxes)
Labels.json	Annotation file for custom training
Feedback Loop	Continuous improvement mechanism via human corrections
Container	On-premises deployment for compliance / latency
Composed Model	Combination of multiple extractors
Classifier	Automatically identify the document type

Model selection:

Standardized market documents?  → Prebuilt model
  └── Receipts? → prebuilt-receipt
  └── Invoices? → prebuilt-invoice
  └── IDs? → prebuilt-idDocument

Internal/proprietary forms? → Custom model
  └── Fixed layout? → Template mode
  └── Variable layout? → Neural mode

12. Glossary

Term	Definition
Azure AI Document Intelligence	Azure service for automatic extraction of information from documents (formerly Form Recognizer)
Bounding Box	Rectangle delimiting the position of text or a field in a document
Classifier (Custom)	Model for identifying the type of a document among several classes
Composed Model	Combination of multiple extraction models to handle a document portfolio
Custom Extraction	Model trained on your own documents to extract your specific fields
Document Intelligence Studio	Web interface for testing and managing DI models
Key-Value Pairs	Automatically extracted field/value pairs (“Total: $42.50”)
Labels.json	Annotation file defining the fields to extract in training documents
Layout Model	Model for extracting the structure of a document (tables, checkboxes, columns)
Neural Model	Custom model type based on neural networks (recommended)
OCR.json	File containing the raw OCR result for a training document
Polygon	Precise coordinates (not just a rectangle) of a text area
Prebuilt Model	Pre-trained Microsoft model for standardized document types
Read Model	DI model for raw text extraction (OCR)
Template Model	Custom model type based on a fixed template (strict layout)

Additional Resources:

Search Terms

information · extraction · azure · ai · document · intelligence · services · artificial · generative · models · container · custom · prebuilt · architecture · feedback · model · python · studio · cases · deployment · docker · loop · processing · sdk

Table of Contents