Update app.py

app.py

@@ -7,51 +7,23 @@ Deploy: HuggingFace Spaces (CPU Basic — free tier)
 Pipeline
   Index  : Image → YOLOv8 detection → CLIP embedding → FAISS vector store
   Query  : Text → CLIP text embedding → cosine k-NN → LLM answer generation
+
+FIXES vs previous version
+  1. Removed gradio_client monkey-patch — it targeted private functions that
+     were renamed in gradio 4.44.x, causing silent schema corruption on import.
+  2. requirements.txt now includes opencv-python-headless (YOLO crashes without it)
+     and explicit gradio==4.44.1 pin.
+  3. gr.Image outputs return PIL Images consistently (Gradio accepts both PIL and
+     numpy; PIL is safer across versions).
 """
 
 import json
 import os
 from datetime import datetime
-from typing import List, Dict, Tuple, Optional
 
 import faiss
 import numpy as np
 import torch
-
-# ── Monkey-patch gradio_client 4.44.x schema-introspection bug ────────────────
-# gradio_client/utils.py calls `if "const" in schema` where schema can be a
-# plain bool (valid JSON Schema: additionalProperties=True/False).
-# The `in` operator on a bool raises TypeError and crashes the whole Space.
-# We patch both affected functions before importing gradio.
-try:
-    import gradio_client.utils as _gcu
-
-    # Patch 1: get_type() — guard against non-dict schema argument
-    _orig_get_type = _gcu.get_type
-
-    def _safe_get_type(schema):
-        if not isinstance(schema, dict):
-            return "Any"
-        return _orig_get_type(schema)
-
-    _gcu.get_type = _safe_get_type
-
-    # Patch 2: _json_schema_to_python_type() — normalise bool additionalProperties
-    _orig_j2p = _gcu._json_schema_to_python_type
-
-    def _safe_j2p(schema, defs=None):
-        if not isinstance(schema, dict):
-            return "Any"
-        if isinstance(schema.get("additionalProperties"), bool):
-            schema = {**schema, "additionalProperties": {}}
-        return _orig_j2p(schema, defs)
-
-    _gcu._json_schema_to_python_type = _safe_j2p
-    print("✅ gradio_client monkey-patch applied.")
-except Exception as _e:
-    print(f"⚠️  gradio_client patch skipped: {_e}")
-# ──────────────────────────────────────────────────────────────────────────────
-
 import gradio as gr
 from PIL import Image
 from transformers import CLIPModel, CLIPProcessor
@@ -60,35 +32,32 @@ from huggingface_hub import InferenceClient
 
 # ── Model loading (runs once at Space start-up) ────────────────────────────────
 print("⏳ Loading CLIP ViT-B/32 ...")
-CLIP_MODEL_ID = "openai/clip-vit-base-patch32"
+CLIP_MODEL_ID  = "openai/clip-vit-base-patch32"
 clip_model     = CLIPModel.from_pretrained(CLIP_MODEL_ID)
 clip_processor = CLIPProcessor.from_pretrained(CLIP_MODEL_ID)
 clip_model.eval()
 
 print("⏳ Loading YOLOv8n ...")
-yolo = YOLO("yolov8n.pt")       # auto-downloads ~6 MB on first run
+yolo = YOLO("yolov8n.pt")          # auto-downloads ~6 MB on first run
 
 print("⏳ Initialising LLM inference client ...")
-# Uses HF Serverless Inference — free for open models, no GPU in Space needed.
-# Add HF_TOKEN as a Space secret for higher rate limits.
+# Free HF Serverless Inference — model runs on HF servers, not in the Space.
+# Add HF_TOKEN as a Space Secret (Settings → Secrets) for higher rate limits.
 HF_TOKEN = os.environ.get("HF_TOKEN", None)
-llm = InferenceClient(
-    model="HuggingFaceH4/zephyr-7b-beta",
-    token=HF_TOKEN,
-)
+llm = InferenceClient(model="HuggingFaceH4/zephyr-7b-beta", token=HF_TOKEN)
 
 print("✅ All models ready.")
 
 # ── FAISS vector store (in-memory, session-scoped) ─────────────────────────────
-EMBED_DIM   = 512                             # CLIP ViT-B/32 output dimension
-faiss_index = faiss.IndexFlatIP(EMBED_DIM)    # inner-product → cosine after L2 norm
-image_store: List[Dict] = []                  # parallel list to FAISS index rows
+EMBED_DIM   = 512                          # CLIP ViT-B/32 output dimension
+faiss_index = faiss.IndexFlatIP(EMBED_DIM) # cosine similarity via L2-normalised inner product
+image_store = []                           # parallel list: one dict per indexed image
 
 
 # ── Embedding helpers ──────────────────────────────────────────────────────────
 
 def embed_image(pil_img: Image.Image) -> np.ndarray:
-    """Return L2-normalised 512-d CLIP image embedding."""
+    """Return L2-normalised 512-d CLIP image embedding (shape: 1 × 512)."""
     inputs = clip_processor(images=pil_img, return_tensors="pt")
     with torch.no_grad():
         features = clip_model.get_image_features(**inputs)
@@ -98,10 +67,9 @@ def embed_image(pil_img: Image.Image) -> np.ndarray:
 
 
 def embed_text(text: str) -> np.ndarray:
-    """Return L2-normalised 512-d CLIP text embedding."""
-    inputs = clip_processor(
-        text=[text], return_tensors="pt", padding=True, truncation=True
-    )
+    """Return L2-normalised 512-d CLIP text embedding (shape: 1 × 512)."""
+    inputs = clip_processor(text=[text], return_tensors="pt",
+                            padding=True, truncation=True)
     with torch.no_grad():
         features = clip_model.get_text_features(**inputs)
     emb = features.numpy().astype("float32")
@@ -111,26 +79,23 @@ def embed_text(text: str) -> np.ndarray:
 
 # ── Computer-vision detection pipeline ────────────────────────────────────────
 
-def run_detection(pil_img: Image.Image) -> Tuple[Image.Image, List[Dict], str]:
+def run_detection(pil_img: Image.Image):
     """
