Azure Form Recognizer + Document Intelligence Pipeline

Solution for the use case Document Processing & Information Extraction Pipeline.

AI tools

Azure Form Recognizer, Azure Document Intelligence, Azure Logic Apps, SharePoint, Power Automate

Language

Python

Accuracy

95%

Published

2025-07-05

Code (Python)

import azure.ai.formrecognizer as fr
from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.identity import DefaultAzureCredential
import json
import requests

class DocumentProcessor:
    def __init__(self):
        self.credential = DefaultAzureCredential()
        self.form_recognizer = fr.DocumentAnalysisClient(
            endpoint=os.getenv("FORM_RECOGNIZER_ENDPOINT"),
            credential=self.credential
        )
        self.doc_intelligence = DocumentIntelligenceClient(
            endpoint=os.getenv("DOC_INTELLIGENCE_ENDPOINT"), 
            credential=self.credential
        )
    
    def process_document(self, document_url, document_type="general"):
        """Main document processing pipeline"""
        
        # Step 1: OCR and Layout Analysis
        layout_result = self.analyze_layout(document_url)
        
        # Step 2: Specialized document analysis
        if document_type == "invoice":
            specialized_result = self.analyze_invoice(document_url)
        elif document_type == "contract":
            specialized_result = self.analyze_contract(document_url)
        else:
            specialized_result = self.analyze_general_document(document_url)
        
        # Step 3: Combine results and validate
        processed_data = self.combine_and_validate(layout_result, specialized_result)
        
        # Step 4: Determine workflow routing
        workflow = self.determine_workflow(processed_data, document_type)
        
        return {
            "document_id": self.generate_document_id(),
            "processing_timestamp": datetime.now().isoformat(),
            "extracted_data": processed_data,
            "workflow": workflow,
            "confidence_score": processed_data.get("confidence", 0),
            "requires_review": processed_data.get("confidence", 0) < 0.85
        }
    
    def analyze_invoice(self, document_url):
        """Specialized invoice processing"""
        
        poller = self.form_recognizer.begin_analyze_document(
            "prebuilt-invoice", 
            document=self.get_document_stream(document_url)
        )
        result = poller.result()
        
        extracted_data = {}
        
        for invoice in result.documents:
            fields = invoice.fields
            
            extracted_data.update({
                "vendor_name": fields.get("VendorName", {}).get("value", ""),
                "vendor_address": fields.get("VendorAddress", {}).get("value", ""),
                "customer_name": fields.get("CustomerName", {}).get("value", ""),
                "invoice_id": fields.get("InvoiceId", {}).get("value", ""),
                "invoice_date": fields.get("InvoiceDate", {}).get("value", ""),
                "due_date": fields.get("DueDate", {}).get("value", ""),
                "subtotal": fields.get("SubTotal", {}).get("value", 0),
                "tax_amount": fields.get("TotalTax", {}).get("value", 0),
                "total_amount": fields.get("InvoiceTotal", {}).get("value", 0),
                "currency": fields.get("CurrencyCode", {}).get("value", "USD")
            })
            
            # Extract line items
            line_items = []
            if "Items" in fields:
                for item in fields["Items"].value:
                    line_item = {
                        "description": item.value.get("Description", {}).get("value", ""),
                        "quantity": item.value.get("Quantity", {}).get("value", 0),
                        "unit_price": item.value.get("UnitPrice", {}).get("value", 0),
                        "amount": item.value.get("Amount", {}).get("value", 0)
                    }
                    line_items.append(line_item)
            
            extracted_data["line_items"] = line_items
        
        return extracted_data
    
    def analyze_contract(self, document_url):
        """Specialized contract processing"""
        
        # Use Document Intelligence for comple
…

Prompts

DOCUMENT_ANALYSIS_PROMPT = """
Analyze this document and extract structured information:

Document Type: {document_type}
Content: {ocr_text}
Layout Analysis: {layout_data}

Extract the following information:
1. Document metadata (date, document number, sender, recipient)
2. Key financial data (amounts, totals, tax information)
3. Contract terms (dates, parties, obligations)
4. Compliance requirements (signatures, approvals, certifications)

Return structured JSON:
{
  "document_type": "invoice|contract|resume|report",
  "confidence": 0.95,
  "extracted_data": {
    "date": "2024-01-15",
    "amount": 1250.00,
    "currency": "USD",
    "vendor": "ABC Corp",
    "line_items": [],
    "approval_required": true
  },
  "validation_status": "valid|requires_review|invalid",
  "next_actions": ["route_to_approver", "validate_amount", "archive"]
}
"""

WORKFLOW_ROUTING_PROMPT = """
Based on document analysis, determine the appropriate workflow routing:

Document: {document_summary}
Extracted Data: {extracted_data}
Business Rules: {routing_rules}
User Permissions: {user_roles}

Routing Decision Matrix:
- Invoices > $5000: Requires CFO approval
- Contracts: Legal review + department head approval  
- Resumes: HR department + hiring manager
- Reports: Department specific routing

Generate workflow:
{
  "primary_route": "finance_approval",
  "approvers": ["john.doe@company.com", "cfo@company.com"],
  "deadline": "2024-01-20",
  "escalation_rules": {
    "after_hours": 48,
    "action": "notify_backup_approver"
  },
  "required_actions": ["budget_check", "vendor_verification"]
}
"""