FinRAG / scripts /ingest_all.py
Harshavard21's picture
Initial commit
454f118
Raw History Blame Contribute Delete
7.57 kB
"""
scripts/ingest_all.py
======================
Master ingestion script — run this ONCE to process all BSE PDFs.
What it does:
1. Parses all 23 PDFs (text + tables + OCR fallback)
2. Chunks them hierarchically (parent-child + atomic tables)
3. Embeds all index chunks using BAAI/bge-large (GPU-accelerated)
4. Stores in Qdrant HNSW vector index
5. Builds BM25 sparse keyword index
6. Prints a summary report
Estimated time on RTX 3050:
- Parsing: ~3-5 minutes (23 PDFs)
- Embedding: ~5-10 minutes (GPU accelerated)
- Total: ~10-15 minutes
Run with:
python scripts/ingest_all.py
To force re-process (ignore cache):
python scripts/ingest_all.py --force
"""
import sys
import os
import argparse
import time
# FORCE UTF-8 output on Windows (fixes Rich Unicode encoding errors)
os.environ["PYTHONUTF8"] = "1"
if sys.platform == "win32":
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
# Project root on path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# CRITICAL: import config FIRST to set USE_TF=0 before any ML imports
from config.settings import settings
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from src.utils.logger import logger
from src.ingestion.pipeline import IngestionPipeline
from src.chunking.hierarchical_chunker import HierarchicalChunker
from src.embeddings.embedder import Embedder
from src.vectorstore.qdrant_store import QdrantStore
from src.vectorstore.bm25_index import BM25Index
console = Console(force_terminal=True)
def parse_args():
parser = argparse.ArgumentParser(description="BSE Financial RAG — Ingestion Pipeline")
parser.add_argument(
"--force",
action="store_true",
help="Force re-processing of all PDFs (ignore cache)",
)
parser.add_argument(
"--recreate-index",
action="store_true",
help="Drop and recreate Qdrant collection (use when changing embedding model)",
)
parser.add_argument(
"--skip-ocr",
action="store_true",
help="Skip OCR for scanned pages (faster, may miss some content)",
)
return parser.parse_args()
def main():
args = parse_args()
console.print(Panel.fit(
"[bold cyan]BSE Financial RAG -- Ingestion Pipeline[/bold cyan]\n"
"[dim]Production-grade PDF -> Vector DB pipeline[/dim]",
border_style="cyan"
))
console.print()
# Ensure all data directories exist
settings.ensure_dirs()
start_time = time.time()
# ---------------------------------------------------------------- #
# Step 1: Ingest PDFs
# ---------------------------------------------------------------- #
console.rule("[bold]Step 1/5: PDF Ingestion[/bold]")
console.print(f"[dim]Processing PDFs from: {settings.raw_pdfs_path}[/dim]\n")
pipeline = IngestionPipeline()
documents = pipeline.run_all(force_reprocess=args.force)
console.print(f"\n✅ Ingested [bold green]{len(documents)}[/bold green] documents")
# Show ingestion summary table
summary_table = Table(title="Ingested Documents", show_lines=True)
summary_table.add_column("Company", style="cyan")
summary_table.add_column("Fiscal Year", style="yellow")
summary_table.add_column("Source File", style="dim")
summary_table.add_column("Pages", justify="right")
summary_table.add_column("Content Items", justify="right")
for doc in documents:
summary_table.add_row(
doc.company,
doc.fiscal_year,
doc.source_file,
str(doc.total_pages),
str(len(doc.items)),
)
console.print(summary_table)
console.print()
# ---------------------------------------------------------------- #
# Step 2: Chunk Documents
# ---------------------------------------------------------------- #
console.rule("[bold]Step 2/5: Hierarchical Chunking[/bold]")
chunker = HierarchicalChunker()
chunked_docs = chunker.chunk_documents(documents)
# Collect all index chunks (child + atomic)
all_index_chunks = []
all_parent_chunks = []
for cdoc in chunked_docs:
all_index_chunks.extend(cdoc.all_index_chunks)
all_parent_chunks.extend(cdoc.parent_chunks)
console.print(f"\n✅ Chunking complete:")
console.print(f" Parent chunks : [bold]{len(all_parent_chunks)}[/bold]")
console.print(f" Index chunks : [bold]{len(all_index_chunks)}[/bold] (child + atomic tables)")
console.print()
# ---------------------------------------------------------------- #
# Step 3: Embed Index Chunks
# ---------------------------------------------------------------- #
console.rule("[bold]Step 3/5: Embedding (GPU: RTX 3050)[/bold]")
console.print(f"[dim]Model: {settings.embedding_model} | Batch size: {settings.embed_batch_size}[/dim]\n")
embedder = Embedder()
texts_to_embed = [chunk.content for chunk in all_index_chunks]
embeddings = embedder.embed_documents(texts_to_embed)
console.print(f"\n✅ Embedded [bold green]{len(embeddings)}[/bold green] chunks | shape: {embeddings.shape}")
console.print()
# ---------------------------------------------------------------- #
# Step 4: Store in Qdrant
# ---------------------------------------------------------------- #
console.rule("[bold]Step 4/5: Qdrant Vector Store[/bold]")
qdrant = QdrantStore()
qdrant.create_collection(recreate=args.recreate_index)
# Also upsert parent chunks (for parent lookup during retrieval)
# Parents are stored with chunk_level="parent" — not searched, just fetched
console.print("[dim]Upserting parent chunks for context promotion...[/dim]")
parent_texts = [chunk.content for chunk in all_parent_chunks]
if parent_texts:
parent_embeddings = embedder.embed_documents(parent_texts)
qdrant.upsert_chunks(all_parent_chunks, parent_embeddings)
console.print("[dim]Upserting index chunks (child + atomic)...[/dim]")
qdrant.upsert_chunks(all_index_chunks, embeddings)
info = qdrant.get_collection_info()
console.print(f"\n✅ Qdrant store: [bold green]{info['vectors_count']}[/bold green] vectors indexed")
console.print()
# ---------------------------------------------------------------- #
# Step 5: Build BM25 Index
# ---------------------------------------------------------------- #
console.rule("[bold]Step 5/5: BM25 Sparse Index[/bold]")
bm25 = BM25Index()
bm25.build(all_index_chunks)
console.print(f"\n✅ BM25 index built over [bold green]{len(all_index_chunks)}[/bold green] chunks")
console.print()
# ---------------------------------------------------------------- #
# Final Summary
# ---------------------------------------------------------------- #
elapsed = time.time() - start_time
console.print(Panel(
f"[bold green]Ingestion Complete![/bold green]\n\n"
f" Documents ingested : {len(documents)}\n"
f" Parent chunks : {len(all_parent_chunks)}\n"
f" Index chunks : {len(all_index_chunks)}\n"
f" Total vectors : {info['vectors_count']}\n"
f" Time elapsed : {elapsed:.1f}s ({elapsed/60:.1f} min)\n\n"
f"[dim]Next: run [cyan]streamlit run app/main.py[/cyan] to start the app![/dim]",
title="Summary",
border_style="green",
))
if __name__ == "__main__":
main()