diff --git a/indexing/Dockerfile b/indexing/Dockerfile deleted file mode 100644 index 3e161b95..00000000 --- a/indexing/Dockerfile +++ /dev/null @@ -1,27 +0,0 @@ -# Use the official AWS Lambda Python 3.9 runtime base image -FROM public.ecr.aws/lambda/python:3.9 - -# Create and set the working directory inside the container -WORKDIR /var/task - -# Copy the requirements file to the working directory -COPY indexing/requirements.txt . - -# Install dependencies into the /var/task directory (where Lambda expects them) -RUN pip install --no-cache-dir -r requirements.txt --target . - -# Copy the necessary files and directories -COPY baseclasses/ baseclasses/ -COPY config/ config/ -COPY core/ core/ -COPY indexing/ indexing/ -COPY util/ util/ -COPY constants/ constants/ -COPY lambda_handlers/indexing_handler.py . - -# Set environment variables -ENV PYTHONPATH=/var/task -ENV PYTHONUNBUFFERED=1 - -# Lambda runtime will look for the handler function here -CMD ["indexing_handler.lambda_handler"] diff --git a/indexing/__init__.py b/indexing/__init__.py deleted file mode 100644 index 6faa0285..00000000 --- a/indexing/__init__.py +++ /dev/null @@ -1 +0,0 @@ -from .indexing import chunk_embed_store diff --git a/indexing/fargate_indexing.Dockerfile b/indexing/fargate_indexing.Dockerfile deleted file mode 100644 index 1dca7221..00000000 --- a/indexing/fargate_indexing.Dockerfile +++ /dev/null @@ -1,26 +0,0 @@ -# Use the official AWS Lambda Python 3.9 runtime base image -FROM --platform=linux/amd64 python:3.9-slim - -# Create and set the working directory inside the container -WORKDIR /var/task - -# Copy the requirements file to the working directory -COPY indexing/requirements.txt . - -# Install dependencies into the /var/task directory (where Lambda expects them) -RUN pip install --no-cache-dir -r requirements.txt --target . - -# Copy the necessary files and directories -COPY baseclasses/ baseclasses/ -COPY config/ config/ -COPY core/ core/ -COPY indexing/ indexing/ -COPY util/ util/ -COPY handlers/task_processor.py . -COPY handlers/fargate_indexing_handler.py . - -# Set environment variables -ENV PYTHONPATH=/var/task -ENV PYTHONUNBUFFERED=1 - -CMD ["python", "fargate_indexing_handler.py"] \ No newline at end of file diff --git a/indexing/indexing.py b/indexing/indexing.py deleted file mode 100644 index d1d4180a..00000000 --- a/indexing/indexing.py +++ /dev/null @@ -1,127 +0,0 @@ -from core.processors import ChunkingProcessor, EmbedProcessor -from core.opensearch_vectorstore import OpenSearchVectorDatabase -from util.s3util import S3Util -from util.pdf_utils import process_pdf_from_folder -import logging -from typing import Dict, List, Any -from opensearchpy.helpers import bulk -import os -import uuid -import json -from config.experimental_config import ExperimentalConfig -from config.config import Config -from core.dynamodb import DynamoDBOperations -import re - -logger = logging.getLogger() -logger.setLevel(logging.INFO) - -def clean_text_for_vector_db(text): - """ - Cleans the input text by removing quotes, special symbols, extra whitespaces, - newline (\n), and tab (\t) characters. - - Args: - text (str): The input text to clean. - - Returns: - str: The cleaned text. - """ - # Remove single and double quotes - text = text.replace('"', '').replace("'", "") - # Remove special symbols (keeping alphanumerics and spaces) - text = re.sub(r'[^a-zA-Z0-9\s]', '', text) - # Remove newlines and tabs - text = text.replace('\n', ' ').replace('\t', ' ') - # Normalize whitespace - text = re.sub(r'\s+', ' ', text) - # Strip leading and trailing spaces - return text.strip() - -def chunk_embed_store(config : Config, experimentalConfig : ExperimentalConfig)-> None: - """Main function to run the chunking and embedding pipeline.""" - experiment_dynamodb = DynamoDBOperations(region=config.aws_region, table_name=config.experiment_table) - logger.info(experiment_dynamodb.table) - try: - """Main function to run the pipeline.""" - - if not experimentalConfig.kb_data: - raise ValueError("S3 path is missing in the kb_data field.") - - pdf_folder_path = S3Util().download_directory_from_s3(experimentalConfig.kb_data) - - # Step 1: Chunking - chunks = ChunkingProcessor(experimentalConfig).chunk(process_pdf_from_folder(pdf_folder_path)) - - embed_chunks = chunks - if experimentalConfig.chunking_strategy.lower() == 'hierarchical': - embed_chunks = [] - for chunk in chunks: - embed_chunks.append(chunk[2]) # Child Chunk only - - # Step 2: Embedding - embedding_results = EmbedProcessor(experimentalConfig).embed(embed_chunks) - - total_index_embed_tokens = 0 - for _, _, metadata in embedding_results: - total_index_embed_tokens += int(metadata['inputTokens']) - - if experimentalConfig.chunking_strategy.lower() == 'hierarchical': - temp_results = [] - for i, chunk in enumerate(chunks): - temp_embedding = list(embedding_results[i]) - temp_embedding.extend([chunk[0], chunk[1]]) - temp_results.append(temp_embedding) - embedding_results = temp_results - documents = [ - { - "_index": experimentalConfig.index_id, - "execution_id":experimentalConfig.execution_id, - "chunk_id": str(uuid.uuid4()), # Generate a unique UUID for each chunk - "text": clean_text_for_vector_db(parent_chunk), - "child_text": clean_text_for_vector_db(chunk), - "parent_id": parent_id, - config.vector_field: embedding, - "metadata": metadata # Optional metadata, defaulting to an empty dictionary - } - for embedding, chunk, metadata, parent_id, parent_chunk in embedding_results # Enumerate is unnecessary since UUIDs are used - ] - else: - documents = [ - { - "_index": experimentalConfig.index_id, - "execution_id":experimentalConfig.execution_id, - "chunk_id": str(uuid.uuid4()), # Generate a unique UUID for each chunk - "text": clean_text_for_vector_db(chunk), - config.vector_field: embedding, - "metadata": metadata # Optional metadata, defaulting to an empty dictionary - } - for embedding, chunk, metadata in embedding_results # Enumerate is unnecessary since UUIDs are used - ] - - logger.info(f"Experiment {experimentalConfig.experiment_id} Indexing Embed Tokens : {total_index_embed_tokens}") - - experiment_dynamodb.update_item( - key={'id': experimentalConfig.experiment_id}, - update_expression="SET index_embed_tokens = :embed", - expression_values={':embed': total_index_embed_tokens} - ) - - _insert_to_opensearch(config, documents) - except Exception as e: - logger.exception(f"Pipeline failed: {e}") - raise e - -def _insert_to_opensearch(config: Config, documents: List[Dict[str, Any]]): - vector_database = OpenSearchVectorDatabase(host=config.opensearch_host, is_serverless=config.opensearch_serverless, region=config.aws_region,username=config.opensearch_username, - password=config.opensearch_password) - chunks_length = len(documents) - chunk_size = 500 # Default chunk size streaming by Opensearch - if chunks_length < chunk_size: - chunk_size = chunks_length - logger.info(f"Opensearch Bulk insert initiated") - bulk(vector_database.client, documents, chunk_size=chunk_size, max_retries=1) - logger.info("Opensearch Bulk insert successful \n Pipeline completed successfully.") - - - diff --git a/indexing/requirements.txt b/indexing/requirements.txt deleted file mode 100644 index d57e5c7b..00000000 --- a/indexing/requirements.txt +++ /dev/null @@ -1,12 +0,0 @@ -boto3==1.35.87 -botocore==1.35.87 -langchain -PyPDF2 -langchain-aws -llama-index -python-dotenv -opensearch-py -sagemaker -ragas==0.2.6 -langchain_aws==0.2.7 -pymupdf \ No newline at end of file