Sesja 11: Inteligentna analiza dokumentów

Zaawansowane przetwarzanie dokumentów

🎯 Cele sesji

• Implementacja Azure Form Recognizer i Document Intelligence
• Przetwarzanie różnych formatów dokumentów (PDF, obrazy, formularze)
• Wydobywanie strukturalnych danych z dokumentów
• Tworzenie custom document models dla specyficznych typów dokumentów

---

📄 Azure Form Recognizer & Document Intelligence

Podstawy przetwarzania dokumentów

Form Recognizer to usługa AI do automatycznego wydobywania tekstu, par klucz-wartość, tabel i struktur z dokumentów.

Kluczowe możliwości:

• OCR (Optical Character Recognition)
• Rozpoznawanie formularzy i faktur
• Wydobywanie tabel i struktur
• Custom models dla specyficznych dokumentów

Proces przetwarzania dokumentów

PRZESYŁANIE → PREPROCESSING → ANALIZA UKŁADU → WYDOBYWANIE TREŚCI → STRUKTURYZACJA

Preprocessing:

• Poprawa jakości obrazu
• Korekcja orientacji
• Segmentacja strony

Analiza układu:

• Wykrywanie tabel
• Identyfikacja pól formularzy
• Określanie kolejności czytania

Wydobywanie treści:

• OCR dla tekstu
• Rozpoznawanie pisma odręcznego
• Wykrywanie checkboxów i podpisów

---

🛠️ Praktyczne zastosowania

Implementacja Document Intelligence

from azure.ai.formrecognizer import DocumentAnalysisClient
from azure.core.credentials import AzureKeyCredential

# Inicjalizacja klienta
client = DocumentAnalysisClient(endpoint, AzureKeyCredential(key))

def analyze_document(document_url):
    """Analiza dokumentu z pomocą prebuilt-document model"""
    
    poller = client.begin_analyze_document_from_url(
        "prebuilt-document", document_url
    )
    result = poller.result()
    
    # Wydobywanie kluczowych informacji
    document_data = {
        "content": result.content,
        "tables": [],
        "key_value_pairs": [],
        "entities": []
    }
    
    # Przetwarzanie tabel
    for table in result.tables:
        table_data = {
            "row_count": table.row_count,
            "column_count": table.column_count,
            "cells": []
        }
        
        for cell in table.cells:
            table_data["cells"].append({
                "row_index": cell.row_index,
                "column_index": cell.column_index,
                "content": cell.content,
                "confidence": cell.confidence
            })
        
        document_data["tables"].append(table_data)
    
    # Wydobywanie par klucz-wartość
    for kv_pair in result.key_value_pairs:
        if kv_pair.key and kv_pair.value:
            document_data["key_value_pairs"].append({
                "key": kv_pair.key.content,
                "value": kv_pair.value.content,
                "confidence": kv_pair.confidence
            })
    
    return document_data

Custom Document Models

from azure.ai.formrecognizer import DocumentModelAdministrationClient

def train_custom_model(training_data_url):
    """Trenowanie custom modelu dla specyficznych dokumentów"""
    
    admin_client = DocumentModelAdministrationClient(endpoint, credential)
    
    # Rozpoczęcie treningu
    poller = admin_client.begin_build_document_model(
        build_mode="template",  # lub "neural"
        blob_container_url=training_data_url,
        description="Model dla faktur firmowych"
    )
    
    model = poller.result()
    print(f"Model ID: {model.model_id}")
    
    return model

def analyze_with_custom_model(document_url, model_id):
    """Analiza dokumentu z custom modelem"""
    
    poller = client.begin_analyze_document_from_url(
        model_id, document_url
    )
    
    result = poller.result()
    
    # Wydobywanie danych zgodnie z custom modelem
    extracted_data = {}
    
    for document in result.documents:
        for field_name, field in document.fields.items():
            extracted_data[field_name] = {
                "value": field.value,
                "confidence": field.confidence
            }
    
    return extracted_data

---

📊 Zaawansowane scenariusze

Przetwarzanie wsadowe dokumentów

import asyncio
from concurrent.futures import ThreadPoolExecutor

