Retrieval fixes behind default-off flags: vector type tagging, input types, Cohere rerank, score ordering

README.md

@@ -65,11 +65,26 @@ cp .env.example .env.local   # fill in Supabase URL and API base URL
 npm run dev
 ```
 
+### Retrieval feature flags
+
+All default off. With none set, indexing and serving behave exactly as they did before the flags existed.
+
+| Flag | Where | Effect when on |
+| --- | --- | --- |
+| `RETRIEVAL_RERANK_ENABLED` | backend + frontend | Ordering only. Pulls a wider Pinecone candidate pool, scores pre-computed and vector candidates in one Cohere rerank pass, and orders the served list by that score. The Cohere score is written to `rerank_score`; the stored 0-10 `relevance_score` is left alone. If rerank is unavailable the response degrades to the unranked ordering rather than dropping results. |
+| `CREATORS_API_EMIT_MATCHES` | frontend | Response shape only, no reordering. Adds the `matches` array (each entry carrying a nested `creator_videos`) to `/api/products/[id]/creators`. `/dashboard/reels?product_id=...` reads `matches`, so that view renders empty without it. The Python endpoint has always emitted `matches`; this brings the Next.js route into line. |
+| `RETRIEVAL_DOC_INPUT_TYPE_ENABLED` | backend | Indexing only. Embeds corpus content with `input_type=search_document`. Vectors written this way are not comparable with the `search_query` vectors already in the index, so it stays off until the index is fully re-embedded. |
+
+Tuning knobs: `COHERE_RERANK_MODEL`, `RETRIEVAL_CANDIDATE_POOL`, `RETRIEVAL_TOP_N`.
+
 ### Running Tests
 
 ```bash
 cd backend
