diff --git a/README.md b/README.md index 450d2cc71ee9819f16e2190380e0af56badf549f..d386d44e40069866374ad0cfc6a6e939b313b66c 100644 --- a/README.md +++ b/README.md @@ -1,13 +1,24 @@ --- -title: Vera Api -emoji: πŸš€ -colorFrom: purple -colorTo: purple +title: VERA API +colorFrom: blue +colorTo: gray sdk: gradio -sdk_version: 6.27.0 -python_version: '3.12' +sdk_version: 5.49.1 +python_version: "3.12" app_file: app.py +suggested_hardware: zero-a10g pinned: false +short_description: VERA earbud review analysis API (FastAPI + Celery worker) --- -Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference +# VERA API (Hugging Face Space) + +Backend for the VERA review analyzer. `app.py` starts loopback Redis, applies database +migrations, starts one Celery worker (which downloads the models from the private +`HF_MODEL_REPO`), and serves the FastAPI app on port 7860. + +Endpoints: `GET /health`, `GET /ready`, `POST /api/reviews`, `GET /api/reviews/{id}`, +`GET /api/reviews/{id}/status`, `GET /api/products`, `GET /api/products/{asin}`. + +Secrets (Settings -> Variables and secrets): `DATABASE_URL` (Supabase session pooler, +`?sslmode=require`), `HF_MODEL_REPO`, `HF_TOKEN` (read), `SECRET_KEY`, `CORS_ORIGINS`. diff --git a/app.py b/app.py new file mode 100644 index 0000000000000000000000000000000000000000..c74f9ded805a89394cb8326c313d481f0449d897 --- /dev/null +++ b/app.py @@ -0,0 +1,124 @@ +"""VERA API + worker launcher for a free Hugging Face Gradio Space on ZeroGPU hardware. + +Why this shape: Docker Spaces and CPU-basic Gradio Spaces require a PRO plan, while a Gradio +Space on ZeroGPU hardware is free. ZeroGPU stops any app that never reports startup, and that +report is normally sent by ``gradio.Blocks.launch``. This launcher instead serves the existing +FastAPI app with uvicorn and sends the report itself (``spaces.zero.startup``). All inference +runs on CPU with onnxruntime; the GPU is never requested, so the ZeroGPU quota is not used. + +One container owns: loopback Redis (Celery broker), Alembic migrations, one solo Celery worker +(which downloads the models from ``HF_MODEL_REPO`` and runs the outbox dispatcher), and the API +on port 7860. If Redis or the worker exits, the process exits so the Space restarts. + +Required Space secrets: DATABASE_URL, HF_MODEL_REPO, HF_TOKEN (read), SECRET_KEY, CORS_ORIGINS. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +import threading +import time +from pathlib import Path + +import spaces # must be imported before any CUDA-related package + +ROOT = Path(__file__).resolve().parent +BACKEND = ROOT / "backend" +MODEL_DIR = Path(os.environ.get("MODEL_DIR", "/home/user/models")) +REDIS_URL = "redis://127.0.0.1:6379/0" + +for name in ("DATABASE_URL", "HF_MODEL_REPO"): + if not os.environ.get(name): + sys.exit(f"Set {name} in the Space secrets") + +os.environ.update( + ENVIRONMENT="production", + MODEL_DIR=str(MODEL_DIR), + REDIS_URL=REDIS_URL, + CELERY_BROKER_URL=REDIS_URL, + CELERY_RESULT_BACKEND="", +) +MODEL_DIR.mkdir(parents=True, exist_ok=True) +sys.path.insert(0, str(BACKEND)) +os.chdir(BACKEND) + + +@spaces.GPU +def _zerogpu_placeholder() -> str: + """ZeroGPU only reports startup when at least one GPU function exists. Never called.""" + return "unused" + + +def _log(message: str) -> None: + print(f"[vera-launcher] {message}", flush=True) + + +def start_redis() -> subprocess.Popen[bytes]: + Path("/tmp/redis").mkdir(exist_ok=True) + proc = subprocess.Popen( + ["redis-server", "--bind", "127.0.0.1", "--protected-mode", "yes", "--port", "6379", + "--dir", "/tmp/redis", "--save", "", "--appendonly", "no"] + ) + for _ in range(30): + ping = subprocess.run(["redis-cli", "-h", "127.0.0.1", "ping"], capture_output=True) + if ping.stdout.strip() == b"PONG": + _log("redis ready") + return proc + if proc.poll() is not None: + break + time.sleep(1) + sys.exit("redis-server failed to start") + + +def run_migrations() -> None: + subprocess.run([sys.executable, "-m", "alembic", "upgrade", "head"], check=True) + _log("database migrations applied") + + +def start_worker() -> subprocess.Popen[bytes]: + return subprocess.Popen( + [sys.executable, "-m", "celery", "-A", "app.worker.celery_app", "worker", + "--pool=solo", "--concurrency=1", "--loglevel=info"] + ) + + +def supervise(children: dict[str, subprocess.Popen[bytes]]) -> None: + """Exit the whole Space if a required child dies, so the platform restarts it.""" + while True: + for name, proc in children.items(): + if proc.poll() is not None: + _log(f"{name} exited with code {proc.returncode}; stopping the Space") + for other in children.values(): + if other.poll() is None: + other.terminate() + os._exit(1) + time.sleep(5) + + +def main() -> None: + import uvicorn + + redis_proc = start_redis() + run_migrations() + worker = start_worker() + threading.Thread( + target=supervise, args=({"redis": redis_proc, "worker": worker},), daemon=True + ).start() + + from app.main import app + + @app.get("/", include_in_schema=False) + def root() -> dict[str, str]: + return {"service": "VERA review analysis API", "health": "/health", "ready": "/ready"} + + zero = getattr(spaces, "zero", None) + if zero is not None and hasattr(zero, "startup"): + zero.startup() + _log("ZeroGPU startup reported") + uvicorn.run(app, host="0.0.0.0", port=7860, proxy_headers=False, access_log=False) + + +if __name__ == "__main__": + main() diff --git a/backend/alembic.ini b/backend/alembic.ini new file mode 100644 index 0000000000000000000000000000000000000000..0e9892a3e2f8b1540cd11873e251e05bf22a499c --- /dev/null +++ b/backend/alembic.ini @@ -0,0 +1,37 @@ +[alembic] +script_location = migrations +prepend_sys_path = . +sqlalchemy.url = postgresql://unused:unused@localhost/vera + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/backend/app/__init__.py b/backend/app/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..c876682d9783d48e2a4d288cf8147ca045f979f9 --- /dev/null +++ b/backend/app/__init__.py @@ -0,0 +1,3 @@ +"""VERA β€” Verified Experience & Review Analyzer (backend package).""" + +__version__ = "1.0.0" diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..f82b28cbd9510cef4faf9fbed07f2526ab8d7c59 --- /dev/null +++ b/backend/app/api/__init__.py @@ -0,0 +1 @@ +"""FastAPI routing and dependencies.""" diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py new file mode 100644 index 0000000000000000000000000000000000000000..b782e2a9ebb1814670be9d5b92645375f96de15c --- /dev/null +++ b/backend/app/api/deps.py @@ -0,0 +1,18 @@ +"""Shared FastAPI dependencies.""" + +from __future__ import annotations + +from collections.abc import Generator + +from sqlalchemy.orm import Session + +from app.database import SessionLocal + + +def get_db() -> Generator[Session, None, None]: + """Yield one request-scoped SQLAlchemy session.""" + db = SessionLocal() + try: + yield db + finally: + db.close() diff --git a/backend/app/api/logging.py b/backend/app/api/logging.py new file mode 100644 index 0000000000000000000000000000000000000000..1ee980946351f1835bad7fe58a4129c80855c104 --- /dev/null +++ b/backend/app/api/logging.py @@ -0,0 +1,50 @@ +"""Structured API logging with request correlation and bounded, safe access fields.""" + +from __future__ import annotations + +import json +import logging +from contextvars import ContextVar +from datetime import UTC, datetime +from logging.config import dictConfig + +request_id_context: ContextVar[str] = ContextVar("request_id", default="-") + + +class JsonFormatter(logging.Formatter): + """Serialize log records without exception bodies, query strings, or request payloads.""" + + def format(self, record: logging.LogRecord) -> str: + payload = { + "timestamp": datetime.fromtimestamp(record.created, UTC).isoformat(), + "level": record.levelname, + "logger": record.name, + "message": record.getMessage(), + "request_id": request_id_context.get(), + } + for field in ("method", "route", "status_code", "duration_ms", "exception_type"): + if hasattr(record, field): + payload[field] = getattr(record, field) + if record.exc_info and record.exc_info[0]: + payload["exception_type"] = record.exc_info[0].__name__ + return json.dumps(payload, ensure_ascii=True, default=str) + + +def configure_json_logging() -> None: + """Configure the API's console logs; access logging is owned by the middleware.""" + dictConfig( + { + "version": 1, + "disable_existing_loggers": False, + "formatters": {"json": {"()": JsonFormatter}}, + "handlers": { + "console": {"class": "logging.StreamHandler", "formatter": "json"}, + "null": {"class": "logging.NullHandler"}, + }, + "root": {"level": "INFO", "handlers": ["console"]}, + "loggers": { + "uvicorn": {"handlers": ["console"], "propagate": False}, + "uvicorn.access": {"handlers": ["null"], "propagate": False}, + }, + } + ) diff --git a/backend/app/api/middleware.py b/backend/app/api/middleware.py new file mode 100644 index 0000000000000000000000000000000000000000..5ad2555ec011759e092a665230ab9dcb5536f9ff --- /dev/null +++ b/backend/app/api/middleware.py @@ -0,0 +1,192 @@ +"""Bounded ASGI request handling and process-local submission throttling.""" + +from __future__ import annotations + +import logging +import math +import re +import time +from collections import OrderedDict, deque +from ipaddress import ip_address, ip_network +from threading import Lock +from uuid import uuid4 + +from starlette.datastructures import Headers, MutableHeaders +from starlette.responses import JSONResponse +from starlette.types import ASGIApp, Message, Receive, Scope, Send + +from app.api.logging import request_id_context +from app.config import Settings + +logger = logging.getLogger("vera.requests") +REQUEST_ID_PATTERN = re.compile(r"[A-Za-z0-9._-]{1,64}\Z") + + +class SlidingWindowLimiter: + """Bound memory and requests per client; restart/multiple-process limits are documented.""" + + def __init__(self, limit: int, window: int, max_clients: int) -> None: + self.limit, self.window, self.max_clients = limit, window, max_clients + self.clients: OrderedDict[str, deque[float]] = OrderedDict() + self.lock = Lock() + + def check(self, client: str, now: float | None = None) -> int: + """Return zero when allowed, otherwise seconds until a retry may be accepted.""" + now = time.monotonic() if now is None else now + with self.lock: + while self.clients and next(iter(self.clients.values()))[-1] <= now - self.window: + self.clients.popitem(last=False) + bucket = self.clients.get(client) + if bucket is None: + if len(self.clients) >= self.max_clients: + return self.window + bucket = self.clients[client] = deque() + while bucket and bucket[0] <= now - self.window: + bucket.popleft() + if len(bucket) >= self.limit: + return max(1, math.ceil(bucket[0] + self.window - now)) + bucket.append(now) + self.clients.move_to_end(client) + return 0 + + +class RequestGuard: + """Apply correlation IDs, body bounds, security headers, and POST rate limits.""" + + def __init__(self, app: ASGIApp, settings: Settings) -> None: + self.app = app + self.settings = settings + self.limiter = SlidingWindowLimiter( + settings.REVIEW_RATE_LIMIT, + settings.REVIEW_RATE_WINDOW_SECONDS, + settings.RATE_LIMIT_MAX_CLIENTS, + ) + self.proxies = [ + ip_network(value.strip(), strict=False) + for value in settings.TRUSTED_PROXY_CIDRS.split(",") + if value.strip() + ] + + def client_key(self, scope: Scope, headers: Headers) -> str: + """Strip only explicitly trusted proxy hops from the right of a forwarded chain.""" + peer = (scope.get("client") or ("unknown", 0))[0] + try: + address = ip_address(peer) + if not any(address in network for network in self.proxies): + return str(address) + hops = headers.get("x-forwarded-for", "").split(",") + if len(hops) > 20: + return peer + for hop in reversed(hops): + address = ip_address(hop.strip()) + if not any(address in network for network in self.proxies): + return str(address) + return str(address) + except ValueError: + return peer + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http": + await self.app(scope, receive, send) + return + headers = Headers(scope=scope) + supplied = headers.get("x-request-id", "") + request_id = supplied if REQUEST_ID_PATTERN.fullmatch(supplied) else uuid4().hex + token = request_id_context.set(request_id) + scope.setdefault("state", {})["request_id"] = request_id + started = time.monotonic() + status_code = 500 + response_started = False + + async def guarded_send(message: Message) -> None: + nonlocal status_code, response_started + if message["type"] == "http.response.start": + response_started = True + status_code = message["status"] + outgoing = MutableHeaders(scope=message) + outgoing["X-Request-ID"] = request_id + outgoing["X-Content-Type-Options"] = "nosniff" + outgoing["X-Frame-Options"] = "DENY" + outgoing["Referrer-Policy"] = "no-referrer" + outgoing["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()" + outgoing["Cache-Control"] = "no-store" + origin = headers.get("origin") + if origin and ( + origin in self.settings.cors_origins or "*" in self.settings.cors_origins + ): + outgoing["Access-Control-Allow-Origin"] = origin + outgoing["Access-Control-Expose-Headers"] = ( + "X-Total-Count, X-Request-ID, Retry-After" + ) + outgoing.add_vary_header("Origin") + if self.settings.is_production: + outgoing["Strict-Transport-Security"] = "max-age=31536000" + await send(message) + + async def reject(code: int, detail: str, extra: dict[str, str] | None = None) -> None: + await JSONResponse( + {"detail": detail, "request_id": request_id}, status_code=code, headers=extra + )(scope, receive, guarded_send) + + try: + length = headers.get("content-length") + if length is not None: + if not length.isascii() or not length.isdigit() or len(length) > 10: + await reject(400, "Invalid Content-Length header.") + return + if int(length) > self.settings.MAX_REQUEST_BODY_BYTES: + await reject(413, "Request body is too large.") + return + if headers.get("content-encoding", "identity").lower() != "identity": + await reject(415, "Compressed request bodies are not supported.") + return + if scope["method"] == "POST" and scope["path"].rstrip("/") == "/api/reviews": + retry = self.limiter.check(self.client_key(scope, headers)) + if retry: + await reject( + 429, + "Too many reviews submitted. Please wait before trying again.", + {"Retry-After": str(retry)}, + ) + return + # Buffer at most the configured limit before any handler can persist side effects. + body = bytearray() + while True: + message = await receive() + if message["type"] == "http.disconnect": + status_code = 499 + return + chunk = message.get("body", b"") + if len(body) + len(chunk) > self.settings.MAX_REQUEST_BODY_BYTES: + await reject(413, "Request body is too large.") + return + body.extend(chunk) + if not message.get("more_body", False): + break + delivered = False + + async def replay() -> Message: + nonlocal delivered + if not delivered: + delivered = True + return {"type": "http.request", "body": bytes(body), "more_body": False} + return await receive() + + await self.app(scope, replay, guarded_send) + except Exception as exc: + logger.error("request_failed", extra={"exception_type": type(exc).__name__}) + if response_started: + raise + await reject(500, "The service could not complete this request.") + finally: + route = getattr(scope.get("route"), "path", "unmatched") + logger.info( + "request_completed", + extra={ + "method": scope["method"], + "route": route, + "status_code": status_code, + "duration_ms": round((time.monotonic() - started) * 1000, 2), + }, + ) + request_id_context.reset(token) diff --git a/backend/app/api/routes/__init__.py b/backend/app/api/routes/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..cd2194dfc97b6a69165f2d74a5bd9893374c2131 --- /dev/null +++ b/backend/app/api/routes/__init__.py @@ -0,0 +1 @@ +"""HTTP route modules.""" diff --git a/backend/app/api/routes/health.py b/backend/app/api/routes/health.py new file mode 100644 index 0000000000000000000000000000000000000000..0a3e4303d060135f42ad0d9798b6282e6a243299 --- /dev/null +++ b/backend/app/api/routes/health.py @@ -0,0 +1,58 @@ +"""Liveness and diagnostic health endpoint.""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter +from fastapi.responses import JSONResponse +from sqlalchemy import text + +from app.config import settings +from app.database import SessionLocal +from app.ml.models.loader import artifact_status + +router = APIRouter(tags=["health"]) + + +def database_ready() -> bool: + """Check database reachability without loading models or publishing queue work.""" + try: + with SessionLocal() as db: + db.execute(text("SELECT 1")) + return True + except Exception: + return False + + +@router.get("/ready") +def ready() -> JSONResponse: + """Readiness is database reachability, not a guarantee of worker/model readiness.""" + available = database_ready() + return JSONResponse( + {"status": "ready" if available else "unavailable", "database": available}, + status_code=200 if available else 503, + headers={} if available else {"Retry-After": "10"}, + ) + + +@router.get("/health") +def health() -> dict[str, Any]: + """Return 200 for liveness, with explicit dependency diagnostics. + + The API itself does not load ML artifacts: workers own inference and fail + closed on missing artifacts. Keeping liveness independent means uptime + monitors can report a useful degraded state during a model rollout. + """ + database_ok = database_ready() + try: + models = artifact_status(settings.MODEL_DIR) + except OSError: + models = {"accessible": False} + ready = database_ok and all(models.values()) + return { + "status": "healthy" if ready else "degraded", + "database": database_ok, + "model_artifacts": models, + "environment": settings.ENVIRONMENT, + } diff --git a/backend/app/api/routes/products.py b/backend/app/api/routes/products.py new file mode 100644 index 0000000000000000000000000000000000000000..c5e21616b1c5c2f9b1957ebb9545809995dea361 --- /dev/null +++ b/backend/app/api/routes/products.py @@ -0,0 +1,80 @@ +"""Product ranking and detail routes (PRD Β§14.2, Β§16.4).""" + +from __future__ import annotations + +from typing import Any, Literal + +from fastapi import APIRouter, Depends, HTTPException, Query, Response, status +from sqlalchemy import asc, desc, func, nulls_last +from sqlalchemy.orm import Session + +from app.api.deps import get_db +from app.models.product import Product +from app.models.review import Review, ReviewStatus +from app.schemas.product import ProductDetailResponse, ProductResponse +from app.schemas.review import ReviewResponse + +router = APIRouter(prefix="/api/products", tags=["products"]) + +SortBy = Literal["corrected_score", "star_average", "review_count", "uncertainty", "rank_position"] +SortOrder = Literal["asc", "desc"] + + +@router.get("", response_model=list[ProductResponse]) +def list_products( + response: Response, + sort_by: SortBy = "corrected_score", + order: SortOrder = "desc", + limit: int = Query(default=50, ge=1, le=100), + offset: int = Query(default=0, ge=0), + db: Session = Depends(get_db), +) -> list[Product]: + """Paginated ranking of scored products. + + Product shells without a completed review are excluded. Ties break on ASIN + so pagination is stable. ``X-Total-Count`` carries the unpaginated total. + """ + column = getattr(Product, sort_by) + direction: Any = desc(column) if order == "desc" else asc(column) + query = db.query(Product).filter(Product.corrected_score.is_not(None)) + response.headers["X-Total-Count"] = str(query.count()) + return ( + query.order_by(nulls_last(direction), asc(Product.asin)).offset(offset).limit(limit).all() + ) + + +@router.get("/{asin}", response_model=ProductDetailResponse) +def get_product( + asin: str, + reviews_limit: int = Query(default=20, ge=1, le=100), + reviews_offset: int = Query(default=0, ge=0), + db: Session = Depends(get_db), +) -> ProductDetailResponse: + """Product aggregate plus a newest-first page of its completed reviews.""" + product = db.get(Product, asin) + if product is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found") + + base = db.query(Review).filter( + Review.product_asin == asin, Review.status == ReviewStatus.DONE.value + ) + total = base.with_entities(func.count(Review.id)).scalar() or 0 + reviews = ( + base.order_by(desc(Review.created_at), desc(Review.id)) + .offset(reviews_offset) + .limit(reviews_limit) + .all() + ) + return ProductDetailResponse.model_validate( + { + **ProductResponse.model_validate(product).model_dump(), + "eb_shrinkage_factor": product.eb_shrinkage_factor, + "aspect_summary": product.aspect_summary, + "last_aggregated_at": product.last_aggregated_at, + "reviews": [ReviewResponse.model_validate(r) for r in reviews], + "reviews_total": total, + "reviews_limit": reviews_limit, + "reviews_offset": reviews_offset, + "reviews_has_more": reviews_offset + len(reviews) < total, + } + ) diff --git a/backend/app/api/routes/reviews.py b/backend/app/api/routes/reviews.py new file mode 100644 index 0000000000000000000000000000000000000000..d339632e2ca2a562e2145ee8f51fc5690375001f --- /dev/null +++ b/backend/app/api/routes/reviews.py @@ -0,0 +1,115 @@ +"""Review submission and polling routes (PRD Β§14.1–14.2).""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime + +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app.api.deps import get_db +from app.models.job import Job, JobKind +from app.models.product import Product +from app.models.review import Review, ReviewStatus +from app.schemas.review import ReviewCreate, ReviewResponse, ReviewTaskStatus +from app.worker.outbox import send_wakeup + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/reviews", tags=["reviews"]) + +STATUS_MESSAGES: dict[str, str] = { + ReviewStatus.PENDING.value: "Review is queued for processing", + ReviewStatus.PROCESSING.value: "Review is being processed", + ReviewStatus.DONE.value: "Review processing completed", + ReviewStatus.FAILED.value: "Review processing failed", +} + + +def _get_review_or_404(db: Session, review_id: int) -> Review: + review = db.get(Review, review_id) + if review is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Review not found") + return review + + +def _ensure_product_shell(db: Session, asin: str, title: str | None) -> None: + """Create or update the product row without losing concurrent first reviews.""" + product = db.get(Product, asin) + if product is None: + try: + # The savepoint lets a concurrent ASIN insert fail locally while + # preserving the review/job transaction that follows it. + with db.begin_nested(): + db.add(Product(asin=asin, title=title)) + db.flush() + except IntegrityError: + product = db.get(Product, asin) + if product is None: + raise + if product is not None and title and not product.title: + product.title = title + + +@router.post("", response_model=ReviewTaskStatus, status_code=status.HTTP_202_ACCEPTED) +def submit_review(payload: ReviewCreate, db: Session = Depends(get_db)) -> ReviewTaskStatus: + """Persist a review and its processing job atomically; return for polling. + + The product shell is created first because ``reviews.product_asin`` is a + foreign key (the PRD's schema made every first review of an ASIN fail). + After commit a wake-up is sent to Redis; if Redis is down the job is still + durable and the outbox dispatcher delivers it later, so this never 503s. + """ + asin = payload.product_asin + if asin: + _ensure_product_shell(db, asin, payload.product_title) + + review = Review( + text=payload.text, + star_rating=payload.star_rating, + helpful_votes=payload.helpful_votes, + verified_purchase=payload.verified_purchase, + product_title=payload.product_title, + product_asin=asin, + status=ReviewStatus.PENDING.value, + ) + db.add(review) + db.flush() # assigns review.id + job = Job(kind=JobKind.REVIEW.value, review_id=review.id) + db.add(job) + db.commit() + + try: + send_wakeup(job.id) + job.last_dispatched_at = datetime.now(UTC) + db.commit() + except Exception: + db.rollback() + logger.warning("Broker unavailable for job %s; outbox will retry", job.id, exc_info=True) + + return ReviewTaskStatus( + review_id=review.id, + task_id=job.id, + status=ReviewStatus.PENDING.value, + message="Review queued for processing", + ) + + +@router.get("/{review_id}", response_model=ReviewResponse) +def get_review(review_id: int, db: Session = Depends(get_db)) -> Review: + """Get a review and, when available, its pipeline output.""" + return _get_review_or_404(db, review_id) + + +@router.get("/{review_id}/status", response_model=ReviewTaskStatus) +def get_review_status(review_id: int, db: Session = Depends(get_db)) -> ReviewTaskStatus: + """Lightweight lifecycle state for polling clients.""" + review = _get_review_or_404(db, review_id) + job_id = db.query(Job.id).filter(Job.review_id == review.id).scalar() + return ReviewTaskStatus( + review_id=review.id, + task_id=job_id, + status=review.status, + message=STATUS_MESSAGES.get(review.status, "Unknown status"), + ) diff --git a/backend/app/config.py b/backend/app/config.py new file mode 100644 index 0000000000000000000000000000000000000000..5686de87de7080db2ab0854344e8c02979243e6a --- /dev/null +++ b/backend/app/config.py @@ -0,0 +1,134 @@ +"""Application settings loaded from environment variables / the repo-root `.env`. + +Every setting is documented in `.env.example`. Settings are read once and +cached via :func:`get_settings`; import the module-level ``settings`` object for +convenience. +""" + +from __future__ import annotations + +from functools import lru_cache +from ipaddress import ip_network +from pathlib import Path +from typing import Literal +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +from pydantic import Field, field_validator +from pydantic_settings import BaseSettings, SettingsConfigDict + +#: Repository root (``backend/app/config.py`` -> two parents up from ``app``). +REPO_ROOT = Path(__file__).resolve().parents[2] + + +def _normalize_database_url(url: str) -> str: + """Force the psycopg2 driver for Postgres URLs. + + Supabase/Heroku-style ``postgres://`` URLs are rejected by SQLAlchemy 2.x, + so both ``postgres://`` and bare ``postgresql://`` are rewritten to + ``postgresql+psycopg2://``. Other schemes (e.g. ``sqlite://``) pass through. + """ + for prefix in ("postgres://", "postgresql://"): + if url.startswith(prefix): + return "postgresql+psycopg2://" + url[len(prefix) :] + return url + + +def _normalize_redis_url(url: str) -> str: + """Ensure TLS Redis URLs carry ``ssl_cert_reqs``. + + Celery refuses ``rediss://`` broker URLs without an explicit + ``ssl_cert_reqs`` query parameter. Upstash presents a valid public + certificate, so ``required`` (full verification) is the safe default. + """ + if not url.startswith("rediss://"): + return url + parts = urlsplit(url) + query = dict(parse_qsl(parts.query)) + query.setdefault("ssl_cert_reqs", "required") + return urlunsplit(parts._replace(query=urlencode(query))) + + +class Settings(BaseSettings): + """Typed runtime configuration.""" + + model_config = SettingsConfigDict( + env_file=REPO_ROOT / ".env", + env_file_encoding="utf-8", + extra="ignore", + case_sensitive=True, + ) + + # --- Database / queue --------------------------------------------------- + DATABASE_URL: str = Field(..., description="SQLAlchemy URL (Supabase or local Postgres).") + REDIS_URL: str = Field(..., description="Redis URL (Upstash rediss:// or local redis://).") + CELERY_BROKER_URL: str = "" + CELERY_RESULT_BACKEND: str = "" + + # --- App ------------------------------------------------------------------ + ENVIRONMENT: Literal["development", "test", "production"] = "development" + SECRET_KEY: str = "" + CORS_ORIGINS: str = "http://localhost:3000" + REVIEW_RATE_LIMIT: int = Field(default=10, ge=1, le=10000) + REVIEW_RATE_WINDOW_SECONDS: int = Field(default=60, ge=1, le=86400) + RATE_LIMIT_MAX_CLIENTS: int = Field(default=10000, ge=1, le=100000) + MAX_REQUEST_BODY_BYTES: int = Field(default=65536, ge=1024, le=1048576) + TRUSTED_PROXY_CIDRS: str = "" + + @field_validator("TRUSTED_PROXY_CIDRS") + @classmethod + def _trusted_proxies(cls, value: str) -> str: + for network in value.split(","): + if network.strip(): + ip_network(network.strip(), strict=False) + return value.strip() + + # --- Models ----------------------------------------------------------------- + MODEL_DIR: Path = Path("data/models") + HF_MODEL_REPO: str = "" + HF_TOKEN: str = "" + + @field_validator("DATABASE_URL") + @classmethod + def _db_url(cls, value: str) -> str: + return _normalize_database_url(value.strip()) + + @field_validator("REDIS_URL", "CELERY_BROKER_URL", "CELERY_RESULT_BACKEND") + @classmethod + def _redis_url(cls, value: str) -> str: + return _normalize_redis_url(value.strip()) + + @field_validator("MODEL_DIR") + @classmethod + def _model_dir(cls, value: Path) -> Path: + # Relative paths are anchored at the repo root so the API (run from + # `backend/`) and notebooks (run from the root) resolve the same folder. + return value if value.is_absolute() else (REPO_ROOT / value).resolve() + + # --- Derived -------------------------------------------------------------- + @property + def broker_url(self) -> str: + """Celery broker; falls back to ``REDIS_URL``.""" + return self.CELERY_BROKER_URL or self.REDIS_URL + + @property + def result_backend(self) -> str: + """Celery result backend; falls back to ``REDIS_URL``.""" + return self.CELERY_RESULT_BACKEND or self.REDIS_URL + + @property + def cors_origins(self) -> list[str]: + """``CORS_ORIGINS`` split on commas, blanks removed.""" + return [origin.strip() for origin in self.CORS_ORIGINS.split(",") if origin.strip()] + + @property + def is_production(self) -> bool: + return self.ENVIRONMENT == "production" + + +@lru_cache(maxsize=1) +def get_settings() -> Settings: + """Return the cached settings instance.""" + return Settings() # values come from env + + +settings = get_settings() diff --git a/backend/app/database.py b/backend/app/database.py new file mode 100644 index 0000000000000000000000000000000000000000..5105e747c0439c75fc47b537d4a34feab3ddcf5b --- /dev/null +++ b/backend/app/database.py @@ -0,0 +1,60 @@ +"""SQLAlchemy engine, session factory, and declarative base.""" + +from __future__ import annotations + +from collections.abc import Iterator +from contextlib import contextmanager + +from sqlalchemy import JSON, create_engine +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.engine import Engine +from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker +from sqlalchemy.pool import StaticPool + +from app.config import settings + +#: JSON column type: native JSONB on Postgres, plain JSON elsewhere (SQLite tests/CI). +JSONType = JSON().with_variant(JSONB(), "postgresql") + + +class Base(DeclarativeBase): + """Declarative base for all ORM models.""" + + +def build_engine(url: str) -> Engine: + """Create an engine with settings appropriate for the URL's dialect.""" + if url.startswith("sqlite"): + # In-memory SQLite needs a single shared connection across threads. + in_memory = ":memory:" in url or url.rstrip("/") == "sqlite:" + return create_engine( + url, + connect_args={"check_same_thread": False}, + poolclass=StaticPool if in_memory else None, + ) + return create_engine( + url, + pool_pre_ping=True, # survive Supabase/pooler idle disconnects + pool_size=5, + max_overflow=5, + pool_recycle=1800, + ) + + +engine = build_engine(settings.DATABASE_URL) +SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False) + + +@contextmanager +def session_scope() -> Iterator[Session]: + """Context-managed session for non-request code (Celery tasks, scripts). + + Rolls back on any exception and always closes the session. + """ + db = SessionLocal() + try: + yield db + except Exception: + db.rollback() + raise + finally: + db.close() diff --git a/backend/app/main.py b/backend/app/main.py new file mode 100644 index 0000000000000000000000000000000000000000..36b39e48273ec1ab195dbb8d2e2a048367d3611f --- /dev/null +++ b/backend/app/main.py @@ -0,0 +1,33 @@ +"""FastAPI application entry point.""" + +from __future__ import annotations + +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware + +from app.api.logging import configure_json_logging +from app.api.middleware import RequestGuard +from app.api.routes import health, products, reviews +from app.config import settings + + +def create_app() -> FastAPI: + """Construct the HTTP API without loading worker-owned ML models.""" + app = FastAPI(title="VERA API", version="1.0.0") + configure_json_logging() + app.add_middleware( + CORSMiddleware, + allow_origins=settings.cors_origins, + allow_credentials=False, + allow_methods=["GET", "POST"], + allow_headers=["Content-Type", "Authorization", "X-Request-ID"], + expose_headers=["X-Total-Count", "X-Request-ID", "Retry-After"], + ) + app.add_middleware(RequestGuard, settings=settings) + app.include_router(health.router) + app.include_router(reviews.router) + app.include_router(products.router) + return app + + +app = create_app() diff --git a/backend/app/ml/__init__.py b/backend/app/ml/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..1d5aba5184668adebdbcdf25832ab1e46e04ec2b --- /dev/null +++ b/backend/app/ml/__init__.py @@ -0,0 +1 @@ +"""VERA ML pipeline: seven signal agents, fusion, and calibration (PRD Β§8–§11).""" diff --git a/backend/app/ml/agents/__init__.py b/backend/app/ml/agents/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..a9f5fae272d7ed6dc8b9ef8c8a52168b61828a05 --- /dev/null +++ b/backend/app/ml/agents/__init__.py @@ -0,0 +1,10 @@ +"""The seven VERA signal agents (PRD Β§9). + +1. :class:`~app.ml.agents.preprocessor.TextPreprocessor` +2. :class:`~app.ml.agents.experience_extractor.ExperienceExtractor` +3. :class:`~app.ml.agents.absa.ABSAModule` +4. :class:`~app.ml.agents.consistency.ConsistencyScorer` +5. :class:`~app.ml.agents.quality.QualityAssessor` +6. :class:`~app.ml.agents.domain_rating.DomainRatingModel` +7. :class:`~app.ml.agents.fusion.FusionModel` +""" diff --git a/backend/app/ml/agents/absa.py b/backend/app/ml/agents/absa.py new file mode 100644 index 0000000000000000000000000000000000000000..fa9b48227405273be1962e0599c8935ef113fa38 --- /dev/null +++ b/backend/app/ml/agents/absa.py @@ -0,0 +1,135 @@ +"""Agent 3 β€” Aspect-Based Sentiment Analysis (PRD Β§9, Agent 3). + +Uses ``yangheng/deberta-v3-base-absa-v1.1`` exported to ONNX, unchanged (no +fine-tuning). The learned :class:`~app.ml.agents.aspect_detector.AspectDetector` +decides which of the 10 earbud aspects are mentioned (keyword matching is the +fallback when no detector is loaded); the model then classifies sentiment for +those aspects in ONE batched inference call. + +Input format: ``[CLS] review_text [SEP] aspect [SEP]`` β€” review FIRST, aspect +SECOND (PRD weakness #4). Only the review is ever truncated. +""" + +from __future__ import annotations + +import json +import re +from collections.abc import Mapping +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import numpy as np + +from app.ml.constants import ASPECT_KEYWORDS, ASPECT_PHRASES, EARBUD_ASPECTS +from app.ml.models.onnx_session import InferenceSessionLike, build_feed, create_session, softmax +from app.ml.models.tokenizer_wrapper import LightTokenizer + +if TYPE_CHECKING: + from app.ml.agents.aspect_detector import AspectDetector + +_VALID_SENTIMENTS = frozenset({"negative", "neutral", "positive"}) + + +def _compile_keyword_patterns() -> dict[str, re.Pattern[str]]: + """One regex per aspect; keywords must start at a word boundary.""" + return { + aspect: re.compile( + r"\b(?:" + "|".join(re.escape(kw) for kw in ASPECT_KEYWORDS[aspect]) + r")", + re.IGNORECASE, + ) + for aspect in EARBUD_ASPECTS + } + + +_KEYWORD_PATTERNS = _compile_keyword_patterns() + + +def load_id2label(config_path: str | Path) -> dict[int, str]: + """Read the ABSA label mapping written by the export notebook. + + The PRD hardcoded ``{0: negative, 1: neutral, 2: positive}``; reading the + exported ``config.id2label`` removes the risk of a silent label swap. + """ + with open(config_path, encoding="utf-8") as fh: + raw = json.load(fh)["id2label"] + mapping = {int(k): str(v).strip().lower() for k, v in raw.items()} + if set(mapping.values()) != _VALID_SENTIMENTS: + raise ValueError(f"absa_config.json id2label must cover {_VALID_SENTIMENTS}, got {mapping}") + return mapping + + +def not_mentioned() -> dict[str, Any]: + """Aspect result for an aspect the review does not mention.""" + return {"detected": False, "sentiment": "not_mentioned", "confidence": 0.0} + + +class ABSAModule: + """Agent 3: per-aspect detection + sentiment.""" + + def __init__( + self, + model_path: str | Path, + tokenizer_path: str | Path, + config_path: str | Path, + *, + session: InferenceSessionLike | None = None, + tokenizer: LightTokenizer | None = None, + id2label: dict[int, str] | None = None, + detector: AspectDetector | None = None, + ) -> None: + """ + Args: + detector: Learned mention detector. Without it, keyword matching decides + which aspects are mentioned (test/legacy behaviour). + """ + self.id2label = id2label or load_id2label(config_path) + self.tokenizer = tokenizer or LightTokenizer(tokenizer_path) + self.session = session or create_session(model_path) + self.detector = detector + + @staticmethod + def keyword_detect(text: str, aspect: str) -> bool: + """Fast pre-filter: is the aspect plausibly mentioned?""" + return bool(_KEYWORD_PATTERNS[aspect].search(text)) + + def classify_aspects(self, text: str, aspects: list[str]) -> dict[str, dict[str, Any]]: + """Classify sentiment for several aspects of one review in a single batch.""" + if not aspects: + return {} + encoded = self.tokenizer.encode_pairs([(text, ASPECT_PHRASES[a]) for a in aspects]) + logits = np.asarray(self.session.run(None, build_feed(self.session, encoded))[0]) + probs = softmax(logits.astype(np.float64), axis=-1) + + label2id = {label: idx for idx, label in self.id2label.items()} + results: dict[str, dict[str, Any]] = {} + for row, aspect in zip(probs, aspects, strict=True): + pred_idx = int(np.argmax(row)) + results[aspect] = { + "detected": True, + "sentiment": self.id2label[pred_idx], + "confidence": round(float(row[pred_idx]), 4), + "prob_neg": round(float(row[label2id["negative"]]), 4), + "prob_neu": round(float(row[label2id["neutral"]]), 4), + "prob_pos": round(float(row[label2id["positive"]]), 4), + } + return results + + def mentioned(self, clean_text: str) -> dict[str, bool]: + """Which aspects the review mentions: learned detector if loaded, else keywords.""" + if self.detector is not None: + return self.detector.detect(clean_text) + return {a: self.keyword_detect(clean_text, a) for a in EARBUD_ASPECTS} + + def process( + self, clean_text: str, mentioned: Mapping[str, bool] | None = None + ) -> dict[str, Any]: + """Return ``{"aspects": {aspect_key: result}}`` for all 10 aspects, in canonical order. + + Args: + mentioned: Precomputed mention decisions. Training passes out-of-fold + detector decisions here so fusion never sees in-sample detections. + """ + decisions = mentioned if mentioned is not None else self.mentioned(clean_text) + detected = [a for a in EARBUD_ASPECTS if decisions.get(a, False)] + classified = self.classify_aspects(clean_text, detected) + return {"aspects": {a: classified.get(a, not_mentioned()) for a in EARBUD_ASPECTS}} diff --git a/backend/app/ml/agents/aspect_detector.py b/backend/app/ml/agents/aspect_detector.py new file mode 100644 index 0000000000000000000000000000000000000000..359f549792474674e88265a71ca5fbe18379414f --- /dev/null +++ b/backend/app/ml/agents/aspect_detector.py @@ -0,0 +1,116 @@ +"""Learned aspect detector β€” decides WHICH of the 10 earbud aspects a review mentions. + +Replaces the keyword-only pre-filter of Agent 3 (PRD Β§9). Keyword lists alone +reached detection F1 0.54 (build quality) and 0.66 (connectivity) on the round-1 +test split: durability phrasings were missed and generic words ("pair of +headphones") fired falsely. + +Model, per aspect: logistic regression on word 1–2-gram TF-IDF features plus +the keyword hits of ALL aspects as extra features, with a decision threshold +chosen by product-grouped out-of-fold F1 on training rows only +(``training/vera_training/aspects.py``). Features are built by +:func:`detector_features`, shared by training and serving. + +SECURITY: the artifact is a joblib pickle β€” only load files produced by the VERA +training pipeline. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from pathlib import Path +from typing import Any, Final + +import joblib +import numpy as np +import scipy.sparse as sp +import sklearn + +from app.ml.agents.absa import ABSAModule +from app.ml.constants import EARBUD_ASPECTS + +#: Scale of the keyword indicator columns relative to TF-IDF values (β‰ˆ unit-norm rows). +KEYWORD_FEATURE_WEIGHT: Final = 3.0 +ARTIFACT_KEYS: Final = frozenset({"vectorizer", "models", "thresholds", "aspects"}) + + +def keyword_matrix(texts: Sequence[str]) -> np.ndarray: + """``(n_texts, 10)`` float matrix of keyword hits in canonical aspect order.""" + return np.array( + [[ABSAModule.keyword_detect(text, aspect) for aspect in EARBUD_ASPECTS] for text in texts], + dtype=np.float64, + ) + + +def detector_features(vectorizer: Any, texts: Sequence[str]) -> sp.csr_matrix: + """TF-IDF features concatenated with weighted keyword indicators.""" + tfidf = vectorizer.transform(list(texts)) + keywords = sp.csr_matrix(keyword_matrix(texts) * KEYWORD_FEATURE_WEIGHT) + return sp.hstack([tfidf, keywords]).tocsr() + + +class AspectDetector: + """Per-aspect probabilities and thresholded mention decisions.""" + + def __init__(self, artifact_path: str | Path) -> None: + artifacts = joblib.load(artifact_path) + self._init_from(artifacts, source=str(artifact_path)) + + @classmethod + def from_components( + cls, + vectorizer: Any, + models: dict[str, Any], + thresholds: dict[str, float], + ) -> AspectDetector: + """Build from in-memory objects (training code and tests).""" + instance = cls.__new__(cls) + instance._init_from( + { + "vectorizer": vectorizer, + "models": models, + "thresholds": thresholds, + "aspects": list(EARBUD_ASPECTS), + }, + source="memory", + ) + return instance + + def _init_from(self, artifacts: dict[str, Any], source: str) -> None: + missing = ARTIFACT_KEYS - set(artifacts) + if missing: + raise ValueError(f"{source} is missing keys {sorted(missing)}") + if tuple(artifacts["aspects"]) != EARBUD_ASPECTS: + raise ValueError(f"{source} aspect order differs from EARBUD_ASPECTS; retrain it") + trained_with = artifacts.get("sklearn_version") + if trained_with is not None and trained_with != sklearn.__version__: + raise ValueError( + f"{source} trained with scikit-learn {trained_with}, runtime has " + f"{sklearn.__version__}; retrain with the pinned version." + ) + self.vectorizer = artifacts["vectorizer"] + self.models: dict[str, Any] = artifacts["models"] + self.thresholds: dict[str, float] = { + k: float(v) for k, v in artifacts["thresholds"].items() + } + + def probabilities(self, texts: Sequence[str]) -> np.ndarray: + """``(n_texts, 10)`` mention probabilities in canonical aspect order.""" + features = detector_features(self.vectorizer, texts) + return np.column_stack( + [self.models[aspect].predict_proba(features)[:, 1] for aspect in EARBUD_ASPECTS] + ) + + def decide(self, probabilities: np.ndarray) -> list[dict[str, bool]]: + """Apply per-aspect thresholds to a probability matrix.""" + return [ + { + aspect: bool(row[i] >= self.thresholds[aspect]) + for i, aspect in enumerate(EARBUD_ASPECTS) + } + for row in np.atleast_2d(probabilities) + ] + + def detect(self, text: str) -> dict[str, bool]: + """Mention decisions for one review.""" + return self.decide(self.probabilities([text]))[0] diff --git a/backend/app/ml/agents/consistency.py b/backend/app/ml/agents/consistency.py new file mode 100644 index 0000000000000000000000000000000000000000..3d3197ed60422c76b6f95ec2425b54abb22b89da --- /dev/null +++ b/backend/app/ml/agents/consistency.py @@ -0,0 +1,40 @@ +"""Agent 4 β€” Star-Text Consistency Scorer (PRD Β§9, Agent 4). + +Measures how far the clicked star rating is from the text-derived experience +score. A large gap means the star is unreliable. +""" + +from __future__ import annotations + +from typing import Any, Final + +from app.ml.constants import SCORE_RANGE + +#: Gap (in stars) beyond which the star is flagged as inflated / deflated. +INFLATION_THRESHOLD: Final = 1.0 + + +class ConsistencyScorer: + """Agent 4: star vs text agreement.""" + + @staticmethod + def process(star_rating: float, experience_score: float) -> dict[str, Any]: + """Compare the star with Agent 2's score. + + Returns: + ``star_text_gap`` (positive = star higher than text), its absolute + value, inflation/deflation flags, ``consistency_score`` in [0, 1] + (1 = perfect agreement; max gap on a 1–5 scale is 4), and + ``star_trust_weight`` = consistencyΒ² (downweights stars harder). + """ + gap = float(star_rating) - float(experience_score) + abs_gap = abs(gap) + consistency = max(0.0, 1.0 - abs_gap / SCORE_RANGE) + return { + "star_text_gap": round(gap, 3), + "absolute_gap": round(abs_gap, 3), + "star_inflated": gap > INFLATION_THRESHOLD, + "star_deflated": gap < -INFLATION_THRESHOLD, + "consistency_score": round(consistency, 3), + "star_trust_weight": round(consistency**2, 3), + } diff --git a/backend/app/ml/agents/domain_rating.py b/backend/app/ml/agents/domain_rating.py new file mode 100644 index 0000000000000000000000000000000000000000..c3f68e753c2ae5602eb189fd34652546b692c7fd --- /dev/null +++ b/backend/app/ml/agents/domain_rating.py @@ -0,0 +1,67 @@ +"""Agent 6 β€” Domain Rating Model (PRD Β§9, Agent 6). + +TF-IDF -> Ridge regression trained on the UNLABELED earbud corpus, using its +star ratings as a noisy proxy target. Training lives offline in +``training/vera_training/domain.py`` (with the held-out ASIN leakage guard); +this module only loads and applies the artifact. + +SECURITY: the artifact is a joblib pickle. Only load files produced by the VERA +training notebooks from your own HF repo β€” unpickling untrusted files executes code. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import joblib +import numpy as np +import sklearn + +from app.ml.constants import SCORE_MAX, SCORE_MIN + + +class DomainRatingModel: + """Agent 6: text -> implied rating from domain vocabulary.""" + + def __init__(self, model_path: str | Path) -> None: + try: + artifacts = joblib.load(model_path) + except ModuleNotFoundError as exc: # e.g. "numpy._core" from a numpy-2 pickle + raise ValueError( + f"{model_path} was pickled with incompatible library versions ({exc}). " + "Retrain it in the production-pinned .venv (see notebooks/05)." + ) from exc + if not {"vectorizer", "model"} <= set(artifacts): + raise ValueError(f"{model_path} must contain 'vectorizer' and 'model'") + trained_with = artifacts.get("sklearn_version") + if trained_with is not None and trained_with != sklearn.__version__: + raise ValueError( + f"{model_path} trained with scikit-learn {trained_with}, runtime has " + f"{sklearn.__version__}; retrain with the pinned version." + ) + self.vectorizer = artifacts["vectorizer"] + self.model = artifacts["model"] + + @classmethod + def from_components(cls, vectorizer: Any, model: Any) -> DomainRatingModel: + """Build from in-memory objects (training notebooks and tests).""" + instance = cls.__new__(cls) + instance.vectorizer = vectorizer + instance.model = model + return instance + + def process(self, clean_text: str) -> dict[str, Any]: + """Predict the implied rating. + + ``domain_confidence`` is the PRD's proxy: 1 βˆ’ distance to the nearest + integer star (Ridge gives no native uncertainty). + """ + features = self.vectorizer.transform([clean_text]) + raw_score = float(self.model.predict(features)[0]) + domain_score = float(np.clip(raw_score, SCORE_MIN, SCORE_MAX)) + domain_confidence = 1.0 - abs(domain_score - round(domain_score)) + return { + "domain_score": round(domain_score, 3), + "domain_confidence": round(domain_confidence, 3), + } diff --git a/backend/app/ml/agents/experience_extractor.py b/backend/app/ml/agents/experience_extractor.py new file mode 100644 index 0000000000000000000000000000000000000000..b45441e728cf64657957399f0407ecdce591f3a2 --- /dev/null +++ b/backend/app/ml/agents/experience_extractor.py @@ -0,0 +1,79 @@ +"""Agent 2 β€” Experience Extractor (PRD Β§9, Agent 2). + +Fine-tuned DeBERTa-v3-base regression head, exported to INT8 ONNX, estimates +the experience score (1–5) from review text alone. + +The model input is produced by :func:`format_experience_input`, which the +fine-tuning notebook imports too β€” training and serving must see identical +text. (The PRD prepended ``"{title} [SEP]"`` at inference but trained on the +bare text: a train/serve skew. Both now use the bare cleaned text.) +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import numpy as np + +from app.ml.constants import SCORE_MAX, SCORE_MIN +from app.ml.models.onnx_session import InferenceSessionLike, build_feed, create_session +from app.ml.models.tokenizer_wrapper import LightTokenizer + + +def format_experience_input(clean_text: str) -> str: + """Canonical model input text. Shared with ``vera_training.finetune``.""" + return clean_text.strip() + + +class ExperienceExtractor: + """Agent 2: text -> predicted experience score.""" + + def __init__( + self, + model_path: str | Path, + tokenizer_path: str | Path, + *, + session: InferenceSessionLike | None = None, + tokenizer: LightTokenizer | None = None, + ) -> None: + """Load the ONNX session and tokenizer. + + ``session`` / ``tokenizer`` may be injected (tests); otherwise they are + built from the paths. + """ + # max_length comes from the exported tokenizer.json (= fine-tuning length). + self.tokenizer = tokenizer or LightTokenizer(tokenizer_path) + self.session = session or create_session(model_path) + + def predict_raw(self, texts: list[str]) -> np.ndarray: + """Return unclipped regression outputs, shape ``(len(texts),)``.""" + encoded = self.tokenizer.encode_batch([format_experience_input(t) for t in texts]) + outputs = self.session.run(None, build_feed(self.session, encoded)) + logits = np.asarray(outputs[0], dtype=np.float64) + return logits.reshape(len(texts), -1)[:, 0] + + @staticmethod + def output_from_raw(raw_score: float) -> dict[str, Any]: + """Agent 2 output dict from one raw regression value. + + Shared with training, where out-of-fold predictions come from PyTorch + rather than this ONNX session. + + Returns: + ``predicted_score`` clipped to [1, 5], the ``raw_score``, and + ``model_uncertainty`` β€” distance of the raw output from the nearest + 0.5 label step (a fixed proxy feature defined by the PRD; the + calibrated uncertainty comes from conformal prediction, not this). + """ + raw = float(raw_score) + nearest_half = round(raw * 2) / 2 + return { + "predicted_score": float(np.clip(raw, SCORE_MIN, SCORE_MAX)), + "raw_score": raw, + "model_uncertainty": abs(raw - nearest_half), + } + + def predict(self, clean_text: str) -> dict[str, Any]: + """Predict one review (see :meth:`output_from_raw` for the fields).""" + return self.output_from_raw(float(self.predict_raw([clean_text])[0])) diff --git a/backend/app/ml/agents/fusion.py b/backend/app/ml/agents/fusion.py new file mode 100644 index 0000000000000000000000000000000000000000..d1a16fc3e7ca117df550a894e6a9ec6724b19e88 --- /dev/null +++ b/backend/app/ml/agents/fusion.py @@ -0,0 +1,105 @@ +"""Agent 7 β€” Fusion (PRD Β§10.2). + +A bootstrap ensemble of XGBoost regressors maps the 40 fusion features to the +corrected experience score. The ensemble mean is the point estimate; the +ensemble standard deviation feeds normalised conformal intervals. + +Artifact format β€” ``fusion_ensemble.json`` (see DECISIONS.md: JSON instead of +the PRD's joblib pickle, because pickled XGBoost objects break across library +versions and unpickling is a code-execution risk):: + + { + "format_version": 1, + "feature_names": [...], # must equal app.ml.features.FEATURE_NAMES + "imputation_means": {...}, # Agent 5 training means + "members": ["", ...], + "metadata": {...} # free-form: n_train, xgboost_version, ... + } +""" + +from __future__ import annotations + +import json +from collections.abc import Sequence +from pathlib import Path +from typing import Any, Final + +import numpy as np +import xgboost as xgb + +from app.ml.agents.quality import IMPUTATION_KEYS +from app.ml.constants import SCORE_MAX, SCORE_MIN +from app.ml.features import FEATURE_NAMES, N_FEATURES + +FORMAT_VERSION: Final = 1 + + +def save_fusion_artifact( + path: str | Path, + members: Sequence[xgb.XGBRegressor | xgb.Booster], + imputation_means: dict[str, float], + metadata: dict[str, Any] | None = None, +) -> None: + """Serialise an ensemble to the portable JSON artifact.""" + if not members: + raise ValueError("ensemble needs at least one member") + absent = [k for k in IMPUTATION_KEYS if k not in imputation_means] + if absent: + raise ValueError(f"imputation_means missing keys: {absent}") + + serialised = [] + for member in members: + booster = member.get_booster() if isinstance(member, xgb.XGBRegressor) else member + serialised.append(bytes(booster.save_raw(raw_format="json")).decode("utf-8")) + + payload = { + "format_version": FORMAT_VERSION, + "feature_names": list(FEATURE_NAMES), + "imputation_means": {k: float(v) for k, v in imputation_means.items()}, + "members": serialised, + "metadata": {"xgboost_version": xgb.__version__, **(metadata or {})}, + } + Path(path).write_text(json.dumps(payload), encoding="utf-8") + + +class FusionModel: + """Agent 7: loads the ensemble and predicts mean Β± std.""" + + def __init__(self, artifact_path: str | Path) -> None: + payload = json.loads(Path(artifact_path).read_text(encoding="utf-8")) + + if payload.get("format_version") != FORMAT_VERSION: + raise ValueError(f"Unsupported fusion artifact version {payload.get('format_version')}") + if tuple(payload["feature_names"]) != FEATURE_NAMES: + raise ValueError( + "Fusion artifact feature_names differ from app.ml.features.FEATURE_NAMES β€” " + "retrain the ensemble with the current feature contract." + ) + + self.imputation_means: dict[str, float] = payload["imputation_means"] + self.metadata: dict[str, Any] = payload.get("metadata", {}) + self.members: list[xgb.Booster] = [] + for raw in payload["members"]: + booster = xgb.Booster() + booster.load_model(bytearray(raw.encode("utf-8"))) + if booster.num_features() != N_FEATURES: + raise ValueError( + f"member expects {booster.num_features()} features, not {N_FEATURES}" + ) + self.members.append(booster) + if not self.members: + raise ValueError("fusion artifact contains no members") + + def predict_members(self, features: np.ndarray) -> np.ndarray: + """Raw member predictions, shape ``(n_members, n_rows)``.""" + matrix = np.atleast_2d(np.asarray(features, dtype=np.float32)) + return np.stack([np.asarray(m.inplace_predict(matrix)) for m in self.members]) + + def predict(self, features: np.ndarray) -> dict[str, float]: + """Predict one review: clipped ensemble mean and member std.""" + preds = self.predict_members(features)[:, 0] + return { + "score": float(np.clip(preds.mean(), SCORE_MIN, SCORE_MAX)), + "raw_mean": float(preds.mean()), + "ensemble_std": float(preds.std()), + } diff --git a/backend/app/ml/agents/preprocessor.py b/backend/app/ml/agents/preprocessor.py new file mode 100644 index 0000000000000000000000000000000000000000..278320a67b912eab5119940c064cea27e1f5726f --- /dev/null +++ b/backend/app/ml/agents/preprocessor.py @@ -0,0 +1,168 @@ +"""Agent 1 β€” Text Preprocessor (PRD Β§9, Agent 1). + +Gates downstream agents with clean text, language detection, an earbud +contamination check, and text-quality metrics. + +DESIGN CONSTRAINT: zero torch / transformers imports. Loading +SentenceTransformer here would add ~350 MB RAM and blow Render's 512 MB limit. +Instead: ``langdetect`` (pure Python) + a tiny sklearn TF-IDF over anchor +phrases + word-boundary keyword matching. +""" + +from __future__ import annotations + +import html +import re +from typing import Any, Final + +from langdetect import DetectorFactory, detect +from sklearn.feature_extraction.text import TfidfVectorizer +from sklearn.metrics.pairwise import cosine_similarity + +DetectorFactory.seed = 42 # make langdetect deterministic + +_URL_RE: Final = re.compile(r"(?:https?://|www\.)\S+", re.IGNORECASE) +_BREAK_TAG_RE: Final = re.compile(r"<\s*(?:br|/p|p)\s*/?\s*>", re.IGNORECASE) +_TAG_RE: Final = re.compile(r"<[^>]+>") +_WS_RE: Final = re.compile(r"\s+") +_SENTENCE_SPLIT_RE: Final = re.compile(r"[.!?]+") + +#: Minimum characters for langdetect to be meaningful (PRD weakness #14). +MIN_LANGDETECT_CHARS: Final = 10 +#: TF-IDF cosine threshold for the body-text earbud check. +EARBUD_SIMILARITY_THRESHOLD: Final = 0.25 + + +class TextPreprocessor: + """Agent 1: clean, validate, and characterise review text.""" + + EARBUD_KEYWORDS: Final[tuple[str, ...]] = ( + "earbud", + "earbuds", + "earphone", + "earphones", + "in-ear", + "headphone", + "headphones", + "headset", + "tws", + "true wireless", + "noise cancel", + "noise cancelling", + "noise canceling", + "airpod", + "airpods", + "bluetooth ear", + "over-ear", + "on-ear", + "anc", + "wireless ear", + ) + + #: Anchor phrases for the TF-IDF cosine fallback (no model download). + _ANCHORS: Final[tuple[str, ...]] = ( + "sound quality is great", + "bass is powerful", + "noise cancellation works", + "battery life lasts hours", + "fit in my ear comfortably", + "earbud fell out", + "earphone sounds", + "in-ear headphone", + "bluetooth connection drops", + "microphone quality for calls", + "charging case", + "pairing device", + ) + + def __init__(self) -> None: + # Tiny TF-IDF fitted once on the anchors (~KBs of RAM). + self._vec = TfidfVectorizer(ngram_range=(1, 2), max_features=2000) + self._vec.fit(self._ANCHORS) + self._anchor_matrix = self._vec.transform(self._ANCHORS) + # Word-boundary match: "anc" must not fire on "dance" / "balance". + self._keyword_re = re.compile( + r"\b(?:" + "|".join(re.escape(k) for k in self.EARBUD_KEYWORDS) + r")\b", + re.IGNORECASE, + ) + + # ------------------------------------------------------------------ + @staticmethod + def clean_text(text: str) -> str: + """Strip URLs and HTML, unescape entities, and normalise whitespace. + + Line-break tags become spaces first β€” Amazon reviews use ``
`` + heavily and removing it outright glues adjacent words together. + """ + text = html.unescape(text or "") + text = _URL_RE.sub(" ", text) + text = _BREAK_TAG_RE.sub(" ", text) + text = _TAG_RE.sub(" ", text) + return _WS_RE.sub(" ", text).strip() + + @staticmethod + def detect_language(text: str) -> str: + """Return an ISO 639-1 code; fail open to ``'en'``. + + langdetect raises on very short, emoji-only, or punctuation-only text. + Those are expected inputs, not retryable errors, so they must never + crash a Celery task (PRD weakness #14). + """ + cleaned = text.strip() + if len(cleaned) < MIN_LANGDETECT_CHARS: + return "en" + try: + return str(detect(cleaned[:500])) + except Exception: + return "en" + + def is_earbud_content(self, text: str, title: str | None) -> tuple[bool, float]: + """Two-stage contamination filter. + + Stage 1: word-boundary keyword match on title + first 200 body chars. + Stage 2: max TF-IDF cosine between body text and earbud anchor phrases. + + Returns: + ``(is_earbud, confidence)`` where confidence is 1.0 for a keyword hit. + """ + combined = f"{title or ''} {text[:200]}" + if self._keyword_re.search(combined): + return True, 1.0 + + text_vec = self._vec.transform([text[:500]]) + max_sim = float(cosine_similarity(text_vec, self._anchor_matrix)[0].max()) + return max_sim > EARBUD_SIMILARITY_THRESHOLD, round(max_sim, 4) + + @staticmethod + def quality_metrics(text: str) -> dict[str, Any]: + """Length and lexical-diversity metrics on cleaned text.""" + words = text.split() + sentences = [s for s in (p.strip() for p in _SENTENCE_SPLIT_RE.split(text)) if len(s) > 3] + n_words = len(words) + n_sentences = max(len(sentences), 1) + return { + "word_count": n_words, + "sentence_count": n_sentences, + "avg_sentence_length": n_words / n_sentences, + "vocab_richness": len({w.lower() for w in words}) / max(n_words, 1), + } + + def process(self, text: str, title: str | None = "") -> dict[str, Any]: + """Run Agent 1 on raw review text. + + ``has_url`` is computed on the RAW text: the PRD computed it after URLs + were stripped, so it was always ``False``. + """ + raw = text or "" + clean = self.clean_text(raw) + lang = self.detect_language(clean) + is_earbud, earbud_conf = self.is_earbud_content(clean, title) + return { + "clean_text": clean, + "language": lang, + "is_english": lang == "en", + "is_earbud": is_earbud, + "earbud_confidence": earbud_conf, + "has_url": bool(_URL_RE.search(raw)), + **self.quality_metrics(clean), + } diff --git a/backend/app/ml/agents/quality.py b/backend/app/ml/agents/quality.py new file mode 100644 index 0000000000000000000000000000000000000000..fa60b0f8f4f30a9609305841b91ae8292e1ce660 --- /dev/null +++ b/backend/app/ml/agents/quality.py @@ -0,0 +1,131 @@ +"""Agent 5 β€” Review Quality Assessor (PRD Β§9, Agent 5). + +Interpretable heuristics (not a learned model) scoring how trustworthy a review +is as a signal. The result weights reviews in product aggregation. + +Fixes vs the PRD sketch (see DECISIONS.md): +* The PRD formula referenced ``quality_score`` inside its own definition via a + ``'quality_score' in dir()`` hack (always False there) and double-counted the + verified flag. The composite is now a plain weighted average. +* ``exp(-age/730)`` is an e-folding time, not a "2-year half-life". + Freshness now uses ``0.5 ** (age / 730)``, and future dates clip to age 0 + (previously produced freshness > 1). +* Missing metadata (``None``) is imputed with training-set means stored in the + fusion artifact, never hard zeros (PRD Β§10.1, weakness #15). +""" + +from __future__ import annotations + +import math +from datetime import UTC, datetime +from typing import Any, Final + +#: Dataset end date β€” freshness reference point for timestamped corpus reviews. +CUTOFF_DATE: Final = datetime(2023, 9, 1, tzinfo=UTC) +FRESHNESS_HALF_LIFE_DAYS: Final = 730.0 +HELPFUL_VOTES_SATURATION: Final = 100 +LENGTH_SATURATION_WORDS: Final = 500 + +WEIGHTS: Final[dict[str, float]] = { + "helpfulness": 0.35, + "freshness": 0.25, + "length": 0.25, + "verified": 0.15, +} +QUALITY_FLOOR: Final = 0.05 + +#: Keys the imputation mapping must provide (training-set means of the agent outputs). +IMPUTATION_KEYS: Final[tuple[str, ...]] = ( + "helpfulness_score", + "freshness_score", + "verified_purchase", +) + + +def _clip01(value: float) -> float: + return min(max(value, 0.0), 1.0) + + +class QualityAssessor: + """Agent 5: metadata -> trust weight.""" + + def __init__(self, imputation: dict[str, float] | None = None) -> None: + """ + Args: + imputation: Training-set means for :data:`IMPUTATION_KEYS`, loaded + from ``fusion_ensemble.json``. May be ``None`` only when every + review carries full metadata (the offline training corpus). + """ + if imputation is not None: + absent = [k for k in IMPUTATION_KEYS if k not in imputation] + if absent: + raise ValueError(f"imputation mapping missing keys: {absent}") + self.imputation = imputation + + def _impute(self, key: str) -> float: + if self.imputation is None: + raise ValueError( + f"Metadata for {key!r} is missing and no imputation means were provided" + ) + return float(self.imputation[key]) + + @staticmethod + def helpfulness(helpful_votes: int) -> float: + """Log-normalised helpful votes, saturating at 100 votes.""" + votes = max(int(helpful_votes), 0) + return _clip01(math.log1p(votes) / math.log1p(HELPFUL_VOTES_SATURATION)) + + @staticmethod + def freshness(timestamp_ms: int) -> float: + """True half-life decay relative to :data:`CUTOFF_DATE`.""" + review_date = datetime.fromtimestamp(timestamp_ms / 1000, tz=UTC) + age_days = max((CUTOFF_DATE - review_date).days, 0) + return float(0.5 ** (age_days / FRESHNESS_HALF_LIFE_DAYS)) + + def process( + self, + helpful_votes: int | None, + verified: bool | None, + timestamp_ms: int | None, + word_count: int, + ) -> dict[str, Any]: + """Score one review; ``None`` inputs are imputed with training means.""" + imputed: list[str] = [] + + if helpful_votes is None: + help_score = self._impute("helpfulness_score") + imputed.append("helpful_votes") + else: + help_score = self.helpfulness(helpful_votes) + + if timestamp_ms is None: + fresh_score = self._impute("freshness_score") + imputed.append("timestamp") + else: + fresh_score = self.freshness(timestamp_ms) + + if verified is None: + verified_score = self._impute("verified_purchase") + imputed.append("verified_purchase") + else: + verified_score = 1.0 if verified else 0.0 + + length_score = _clip01(max(word_count, 0) / LENGTH_SATURATION_WORDS) + + quality = ( + WEIGHTS["helpfulness"] * help_score + + WEIGHTS["freshness"] * fresh_score + + WEIGHTS["length"] * length_score + + WEIGHTS["verified"] * verified_score + ) + quality = min(max(quality, QUALITY_FLOOR), 1.0) + + return { + "helpfulness_score": round(help_score, 3), + "freshness_score": round(fresh_score, 3), + "length_score": round(length_score, 3), + "verified_purchase": round(verified_score, 3), + "quality_score": round(quality, 3), + "trust_weight": round(quality, 3), + "imputed_fields": imputed, + } diff --git a/backend/app/ml/calibration/__init__.py b/backend/app/ml/calibration/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..c1618075b634092d8f8f73c8c66452b63bd46aab --- /dev/null +++ b/backend/app/ml/calibration/__init__.py @@ -0,0 +1 @@ +"""Conformal calibration (per review) and Empirical Bayes aggregation (per product).""" diff --git a/backend/app/ml/calibration/conformal.py b/backend/app/ml/calibration/conformal.py new file mode 100644 index 0000000000000000000000000000000000000000..98f9d35c491fed0e351bb79100c58dd87bbe346e --- /dev/null +++ b/backend/app/ml/calibration/conformal.py @@ -0,0 +1,80 @@ +"""Production split-conformal interval predictor (PRD Β§10.3). + +Reads the pre-fitted quantile from ``conformal_qhat.json`` β€” no MAPIE, no +sklearn, only ``json`` + ``numpy`` (PRD weakness #16). Fitting happens offline +in ``training/vera_training/calibration.py`` (notebook 06). + +Two nonconformity methods are supported (see DECISIONS.md): + +* ``absolute`` β€” score ``|y βˆ’ Ε·|``; interval ``Ε· Β± qΜ‚``. Constant width, so + "confidence" only varies through clipping at the scale edges. +* ``normalized`` β€” score ``|y βˆ’ Ε·| / (ΟƒΜ‚ + floor)`` with ΟƒΜ‚ = ensemble std; + interval ``Ε· Β± qΜ‚Β·(ΟƒΜ‚ + floor)``. Same finite-sample coverage guarantee, but + widths adapt per review, which makes the confidence value informative. + +JSON schema:: + + {"method": "absolute"|"normalized", "q_hat": float, "alpha": float, + "nominal_coverage": float, "empirical_coverage": float, + "calibration_n": int, "sigma_floor": float (normalized only)} +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Final + +import numpy as np + +from app.ml.constants import SCORE_MAX, SCORE_MIN, SCORE_RANGE + +METHODS: Final = frozenset({"absolute", "normalized"}) + + +def interval_confidence(lower: float, upper: float) -> float: + """Map interval width to [0, 1]: a 0.5-wide interval -> 0.875, 2-wide -> 0.5.""" + return float(np.clip(1.0 - (upper - lower) / SCORE_RANGE, 0.0, 1.0)) + + +class ConformalPredictor: + """Online conformal predictor; loaded once per worker process.""" + + def __init__(self, qhat_path: str | Path) -> None: + data: dict[str, Any] = json.loads(Path(qhat_path).read_text(encoding="utf-8")) + self.method: str = data.get("method", "absolute") + if self.method not in METHODS: + raise ValueError(f"Unknown conformal method {self.method!r}") + self.q_hat = float(data["q_hat"]) + if not np.isfinite(self.q_hat) or self.q_hat < 0: + raise ValueError(f"q_hat must be a finite non-negative float, got {self.q_hat}") + self.nominal_coverage = float(data["nominal_coverage"]) + self.sigma_floor = float(data.get("sigma_floor", 0.0)) + if self.method == "normalized" and self.sigma_floor <= 0: + raise ValueError("normalized conformal requires sigma_floor > 0") + self._meta = data + + def half_width(self, ensemble_std: float = 0.0) -> float: + """Interval half-width for one prediction.""" + if self.method == "normalized": + return self.q_hat * (max(float(ensemble_std), 0.0) + self.sigma_floor) + return self.q_hat + + def predict_interval( + self, base_pred: float, ensemble_std: float = 0.0 + ) -> tuple[float, float, float]: + """Return ``(lower, upper, confidence)`` clipped to the 1–5 scale.""" + hw = self.half_width(ensemble_std) + lower = float(np.clip(base_pred - hw, SCORE_MIN, SCORE_MAX)) + upper = float(np.clip(base_pred + hw, SCORE_MIN, SCORE_MAX)) + return lower, upper, interval_confidence(lower, upper) + + def health_info(self) -> dict[str, Any]: + """Calibration metadata for diagnostics.""" + return { + "method": self.method, + "q_hat": self.q_hat, + "nominal_coverage": self.nominal_coverage, + "empirical_coverage": self._meta.get("empirical_coverage"), + "calibration_n": self._meta.get("calibration_n"), + } diff --git a/backend/app/ml/calibration/empirical_bayes.py b/backend/app/ml/calibration/empirical_bayes.py new file mode 100644 index 0000000000000000000000000000000000000000..cc4bb9436ef865cc2b06d24a92f75eaa5a3e2201 --- /dev/null +++ b/backend/app/ml/calibration/empirical_bayes.py @@ -0,0 +1,276 @@ +"""Product-level Empirical Bayes aggregation (PRD Β§11). + +A product with one 5.0 review must not outrank a product with 500 reviews +averaging 4.6. Each product's quality-weighted mean is shrunk toward the global +mean in proportion to how noisy that mean is (normal–normal model). + +Model, for product j with review scores y_ij and trust weights w_ij:: + + m_j = Ξ£ w_ij y_ij / Ξ£ w_ij (trust-weighted mean) + n_eff = (Ξ£ w_ij)Β² / Ξ£ w_ijΒ² (Kish effective sample size) + σ²_j = max(weighted var_j, 0.05) (noise floor, PRD weakness #5) + β€” conservative default variance when the product has one review + sΒ²_j = σ²_j / n_eff (sampling variance of m_j) + B_j = sΒ²_j / (sΒ²_j + τ²) (shrinkage factor) + ΞΈΜ‚_j = (1 βˆ’ B_j)Β·m_j + B_jΒ·ΞΌ (EB estimate) + sd_j = sqrt(B_j Β· τ²) (posterior sd = uncertainty) + +Prior (fitted across all products, method of moments):: + + ΞΌ = mean of all review scores + τ² = max(Var_j(m_j) βˆ’ mean_j(sΒ²_j), 0.01) (between-product variance) + +Corrections vs the PRD sketch (see DECISIONS.md): the PRD set τ² to the variance +of *individual review scores* (that is total variance, not between-product +variance), used τ² as σ² for single-review products, and reported an ad-hoc +``sqrt(σ²/n + B²τ²)`` instead of the posterior standard deviation. +""" + +from __future__ import annotations + +from collections import Counter +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass, field +from typing import Any, Final + +import numpy as np + +from app.ml.constants import EARBUD_ASPECTS, SCORE_MAX, SCORE_MIN + +SIGMA_SQ_FLOOR: Final = 0.05 +TAU_SQ_FLOOR: Final = 0.01 +#: Below this many products the between-product variance is not estimable. +MIN_PRODUCTS_FOR_PRIOR: Final = 3 +#: Weakly informative fallbacks: product means within ~Β±1 star, reviews within ~Β±0.7. +DEFAULT_TAU_SQ: Final = 0.25 +DEFAULT_SIGMA_SQ: Final = 0.5 + +_SENTIMENT_TIE_ORDER: Final = ("negative", "positive", "neutral") + + +@dataclass(frozen=True) +class ReviewSignal: + """The per-review quantities aggregation needs.""" + + product_asin: str + corrected_score: float + quality_weight: float + star_rating: float + aspects: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class EBPrior: + global_mean: float + tau_sq: float + pooled_sigma_sq: float + n_products: int + + +@dataclass(frozen=True) +class ProductAggregate: + product_asin: str + corrected_score: float + uncertainty: float + review_count: int + trust_weighted_score: float + star_average: float | None + eb_shrinkage_factor: float + aspect_summary: dict[str, Any] + + +@dataclass(frozen=True) +class _ProductStats: + n: int + weighted_mean: float + weighted_var: float | None # None when n == 1 + n_eff: float + + +def _product_stats(scores: Sequence[float], weights: Sequence[float]) -> _ProductStats: + y = np.asarray(scores, dtype=np.float64) + w = np.asarray(weights, dtype=np.float64) + if y.size == 0: + raise ValueError("cannot aggregate a product with no reviews") + if y.shape != w.shape: + raise ValueError("scores and weights must have the same length") + w = np.where(np.isfinite(w) & (w > 0), w, 0.0) + if w.sum() <= 0: # all weights unusable -> uniform + w = np.ones_like(y) + w = w / w.sum() + + mean = float(np.dot(w, y)) + n_eff = float(1.0 / np.sum(w**2)) + var: float | None = None + if y.size > 1: + denom = 1.0 - float(np.sum(w**2)) # unbiased reliability-weighted variance + var = float(np.dot(w, (y - mean) ** 2) / denom) if denom > 1e-12 else 0.0 + return _ProductStats(n=int(y.size), weighted_mean=mean, weighted_var=var, n_eff=n_eff) + + +class EmpiricalBayesAggregator: + """Fit a prior across products, then shrink each product's mean toward it.""" + + def __init__(self) -> None: + self.prior: EBPrior | None = None + + # ------------------------------------------------------------------ + def fit(self, grouped: Mapping[str, tuple[Sequence[float], Sequence[float]]]) -> EBPrior: + """Estimate ΞΌ, τ² and the pooled within-product variance. + + Args: + grouped: ``asin -> (corrected_scores, quality_weights)``. + """ + if not grouped: + raise ValueError("fit() needs at least one product") + stats = {asin: _product_stats(*sw) for asin, sw in grouped.items()} + all_scores = np.concatenate([np.asarray(s, dtype=np.float64) for s, _ in grouped.values()]) + + multi = [s for s in stats.values() if s.weighted_var is not None] + if multi: + dof = np.array([s.n - 1 for s in multi], dtype=np.float64) + variances = np.array([s.weighted_var for s in multi], dtype=np.float64) + pooled = float(np.dot(dof, variances) / dof.sum()) + else: + pooled = DEFAULT_SIGMA_SQ + pooled = max(pooled, SIGMA_SQ_FLOOR) + + if len(stats) >= MIN_PRODUCTS_FOR_PRIOR: + means = np.array([s.weighted_mean for s in stats.values()]) + sampling = np.array([self._sigma_sq(s, pooled) / s.n_eff for s in stats.values()]) + tau_sq = max(float(means.var(ddof=1) - sampling.mean()), TAU_SQ_FLOOR) + else: + tau_sq = DEFAULT_TAU_SQ + + self.prior = EBPrior( + global_mean=float(all_scores.mean()), + tau_sq=tau_sq, + pooled_sigma_sq=pooled, + n_products=len(stats), + ) + return self.prior + + @staticmethod + def _sigma_sq(stats: _ProductStats, pooled_sigma_sq: float) -> float: + # A one-review product has no within-product variance estimate. The + # observed pooled variance can be deceptively near zero when existing + # multi-review products are unanimous, so using it here would allow an + # isolated 5.0 to outrank a well-supported 4.6. Use the documented + # weakly informative variance instead; observed variance takes over as + # soon as a product has two reviews. + raw = ( + stats.weighted_var + if stats.weighted_var is not None + else max(pooled_sigma_sq, DEFAULT_SIGMA_SQ) + ) + return max(raw, SIGMA_SQ_FLOOR) + + # ------------------------------------------------------------------ + def aggregate_product( + self, + corrected_scores: Sequence[float], + quality_weights: Sequence[float], + aspect_sentiments: Sequence[Mapping[str, Any]], + star_ratings: Sequence[float] | None = None, + product_asin: str = "", + ) -> ProductAggregate: + """Compute the EB-shrunk aggregate for one product. Requires :meth:`fit` first.""" + if self.prior is None: + raise RuntimeError("call fit() before aggregate_product()") + prior = self.prior + stats = _product_stats(corrected_scores, quality_weights) + + s_sq = self._sigma_sq(stats, prior.pooled_sigma_sq) / stats.n_eff + shrink = float(np.clip(s_sq / (s_sq + prior.tau_sq), 0.0, 1.0)) + eb_score = (1.0 - shrink) * stats.weighted_mean + shrink * prior.global_mean + posterior_sd = float(np.sqrt(shrink * prior.tau_sq)) + + stars = list(star_ratings) if star_ratings is not None else [] + return ProductAggregate( + product_asin=product_asin, + corrected_score=round(float(np.clip(eb_score, SCORE_MIN, SCORE_MAX)), 2), + uncertainty=round(posterior_sd, 3), + review_count=stats.n, + trust_weighted_score=round(stats.weighted_mean, 2), + star_average=round(float(np.mean(stars)), 2) if stars else None, + eb_shrinkage_factor=round(shrink, 3), + aspect_summary=aggregate_aspects(aspect_sentiments), + ) + + def aggregate_all( + self, reviews: Iterable[ReviewSignal] + ) -> tuple[EBPrior, dict[str, ProductAggregate]]: + """Fit the prior on every product and aggregate each one.""" + by_asin: dict[str, list[ReviewSignal]] = {} + for review in reviews: + by_asin.setdefault(review.product_asin, []).append(review) + if not by_asin: + raise ValueError("aggregate_all() needs at least one review") + + prior = self.fit( + { + asin: ([r.corrected_score for r in rs], [r.quality_weight for r in rs]) + for asin, rs in by_asin.items() + } + ) + aggregates = { + asin: self.aggregate_product( + [r.corrected_score for r in rs], + [r.quality_weight for r in rs], + [r.aspects for r in rs], + star_ratings=[r.star_rating for r in rs], + product_asin=asin, + ) + for asin, rs in by_asin.items() + } + return prior, aggregates + + +def rank_products(aggregates: Mapping[str, ProductAggregate]) -> list[str]: + """ASINs best-first: EB score desc, then more reviews, then lower uncertainty, then ASIN.""" + return sorted( + aggregates, + key=lambda a: ( + -aggregates[a].corrected_score, + -aggregates[a].review_count, + aggregates[a].uncertainty, + a, + ), + ) + + +def aggregate_aspects(aspect_sentiments: Sequence[Mapping[str, Any]]) -> dict[str, Any]: + """Per-aspect mention rate, sentiment counts, net sentiment, and majority label. + + ``net_sentiment`` = (positive βˆ’ negative) / mentions, in [βˆ’1, 1] (drives the + radar chart). Ties for the majority label resolve negative > positive > + neutral, so a split verdict surfaces the complaint. Reviews whose aspect + dict lacks a key (e.g. legacy rows) count as "not mentioned". + """ + total = len(aspect_sentiments) + summary: dict[str, Any] = {} + for aspect in EARBUD_ASPECTS: + mentions = [ + a[aspect]["sentiment"] + for a in aspect_sentiments + if isinstance(a.get(aspect), Mapping) and a[aspect].get("detected") + ] + counts = Counter(mentions) + n = len(mentions) + if n == 0: + dominant = "not_mentioned" + else: + top = max(counts.values()) + dominant = next(s for s in _SENTIMENT_TIE_ORDER if counts.get(s, 0) == top) + summary[aspect] = { + "mention_count": n, + "detected_fraction": round(n / total, 3) if total else 0.0, + "positive": counts.get("positive", 0), + "negative": counts.get("negative", 0), + "neutral": counts.get("neutral", 0), + "net_sentiment": ( + round((counts.get("positive", 0) - counts.get("negative", 0)) / n, 3) if n else 0.0 + ), + "dominant_sentiment": dominant, + } + return summary diff --git a/backend/app/ml/constants.py b/backend/app/ml/constants.py new file mode 100644 index 0000000000000000000000000000000000000000..ed0b1f85fb63eedda68e460eeb4ee75244b22ff0 --- /dev/null +++ b/backend/app/ml/constants.py @@ -0,0 +1,142 @@ +"""Constants shared by production agents, the training package, and tests. + +Aspect keys are snake_case everywhere (API, DB JSON, features, labels). The +natural-language phrase in :data:`ASPECT_PHRASES` is only used as the second +sentence fed to the ABSA model. +""" + +from __future__ import annotations + +from typing import Final, Literal + +#: Canonical, ordered aspect keys. ORDER IS PART OF THE FUSION FEATURE CONTRACT β€” +#: never reorder without retraining the ensemble. +EARBUD_ASPECTS: Final[tuple[str, ...]] = ( + "sound_quality", + "bass", + "battery_life", + "comfort", + "fit", + "build_quality", + "connectivity", + "microphone", + "noise_cancellation", + "value", +) + +#: Aspect key -> phrase given to the ABSA model as the aspect sentence. +ASPECT_PHRASES: Final[dict[str, str]] = {key: key.replace("_", " ") for key in EARBUD_ASPECTS} + +#: Keyword pre-filter per aspect. Matched on word boundaries at the START of a +#: word (so "isolat" matches "isolation" but "fit" does not match "benefit"). +#: "cheap" is deliberately only under `value`: "cheap plastic" is still caught +#: by `build_quality` through "plastic". +ASPECT_KEYWORDS: Final[dict[str, tuple[str, ...]]] = { + "sound_quality": ( + "sound", + "audio", + "music", + "clarity", + "treble", + "highs", + "mids", + "tinny", + "muddy", + ), + "bass": ("bass", "low end", "boom", "thump", "sub-bass", "subwoofer"), + "battery_life": ("battery", "charge", "charging", "playtime", "endurance", "hours of"), + "comfort": ("comfort", "uncomfortable", "ear pain", "hurt", "wear", "wearing", "sore"), + "fit": ("fit", "fits", "stay in", "stays in", "fall out", "falls out", "ear tip", "seal"), + # Durability counts as build quality in the labeling protocol ("broke after a week", + # "connector bent"), which the original 7 keywords missed (train-split recall 0.35). + "build_quality": ( + "build", + "built", + "construct", + "durab", + "sturd", + "flims", + "fragil", + "well made", + "well-made", + "cheaply made", + "poorly made", + "plastic", + "material", + "hinge", + "crack", + "broke", + "break", + "snapped", + "fell apart", + "falling apart", + "came apart", + "cord", + "cable", + "wire", + "jack", + "connector", + "stopped working", + "quit working", + "died", + "defect", + "peel", + "wore out", + "worn out", + ), + # "pair" / "range" were removed: "a pair of headphones" and "price range" made + # connectivity precision 0.52 on the train split. + "connectivity": ( + "bluetooth", + "connect", + "disconnect", + "reconnect", + "pairing", + "paired", + "pair with", + "pair to", + "latency", + "lag", + "delay", + "drops out", + "dropped out", + "dropout", + "cuts out", + "cutting out", + "signal", + "out of range", + "sync", + "stutter", + "multipoint", + ), + "microphone": ("mic", "microphone", "calls", "phone call", "voice", "recording"), + "noise_cancellation": ( + "anc", + "noise cancel", + "noise-cancel", + "passive isolation", + "isolat", + "block out", + "blocks out", + "transparency mode", + ), + "value": ("price", "value", "worth", "money", "cheap", "expensive", "cost", "budget"), +} + +Sentiment = Literal["positive", "negative", "neutral", "not_mentioned"] + +#: Numeric encoding of aspect sentiment for fusion features (PRD Β§10.1). +SENTIMENT_TO_NUMERIC: Final[dict[str, int]] = { + "negative": -1, + "neutral": 0, + "positive": 1, + "not_mentioned": 0, +} + +#: Valid experience-score label values produced by the labeler (PRD Β§7.2). +LABEL_SCORE_VALUES: Final[tuple[float, ...]] = (1.0, 1.5, 2.0, 2.5, 3.0, 3.5, 4.0, 4.5, 5.0) + +SCORE_MIN: Final[float] = 1.0 +SCORE_MAX: Final[float] = 5.0 +#: Largest possible distance on the 1–5 scale. +SCORE_RANGE: Final[float] = SCORE_MAX - SCORE_MIN diff --git a/backend/app/ml/features.py b/backend/app/ml/features.py new file mode 100644 index 0000000000000000000000000000000000000000..9352ce9d50688f620b183893164a1a1ea81b2ccd --- /dev/null +++ b/backend/app/ml/features.py @@ -0,0 +1,125 @@ +"""Fusion feature contract (PRD Β§10.1). + +This module is the SINGLE source of truth for the 40-dimensional feature vector. +Notebook 05 (training), notebook 06 (calibration), and the production pipeline +all call :func:`build_feature_vector`, so train/serve skew is impossible by +construction. Changing the order or meaning of any feature requires retraining +the ensemble and re-fitting the calibrator. + +Imputation of missing metadata happens upstream in Agent 5 +(:class:`~app.ml.agents.quality.QualityAssessor`); the PRD's +``build_feature_vector(imputation=...)`` parameter was never actually used. +""" + +from __future__ import annotations + +from typing import Any, Final + +import numpy as np + +from app.ml.constants import EARBUD_ASPECTS, SENTIMENT_TO_NUMERIC + +FEATURE_NAMES: Final[tuple[str, ...]] = ( + # Base signals (3) + "star_rating", + "experience_score", + "domain_score", + # Consistency signals (4) + "star_text_gap", + "absolute_gap", + "star_inflated", + "consistency_score", + # Quality signals (4) + "quality_score", + "helpfulness_score", + "freshness_score", + "verified_purchase", + # Text quality signals (4) + "word_count_norm", + "vocab_richness", + "avg_sentence_length_norm", + "experience_model_uncertainty", + # Aspect detection (10) + aspect sentiment (10) + *(f"aspect_detected__{a}" for a in EARBUD_ASPECTS), + *(f"aspect_sentiment__{a}" for a in EARBUD_ASPECTS), + # Interactions (5) + "star_x_experience", + "abs_star_minus_domain", + "experience_x_quality", + "consistency_x_trust", + "domain_confidence_x_domain_score", +) + +N_FEATURES: Final = len(FEATURE_NAMES) + +#: Feature-name groups used by the ablation study (PRD Β§13.2). +FEATURE_GROUPS: Final[dict[str, tuple[str, ...]]] = { + "star": ("star_rating",), + "experience": ("experience_score", "experience_model_uncertainty", "star_x_experience"), + "domain": ("domain_score", "abs_star_minus_domain", "domain_confidence_x_domain_score"), + "consistency": ("star_text_gap", "absolute_gap", "star_inflated", "consistency_score"), + "quality": ( + "quality_score", + "helpfulness_score", + "freshness_score", + "verified_purchase", + "word_count_norm", + "vocab_richness", + "avg_sentence_length_norm", + "experience_x_quality", + "consistency_x_trust", + ), + "absa": tuple(n for n in FEATURE_NAMES if n.startswith("aspect_")), +} + + +def build_feature_vector(agent_outputs: dict[str, Any]) -> np.ndarray: + """Assemble the fusion features from Agents 1–6. + + Args: + agent_outputs: Mapping with keys ``preprocessor``, ``experience_extractor``, + ``absa``, ``consistency``, ``quality``, ``domain_rating`` (each the + agent's ``process``/``predict`` output) and ``star_rating``. + + Returns: + ``float32`` array of shape ``(N_FEATURES,)``. + """ + pre = agent_outputs["preprocessor"] + exp = agent_outputs["experience_extractor"] + absa = agent_outputs["absa"]["aspects"] + cons = agent_outputs["consistency"] + qual = agent_outputs["quality"] + dom = agent_outputs["domain_rating"] + star = float(agent_outputs["star_rating"]) + + feats: list[float] = [ + star, + exp["predicted_score"], + dom["domain_score"], + cons["star_text_gap"], + cons["absolute_gap"], + float(cons["star_inflated"]), + cons["consistency_score"], + qual["quality_score"], + qual["helpfulness_score"], + qual["freshness_score"], + float(qual["verified_purchase"]), + pre["word_count"] / 500, + pre["vocab_richness"], + pre["avg_sentence_length"] / 20, + exp["model_uncertainty"], + ] + feats += [float(absa[a]["detected"]) for a in EARBUD_ASPECTS] + feats += [float(SENTIMENT_TO_NUMERIC[absa[a]["sentiment"]]) for a in EARBUD_ASPECTS] + feats += [ + star * exp["predicted_score"], + abs(star - dom["domain_score"]), + exp["predicted_score"] * qual["quality_score"], + cons["consistency_score"] * qual["trust_weight"], + dom["domain_confidence"] * dom["domain_score"], + ] + + vector = np.asarray(feats, dtype=np.float32) + if vector.shape != (N_FEATURES,): # guards against silent contract drift + raise AssertionError(f"feature vector has {vector.shape[0]} dims, expected {N_FEATURES}") + return vector diff --git a/backend/app/ml/models/__init__.py b/backend/app/ml/models/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..5b8d38a9b4e4e7cdd7cc94c220daf80e6950b86d --- /dev/null +++ b/backend/app/ml/models/__init__.py @@ -0,0 +1 @@ +"""Model artifact management: tokenizer wrapper, ONNX session helpers, HF Hub loader.""" diff --git a/backend/app/ml/models/loader.py b/backend/app/ml/models/loader.py new file mode 100644 index 0000000000000000000000000000000000000000..9dc0383ef737fa05836bedcd7490b50d974f790f --- /dev/null +++ b/backend/app/ml/models/loader.py @@ -0,0 +1,128 @@ +"""Model artifact registry and HuggingFace Hub download (PRD Β§17.3). + +Production policy (user decision, see DECISIONS.md): the worker **hard-fails** +if any artifact is missing. There is no heuristic fallback. + +CLI (from ``backend/``):: + + python -m app.ml.models.loader # download missing artifacts, then verify + python -m app.ml.models.loader --check # verify only, no network +""" + +from __future__ import annotations + +import argparse +import logging +import sys +from pathlib import Path +from typing import Final + +logger = logging.getLogger(__name__) + +#: Every file the production pipeline needs, with the notebook that produces it. +REQUIRED_ARTIFACTS: Final[dict[str, str]] = { + "experience_model.onnx": "notebooks/04_finetune_experience.ipynb (INT8 ONNX)", + "experience_tokenizer.json": "notebooks/04_finetune_experience.ipynb", + "absa_model.onnx": "notebooks/04_finetune_experience.ipynb (ABSA export section)", + "absa_tokenizer.json": "notebooks/04_finetune_experience.ipynb (ABSA export section)", + "absa_config.json": "notebooks/04_finetune_experience.ipynb (ABSA export section)", + "aspect_detector.joblib": "training/scripts/train_all.py --stages aspects", + "domain_tfidf.joblib": "notebooks/05_train_fusion.ipynb (domain model section)", + "fusion_ensemble.json": "notebooks/05_train_fusion.ipynb", + "conformal_qhat.json": "notebooks/06_calibration.ipynb", +} + + +class ModelArtifactsMissingError(RuntimeError): + """Raised when required model artifacts are absent after any download attempt.""" + + def __init__(self, model_dir: Path, missing: list[str]) -> None: + self.model_dir = model_dir + self.missing = missing + details = "\n".join( + f" - {name} <- produced by {REQUIRED_ARTIFACTS[name]}" for name in missing + ) + super().__init__( + f"{len(missing)} required model artifact(s) missing in {model_dir}:\n{details}\n" + "Run notebooks 04 (Colab) and 05-06 (local), then upload the files to " + "HF_MODEL_REPO or copy them into MODEL_DIR." + ) + + +def missing_artifacts(model_dir: Path) -> list[str]: + """Return the names of required artifacts not present (or empty) in ``model_dir``.""" + return [ + name + for name in REQUIRED_ARTIFACTS + if not (model_dir / name).is_file() or (model_dir / name).stat().st_size == 0 + ] + + +def artifact_status(model_dir: Path) -> dict[str, bool]: + """Map each required artifact to whether it is present. Used by ``/health``.""" + missing = set(missing_artifacts(model_dir)) + return {name: name not in missing for name in REQUIRED_ARTIFACTS} + + +def ensure_models(model_dir: Path, hf_repo: str = "", hf_token: str = "") -> dict[str, Path]: + """Download missing artifacts from the Hub (if configured) and verify all exist. + + Args: + model_dir: Local artifact directory (created if absent). + hf_repo: ``user/repo`` on HuggingFace Hub; empty disables downloading. + hf_token: Optional token for private repos. + + Returns: + Mapping of artifact name to its local path. + + Raises: + ModelArtifactsMissingError: If anything is still missing afterwards. + """ + model_dir.mkdir(parents=True, exist_ok=True) + missing = missing_artifacts(model_dir) + + if missing and hf_repo: + from huggingface_hub import hf_hub_download # imported lazily: network-only path + + for name in missing: + logger.info("Downloading %s from %s ...", name, hf_repo) + try: + hf_hub_download( + repo_id=hf_repo, + filename=name, + local_dir=model_dir, + token=hf_token or None, + ) + except Exception as exc: # report every failure, then fail once below + logger.error("Failed to download %s: %s", name, exc) + missing = missing_artifacts(model_dir) + + if missing: + raise ModelArtifactsMissingError(model_dir, missing) + + logger.info("All %d model artifacts ready in %s", len(REQUIRED_ARTIFACTS), model_dir) + return {name: model_dir / name for name in REQUIRED_ARTIFACTS} + + +def _main(argv: list[str] | None = None) -> int: + from app.config import settings + + parser = argparse.ArgumentParser(description="Download / verify VERA model artifacts.") + parser.add_argument("--check", action="store_true", help="verify only; never download") + args = parser.parse_args(argv) + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") + + try: + ensure_models( + settings.MODEL_DIR, + hf_repo="" if args.check else settings.HF_MODEL_REPO, + hf_token=settings.HF_TOKEN, + ) + except ModelArtifactsMissingError as exc: + logger.error("%s", exc) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(_main()) diff --git a/backend/app/ml/models/onnx_session.py b/backend/app/ml/models/onnx_session.py new file mode 100644 index 0000000000000000000000000000000000000000..f2f0f3aa3fb727df6b7bb88ca68019aa3a805dc7 --- /dev/null +++ b/backend/app/ml/models/onnx_session.py @@ -0,0 +1,63 @@ +"""Thin onnxruntime helpers shared by the transformer agents (Agents 2 and 3).""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Protocol + +import numpy as np + + +class InferenceSessionLike(Protocol): + """Subset of ``onnxruntime.InferenceSession`` the agents rely on (eases testing).""" + + def get_inputs(self) -> list[Any]: ... + + def run( + self, output_names: list[str] | None, input_feed: dict[str, np.ndarray] + ) -> list[Any]: ... + + +def create_session(model_path: str | Path, intra_op_threads: int = 2) -> InferenceSessionLike: + """Create a CPU inference session with full graph optimisation.""" + import onnxruntime as ort # local import keeps module import cheap for tests + + path = Path(model_path) + if not path.is_file(): + raise FileNotFoundError(f"ONNX model not found: {path}") + opts = ort.SessionOptions() + opts.intra_op_num_threads = intra_op_threads + opts.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL + session: InferenceSessionLike = ort.InferenceSession( + str(path), sess_options=opts, providers=["CPUExecutionProvider"] + ) + return session + + +def build_feed( + session: InferenceSessionLike, encoded: dict[str, np.ndarray] +) -> dict[str, np.ndarray]: + """Select exactly the inputs the graph declares. + + DeBERTa-v3 has ``type_vocab_size=0``, so its ONNX export has no + ``token_type_ids`` input; feeding an undeclared input makes onnxruntime + raise ``INVALID_ARGUMENT``. Conversely, BERT-style exports require it. + """ + feed: dict[str, np.ndarray] = {} + for spec in session.get_inputs(): + name = spec.name + if name in encoded: + feed[name] = encoded[name].astype(np.int64, copy=False) + elif name == "token_type_ids": + feed[name] = np.zeros_like(encoded["input_ids"], dtype=np.int64) + else: + raise KeyError(f"ONNX graph expects unsupported input {name!r}") + return feed + + +def softmax(logits: np.ndarray, axis: int = -1) -> np.ndarray: + """Numerically stable softmax (the PRD's ``exp(x)/sum(exp(x))`` overflows for large logits).""" + shifted = logits - np.max(logits, axis=axis, keepdims=True) + exp = np.exp(shifted) + result: np.ndarray = exp / np.sum(exp, axis=axis, keepdims=True) + return result diff --git a/backend/app/ml/models/tokenizer_wrapper.py b/backend/app/ml/models/tokenizer_wrapper.py new file mode 100644 index 0000000000000000000000000000000000000000..d6d0e72296ab2cfdd6fe2157520219da6d8a395c --- /dev/null +++ b/backend/app/ml/models/tokenizer_wrapper.py @@ -0,0 +1,82 @@ +"""Torch-free tokenizer (PRD Β§4.4). + +Drop-in replacement for ``transformers.AutoTokenizer`` in production. The +HuggingFace ``tokenizers`` library is a small Rust extension that reads the same +``tokenizer.json`` a fine-tuned model ships with, but never imports torch. + +Differences from the PRD sketch (see DECISIONS.md): +* Loads from a local ``tokenizer.json`` file (downloaded with the ONNX model) + instead of ``Tokenizer.from_pretrained`` β€” no network at inference time, and + the base ``microsoft/deberta-v3-base`` repo does not ship a ``tokenizer.json``. +* Pads to the longest sequence in the batch instead of always to 512 tokens. + The ONNX graphs have dynamic axes, so this is numerically identical and + several times faster on a shared CPU. +* The pad id/token are read from the vocabulary instead of hardcoded to 0. +* Pair inputs use ``only_first`` truncation so the (short) aspect sentence is + never cut β€” only the review body is truncated. +""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +from tokenizers import Encoding, Tokenizer + +_PAD_CANDIDATES = ("[PAD]", "", "") +DEFAULT_MAX_LENGTH = 512 + + +class LightTokenizer: + """Rust-backed tokenizer producing int64 numpy batches for onnxruntime.""" + + def __init__(self, tokenizer_path: str | Path, max_length: int | None = None) -> None: + """ + Args: + tokenizer_path: ``tokenizer.json`` exported with the model. + max_length: Truncation length. ``None`` uses the length stored in the + file's truncation settings (written at export so serving matches + training), falling back to 512. + """ + path = Path(tokenizer_path) + if not path.is_file(): + raise FileNotFoundError(f"tokenizer.json not found: {path}") + self._tok = Tokenizer.from_file(str(path)) + stored = (self._tok.truncation or {}).get("max_length") + self.max_length = int(max_length or stored or DEFAULT_MAX_LENGTH) + + pad_token = next((t for t in _PAD_CANDIDATES if self._tok.token_to_id(t) is not None), None) + if pad_token is None: + raise ValueError(f"No padding token {_PAD_CANDIDATES} in vocabulary of {path}") + self.pad_token = pad_token + self.pad_id = int(self._tok.token_to_id(pad_token)) + self._tok.enable_padding(pad_id=self.pad_id, pad_token=self.pad_token) + + # ------------------------------------------------------------------ + def __call__(self, text: str, text_pair: str | None = None) -> dict[str, np.ndarray]: + """Encode one text (or one text pair) into a batch of size 1.""" + if text_pair is None: + return self.encode_batch([text]) + return self.encode_pairs([(text, text_pair)]) + + def encode_batch(self, texts: list[str]) -> dict[str, np.ndarray]: + """Encode single sequences, padded to the longest in the batch.""" + if not texts: + raise ValueError("encode_batch() needs at least one text") + self._tok.enable_truncation(max_length=self.max_length, strategy="longest_first") + return self._to_arrays(self._tok.encode_batch(texts)) + + def encode_pairs(self, pairs: list[tuple[str, str]]) -> dict[str, np.ndarray]: + """Encode (first, second) pairs; only the first sequence is ever truncated.""" + if not pairs: + raise ValueError("encode_pairs() needs at least one pair") + self._tok.enable_truncation(max_length=self.max_length, strategy="only_first") + return self._to_arrays(self._tok.encode_batch(pairs)) + + @staticmethod + def _to_arrays(encodings: list[Encoding]) -> dict[str, np.ndarray]: + return { + "input_ids": np.asarray([e.ids for e in encodings], dtype=np.int64), + "attention_mask": np.asarray([e.attention_mask for e in encodings], dtype=np.int64), + "token_type_ids": np.asarray([e.type_ids for e in encodings], dtype=np.int64), + } diff --git a/backend/app/ml/pipeline.py b/backend/app/ml/pipeline.py new file mode 100644 index 0000000000000000000000000000000000000000..d025c0cc92473485a13cb8790f36a573e5845f8f --- /dev/null +++ b/backend/app/ml/pipeline.py @@ -0,0 +1,218 @@ +"""VERA pipeline orchestrator (PRD Β§8.2). + +Runs Agents 1–6, fuses their signals with Agent 7, and wraps the point estimate +in a conformal interval. Product-level Empirical Bayes aggregation runs +separately (``app.worker.tasks.aggregate_products_task``) because it needs every +review of every product. + +Model policy: :class:`VERAPipeline` **hard-fails** at construction when any +artifact is missing (user decision, DECISIONS.md) β€” see +:mod:`app.ml.models.loader`. +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from app.ml.agents.absa import ABSAModule +from app.ml.agents.aspect_detector import AspectDetector +from app.ml.agents.consistency import ConsistencyScorer +from app.ml.agents.domain_rating import DomainRatingModel +from app.ml.agents.experience_extractor import ExperienceExtractor +from app.ml.agents.fusion import FusionModel +from app.ml.agents.preprocessor import TextPreprocessor +from app.ml.agents.quality import QualityAssessor +from app.ml.calibration.conformal import ConformalPredictor +from app.ml.constants import EARBUD_ASPECTS +from app.ml.features import build_feature_vector +from app.ml.models.loader import ensure_models + +logger = logging.getLogger(__name__) + + +@dataclass +class SignalAgents: + """Agents 1–6. ``experience`` may be ``None`` when callers supply its output + directly (offline training uses out-of-fold predictions).""" + + preprocessor: TextPreprocessor + experience: ExperienceExtractor | None + absa: ABSAModule + consistency: ConsistencyScorer + quality: QualityAssessor + domain: DomainRatingModel + + +def extract_signals( + agents: SignalAgents, + *, + text: str, + star_rating: float, + helpful_votes: int | None = None, + verified_purchase: bool | None = None, + timestamp_ms: int | None = None, + product_title: str | None = "", + experience_output: dict[str, Any] | None = None, + aspect_mentions: dict[str, bool] | None = None, +) -> dict[str, Any]: + """Run Agents 1–6 and return the ``agent_outputs`` mapping for fusion. + + Shared by production and the training notebooks so both build features + through exactly the same code path. Training supplies out-of-fold + ``experience_output`` and ``aspect_mentions``; production computes both. + """ + pre = agents.preprocessor.process(text, product_title) + clean = pre["clean_text"] + + if experience_output is None: + if agents.experience is None: + raise ValueError("experience_output is required when no ExperienceExtractor is loaded") + experience_output = agents.experience.predict(clean) + + quality = agents.quality.process( + helpful_votes=helpful_votes, + verified=verified_purchase, + timestamp_ms=timestamp_ms, + word_count=pre["word_count"], + ) + return { + "preprocessor": pre, + "experience_extractor": experience_output, + "absa": agents.absa.process(clean, aspect_mentions), + "consistency": agents.consistency.process( + star_rating, experience_output["predicted_score"] + ), + "quality": quality, + "domain_rating": agents.domain.process(clean), + "star_rating": float(star_rating), + } + + +def build_explanation(agent_outputs: dict[str, Any], corrected_score: float) -> str: + """Human-readable summary for the ScoreCard (PRD Β§16.2).""" + star = agent_outputs["star_rating"] + cons = agent_outputs["consistency"] + aspects = agent_outputs["absa"]["aspects"] + pre = agent_outputs["preprocessor"] + + def names(sentiment: str) -> list[str]: + return [a.replace("_", " ") for a in EARBUD_ASPECTS if aspects[a]["sentiment"] == sentiment] + + parts: list[str] = [] + if cons["star_inflated"]: + parts.append( + f"The text implies about {corrected_score:.1f} stars despite the {star:g}-star rating." + ) + elif cons["star_deflated"]: + parts.append( + f"The text is more positive than the {star:g}-star rating suggests " + f"(about {corrected_score:.1f})." + ) + else: + parts.append("The star rating and the review text broadly agree.") + + negatives, positives = names("negative"), names("positive") + if negatives: + parts.append(f"Concerns: {', '.join(negatives)}.") + if positives: + parts.append(f"Praised: {', '.join(positives)}.") + if not pre["is_english"]: + parts.append(f"Warning: detected language '{pre['language']}'; scores are English-trained.") + if not pre["is_earbud"]: + parts.append("Warning: this review may not be about earbuds or headphones.") + return " ".join(parts) + + +class VERAPipeline: + """End-to-end per-review inference. Construct once per worker process.""" + + def __init__(self, model_dir: Path, hf_repo: str = "", hf_token: str = "") -> None: + started = time.perf_counter() + paths = ensure_models(model_dir, hf_repo=hf_repo, hf_token=hf_token) # hard-fail + + self.fusion = FusionModel(paths["fusion_ensemble.json"]) + self.conformal = ConformalPredictor(paths["conformal_qhat.json"]) + self.agents = SignalAgents( + preprocessor=TextPreprocessor(), + experience=ExperienceExtractor( + paths["experience_model.onnx"], paths["experience_tokenizer.json"] + ), + absa=ABSAModule( + paths["absa_model.onnx"], + paths["absa_tokenizer.json"], + paths["absa_config.json"], + detector=AspectDetector(paths["aspect_detector.joblib"]), + ), + consistency=ConsistencyScorer(), + quality=QualityAssessor(imputation=self.fusion.imputation_means), + domain=DomainRatingModel(paths["domain_tfidf.joblib"]), + ) + logger.info("VERA pipeline loaded in %.1fs", time.perf_counter() - started) + + @classmethod + def from_components( + cls, agents: SignalAgents, fusion: FusionModel, conformal: ConformalPredictor + ) -> VERAPipeline: + """Assemble from pre-built parts (tests, notebooks).""" + instance = cls.__new__(cls) + instance.agents = agents + instance.fusion = fusion + instance.conformal = conformal + return instance + + def process( + self, + *, + text: str, + star_rating: float, + helpful_votes: int | None = None, + verified_purchase: bool | None = None, + product_title: str | None = "", + timestamp_ms: int | None = None, + ) -> dict[str, Any]: + """Score one review (PRD Β§3.1 output contract).""" + started = time.perf_counter() + outputs = extract_signals( + self.agents, + text=text, + star_rating=star_rating, + helpful_votes=helpful_votes, + verified_purchase=verified_purchase, + timestamp_ms=timestamp_ms, + product_title=product_title, + ) + fused = self.fusion.predict(build_feature_vector(outputs)) + lower, upper, confidence = self.conformal.predict_interval( + fused["score"], fused["ensemble_std"] + ) + corrected = round(fused["score"], 3) + + pre, cons, qual = outputs["preprocessor"], outputs["consistency"], outputs["quality"] + return { + "corrected_score": corrected, + "interval": [round(lower, 3), round(upper, 3)], + "confidence": round(confidence, 3), + "aspects": outputs["absa"]["aspects"], + # Agent 4 definition: star βˆ’ text-derived experience score (positive = inflated). + "star_text_gap": cons["star_text_gap"], + "review_quality": qual["quality_score"], + "ensemble_std": round(fused["ensemble_std"], 4), + "flags": { + "language": pre["language"], + "is_english": pre["is_english"], + "is_earbud": pre["is_earbud"], + "earbud_confidence": pre["earbud_confidence"], + "has_url": pre["has_url"], + "star_inflated": cons["star_inflated"], + "star_deflated": cons["star_deflated"], + "experience_score": round(outputs["experience_extractor"]["predicted_score"], 3), + "domain_score": outputs["domain_rating"]["domain_score"], + "imputed_fields": qual["imputed_fields"], + }, + "explanation": build_explanation(outputs, corrected), + "latency_ms": round((time.perf_counter() - started) * 1000, 1), + } diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..ade85dd0527e690bd9d002493cb66104932f1b33 --- /dev/null +++ b/backend/app/models/__init__.py @@ -0,0 +1,9 @@ +"""SQLAlchemy ORM models. Importing this package registers every table on ``Base``.""" + +from app.models.calibration import CalibrationCheckpoint +from app.models.job import Job +from app.models.label_run import LabelRun +from app.models.product import Product +from app.models.review import Review, ReviewStatus + +__all__ = ["CalibrationCheckpoint", "Job", "LabelRun", "Product", "Review", "ReviewStatus"] diff --git a/backend/app/models/calibration.py b/backend/app/models/calibration.py new file mode 100644 index 0000000000000000000000000000000000000000..7aa01ae693418e60321d6823d2aedcd1793af19e --- /dev/null +++ b/backend/app/models/calibration.py @@ -0,0 +1,24 @@ +"""`calibration_checkpoints` table β€” history of fitted conformal q_hat values (PRD Β§10.3).""" + +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import DateTime, Float, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +class CalibrationCheckpoint(Base): + __tablename__ = "calibration_checkpoints" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + method: Mapped[str] = mapped_column(String(20), nullable=False, default="absolute") + q_hat: Mapped[float] = mapped_column(Float, nullable=False) + empirical_coverage: Mapped[float] = mapped_column(Float, nullable=False) + nominal_coverage: Mapped[float] = mapped_column(Float, nullable=False) + calibration_set_size: Mapped[int] = mapped_column(Integer, nullable=False) + fitted_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) diff --git a/backend/app/models/job.py b/backend/app/models/job.py new file mode 100644 index 0000000000000000000000000000000000000000..6cc2c10ddcb27301a7d03ce39a240b0dc64f18a9 --- /dev/null +++ b/backend/app/models/job.py @@ -0,0 +1,49 @@ +"""`jobs` table β€” transactional outbox for background work. + +Why an outbox: Postgres and Redis cannot commit atomically. The API writes the +review AND its job row in one database transaction, so a job can never be lost +between "review saved" and "task queued". Redis only carries wake-up messages; +the database owns attempts, back-off, and completion. Duplicate deliveries are +harmless because consumers lock the job row and skip finished jobs. +""" + +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum +from uuid import uuid4 + +from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +class JobKind(StrEnum): + REVIEW = "review" # run the ML pipeline for one review + AGGREGATE = "aggregate" # recompute every product aggregate + ranking + + +class Job(Base): + __tablename__ = "jobs" + __table_args__ = (Index("idx_jobs_dispatch", "finished_at", "available_at"),) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid4())) + kind: Mapped[str] = mapped_column(String(20), nullable=False) + review_id: Mapped[int | None] = mapped_column( + ForeignKey("reviews.id", ondelete="CASCADE"), unique=True + ) + attempts: Mapped[int] = mapped_column(Integer, default=0, server_default="0", nullable=False) + #: Earliest time the job may run (pushed forward by retry back-off). + available_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + #: Last time a wake-up was sent to Redis; NULL means "send now". + last_dispatched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + + def __repr__(self) -> str: + return f"" diff --git a/backend/app/models/label_run.py b/backend/app/models/label_run.py new file mode 100644 index 0000000000000000000000000000000000000000..0e9a6a4fde70733495d5db941517b8c03df30d20 --- /dev/null +++ b/backend/app/models/label_run.py @@ -0,0 +1,28 @@ +"""`label_runs` table β€” audit trail of LLM-generated labels (PRD Β§7, Β§15).""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from sqlalchemy import DateTime, Float, Index, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base, JSONType + + +class LabelRun(Base): + __tablename__ = "label_runs" + __table_args__ = (Index("idx_label_runs_hash", "review_text_hash"),) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + review_text_hash: Mapped[str] = mapped_column(String(64), nullable=False) # sha256 hex + labeler: Mapped[str] = mapped_column(String(100), nullable=False) + experience_score: Mapped[float | None] = mapped_column(Float) + aspects: Mapped[dict[str, Any] | None] = mapped_column(JSONType) + has_irony: Mapped[bool | None] = mapped_column() + review_quality: Mapped[str | None] = mapped_column(String(10)) + raw_response: Mapped[dict[str, Any] | None] = mapped_column(JSONType) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) diff --git a/backend/app/models/product.py b/backend/app/models/product.py new file mode 100644 index 0000000000000000000000000000000000000000..cd8a98bf044c615c63a698ffc85e8234129c830c --- /dev/null +++ b/backend/app/models/product.py @@ -0,0 +1,44 @@ +"""`products` table β€” Empirical-Bayes product aggregates (PRD Β§11, Β§15).""" + +from __future__ import annotations + +from datetime import datetime +from typing import TYPE_CHECKING, Any + +from sqlalchemy import DateTime, Float, Index, Integer, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from app.database import Base, JSONType + +if TYPE_CHECKING: + from app.models.review import Review + + +class Product(Base): + __tablename__ = "products" + __table_args__ = (Index("idx_products_rank", "rank_position"),) + + asin: Mapped[str] = mapped_column(String(20), primary_key=True) + title: Mapped[str | None] = mapped_column(Text) + + corrected_score: Mapped[float | None] = mapped_column(Float) + uncertainty: Mapped[float | None] = mapped_column(Float) + trust_weighted_score: Mapped[float | None] = mapped_column(Float) + star_average: Mapped[float | None] = mapped_column(Float) + eb_shrinkage_factor: Mapped[float | None] = mapped_column(Float) + review_count: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default="0" + ) + aspect_summary: Mapped[dict[str, Any] | None] = mapped_column(JSONType) + + rank_position: Mapped[int | None] = mapped_column(Integer) + previous_rank_position: Mapped[int | None] = mapped_column(Integer) + last_aggregated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + + reviews: Mapped[list[Review]] = relationship(back_populates="product") + + def __repr__(self) -> str: + return f"" diff --git a/backend/app/models/review.py b/backend/app/models/review.py new file mode 100644 index 0000000000000000000000000000000000000000..2d0f4d38433792c7207385580e067db2d04c0e49 --- /dev/null +++ b/backend/app/models/review.py @@ -0,0 +1,87 @@ +"""`reviews` table β€” one submitted review and its pipeline output (PRD Β§15).""" + +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum +from typing import TYPE_CHECKING, Any + +from sqlalchemy import ( + CheckConstraint, + DateTime, + Float, + ForeignKey, + Index, + Integer, + String, + Text, + func, +) +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from app.database import Base, JSONType + +if TYPE_CHECKING: + from app.models.product import Product + + +class ReviewStatus(StrEnum): + """Processing lifecycle of a review.""" + + PENDING = "pending" + PROCESSING = "processing" + DONE = "done" + FAILED = "failed" + + +class Review(Base): + __tablename__ = "reviews" + __table_args__ = ( + CheckConstraint("star_rating >= 1 AND star_rating <= 5", name="ck_reviews_star_range"), + CheckConstraint( + "status IN ('pending', 'processing', 'done', 'failed')", name="ck_reviews_status" + ), + Index("idx_reviews_asin", "product_asin"), + Index("idx_reviews_status", "status"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + + # --- Input ---------------------------------------------------------------- + text: Mapped[str] = mapped_column(Text, nullable=False) + star_rating: Mapped[float] = mapped_column(Float, nullable=False) + # NULL means "not supplied" β€” the pipeline imputes training means (PRD Β§10.1). + helpful_votes: Mapped[int | None] = mapped_column(Integer, nullable=True) + verified_purchase: Mapped[bool | None] = mapped_column(nullable=True) + product_title: Mapped[str | None] = mapped_column(Text, nullable=True) + product_asin: Mapped[str | None] = mapped_column( + String(20), ForeignKey("products.asin", ondelete="SET NULL"), nullable=True + ) + + # --- Pipeline output ------------------------------------------------------ + corrected_score: Mapped[float | None] = mapped_column(Float) + interval_lower: Mapped[float | None] = mapped_column(Float) + interval_upper: Mapped[float | None] = mapped_column(Float) + confidence_value: Mapped[float | None] = mapped_column(Float) + star_text_gap: Mapped[float | None] = mapped_column(Float) + review_quality: Mapped[float | None] = mapped_column(Float) + aspects: Mapped[dict[str, Any] | None] = mapped_column(JSONType) + flags: Mapped[dict[str, Any] | None] = mapped_column(JSONType) + explanation: Mapped[str | None] = mapped_column(Text) + error_message: Mapped[str | None] = mapped_column(Text) + + # --- Bookkeeping ---------------------------------------------------------- + status: Mapped[str] = mapped_column( + String(20), nullable=False, default=ReviewStatus.PENDING.value + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False + ) + + product: Mapped[Product | None] = relationship(back_populates="reviews") + + def __repr__(self) -> str: + return f"" diff --git a/backend/app/schemas/__init__.py b/backend/app/schemas/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..201233e5909632f466d789cab86393a2fd5357d3 --- /dev/null +++ b/backend/app/schemas/__init__.py @@ -0,0 +1 @@ +"""Pydantic request and response schemas for the public API.""" diff --git a/backend/app/schemas/product.py b/backend/app/schemas/product.py new file mode 100644 index 0000000000000000000000000000000000000000..0ddecae0aa264fdee19dfce471f364b2597a7158 --- /dev/null +++ b/backend/app/schemas/product.py @@ -0,0 +1,39 @@ +"""Product API contracts.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from pydantic import BaseModel, ConfigDict + +from app.schemas.review import ReviewResponse + + +class ProductResponse(BaseModel): + """One row of the product ranking.""" + + model_config = ConfigDict(from_attributes=True) + + asin: str + title: str | None + corrected_score: float | None + uncertainty: float | None + trust_weighted_score: float | None + star_average: float | None + review_count: int + rank_position: int | None + previous_rank_position: int | None + aspect_summary: dict[str, Any] | None + + +class ProductDetailResponse(ProductResponse): + """Product aggregate plus one page of completed reviews.""" + + eb_shrinkage_factor: float | None + last_aggregated_at: datetime | None + reviews: list[ReviewResponse] + reviews_total: int + reviews_limit: int + reviews_offset: int + reviews_has_more: bool diff --git a/backend/app/schemas/review.py b/backend/app/schemas/review.py new file mode 100644 index 0000000000000000000000000000000000000000..0cf0185bed03d4329a789715d3fa97df66595b39 --- /dev/null +++ b/backend/app/schemas/review.py @@ -0,0 +1,95 @@ +"""Review API contracts.""" + +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from app.ml.constants import Sentiment + + +class AspectSentiment(BaseModel): + """Inference result for one earbud aspect.""" + + detected: bool + sentiment: Sentiment + confidence: float = Field(ge=0.0, le=1.0) + prob_neg: float | None = Field(default=None, ge=0.0, le=1.0) + prob_neu: float | None = Field(default=None, ge=0.0, le=1.0) + prob_pos: float | None = Field(default=None, ge=0.0, le=1.0) + + +class ReviewCreate(BaseModel): + """A review submitted for asynchronous scoring.""" + + text: str = Field(min_length=20, max_length=10_000) + star_rating: float = Field(ge=1.0, le=5.0) + helpful_votes: int | None = Field(default=None, ge=0) + verified_purchase: bool | None = None + product_title: str | None = Field(default=None, max_length=1_000) + product_asin: str | None = Field(default=None, max_length=20) + + @field_validator("text") + @classmethod + def text_must_contain_non_whitespace(cls, value: str) -> str: + value = value.strip() + if not value: + raise ValueError("text must contain non-whitespace characters") + return value + + @field_validator("product_title", "product_asin", mode="before") + @classmethod + def strip_optional_strings(cls, value: object) -> str | None: + if value is None: + return None + if not isinstance(value, str): + raise ValueError("must be a string") + return value.strip() or None + + +class ReviewTaskStatus(BaseModel): + """Polling response returned after a review is submitted.""" + + review_id: int + task_id: str | None + status: str + message: str + + +class ReviewResponse(BaseModel): + """Persisted review plus any completed pipeline result.""" + + model_config = ConfigDict(from_attributes=True) + + id: int + text: str + star_rating: float + helpful_votes: int | None + verified_purchase: bool | None + product_title: str | None + product_asin: str | None + corrected_score: float | None + interval_lower: float | None + interval_upper: float | None + confidence_value: float | None + star_text_gap: float | None + review_quality: float | None + aspects: dict[str, AspectSentiment] | None + flags: dict[str, Any] | None + explanation: str | None + error_message: str | None + status: str + created_at: datetime + updated_at: datetime + + +class ReviewLifecycle(StrEnum): + """Public lifecycle values; mirrors the persisted ORM enum.""" + + PENDING = "pending" + PROCESSING = "processing" + DONE = "done" + FAILED = "failed" diff --git a/backend/app/worker/__init__.py b/backend/app/worker/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..40241535f1fc7b6a6ca49ebd485cabfcb3e299e2 --- /dev/null +++ b/backend/app/worker/__init__.py @@ -0,0 +1 @@ +"""Celery application and asynchronous VERA tasks.""" diff --git a/backend/app/worker/celery_app.py b/backend/app/worker/celery_app.py new file mode 100644 index 0000000000000000000000000000000000000000..0d231e3c39b675835fc0b05a76dff639e2bbd2cf --- /dev/null +++ b/backend/app/worker/celery_app.py @@ -0,0 +1,71 @@ +"""Celery application (PRD Β§14.4), tuned for the Upstash free tier. + +UPSTASH BUDGET: the free tier allows 10,000 Redis commands/day. Celery's default +0.1 s broker polling would burn ~864K/day, so ``polling_interval`` is 2.0 s. +Even that is ~43K polls/day for ONE worker β€” above the free quota. Run a single +worker, and expect to need a paid Upstash plan (or local Redis) for sustained +use. Results are ignored (the DB stores outcomes), saving result-backend writes. + +Worker start:: + + cd backend && celery -A app.worker.celery_app worker --loglevel=info + +Startup behaviour: +* ``worker_init`` loads the ML pipeline and exits the process if any artifact is + missing (hard-fail policy, DECISIONS.md) β€” before consuming any message. +* ``worker_ready`` starts the outbox dispatcher thread (:mod:`app.worker.outbox`). +""" + +from __future__ import annotations + +import logging +from typing import Any + +from celery import Celery +from celery.signals import worker_init, worker_ready + +from app.config import settings + +logger = logging.getLogger(__name__) + +celery_app = Celery("vera", broker=settings.broker_url, include=["app.worker.tasks"]) +celery_app.conf.update( + task_serializer="json", + accept_content=["json"], + timezone="UTC", + enable_utc=True, + task_ignore_result=True, + broker_transport_options={"visibility_timeout": 3600, "polling_interval": 2.0}, + broker_connection_timeout=3, + broker_connection_retry_on_startup=True, + worker_prefetch_multiplier=1, + task_acks_late=True, + task_reject_on_worker_lost=True, + # Serialize inference so this worker does not duplicate the loaded FP32 + # models. Deployment memory must be sized and verified separately; the + # free-tier 512 MB target is below the documented production requirement. + worker_pool="solo", + worker_concurrency=1, +) + + +@worker_init.connect +def load_models_or_exit(**_: Any) -> None: + """Fail fast: a worker without valid artifacts must not consume jobs.""" + from app.worker.tasks import get_pipeline + + try: + get_pipeline() + except Exception as exc: + logger.critical("VERA model initialisation failed: %s", exc) + # SystemExit (a BaseException) escapes Celery's signal error handling. + raise SystemExit(1) from exc + + +@worker_ready.connect +def start_outbox(**_: Any) -> None: + """Run the outbox dispatcher alongside the worker (no extra process needed).""" + from app.worker.outbox import start_background_dispatcher + + start_background_dispatcher() + logger.info("Outbox dispatcher thread started") diff --git a/backend/app/worker/outbox.py b/backend/app/worker/outbox.py new file mode 100644 index 0000000000000000000000000000000000000000..cd0438d8e01d5de08170a7be950daa54ceb42d15 --- /dev/null +++ b/backend/app/worker/outbox.py @@ -0,0 +1,93 @@ +"""Outbox dispatcher: turns unfinished `jobs` rows into Celery wake-ups. + +Runs as a daemon thread inside the worker (started on ``worker_ready`` in +:mod:`app.worker.celery_app`), so a free-tier deployment needs no extra process. +It can also run standalone:: + + cd backend && python -m app.worker.outbox + +A job is (re)sent when it is unfinished, due (``available_at <= now``), and was +never dispatched or was dispatched more than :data:`REDISPATCH_AFTER` ago β€” this +recovers messages lost to a Redis outage or a crashed worker. Importing this +module never loads ML models. +""" + +from __future__ import annotations + +import contextlib +import logging +import threading +from datetime import UTC, datetime, timedelta +from typing import Final + +from sqlalchemy import or_ + +from app.database import SessionLocal +from app.models.job import Job + +logger = logging.getLogger(__name__) + +POLL_INTERVAL_SECONDS: Final = 5.0 +REDISPATCH_AFTER: Final = timedelta(seconds=60) +RUN_JOB_TASK: Final = "vera.run_job" + + +def send_wakeup(job_id: str) -> None: + """Publish one wake-up message. Raises if the broker is unreachable.""" + from app.worker.celery_app import celery_app # lazy: avoid import cycle + + celery_app.send_task(RUN_JOB_TASK, args=[job_id], retry=False) + + +def dispatch_pending(limit: int = 100) -> int: + """Send wake-ups for due jobs. Returns the number sent.""" + now = datetime.now(UTC) + sent = 0 + with SessionLocal() as db: + jobs = ( + db.query(Job) + .filter( + Job.finished_at.is_(None), + Job.available_at <= now, + or_( + Job.last_dispatched_at.is_(None), + Job.last_dispatched_at < now - REDISPATCH_AFTER, + ), + ) + .order_by(Job.available_at, Job.id) + .with_for_update(skip_locked=True) + .limit(limit) + .all() + ) + for job in jobs: + send_wakeup(job.id) + job.last_dispatched_at = now + sent += 1 + db.commit() + return sent + + +def run_forever(stop: threading.Event | None = None) -> None: + """Poll until ``stop`` is set. Errors are logged; the loop never dies.""" + stop = stop or threading.Event() + while not stop.is_set(): + try: + count = dispatch_pending() + if count: + logger.info("Outbox dispatched %d job(s)", count) + except Exception: + logger.exception("Outbox dispatch failed; will retry") + stop.wait(POLL_INTERVAL_SECONDS) + + +def start_background_dispatcher() -> threading.Event: + """Start :func:`run_forever` in a daemon thread; set the returned event to stop it.""" + stop = threading.Event() + threading.Thread(target=run_forever, args=(stop,), name="vera-outbox", daemon=True).start() + return stop + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + with contextlib.suppress(KeyboardInterrupt): + run_forever() diff --git a/backend/app/worker/tasks.py b/backend/app/worker/tasks.py new file mode 100644 index 0000000000000000000000000000000000000000..6fd6f58667138de062fb94dd52f48b80caa503a5 --- /dev/null +++ b/backend/app/worker/tasks.py @@ -0,0 +1,238 @@ +"""Celery task: execute one outbox job (review inference or product aggregation). + +Delivery model (see :mod:`app.models.job`): Redis messages are only wake-ups +and may be duplicated or lost; the ``jobs`` row is the source of truth. Every +run locks its job row, skips finished jobs, and records retries durably with +exponential back-off. :mod:`app.worker.outbox` re-sends anything unfinished. + +ML models load lazily through :func:`get_pipeline`, so importing this module +(e.g. from tests) never touches model artifacts. +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime, timedelta +from functools import lru_cache +from typing import TYPE_CHECKING, Any, Final + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.config import settings +from app.database import SessionLocal +from app.models.job import Job, JobKind +from app.models.product import Product +from app.models.review import Review, ReviewStatus +from app.worker.celery_app import celery_app + +if TYPE_CHECKING: + from app.ml.pipeline import VERAPipeline + +logger = logging.getLogger(__name__) + +MAX_REVIEW_ATTEMPTS: Final = 4 +MAX_AGGREGATE_ATTEMPTS: Final = 10 +MAX_BACKOFF_SECONDS: Final = 300 +#: Arbitrary constant key for the Postgres advisory lock around aggregation. +AGGREGATION_LOCK_KEY: Final = 19842026 +FAILED_REVIEW_MESSAGE: Final = "Review analysis failed. Please try again later." + + +@lru_cache(maxsize=1) +def get_pipeline() -> VERAPipeline: + """Load production artifacts once per worker process. Never falls back.""" + from app.ml.pipeline import VERAPipeline + + return VERAPipeline(settings.MODEL_DIR, settings.HF_MODEL_REPO, settings.HF_TOKEN) + + +def _as_utc(value: datetime) -> datetime: + """Normalise DB datetimes: SQLite returns naive values, Postgres aware ones.""" + return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) + + +def backoff_seconds(attempts: int) -> int: + """5 s, 10 s, 20 s, … capped at :data:`MAX_BACKOFF_SECONDS`.""" + return min(MAX_BACKOFF_SECONDS, 5 * 2 ** min(attempts, 6)) + + +# ---------------------------------------------------------------------------- +# Job handlers β€” run inside the caller's transaction; the caller commits. +# ---------------------------------------------------------------------------- +def process_review(db: Session, review_id: int) -> None: + """Run the pipeline for one review and store its result. + + The review row stays locked for the whole inference, so a crash rolls back + cleanly and a duplicate delivery cannot write concurrently. The status is not + committed as "processing" separately; clients see "pending" until "done". + """ + review = db.query(Review).filter(Review.id == review_id).with_for_update().one_or_none() + if review is None or review.status in (ReviewStatus.DONE.value, ReviewStatus.FAILED.value): + return + + result = get_pipeline().process( + text=review.text, + star_rating=review.star_rating, + helpful_votes=review.helpful_votes, + verified_purchase=review.verified_purchase, + product_title=review.product_title, + ) + review.corrected_score = result["corrected_score"] + review.interval_lower, review.interval_upper = result["interval"] + review.confidence_value = result["confidence"] + review.star_text_gap = result["star_text_gap"] + review.review_quality = result["review_quality"] + review.aspects = result["aspects"] + review.flags = result["flags"] + review.explanation = result["explanation"] + review.error_message = None + review.status = ReviewStatus.DONE.value + + # Reviews without an ASIN never create aggregation work (PRD weakness #3). + if review.product_asin: + db.add(Job(kind=JobKind.AGGREGATE.value)) + + +def aggregate_products(db: Session, current_job_id: str | None = None) -> int: + """Recompute EB aggregates and ranks for ALL products. Returns product count. + + The EB prior is a population quantity, so every product is refreshed under + one lock (Postgres advisory transaction lock β€” released on commit/rollback, + no TTL to expire mid-run). Other unfinished aggregate jobs visible *before* + the review snapshot is read are marked finished: their reviews are already + committed and therefore included, so re-running them would be pure waste. + """ + from app.ml.calibration.empirical_bayes import ( + EmpiricalBayesAggregator, + ReviewSignal, + rank_products, + ) + + if db.get_bind().dialect.name == "postgresql": + db.execute(text("SELECT pg_advisory_xact_lock(:key)"), {"key": AGGREGATION_LOCK_KEY}) + + superseded = ( + db.query(Job) + .filter( + Job.kind == JobKind.AGGREGATE.value, + Job.finished_at.is_(None), + Job.id != (current_job_id or ""), + ) + .with_for_update(skip_locked=True) + .all() + ) + + rows = ( + db.query( + Review.product_asin, + Review.corrected_score, + Review.review_quality, + Review.star_rating, + Review.aspects, + ) + .filter( + Review.status == ReviewStatus.DONE.value, + Review.product_asin.is_not(None), + Review.corrected_score.is_not(None), + ) + .all() + ) + now = datetime.now(UTC) + for job in superseded: + job.finished_at = now + + signals = [ + ReviewSignal( + product_asin=row.product_asin, + corrected_score=float(row.corrected_score), + quality_weight=float(row.review_quality) if row.review_quality is not None else 0.5, + star_rating=float(row.star_rating), + aspects=row.aspects or {}, + ) + for row in rows + ] + if not signals: + return 0 + + _, aggregates = EmpiricalBayesAggregator().aggregate_all(signals) + for rank, asin in enumerate(rank_products(aggregates), start=1): + aggregate = aggregates[asin] + product = db.get(Product, asin) + if product is None: # FK makes this impossible unless data was edited by hand + raise RuntimeError(f"Completed review references missing product {asin!r}") + product.previous_rank_position = product.rank_position + product.rank_position = rank + product.corrected_score = aggregate.corrected_score + product.uncertainty = aggregate.uncertainty + product.trust_weighted_score = aggregate.trust_weighted_score + product.star_average = aggregate.star_average + product.eb_shrinkage_factor = aggregate.eb_shrinkage_factor + product.review_count = aggregate.review_count + product.aspect_summary = aggregate.aspect_summary + product.last_aggregated_at = now + return len(aggregates) + + +# ---------------------------------------------------------------------------- +# Celery entry point +# ---------------------------------------------------------------------------- +def _record_failure(db: Session, job_id: str) -> None: + """Increment attempts; schedule a retry or mark the job (and review) failed.""" + job = db.query(Job).filter(Job.id == job_id).with_for_update().one_or_none() + if job is None or job.finished_at is not None: + return # a duplicate delivery finished it meanwhile + + job.attempts += 1 + limit = MAX_REVIEW_ATTEMPTS if job.kind == JobKind.REVIEW.value else MAX_AGGREGATE_ATTEMPTS + now = datetime.now(UTC) + if job.attempts >= limit: + job.finished_at = now + if job.kind == JobKind.REVIEW.value and job.review_id is not None: + review = db.get(Review, job.review_id) + if review is not None and review.status != ReviewStatus.DONE.value: + review.status = ReviewStatus.FAILED.value + review.error_message = FAILED_REVIEW_MESSAGE + logger.error("Job %s (%s) failed permanently after %d attempts", job.id, job.kind, limit) + else: + job.available_at = now + timedelta(seconds=backoff_seconds(job.attempts)) + job.last_dispatched_at = None # let the outbox re-send once due + db.commit() + + +def execute_job(job_id: str) -> str: + """Run one job to completion or record its failure. Returns an outcome label.""" + with SessionLocal() as db: + try: + job = ( + db.query(Job) + .filter(Job.id == job_id) + .with_for_update(skip_locked=True) + .one_or_none() + ) + if job is None or job.finished_at is not None: + return "ignored" # unknown, finished, or locked by a concurrent delivery + if _as_utc(job.available_at) > datetime.now(UTC): + return "deferred" + + if job.kind == JobKind.REVIEW.value and job.review_id is not None: + process_review(db, job.review_id) + elif job.kind == JobKind.AGGREGATE.value: + aggregate_products(db, current_job_id=job.id) + else: + raise ValueError(f"Unknown job kind {job.kind!r}") + + job.finished_at = datetime.now(UTC) + db.commit() + return "done" + except Exception: + db.rollback() + logger.exception("Job %s failed; recording durable retry", job_id) + _record_failure(db, job_id) + return "retry_or_failed" + + +@celery_app.task(name="vera.run_job", ignore_result=True) +def run_job(job_id: str) -> dict[str, Any]: + """Celery wrapper around :func:`execute_job`.""" + return {"job_id": job_id, "status": execute_job(job_id)} diff --git a/backend/migrations/env.py b/backend/migrations/env.py new file mode 100644 index 0000000000000000000000000000000000000000..fdcd7f39bd04768ec4e0e700fb55afc926b14339 --- /dev/null +++ b/backend/migrations/env.py @@ -0,0 +1,52 @@ +"""Alembic environment configured from VERA's runtime settings.""" + +from __future__ import annotations + +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import engine_from_config, pool + +import app.models # noqa: F401 # Register all ORM metadata before autogeneration. +from app.config import settings +from app.database import Base + +config = context.config +# ConfigParser treats "%" as interpolation; URL-encoded passwords contain it. +config.set_main_option("sqlalchemy.url", settings.DATABASE_URL.replace("%", "%%")) +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = Base.metadata + + +def run_migrations_offline() -> None: + """Run migrations without an Engine, e.g. to generate SQL.""" + context.configure( + url=settings.DATABASE_URL, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + compare_type=True, + ) + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + """Run migrations against the configured database.""" + connectable = engine_from_config( + config.get_section(config.config_ini_section, {}), + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata, compare_type=True) + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/backend/migrations/script.py.mako b/backend/migrations/script.py.mako new file mode 100644 index 0000000000000000000000000000000000000000..8c1224053a0c87ae2dba394d777061cae847cf24 --- /dev/null +++ b/backend/migrations/script.py.mako @@ -0,0 +1,24 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = ${repr(up_revision)} +down_revision = ${repr(down_revision)} +branch_labels = ${repr(branch_labels)} +depends_on = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/backend/migrations/versions/001_initial.py b/backend/migrations/versions/001_initial.py new file mode 100644 index 0000000000000000000000000000000000000000..31d03161a84c62dfbf36302140270d70747e9029 --- /dev/null +++ b/backend/migrations/versions/001_initial.py @@ -0,0 +1,119 @@ +"""Create the VERA application schema. + +Revision ID: 001_initial +Revises: +Create Date: 2026-09-13 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision = "001_initial" +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade() -> None: + """Create products first because reviews references its ASIN.""" + op.create_table( + "products", + sa.Column("asin", sa.String(length=20), primary_key=True), + sa.Column("title", sa.Text()), + sa.Column("corrected_score", sa.Float()), + sa.Column("uncertainty", sa.Float()), + sa.Column("trust_weighted_score", sa.Float()), + sa.Column("star_average", sa.Float()), + sa.Column("eb_shrinkage_factor", sa.Float()), + sa.Column("review_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("aspect_summary", sa.JSON()), + sa.Column("rank_position", sa.Integer()), + sa.Column("previous_rank_position", sa.Integer()), + sa.Column("last_aggregated_at", sa.DateTime(timezone=True)), + sa.Column( + "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False + ), + ) + op.create_index("idx_products_rank", "products", ["rank_position"]) + + op.create_table( + "reviews", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("text", sa.Text(), nullable=False), + sa.Column("star_rating", sa.Float(), nullable=False), + sa.Column("helpful_votes", sa.Integer()), + sa.Column("verified_purchase", sa.Boolean()), + sa.Column("product_title", sa.Text()), + sa.Column( + "product_asin", + sa.String(length=20), + sa.ForeignKey("products.asin", ondelete="SET NULL"), + ), + sa.Column("corrected_score", sa.Float()), + sa.Column("interval_lower", sa.Float()), + sa.Column("interval_upper", sa.Float()), + sa.Column("confidence_value", sa.Float()), + sa.Column("star_text_gap", sa.Float()), + sa.Column("review_quality", sa.Float()), + sa.Column("aspects", sa.JSON()), + sa.Column("flags", sa.JSON()), + sa.Column("explanation", sa.Text()), + sa.Column("error_message", sa.Text()), + sa.Column("status", sa.String(length=20), nullable=False, server_default="pending"), + sa.Column( + "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False + ), + sa.Column( + "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False + ), + sa.CheckConstraint("star_rating >= 1 AND star_rating <= 5", name="ck_reviews_star_range"), + sa.CheckConstraint( + "status IN ('pending', 'processing', 'done', 'failed')", name="ck_reviews_status" + ), + ) + op.create_index("idx_reviews_asin", "reviews", ["product_asin"]) + op.create_index("idx_reviews_status", "reviews", ["status"]) + + op.create_table( + "label_runs", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("review_text_hash", sa.String(length=64), nullable=False), + sa.Column("labeler", sa.String(length=100), nullable=False), + sa.Column("experience_score", sa.Float()), + sa.Column("aspects", sa.JSON()), + sa.Column("has_irony", sa.Boolean()), + sa.Column("review_quality", sa.String(length=10)), + sa.Column("raw_response", sa.JSON()), + sa.Column( + "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False + ), + ) + op.create_index("idx_label_runs_hash", "label_runs", ["review_text_hash"]) + + op.create_table( + "calibration_checkpoints", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("method", sa.String(length=20), nullable=False, server_default="absolute"), + sa.Column("q_hat", sa.Float(), nullable=False), + sa.Column("empirical_coverage", sa.Float(), nullable=False), + sa.Column("nominal_coverage", sa.Float(), nullable=False), + sa.Column("calibration_set_size", sa.Integer(), nullable=False), + sa.Column( + "fitted_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False + ), + ) + + +def downgrade() -> None: + """Drop in dependency order.""" + op.drop_table("calibration_checkpoints") + op.drop_index("idx_label_runs_hash", table_name="label_runs") + op.drop_table("label_runs") + op.drop_index("idx_reviews_status", table_name="reviews") + op.drop_index("idx_reviews_asin", table_name="reviews") + op.drop_table("reviews") + op.drop_index("idx_products_rank", table_name="products") + op.drop_table("products") diff --git a/backend/migrations/versions/002_jobs_outbox.py b/backend/migrations/versions/002_jobs_outbox.py new file mode 100644 index 0000000000000000000000000000000000000000..4eec233b9eb2b37f997e173ff75cdeb4f2a26ae0 --- /dev/null +++ b/backend/migrations/versions/002_jobs_outbox.py @@ -0,0 +1,48 @@ +"""Add the `jobs` transactional-outbox table. + +Revision ID: 002_jobs_outbox +Revises: 001_initial +Create Date: 2026-09-14 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "002_jobs_outbox" +down_revision = "001_initial" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "jobs", + sa.Column("id", sa.String(length=36), primary_key=True), + sa.Column("kind", sa.String(length=20), nullable=False), + sa.Column( + "review_id", + sa.Integer(), + sa.ForeignKey("reviews.id", ondelete="CASCADE"), + unique=True, + ), + sa.Column("attempts", sa.Integer(), nullable=False, server_default="0"), + sa.Column( + "available_at", + sa.DateTime(timezone=True), + server_default=sa.func.now(), + nullable=False, + ), + sa.Column("last_dispatched_at", sa.DateTime(timezone=True)), + sa.Column("finished_at", sa.DateTime(timezone=True)), + sa.Column( + "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False + ), + ) + op.create_index("idx_jobs_dispatch", "jobs", ["finished_at", "available_at"]) + + +def downgrade() -> None: + op.drop_index("idx_jobs_dispatch", table_name="jobs") + op.drop_table("jobs") diff --git a/packages.txt b/packages.txt new file mode 100644 index 0000000000000000000000000000000000000000..20a006aa2658d432438a7106207b80961efbdcc7 --- /dev/null +++ b/packages.txt @@ -0,0 +1 @@ +redis-server diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000000000000000000000000000000000000..824e2591eda23c209c11328cde68e908720e49aa --- /dev/null +++ b/requirements.txt @@ -0,0 +1,26 @@ +# Hugging Face Gradio Space (ZeroGPU hardware) runtime for the VERA API + worker. +# The Space image preinstalls gradio (sdk_version in README.md) and `spaces`; their web stack +# (fastapi, starlette, pydantic, httpx, uvicorn, huggingface_hub) is left unpinned here so pip +# resolves versions compatible with gradio. ML libraries stay pinned: the artifacts were built +# with them (scikit-learn version guard, xgboost JSON format, ONNX opset, tokenizers JSON). +onnxruntime==1.19.0 +tokenizers==0.20.0 +scikit-learn==1.5.2 +xgboost==2.1.1 +joblib==1.4.2 +numpy==1.26.4 +pandas==2.2.3 +pyarrow==17.0.0 +langdetect==1.0.9 +huggingface_hub>=0.28.1 +fastapi +uvicorn[standard] +celery==5.4.0 +redis==5.1.1 +sqlalchemy==2.0.35 +alembic==1.13.3 +psycopg2-binary==2.9.9 +pydantic>=2.9,<2.12 +pydantic-settings>=2.5 +python-dotenv +httpx