feat: optional multimodal ingest + visual retrieval (CLIP cross-modal + text-page coupling)

app/config.py

@@ -90,6 +90,45 @@ def get_env_variable(
 env_value = get_env_variable("PDF_EXTRACT_IMAGES", "False").lower()
 PDF_EXTRACT_IMAGES = True if env_value == "true" else False
 
+# Optional pre-extraction webhook: when text extraction returns pages averaging
+# fewer than PRE_EXTRACTION_WEBHOOK_MIN_CHARS characters, the file is POSTed to
+# PRE_EXTRACTION_WEBHOOK_URL. The webhook is expected to respond with
+# `{"text": "...", "provider": "..."}` and that text replaces the original
+# extraction. Disabled when the URL is empty (the default).
+PRE_EXTRACTION_WEBHOOK_URL = get_env_variable("PRE_EXTRACTION_WEBHOOK_URL", "")
+PRE_EXTRACTION_WEBHOOK_MIN_CHARS = int(
+    get_env_variable("PRE_EXTRACTION_WEBHOOK_MIN_CHARS", "100")
+)
+PRE_EXTRACTION_WEBHOOK_TIMEOUT = int(
+    get_env_variable("PRE_EXTRACTION_WEBHOOK_TIMEOUT", "60")
+)
+
+# Optional multimodal-RAG visual pipeline. When VISUAL_EMBED_URL is set, each
+# ingested PDF is rendered to page PNGs via pdftoppm and each page is sent to
+# the CLIP embed sidecar; the resulting vectors are stored in the
+# `visual_chunks` pgvector table for retrieval. All steps soft-fail: if
+# pdftoppm, the sidecar, or the DB is unreachable, the text ingest path still
+# succeeds.
+VISUAL_EMBED_URL = get_env_variable("VISUAL_EMBED_URL", "")
+# Text-query endpoint of the same CLIP sidecar. Empty → derive from
+# VISUAL_EMBED_URL by swapping /embed/image for /embed/text.
+VISUAL_TEXT_EMBED_URL = get_env_variable("VISUAL_TEXT_EMBED_URL", "")
+VISUAL_PAGE_DPI = int(get_env_variable("VISUAL_PAGE_DPI", "100"))
+VISUAL_STORAGE_ROOT = get_env_variable("VISUAL_STORAGE_ROOT", "/var/rag-visual")
+VISUAL_EMBED_TIMEOUT = int(get_env_variable("VISUAL_EMBED_TIMEOUT", "30"))
+VISUAL_SCORE_THRESHOLD = float(get_env_variable("VISUAL_SCORE_THRESHOLD", "0.25"))
+VISUAL_QUERY_TOP_K = int(get_env_variable("VISUAL_QUERY_TOP_K", "3"))
+# Text-to-visual-page-coupling (Phase 4). When True, /query uses the pages of
+# the text-search hits as the primary source of visual_matches, with CLIP
+# cross-modal retrieval as a secondary signal. Rationale: CLIP-style models
+# weight text-inside-images, so a page with a matching headline can outrank
+# the page with the actually-relevant photo. The text pipeline already knows
+# which pages are relevant — couple the visuals to that decision.
+VISUAL_TEXT_COUPLED = get_env_variable("VISUAL_TEXT_COUPLED", "true").lower() == "true"
+VISUAL_TEXT_COUPLED_MAX_PAGES = int(
+    get_env_variable("VISUAL_TEXT_COUPLED_MAX_PAGES", "4")
+)
+
 if POSTGRES_USE_UNIX_SOCKET:
     connection_suffix = f"{urllib.parse.quote_plus(POSTGRES_USER)}:{urllib.parse.quote_plus(POSTGRES_PASSWORD)}@/{urllib.parse.quote_plus(POSTGRES_DB)}?host={urllib.parse.quote_plus(DB_HOST)}"
 else:

app/models.py

@@ -31,6 +31,11 @@ class QueryRequestBody(BaseModel):
     file_id: str
     k: int = 4
     entity_id: Optional[str] = None
+    # When True, the /query response is wrapped as
+    # {"chunks": [...], "visual_matches": [...]} and the visual pipeline
+    # (multimodal-RAG) is consulted in addition to the text index.
+    # Default False keeps the legacy flat-list shape.
+    include_visual: bool = False
 
 
 class CleanupMethod(str, Enum):
@@ -42,3 +47,11 @@ class QueryMultipleBody(BaseModel):
     query: str
     file_ids: List[str]
     k: int = 4
+    include_visual: bool = False
+
+
+class VisualMatch(BaseModel):
+    file_id: str
+    page_number: int
+    image_path: str
+    score: float

app/routes/document_routes.py

@@ -52,6 +52,19 @@
     clean_text,
     process_documents,
     cleanup_temp_encoding_file,
+    maybe_enrich_with_webhook,
+)
+from app.utils.visual_embed import (
+    embed_text_query,
+    fetch_visual_chunks_for_pages,
+    maybe_embed_visuals,
+    similarity_search_visual,
+)
+from app.config import (
+    VISUAL_EMBED_URL,
+    VISUAL_QUERY_TOP_K,
+    VISUAL_TEXT_COUPLED,
+    VISUAL_TEXT_COUPLED_MAX_PAGES,
 )
 from app.utils.health import is_health_ok
 