-    Run YOLOv8 on a PIL image.
-    Returns: (annotated_pil, detections_list, summary_str)
+    Run YOLOv8n on a PIL image.
+    Returns: (annotated_PIL, detections_list, summary_str)
     """
-    results = yolo(np.array(pil_img))[0]
-    annotated = Image.fromarray(results.plot())
+    results   = yolo(np.array(pil_img))[0]
+    annotated = Image.fromarray(results.plot())   # PIL Image with bounding boxes
 
-    detections: List[Dict] = []
+    detections = []
     if results.boxes is not None:
         for box in results.boxes:
-            cls_id = int(box.cls[0])
-            conf   = float(box.conf[0])
             detections.append({
-                "label":      yolo.names[cls_id],
-                "confidence": round(conf, 3),
+                "label":      yolo.names[int(box.cls[0])],
+                "confidence": round(float(box.conf[0]), 3),
             })
 
-    # Object count summary, e.g. "2 person, 1 car"
-    counts: Dict[str, int] = {}
+    counts = {}
     for d in detections:
         counts[d["label"]] = counts.get(d["label"], 0) + 1
     summary = ", ".join(f"{v} {k}" for k, v in counts.items()) or "no objects detected"
@@ -141,10 +106,7 @@ def run_detection(pil_img: Image.Image) -> Tuple[Image.Image, List[Dict], str]:
 # ── Index pipeline ─────────────────────────────────────────────────────────────
 
 def index_image(image_np, note: str):
-    """
-    Detect objects, compute CLIP embedding, add to FAISS index.
-    Triggered by the 'Detect & Index' button.
-    """
+    """Detect → embed → store. Called by the 'Detect & Index' button."""
     if image_np is None:
         return None, "❌ Please upload an image first.", _store_badge()
 
@@ -152,21 +114,20 @@ def index_image(image_np, note: str):
     annotated, detections, summary = run_detection(pil_img)
     embedding                      = embed_image(pil_img)
 
-    # Persist to vector store
     faiss_index.add(embedding)
-    record = {
+    image_store.append({
         "id":         len(image_store),
         "image":      pil_img.copy(),
-        "annotated":  annotated,
+        "annotated":  annotated,         # PIL Image
         "detections": detections,
         "summary":    summary,
         "note":       note.strip() or "—",
         "ts":         datetime.now().strftime("%H:%M:%S"),
-    }
-    image_store.append(record)
+    })
 
-    msg = f"✅ Image #{record['id']} indexed · Found: {summary}"
-    return np.array(annotated), msg, _store_badge()
+    msg = f"✅ Image #{len(image_store) - 1} indexed · Found: {summary}"
+    # FIX 3: return PIL Image directly — avoids potential numpy dtype mismatches
+    return annotated, msg, _store_badge()
 
 
 def _store_badge() -> str:
@@ -177,24 +138,20 @@ def _store_badge() -> str:
 
 def query_images(text_query: str, top_k: int):
     """
