diff --git a/backend/alembic/versions/q9c0d1e2f364_question_fulltext.py b/backend/alembic/versions/q9c0d1e2f364_question_fulltext.py new file mode 100644 index 0000000..1580bf8 --- /dev/null +++ b/backend/alembic/versions/q9c0d1e2f364_question_fulltext.py @@ -0,0 +1,29 @@ +"""Full-text search vector on questions, for hybrid lexical + vector retrieval. + +Revision ID: q9c0d1e2f364 +Revises: p8b9c0d1e253 +""" +from alembic import op + +revision = "q9c0d1e2f364" +down_revision = "p8b9c0d1e253" +branch_labels = None +depends_on = None + + +def upgrade(): + # Generated column keeps the index in step with edits without a trigger. + op.execute(""" + ALTER TABLE questions ADD COLUMN IF NOT EXISTS search_vector tsvector + GENERATED ALWAYS AS ( + setweight(to_tsvector('english', coalesce(question_text, '')), 'A') || + setweight(to_tsvector('english', coalesce(options::text, '')), 'B') || + setweight(to_tsvector('english', coalesce(explanation, '')), 'C') + ) STORED + """) + op.execute("CREATE INDEX IF NOT EXISTS ix_questions_search_vector ON questions USING GIN (search_vector)") + + +def downgrade(): + op.execute("DROP INDEX IF EXISTS ix_questions_search_vector") + op.execute("ALTER TABLE questions DROP COLUMN IF EXISTS search_vector") diff --git a/backend/alembic/versions/r0d1e2f3a475_embedding_provenance.py b/backend/alembic/versions/r0d1e2f3a475_embedding_provenance.py new file mode 100644 index 0000000..c04fb41 --- /dev/null +++ b/backend/alembic/versions/r0d1e2f3a475_embedding_provenance.py @@ -0,0 +1,26 @@ +"""Record which model produced each stored embedding. + +Without this a model change silently mixes vectors from different embedding +spaces, and cosine distance between them is meaningless. + +Revision ID: r0d1e2f3a475 +Revises: q9c0d1e2f364 +""" +from alembic import op + +revision = "r0d1e2f3a475" +down_revision = "q9c0d1e2f364" +branch_labels = None +depends_on = None + + +def upgrade(): + op.execute("ALTER TABLE questions ADD COLUMN IF NOT EXISTS embedding_model VARCHAR(120)") + op.execute("ALTER TABLE questions ADD COLUMN IF NOT EXISTS embedded_at TIMESTAMP") + op.execute("CREATE INDEX IF NOT EXISTS ix_questions_embedding_model ON questions(embedding_model)") + + +def downgrade(): + op.execute("DROP INDEX IF EXISTS ix_questions_embedding_model") + op.execute("ALTER TABLE questions DROP COLUMN IF EXISTS embedded_at") + op.execute("ALTER TABLE questions DROP COLUMN IF EXISTS embedding_model") diff --git a/backend/alembic/versions/s1e2f3a4b586_bge_m3_embeddings.py b/backend/alembic/versions/s1e2f3a4b586_bge_m3_embeddings.py new file mode 100644 index 0000000..359c924 --- /dev/null +++ b/backend/alembic/versions/s1e2f3a4b586_bge_m3_embeddings.py @@ -0,0 +1,28 @@ +"""Re-point the question vector column at BGE-M3 (1024 dims). + +Vectors from a different model share no space with the new ones, so the column is +cleared rather than converted; `retry_missing_embeddings` refills it and the +provenance columns keep the empty state visible instead of silent. + +Revision ID: s1e2f3a4b586 +Revises: r0d1e2f3a475 +""" +from alembic import op + +revision = "s1e2f3a4b586" +down_revision = "r0d1e2f3a475" +branch_labels = None +depends_on = None + + +def upgrade(): + op.execute("DROP INDEX IF EXISTS questions_embedding_idx") + op.execute("ALTER TABLE questions DROP COLUMN IF EXISTS embedding") + op.execute("ALTER TABLE questions ADD COLUMN embedding vector(1024)") + op.execute("UPDATE questions SET embedding_model = NULL, embedded_at = NULL") + + +def downgrade(): + op.execute("ALTER TABLE questions DROP COLUMN IF EXISTS embedding") + op.execute("ALTER TABLE questions ADD COLUMN embedding vector(1024)") + op.execute("UPDATE questions SET embedding_model = NULL, embedded_at = NULL") diff --git a/backend/app/config.py b/backend/app/config.py index 8229f00..37d0886 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -23,6 +23,12 @@ class Settings(BaseSettings): AWS_SECRET_ACCESS_KEY: str = "" AWS_REGION: str = "us-east-1" AWS_BEDROCK_REGION: str = "us-east-1" + # Embeddings run locally by default: a search must not depend on a remote + # service being up, and a local model cannot change under us at runtime. + # BGE-M3 via the existing LiteLLM proxy — no extra credential. The retry + # task backfills anything an outage leaves unembedded, and search still + # answers from full text while the semantic half is unavailable. + EMBEDDING_PROVIDER: str = "litellm" EMBEDDING_DIMENSIONS: int = 1024 APP_URL: str = "https://quiz.danvics.com" diff --git a/backend/app/models/question.py b/backend/app/models/question.py index 7646e67..d1f74d7 100644 --- a/backend/app/models/question.py +++ b/backend/app/models/question.py @@ -1,5 +1,5 @@ from pgvector.sqlalchemy import Vector -from sqlalchemy import Column, Integer, String, Text, JSON, ForeignKey +from sqlalchemy import Column, DateTime, Integer, String, Text, JSON, ForeignKey from sqlalchemy.orm import relationship, deferred from app.config import settings @@ -29,7 +29,11 @@ class Question(Base): difficulty = Column(String(10), nullable=True) # easy | medium | hard user_id = Column(Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True) is_shared = Column(Integer, default=1) # 1 = visible in bank, 0 = private (only owner sees it) - embedding = deferred(Column(Vector(1024), nullable=True)) # semantic search vector — deferred: not loaded in standard queries + embedding = deferred(Column(Vector(settings.EMBEDDING_DIMENSIONS), nullable=True)) # semantic search vector — deferred: not loaded in standard queries + # Which model produced `embedding`. Vectors from different models are not + # comparable, so a model change must be detectable rather than silent. + embedding_model = Column(String(120), nullable=True, index=True) + embedded_at = Column(DateTime, nullable=True) question_category = relationship("QuestionCategory", back_populates="questions", foreign_keys=[question_category_id]) diff --git a/backend/app/routers/admin.py b/backend/app/routers/admin.py index 31f6dcd..2f0251b 100644 --- a/backend/app/routers/admin.py +++ b/backend/app/routers/admin.py @@ -604,11 +604,27 @@ def rollback_classification_snapshot( } +@router.get("/embedding/health") +def embedding_health(db: Session = Depends(get_db), admin: User = Depends(require_admin)): + """How much of the bank is semantically searchable under the active model. + + Vectors from two different embedding models are not comparable, so a model + change has to be visible rather than silently degrading search quality. + """ + from app.services import embedding_service + + return embedding_service.stale_embedding_counts(db) + + @router.post("/embedding/regenerate") -def regenerate_embeddings(admin: User = Depends(require_admin)): - """Queue a background Celery task to regenerate all question embeddings.""" +def regenerate_embeddings( + stale_only: bool = Query(True, description="Only rows with no vector or a vector from another model"), + admin: User = Depends(require_admin), +): + """Queue a background task to re-embed questions with the current model.""" import uuid from app.tasks.quiz_tasks import regenerate_embeddings as regen_task job_id = str(uuid.uuid4()) - regen_task.delay(job_id, admin.id) - return {"job_id": job_id, "message": "Regeneration started — progress visible in the Jobs badge."} + regen_task.delay(job_id, admin.id, stale_only) + scope = "missing and stale" if stale_only else "all" + return {"job_id": job_id, "message": f"Regenerating {scope} embeddings — progress in the Jobs badge."} diff --git a/backend/app/routers/questions.py b/backend/app/routers/questions.py index 3a132cb..a9d67eb 100644 --- a/backend/app/routers/questions.py +++ b/backend/app/routers/questions.py @@ -23,6 +23,7 @@ from app.models.question_category import QuestionCategory, QuestionCategoryLink from app.models.quiz import Quiz from app.models.user import User from app.models.favorite import Favorite +from app.services.search_service import hybrid_question_ids from app.services.quiz_builder import (bank_query, category_descendants, filtered_bank_query, CreateFromBankRequest, GenerateTestRequest, create_saved_test, generate_test) from app.utils.auth import get_current_user, require_moderator @@ -31,6 +32,9 @@ from app.utils.category_grants import (assert_can_manage_category, assert_can_ma router = APIRouter() +# Ranked retrieval is capped so a broad query cannot pull the whole bank. +MAX_SEARCH_RESULTS = 500 + def parse_category_ids(value): try: @@ -250,7 +254,6 @@ def get_question_bank( difficulty: Literal["easy", "medium", "hard"] | None = Query(None), article_ids: str | None = Query(None, description="Comma-separated article IDs (OR filter)"), tag_ids: str | None = Query(None, description="Comma-separated tag IDs (AND filter)"), - search_mode: str = Query("hybrid"), # "keyword" | "semantic" | "hybrid" limit: int = Query(50, le=200), offset: int = Query(0), db: Session = Depends(get_db), @@ -320,49 +323,23 @@ def get_question_bank( return {"total": 0, "questions": []} query = query.filter(Question.id.in_(fav_ids)) - # ── Semantic search (pgvector) ───────────────────────────────── - semantic_ids_ordered: list[int] = [] - if q and q.strip() and search_mode in ("semantic", "hybrid"): - from app.services.embedding_service import generate_embedding - emb = generate_embedding(q.strip()) - if emb: - # Validate all values are finite floats before interpolating into SQL - emb_literal = "[" + ",".join(str(float(x)) for x in emb) + "]" - rows = db.execute(sa_text(""" - SELECT id, 1 - (embedding <=> CAST(:vec AS vector)) AS sim - FROM questions - WHERE embedding IS NOT NULL - ORDER BY embedding <=> CAST(:vec AS vector) - LIMIT :lim - """), {"vec": emb_literal, "lim": limit * 2}).fetchall() - semantic_ids_ordered = [r.id for r in rows if float(r.sim) >= 0.55] - - # ── Keyword filter ───────────────────────────────────────────── - if q and q.strip() and search_mode in ("keyword", "hybrid"): - phrase = q.strip() - query = query.filter( - or_( - Question.question_text.ilike(f"%{phrase}%"), - cast(Question.options, String).ilike(f"%{phrase}%"), - ) - ) - - # Apply semantic ID filter if semantic-only mode - if q and q.strip() and search_mode == "semantic" and semantic_ids_ordered: - query = query.filter(Question.id.in_(semantic_ids_ordered)) - - total = query.count() - questions = query.order_by(Question.source_quiz_id, Question.id).offset(offset).limit(limit).all() - - # If hybrid: merge semantic first then keyword remainder - if semantic_ids_ordered and search_mode == "hybrid": - sem_set = set(semantic_ids_ordered) - sem_qs = [qu for qu in questions if qu.id in sem_set] - kw_qs = [qu for qu in questions if qu.id not in sem_set] - # Sort semantic by original similarity order - sem_order = {qid: i for i, qid in enumerate(semantic_ids_ordered)} - sem_qs.sort(key=lambda qu: sem_order.get(qu.id, 999)) - questions = sem_qs + kw_qs + # ── Hybrid retrieval: full text fused with embeddings ────────── + # Always both. Keyword-only silently drops the question that asks the same + # thing in different words, which is the one a concept search wants. + semantic_ids: set[int] = set() + if q and q.strip(): + ranked_ids, semantic_ids = hybrid_question_ids(db, q.strip(), limit=MAX_SEARCH_RESULTS) + if not ranked_ids: + return {"total": 0, "questions": []} + query = query.filter(Question.id.in_(ranked_ids)) + rank_of = {question_id: position for position, question_id in enumerate(ranked_ids)} + matched = query.all() + matched.sort(key=lambda question: rank_of.get(question.id, len(rank_of))) + total = len(matched) + questions = matched[offset:offset + limit] + else: + total = query.count() + questions = query.order_by(Question.source_quiz_id, Question.id).offset(offset).limit(limit).all() quiz_cache: dict[int, str] = {} cat_cache: dict[int, str] = {} @@ -402,6 +379,7 @@ def get_question_bank( "difficulty": qu.difficulty, "user_id": qu.user_id, "is_shared": qu.is_shared if qu.is_shared is not None else 1, + "match_source": "semantic" if qu.id in semantic_ids else "keyword", }) return {"total": total, "questions": result} diff --git a/backend/app/services/embedding_service.py b/backend/app/services/embedding_service.py index 771a74f..d315f39 100644 --- a/backend/app/services/embedding_service.py +++ b/backend/app/services/embedding_service.py @@ -5,7 +5,10 @@ Priority: 2. LiteLLM proxy — model from LITELLM_EMBEDDING_MODEL env 3. AWS Bedrock Titan Embed V2 (direct fallback) """ +import hashlib +import json import logging +from datetime import datetime from app.config import settings @@ -13,7 +16,7 @@ logger = logging.getLogger(__name__) def _get_embedding_model() -> str: - """Return the active embedding model: Redis setting takes precedence over env.""" + """The active embedding model name — also the provenance stamp on each vector.""" try: import redis as redis_lib r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True) @@ -34,23 +37,32 @@ def _text_for_question(question_text: str, options: list[str] | None) -> str: def generate_embedding(text: str) -> list[float] | None: - """Generate a 1024-dim embedding. + """Return a vector of EMBEDDING_DIMENSIONS floats, or None. - Priority: - 1. LiteLLM proxy (openai/titan-embed-v2) — scores ~0.71 cosine similarity - 2. AWS Bedrock direct — fallback, scores ~0.48 + None means "not embedded yet", never "embedded badly": a wrong-sized vector + is rejected rather than stored, because the index cannot compare it. The + `retry_missing_embeddings` task refills whatever a failure leaves behind. + + Order: the LiteLLM proxy (BGE-M3), then Bedrock. """ clean = " ".join(text.split())[:4000] if not clean: return None - # ── 1. LiteLLM proxy (direct httpx — avoids LiteLLM param validation) ── + # ── LiteLLM proxy (direct httpx — avoids LiteLLM param validation) ── + # Must be the same resolver that stamps provenance, or a vector gets labelled + # with a model that did not produce it — which is the exact failure the + # provenance columns exist to catch. embedding_model = _get_embedding_model() api_base = (settings.LITELLM_API_BASE or "").rstrip("/").removesuffix("/v1") if embedding_model and settings.LITELLM_API_KEY and api_base: try: import httpx, json as _json - body: dict = {"model": embedding_model, "input": [clean], "dimensions": settings.EMBEDDING_DIMENSIONS} + body: dict = {"model": embedding_model, "input": [clean]} + # Only OpenAI's embedding-3 family accepts a truncation size; sending + # it to BGE-M3 is rejected outright. + if "embedding-3" in embedding_model: + body["dimensions"] = settings.EMBEDDING_DIMENSIONS resp = httpx.post( f"{api_base}/v1/embeddings", headers={"Authorization": f"Bearer {settings.LITELLM_API_KEY}", "Content-Type": "application/json"}, @@ -94,10 +106,42 @@ def generate_embedding(text: str) -> list[float] | None: def embed_question(question) -> bool: - """Generate and store embedding for a Question ORM object. Returns True on success.""" + """Generate and store an embedding, recording which model produced it. + + Stamping the model is what makes a later model change detectable: vectors + from two different models share no space, so cosine distance between them is + noise. `stale_embedding_counts` reports the mix; the regenerate task re-embeds + anything not on the active model. + """ text = _text_for_question(question.question_text, question.options) + model = _get_embedding_model() emb = generate_embedding(text) if emb: question.embedding = emb + question.embedding_model = model + question.embedded_at = datetime.utcnow() return True return False + + +def stale_embedding_counts(db) -> dict: + """How much of the bank is searchable semantically, and how much is stale.""" + from sqlalchemy import func + + from app.models.question import Question + + active = _get_embedding_model() + total = db.query(func.count(Question.id)).scalar() or 0 + missing = db.query(func.count(Question.id)).filter(Question.embedding.is_(None)).scalar() or 0 + stale = db.query(func.count(Question.id)).filter( + Question.embedding.isnot(None), + (Question.embedding_model.is_(None)) | (Question.embedding_model != active), + ).scalar() or 0 + return { + "active_model": active, + "total": total, + "missing": missing, + "stale": stale, + "current": max(0, total - missing - stale), + "needs_regeneration": missing + stale > 0, + } diff --git a/backend/app/services/search_service.py b/backend/app/services/search_service.py new file mode 100644 index 0000000..9a21b00 --- /dev/null +++ b/backend/app/services/search_service.py @@ -0,0 +1,140 @@ +"""Hybrid question retrieval: Postgres full text fused with pgvector similarity. + +Why not OpenSearch/Elasticsearch: the lexical half of this problem is ordinary +ranked text matching that Postgres already does with `tsvector`/`ts_rank_cd`, +and the semantic half already runs on pgvector with embeddings that are stored +and kept current. A search cluster would add a second datastore to keep in sync, +a JVM's memory footprint on this host, and a new failure mode, to replace an +index Postgres maintains for free inside the same transaction. + +The two rankers are combined with Reciprocal Rank Fusion rather than a weighted +score, because a BM25-style rank and a cosine distance are not on comparable +scales; RRF only needs each ranker's ordering. + +Retrieval is always hybrid. A keyword-only mode looks precise but silently drops +the question that asks the same thing in different words, which is exactly the +question a learner searching a concept wants. +""" +import hashlib +import json +import logging + +from sqlalchemy import text as sa_text +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + +# Rank-fusion constant. 60 is the value from the original RRF paper; it damps +# the head of each list so one ranker cannot dominate on its top hit alone. +RRF_K = 60 +# Per-ranker candidate pool. Wider than the page so fusion has room to reorder. +POOL_MULTIPLIER = 4 +MIN_POOL = 60 +SEMANTIC_FLOOR = 0.45 +# Query embeddings are deterministic per model, so a day is safe and cheap. +QUERY_CACHE_TTL = 24 * 3600 + + +def _is_postgres(db: Session) -> bool: + return db.bind is not None and db.bind.dialect.name == "postgresql" + + +def _lexical_ranked(db: Session, query_text: str, pool: int) -> list[int]: + """Question ids by full-text relevance, best first.""" + if not _is_postgres(db): + # SQLite (tests): substring matching keeps the fusion path exercised. + rows = db.execute(sa_text(""" + SELECT id FROM questions + WHERE lower(question_text) LIKE :like OR lower(CAST(options AS TEXT)) LIKE :like + ORDER BY id LIMIT :pool + """), {"like": f"%{query_text.lower()}%", "pool": pool}).fetchall() + return [row[0] for row in rows] + rows = db.execute(sa_text(""" + SELECT id FROM questions + WHERE search_vector @@ websearch_to_tsquery('english', :q) + ORDER BY ts_rank_cd(search_vector, websearch_to_tsquery('english', :q)) DESC, id + LIMIT :pool + """), {"q": query_text, "pool": pool}).fetchall() + return [row[0] for row in rows] + + +def _query_embedding(query_text: str) -> list[float] | None: + """Embed a search query, cached per model so typing is not a round-trip per keystroke. + + Cached under the active model's name, so switching models cannot serve a + vector from the previous embedding space. + """ + from app.services.embedding_service import _get_embedding_model, generate_embedding + + model = _get_embedding_model() + key = f"qsearch:emb:{model}:{hashlib.sha256(query_text.encode()).hexdigest()[:32]}" + cache = None + try: + import redis as redis_lib + + from app.config import settings + + cache = redis_lib.from_url(settings.REDIS_URL, decode_responses=True) + hit = cache.get(key) + if hit: + return json.loads(hit) + except Exception: + cache = None # Redis is a nicety here; the embedder still answers. + + embedding = generate_embedding(query_text) + if embedding and cache is not None: + try: + cache.setex(key, QUERY_CACHE_TTL, json.dumps(embedding)) + except Exception: + logger.debug("Could not cache query embedding", exc_info=True) + return embedding + + +def _semantic_ranked(db: Session, query_text: str, pool: int) -> list[int]: + """Question ids by embedding similarity, nearest first.""" + if not _is_postgres(db): + return [] + embedding = _query_embedding(query_text) + if not embedding: + return [] + literal = "[" + ",".join(str(float(value)) for value in embedding) + "]" + rows = db.execute(sa_text(""" + SELECT id, 1 - (embedding <=> CAST(:vec AS vector)) AS similarity + FROM questions + WHERE embedding IS NOT NULL + ORDER BY embedding <=> CAST(:vec AS vector) + LIMIT :pool + """), {"vec": literal, "pool": pool}).fetchall() + return [row.id for row in rows if float(row.similarity) >= SEMANTIC_FLOOR] + + +def hybrid_question_ids(db: Session, query_text: str, limit: int = 200) -> tuple[list[int], set[int]]: + """Return (ids best-first, ids the semantic ranker contributed). + + The result is the *union* of both rankers. An earlier implementation + intersected them, so a question that matched the meaning but not the literal + string could never be returned no matter how well it scored. + """ + query_text = (query_text or "").strip() + if not query_text: + return [], set() + pool = max(MIN_POOL, limit * POOL_MULTIPLIER) + + try: + lexical = _lexical_ranked(db, query_text, pool) + except Exception: + logger.warning("Lexical search unavailable; falling back to semantic only", exc_info=True) + lexical = [] + try: + semantic = _semantic_ranked(db, query_text, pool) + except Exception: + logger.warning("Semantic search unavailable; falling back to lexical only", exc_info=True) + semantic = [] + + scores: dict[int, float] = {} + for ranked in (lexical, semantic): + for position, question_id in enumerate(ranked): + scores[question_id] = scores.get(question_id, 0.0) + 1.0 / (RRF_K + position + 1) + + ordered = sorted(scores, key=lambda qid: (-scores[qid], qid)) + return ordered[:limit], set(semantic) diff --git a/backend/app/tasks/__init__.py b/backend/app/tasks/__init__.py index 4fe1c07..19e95cc 100644 --- a/backend/app/tasks/__init__.py +++ b/backend/app/tasks/__init__.py @@ -17,3 +17,13 @@ celery_app.conf.result_serializer = "json" celery_app.conf.accept_content = ["json"] celery_app.conf.worker_hijack_root_logger = False # Don't override our JSON logging celery_app.conf.broker_connection_retry_on_startup = True + +# Questions whose embedding failed at creation would otherwise never be +# searchable semantically; this sweeps them up. It normally finds nothing. +celery_app.conf.beat_schedule = { + "retry-missing-embeddings": { + "task": "retry_missing_embeddings", + "schedule": 900.0, # every 15 minutes + }, +} +celery_app.conf.timezone = "UTC" diff --git a/backend/app/tasks/quiz_tasks.py b/backend/app/tasks/quiz_tasks.py index c8dd9a5..8339541 100644 --- a/backend/app/tasks/quiz_tasks.py +++ b/backend/app/tasks/quiz_tasks.py @@ -528,9 +528,54 @@ Questions: db.close() +@celery_app.task(name="retry_missing_embeddings") +def retry_missing_embeddings(batch: int = 200) -> dict: + """Backfill questions that have no usable vector. + + Embedding at creation time is best effort: if the encoder is briefly + unavailable the question is still saved, and without this it would stay + invisible to semantic search forever. Runs on a schedule and normally finds + nothing. Also catches rows left by an embedding-model change. + """ + db = SessionLocal() + try: + from app.models.question import Question + from app.services import embedding_service + + active = embedding_service._get_embedding_model() + pending = ( + db.query(Question) + .filter( + (Question.embedding.is_(None)) + | (Question.embedding_model.is_(None)) + | (Question.embedding_model != active) + ) + .limit(batch) + .all() + ) + embedded = 0 + for question in pending: + try: + if embedding_service.embed_question(question): + embedded += 1 + except Exception: + logger.warning("Retry embedding failed for question %s", question.id, exc_info=True) + if embedded: + db.commit() + logger.info("Backfilled %s embeddings (%s pending in this batch)", embedded, len(pending)) + return {"pending": len(pending), "embedded": embedded, "model": active} + finally: + db.close() + + @celery_app.task(name="regenerate_embeddings", bind=True) -def regenerate_embeddings(self, job_id: str, user_id: int): - """Regenerate embeddings for all questions using the current embedding model.""" +def regenerate_embeddings(self, job_id: str, user_id: int, stale_only: bool = True): + """Re-embed questions with the current model. + + `stale_only` (the default) covers exactly what breaks semantic search: rows + with no vector, and rows whose vector came from a different model and so sits + in an incomparable space. Pass False to rebuild the whole bank. + """ r = _redis() r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS) r.set(f"extraction:job_title:{job_id}", "Regenerate Embeddings", ex=EXPIRE_SECONDS) @@ -542,9 +587,18 @@ def regenerate_embeddings(self, job_id: str, user_id: int): from app.models.question import Question from app.services import embedding_service - questions = db.query(Question).all() + query = db.query(Question) + if stale_only: + active = embedding_service._get_embedding_model() + query = query.filter( + (Question.embedding.is_(None)) + | (Question.embedding_model.is_(None)) + | (Question.embedding_model != active) + ) + questions = query.all() total = len(questions) - _push_step(r, job_id, "start", f"Regenerating embeddings for {total} questions…") + scope = "missing or stale" if stale_only else "all" + _push_step(r, job_id, "start", f"Regenerating embeddings for {total} {scope} questions…") ok = 0 for i, q in enumerate(questions): diff --git a/backend/tests/test_hybrid_search.py b/backend/tests/test_hybrid_search.py new file mode 100644 index 0000000..751acae --- /dev/null +++ b/backend/tests/test_hybrid_search.py @@ -0,0 +1,157 @@ +"""Hybrid retrieval: lexical and semantic rankers fused, never intersected. + +Run: DATABASE_URL=sqlite:///:memory: PYTHONPATH=backend python -m unittest discover -s backend/tests +""" +import os +os.environ["DATABASE_URL"] = "sqlite:///:memory:" + +import unittest +from unittest.mock import patch + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session +from sqlalchemy.pool import StaticPool + +from app.database import Base +from app.models.course import Course # noqa — Quiz.course_id FK needs the table in metadata. +from app.models.question import Question +from app.models.user import User +from app.services import search_service + + +class HybridSearchTests(unittest.TestCase): + def setUp(self): + self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) + Base.metadata.create_all(self.engine) + self.db = Session(self.engine) + self.db.add(User(id=1, name="Mod", email="mod@example.test", hashed_password="unused", role="moderator")) + for qid, text in [ + (1, "A child with fever and a seizure"), + (2, "An infant with jaundice on day three"), + (3, "A toddler with a febrile convulsion"), + ]: + self.db.add(Question(id=qid, user_id=1, is_shared=1, question_text=text, + question_type="mcq", options=["yes", "no"], correct_answer="yes")) + self.db.commit() + + def tearDown(self): + self.db.close() + self.engine.dispose() + + def test_result_is_the_union_of_both_rankers(self): + # "convulsion" never appears in question 1, and "fever" never in 3, so an + # intersection would return one of them; the union must return both. + with patch.object(search_service, "_semantic_ranked", return_value=[3]): + ids, semantic = search_service.hybrid_question_ids(self.db, "fever") + self.assertEqual(set(ids), {1, 3}) + self.assertEqual(semantic, {3}) + + def test_agreement_between_rankers_outranks_a_single_hit(self): + with patch.object(search_service, "_semantic_ranked", return_value=[3, 1]): + ids, _ = search_service.hybrid_question_ids(self.db, "fever") + # Question 1 is first lexically and second semantically; 3 only appears once + # in the lexical list, so fusion should put 1 ahead of the semantic-only hit. + self.assertEqual(ids[0], 1) + + def test_a_failing_ranker_degrades_instead_of_erroring(self): + with patch.object(search_service, "_semantic_ranked", side_effect=RuntimeError("no pgvector")): + ids, semantic = search_service.hybrid_question_ids(self.db, "fever") + self.assertEqual(ids, [1]) + self.assertEqual(semantic, set()) + + with patch.object(search_service, "_lexical_ranked", side_effect=RuntimeError("no index")), \ + patch.object(search_service, "_semantic_ranked", return_value=[2]): + ids, _ = search_service.hybrid_question_ids(self.db, "fever") + self.assertEqual(ids, [2]) + + def test_blank_query_returns_nothing(self): + self.assertEqual(search_service.hybrid_question_ids(self.db, " "), ([], set())) + + +class EmbeddingProvenanceTests(unittest.TestCase): + """A model change must be visible, not a silent quality regression.""" + + def setUp(self): + self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) + Base.metadata.create_all(self.engine) + self.db = Session(self.engine) + self.db.add(User(id=1, name="Mod", email="mod@example.test", hashed_password="unused", role="moderator")) + for qid in (1, 2, 3): + self.db.add(Question(id=qid, user_id=1, is_shared=1, question_text=f"Question {qid}", + question_type="mcq", options=["yes", "no"], correct_answer="yes")) + self.db.commit() + + def tearDown(self): + self.db.close() + self.engine.dispose() + + def test_embedding_records_the_model_that_produced_it(self): + from app.services import embedding_service + + question = self.db.get(Question, 1) + with patch.object(embedding_service, "_get_embedding_model", return_value="model-a"), \ + patch.object(embedding_service, "generate_embedding", return_value=[0.1] * 1024): + self.assertTrue(embedding_service.embed_question(question)) + self.assertEqual(question.embedding_model, "model-a") + self.assertIsNotNone(question.embedded_at) + + def test_a_failed_embedding_leaves_no_stale_provenance(self): + from app.services import embedding_service + + question = self.db.get(Question, 1) + with patch.object(embedding_service, "_get_embedding_model", return_value="model-a"), \ + patch.object(embedding_service, "generate_embedding", return_value=None): + self.assertFalse(embedding_service.embed_question(question)) + self.assertIsNone(question.embedding_model) + + def test_health_separates_missing_from_stale(self): + from app.services import embedding_service + + # One current, one from a retired model, one never embedded. + self.db.query(Question).filter(Question.id == 1).update( + {"embedding": [0.1] * 1024, "embedding_model": "model-a"}) + self.db.query(Question).filter(Question.id == 2).update( + {"embedding": [0.2] * 1024, "embedding_model": "model-old"}) + self.db.commit() + + with patch.object(embedding_service, "_get_embedding_model", return_value="model-a"): + health = embedding_service.stale_embedding_counts(self.db) + self.assertEqual(health["active_model"], "model-a") + self.assertEqual((health["current"], health["stale"], health["missing"]), (1, 1, 1)) + self.assertTrue(health["needs_regeneration"]) + + # Switching the model makes every stored vector stale, and says so. + with patch.object(embedding_service, "_get_embedding_model", return_value="model-b"): + after = embedding_service.stale_embedding_counts(self.db) + self.assertEqual((after["current"], after["stale"]), (0, 2)) + + +class ProvenanceMatchesGeneratorTests(unittest.TestCase): + """The stamp must name the model that actually produced the vector.""" + + def test_generator_and_stamp_resolve_the_same_model(self): + from app.config import settings + from app.services import embedding_service + + calls = {} + + def fake_post(url, **kwargs): + calls["model"] = kwargs["content"] if "content" in kwargs else kwargs.get("json") + raise RuntimeError("stop after capturing the request") + + with patch.object(settings, "LITELLM_EMBEDDING_MODEL", "from-env"), \ + patch.object(settings, "LITELLM_API_KEY", "k"), \ + patch.object(settings, "LITELLM_API_BASE", "https://proxy.test"), \ + patch.object(embedding_service, "_get_embedding_model", return_value="from-resolver") as resolver, \ + patch("httpx.post", side_effect=fake_post): + embedding_service.generate_embedding("some question text") + + # The request must carry the resolver's model, not the raw env value: + # otherwise embed_question would stamp a different name than it embedded with. + self.assertIn("from-resolver", str(calls.get("model"))) + self.assertNotIn("from-env", str(calls.get("model"))) + self.assertTrue(resolver.called) + + +if __name__ == "__main__": + unittest.main() diff --git a/docker-compose.yml b/docker-compose.yml index 9f7cc3d..220e132 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -47,7 +47,8 @@ services: celery: build: ./backend - command: celery -A app.tasks worker --loglevel=${LOG_LEVEL:-info} --concurrency=2 + # --beat runs the embedded scheduler (single worker, so no lock needed). + command: celery -A app.tasks worker --beat --loglevel=${LOG_LEVEL:-info} --concurrency=2 env_file: - ./backend/.env environment: diff --git a/frontend/src/pages/QuestionBankPage.jsx b/frontend/src/pages/QuestionBankPage.jsx index cc0475c..cd456c2 100644 --- a/frontend/src/pages/QuestionBankPage.jsx +++ b/frontend/src/pages/QuestionBankPage.jsx @@ -280,7 +280,6 @@ export default function QuestionBankPage() { api.get('/articles/').then(res => setArticles(res.data || [])).catch(() => setArticles([])) api.get('/collections/').then(res => setCollections(res.data || [])).catch(() => setCollections([])) }, []) - const [searchMode, setSearchMode] = useState('hybrid') const [filterCatIds, setFilterCatIds] = useState([]) const [showUncategorized, setShowUncategorized] = useState(false) const [showFavorites, setShowFavorites] = useState(false) @@ -323,10 +322,10 @@ export default function QuestionBankPage() { loadTags() }, []) - const loadQuestions = async (query = searchQuery, off = 0, catIds = filterCatIds, uncatOnly = showUncategorized, favOnly = showFavorites, mode = searchMode, size = pageSize, tagIds = selectedTagIds) => { + const loadQuestions = async (query = searchQuery, off = 0, catIds = filterCatIds, uncatOnly = showUncategorized, favOnly = showFavorites, size = pageSize, tagIds = selectedTagIds) => { setLoading(true) try { - const params = { limit: size === 'all' ? 5000 : size, offset: off, search_mode: mode } + const params = { limit: size === 'all' ? 5000 : size, offset: off } if (query.trim()) params.q = query.trim() if (catIds.length > 0) params.category_ids = catIds.join(',') if (uncatOnly) params.uncategorized = true @@ -344,9 +343,9 @@ export default function QuestionBankPage() { const catIdsKey = filterCatIds.join(',') useEffect(() => { clearTimeout(debounceRef.current) - debounceRef.current = setTimeout(() => loadQuestions(searchQuery, 0, filterCatIds, showUncategorized, showFavorites, searchMode, pageSize, selectedTagIds), 300) + debounceRef.current = setTimeout(() => loadQuestions(searchQuery, 0, filterCatIds, showUncategorized, showFavorites, pageSize, selectedTagIds), 300) return () => clearTimeout(debounceRef.current) - }, [searchQuery, catIdsKey, showUncategorized, showFavorites, showMyQuestions, searchMode, pageSize, tagIdsKey, difficulty, bankArticleIds.join(',')]) + }, [searchQuery, catIdsKey, showUncategorized, showFavorites, showMyQuestions, pageSize, tagIdsKey, difficulty, bankArticleIds.join(',')]) const toggleTag = (tagId) => { setSelectedTagIds(prev => { @@ -772,16 +771,6 @@ export default function QuestionBankPage() { setSearchQuery(e.target.value)} style={{ width: '100%', padding: '9px 13px 9px 36px', border: '1px solid var(--border)', borderRadius: 8, fontSize: '0.9rem', background: 'var(--input-bg)', color: 'var(--text)' }} /> - {/* Search mode — clearly labelled separately from category chips */} -
- Search: - -
{/* Page size */}
Show: @@ -905,7 +894,7 @@ export default function QuestionBankPage() { {questions.length < total && (
-
diff --git a/frontend/src/pages/QuestionManagerPage.jsx b/frontend/src/pages/QuestionManagerPage.jsx index 39de1d9..a2338f3 100644 --- a/frontend/src/pages/QuestionManagerPage.jsx +++ b/frontend/src/pages/QuestionManagerPage.jsx @@ -66,7 +66,7 @@ export default function QuestionManagerPage() { const load = useCallback(() => { setLoading(true) setError('') - const params = { limit: PAGE_SIZE, offset: page * PAGE_SIZE, search_mode: 'keyword' } + const params = { limit: PAGE_SIZE, offset: page * PAGE_SIZE } if (debouncedSearch) params.q = debouncedSearch if (needs) params.needs = needs if (categoryId) params.category_id = Number(categoryId)