Build a Technical Specification RAG Pipeline with Docling and Haystack

                      
                     
from typing import Dict, Any
import os
import requests
from haystack.document_stores import InMemoryDocumentStore
from haystack.pipelines import DocumentSearchPipeline
from haystack.nodes import DensePassageRetriever, Reader

# Configuration
API_KEY = os.getenv('DOCLING_API_KEY')
DOC_URL = os.getenv('DOC_URL')

# Initialize Document Store and Pipeline
document_store = InMemoryDocumentStore()
dpr = DensePassageRetriever(document_store=document_store)
reader = Reader(model='deepset/roberta-base-squad2')
pipeline = DocumentSearchPipeline(retriever=dpr, reader=reader)

def fetch_documents() -> None:
    try:
        response = requests.get(DOC_URL, headers={'Authorization': f'Bearer {API_KEY}'})
        response.raise_for_status()  # Raise an error for bad responses
        documents = response.json().get('documents', [])
        document_store.write_documents(documents)
    except requests.RequestException as e:
        print(f'Error fetching documents: {e}')

def query_pipeline(query: str) -> Dict[str, Any]:
    try:
        results = pipeline.run(query=query, params={'Retriever': {'top_k': 5}, 'Reader': {'top_k': 1}})
        return results
    except Exception as e:
        return {'error': str(e)}

if __name__ == '__main__':
    fetch_documents()  # Populate the document store
    sample_query = 'What is the RAG pipeline?'
    print(query_pipeline(sample_query))