-    1. Embed query text with CLIP.
-    2. Cosine k-NN search in FAISS.
-    3. Build RAG prompt from retrieved context.
-    4. Generate answer with Zephyr-7B via HF Inference API.
+    Text → CLIP embed → FAISS k-NN → RAG prompt → Zephyr-7B answer.
     """
     if not image_store:
-        return None, "❌ No images indexed yet — go to 'Detect & Index' first.", []
+        return None, "❌ No images indexed yet — upload images in the 'Detect & Index' tab first.", ""
 
     if not text_query.strip():
-        return None, "❌ Please enter a question.", []
+        return None, "❌ Please enter a question.", ""
 
     query_emb    = embed_text(text_query)
     k            = min(int(top_k), len(image_store))
     scores, idxs = faiss_index.search(query_emb, k)
 
-    # Build retrieval results & context string for the LLM
-    hits: List[Dict]    = []
-    ctx_lines: List[str] = []
+    hits      = []
+    ctx_lines = []
     for rank, (score, idx) in enumerate(zip(scores[0], idxs[0])):
         if idx < 0:
             continue
@@ -214,7 +171,6 @@ def query_images(text_query: str, top_k: int):
 
     context = "\n".join(ctx_lines)
 
-    # Zephyr-7B chat prompt (ChatML format)
     prompt = (
         "<|system|>\n"
         "You are a concise visual-AI assistant. "
@@ -235,15 +191,11 @@ def query_images(text_query: str, top_k: int):
             stop_sequences=["<|user|>", "<|system|>"],
         ).strip()
     except Exception as exc:
-        answer = (
-            f"⚠️ LLM call failed ({exc}).\n\n"
-            f"Raw retrieval context:\n{context}"
-        )
+        answer = f"⚠️ LLM call failed ({exc}).\n\nRaw retrieval context:\n{context}"
 
     best_idx   = int(idxs[0][0]) if len(idxs[0]) > 0 and idxs[0][0] >= 0 else None
-    best_image = (
-        np.array(image_store[best_idx]["annotated"]) if best_idx is not None else None
-    )
+    # FIX 3 continued: return PIL Image directly
+    best_image = image_store[best_idx]["annotated"] if best_idx is not None else None
 
     return best_image, answer, json.dumps(hits, indent=2)
 
@@ -271,7 +223,6 @@ with gr.Blocks(
     css=CUSTOM_CSS,
 ) as demo:
 
-    # ── Header ─────────────────────────────────────────────────────────────────
     gr.HTML("""
     <div class="hero">
       <h1>🔍 VisualRAG</h1>
