Classify Manufacturing Regulations with LayoutParser and Haystack

                      
                     
from typing import Dict, Any
import os
import logging
from haystack.document_stores import InMemoryDocumentStore
from haystack.nodes import DensePassageRetriever, FARMReader
from haystack.pipelines import DocumentSearchPipeline
from layoutparser import Layout, load_model

# Configuration
logging.basicConfig(level=logging.INFO)
API_KEY = os.getenv('LAYOUTPARSER_API_KEY')
document_store = InMemoryDocumentStore()

# Load LayoutParser model
layout_model = load_model('lp://layoutparser')

# Initialize Haystack components
retriever = DensePassageRetriever(document_store=document_store)
reader = FARMReader(model_name_or_path='deepset/roberta-base-squad2')

# Pipeline for document classification
pipeline = DocumentSearchPipeline(retriever=retriever, reader=reader)

# Function to classify document
async def classify_document(file_path: str) -> Dict[str, Any]:
    try:
        layout = layout_model.detect(file_path)
        documents = layout.to_documents()
        document_store.write_documents(documents)
        results = pipeline.run(query='Manufacturing Regulation', params={'top_k': 5})
        return {'success': True, 'data': results}
    except Exception as e:
        logging.error(f'Error classifying document: {str(e)}')
        return {'success': False, 'error': str(e)}

if __name__ == '__main__':
    sample_file = 'path/to/sample_document.pdf'
    classification_result = classify_document(sample_file)
    print(classification_result)