diff --git a/README.md b/README.md index eab9ab2..eeec432 100644 --- a/README.md +++ b/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 diff --git a/backend/.env.example b/backend/.env.example index ba7680d..43a40ff 100644 --- a/backend/.env.example +++ b/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= diff --git a/backend/API.py b/backend/API.py index 4eb85c9..d7adff8 100644 --- a/backend/API.py +++ b/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: diff --git a/backend/background_worker.py b/backend/background_worker.py index 7a9ba44..955a455 100644 --- a/backend/background_worker.py +++ b/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], diff --git a/backend/jobs/tasks.py b/backend/jobs/tasks.py index 19c4841..938bf30 100644 --- a/backend/jobs/tasks.py +++ b/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([{ diff --git a/backend/requirements.txt b/backend/requirements.txt index 190757c..08955e1 100644 --- a/backend/requirements.txt +++ b/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 diff --git a/backend/scripts/eval_rerank.py b/backend/scripts/eval_rerank.py new file mode 100644 index 0000000..f0c7035 --- /dev/null +++ b/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 + 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())) diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 5e27fb0..aadbfe8 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -1,5 +1,8 @@ -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") @@ -7,3 +10,25 @@ 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 diff --git a/backend/tests/test_retrieval.py b/backend/tests/test_retrieval.py new file mode 100644 index 0000000..a2463f3 --- /dev/null +++ b/backend/tests/test_retrieval.py @@ -0,0 +1,550 @@ +"""Tests for the retrieval flags, candidate identification, and rerank stage. + +Fully offline: the Cohere and Pinecone clients are mocked (conftest stubs both +constructors for the whole session). +""" +import importlib +from types import SimpleNamespace + +import cohere +import pytest +from unittest.mock import create_autospec, patch + +import utils.rerank as rerank_module +from utils.creator_ranking import ( + creator_document_text, + rank_creator_candidates, + stored_match_score, +) +from utils.feature_flags import document_input_type_enabled, rerank_serving_enabled +from utils.rerank import ( + candidate_pool, + rerank_documents, + rerank_model, + reset_client_cache, + top_n, +) + + +@pytest.fixture(autouse=True) +def reset_rerank_client_cache(): + reset_client_cache() + yield + reset_client_cache() + + +@pytest.fixture +def vectordb(monkeypatch): + monkeypatch.setenv("COHERE_KEY", "test-cohere-key") + monkeypatch.setenv("PINECONE_KEY", "test-pinecone-key") + monkeypatch.setenv("INDEX_NAME", "test-index") + # patch() only bites on the first import, and utils.vectordb may already be + # in sys.modules from another test file. Reload so the module-level clients + # are rebuilt against the mocks either way. + with patch("cohere.ClientV2"), patch("pinecone.Pinecone"): + import utils.vectordb as module + + importlib.reload(module) + yield module + + +@pytest.fixture +def make_client(real_cohere_client_v2): + """Build a client autospecced from the real cohere.ClientV2. + + Autospec enforces the live signature, so renaming a keyword (documents -> + docs, query -> q) or passing positionally raises TypeError instead of + silently passing the way a bare MagicMock would. + """ + + def build(results=(), error=None): + client = create_autospec(real_cohere_client_v2, instance=True) + if error is not None: + client.rerank.side_effect = error + else: + client.rerank.return_value = SimpleNamespace(results=list(results)) + return client + + return build + + +def result(index, score): + return SimpleNamespace(index=index, relevance_score=score) + + +def match_row(video_id, **fields): + row = { + "video_id": video_id, + "creator_videos": {"video_id": video_id, "title": f"{video_id} review"}, + } + row.update(fields) + return row + + +def vector_row(video_id, score): + return {"video_id": video_id, "score": score, "title": f"{video_id} short", "channel": "c"} + + +class TestFlags: + def test_default_off(self, monkeypatch): + monkeypatch.delenv("RETRIEVAL_RERANK_ENABLED", raising=False) + monkeypatch.delenv("RETRIEVAL_DOC_INPUT_TYPE_ENABLED", raising=False) + assert rerank_serving_enabled() is False + assert document_input_type_enabled() is False + + @pytest.mark.parametrize("value", ["true", "TRUE", "1", "yes", "on"]) + def test_truthy_values(self, monkeypatch, value): + monkeypatch.setenv("RETRIEVAL_RERANK_ENABLED", value) + assert rerank_serving_enabled() is True + + @pytest.mark.parametrize("value", ["false", "0", "", "off", "maybe"]) + def test_falsy_values(self, monkeypatch, value): + monkeypatch.setenv("RETRIEVAL_RERANK_ENABLED", value) + assert rerank_serving_enabled() is False + + def test_flags_are_independent(self, monkeypatch): + monkeypatch.setenv("RETRIEVAL_RERANK_ENABLED", "true") + monkeypatch.delenv("RETRIEVAL_DOC_INPUT_TYPE_ENABLED", raising=False) + assert rerank_serving_enabled() is True + assert document_input_type_enabled() is False + + +class TestCorpusInputType: + def test_defaults_to_search_query(self, vectordb, monkeypatch): + monkeypatch.delenv("RETRIEVAL_DOC_INPUT_TYPE_ENABLED", raising=False) + assert vectordb.corpus_input_type() == "search_query" + + def test_search_document_when_enabled(self, vectordb, monkeypatch): + monkeypatch.setenv("RETRIEVAL_DOC_INPUT_TYPE_ENABLED", "true") + assert vectordb.corpus_input_type() == "search_document" + + def test_query_embedding_is_always_search_query(self, vectordb, monkeypatch): + monkeypatch.setenv("RETRIEVAL_DOC_INPUT_TYPE_ENABLED", "true") + vectordb.co.embed.reset_mock() + vectordb.text_to_embedding("a query") + assert vectordb.co.embed.call_args.kwargs["input_type"] == "search_query" + + def test_document_embedding_follows_flag(self, vectordb, monkeypatch): + monkeypatch.setenv("RETRIEVAL_DOC_INPUT_TYPE_ENABLED", "true") + vectordb.co.embed.reset_mock() + vectordb.document_to_embedding("corpus content") + assert vectordb.co.embed.call_args.kwargs["input_type"] == "search_document" + + +class TestIsCreatorVideoMatch: + def test_typed_video(self, vectordb): + assert vectordb.is_creator_video_match({"type": "creator_video", "video_id": "abc"}) + + def test_typed_product(self, vectordb): + assert not vectordb.is_creator_video_match({"type": "product", "title": "Snowboard"}) + + def test_legacy_video_without_type(self, vectordb): + assert vectordb.is_creator_video_match({"video_id": "abc", "title": "Review"}) + + def test_legacy_product_without_type(self, vectordb): + assert not vectordb.is_creator_video_match({"title": "Snowboard", "price": 10}) + + def test_empty(self, vectordb): + assert not vectordb.is_creator_video_match({}) + assert not vectordb.is_creator_video_match(None) + + +class TestRerankCallShape: + """The call shape is the contract with cohere; pin it explicitly.""" + + def test_exact_keywords_passed_to_cohere(self, make_client): + client = make_client([result(0, 0.5)]) + rerank_documents("a query", ["a", "b"], limit=2, client=client) + + assert client.rerank.call_args.args == () + kwargs = client.rerank.call_args.kwargs + assert set(kwargs) == {"model", "query", "documents", "top_n"} + assert kwargs["query"] == "a query" + assert kwargs["documents"] == ["a", "b"] + assert kwargs["top_n"] == 2 + + def test_documents_are_passed_as_a_list_of_strings(self, make_client): + client = make_client([result(0, 0.5)]) + rerank_documents("query", ("a", "b"), client=client) + documents = client.rerank.call_args.kwargs["documents"] + assert isinstance(documents, list) + assert all(isinstance(document, str) for document in documents) + + def test_rerank_resolves_to_the_v2_method(self, real_cohere_client_v2): + # ClientV2(V2Client, Client): the v1 rerank takes + # Sequence[RerankRequestDocumentsItem] and max_chunks_per_doc, so an MRO + # change would silently break the call above. requirements.txt pins the + # major to keep this true. + assert real_cohere_client_v2.rerank.__qualname__.startswith("V2Client") + + +class TestRerankDocuments: + def test_returns_ranked_indices_and_scores(self, make_client): + client = make_client([result(2, 0.9), result(0, 0.4)]) + ranked = rerank_documents("query", ["a", "b", "c"], limit=2, client=client) + assert ranked == [(2, 0.9), (0, 0.4)] + + def test_caps_top_n_at_document_count(self, make_client): + client = make_client([result(0, 0.5)]) + rerank_documents("query", ["a"], limit=10, client=client) + assert client.rerank.call_args.kwargs["top_n"] == 1 + + def test_uses_configured_model(self, monkeypatch, make_client): + monkeypatch.setenv("COHERE_RERANK_MODEL", "rerank-english-v3.0") + client = make_client([result(0, 0.5)]) + rerank_documents("query", ["a"], client=client) + assert client.rerank.call_args.kwargs["model"] == "rerank-english-v3.0" + + def test_no_documents(self, make_client): + client = make_client([result(0, 0.5)]) + assert rerank_documents("query", [], client=client) is None + client.rerank.assert_not_called() + + def test_no_query(self, make_client): + client = make_client([result(0, 0.5)]) + assert rerank_documents("", ["a"], client=client) is None + client.rerank.assert_not_called() + + def test_api_error_falls_back(self, make_client): + client = make_client(error=RuntimeError("rerank is down")) + assert rerank_documents("query", ["a", "b"], client=client) is None + + def test_empty_results_fall_back(self, make_client): + assert rerank_documents("query", ["a"], client=make_client([])) is None + + def test_out_of_range_index_is_dropped(self, make_client): + client = make_client([result(99, 0.9), result(1, 0.2)]) + assert rerank_documents("query", ["a", "b"], client=client) == [(1, 0.2)] + + def test_missing_key_falls_back(self, monkeypatch): + monkeypatch.delenv("COHERE_KEY", raising=False) + assert rerank_documents("query", ["a", "b"]) is None + + def test_client_construction_failure_falls_back(self, monkeypatch): + def explode(): + raise ImportError("no cohere here") + + monkeypatch.setattr(rerank_module, "_client", explode) + assert rerank_documents("query", ["a", "b"]) is None + + +class TestClientCache: + @pytest.fixture + def built_with(self, monkeypatch): + """Keys every ClientV2 construction was made with. One entry per client.""" + keys = [] + + def factory(key): + keys.append(key) + return SimpleNamespace(key=key) + + monkeypatch.setattr(cohere, "ClientV2", factory) + return keys + + def test_client_is_reused_across_calls(self, monkeypatch, built_with): + monkeypatch.setenv("COHERE_KEY", "key-1") + assert rerank_module._client() is rerank_module._client() + assert built_with == ["key-1"] + + def test_rotated_key_builds_a_new_client(self, monkeypatch, built_with): + monkeypatch.setenv("COHERE_KEY", "key-1") + first = rerank_module._client() + monkeypatch.setenv("COHERE_KEY", "key-2") + assert rerank_module._client() is not first + assert built_with == ["key-1", "key-2"] + + def test_missing_key_returns_none_and_does_not_cache(self, monkeypatch, built_with): + monkeypatch.delenv("COHERE_KEY", raising=False) + assert rerank_module._client() is None + assert built_with == [] + monkeypatch.setenv("COHERE_KEY", "key-1") + assert rerank_module._client() is not None + + def test_cache_does_not_defeat_client_injection(self, monkeypatch, make_client): + monkeypatch.setenv("COHERE_KEY", "key-1") + client = make_client([result(0, 0.5)]) + rerank_documents("query", ["a"], client=client) + client.rerank.assert_called_once() + assert rerank_module._cached_client is None + + +class TestRerankSettings: + def test_defaults(self, monkeypatch): + monkeypatch.delenv("COHERE_RERANK_MODEL", raising=False) + monkeypatch.delenv("RETRIEVAL_CANDIDATE_POOL", raising=False) + monkeypatch.delenv("RETRIEVAL_TOP_N", raising=False) + assert rerank_model() == "rerank-v3.5" + assert candidate_pool() == 30 + assert top_n() == 10 + + def test_env_overrides(self, monkeypatch): + monkeypatch.setenv("RETRIEVAL_CANDIDATE_POOL", "50") + monkeypatch.setenv("RETRIEVAL_TOP_N", "5") + assert candidate_pool() == 50 + assert top_n() == 5 + + def test_invalid_values_are_handled(self, monkeypatch): + monkeypatch.setenv("RETRIEVAL_CANDIDATE_POOL", "not-a-number") + monkeypatch.setenv("RETRIEVAL_TOP_N", "0") + assert candidate_pool() == 30 + assert top_n() == 1 + + +class TestCreatorDocumentText: + def test_uses_title_channel_and_truncated_description(self): + text = creator_document_text( + {"title": "Snowboard review", "channel_title": "Ride", "description": "z" * 600} + ) + assert text.startswith("Snowboard review Ride ") + assert text.count("z") == 500 + + def test_falls_back_to_channel_key(self): + assert creator_document_text({"title": "T", "channel": "C"}) == "T C" + + def test_empty_video(self): + assert creator_document_text({}) == "" + + +class TestStoredMatchScore: + def test_prefers_relevance_score(self): + assert stored_match_score({"relevance_score": 8.5, "similarity_score": 0.2}) == 8.5 + + def test_falls_back_to_similarity_score(self): + assert stored_match_score({"similarity_score": 0.2}) == 0.2 + + def test_ignores_the_legacy_match_score_column(self): + # match_score belongs to the legacy product_matches table and is never + # present on product_creator_matches rows. + assert stored_match_score({"match_score": 99}) == 0.0 + + def test_non_numeric_is_zero(self): + assert stored_match_score({"relevance_score": "high"}) == 0.0 + + +def never_called(*args, **kwargs): + raise AssertionError("rerank should not have been called") + + +def unavailable(*args, **kwargs): + return None + + +class TestRankCreatorCandidatesFallback: + """With rerank unavailable the response must match the pre-flag one.""" + + def test_keeps_every_row_and_the_vector_list(self): + rows = [match_row(f"m{i}", relevance_score=i) for i in range(12)] + vectors = [vector_row(f"v{i}", 0.5) for i in range(6)] + + matches, vector_matches = rank_creator_candidates( + "query", rows, vectors, limit=50, keep=10, rerank=unavailable + ) + + assert len(matches) == 12 + assert len(vector_matches) == 6 + + def test_respects_the_callers_limit_not_top_n(self): + rows = [match_row(f"m{i}", relevance_score=i) for i in range(12)] + vectors = [vector_row("v1", 0.5)] + + matches, vector_matches = rank_creator_candidates( + "query", rows, vectors, limit=50, keep=10, rerank=unavailable + ) + + assert len(matches) == 12 # not truncated to keep=10 + assert vector_matches + + def test_limit_still_bounds_the_match_list(self): + rows = [match_row(f"m{i}", relevance_score=i) for i in range(12)] + matches, _ = rank_creator_candidates( + "query", rows, [], limit=5, keep=10, rerank=unavailable + ) + assert len(matches) == 5 + + def test_orders_on_relevance_score(self): + rows = [ + match_row("low", relevance_score=2.0), + match_row("high", relevance_score=9.0), + match_row("mid", relevance_score=5.0), + ] + matches, _ = rank_creator_candidates( + "query", rows, [], limit=50, keep=10, rerank=unavailable + ) + assert [row["video_id"] for row in matches] == ["high", "mid", "low"] + + def test_legacy_match_score_does_not_drive_the_order(self): + rows = [ + match_row("legacy", match_score=99), + match_row("real", relevance_score=1.0), + ] + matches, _ = rank_creator_candidates( + "query", rows, [], limit=50, keep=10, rerank=unavailable + ) + assert [row["video_id"] for row in matches] == ["real", "legacy"] + + def test_equal_scores_keep_insertion_order(self): + rows = [match_row("a"), match_row("b"), match_row("c")] + matches, _ = rank_creator_candidates( + "query", rows, [], limit=50, keep=10, rerank=unavailable + ) + assert [row["video_id"] for row in matches] == ["a", "b", "c"] + + def test_vector_list_is_ordered_by_similarity(self): + vectors = [vector_row("low", 0.1), vector_row("high", 0.9)] + _, vector_matches = rank_creator_candidates( + "query", [], vectors, limit=50, keep=10, rerank=unavailable + ) + assert [row["video_id"] for row in vector_matches] == ["high", "low"] + + def test_empty_query_never_calls_rerank(self): + rows = [match_row("a"), match_row("b")] + vectors = [vector_row("v", 0.5)] + + matches, vector_matches = rank_creator_candidates( + "", rows, vectors, limit=50, keep=10, rerank=never_called + ) + + assert [row["video_id"] for row in matches] == ["a", "b"] + assert len(vector_matches) == 1 + + def test_a_raising_rerank_is_contained(self): + def explode(*args, **kwargs): + raise RuntimeError("cohere exploded") + + rows = [match_row("a"), match_row("b")] + matches, vector_matches = rank_creator_candidates( + "query", rows, [vector_row("v", 0.5)], limit=50, keep=10, rerank=explode + ) + + assert len(matches) == 2 + assert len(vector_matches) == 1 + + def test_no_rerank_score_is_added(self): + matches, _ = rank_creator_candidates( + "query", [match_row("a", relevance_score=8.5)], [], limit=50, keep=10, rerank=unavailable + ) + assert "rerank_score" not in matches[0] + + +class TestRankCreatorCandidatesRerank: + def test_relevance_score_survives_and_rerank_score_is_added(self): + rows = [match_row("a", relevance_score=8.5, relevance_reasoning="keyword hit")] + + matches, _ = rank_creator_candidates( + "query", rows, [], limit=50, keep=10, rerank=lambda *a, **kw: [(0, 0.31)] + ) + + assert matches[0]["relevance_score"] == 8.5 + assert matches[0]["rerank_score"] == 0.31 + assert matches[0]["relevance_reasoning"] == "keyword hit" + + def test_the_stored_row_is_not_mutated(self): + row = match_row("a", relevance_score=8.5) + rank_creator_candidates( + "query", [row], [], limit=50, keep=10, rerank=lambda *a, **kw: [(0, 0.31)] + ) + assert "rerank_score" not in row + + def test_truncates_to_top_n_on_success(self): + rows = [match_row(f"m{i}") for i in range(12)] + seen = {} + + def rerank(query, documents, limit=None): + seen["limit"] = limit + return [(i, 1.0 - i / 100) for i in range(limit)] + + matches, _ = rank_creator_candidates("query", rows, [], limit=50, keep=3, rerank=rerank) + + assert seen["limit"] == 3 + assert len(matches) == 3 + + def test_orders_by_rerank_score_across_both_sources(self): + rows = [match_row("m1"), match_row("m2")] + vectors = [vector_row("v1", 0.9)] + + # Cohere returns best first: v1, m2, m1. + matches, vector_matches = rank_creator_candidates( + "query", rows, vectors, limit=50, keep=10, rerank=lambda *a, **kw: [(2, 0.9), (1, 0.5), (0, 0.1)] + ) + + assert [row["video_id"] for row in matches] == ["m2", "m1"] + assert [row["video_id"] for row in vector_matches] == ["v1"] + assert vector_matches[0]["rerank_score"] == 0.9 + + def test_deduplicates_a_video_present_in_both_sources(self): + documents = {} + + def rerank(query, documents_, limit=None): + documents["count"] = len(documents_) + return [(0, 0.9)] + + rank_creator_candidates( + "query", + [match_row("shared")], + [vector_row("shared", 0.9)], + limit=50, + keep=10, + rerank=rerank, + ) + assert documents["count"] == 1 + + def test_reranks_on_the_embedded_video_text(self): + captured = {} + + def rerank(query, documents, limit=None): + captured["documents"] = documents + return [(0, 0.9)] + + rank_creator_candidates( + "query", [match_row("abc")], [], limit=50, keep=10, rerank=rerank + ) + assert captured["documents"] == ["abc review"] + + def test_row_without_video_text_falls_back_to_the_video_id(self): + captured = {} + + def rerank(query, documents, limit=None): + captured["documents"] = documents + return [(0, 0.9)] + + rank_creator_candidates( + "query", + [{"video_id": "abc", "creator_videos": None}], + [], + limit=50, + keep=10, + rerank=rerank, + ) + assert captured["documents"] == ["abc"] + + +class TestFlagOffContract: + """With RETRIEVAL_RERANK_ENABLED unset the endpoint never reaches the + ranking helper; when it does reach it without a usable rerank, the result is + the pre-flag response.""" + + def test_flag_defaults_to_off(self, monkeypatch): + monkeypatch.delenv("RETRIEVAL_RERANK_ENABLED", raising=False) + assert rerank_serving_enabled() is False + + def test_unavailable_rerank_returns_the_same_rows(self): + rows = [match_row(f"m{i}") for i in range(12)] + vectors = [vector_row(f"v{i}", 0.5) for i in range(20)] + + matches, vector_matches = rank_creator_candidates( + "query", rows, vectors, limit=50, keep=10, rerank=unavailable + ) + + assert [row["video_id"] for row in matches] == [row["video_id"] for row in rows] + assert [row["video_id"] for row in vector_matches] == [row["video_id"] for row in vectors] + + def test_missing_cohere_key_takes_the_fallback(self, monkeypatch): + monkeypatch.delenv("COHERE_KEY", raising=False) + rows = [match_row(f"m{i}") for i in range(12)] + + matches, vector_matches = rank_creator_candidates( + "query", rows, [vector_row("v", 0.5)], limit=50, keep=10 + ) + + assert len(matches) == 12 + assert len(vector_matches) == 1 diff --git a/backend/utils/creator_ranking.py b/backend/utils/creator_ranking.py new file mode 100644 index 0000000..94a19b4 --- /dev/null +++ b/backend/utils/creator_ranking.py @@ -0,0 +1,107 @@ +""" +Ordering for the creator list served by /products/{id}/creators. + +Two candidate sources with incomparable scores: pre-computed rows from +product_creator_matches carry the keyword scorer's 0-10 `relevance_score`, and +vector rows carry a Pinecone cosine `score`. The rerank path rescores both in one +pass so they land on a single scale. That Cohere score is written to +`rerank_score`; the stored `relevance_score` is left alone, because +utils.relevance compares it against a 0-10 threshold. +""" +from typing import Callable, List, Optional, Sequence, Tuple + +from utils.rerank import rerank_documents + +RerankFn = Callable[..., Optional[List[Tuple[int, float]]]] + + +def creator_document_text(video: dict) -> str: + """Text a creator video is reranked on.""" + parts = [ + video.get("title") or "", + video.get("channel_title") or video.get("channel") or "", + (video.get("description") or "")[:500], + ] + return " ".join(part for part in parts if part).strip() + + +def stored_match_score(row: dict) -> float: + """The score a pre-computed match row actually carries. + + `match_score` belongs to the legacy product_matches table and is never + present here, so sorting on it was a no-op. + """ + for key in ("relevance_score", "similarity_score"): + value = row.get(key) + if value is not None: + try: + return float(value) + except (TypeError, ValueError): + return 0.0 + return 0.0 + + +def _vector_score(row: dict) -> float: + try: + return float(row.get("score") or 0) + except (TypeError, ValueError): + return 0.0 + + +def rank_creator_candidates( + query: str, + match_rows: Sequence[dict], + vector_matches: Sequence[dict], + limit: int, + keep: int, + rerank: RerankFn = rerank_documents, +) -> Tuple[List[dict], List[dict]]: + """Score pre-computed and vector candidates in one rerank pass. + + On success the combined list is truncated to `keep` (the configured top_n). + On any failure the caller's `limit` applies instead: both lists come back + whole, only re-sorted on the score each one actually carries, so an + unavailable rerank degrades to the pre-flag response rather than dropping + results and emptying the vector list. + """ + ranked = None + candidates: List[Tuple[str, dict, str]] = [] + + if query: + try: + seen = set() + for row in match_rows: + video_id = row.get("video_id") + if video_id in seen: + continue + seen.add(video_id) + video = row.get("creator_videos") or {} + candidates.append(("match", row, creator_document_text(video) or str(video_id or ""))) + for vector_match in vector_matches: + video_id = vector_match.get("video_id") + if video_id in seen: + continue + seen.add(video_id) + candidates.append( + ("vector", vector_match, creator_document_text(vector_match) or str(video_id or "")) + ) + + if candidates: + ranked = rerank(query, [doc for _, _, doc in candidates], limit=min(limit, keep)) + except Exception as e: + print(f"⚠️ Rerank stage failed, keeping similarity order: {e}") + ranked = None + + if not ranked: + return ( + sorted(match_rows, key=stored_match_score, reverse=True)[:limit], + sorted(vector_matches, key=_vector_score, reverse=True), + ) + + ranked_candidates = [ + (candidates[i][0], {**candidates[i][1], "rerank_score": score}) for i, score in ranked + ] + return ( + [payload for kind, payload in ranked_candidates if kind == "match"], + [payload for kind, payload in ranked_candidates if kind == "vector"], + ) diff --git a/backend/utils/feature_flags.py b/backend/utils/feature_flags.py new file mode 100644 index 0000000..f4bf011 --- /dev/null +++ b/backend/utils/feature_flags.py @@ -0,0 +1,35 @@ +""" +Retrieval feature flags. + +Both default off. With neither set, indexing and serving behave exactly as they +did before these flags existed. +""" +import os + +TRUTHY = {"1", "true", "yes", "on"} + + +def _enabled(name: str) -> bool: + return os.getenv(name, "false").strip().lower() in TRUTHY + + +def rerank_serving_enabled() -> bool: + """Serving path, ordering only. + + Widens the Pinecone candidate pool and orders the served creator list by + Cohere rerank score, written to `rerank_score`. The response keys are the + same either way, and an unavailable rerank degrades to the unranked + ordering. The frontend route has a second, separate flag + (CREATORS_API_EMIT_MATCHES) because its response was missing a key this + endpoint has always returned. + """ + return _enabled("RETRIEVAL_RERANK_ENABLED") + + +def document_input_type_enabled() -> bool: + """Indexing path: embed corpus content with input_type=search_document. + + Vectors written this way are not comparable with the search_query vectors + already in the index, so this stays off until the index is fully re-embedded. + """ + return _enabled("RETRIEVAL_DOC_INPUT_TYPE_ENABLED") diff --git a/backend/utils/rerank.py b/backend/utils/rerank.py new file mode 100644 index 0000000..2238da3 --- /dev/null +++ b/backend/utils/rerank.py @@ -0,0 +1,110 @@ +""" +Cohere Rerank stage for retrieval. + +Pinecone similarity is a first-pass filter; rerank scores each candidate against +the query text directly. Callers ask for a wide candidate pool and keep the top N. + +Every failure mode returns None so callers can fall back to similarity order +instead of failing the request. That includes constructing the client, so an +unimportable or misconfigured cohere never turns into a 500. +""" +import os +import threading +from typing import Any, List, Optional, Sequence, Tuple + +DEFAULT_RERANK_MODEL = "rerank-v3.5" +DEFAULT_CANDIDATE_POOL = 30 +DEFAULT_TOP_N = 10 + +# cohere.ClientV2 owns an httpx.Client, so a per-request instance leaks a +# connection pool. Cached against the key it was built from so a rotated or +# removed COHERE_KEY is still honoured. +_client_lock = threading.Lock() +_cached_client: Optional[Any] = None +_cached_client_key: Optional[str] = None + + +def rerank_model() -> str: + return os.getenv("COHERE_RERANK_MODEL", DEFAULT_RERANK_MODEL) + + +def candidate_pool() -> int: + try: + return max(1, int(os.getenv("RETRIEVAL_CANDIDATE_POOL", str(DEFAULT_CANDIDATE_POOL)))) + except ValueError: + return DEFAULT_CANDIDATE_POOL + + +def top_n() -> int: + try: + return max(1, int(os.getenv("RETRIEVAL_TOP_N", str(DEFAULT_TOP_N)))) + except ValueError: + return DEFAULT_TOP_N + + +def reset_client_cache() -> None: + """Drop the cached client. Used by tests; production never needs it.""" + global _cached_client, _cached_client_key + with _client_lock: + _cached_client = None + _cached_client_key = None + + +def _client() -> Optional[Any]: + global _cached_client, _cached_client_key + + key = os.getenv("COHERE_KEY") + if not key: + return None + + with _client_lock: + if _cached_client is None or _cached_client_key != key: + import cohere + + _cached_client = cohere.ClientV2(key) + _cached_client_key = key + return _cached_client + + +def rerank_documents( + query: str, + documents: Sequence[str], + limit: Optional[int] = None, + client: Optional[Any] = None, +) -> Optional[List[Tuple[int, float]]]: + """Rerank documents against query. + + Returns (index into documents, relevance score) best first, or None when + rerank is unavailable — missing key, empty input, or an API error. + """ + if not query or not documents: + return None + + requested = limit or top_n() + try: + co = client or _client() + if co is None: + print("⚠️ Rerank skipped: COHERE_KEY not set") + return None + + response = co.rerank( + model=rerank_model(), + query=query, + documents=list(documents), + top_n=min(requested, len(documents)), + ) + + results = getattr(response, "results", None) + if not results: + return None + + ranked: List[Tuple[int, float]] = [] + for item in results: + index = getattr(item, "index", None) + if index is None or not 0 <= index < len(documents): + continue + ranked.append((index, float(getattr(item, "relevance_score", 0.0) or 0.0))) + return ranked or None + except Exception as e: + print(f"⚠️ Rerank failed, keeping similarity order: {e}") + return None diff --git a/backend/utils/vectordb.py b/backend/utils/vectordb.py index b0e991c..70367ec 100644 --- a/backend/utils/vectordb.py +++ b/backend/utils/vectordb.py @@ -6,6 +6,7 @@ import time from typing import List, Dict, Any, TypedDict, Optional +from utils.feature_flags import document_input_type_enabled from utils.shopify import Product class ImageUrlContent(TypedDict): @@ -66,16 +67,41 @@ def imageurl_to_embedding(image_url: str) -> Any: inputs=imageurl_to_input(image_url) ) -def text_to_embedding(text: str) -> Any: +def corpus_input_type() -> str: + """Input type for corpus content. + + embed-english-v3.0 is asymmetric: corpus text belongs in search_document and + live queries in search_query. The index was built with search_query, so + switching requires a full re-embed and stays behind a flag until then. + """ + return "search_document" if document_input_type_enabled() else "search_query" + +def text_to_embedding(text: str, input_type: str = "search_query") -> Any: return co.embed( model="embed-english-v3.0", - input_type="search_query", + input_type=input_type, embedding_types=["float"], inputs=[{"content": [ {"type": "text", "text": text}, ]}] ) +def document_to_embedding(text: str) -> Any: + """Embed corpus content (products, creator videos).""" + return text_to_embedding(text, input_type=corpus_input_type()) + +def is_creator_video_match(metadata: Optional[Dict[str, Any]]) -> bool: + """Identify creator video vectors in an index shared with products. + + Vectors indexed before the type key existed are identified by video_id. + """ + if not metadata: + return False + vector_type = metadata.get("type") + if vector_type: + return vector_type == "creator_video" + return bool(metadata.get("video_id")) + def embed_products(products: List[Product]) -> List[EmbeddingItem]: items: List[EmbeddingItem] = [] @@ -94,13 +120,14 @@ def embed_products(products: List[Product]) -> List[EmbeddingItem]: print(f"🖼️ Using image embedding for '{product['name']}'") embedding = imageurl_to_embedding(image_url).embeddings.float_[0] else: - embedding = text_to_embedding(text).embeddings.float_[0] + embedding = document_to_embedding(text).embeddings.float_[0] except Exception as e: print(f"⚠️ Failed to create embedding for '{product['name']}': {e}") continue # Build metadata, filtering out None/null values (Pinecone doesn't accept them) metadata = { + "type": "product", "title": product["name"], "price": product.get("price", 0) } diff --git a/frontend/.env.example b/frontend/.env.example index fa94a47..a2611da 100644 --- a/frontend/.env.example +++ b/frontend/.env.example @@ -14,3 +14,15 @@ NEXTAUTH_SECRET=your-secret-here # Supabase (if using) NEXT_PUBLIC_SUPABASE_URL=your-supabase-url NEXT_PUBLIC_SUPABASE_ANON_KEY=your-supabase-anon-key + +# Vector search - used by /api/products/[id]/creators +PINECONE_API_KEY= +PINECONE_INDEX_NAME=products +COHERE_API_KEY= + +# Retrieval feature flags - all default off, unset means unchanged behaviour +RETRIEVAL_RERANK_ENABLED=false # Ordering only: order the served creator list by Cohere rerank score +CREATORS_API_EMIT_MATCHES=false # Response shape only: emit the `matches` array the reels view reads +COHERE_RERANK_MODEL=rerank-v3.5 +RETRIEVAL_CANDIDATE_POOL=30 +RETRIEVAL_TOP_N=10 diff --git a/frontend/package.json b/frontend/package.json index 667af55..171c5e9 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -6,7 +6,8 @@ "dev": "next dev --turbopack", "build": "next build --turbopack", "start": "next start", - "lint": "eslint" + "lint": "eslint", + "test": "node --test \"tests/*.test.ts\"" }, "dependencies": { "@dnd-kit/core": "^6.3.1", diff --git a/frontend/src/app/api/products/[id]/creators/route.ts b/frontend/src/app/api/products/[id]/creators/route.ts index b889ae0..8ab2645 100644 --- a/frontend/src/app/api/products/[id]/creators/route.ts +++ b/frontend/src/app/api/products/[id]/creators/route.ts @@ -1,5 +1,13 @@ import { supabaseAdmin } from '@/lib/supabaseAdmin'; -import { queryByText } from '@/lib/vectordb'; +import { isCreatorVideoMatch, queryByText } from '@/lib/vectordb'; +import { candidatePool, rerankDocuments, rerankTopN } from '@/lib/rerank'; +import { emitCreatorMatchesEnabled, rerankServingEnabled } from '@/lib/featureFlags'; +import { + CreatorEntry, + PRE_COMPUTED, + rankCreators, + toMatchRows, +} from '@/lib/creatorRanking'; import { NextRequest, NextResponse } from 'next/server'; export async function GET( @@ -10,6 +18,8 @@ export async function GET( const { id: productId } = await params; const { searchParams } = new URL(req.url); const limit = parseInt(searchParams.get('limit') || '50', 10); + const rerankEnabled = rerankServingEnabled(); + const emitMatches = emitCreatorMatchesEnabled(); if (!productId) { return NextResponse.json({ error: 'Product ID is required' }, { status: 400 }); @@ -30,14 +40,16 @@ export async function GET( console.error('Error fetching creator matches:', matchError); } - const preMatchedCreators = (preMatches || []).map((match) => ({ + const preMatchedCreators: CreatorEntry[] = (preMatches || []).map((match) => ({ id: match.id, video_id: match.video_id, product_id: match.product_id, match_score: match.match_score, + relevance_score: match.relevance_score, + similarity_score: match.similarity_score, created_at: match.created_at, video: match.creator_videos, - source: 'pre_computed', + source: PRE_COMPUTED, })); // Also do real-time vector search for fresh matches @@ -47,23 +59,27 @@ export async function GET( .eq('id', productId) .single(); - let vectorMatches: Array<{ - id: string; - video_id: string; - product_id: string; - match_score: number; - video: unknown; - source: string; - }> = []; + let vectorMatches: CreatorEntry[] = []; + // Built from the fields that are actually present, so a missing product + // yields '' and short-circuits to the similarity fallback instead of + // calling rerank with a dead query. + const rerankQuery = [product?.title, product?.description] + .filter(Boolean) + .join(' ') + .slice(0, 500); if (product) { try { // Search using product title + description const searchText = `${product.title} ${product.description || ''}`.slice(0, 500); - const vectorResults = await queryByText(searchText, 20); + const vectorResults = await queryByText(searchText, rerankEnabled ? candidatePool() : 20); + + const candidates = rerankEnabled + ? vectorResults.matches.filter((m) => isCreatorVideoMatch(m.metadata)) + : vectorResults.matches; // Get video details for vector matches - const videoIds = vectorResults.matches.map((m) => m.metadata.video_id).filter(Boolean); + const videoIds = candidates.map((m) => m.metadata.video_id).filter(Boolean); if (videoIds.length > 0) { const { data: videos } = await supabaseAdmin @@ -71,7 +87,7 @@ export async function GET( .select('*') .in('video_id', videoIds); - vectorMatches = vectorResults.matches + vectorMatches = candidates .map((match) => { const video = videos?.find((v) => v.video_id === match.metadata.video_id); return video @@ -85,7 +101,7 @@ export async function GET( } : null; }) - .filter((m) => m !== null) as typeof vectorMatches; + .filter((m) => m !== null) as CreatorEntry[]; } } catch (vectorError) { console.error('Vector search failed:', vectorError); @@ -100,11 +116,28 @@ export async function GET( index === self.findIndex((c) => c.video_id === creator.video_id) ); + let creators = uniqueCreators.slice(0, limit); + let ranking: string | undefined; + + if (rerankEnabled) { + ({ creators, ranking } = await rankCreators( + rerankQuery, + uniqueCreators, + limit, + rerankTopN(), + rerankDocuments + )); + } + return NextResponse.json({ - creators: uniqueCreators.slice(0, limit), - count: uniqueCreators.length, + creators, + // `count` has always reported the untruncated total on the unranked path. + // Left as-is so the flag-off response is unchanged. + count: rerankEnabled ? creators.length : uniqueCreators.length, pre_computed_count: preMatchedCreators.length, vector_search_count: vectorMatches.length, + ...(ranking ? { ranking } : {}), + ...(emitMatches ? { matches: toMatchRows(creators) } : {}), }); } catch (error) { console.error('Error fetching product creators:', error); diff --git a/frontend/src/app/dashboard/reels/page.tsx b/frontend/src/app/dashboard/reels/page.tsx index 7e23c65..f8a9dff 100644 --- a/frontend/src/app/dashboard/reels/page.tsx +++ b/frontend/src/app/dashboard/reels/page.tsx @@ -42,7 +42,7 @@ function ReelsPageContent() { }; type CreatorMatch = { - creator_videos: CreatorVideo; + creator_videos: CreatorVideo | null; }; const [data, setData] = useState(null); @@ -113,29 +113,30 @@ function ReelsPageContent() { } const result = await response.json(); - const creators = result.matches || []; - - // Filter out creators that have been interacted with - const filteredCreators = creators.filter((match: CreatorMatch) => { - const video = match.creator_videos; - return !interactedVideoIds.has(video.video_id); - }); + // `matches` is only present when CREATORS_API_EMIT_MATCHES is on; + // without it this view has always rendered empty. + const creators: CreatorMatch[] = result.matches || []; + + // A match whose creator_videos embed missed has nothing to + // render. The API drops those, this is the belt for the braces. + const videos = creators + .map((match) => match.creator_videos) + .filter((video): video is CreatorVideo => + video ? !interactedVideoIds.has(video.video_id) : false + ); // Transform API response to Reel format - const reels: Reel[] = filteredCreators.map((match: CreatorMatch) => { - const video = match.creator_videos; - return { - id: video.id, - company: video.shop_domain || user.companyId, - yt_short_url: video.url, - product_imgs: video.thumbnail ? [video.thumbnail] : [], - product_titles: [video.title], - short_id: video.video_id, - email: video.email || "", - channel_id: video.channel_id, - company_id: user.companyId - }; - }); + const reels: Reel[] = videos.map((video) => ({ + id: video.id, + company: video.shop_domain || user.companyId, + yt_short_url: video.url, + product_imgs: video.thumbnail ? [video.thumbnail] : [], + product_titles: [video.title], + short_id: video.video_id, + email: video.email || "", + channel_id: video.channel_id, + company_id: user.companyId + })); console.log(`Loaded ${reels.length} creators for product ${productId} (${creators.length - reels.length} filtered out)`); setData(reels); diff --git a/frontend/src/lib/creatorRanking.ts b/frontend/src/lib/creatorRanking.ts new file mode 100644 index 0000000..45e33f2 --- /dev/null +++ b/frontend/src/lib/creatorRanking.ts @@ -0,0 +1,110 @@ +/** + * Ordering for the creator list served by /api/products/[id]/creators. + * Mirrors backend/utils/creator_ranking.py. + * + * Two candidate sources with incomparable scores: pre-computed rows from + * product_creator_matches carry the keyword scorer's 0-10 `relevance_score`, + * vector rows carry a Pinecone cosine score. The rerank path rescores both in + * one pass and writes the Cohere 0-1 score to `rerank_score`, so + * `relevance_score` keeps meaning what the database says it means. + * + * Pure and dependency-free so it can be exercised offline. + */ + +export const PRE_COMPUTED = 'pre_computed'; + +export type CreatorVideoRow = { + video_id?: string; + title?: string; + channel_title?: string; + description?: string; + [key: string]: unknown; +}; + +export type CreatorEntry = { + id: string; + video_id: string; + product_id: string; + /** Pinecone cosine score on vector rows. Never set on pre-computed rows. */ + match_score?: number; + /** Stored 0-10 keyword score, pre-computed rows only. */ + relevance_score?: number; + similarity_score?: number; + /** Cohere 0-1 score, only present on the successful rerank path. */ + rerank_score?: number; + created_at?: string; + video: CreatorVideoRow | null; + source: string; +}; + +export type RerankFn = ( + query: string, + documents: string[], + limit?: number +) => Promise | null>; + +// Text a creator video is reranked on. +export function creatorDocumentText(entry: CreatorEntry): string { + const video = entry.video || {}; + return ( + [video.title || '', video.channel_title || '', (video.description || '').slice(0, 500)] + .filter(Boolean) + .join(' ') + .trim() || entry.video_id + ); +} + +// `match_score` is not a column on product_creator_matches, it belongs to the +// legacy product_matches table, so sorting pre-computed rows on it was a no-op. +export function entryScore(entry: CreatorEntry): number { + const raw = + entry.source === PRE_COMPUTED + ? entry.relevance_score ?? entry.similarity_score + : entry.match_score; + const value = Number(raw ?? 0); + return Number.isFinite(value) ? value : 0; +} + +/** + * On success the combined list is truncated to `keep` (the configured top_n). + * On any failure the caller's `limit` applies instead and nothing is dropped, + * so an unavailable rerank degrades to the pre-flag response. + */ +export async function rankCreators( + query: string, + creators: CreatorEntry[], + limit: number, + keep: number, + rerank: RerankFn +): Promise<{ creators: CreatorEntry[]; ranking: 'rerank' | 'similarity' }> { + const ranked = + query && creators.length > 0 + ? await rerank(query, creators.map(creatorDocumentText), Math.min(limit, keep)) + : null; + + if (!ranked) { + const byScore = (a: CreatorEntry, b: CreatorEntry) => entryScore(b) - entryScore(a); + const preComputed = creators.filter((c) => c.source === PRE_COMPUTED).sort(byScore); + const vector = creators.filter((c) => c.source !== PRE_COMPUTED).sort(byScore); + return { creators: [...preComputed, ...vector].slice(0, limit), ranking: 'similarity' }; + } + + return { + creators: ranked.map(({ index, relevanceScore }) => ({ + ...creators[index], + rerank_score: relevanceScore, + })), + ranking: 'rerank', + }; +} + +// The reels UI reads `matches` with a nested `creator_videos`, the shape the +// Python endpoint returns. A row whose FK embed missed has no video and would +// null-deref there, so it is dropped rather than emitted. +export function toMatchRows( + creators: CreatorEntry[] +): Array { + return creators + .filter((creator) => creator.video) + .map((creator) => ({ ...creator, creator_videos: creator.video as CreatorVideoRow })); +} diff --git a/frontend/src/lib/featureFlags.ts b/frontend/src/lib/featureFlags.ts new file mode 100644 index 0000000..09207b1 --- /dev/null +++ b/frontend/src/lib/featureFlags.ts @@ -0,0 +1,28 @@ +/** + * Retrieval feature flags. + * + * All default off. With none set, this route behaves exactly as it did before + * the flags existed. Mirrors backend/utils/feature_flags.py. + */ + +const TRUTHY = new Set(['1', 'true', 'yes', 'on']); + +function enabled(value: string | undefined): boolean { + return TRUTHY.has((value || 'false').trim().toLowerCase()); +} + +// Ordering only: rerank the merged candidates and order the served list by the +// Cohere score instead of by relevance/similarity score. Does not change which +// keys the response carries. +export function rerankServingEnabled(): boolean { + return enabled(process.env.RETRIEVAL_RERANK_ENABLED); +} + +// Response shape, not ordering: add the `matches` array (each entry carrying a +// nested `creator_videos`) that the reels view reads. This route has only ever +// returned `creators`, so /dashboard/reels?product_id=... renders nothing +// without it. Flagged rather than unconditional so turning it on is a +// deliberate, revertible change to a live response. +export function emitCreatorMatchesEnabled(): boolean { + return enabled(process.env.CREATORS_API_EMIT_MATCHES); +} diff --git a/frontend/src/lib/rerank.ts b/frontend/src/lib/rerank.ts new file mode 100644 index 0000000..1bd927b --- /dev/null +++ b/frontend/src/lib/rerank.ts @@ -0,0 +1,66 @@ +/** + * Cohere Rerank stage for retrieval. + * + * Pinecone similarity is a first-pass filter; rerank scores each candidate + * against the query text directly. Callers ask for a wide candidate pool and + * keep the top N. Mirrors backend/utils/rerank.py. + * + * Every failure mode returns null so callers can fall back to similarity order + * instead of failing the request. + */ + +import { getCohere } from '@/lib/vectordb'; + +export const DEFAULT_RERANK_MODEL = 'rerank-v3.5'; +const DEFAULT_CANDIDATE_POOL = 30; +const DEFAULT_TOP_N = 10; + +function intFromEnv(value: string | undefined, fallback: number): number { + const parsed = parseInt(value || '', 10); + return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; +} + +export function rerankModel(): string { + return process.env.COHERE_RERANK_MODEL || DEFAULT_RERANK_MODEL; +} + +export function candidatePool(): number { + return intFromEnv(process.env.RETRIEVAL_CANDIDATE_POOL, DEFAULT_CANDIDATE_POOL); +} + +export function rerankTopN(): number { + return intFromEnv(process.env.RETRIEVAL_TOP_N, DEFAULT_TOP_N); +} + +export async function rerankDocuments( + query: string, + documents: string[], + limit?: number +): Promise | null> { + if (!query || documents.length === 0) return null; + + const requested = limit ?? rerankTopN(); + + try { + const cohere = getCohere(); + const response = await cohere.rerank({ + model: rerankModel(), + query, + documents, + topN: Math.min(requested, documents.length), + }); + + const results = (response.results || []).filter( + (result) => result.index >= 0 && result.index < documents.length + ); + if (results.length === 0) return null; + + return results.map((result) => ({ + index: result.index, + relevanceScore: result.relevanceScore ?? 0, + })); + } catch (error) { + console.error('Rerank failed, keeping similarity order:', error); + return null; + } +} diff --git a/frontend/src/lib/vectordb.ts b/frontend/src/lib/vectordb.ts index 7863745..cbdc1bd 100644 --- a/frontend/src/lib/vectordb.ts +++ b/frontend/src/lib/vectordb.ts @@ -25,7 +25,7 @@ function getPinecone() { } // Lazy initialize Cohere client -function getCohere() { +export function getCohere() { if (!cohereClient) { if (!process.env.COHERE_API_KEY) { throw new Error('COHERE_API_KEY not configured'); @@ -42,13 +42,21 @@ export function getIndex() { return getPinecone().index(INDEX_NAME); } +// embed-english-v3.0 is asymmetric: corpus content belongs in 'search_document' +// and live queries in 'search_query'. The index was built with 'search_query', +// so corpus writers (the Python indexer) only switch once it is re-embedded. +export type EmbedInputType = 'search_query' | 'search_document'; + // Text to embedding using Cohere -export async function textToEmbedding(text: string): Promise { +export async function textToEmbedding( + text: string, + inputType: EmbedInputType = 'search_query' +): Promise { const cohere = getCohere(); const response = await cohere.embed({ texts: [text], model: 'embed-english-v3.0', - inputType: 'search_query', + inputType, }); const embeddings = response.embeddings; @@ -75,6 +83,15 @@ export async function imageUrlToEmbedding(imageUrl: string): Promise { throw new Error('Failed to generate embedding'); } +// Identify creator video vectors in an index shared with products. +// Vectors indexed before the type key existed are identified by video_id. +export function isCreatorVideoMatch(metadata: Record): boolean { + if (!metadata) return false; + const vectorType = metadata.type; + if (vectorType) return vectorType === 'creator_video'; + return Boolean(metadata.video_id); +} + // Query products by text export async function queryByText( query: string, diff --git a/frontend/tests/creatorRanking.test.ts b/frontend/tests/creatorRanking.test.ts new file mode 100644 index 0000000..e8a91a8 --- /dev/null +++ b/frontend/tests/creatorRanking.test.ts @@ -0,0 +1,185 @@ +import { test, describe } from 'node:test'; +import assert from 'node:assert/strict'; + +import type { CreatorEntry } from '../src/lib/creatorRanking.ts'; +import { + PRE_COMPUTED, + entryScore, + rankCreators, + toMatchRows, +} from '../src/lib/creatorRanking.ts'; + +function preComputed(videoId: string, fields: Partial = {}): CreatorEntry { + return { + id: videoId, + video_id: videoId, + product_id: 'p1', + video: { video_id: videoId, title: `${videoId} review` }, + source: PRE_COMPUTED, + ...fields, + }; +} + +function vectorEntry(videoId: string, score: number): CreatorEntry { + return { + id: `vector_${videoId}`, + video_id: videoId, + product_id: 'p1', + match_score: score, + video: { video_id: videoId, title: `${videoId} short` }, + source: 'real_time_vector_search', + }; +} + +const unavailable = async () => null; +const neverCalled = async () => { + throw new Error('rerank should not have been called'); +}; + +describe('entryScore', () => { + test('pre-computed rows score on relevance_score', () => { + assert.equal(entryScore(preComputed('a', { relevance_score: 8.5 })), 8.5); + }); + + test('pre-computed rows fall back to similarity_score', () => { + assert.equal(entryScore(preComputed('a', { similarity_score: 0.4 })), 0.4); + }); + + // match_score belongs to the legacy product_matches table and is never + // present on product_creator_matches rows. + test('the legacy match_score column is ignored on pre-computed rows', () => { + assert.equal(entryScore(preComputed('a', { match_score: 99 })), 0); + }); + + test('vector rows score on their Pinecone score', () => { + assert.equal(entryScore(vectorEntry('v', 0.72)), 0.72); + }); +}); + +describe('rankCreators fallback', () => { + test('keeps every candidate when rerank is unavailable', async () => { + const creators = [ + ...Array.from({ length: 12 }, (_, i) => preComputed(`m${i}`, { relevance_score: i })), + ...Array.from({ length: 6 }, (_, i) => vectorEntry(`v${i}`, 0.5)), + ]; + + const result = await rankCreators('query', creators, 50, 10, unavailable); + + assert.equal(result.ranking, 'similarity'); + assert.equal(result.creators.length, 18); + }); + + test("respects the caller's limit rather than top_n", async () => { + const creators = Array.from({ length: 12 }, (_, i) => + preComputed(`m${i}`, { relevance_score: i }) + ); + + const result = await rankCreators('query', creators, 50, 10, unavailable); + assert.equal(result.creators.length, 12); + + const limited = await rankCreators('query', creators, 5, 10, unavailable); + assert.equal(limited.creators.length, 5); + }); + + test('orders pre-computed rows by relevance_score', async () => { + const creators = [ + preComputed('low', { relevance_score: 2 }), + preComputed('high', { relevance_score: 9 }), + preComputed('mid', { relevance_score: 5 }), + ]; + + const result = await rankCreators('query', creators, 50, 10, unavailable); + assert.deepEqual( + result.creators.map((c) => c.video_id), + ['high', 'mid', 'low'] + ); + }); + + test('equal scores keep insertion order', async () => { + const creators = [preComputed('a'), preComputed('b'), preComputed('c')]; + const result = await rankCreators('query', creators, 50, 10, unavailable); + assert.deepEqual( + result.creators.map((c) => c.video_id), + ['a', 'b', 'c'] + ); + }); + + test('the vector candidates are not dropped', async () => { + const creators = [preComputed('m1', { relevance_score: 1 }), vectorEntry('v1', 0.9)]; + const result = await rankCreators('query', creators, 50, 10, unavailable); + assert.ok(result.creators.some((c) => c.video_id === 'v1')); + }); + + test('an empty query never calls rerank', async () => { + const creators = [preComputed('a'), preComputed('b')]; + const result = await rankCreators('', creators, 50, 10, neverCalled); + assert.equal(result.ranking, 'similarity'); + assert.equal(result.creators.length, 2); + }); + + test('no rerank_score is added on the fallback path', async () => { + const result = await rankCreators('query', [preComputed('a')], 50, 10, unavailable); + assert.equal('rerank_score' in result.creators[0], false); + }); +}); + +describe('rankCreators rerank path', () => { + test('relevance_score survives untouched and rerank_score is added', async () => { + const creators = [preComputed('a', { relevance_score: 8.5 })]; + + const result = await rankCreators('query', creators, 50, 10, async () => [ + { index: 0, relevanceScore: 0.31 }, + ]); + + assert.equal(result.ranking, 'rerank'); + assert.equal(result.creators[0].relevance_score, 8.5); + assert.equal(result.creators[0].rerank_score, 0.31); + }); + + test('asks cohere for min(limit, top_n) and reranks the video text', async () => { + const captured: { documents?: string[]; limit?: number } = {}; + const creators = [preComputed('a'), vectorEntry('v', 0.5)]; + + await rankCreators('query', creators, 50, 3, async (_query, documents, limit) => { + captured.documents = documents; + captured.limit = limit; + return [{ index: 0, relevanceScore: 0.9 }]; + }); + + assert.equal(captured.limit, 3); + assert.deepEqual(captured.documents, ['a review', 'v short']); + }); + + test('orders by the returned ranking across both sources', async () => { + const creators = [preComputed('m1'), preComputed('m2'), vectorEntry('v1', 0.9)]; + + const result = await rankCreators('query', creators, 50, 10, async () => [ + { index: 2, relevanceScore: 0.9 }, + { index: 1, relevanceScore: 0.5 }, + { index: 0, relevanceScore: 0.1 }, + ]); + + assert.deepEqual( + result.creators.map((c) => c.video_id), + ['v1', 'm2', 'm1'] + ); + }); +}); + +describe('toMatchRows', () => { + test('nests the video under creator_videos, the shape the reels view reads', () => { + const rows = toMatchRows([preComputed('a')]); + assert.equal(rows.length, 1); + assert.equal(rows[0].creator_videos.video_id, 'a'); + }); + + // reels/page.tsx dereferences match.creator_videos.video_id, so a row whose + // FK embed missed must never reach it. + test('drops rows whose creator_videos embed missed', () => { + const rows = toMatchRows([preComputed('a'), preComputed('b', { video: null })]); + assert.deepEqual( + rows.map((r) => r.video_id), + ['a'] + ); + }); +}); diff --git a/frontend/tests/featureFlags.test.ts b/frontend/tests/featureFlags.test.ts new file mode 100644 index 0000000..4b42295 --- /dev/null +++ b/frontend/tests/featureFlags.test.ts @@ -0,0 +1,44 @@ +import { test, describe, beforeEach } from 'node:test'; +import assert from 'node:assert/strict'; + +import { emitCreatorMatchesEnabled, rerankServingEnabled } from '../src/lib/featureFlags.ts'; + +const FLAGS = ['RETRIEVAL_RERANK_ENABLED', 'CREATORS_API_EMIT_MATCHES']; + +describe('retrieval feature flags', () => { + beforeEach(() => { + for (const flag of FLAGS) delete process.env[flag]; + }); + + test('both default off', () => { + assert.equal(rerankServingEnabled(), false); + assert.equal(emitCreatorMatchesEnabled(), false); + }); + + test('truthy values enable', () => { + for (const value of ['true', 'TRUE', '1', 'yes', 'on', ' on ']) { + process.env.RETRIEVAL_RERANK_ENABLED = value; + assert.equal(rerankServingEnabled(), true, value); + } + }); + + test('anything else stays off', () => { + for (const value of ['false', '0', '', 'off', 'maybe']) { + process.env.RETRIEVAL_RERANK_ENABLED = value; + assert.equal(rerankServingEnabled(), false, value); + } + }); + + // The ordering flag and the response-shape flag are separate on purpose: + // turning on rerank must not start emitting `matches`, and vice versa. + test('the two flags are independent', () => { + process.env.RETRIEVAL_RERANK_ENABLED = 'true'; + assert.equal(rerankServingEnabled(), true); + assert.equal(emitCreatorMatchesEnabled(), false); + + delete process.env.RETRIEVAL_RERANK_ENABLED; + process.env.CREATORS_API_EMIT_MATCHES = 'true'; + assert.equal(rerankServingEnabled(), false); + assert.equal(emitCreatorMatchesEnabled(), true); + }); +}); diff --git a/frontend/tsconfig.json b/frontend/tsconfig.json index f3282ef..99330b4 100644 --- a/frontend/tsconfig.json +++ b/frontend/tsconfig.json @@ -38,6 +38,7 @@ ".next/types/**/*.ts" ], "exclude": [ - "node_modules" + "node_modules", + "tests" ] }