import os import faiss import pickle import numpy as np from typing import List, Dict from docarray import Document, DocumentArray from jina import Executor, requests from sentence_transformers import SentenceTransformer from transformers import BlipProcessor, BlipForConditionalGeneration, AutoTokenizer, AutoModelForSeq2SeqLM from pdfminer.high_level import extract_text import fitz # PyMuPDF - better than pdf2image from PIL import Image import traceback import torch import re class MultimodalRAGExecutor(Executor): def __init__( self, llm_model_name: str = "google/flan-t5-large", # FREE and reliable embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2", # FREE embeddings image_model: str = "Salesforce/blip-image-captioning-base", # FREE vision model index_file: str = "faiss_index.bin", metadata_file: str = "metadata.pkl", dim: int = 384, # MiniLM embedding dimension use_huggingface_api: bool = False, **kwargs, ): super().__init__(**kwargs) self.llm_model_name = llm_model_name self.embedding_model = embedding_model self.image_model = image_model self.index_file = index_file self.metadata_file = metadata_file self.dim = dim self.use_huggingface_api = use_huggingface_api # Initialize FREE embedding model print(f"📥 Loading embedding model: {embedding_model}...") self.embedder = SentenceTransformer(embedding_model) print("✅ Embedding model loaded!") # Initialize FREE image captioning model print(f"📥 Loading image model: {image_model}...") try: self.blip_processor = BlipProcessor.from_pretrained(image_model) self.blip_model = BlipForConditionalGeneration.from_pretrained(image_model) print("✅ Image model loaded!") except Exception as e: print(f"⚠️ Warning: Could not load BLIP model: {e}") self.blip_processor = None self.blip_model = None # Initialize LLM self._initialize_llm() # Load or create FAISS index self._load_or_create_index() def _initialize_llm(self): """Initialize LLM (API or local)""" if self.use_huggingface_api: hf_token = os.getenv("HUGGINGFACE_API_KEY") if hf_token: try: from huggingface_hub import InferenceClient self.llm_client = InferenceClient(token=hf_token) self.llm_mode = "api" print("✅ Using HuggingFace API for LLM!") return except Exception as e: print(f"⚠️ HF API initialization failed: {e}") # Load local LLM (FLAN-T5 works best) print(f"📥 Loading local LLM: {self.llm_model_name}...") try: self.tokenizer = AutoTokenizer.from_pretrained(self.llm_model_name) self.llm_model = AutoModelForSeq2SeqLM.from_pretrained(self.llm_model_name) self.llm_mode = "local" print("✅ Local LLM loaded!") except Exception as e: print(f"⚠️ LLM loading failed: {e}") self.llm_mode = "extractive" print("⚠️ Falling back to extractive QA") def _load_or_create_index(self): """Load existing index or create new one""" if os.path.exists(self.index_file) and os.path.exists(self.metadata_file): try: self.index = faiss.read_index(self.index_file) with open(self.metadata_file, "rb") as f: self.metadata = pickle.load(f) print(f"✅ Loaded existing index with {self.index.ntotal} vectors") except Exception as e: print(f"⚠️ Could not load index: {e}. Creating new one.") self.index = faiss.IndexFlatL2(self.dim) self.metadata = [] else: self.index = faiss.IndexFlatL2(self.dim) self.metadata = [] print("✅ Created new empty index") def _get_embedding(self, text: str) -> np.ndarray: """Generate embeddings using sentence-transformers""" embedding = self.embedder.encode(text, convert_to_numpy=True) return embedding.astype(np.float32) def _analyze_image(self, image: Image.Image, context: str = "") -> str: """Analyze image using BLIP model""" if not self.blip_processor or not self.blip_model: return "Image analysis unavailable" try: # Generate caption inputs = self.blip_processor(image, return_tensors="pt") out = self.blip_model.generate(**inputs, max_length=100) caption = self.blip_processor.decode(out[0], skip_special_tokens=True) # Generate detailed description text_prompt = "a detailed description of" inputs = self.blip_processor(image, text_prompt, return_tensors="pt") out = self.blip_model.generate(**inputs, max_length=150) detailed = self.blip_processor.decode(out[0], skip_special_tokens=True) description = f"Caption: {caption}. Details: {detailed}" if context: description += f". Context: {context[:100]}" return description except Exception as e: print(f"Image analysis error: {e}") return "Image analysis failed" def _extract_text_from_pdf(self, pdf_path: str) -> List[str]: """Extract text from PDF using PyMuPDF""" texts = [] try: doc = fitz.open(pdf_path) for page_num, page in enumerate(doc, start=1): text = page.get_text("text") if text and text.strip(): texts.append(f"Page {page_num}:\n{text.strip()}") doc.close() except Exception as e: print(f"Text extraction error: {e}") return texts def _extract_images_from_pdf(self, pdf_path: str) -> List[Dict]: """Extract images from PDF using PyMuPDF""" images_data = [] try: doc = fitz.open(pdf_path) for page_num, page in enumerate(doc, start=1): image_list = page.get_images(full=True) for img_idx, img in enumerate(image_list): try: xref = img[0] base_image = doc.extract_image(xref) image_bytes = base_image["image"] # Convert to PIL Image import io pil_image = Image.open(io.BytesIO(image_bytes)).convert('RGB') # Filter small images width, height = pil_image.size if width >= 100 and height >= 100: images_data.append({ 'image': pil_image, 'page': page_num, 'index': img_idx }) except Exception as e: print(f"Error extracting image on page {page_num}: {e}") continue doc.close() except Exception as e: print(f"Image extraction error: {e}") return images_data def _generate_answer_local(self, prompt: str) -> str: """Generate answer using local FLAN-T5 model""" try: inputs = self.tokenizer( prompt, return_tensors="pt", max_length=512, truncation=True ) outputs = self.llm_model.generate( inputs.input_ids, max_new_tokens=200, temperature=0.3, do_sample=False, num_beams=4 ) answer = self.tokenizer.decode(outputs[0], skip_special_tokens=True) return answer.strip() except Exception as e: print(f"Local generation error: {e}") return "" def _generate_answer_api(self, prompt: str) -> str: """Generate answer using HuggingFace API""" try: response = self.llm_client.text_generation( prompt, model="google/flan-t5-xxl", max_new_tokens=250, temperature=0.3 ) return response.strip() except Exception as e: print(f"API generation error: {e}") return "" def _generate_extractive_answer(self, query: str, context: str) -> str: """Fallback extractive QA""" sentences = re.split(r'[.!?]+', context) sentences = [s.strip() for s in sentences if len(s.strip()) > 20] query_words = set(query.lower().split()) scored = [] for sent in sentences[:30]: sent_words = set(sent.lower().split()) overlap = len(query_words.intersection(sent_words)) if overlap > 0: scored.append((overlap, sent)) scored.sort(reverse=True, key=lambda x: x[0]) if scored: top_sentences = [s for _, s in scored[:3]] return ". ".join(top_sentences) + "." return "I couldn't find relevant information to answer your question." def _generate_answer(self, prompt: str, query: str = "", context: str = "") -> str: """Generate answer with fallback chain""" if self.llm_mode == "api": answer = self._generate_answer_api(prompt) if answer and len(answer) > 20: return answer if self.llm_mode == "local": answer = self._generate_answer_local(prompt) if answer and len(answer) > 20: return answer # Fallback to extractive return self._generate_extractive_answer(query, context) def _clean_answer(self, answer: str) -> str: """Clean unwanted artifacts from answer""" answer = re.sub(r'^(Answer|Context|Question):\s*', '', answer, flags=re.IGNORECASE) answer = re.sub(r'[#*_`]+', '', answer) answer = re.sub(r'\s+', ' ', answer).strip() return answer @requests(on="/upload") def upload(self, docs: DocumentArray, **kwargs): """Upload PDF, extract text + images, index them""" try: for doc in docs: file_path = doc.uri if file_path.startswith("file://"): file_path = file_path.replace("file://", "") if not os.path.exists(file_path) or not file_path.endswith(".pdf"): continue print(f"📄 Processing PDF: {file_path}") # Extract text text_chunks = self._extract_text_from_pdf(file_path) # Index text chunks text_count = 0 context_text = "\n".join(text_chunks[:2])[:1000] # Context for images for chunk in text_chunks: # Split into smaller chunks paragraphs = chunk.split("\n\n") for para in paragraphs: if para.strip() and len(para.strip()) > 50: emb = self._get_embedding(para.strip()) self.index.add(np.array([emb])) self.metadata.append({ "type": "text", "content": para.strip() }) text_count += 1 print(f"✅ Indexed {text_count} text chunks") # Extract and analyze images images_data = self._extract_images_from_pdf(file_path) image_count = 0 for img_data in images_data: # Analyze image description = self._analyze_image(img_data['image'], context_text) # Save image temporarily img_path = os.path.abspath( f"img_p{img_data['page']}_i{img_data['index']}.png" ) img_data['image'].save(img_path, "PNG") # Embed description embed_text = f"Image from page {img_data['page']}: {description}" emb = self._get_embedding(embed_text) self.index.add(np.array([emb])) self.metadata.append({ "type": "image", "content": f"file://{img_path}", "description": description, "page": img_data['page'] }) image_count += 1 print(f"✅ Analyzed and indexed {image_count} images") # Save index faiss.write_index(self.index, self.index_file) with open(self.metadata_file, "wb") as f: pickle.dump(self.metadata, f) summary = f"✅ Upload complete!\n" summary += f"📊 Total vectors: {self.index.ntotal}\n" summary += f"📝 Text chunks: {text_count}\n" summary += f"🖼️ Images: {image_count}\n" summary += f"💚 100% FREE processing!" return DocumentArray([Document(text=summary)]) except Exception as e: error_msg = f"❌ Upload failed:\n{traceback.format_exc()}" print(error_msg) return DocumentArray([Document(text=error_msg)]) @requests(on="/query") def query(self, docs: DocumentArray, **kwargs): """Query with multimodal RAG""" results = DocumentArray() if self.index.ntotal == 0: return DocumentArray([ Document(text="⚠️ No documents uploaded. Please upload a PDF first.") ]) for doc in docs: try: query_text = doc.text # Get query embedding and search query_emb = self._get_embedding(query_text) D, I = self.index.search(np.array([query_emb]), k=8) context_parts = [] matched_images = [] image_descriptions = [] for idx in I[0]: if idx < len(self.metadata): meta = self.metadata[idx] if meta["type"] == "text": context_parts.append(meta["content"]) elif meta["type"] == "image": matched_images.append(Document(uri=meta["content"])) image_descriptions.append( f"[Image Page {meta.get('page', '?')}]: {meta['description']}" ) # Build context context_text = "\n\n".join(context_parts[:4]) if image_descriptions: context_text += "\n\nRelevant Images:\n" + "\n".join(image_descriptions[:2]) # Truncate if too long if len(context_text) > 2000: context_text = context_text[:2000] + "..." # Create prompt prompt = ( f"Answer the question based on the context from a research paper.\n\n" f"Context:\n{context_text}\n\n" f"Question: {query_text}\n\n" f"Answer:" ) # Generate answer answer = self._generate_answer(prompt, query_text, context_text) answer = self._clean_answer(answer) # Create result answer_doc = Document(text=answer) if matched_images: answer_doc.chunks = DocumentArray(matched_images[:4]) results.append(answer_doc) except Exception as e: error_msg = f"❌ Query failed: {str(e)}\n{traceback.format_exc()}" print(error_msg) results.append(Document(text=error_msg)) return results @requests(on="/stats") def stats(self, docs: DocumentArray, **kwargs): """Get index statistics""" text_count = sum(1 for m in self.metadata if m["type"] == "text") image_count = sum(1 for m in self.metadata if m["type"] == "image") stats_text = ( f"📊 Index Statistics:\n" f"Total vectors: {self.index.ntotal}\n" f"Text chunks: {text_count}\n" f"Images: {image_count}\n" f"LLM mode: {self.llm_mode}\n" f"💚 100% FREE models!" ) return DocumentArray([Document(text=stats_text)]) @requests(on="/reset") def reset(self, docs: DocumentArray, **kwargs): """Reset index and metadata""" try: self.index = faiss.IndexFlatL2(self.dim) self.metadata = [] if os.path.exists(self.index_file): os.remove(self.index_file) if os.path.exists(self.metadata_file): os.remove(self.metadata_file) return DocumentArray([Document(text="✅ Index reset successfully!")]) except Exception as e: return DocumentArray([ Document(text=f"❌ Reset failed: {str(e)}") ])