-python -m pytest tests/ -v
+python -m pytest tests/ -v            # note: test_real_search.py and test_youtube.py call the live YouTube API
+
+cd ../frontend
+npm test                              # offline, node --test
 ```
 
 ### Docker

backend/.env.example

@@ -6,6 +6,13 @@ INDEX_NAME=
 GEMINI_KEY=
 USE_IMAGE_EMBEDDINGS=false                 # Set to true to use image embeddings (requires more Pinecone storage)
 
+# Retrieval feature flags - both default off, unset means unchanged behaviour
+RETRIEVAL_RERANK_ENABLED=false             # Serving, ordering only: order the served list by Cohere rerank score
+RETRIEVAL_DOC_INPUT_TYPE_ENABLED=false     # Indexing: embed corpus with input_type=search_document (requires a full re-embed)
+COHERE_RERANK_MODEL=rerank-v3.5            # Rerank model
+RETRIEVAL_CANDIDATE_POOL=30                # Candidates retrieved from Pinecone before reranking
+RETRIEVAL_TOP_N=10                         # Results kept after reranking
+
 # Database
 SUPABASE_URL=
 SUPABASE_SERVICE_ROLE_KEY=

backend/API.py

@@ -21,6 +21,9 @@
     APP_URL
 )
 from utils.shopify_api import ShopifyAPIClient
+from utils.feature_flags import rerank_serving_enabled
+from utils.rerank import candidate_pool, top_n
+from utils.creator_ranking import rank_creator_candidates
 
 # Redis caching and rate limiting
 from utils.redis_client import (
@@ -1226,28 +1229,63 @@ async def get_product_creators(product_id: str, request: Request):
             .single()\
             .execute()
 
+        rerank_enabled = rerank_serving_enabled()
+        product_data = product.data or {}
+        # Unchanged: the exact text vector search has always embedded.
+        search_text = f"{product_data.get('title', '')} {product_data.get('description', '')}"[:500]
+        # Built outside the vector-search branch, from the fields that are
+        # actually present. A product that was never indexed still has text to
+        # rerank against, and a missing product yields "" and short-circuits to
+        # the similarity fallback instead of calling rerank with a dead query.
+        rerank_query = " ".join(
+            part for part in (product_data.get("title"), product_data.get("description")) if part
+        )[:500]
+
         vector_matches = []
-        if product.data and product.data.get("pinecone_id"):
+        if product_data.get("pinecone_id"):
             try:
-                from utils.vectordb import query_text
+                from utils.vectordb import query_text, is_creator_video_match
 
-                search_text = f"{product.data['title']} {product.data.get('description', '')}"
-                vector_results = query_text(search_text[:500], top_k=20)  # Limit text length
+                vector_results = query_text(
+                    search_text,
+                    top_k=candidate_pool() if rerank_enabled else 20
+                )
 
                 # Convert Pinecone results to our format
                 for match in vector_results.matches:
-                    if match.metadata.get("type") == "creator_video":
+                    metadata = match.metadata or {}
+                    if rerank_enabled:
+                        if not is_creator_video_match(metadata):
+                            continue
+                        vector_matches.append({
+                            "video_id": metadata.get("video_id"),
+                            "score": match.score,
+                            "title": metadata.get("title", ""),
+                            "channel": metadata.get("channel", "")
+                        })
+                    elif metadata.get("type") == "creator_video":
                         vector_matches.append({
-                            "video_id": match.metadata.get("video_id"),
+                            "video_id": metadata.get("video_id"),
                             "score": match.score
                         })
             except Exception as vector_error:
                 print(f"Vector search error: {vector_error}")
 
+        match_rows = matches.data if matches.data else []
+
+        if rerank_enabled:
+            match_rows, vector_matches = rank_creator_candidates(
+                rerank_query,
+                match_rows,
+                vector_matches,
+                limit=limit,
+                keep=top_n(),
+            )
+
         return json({
-            "matches": matches.data if matches.data else [],
+            "matches": match_rows,
             "vector_matches": vector_matches,
-            "count": len(matches.data) if matches.data else 0
+            "count": len(match_rows)
         })
 
     except Exception as e:

backend/background_worker.py

@@ -12,7 +12,7 @@
 from dotenv import load_dotenv
 from utils.yt_search import fetch_top_shorts
 from utils.video import parse_video
-from utils.vectordb import text_to_embedding, upsert_embeddings
+from utils.vectordb import document_to_embedding, upsert_embeddings
 from utils.supabase import SupabaseClient
 
 # Load environment variables
@@ -412,7 +412,7 @@ async def process_creator_video(video: dict, product: dict, source_keyword: str,
 
         # 2. Create embedding
         embedding_text = f"{video['title']} {video['description']} {analysis_data.get('aesthetic', '')} {analysis_data.get('tone_vibe', '')}"
-        embedding_response = text_to_embedding(embedding_text)
+        embedding_response = document_to_embedding(embedding_text)
         embedding_vector = embedding_response.embeddings.float_[0]
 
         # 3. Store in Pinecone
@@ -429,6 +429,7 @@ async def process_creator_video(video: dict, product: dict, source_keyword: str,
             "id": video_pinecone_id,
             "values": embedding_vector,
             "metadata": {
+                "type": "creator_video",
                 "video_id": video_id,
                 "title": video["title"][:200],  # Truncate for metadata limits
                 "channel": video["channelTitle"][:100],

backend/jobs/tasks.py

@@ -163,7 +163,7 @@ async def generate_embeddings_job(
         Job result with status and embedding info
     """
     from utils.redis_client import DistributedLock, cohere_limiter
-    from utils.vectordb import text_to_embedding, upsert_embeddings
+    from utils.vectordb import document_to_embedding, upsert_embeddings
 
     print(f"[Job] Generating embedding for video: {video_id}")
 
@@ -177,7 +177,7 @@ async def generate_embeddings_job(
             return {"status": "skipped", "reason": "already_processing"}
 
         try:
-            embedding_response = text_to_embedding(text)
+            embedding_response = document_to_embedding(text)
             embedding_vector = embedding_response.embeddings.float_[0]
 
             upsert_embeddings([{

backend/requirements.txt

@@ -1,7 +1,10 @@
 blacksheep
 uvicorn
-cohere
-pinecone
+# Majors pinned: ClientV2.rerank resolves to V2Client.rerank through the
+# ClientV2(V2Client, Client) MRO, and the v1 signature is incompatible
+# (Sequence[RerankRequestDocumentsItem], max_chunks_per_doc).
+cohere>=5.18,<6
+pinecone>=8,<9
 python-dotenv
 supabase
 google-api-python-client

backend/scripts/eval_rerank.py

@@ -0,0 +1,147 @@
+#!/usr/bin/env python3
+"""
+Show what Cohere Rerank does to a product's creator candidates.
+
+Retrieves a candidate pool from Pinecone and prints the top N in similarity
+order next to the top N in rerank order, with scores, so the ordering change is
+visible before turning it on in production.
+
+Reads only. Run from the backend directory with your own .env:
+
+    python scripts/eval_rerank.py --list
+    python scripts/eval_rerank.py --product-id <uuid>
+    python scripts/eval_rerank.py --query "merino wool base layer"
+
+Neither retrieval flag needs to be set: the script calls the rerank stage
+directly so you can compare today's ordering against the flagged path.
+"""
+import argparse
+import asyncio
+import sys
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+from utils.creator_ranking import creator_document_text
+from utils.rerank import candidate_pool, rerank_documents, rerank_model, top_n
+from utils.supabase import SupabaseClient
+from utils.vectordb import is_creator_video_match, query_text
+
+COLUMN_WIDTH = 46
+
+
+def truncate(text: str, width: int) -> str:
+    text = " ".join((text or "").split())
+    return text if len(text) <= width else text[: width - 1] + "…"
+
+
+def label(video: dict) -> str:
+    title = video.get("title") or video.get("video_id") or "?"
+    channel = video.get("channel_title") or video.get("channel") or ""
+    return f"{title} — {channel}" if channel else title
+
+
+async def load_products(supabase: SupabaseClient, product_id: str = None) -> list:
+    query = supabase.client.table("company_products").select("id, title, description")
+    if product_id:
+        query = query.eq("id", product_id)
+    result = await query.limit(25).execute()
+    return result.data or []
+
+
+async def load_videos(supabase: SupabaseClient, video_ids: list) -> dict:
+    if not video_ids:
+        return {}
+    result = await supabase.client.table("creator_videos")\
+        .select("video_id, title, description, channel_title")\
+        .in_("video_id", video_ids)\
+        .execute()
+    return {row["video_id"]: row for row in (result.data or [])}
+
+
+async def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--product-id", help="Product to evaluate")
+    parser.add_argument("--query", help="Raw query text instead of a product")
+    parser.add_argument("--list", action="store_true", help="List products and exit")
+    parser.add_argument("--pool", type=int, default=candidate_pool(), help="Pinecone candidates to retrieve")
+    parser.add_argument("--top-n", type=int, default=top_n(), help="Results to display")
+    args = parser.parse_args()
+
+    supabase = SupabaseClient()
+    await supabase.initialize()
+
+    if args.list:
+        for product in await load_products(supabase):
+            print(f"{product['id']}  {product['title']}")
+        return 0
+
+    if args.query:
+        query = args.query
+    elif args.product_id:
+        products = await load_products(supabase, args.product_id)
+        if not products:
+            print(f"No product with id {args.product_id}")
+            return 1
+        product = products[0]
+        query = f"{product['title']} {product.get('description') or ''}"
+        print(f"Product: {product['title']}")
+    else:
+        parser.error("pass --product-id, --query, or --list")
+
+    query = query[:500]
+    print(f"Query: {truncate(query, 120)}")
+    print(f"Pool: {args.pool}   Top N: {args.top_n}   Rerank model: {rerank_model()}\n")
+
+    results = query_text(query, top_k=args.pool)
+    candidates = [m for m in results.matches if is_creator_video_match(m.metadata or {})]
+    if not candidates:
+        print("No creator video vectors in the candidate pool.")
+        print("Videos indexed before the type metadata key are matched on video_id;")
+        print("if this is empty the pool is all products, or the index is empty.")
+        return 1
+
+    videos = await load_videos(supabase, [m.metadata.get("video_id") for m in candidates])
+    enriched = []
+    for match in candidates:
+        video_id = match.metadata.get("video_id")
+        video = videos.get(video_id) or dict(match.metadata or {})
+        video.setdefault("video_id", video_id)
+        enriched.append((video, match.score))
+
+    ranked = rerank_documents(query, [creator_document_text(v) for v, _ in enriched], limit=args.top_n)
+    if ranked is None:
+        print("Rerank unavailable — the serving path would keep similarity order.")
+        return 1
+
+    print(f"{'similarity (today)'.upper():<{COLUMN_WIDTH + 8}}{'rerank (flagged path)'.upper()}")
+    print("-" * (COLUMN_WIDTH + 8) + "-" * (COLUMN_WIDTH + 8))
+
+    left = enriched[: args.top_n]
+    for rank in range(max(len(left), len(ranked))):
+        if rank < len(left):
+            video, score = left[rank]
+            before = f"{rank + 1:>2}. [{score:.3f}] {truncate(label(video), COLUMN_WIDTH - 12)}"
+        else:
+            before = ""
+        if rank < len(ranked):
+            index, score = ranked[rank]
+            video = enriched[index][0]
+            after = f"{rank + 1:>2}. [{score:.3f}] {truncate(label(video), COLUMN_WIDTH - 12)}"
+        else:
+            after = ""
+        print(f"{before:<{COLUMN_WIDTH + 8}}{after}")
+
+    moved = sum(
+        1
+        for position, (index, _) in enumerate(ranked)
+        if index != position
+    )
+    print(f"\n{moved} of {len(ranked)} positions changed.")
+    print("The served endpoint also merges pre-computed keyword matches into this pool.")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

backend/tests/conftest.py

@@ -1,9 +1,34 @@
-import pytest
 import os
+from unittest.mock import MagicMock, patch
+
+import cohere
+import pytest
 
 os.environ.setdefault("SUPABASE_URL", "https://test.supabase.co")
 os.environ.setdefault("SUPABASE_SERVICE_ROLE_KEY", "test-key")
 os.environ.setdefault("SHOPIFY_API_KEY", "test-shopify-key")
 os.environ.setdefault("SHOPIFY_API_SECRET", "test-shopify-secret")
 os.environ.setdefault("SHOPIFY_REDIRECT_URI", "https://api.test.com/shopify/callback")
 os.environ.setdefault("APP_URL", "https://test.maatchaa.vercel.app")
+
+# Dummy values so a stray import of utils.vectordb during a test run cannot pick
+# up real keys from .env (load_dotenv does not override what is already set).
+os.environ.setdefault("COHERE_KEY", "test-cohere-key")
+os.environ.setdefault("PINECONE_KEY", "test-pinecone-key")
+os.environ.setdefault("INDEX_NAME", "test-index")
+
+REAL_COHERE_CLIENT_V2 = cohere.ClientV2
+
+# utils.vectordb builds its clients at import time, and pc.Index(name) resolves
+# the index host through a live call to api.pinecone.io. Job tasks import the
+# module from inside their bodies, so that import can happen at any point in a
+# run. Stub both constructors for the whole session; started and never stopped
+# on purpose.
+patch("cohere.ClientV2", MagicMock()).start()
+patch("pinecone.Pinecone", MagicMock()).start()
+
+
+@pytest.fixture(scope="session")
+def real_cohere_client_v2():
+    """The unpatched cohere.ClientV2, for autospeccing against its real signature."""
+    return REAL_COHERE_CLIENT_V2