@@ -281,18 +232,18 @@ with gr.Blocks(
         <span class="badge">CLIP ViT-B/32</span>
         <span class="badge">FAISS</span>
         <span class="badge">Zephyr-7B</span>
-        <span class="badge">Gradio</span>
+        <span class="badge">Gradio 4.44.1</span>
       </div>
     </div>
     """)
 
     with gr.Tabs():
 
-        # ── TAB 1 : Detect & Index ─────────────────────────────────────────────
+        # ── TAB 1: Detect & Index ──────────────────────────────────────────────
         with gr.Tab("📤 Detect & Index"):
             gr.Markdown(
-                "Upload any image. YOLOv8 will detect objects, then CLIP encodes it "
-                "into a 512-d embedding stored in FAISS for later retrieval."
+                "Upload any image. YOLOv8n detects objects, then CLIP ViT-B/32 encodes it "
+                "into a 512-d embedding stored in FAISS."
             )
             with gr.Row():
                 with gr.Column(scale=1):
@@ -305,17 +256,14 @@ with gr.Blocks(
                         label="Context note (optional)",
                         placeholder="e.g. 'Warehouse entrance camera, aisle 3'",
                     )
-                    index_btn = gr.Button(
-                        "🔍 Detect & Index", variant="primary", size="lg"
-                    )
+                    index_btn = gr.Button("🔍 Detect & Index", variant="primary", size="lg")
 
                 with gr.Column(scale=1):
+                    # type not specified on output gr.Image → Gradio auto-detects PIL or numpy
                     detection_output = gr.Image(label="Detection result (annotated)")
                     status_output    = gr.Textbox(label="Status", interactive=False)
                     store_output     = gr.Textbox(
-                        label="Vector store",
-                        interactive=False,
-                        value=_store_badge(),
+                        label="Vector store", interactive=False, value=_store_badge()
                     )
 
             index_btn.click(
@@ -324,18 +272,18 @@ with gr.Blocks(
                 outputs=[detection_output, status_output, store_output],
             )
 
-        # ── TAB 2 : Query (RAG) ────────────────────────────────────────────────
+        # ── TAB 2: Query (RAG) ────────────────────────────────────────────────
         with gr.Tab("💬 Query (RAG)"):
             gr.Markdown(
-                "Type any question about your indexed images. "
-                "CLIP embeds the query, FAISS retrieves the most similar images, "
+                "Ask any question about your indexed images. CLIP embeds the query, "
+                "FAISS retrieves the most similar images by cosine similarity, "
                 "and Zephyr-7B generates a grounded answer."
             )
             with gr.Row():
                 with gr.Column(scale=1):
                     query_input = gr.Textbox(
                         label="Your question",
-                        placeholder="e.g. 'How many people are visible?'  or  'Are there any vehicles?'",
+                        placeholder="e.g. 'How many people are visible?' or 'Are there any vehicles?'",
                         lines=3,
                     )
                     topk_slider = gr.Slider(
@@ -349,43 +297,39 @@ with gr.Blocks(
                 with gr.Column(scale=1):
                     match_img  = gr.Image(label="Best matching image")
                     llm_answer = gr.Textbox(
-                        label="AI Answer (RAG-grounded)",
-                        lines=6,
-                        interactive=False,
+                        label="AI Answer (RAG-grounded)", lines=6, interactive=False
                     )
 
-            hits_json = gr.Textbox(
-                label="Retrieval scores (all top-K)",
-                interactive=False,
-                lines=6,
+            hits_out = gr.Textbox(
+                label="Retrieval scores (all top-K)", interactive=False, lines=8
             )
 
             query_btn.click(
                 fn=query_images,
                 inputs=[query_input, topk_slider],
-                outputs=[match_img, llm_answer, hits_json],
+                outputs=[match_img, llm_answer, hits_out],
             )
 
-        # ── TAB 3 : How it works ───────────────────────────────────────────────
+        # ── TAB 3: How it works ───────────────────────────────────────────────
         with gr.Tab("🏗️ How it works"):
             gr.Markdown("""
-## System Overview
+## System overview
 
-### Index Pipeline
+### Index pipeline
 ```
-Image → YOLOv8 detection (objects + confidence)
-      → CLIP ViT-B/32 image encoder (512-d embedding)
+Image → YOLOv8n detection (objects + confidence scores)
+      → CLIP ViT-B/32 image encoder  →  512-d embedding
       → L2 normalisation
-      → FAISS IndexFlatIP (cosine similarity store)
+      → FAISS IndexFlatIP  (cosine similarity store)
 ```
 
-### Query / RAG Pipeline
+### Query / RAG pipeline
 ```
-Text query → CLIP text encoder (512-d embedding)
+Text query → CLIP text encoder  →  512-d query embedding
            → L2 normalisation
            → FAISS k-NN search (cosine similarity, top-K)
-           → Build RAG prompt: query + retrieved context
-           → Zephyr-7B-β via HF Serverless Inference API
+           → RAG prompt = query + retrieved context
+           → Zephyr-7B-β  (HF Serverless Inference API)
            → Natural language answer
 ```
 
@@ -396,22 +340,19 @@ Text query → CLIP text encoder (512-d embedding)
 | Object detection | YOLOv8n (Ultralytics) | Fast CPU inference, production-proven |
 | Visual embedding | CLIP ViT-B/32 | Joint image–text space enables text→image search |
 | Vector index | FAISS IndexFlatIP | Exact cosine k-NN, zero infra |
-| LLM | Zephyr-7B-β (HF API) | Free serverless, good instruction following |
-| UI | Gradio 4.x | Native HF Spaces support |
-
-## Skills Demonstrated
-- **Computer vision pipeline**: detect → annotate → embed → index
-- **RAG architecture**: embed → retrieve → generate
-- **Multi-modal reasoning**: text queries over an image knowledge base
-- **Vector database**: FAISS with cosine similarity (L2-normalised inner product)
-- **LLM integration**: HF Inference API with prompt engineering
-- **Production patterns**: stateless functions, error handling, model caching
+| LLM | Zephyr-7B-β (HF API) | Free serverless, strong instruction following |
+| UI | Gradio 4.44.1 | Native HF Spaces support |
+
+## What was fixed in this version
+- `opencv-python-headless` added to requirements — YOLO needs `cv2` at startup
+- `gradio==4.44.1` pinned explicitly in requirements
+- Removed fragile `gradio_client` monkey-patch — caused silent schema corruption in 4.44.x
+- Return types unified to PIL Image (avoids numpy dtype edge cases across Gradio versions)
             """)
 
-    # ── Footer ─────────────────────────────────────────────────────────────────
     gr.HTML("""
-    <div style="text-align:center; padding:16px 0 4px; color:var(--color-subdued); font-size:12px">
-      VisualRAG · YOLOv8 + CLIP + FAISS + LLM · Built for HuggingFace Spaces
+    <div style="text-align:center; padding:14px 0 4px; color:var(--color-subdued); font-size:12px">
+      VisualRAG · YOLOv8 + CLIP + FAISS + LLM · HuggingFace Spaces
     </div>
     """)