Initial commit: Iqbal Poetry RAG system

.gitignore

@@ -0,0 +1,37 @@
+# === Python virtual environments ===
+.env_iqbal_rag/
+
+# === Python cache and compiled files ===
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+
+# === Dataset (do not track poetry data) ===
+data/
+
+# === Not Implemented Features ===
+sft/
+
+
+# === Environment variable/config files ===
+.env
+
+# === Logs ===
+*.log
+
+# === IDE/editor folders ===
+.vscode/
+.idea/
+
+# === OS generated files ===
+.DS_Store
+Thumbs.db
+
+# === Test and coverage outputs ===
+.pytest_cache/
+.coverage
+htmlcov/
+
+# === Jupyter Notebook checkpoints ===
+.ipynb_checkpoints/

README.md

@@ -0,0 +1,101 @@
+# Iqbal Poetry RAG System
+
+A Retrieval-Augmented Generation (RAG) system for exploring and querying the poetry of Allama Iqbal. This project leverages vector search and large language models (LLMs) to answer questions about Iqbal's poetry, providing relevant poem excerpts as context.
+
+---
+
+## Features
+
+- **Semantic Search**: Retrieve the most relevant poems for a given question using vector embeddings.
+- **LLM-Powered Answers**: Generate answers using a language model, grounded in retrieved poem context.
+- **Configurable**: Easily adjust retrieval thresholds, model settings, and data sources.
+- **Error Handling**: Robust error management for smoother user experience.
+- **(Optional) Feedback Logging**: Log user feedback for continuous improvement.
+
+---
+
+## Installation
+
+### Prerequisites
+
+- Python 3.9+
+- [uv](https://github.com/astral-sh/uv) (a fast Python package installer, drop-in replacement for pip)
+
+### 1. Clone the repository
+
+```bash
+git clone https://github.com/yourusername/iqbal_poetry_rag.git
+cd iqbal_poetry_rag
+```
+
+### 2. Install dependencies
+
+```bash
+uv pip install -r requirements.txt
+```
+
+---
+
+## Usage
+
+1. **Prepare your data**: Place your poems JSON file at the path specified in `app/config.py` (`JSON_FILE_PATH`).
+2. **Run the main application** (example, adjust as needed):
+
+```bash
+python app/main.py
+```
+
+3. **Query the system**: Enter your question about Iqbal's poetry and receive contextually grounded answers.
+
+---
+
+## Project Structure
+
+```bash
+iqbal_poetry_rag/
+│
+├── app/
+│ ├── RAGSystem.py # Main RAG system class
+│ ├── main.py # Entry point for the application
+│ └── config.py # Configuration (thresholds, file paths, etc.)
+│
+├── rag/
+│ ├── vector_store.py # Vector store initialization and building
+│ ├── retriever.py # Retriever configuration
+│ ├── llm.py # LLM initialization and prompt management
+│
+├── utils/
+│ ├── error_handling.py # Error handling decorators
+│ └── feedback_logger.py # (Optional) Feedback logging
+│
+├── requirements.txt # Project dependencies
+└── README.md # This file
+```
+
+---
+
+## Configuration
+
+Edit `app/config.py` to set:
+
+- `SCORE_THRESHOLD`: Minimum similarity score for retrieved poems.
+- `JSON_FILE_PATH`: Path to your poems data file.
+
+---
+
+## Contributing
+
+Contributions are welcome! Please open issues or submit pull requests for improvements or bug fixes.
+
+---
+
+## License
+
+[MIT License](LICENSE)
+
+---
+
+## Acknowledgements
+
+- Inspired by the poetry of Allama Iqbal.
+- Built with Python, vector search, and LLM technologies.

app/RAGSystem.py

@@ -0,0 +1,50 @@
+"""Main RAG system implementation."""
+
+
+from rag.vector_store import initialize_vector_store, build_vector_store_from_json
+from rag.retriever import configure_retriever
+from rag.llm import initialize_llm, get_rag_prompt
+# from utils.feedback_logger import FeedbackLogger
+from utils.error_handling import handle_rag_error
+from app.config import SCORE_THRESHOLD, JSON_FILE_PATH
+
+class IqbalRAGSystem:
+    """Manages the RAG system for Iqbal's poetry."""
+
+    def __init__(self):
+        """Initialize the RAG system components."""
+        # Build or load vector store
+        self.vector_store = build_vector_store_from_json(JSON_FILE_PATH)
+        self.retriever = configure_retriever(self.vector_store)
+        self.llm = initialize_llm()
+        self.prompt = get_rag_prompt()
+        self.chain = self.prompt | self.llm
+        # self.feedback_logger = FeedbackLogger()
+
+    @handle_rag_error
+    def query_rag(self, question):
+        """Process a query through the RAG system."""
+        print(f"**********************************************************************")
+        print(f"query_rag: {question}")
+        docs = self.retriever.invoke(question, config={'score_threshold': SCORE_THRESHOLD})
+        if not docs:
+            return "No relevant poems found", []
+
+        print(f"docs: {docs}")
+        context = "\n\n".join(doc.page_content for doc in docs)
+        print(f"context: {context}")
+        context_ids = [doc.metadata.get("poem_id", "") for doc in docs]
+        print(f"context_ids: {context_ids}")
+        
+        response = self.chain.invoke({
+            'context': context,
+            'question': question
+        })
+        print(f"response: {response}")
+        print(f"**********************************************************************")
+        
+        return response, context_ids
+
+    # def log_feedback(self, query, response, feedback, comment, context_ids):
+    #     """Log user feedback."""
+    #     self.feedback_logger.log_feedback(query, response, feedback, comment, context_ids)

app/config.py

@@ -0,0 +1,38 @@
+"""Configuration settings for the Iqbal Poetry RAG application."""
+
+import os
+import gradio as gr
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+# Application settings
+APP_NAME = "RAG-Iqbal: Q&A with Allama Iqbal based on a poetry dataset"
+DEBUG_MODE = os.getenv("DEBUG_MODE", "False").lower() == "true"
+
+# API Keys
+PERPLEXITY_API_KEY = os.getenv("PERPLEXITY_API_KEY")
+
+# Paths
+BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+CHROMA_DB_DIR = os.path.join(BASE_DIR, "data", "iqbalchroma_db")
+FEEDBACK_DIR = os.path.join(BASE_DIR, "data", "feedback")
+JSON_FILE_PATH = os.path.join(BASE_DIR, "data", "processed_data", "iqbal_poems_rag.json")
+
+# Create necessary directories
+os.makedirs(os.path.dirname(JSON_FILE_PATH), exist_ok=True)
+os.makedirs(CHROMA_DB_DIR, exist_ok=True)
+os.makedirs(FEEDBACK_DIR, exist_ok=True)
+
+# RAG settings
+EMBEDDING_MODEL = "llama3"  # This should match your Ollama model name exactly
+LLM_MODEL = "TinyLlama/TinyLlama-1.1B-Chat-v1.0"   # "microsoft/phi-2"
+RETRIEVER_K = 5
+RETRIEVER_FETCH_K = 20
+RETRIEVER_LAMBDA_MULT = 0.75
+SCORE_THRESHOLD = 0.65
+
+# Gradio settings
+GRADIO_THEME = gr.themes.Soft()  # Import gr here
+GRADIO_SERVER_PORT = int(os.getenv("PORT", 7860))

app/gradio_interface.py

@@ -0,0 +1,138 @@
+"""Gradio interface for the Iqbal Poetry RAG application."""
+
+import gradio as gr
+from app.config import APP_NAME, GRADIO_THEME, GRADIO_SERVER_PORT
+
+def process_query(question, history):
+    """Process user query through the RAG system."""
+    try:
+        response, _ = rag_system.query_rag(question)
+        return response, _
+    except Exception as e:
+        error_message = f"Error processing query: {str(e)}"
+        print(f"Error in process_query: {error_message}")  # Log error for debugging
+        return error_message
+
+def handle_feedback(question, response, feedback, comment):
+    """Handle user feedback submission."""
+    if not question or not response:
+        return "Feedback not logged - missing data"
+    
+    # Since feedback logging is commented out in IqbalRAGSystem, just return success
+    return "Feedback logging is currently disabled"
+
+def create_gradio_interface():
+    """Create and configure the Gradio interface."""
+    with gr.Blocks(theme=GRADIO_THEME, title=APP_NAME) as app:
+        gr.Markdown(f"# {APP_NAME}")
+        gr.Markdown("""
+        Welcome to the Iqbal Poetry RAG system! Ask questions about Iqbal's philosophical poetry.
+        The system will search through Iqbal's poems and provide relevant answers based on the content.
+        """)
+        
+        chatbot = gr.Chatbot(
+            label="Conversation",
+            height=500,
+            type="messages",  # Explicitly set type to 'messages'
+            show_label=True,
+            container=True
+        )
+        
+        with gr.Row():
+            with gr.Column(scale=4):
+                input_question = gr.Textbox(
+                    label="Your Question",
+                    placeholder="Ask about philosophical concepts in Iqbal's poetry...",
+                    lines=3,
+                    show_label=True
+                )
+            
+            with gr.Column(scale=1):
+                submit_btn = gr.Button("Submit", variant="primary")
+        
+        with gr.Accordion("Provide Feedback", open=False):
+            feedback_rating = gr.Radio(
+                ["Helpful", "Partially Helpful", "Incorrect"],
+                label="Feedback Rating",
+                value="Helpful"
+            )
+            
+            feedback_comment = gr.Textbox(
+                label="Additional Comments",
+                placeholder="Provide detailed feedback...",
+                lines=3
+            )
+            
+            feedback_btn = gr.Button("Submit Feedback")
+            feedback_status = gr.Markdown()
+        
+        gr.Examples(
+            examples=[
+                "Explain Iqbal's concept of Khudi",
+                "Analyze the symbolism in 'The Himalayas' poem",
+                "Compare Iqbal's view of nature with Romantic poets",
+                "What is Iqbal's view on Western materialism?",
+                "Discuss the influence of Rumi on Iqbal's philosophy",
+                "What are the main themes in 'The Secrets of the Self'?",
+                "How does Iqbal view the relationship between God and man?"
+            ],
+            inputs=input_question
+        )
+        
+        def user_input(user_message, history):
+            if not user_message.strip():
+                return "", history
+            # Convert to new message format
+            history = history + [{"role": "user", "content": user_message}]
+            return "", history
+        
+        def bot_response(history):
+            user_message = history[-1]["content"]
+            bot_message, _ = process_query(user_message, history)
+            # Convert to new message format
+            history = history + [{"role": "assistant", "content": bot_message}]
+            return history
+        
+        submit_btn.click(
+            user_input,
+            [input_question, chatbot],
+            [input_question, chatbot],
+            queue=False
+        ).then(
+            bot_response,
+            chatbot,
+            chatbot
+        )
+        
+        input_question.submit(
+            user_input,
+            [input_question, chatbot],
+            [input_question, chatbot],
+            queue=False
+        ).then(
+            bot_response,
+            chatbot,
+            chatbot
+        )
+        
+        feedback_btn.click(
+            fn=handle_feedback,
+            inputs=[input_question, chatbot, feedback_rating, feedback_comment],
+            outputs=feedback_status
+        )
+        
+        return app
+
+def launch_gradio_app(system=None):
+    """Launch the Gradio application."""
+    global rag_system
+    if system is not None:
+        globals()['rag_system'] = system
+    
+    app = create_gradio_interface()
+    app.launch(
+        server_name="0.0.0.0",
+        server_port=GRADIO_SERVER_PORT,
+        share=False,
+        show_error=True  # Show detailed error messages in the UI
+    )

rag/embeddings.py

@@ -0,0 +1,17 @@
+"""Embedding functionality for the RAG system."""
+
+import os
+from langchain_ollama import OllamaEmbeddings
+from app.config import EMBEDDING_MODEL
+
+def get_embeddings():
+    """Initialize and return the embedding model."""
+    try:
+        return OllamaEmbeddings(
+            model=EMBEDDING_MODEL,
+            base_url="http://localhost:11434"  # Explicitly set the base URL
+        )
+    except Exception as e:
+        print(f"Warning: Failed to initialize OllamaEmbeddings: {e}")
+        # Fallback to a different embedding method if needed
+        raise

rag/llm.py

@@ -0,0 +1,57 @@
+"""LLM configuration for the RAG system."""
+
+from langchain_community.chat_models import ChatPerplexity
+from langchain_core.prompts import PromptTemplate, ChatPromptTemplate
+from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
+from langchain_community.llms import HuggingFacePipeline
+from app.config import PERPLEXITY_API_KEY, LLM_MODEL
+
+def initialize_llm():
+    """Initialize and return the LLM."""
+    # return ChatPerplexity(
+    #     pplx_api_key=PERPLEXITY_API_KEY,
+    #     model=LLM_MODEL,
+    #     temperature=0.2,
+    #     max_tokens=1024
+    # )
+
+    tokenizer = AutoTokenizer.from_pretrained(LLM_MODEL, trust_remote_code=True)
+    model = AutoModelForCausalLM.from_pretrained(
+        LLM_MODEL,
+        trust_remote_code=True,
+        device_map="auto",
+        torch_dtype="auto"  # enables FP16 if available
+    )
+
+    pipe = pipeline(
+        "text-generation",
+        model=model,
+        tokenizer=tokenizer,
+        do_sample=True,
+        max_new_tokens=1024,
+        temperature=0.7,
+        top_p=0.9,
+        repetition_penalty=1.1,
+        return_full_text=False
+    )
+
+    return HuggingFacePipeline(pipeline=pipe)
+
+
+def get_rag_prompt():
+    """Return the RAG prompt template."""
+    # return ChatPromptTemplate.from_template(
+    #     """You are an expert on Allama Iqbal's poetry based on the context. Use only the context to answer and cite his verses in the answer.
+    #     Context: {context}
+    #     Question: {question}
+    #     Answer in structured Markdown:"""
+    # )
+
+    return PromptTemplate.from_template(
+        """You are an expert on Allama Iqbal's poetry. Use the provided context to answer the question.        
+        ### CONTEXT:{context}        
+        ### QUESTION:{question}        
+        ### ANSWER (Markdown formatted with poetic citations):
+        """
+    )
+

rag/retriever.py

@@ -0,0 +1,14 @@
+"""Retriever configuration for the RAG system."""
+
+from app.config import RETRIEVER_K, RETRIEVER_FETCH_K, RETRIEVER_LAMBDA_MULT
+
+def configure_retriever(vector_store):
+    """Configure and return the retriever."""
+    return vector_store.as_retriever(
+        search_type="mmr",
+        search_kwargs={
+            'k': RETRIEVER_K,
+            'fetch_k': RETRIEVER_FETCH_K,
+            'lambda_mult': RETRIEVER_LAMBDA_MULT
+        }
+    )

rag/vector_store.py

@@ -0,0 +1,70 @@
+"""Vector store management for the RAG system."""
+
+import json
+import os
+from langchain_chroma import Chroma  # Updated import
+from langchain_core.documents import Document
+from app.config import CHROMA_DB_DIR
+from rag.embeddings import get_embeddings
+
+def initialize_vector_store():
+    """Initialize and return the Chroma vector store."""
+    embeddings = get_embeddings()
+    return Chroma(
+        persist_directory=CHROMA_DB_DIR,
+        embedding_function=embeddings
+    )
+
+def build_vector_store_from_json(json_file_path):
+    """Build and persist a vector store from JSON data."""
+    # Check if vector store already exists
+    if os.path.exists(CHROMA_DB_DIR) and os.listdir(CHROMA_DB_DIR):
+        print(f"Vector store already exists at {CHROMA_DB_DIR}. Skipping creation.")
+        return initialize_vector_store()
+    
+    print(f"Building vector store from {json_file_path}...")
+    embeddings = get_embeddings()
+    
+    # Load JSON data
+    with open(json_file_path, 'r', encoding='utf-8') as f:
+        poems_data = json.load(f)
+    
+    # Convert to documents
+    documents = []
+    for poem in poems_data:
+        # Create content with all available information
+        content_parts = []
+        if poem.get('poem_id'):
+            content_parts.append(f"Poem ID: {poem['poem_id']}")
+        if poem.get('book_title'):
+            content_parts.append(f"Book: {poem['book_title']}")
+        if poem.get('full_text'):
+            content_parts.append(f"Text:\n{poem['full_text']}")
+        
+        # Create metadata
+        metadata = {
+            "poem_id": poem.get("poem_id", ""),
+            "book_id": poem.get("book_id", ""),
+            "book_title": poem.get("book_title", "Unknown")
+        }
+        
+        # Create document
+        doc = Document(
+            page_content="\n\n".join(content_parts),
+            metadata=metadata
+        )
+        
+        documents.append(doc)
+    
+    print(f"Creating vector store with {len(documents)} documents...")
+    
+    # Create vector store
+    os.makedirs(CHROMA_DB_DIR, exist_ok=True)
+    vector_store = Chroma.from_documents(
+        documents=documents,
+        embedding=embeddings,
+        persist_directory=CHROMA_DB_DIR
+    )
+    
+    print(f"Vector store created and persisted at {CHROMA_DB_DIR}")
+    return vector_store

run.py

@@ -0,0 +1,76 @@
+"""Entry point for the Iqbal Poetry RAG application."""
+
+import os
+import sys
+import requests
+import subprocess
+from app.RAGSystem import IqbalRAGSystem
+from app.gradio_interface import launch_gradio_app
+from app.config import JSON_FILE_PATH
+
+def check_ollama_availability():
+    """Check if Ollama is running and has the required models."""
+    try:
+        # First check if Ollama service is running
+        response = requests.get("http://localhost:11434/api/tags", timeout=5)
+        
+        if response.status_code == 200:
+            # Check available models
+            available_models = [model["name"].lower() for model in response.json().get("models", [])]
+            
+            if "llama3" not in available_models:
+                print("Warning: llama3 model not found in Ollama.")
+                # Try to check again with list command
+                try:
+                    result = subprocess.run(["ollama", "list"], capture_output=True, text=True)
+                    if "llama3" in result.stdout.lower():
+                        print("Model found via command line. Proceeding...")
+                        return True
+                    else:
+                        print("Please install it with: ollama pull llama3")
+                        return False
+                except Exception:
+                    print("Could not check models via command line.")
+                    return False
+            return True
+        else:
+            print(f"Warning: Ollama is running but API call failed with status {response.status_code}.")
+            return False
+    except requests.exceptions.ConnectionError:
+        print("Error: Ollama server is not running. Please start Ollama with:")
+        print("ollama serve")
+        return False
+    except Exception as e:
+        print(f"Error checking Ollama: {str(e)}")
+        return False
+
+def check_data_file():
+    """Check if the required data file exists."""
+    if not os.path.exists(JSON_FILE_PATH):
+        print(f"Error: Required data file not found at {JSON_FILE_PATH}")
+        print("Please ensure the iqbal_poems_rag.json file is present in the data/processed_data directory.")
+        return False
+    return True
+
+if __name__ == "__main__":
+    # Check if data file exists
+    if not check_data_file():
+        sys.exit(1)
+    
+    # Check if Ollama is available
+    ollama_available = check_ollama_availability()
+    if not ollama_available:
+        print("Warning: Proceeding without verifying Ollama setup...")
+    
+    try:
+        # Initialize RAG system (this will build the vector store if needed)
+        print("Initializing Iqbal Poetry RAG system...")
+        rag_system = IqbalRAGSystem()
+        
+        # Launch Gradio app
+        print("Launching Gradio interface...")
+        print(f"Access the application at http://localhost:{os.getenv('PORT', 7860)}")
+        launch_gradio_app(system=rag_system)
+    except Exception as e:
+        print(f"Error: Failed to start the application: {str(e)}")
+        sys.exit(1)

tests/inspector.py

@@ -0,0 +1,130 @@
+import os
+import json
+from langchain_chroma import Chroma
+from langchain_core.documents import Document
+from pprint import pprint
+import pandas as pd
+
+class ChromaVectorStoreInspector:
+    def __init__(self, persist_dir, embeddings):
+        """
+        Initialize inspector with existing Chroma store
+        
+        Args:
+            persist_dir (str): Directory where Chroma data is stored
+            embeddings: Embeddings model used in the original store
+        """
+        self.vector_store = Chroma(
+            persist_directory=persist_dir,
+            embedding_function=embeddings
+        )
+        self.collection = self.vector_store._collection
+
+    def get_store_metadata(self):
+        """Get critical metadata about the vector store"""
+        return {
+            "document_count": self._get_document_count(),
+            "embedding_function": str(self.vector_store._embedding_function),
+            "persist_directory": self.vector_store._persist_directory,
+            "collection_name": self.collection.name
+        }
+
+    def _get_document_count(self):
+        """Get total number of documents in the collection"""
+        return self.collection.count()
+
+    def sample_documents(self, n=5):
+        """Retrieve sample documents with metadata"""
+        results = self.collection.get(limit=n)
+        return [
+            {
+                "id": doc_id,
+                "metadata": meta,
+                "content": doc[:200] + "..." if len(doc) > 200 else doc
+            }
+            for doc_id, meta, doc in zip(
+                results["ids"],
+                results["metadatas"],
+                results["documents"]
+            )
+        ]
+
+    def analyze_metadata(self):
+        """Analyze metadata distribution patterns"""
+        results = self.collection.get()
+        print(results)
+        # df = pd.DataFrame(results["metadatas"])
+        
+        # analysis = {}
+        # if not df.empty:
+        #     analysis["metadata_fields"] = list(df.columns)
+        #     analysis["book_title_distribution"] = df["book_title"].value_counts().to_dict()
+        #     analysis["missing_values"] = df.isna().sum().to_dict()
+        
+        # return analysis
+
+    def test_semantic_search(self, query, k=3):
+        """Test the vector search functionality"""
+        results = self.vector_store.similarity_search(query, k=k)
+        return [
+            {
+                "content": doc.page_content[:150] + "...",
+                "metadata": doc.metadata,
+                "score": doc.metadata.get("score", 0.0)
+            }
+            for doc in results
+        ]
+
+    def full_health_check(self):
+        """Comprehensive store verification report"""
+        return {
+            "metadata": self.get_store_metadata(),
+            "sample_documents": self.sample_documents(),
+            "metadata_analysis": self.analyze_metadata(),
+            "search_test": self.test_semantic_search("philosophical concepts")
+        }
+
+    def verify_against_source(self, json_path):
+        """Verify vector store contents against source JSON"""
+        with open(json_path, "r") as f:
+            source_data = json.load(f)
+        
+        source_ids = {p["poem_id"] for p in source_data}
+        stored_ids = set(self.collection.get()["ids"])
+        
+        return {
+            "source_count": len(source_ids),
+            "stored_count": len(stored_ids),
+            "missing_in_store": source_ids - stored_ids,
+            "extra_in_store": stored_ids - source_ids
+        }
+
+# Usage example
+if __name__ == "__main__":
+    from rag.embeddings import get_embeddings  # Your existing embeddings setup
+    from app.config import CHROMA_DB_DIR  # Your config
+    
+    # Initialize inspector
+    inspector = ChromaVectorStoreInspector(
+        persist_dir=CHROMA_DB_DIR,
+        embeddings=get_embeddings()
+    )
+    
+    # print("\n=== Vector Store Metadata ===")
+    # pprint(inspector.get_store_metadata())
+    
+    # print("\n=== Document Samples ===")
+    # pprint(inspector.sample_documents())
+    
+    # print("\n=== Metadata Analysis ===")
+    # pprint(inspector.analyze_metadata())
+    
+    # print("\n=== Source Verification ===")
+    # verification = inspector.verify_against_source("data/processed_data/iqbal_poems_rag.json")
+    # pprint(verification)
+    
+    # print("\n=== Search Test Results ===")
+    # pprint(inspector.test_semantic_search("lost"))
+
+    print("\n=== Full Health Check ===")
+    pprint(inspector.full_health_check())

utils/__init__.py

@@ -0,0 +1,15 @@
+# Internal imports
+from .dataset_curator import DatasetCurator
+from .dataset_downloader import DatasetDownloader
+
+
+# Download the dataset from github
+# downloader = DatasetDownloader(output_dir="data")
+# downloader.download_from_github(source_name="github_iqbal_demystified")
+
+# Process the dataset into a single file
+# curator = DatasetCurator(data_path="data", output_dir="data/processed_data")
+# dataset = curator.process_dataset(source="github_iqbal_demystified")
+# print(f"Dataset processing complete. Stats:")
+# print(f"- Books: {dataset['metadata']['total_books']}")
+# print(f"- Poems: {dataset['metadata']['total_poems']}")

utils/dataset_curator.py

@@ -0,0 +1,386 @@
+# Standard library imports
+import os
+import json
+import logging
+from pathlib import Path
+from typing import Dict, List, Any, Optional
+
+# Third-party imports
+import yaml
+from tqdm import tqdm
+
+# Configure logging
+logging.basicConfig(
+    level=logging.DEBUG,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    handlers=[
+        logging.FileHandler("dataset_curation.log", encoding="utf-8"),
+        logging.StreamHandler()
+    ]
+)
+logger = logging.getLogger("DatasetCurator")
+
+
+class DatasetCurator:
+    """
+    A robust dataset curator for processing Allama Iqbal's poetry collection
+    with nested YAML structures into a flattened JSON format optimized for RAG.
+    
+    Features:
+    - Hierarchical data flattening
+    - Multilingual support (with English focus)
+    - Nested structure resolution
+    - Metadata preservation
+    - Data validation and error handling
+    """
+    
+    def __init__(self, data_path: str, output_dir: str):
+        """
+        Initialize the curator with validated paths
+        
+        Args:
+            data_root (str): Root directory containing 'lists' and 'poems' folders
+            output_dir (str): Directory for saving processed datasets
+        """
+        self.data_root = Path(data_path)
+        self.output_dir = Path(output_dir)
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+        
+        self.dataset = {
+            "metadata": {
+                "total_books": 0,
+                "total_poems": 0
+            },
+            "books": [],
+            "poems": []
+        }
+
+
+    def process_dataset(self, source: str = 'github_iqbal_demystified'):
+        """
+        Process the dataset based on the source.
+        """
+        if source == 'github_iqbal_demystified':
+            self.data_root = self.data_root / source
+            self.dataset = self.process_github_iqbal_demystified()
+            logger.info(f"Dataset processed successfully")
+            # logger.debug(f"Dataset: {self.dataset}")
+        else:
+            raise ValueError(f"Unsupported source: {self.source}")
+        
+        # Save the dataset to various formats
+        self._save_dataset()
+        
+        return self.dataset
+    
+    
+    def process_github_iqbal_demystified(self):
+        """
+        Main processing pipeline with error handling and progress tracking
+        """
+        try:
+            book_files = sorted((self.data_root / "lists").glob("List_*.yaml"))
+            logger.info(f"Found {len(book_files)} book files to process")
+            
+            for book_file in tqdm(book_files, desc="Processing books"):
+                book_data = self._load_yaml(book_file)
+                book_id = book_file.stem.split("_")[-1]
+                processed_book = self._process_book(book_id, book_data)
+                self.dataset["books"].append(processed_book)
+                
+                poems = self._process_poems(book_id, processed_book)
+                self.dataset["poems"].extend(poems)          
+                self.dataset["metadata"]["total_poems"] += len(poems)
+            self.dataset["metadata"]["total_books"] = len(self.dataset["books"])  
+            
+            return self.dataset
+            
+        except Exception as e:
+            logger.error(f"Processing failed: {str(e)}")
+            return None
+
+    def _process_book(self, book_id: str, raw_data: Dict) -> Dict:
+        """
+        Process book metadata with nested section structure
+        
+        Args:
+            book_id (str): Unique identifier for the book
+            raw_data (Dict): Raw YAML data from list file
+            
+        Returns:
+            Dict: Processed book structure with flattened metadata
+        """
+        book_structure = {
+            "id": book_id,
+            "titles": {},
+            "sections": [],
+            "metadata": {"total_sections": 0, "total_poems": 0}
+        }
+        
+        # Process multilingual titles
+        for title_entry in raw_data.get("name", []):
+            lang = title_entry.get("lang", "unknown")
+            book_structure["titles"][lang] = title_entry.get("text", "")
+            if lang == 'en':
+                book_structure['primary_title'] = title_entry.get("text", "Unknown")
+        
+        # Process sections
+        current_section = None
+        for section_data in raw_data.get("sections", []):
+            if "sectionName" in section_data:
+                if current_section:
+                    book_structure["sections"].append(current_section)
+                    book_structure["metadata"]["total_sections"] += 1
+                
+                current_section = {
+                    "id": len(book_structure["sections"]) + 1,
+                    "titles": {},
+                    "poems": [],
+                    "poem_ids": [],
+                    "metadata": {"total_poems": 0}
+                }
+                
+                for name_entry in section_data["sectionName"]:
+                    lang = name_entry.get("lang", "unknown")
+                    current_section["titles"][lang] = name_entry.get("text", "")
+            
+            if "poems" in section_data and current_section:
+                poems = self._process_poem_metadata(section_data["poems"])
+                poem_ids = [poem['id'] for poem in poems]
+                current_section["poems"].extend(poems)
+                current_section["poem_ids"].extend(poem_ids)
+                current_section["metadata"]["total_poems"] += len(poems)
+        
+        if current_section:
+            book_structure["sections"].append(current_section)
+            book_structure["metadata"]["total_sections"] += 1
+        
+        book_structure["metadata"]["total_poems"] = sum(
+            len(s["poems"]) for s in book_structure["sections"]
+        )
+        return book_structure
+    
+
+    def _process_poem_metadata(self, poems: List[Dict]) -> List[Dict]:
+        """
+        Flatten poem metadata from nested structure
+        
+        Args:
+            poems (List[Dict]): Raw poem metadata entries
+            
+        Returns:
+            List[Dict]: Processed poem metadata
+        """
+        processed = []
+        for poem in poems:
+            processed_poem = {
+                "id": poem.get("id", ""),
+                "titles": {},
+                "metadata": {"languages": []}  # Changed from set to list
+            }
+            
+            for title_entry in poem.get("poemName", []):
+                lang = title_entry.get("lang", "unknown")
+                processed_poem["titles"][lang] = title_entry.get("text", "")
+                if lang not in processed_poem["metadata"]["languages"]:
+                    processed_poem["metadata"]["languages"].append(lang)
+            
+            processed.append(processed_poem)
+        return processed
+    
+
+    def _process_poems(self, book_id: str, book_data: Dict) -> List[Dict]:
+        """
+        Process poem content files with validation and error handling
+        
+        Args:
+            book_id (str): Parent book identifier
+            book_data (Dict): Processed book structure
+            
+        Returns:
+            List[Dict]: Processed poems with flattened content
+        """
+        poems = []
+        book_name = book_data.get("primary_title", f"book_{book_id}")
+        sections = book_data.get("sections", [])
+        poem_dir = self.data_root / "poems" / book_id
+        
+        if not poem_dir.exists():
+            logger.warning(f"Missing poem directory for book: {book_id}:{book_name}")
+            return []
+            
+        for poem_file in poem_dir.glob("*.yaml"):
+            try:
+                poem_id = poem_file.stem
+                raw_data = self._load_yaml(poem_file)
+                
+                # Create the generator expression, broken for readability
+                sectioninfo_generator = (
+                    (section_info.get('id'), section_info.get('titles', {}).get('en'))
+                    for section_info in sections
+                    if poem_id in section_info.get('poem_ids', [])
+                )
+                # Use next() with the generator and a default tuple
+                section_id, section_name = next(sectioninfo_generator, (None, None))
+                # Create poem structure
+                poem = {
+                    "id": poem_id,
+                    "book_id": book_id,
+                    "book_title": book_name,
+                    "section_id": section_id,
+                    "section_title": section_name,
+                    "metadata": {"languages": []},
+                    "content": {"descriptions": {}, "verses": []}
+                }
+                
+                # Process descriptions
+                for desc_entry in raw_data.get("description", []):
+                    lang = desc_entry.get("lang", "unknown")
+                    poem["content"]["descriptions"][lang] = desc_entry.get("text", "")
+                    if lang not in poem["metadata"]["languages"]:
+                        poem["metadata"]["languages"].append(lang)
+                
+                # Process verses with language detection
+                for verse in raw_data.get("sher", []):
+                    processed_verse = self._process_verse(verse)
+                    poem["content"]["verses"].append(processed_verse)
+                    # Detect verse languages
+                    for content in verse.get("sherContent", []):
+                        lang = content.get("lang", "unknown")
+                        if lang not in poem["metadata"]["languages"]:
+                            poem["metadata"]["languages"].append(lang)
+                
+                # Flatten structure with complete English detection
+                rag_poem = self._flatten_for_rag(poem)
+                if rag_poem:  # Only add if English content exists
+                    poems.append(rag_poem)
+            except Exception as e:
+                logger.error(f"Failed processing poem {poem_id}: {str(e)}")
+        
+        return poems
+
+    def _process_verse(self, verse: Dict) -> Dict:
+        """
+        Process individual verse with multilingual content
+        
+        Args:
+            verse (Dict): Raw verse data from YAML
+            
+        Returns:
+            Dict: Processed verse structure
+        """
+        processed = {
+            "id": verse.get("id", ""),
+            "content": {},
+            "notes": []
+        }
+        
+        for content_entry in verse.get("sherContent", []):
+            lang = content_entry.get("lang", "unknown")
+            processed["content"][lang] = {
+                "text": content_entry.get("text", ""),
+                "notes": [self._process_note(n) for n in content_entry.get("notes", [])]
+            }
+        
+        return processed
+    
+
+    def _process_note(self, note: Dict) -> Dict:
+        """
+        Standardize phrase/note structure
+        
+        Args:
+            note (Dict): Raw note data
+            
+        Returns:
+            Dict: Processed note structure
+        """
+        return {
+            "phrase": note.get("phrase", ""),
+            "meaning": note.get("meaning", ""),
+            "occurrences": note.get("occurrence", 1)
+        }
+    
+
+    def _flatten_for_rag(self, poem: Dict) -> Dict:
+        """
+        Transform poem structure into RAG-optimized format
+        
+        Args:
+            poem (Dict): Original poem structure
+            
+        Returns:
+            Dict: Flattened structure with combined text fields
+        """
+        rag_poem = {
+            "poem_id": poem["id"],
+            "book_id": poem["book_id"],
+            "book_title": poem["book_title"],
+            "section_id": poem["section_id"],
+            "section_title": poem["section_title"],
+            "text_blocks": [],
+            "full_text": ""
+        }
+
+        # Extract English content from all sources
+        en_content = {
+            "descriptions": poem["content"]["descriptions"].get("en", ""),
+            "verses": [],
+            "phrases": []
+        }
+
+        # Process verses
+        for verse in poem["content"]["verses"]:
+            if "en" in verse["content"]:
+                en_content["verses"].append(verse["content"]["en"]["text"])
+                en_content["phrases"].extend(
+                    f"{note['phrase']}: {note['meaning']}"
+                    for note in verse["content"]["en"].get("notes", [])
+                )
+
+        # Build full text if English content exists
+        if en_content["verses"]:
+            rag_poem["full_text"] = "\n\n".join([
+                en_content["descriptions"],
+                "\n".join(en_content["verses"])
+            ])
+            rag_poem["text_blocks"] = en_content["verses"]
+            rag_poem["phrases"] = en_content["phrases"]
+            return rag_poem
+        
+        logger.warning(f"No English content found for poem {poem['id']}")
+        return None
+    
+
+    def _save_dataset(self):
+        """Save datasets with proper serialization checks"""
+        base_path = self.output_dir / "iqbal_poems"
+        
+        # Save full dataset
+        with open(f"{base_path}_full.json", "w", encoding="utf-8") as f:
+            json.dump(self.dataset, f, ensure_ascii=True, indent=2)
+        
+        # Save RAG-optimized poems (only those with English content)
+        rag_data = [p for p in self.dataset["poems"] if p is not None]
+        
+        with open(f"{base_path}_rag.json", "w", encoding="utf-8") as f:
+            json.dump(rag_data, f, ensure_ascii=True, indent=2)
+        
+        logger.info(f"Saved {len(rag_data)} RAG-ready poems")
+
+    def _load_yaml(self, path: Path) -> Dict:
+        """
+        Safe YAML loader with validation
+        
+        Args:
+            path (Path): Path to YAML file
+            
+        Returns:
+            Dict: Parsed YAML content
+        """
+        try:
+            with open(path, "r", encoding="utf-8") as f:
+                return yaml.safe_load(f)
+        except Exception as e:
+            logger.error(f"Failed loading YAML from {path}: {str(e)}")
+            raise

utils/dataset_downloader.py

@@ -0,0 +1,183 @@
+# External imports
+import os
+import requests
+import yaml
+import time
+import logging
+
+from bs4 import BeautifulSoup
+from pathlib import Path
+from tqdm import tqdm
+
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.FileHandler("dataset_downloader.log"),
+        logging.StreamHandler()
+    ]
+)
+logger = logging.getLogger("DatasetDownloader")
+
+
+# Constants at module level
+SOURCES = {
+    "github_iqbal_demystified": "https://raw.githubusercontent.com/AzeemGhumman/iqbal-demystified-dataset/master/data",
+    "iqbal_cyberlibrary": "https://iqbalcyberlibrary.net",
+    "allama_iqbal_poetry": "https://blogs.library.mcgill.ca/islamicstudieslibrary/allama-iqbal-poetry-%DA%A9%D9%84%D8%A7%D9%85-%D8%B9%D9%84%D8%A7%D9%85%DB%81-%D9%85%D8%AD%D9%85%D8%AF-%D8%A7%D9%82%D8%A8%D8%A7%D9%84/",
+    "iqbal_review": "https://www.allamaiqbal.com/publications/journals/review/",
+    "rekhta": "https://www.rekhta.org/poets/allama-iqbal/ghazals"
+}
+
+
+class DatasetDownloader:
+    """
+    A class to download dataset from various sources.
+    """
+    
+    def __init__(self, output_dir: str = "data", number_of_books: int = 11, max_workers: int = 5) -> None:
+        """Initialize the dataset downloader with configuration parameters.
+        
+        Args:
+            output_dir (str): Directory to store the downloaded files. Defaults to "data".
+            max_workers (int): Maximum number of concurrent workers. Defaults to 5.
+        """
+        if max_workers < 1:
+            raise ValueError("max_workers must be at least 1")
+    
+        self.output_dir = Path(output_dir)
+        self.max_workers = max_workers
+        self.number_of_books = number_of_books
+        # Constant variables
+        self.sources = SOURCES
+
+
+    def download_from_github(self, source_name: str = "github_iqbal_demystified"):
+        """Download dataset from GitHub."""
+        logger.info("Downloading dataset from GitHub")
+
+        # Check if the source name is valid
+        if source_name not in self.sources:
+            raise ValueError(f"Source name {source_name} not found in sources")
+        
+        # Get the source name and base url
+        base_url = self.sources[source_name]
+        folders = ["lists", "poems"]
+        
+        # Create the folders for the source
+        for folder in folders:
+            output_path = self.output_dir / source_name / folder
+            os.makedirs(output_path, exist_ok=True)
+
+        # Fetch the list metadata from the GitHub repository
+        book_ids = self._download_github_lists(source_name, base_url, folder="lists")
+        # Fetch the poems from the GitHub repository
+        poem_ids = self._download_github_poems(source_name, base_url, folder="poems", book_ids=book_ids)
+
+        logger.info(f"Completed fetching data from Iqbal Demystified GitHub repository. Total poems fetched: {len(poem_ids)}")
+
+    
+    def _download_github_lists(self, source_name: str, base_url: str, folder: str) -> list:
+        """Fetch the list metadata from the GitHub repository."""
+
+        logger.info(f"Fetching book metadata from {folder} folder")
+        
+        book_ids = []
+        # Fetch the metadata for each book along with the poems
+        for index in tqdm(range(self.number_of_books), desc="Fetching book metadata"):
+            book_id = f"{index+1:03}"
+            # Create the output path for the book
+            output_path = self.output_dir / source_name / folder / f"List_{book_id}.yaml"
+            # Fetch the metadata for the book using requests
+            metadata_url = f"{base_url}/lists/List_{book_id}.yaml"
+            # Skip if already downloaded
+            if output_path.exists():
+                logger.debug(f"List_{book_id}.yaml already exists, skipping download")
+                book_ids.append(book_id)
+                continue
+
+            try:
+                response = requests.get(metadata_url)
+                response.raise_for_status()
+                if response.status_code == 200:
+                    with open(output_path, "w", encoding="utf-8") as f:
+                        f.write(response.text)
+                
+                    book_ids.append(book_id)
+                    logger.info(f"Successfully fetched List_{book_id}.yaml")
+            except Exception as e:
+                logger.error(f"Error fetching metadata for {book_id}: {e}")
+            
+            # Respect rate limits
+            time.sleep(0.5)
+
+        logger.info(f"Fetched {len(book_ids)} book lists")
+        return book_ids
+    
+    
+    def _download_github_poems(self, source_name: str, base_url: str, folder: str, book_ids: list) -> list:
+        """Fetch the poems from the GitHub repository."""
+        # List to store the fetched poems
+        fetched_poems = []
+        # Fetch the poems for each book by first reading the list metadata and then fetching the poems
+        for id in tqdm(book_ids, desc=f"Fetching books metadata, poems and shers"):
+            metadata_path = self.output_dir / source_name / "lists" / f"List_{id}.yaml"
+            if not metadata_path.exists():
+                logger.error(f"Metadata file for book {id} does not exist")
+                continue
+
+            # Create directory for this book's poems
+            poems_path = self.output_dir / source_name / folder / id
+            os.makedirs(poems_path, exist_ok=True)
+
+            # Load and parse the list file
+            try:
+                with open(metadata_path, "r", encoding="utf-8") as f:
+                    book_metadata = yaml.safe_load(f)
+            except Exception as e:
+                logger.error(f"Error parsing list file for book {id}: {str(e)}")
+                continue
+ 
+            # Extract all poem IDs from the list
+            poem_ids = []
+            for section in book_metadata.get('sections', []):
+                if 'poems' in section:
+                    for poem in section['poems']:
+                        if 'id' in poem:
+                            poem_ids.append(poem['id'])
+            
+            # Fetch each poem
+            fetched_poems = []
+            for poem_id in tqdm(poem_ids, desc=f"Fetching poems for book {id}"):
+                poem_url = f"{base_url}/poems/{id}/{poem_id}.yaml"
+                output_path = poems_path / f"{poem_id}.yaml"
+                
+                # Skip if already downloaded
+                if output_path.exists():
+                    logger.debug(f"Poem {poem_id} already exists, skipping download")
+                    fetched_poems.append(poem_id)
+                    continue
+                
+                try:
+                    response = requests.get(poem_url, timeout=10)
+                    
+                    if response.status_code == 200:
+                        with open(output_path, "w", encoding="utf-8") as f:
+                            f.write(response.text)
+                        
+                        fetched_poems.append(poem_id)
+                        print(f"Successfully fetched poem {poem_id}")
+                    else:
+                        print(f"Failed to fetch poem {poem_id}: {response.status_code}")
+                    
+                    # Respect rate limits
+                    time.sleep(0.5)
+                    
+                except Exception as e:
+                    logger.error(f"Error fetching poem {poem_id}: {str(e)}")
+
+            logger.info(f"Fetched {len(fetched_poems)} poems for book {id}")
+        return fetched_poems
+

utils/error_handling.py

@@ -0,0 +1,15 @@
+"""Error handling utilities."""
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+def handle_rag_error(func):
+    """Decorator for handling RAG system errors."""
+    def wrapper(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except Exception as e:
+            logger.error(f"Error in RAG system: {str(e)}")
+            return f"An error occurred: {str(e)}", []
+    return wrapper