"""Main Mulitmodal-RAG pipeline script.""" import os import torch import fitz #PyMuPDF import faiss import re import gc import numpy as np from typing import List, Dict, Tuple from PIL import Image from langchain.text_splitter import RecursiveCharacterTextSplitter from transformers import TextIteratorStreamer from utils import ( save_cache, load_cache, init_faiss_indexflatip, add_embeddings_to_index, search_faiss_index, save_faiss_index, load_faiss_index, cleanup_images, clear_gpu_cache ) from model_setup import embedding_model, model, processor torch.set_num_threads(4) # Just being agnostic device = "cuda" if torch.cuda.is_available() else "cpu" # Function to extract text and images from each page of the PDF # This function uses PyMuPDF (fitz) to extract text and images from each page def extract_pages_text_and_images(pdf_path, image_dir): """Extract text and images page-wise""" doc = fitz.open(pdf_path) os.makedirs(image_dir, exist_ok=True) page_texts = [] page_images = [] for page_num in range(len(doc)): page = doc.load_page(page_num) text = page.get_text() # Store all images on this page, store their paths images = [] for img_index, img in enumerate(page.get_images(full=True)): xref = img[0] base_image = doc.extract_image(xref) image_bytes = base_image["image"] image_ext = base_image["ext"] image_filename = f"page_{page_num + 1}_img_{img_index}.{image_ext}" image_path = os.path.join(image_dir, image_filename) with open(image_path, "wb") as img_file: img_file.write(image_bytes) images.append(image_path) page_texts.append(text) page_images.append(images) if doc: doc.close() return page_texts, page_images # Generate image descriptions using the Gemma3 model # This function will be called in parallel for each page's images def generate_image_descriptions(image_paths): """Generate images and tables descriptions as texts.""" captions = [] if not processor or not model: print("[ERROR] Model or Processor not loaded. Cannot generate image descriptions.") return [] for image_path in image_paths: raw_image = Image.open(image_path) if raw_image.mode != "RGB": image = raw_image.convert("RGB") else: image = raw_image width, height = image.size if width < 32 or height < 32: # Filtering out smaller images that may disrupt the process continue messages = [ {"role": "system", "content": [{"type": "text", "text": "You are a helpful assistant."}]}, {"role": "user", "content": [ {"type": "image", "image": image}, {"type": "text", "text": "Describe the factual content visible in the image. Be concise and accurate as the descriptions will be used for retrieval."} ]} ] try: inputs = processor.apply_chat_template(messages, add_generation_prompt=True, tokenize=True, return_dict=True, return_tensors="pt").to("cpu", dtype=torch.bfloat16) input_len = inputs["input_ids"].shape[-1] # To get rid of the prompt echo with torch.inference_mode(): generated_ids = model.generate( **inputs, max_new_tokens=512, do_sample=False, cache_implementation="offloaded_static" ) raw = processor.decode(generated_ids[0], skip_special_tokens=True) caption = clean_caption(raw) captions.append({"image_path": image_path, "caption": caption}) except Exception as e: print(f"[⚠️ ERROR]: Failed to generate caption for image {image_path}: {e}") captions.append({"image_path": image_path, "caption": "<---image---> (Captioning failed)"}) # Add a placeholder caption continue finally: gc.collect() clear_gpu_cache() return captions # Cleaning the captions from the extracted images # Regex: match everything from "model\n" up through first double-newline after "sent:" prefix_re = re.compile( r"model\s*\n.*?\bsent:\s*\n\n", flags=re.IGNORECASE | re.DOTALL, ) def clean_caption(raw: str) -> str: # 1. Strip off the prompt/header by splitting once. parts = prefix_re.split(raw.strip(), maxsplit=1) if len(parts) == 2: return parts[1].strip() # 2. Fallback: if the caption begins with ** (bold header), return from there. bold_index = raw.find("**") if bold_index >= 0: return raw[bold_index:].strip() # 3. Last resort: return everything except the first paragraph. paras = raw.strip().split('\n\n', 1) return paras[-1].strip() # might still include some leading noise # Generate captions for all images on each page def generate_captions_per_page(page_image_paths_list): """"Generate captions per page's images""" page_captions = [] for image_paths in page_image_paths_list: captions = generate_image_descriptions(image_paths) # Extract the 'caption' strings only captions_texts = [cap['caption'] for cap in captions] page_captions.append(captions_texts) return page_captions # Merge text and captions for each page # This function combines the text and captions for each page into a single string def merge_text_and_captions(page_texts, page_captions): """Merge text, image captions and table descriptions per page""" combined_pages = [] for page_num, (text, captions) in enumerate(zip(page_texts, page_captions), 1): page_content = text.strip() + "\n\n" for cap in captions: page_content += f"[Image Description]: {cap}\n\n" combined_pages.append(page_content) return combined_pages # Chunk the merged pages into smaller text chunks with metadata # This function splits the combined text of each page into smaller chunks def chunk_text_with_metadata(merged_pages): """ Given a list of pages (strings) with combined text and image captions, split each page's content into chunks, attach metadata, and collect all chunks. Args: merged_pages (List[str]): List where each item is the content (text + captions) of a single page. Returns: List[dict]: List of chunked dicts with keys: content, page, chunk_id, type """ text_splitter = RecursiveCharacterTextSplitter( separators=["\n\n", "\n", ".", " ", ""], # Recursive splitting separators, from paragraphs to words chunk_size =1000, chunk_overlap =200, add_start_index=True ) all_chunks = [] chunk_global_id = 0 for page_num, page_content in enumerate(merged_pages, start=1): # Split page content into chunks page_chunks = text_splitter.split_text(page_content) # Tag metadata on each chunk for chunk_num, chunk_text in enumerate(page_chunks, start=1): chunk_dict = { "content": chunk_text, "page": page_num, "chunk_id": chunk_global_id, "chunk_number_on_page": chunk_num, "type": "extracted_texts_and_captions_descriptions" } all_chunks.append(chunk_dict) chunk_global_id += 1 return all_chunks # Preprocess the uploaded PDF def preprocess_pdf(file_path: str, image_dir: str, embedding_model, index_file: str = "index.faiss", chunks_file: str = "chunks.json", use_cache: bool = True) -> Tuple[faiss.IndexFlatIP, List[Dict]]: if not os.path.exists(file_path): raise FileNotFoundError(f"PDF not found at {file_path}") # Loading cache to save on compute time and resources everytime a query is made if use_cache and os.path.exists(index_file) and os.path.exists(chunks_file): print("[INFO] Loading cached FAISS index and chunks...") index = load_faiss_index(index_file) chunks = load_cache(chunks_file) return index, chunks # Cleanup stale cache if not using it or if missing if not use_cache or not (os.path.exists(index_file) and os.path.exists(chunks_file)): if os.path.exists(index_file): os.remove(index_file) if os.path.exists(chunks_file): os.remove(chunks_file) # Otherwise run full processing try: page_texts, page_images = extract_pages_text_and_images(file_path, image_dir) except Exception as e: print(f"Error reading PDF: {e}") raise e page_captions = generate_captions_per_page(page_images) merged_pages = merge_text_and_captions(page_texts, page_captions) # Delete extracted images after captioning cleanup_images(image_dir) # Chunk the merged pages chunks = chunk_text_with_metadata(merged_pages) texts = [chunk['content'] for chunk in chunks] # Geenrate embeddings and initialize faiss index with the dimensions of the embeddings embeddings = embedding_model.encode(texts, normalize_embeddings=True) embeddings = embeddings.astype(np.float32) # Making sure embeddings are in float32 format for FAISS embedding_dim = embeddings.shape[1] index = init_faiss_indexflatip(embedding_dim=embedding_dim) # Add embeddings to index add_embeddings_to_index(index=index, embeddings=embeddings) # Save index and chunks if use_cache: save_faiss_index(index, index_file) save_cache(chunks, chunks_file) return index, chunks # Semantic search funtion that uses preprocessed data def semantic_search(query, embedding_model, index, chunks, top_k=10): # Embed user query query_embedding = embedding_model.encode([query], normalize_embeddings=True) # Retrieve top matches from FAISS distances, indices = search_faiss_index(index, query_embedding, k=top_k) # Retrieve matched chunks retrieved_chunks = [chunks[i] for i in indices[0]] return retrieved_chunks # Generate answer for Gradio interface def generate_answer_stream(query, retrieved_chunks, model, processor): """Feeds tokens gradually from LLM.""" context_texts = [chunk['content'] for chunk in retrieved_chunks] # Combine system instruction, context, and query into a single string for the user role system_instruction = """You are a helpful and precise assistant for question-answering tasks. Use only the following pieces of retrieved context to answer the question. You may provide the response in a structured markdown response if necessary. If the answer is not found in the provided context, state that the information is not available in the document. Do not use any external knowledge or make assumptions. """ # Build the core prompt string, excluding specific turn markers # The processor.apply_chat_template will handle the proper formatting rag_prompt_content = "" if system_instruction: rag_prompt_content += f"{system_instruction.strip()}\n\n" if context_texts: rag_prompt_content += "Context:\n" +"-"+ "\n-".join(context_texts).strip() + "\n\n" rag_prompt_content += f"Question: {query.strip()}\nAnswer:" # Robust format for multimodal processor messages = [ {"role": "user", "content": [{"type": "text", "text": rag_prompt_content}]} ] # Prepare model inputs using apply_chat_template # This will correctly format the prompt for Gemma 3 inputs = processor.apply_chat_template( messages, add_generation_prompt=True, # Tell the model it is the start of its turn tokenize=True, return_dict=True, return_tensors="pt", truncation=True, max_length=4096 # Apply max_length here if needed, truncation will handle it ).to("cpu") streamer = TextIteratorStreamer(processor.tokenizer, skip_prompt=True, decode_kwargs={"skip_special_tokens": True}) with torch.inference_mode(): model.generate(**inputs, streamer=streamer, use_cache=True, max_new_tokens=512) gc.collect() # Free memory after model generation accumulated = "" for new_text in streamer: # time.sleep(0.2) accumulated += new_text yield accumulated # Free memory after streaming is complete clear_gpu_cache() gc.collect()