main.py

# file: main.py
import time
import os
import asyncio
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, HttpUrl
from typing import List, Dict, Any
from dotenv import load_dotenv

# Assuming 'ingestion_router.py' is in the same directory and contains the function
from ingestion_router import ingest_and_parse_document
from chunking_parent import create_parent_child_chunks
from embedding import EmbeddingClient
from retrieval_parent import Retriever
from generation import generate_answer

load_dotenv()

app = FastAPI(
    title="Modular RAG API",
    description="A modular API for Retrieval-Augmented Generation with Parent-Child Retrieval.",
    version="2.3.0", # Updated version
)

GROQ_API_KEY = os.environ.get("GROQ_API_KEY")
embedding_client = EmbeddingClient()
retriever = Retriever(embedding_client=embedding_client)

# --- Pydantic Models ---
class RunRequest(BaseModel):
    documents: HttpUrl
    questions: List[str]

class RunResponse(BaseModel):
    answers: List[str]

class TestRequest(BaseModel):
    documents: HttpUrl

# --- NEW: Test Endpoint for Ingestion and Parsing ---
@app.post("/test/ingestion", response_model=Dict[str, Any], tags=["Testing"])
async def test_ingestion_endpoint(request: TestRequest):
    """
    Tests the complete ingestion and parsing pipeline.
    Downloads a document from a URL, processes it using the modular
    parsing strategy (e.g., parallel for PDF, standard for DOCX),
    and returns the extracted Markdown content and time taken.
    """
    print("--- Running Document Ingestion & Parsing Test ---")
    start_time = time.perf_counter()
    try:
        # Step 1: Call the main ingestion function from your router
        markdown_content = await ingest_and_parse_document(request.documents)

        end_time = time.perf_counter()
        duration = end_time - start_time
        print(f"--- Ingestion and Parsing took {duration:.2f} seconds ---")

        if not markdown_content:
            raise HTTPException(
                status_code=404,
                detail="Document processed, but no content was extracted."
            )

        return {
            "total_time_seconds": duration,
            "character_count": len(markdown_content),
            "extracted_content": markdown_content,
        }
    except Exception as e:
        # Catch potential download errors, parsing errors, or unsupported file types
        raise HTTPException(status_code=500, detail=f"An error occurred during ingestion test: {str(e)}")


# --- Test Endpoint for Parent-Child Chunking ---
@app.post("/test/chunk", response_model=Dict[str, Any], tags=["Testing"])
async def test_chunking_endpoint(request: TestRequest):
    """
    Tests the parent-child chunking strategy.
    Returns parent chunks, child chunks, and the time taken.
    """
    print("--- Running Parent-Child Chunking Test ---")
    start_time = time.perf_counter()

    try:
        # Step 1: Parse the document to get raw text
        markdown_content = await ingest_and_parse_document(request.documents)
        
        # Step 2: Create parent and child chunks
        child_documents, docstore, _ = create_parent_child_chunks(markdown_content)
        
        end_time = time.perf_counter()
        duration = end_time - start_time
        print(f"--- Parsing and Chunking took {duration:.2f} seconds ---")

        # Convert Document objects to a JSON-serializable list for the response
        child_chunk_results = [
            {"page_content": doc.page_content, "metadata": doc.metadata}
            for doc in child_documents
        ]
        
        # Retrieve parent documents from the in-memory store
        parent_docs = docstore.mget(list(docstore.store.keys()))
        parent_chunk_results = [
            {"page_content": doc.page_content, "metadata": doc.metadata}
            for doc in parent_docs if doc
        ]
        
        return {
            "total_time_seconds": duration,
            "parent_chunk_count": len(parent_chunk_results),
            "child_chunk_count": len(child_chunk_results),
            "parent_chunks": parent_chunk_results,
            "child_chunks": child_chunk_results,
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"An error occurred during chunking test: {str(e)}")


@app.post("/hackrx/run", response_model=RunResponse)
async def run_rag_pipeline(request: RunRequest):
    try:
        print("--- Kicking off RAG Pipeline with Parent-Child Strategy ---")
        
        # --- STAGE 1: DOCUMENT INGESTION ---
        markdown_content = await ingest_and_parse_document(request.documents)
        
        # --- STAGE 2: PARENT-CHILD CHUNKING ---
        child_documents, docstore, _ = create_parent_child_chunks(markdown_content)

        if not child_documents:
            raise HTTPException(status_code=400, detail="Document could not be processed into chunks.")

        # --- STAGE 3: INDEXING ---
        retriever.index(child_documents, docstore)

        # --- STAGE 4: CONCURRENT RETRIEVAL & GENERATION ---
        print("Starting retrieval for all questions...")
        retrieval_tasks = [
            retriever.retrieve(q, GROQ_API_KEY)
            for q in request.questions
        ]
        all_retrieved_chunks = await asyncio.gather(*retrieval_tasks)
        print("Retrieval complete. Starting final answer generation...")

        answer_tasks = [
            generate_answer(q, chunks, GROQ_API_KEY)
            for q, chunks in zip(request.questions, all_retrieved_chunks)
        ]
        final_answers = await asyncio.gather(*answer_tasks)

        print("--- RAG Pipeline Completed Successfully ---")
        return RunResponse(answers=final_answers)

    except Exception as e:
        raise HTTPException(
            status_code=500, detail=f"An internal server error occurred: {str(e)}"
        )