Function process_text_task

worker/tasks.py:142–248

(self, textToProcess: str, documentId: str, generateQuestions: bool, generateSummaries: bool)

# Celery task for processing text
@celery_app.task(bind=True, rate_limit="1/m", name="celery_worker.process_text_task")
def process_text_task(self, textToProcess: str, documentId: str, generateQuestions: bool, generateSummaries: bool):
    logging.info(f"Starting process for document {documentId}")
    self.update_state(state=AppConfig.PROCESSING_DOCUMENT, meta={"documentId": documentId})
 
    global active_tasks_count

    # Task Concurrency Management
    with active_tasks_lock:
        # Check if the maximum number of concurrent tasks has been reached
        if active_tasks_count >= MAX_CONCURRENT_TASKS:
            # Requeue or delay the task
            raise self.retry(countdown=60)  # Retry after 60 seconds

        # Increment the count of active tasks
        active_tasks_count += 1

    try: 
        doc = telegram.text_to_docs(textToProcess)  

        # Setup splitters 
        parent_splitter = TokenTextSplitter(chunk_size=512, chunk_overlap=24)
        child_splitter = TokenTextSplitter(chunk_size=100, chunk_overlap=24)
        parent_documents = parent_splitter.split_documents(doc)

        # Setup embeddings
        embeddings = OpenAIEmbeddings(openai_api_key=AppConfig.OPENAI_API_KEY, embedding_dimension=AppConfig.EMBEDDING_DIMENSION)

        # Iterate through parent and child chunks for document and generate structure
        for i, parent in enumerate(parent_documents):

            self.update_state(state=AppConfig.PROCESSING_PAGES, meta={"page": i+1, "total_pages": len(parent_documents), "documentId": documentId})
            logging.info(f"processing chunk {i+1} of {len(parent_documents)} for document {documentId}")
            
            child_documents = child_splitter.split_documents([parent])
            params = {
                "document_uuid": documentId,
                "parent_uuid": str(uuid.uuid4()),
                "name": f"Page {i+1}",
                "parent_text": parent.page_content,
                "parent_id": i,
                "parent_embedding": embeddings.embed_query(parent.page_content),
                "children": [
                    {
                        "text": c.page_content,
                        "id": str(uuid.uuid4()),
                        "name": f"{i}-{ic+1}",
                        "embedding": embeddings.embed_query(c.page_content),
                    }
                    for ic, c in enumerate(child_documents)
                ],
            }

            try:
                # Ingest data
                with driver.session() as session :
                    session.run(
                    """
                        MERGE (p:Page {uuid: $parent_uuid})