Build a Technical Specification RAG Pipeline with Docling and Haystack

                      
                      from typing import Dict, Any
import os
from haystack import Document
from haystack.pipelines import Pipeline
from haystack.nodes import DensePassageRetriever, FARMReader
from docling import DoclingPipeline
import logging

# Configuration
api_key = os.getenv('API_KEY')
docling_api_url = os.getenv('DOCLING_API_URL')

# Initialize logging
logging.basicConfig(level=logging.INFO)

# Initialize Haystack components
retriever = DensePassageRetriever(model='facebook/dpr-question_encoder-single-nq-base')
reader = FARMReader(model='deepset/roberta-base-squad2')

# Create RAG pipeline
pipeline = Pipeline()
pipeline.add_node(retriever, name='Retriever', inputs=['Query'])
pipeline.add_node(reader, name='Reader', inputs=['Retriever'])

# Define function for processing documents
async def process_documents(query: str) -> Dict[str, Any]:
    try:
        docs = await DoclingPipeline(api_url=docling_api_url, api_key=api_key).fetch_docs(query)
        if not docs:
            return {'success': False, 'message': 'No documents found.'}
        results = pipeline.run(query=query, documents=docs)
        return {'success': True, 'data': results}
    except Exception as error:
        logging.error(f'Error processing documents: {str(error)}')
        return {'success': False, 'error': str(error)}

if __name__ == '__main__':
    query = 'What is the technical specification?'
    response = process_documents(query)
    print(response)