class BatchDocumentProcessor:
    def __init__(self, client):
        self.client = client
        self.executor = ThreadPoolExecutor(max_workers=5)
    
    async def process_document_batch(self, document_urls):
        """Przetwarzanie wielu dokumentów równolegle"""
        
        tasks = []
        for url in document_urls:
            task = asyncio.create_task(
                self.analyze_document_async(url)
            )
            tasks.append(task)
        
        results = await asyncio.gather(*tasks)
        return results
    
    async def analyze_document_async(self, document_url):
        """Asynchroniczna analiza dokumentu"""
        
        loop = asyncio.get_event_loop()
        
        # Uruchom synchroniczną analizę w executor
        result = await loop.run_in_executor(
            self.executor,
            self.analyze_single_document,
            document_url
        )
        
        return result
    
    def analyze_single_document(self, document_url):
        """Analiza pojedynczego dokumentu"""
        
        try:
            poller = self.client.begin_analyze_document_from_url(
                "prebuilt-invoice", document_url
            )
            
            result = poller.result()
            
            return {
                "document_url": document_url,
                "status": "success",
                "data": self.extract_invoice_data(result)
            }
            
        except Exception as e:
            return {
                "document_url": document_url,
                "status": "error",
                "error": str(e)
            }

Integracja z workflow biznesowym

from azure.servicebus import ServiceBusClient, ServiceBusMessage
import json

class DocumentWorkflowIntegrator:
    def __init__(self, form_recognizer_client, servicebus_client):
        self.form_client = form_recognizer_client
        self.servicebus = servicebus_client
    
    def process_business_document(self, document_info):
        """Przetwarzanie dokumentu w kontekście biznesowym"""
        
        document_url = document_info["url"]
        document_type = document_info["type"]
        
        # Wybór odpowiedniego modelu
        if document_type == "invoice":
            model = "prebuilt-invoice"
        elif document_type == "receipt":
            model = "prebuilt-receipt"
        elif document_type == "business_card":
            model = "prebuilt-businessCard"
        else:
            model = "prebuilt-document"
        
        # Analiza dokumentu
        analysis_result = self.analyze_with_model(document_url, model)
        
        # Walidacja biznesowa
        validation_result = self.validate_business_rules(
            analysis_result, document_type
        )
        
        # Routing do odpowiedniego workflow
        if validation_result["valid"]:
            self.route_to_approval_workflow(analysis_result)
        else:
            self.route_to_review_queue(analysis_result, validation_result["errors"])
        
        return {
            "document_id": document_info["id"],
            "analysis": analysis_result,
            "validation": validation_result,
            "next_step": validation_result["next_action"]
        }
    
    def validate_business_rules(self, analysis_result, document_type):
        """Walidacja zgodności z regułami biznesowymi"""
        
        errors = []
        
        if document_type == "invoice":
            # Sprawdź wymagane pola
            required_fields = ["vendor_name", "invoice_total", "invoice_date"]
            
            for field in required_fields:
                if field not in analysis_result or not analysis_result[field]:
                    errors.append(f"Brakuje wymaganego pola: {field}")
            
            # Sprawdź wartości
            if "invoice_total" in analysis_result:
                amount = analysis_result["invoice_total"].get("value", 0)
                if amount > 10000:  # Limit zatwierdzania
                    errors.append("Kwota przekracza limit automatycznego zatwierdzania")
        
        return {
            "valid": len(errors) == 0,
            "errors": errors,
            "next_action": "auto_approve" if len(errors) == 0 else "manual_review"
        }

---

✅ Zadania praktyczne

Zadanie 1: Analiza faktur (30 min)

1. Skonfiguruj Azure Form Recognizer
2. Przeanalizuj przykładowe faktury
3. Wydobądź kluczowe informacje (kwota, data, dostawca)
4. Zaimplementuj walidację danych

Zadanie 2: Custom model (45 min)

1. Przygotuj zestaw treningowy (5+ dokumentów tego samego typu)
2. Wytrenuj custom model
3. Przetestuj model na nowych dokumentach
4. Oceń dokładność i popraw

Zadanie 3: Integracja biznesowa (30 min)

1. Zaprojektuj workflow przetwarzania dokumentów
2. Zaimplementuj routing oparty na typie dokumentu
3. Dodaj walidację biznesową
4. Skonfiguruj powiadomienia o błędach

Zadanie 4: Batch processing (15 min)

1. Przygotuj zestaw 10+ dokumentów
2. Zaimplementuj przetwarzanie wsadowe
3. Dodaj raportowanie postępu
4. Przetestuj na większej liczbie dokumentów

---

📚 Materiały dodatkowe

• Azure Form Recognizer Documentation↗
• Document Intelligence Studio↗
• Custom Model Training Guide↗
• Best Practices for Document Processing↗