tasks/data_ingestion_task/data_ingestion_task.py

from tasks.base_task import BaseTask
from utils.llama_index_utils import setup_directories
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, StorageContext

class DataIngestionTask(BaseTask):
 def load_input(self, input_data):
 # No input data required; handles its own input (documents in the data directory)
 self.data_dir, self.persist_dir = setup_directories()

 def process(self):
 # Process the documents and store the index
 documents = SimpleDirectoryReader(self.data_dir).load_data()
 storage_context = StorageContext.from_defaults()
 self.index = VectorStoreIndex.from_documents(documents)
 self.index.storage_context.persist(persist_dir=self.persist_dir)

 def save_output(self, result):
 # No output to return for this task
 pass