@@ -147,6 +160,11 @@ async def load_file_content(
         loader, known_type, file_ext = get_loader(filename, content_type, file_path)
         loop = asyncio.get_running_loop()
         data = await loop.run_in_executor(executor, lambda: list(loader.lazy_load()))
+        # Optional pre-extraction webhook (e.g. OCR sidecar) — no-op when
+        # PRE_EXTRACTION_WEBHOOK_URL is unset.
+        data = await loop.run_in_executor(
+            executor, maybe_enrich_with_webhook, file_path, data
+        )
         return data, known_type, file_ext
     finally:
         # Clean up temporary UTF-8 file if it was created for encoding conversion
@@ -311,6 +329,126 @@ def get_cached_query_embedding(query: str):
     return vector_store.embedding_function.embed_query(query)
 
 
+def _pages_by_file_from_text_docs(text_documents) -> dict:
+    """Extract {file_id: [page_number, ...]} from text similarity_search results.
+
+    Preserves rank order (earliest text hit → first page in the list) and
+    dedupes within each file. Reads ``page_number`` first, falling back to
+    ``page`` — PyPDFLoader writes ``page`` while custom loaders and older
+    chunks may use ``page_number``. Chunks with neither are silently
+    skipped (e.g. text-only formats like .txt that have no page concept).
+    """
+    pages_by_file: dict = {}
+    if not text_documents:
+        return pages_by_file
+    for entry in text_documents:
+        doc = entry[0] if isinstance(entry, tuple) else entry
+        metadata = getattr(doc, "metadata", None) or {}
+        file_id = metadata.get("file_id")
+        page_number = metadata.get("page_number")
+        if page_number is None:
+            page_number = metadata.get("page")
+        if not file_id or page_number is None:
+            continue
+        try:
+            page_number = int(page_number)
+        except (TypeError, ValueError):
+            continue
+        bucket = pages_by_file.setdefault(file_id, [])
+        if page_number not in bucket:
+            bucket.append(page_number)
+    return pages_by_file
+
+
+async def _fetch_visual_matches_for_file_ids(
+    request: Request,
+    query: str,
+    file_ids: List[str],
+    text_documents: Optional[list] = None,
+) -> list:
+    """Cross-modal retrieval for /query's visual_matches field.
+
+    Two signals are merged when ``VISUAL_TEXT_COUPLED`` is enabled:
+
+      1. **Primary** — text-page-coupling: use the pages of the text-search
+         hits (per file_id) to look up visual_chunks directly. Fixes the
+         CLIP-weights-text-inside-images bias: a page whose headline matches
+         the query outranks the page with the actually-relevant photo.
+      2. **Secondary** — CLIP cross-modal: text query embedded via the CLIP
+         text encoder, nearest neighbours in ``visual_chunks``. Keeps
+         visually-relevant pages that the text pipeline missed.
+
+    Dedup is by (file_id, page_number); primary wins. Soft-fails on any
+    error (returns []) so a broken sidecar never breaks the main query path.
+    """
+    if not VISUAL_EMBED_URL or not file_ids:
+        return []
+
+    results: list = []
+    seen: set = set()
+
+    # Primary signal: text-page-coupling.
+    if VISUAL_TEXT_COUPLED:
+        try:
+            from app.services.database import PSQLDatabase
+
+            pool = await PSQLDatabase.get_pool()
+            pages_by_file = _pages_by_file_from_text_docs(text_documents)
+            for file_id, pages in pages_by_file.items():
+                if file_id not in file_ids:
+                    continue
+                capped = pages[:VISUAL_TEXT_COUPLED_MAX_PAGES]
+                rows = await fetch_visual_chunks_for_pages(
+                    pool=pool, file_id=file_id, page_numbers=capped
+                )
+                # Preserve text-rank order by reindexing from ``capped``.
+                by_page = {r["page_number"]: r for r in rows}
+                for page in capped:
+                    row = by_page.get(page)
+                    if not row:
+                        continue
+                    key = (row["file_id"], row["page_number"])
+                    if key in seen:
+                        continue
+                    seen.add(key)
+                    # Synthetic score above VISUAL_SCORE_THRESHOLD so downstream
+                    # filters treat text-coupled hits as top-tier.
+                    results.append({**row, "score": 1.0, "source": "text_coupled"})
+        except Exception as exc:
+            logger.warning("visual retrieval: text-page-coupling failed: %s", exc)
+
+    # Secondary signal: CLIP cross-modal.
+    try:
+        loop = asyncio.get_running_loop()
+        query_embedding = await loop.run_in_executor(
+            request.app.state.thread_pool, embed_text_query, query
+        )
+    except Exception as exc:
+        logger.warning("visual retrieval: text embed failed: %s", exc)
+        return results
+    try:
+        from app.services.database import PSQLDatabase
+
+        pool = await PSQLDatabase.get_pool()
+        clip_hits = await similarity_search_visual(
+            pool=pool,
+            query_embedding=query_embedding,
+            file_ids=file_ids,
+            k=VISUAL_QUERY_TOP_K,
+        )
+    except Exception as exc:
+        logger.warning("visual retrieval: similarity search failed: %s", exc)
+        return results
+
+    for hit in clip_hits:
+        key = (hit["file_id"], hit["page_number"])
+        if key in seen:
+            continue
+        seen.add(key)
+        results.append({**hit, "source": "clip"})
+    return results
+
+
 @router.post("/query")
 async def query_embeddings_by_file_id(
     body: QueryRequestBody,
@@ -341,6 +479,11 @@ async def query_embeddings_by_file_id(
             )
 
         if not documents:
+            if body.include_visual:
+                visual = await _fetch_visual_matches_for_file_ids(
+                    request, body.query, [body.file_id], text_documents=documents
+                )
+                return {"chunks": authorized_documents, "visual_matches": visual}
             return authorized_documents
 
         document, score = documents[0]
@@ -369,6 +512,14 @@ async def query_embeddings_by_file_id(
                     f"Unauthorized access attempt by user {user_authorized} to a document with user_id {doc_user_id}"
                 )
 
+        if body.include_visual:
+            visual = await _fetch_visual_matches_for_file_ids(
+                request,
+                body.query,
+                [body.file_id],
+                text_documents=authorized_documents,
+            )
+            return {"chunks": authorized_documents, "visual_matches": visual}
         return authorized_documents
 
     except HTTPException as http_exc:
@@ -756,6 +907,12 @@ async def embed_local_file(
         data = await loop.run_in_executor(
             request.app.state.thread_pool, lambda: list(loader.lazy_load())
         )
+        data = await loop.run_in_executor(
+            request.app.state.thread_pool,
+            maybe_enrich_with_webhook,
+            file_path,
+            data,
+        )
 
         result = await store_data_in_vector_db(
             data,
@@ -765,6 +922,15 @@ async def embed_local_file(
             executor=request.app.state.thread_pool,
         )
 
+        # Visual pipeline (multimodal-RAG). Soft-fails by design; never raises.
+        await maybe_embed_visuals(
+            file_path=file_path,
+            file_id=document.file_id,
+            file_ext=file_ext,
+            user_id=user_id,
+            executor=request.app.state.thread_pool,
+        )
+
         if result:
             return {
                 "status": True,
@@ -841,6 +1007,15 @@ async def embed_file(
             executor=request.app.state.thread_pool,
         )
 
+        # Visual pipeline (multimodal-RAG). Soft-fails by design; never raises.
+        await maybe_embed_visuals(
+            file_path=validated_file_path,
+            file_id=file_id,
+            file_ext=file_ext,
+            user_id=user_id,
+            executor=request.app.state.thread_pool,
+        )
+
         if not result:
             response_status = False
             response_message = "Failed to process/store the file data."
@@ -977,6 +1152,15 @@ async def embed_file_upload(
             executor=request.app.state.thread_pool,
         )
 
+        # Visual pipeline (multimodal-RAG). Soft-fails by design; never raises.
+        await maybe_embed_visuals(
+            file_path=validated_temp_file_path,
+            file_id=file_id,
+            file_ext=file_ext,
+            user_id=user_id,
+            executor=request.app.state.thread_pool,
+        )
+
         if not result:
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
@@ -1031,6 +1215,14 @@ async def query_embeddings_by_file_ids(request: Request, body: QueryMultipleBody
                 embedding, k=body.k, filter={"file_id": {"$in": body.file_ids}}
             )
 
+        if body.include_visual:
+            visual = await _fetch_visual_matches_for_file_ids(
+                request, body.query, body.file_ids, text_documents=documents
+            )
+            # When include_visual is True, we must not 404 just because the
+            # text path came back empty — visual matches might still be useful.
+            return {"chunks": documents, "visual_matches": visual}
+
         # Ensure documents list is not empty
         if not documents:
             raise HTTPException(