diff --git a/.dockerignore b/.dockerignore index 67537ac..8b96602 100644 --- a/.dockerignore +++ b/.dockerignore @@ -24,6 +24,10 @@ mlruns/ models/*.pt models/*.pth models/*.ckpt +models/ +data/ +benchmarks/ +scripts/ .git .github tests/ diff --git a/.env.example b/.env.example index 922fdae..04a0b31 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,8 @@ -SECRET_KEY=change-this-to-a-long-random-secret -REDIS_URL=redis://localhost:6379 -MLFLOW_TRACKING_URI=http://localhost:5000 -ENVIRONMENT=development +API_KEY=replace-with-a-long-random-service-key +ENVIRONMENT=production +MODEL_BUNDLE_PATH=models/current +MAX_INFERENCE_MS=250 +MAX_CONCURRENT_INFERENCES=8 +CACHE_TTL_SECONDS=30 +REQUESTS_PER_MINUTE=120 LOG_LEVEL=INFO diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml index 275490c..af8bdfd 100644 --- a/.github/workflows/benchmarks.yml +++ b/.github/workflows/benchmarks.yml @@ -24,8 +24,14 @@ jobs: run: | python -m pip install --upgrade pip if [ -f requirements.txt ]; then pip install -r requirements.txt; fi - pip install pytest pytest-benchmark + pip install pytest - name: โฑ๏ธ Profile Code Execution Speeds run: | - pytest --benchmark-only --benchmark-skip || echo "Performance profiles logged successfully." + python -m benchmarks.inference --iterations 200 --warmup 20 > benchmark-results.json + + - name: ๐Ÿ“Ž Upload benchmark evidence + uses: actions/upload-artifact@v4 + with: + name: inference-benchmark + path: benchmark-results.json diff --git a/.github/workflows/ci-cd.yml b/.github/workflows/ci-cd.yml index 47c4ac6..1d73136 100644 --- a/.github/workflows/ci-cd.yml +++ b/.github/workflows/ci-cd.yml @@ -29,4 +29,41 @@ jobs: - name: ๐Ÿงช Execute Recommender Test Suite run: | - pytest --cov=. --cov-report=term-missing tests/ || echo "โš ๏ธ Core Engine: Complete test coverage tracking." + pytest --cov=app --cov=src --cov-report=term-missing --cov-report=xml --cov-fail-under=85 tests/ + + - name: ๐Ÿ“Ž Upload test evidence + uses: actions/upload-artifact@v4 + with: + name: test-evidence + path: coverage.xml + + container-smoke: + name: Verified Bundle Container Smoke Test + runs-on: ubuntu-latest + needs: engine-test + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + cache: pip + - name: Build a pipeline-validation bundle + run: | + pip install -r requirements.txt + python -m scripts.generate_demo_data --output data/ci-events.jsonl + python -m src.training.train --dataset data/ci-events.jsonl --output models/current --epochs 1 --top-k 5 + - name: Build and verify the production image + run: | + docker build -t deepsequence:${{ github.sha }} . + docker run -d --rm --name deepsequence-ci --read-only --tmpfs /tmp \ + -e API_KEY=ci-test-key \ + -v "${PWD}/models/current:/models/current:ro" \ + -p 8000:8000 deepsequence:${{ github.sha }} + trap 'docker logs deepsequence-ci || true; docker stop deepsequence-ci >/dev/null 2>&1 || true' EXIT + for attempt in {1..30}; do + if curl --fail --silent http://127.0.0.1:8000/recommendations/health | grep '"trained_model":true'; then + exit 0 + fi + sleep 2 + done + exit 1 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d1c2b37..399f176 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,8 +24,8 @@ jobs: - name: ๐Ÿ“ฆ Provision Linting Binaries run: | python -m pip install --upgrade pip - pip install flake8 + pip install ruff # FIXED: Added F821 and W292 to ensure type hints or missing file newlines don't break the build - name: ๐Ÿš€ Verify PEP 8 Alignment - run: flake8 . --ignore=W293,W291,F401,E302,E305,E701,E203,E501,E261,W292,F821 --exclude=venv,.git,__pycache__,build,dist + run: ruff check app src benchmarks scripts tests streamlit_app.py diff --git a/.github/workflows/data-validation.yml b/.github/workflows/data-validation.yml index 236a2bc..34d775e 100644 --- a/.github/workflows/data-validation.yml +++ b/.github/workflows/data-validation.yml @@ -25,8 +25,8 @@ jobs: - name: ๐Ÿ“ฆ Install Data Integrity Frameworks run: | python -m pip install --upgrade pip - pip install pydantic great_expectations pytest + pip install pydantic pytest - name: ๐Ÿงฌ Audit Ingestion Model Integrity run: | - pytest tests/test_schema_contracts.py || pytest tests/ || echo "โš ๏ธ Ingestion Schema: Skipping data validation pass." + pytest tests/test_schema_contracts.py diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml index 6776c9c..ed8be9a 100644 --- a/.github/workflows/security.yml +++ b/.github/workflows/security.yml @@ -29,5 +29,5 @@ jobs: - name: ๐Ÿ“ฆ Evaluate Package Insecurities run: | - pip install safety - safety check || echo "โš ๏ธ Security Alert: Review reported package vulnerabilities." + pip install pip-audit + pip-audit -r requirements.txt diff --git a/.gitignore b/.gitignore index 5aeacb5..89237cd 100644 --- a/.gitignore +++ b/.gitignore @@ -24,3 +24,8 @@ mlruns/ models/*.pt models/*.pth models/*.ckpt +models/ +data/ +testtmp/ +benchmark-results.json +coverage.xml diff --git a/Dockerfile b/Dockerfile index 498099a..099afa8 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,23 +1,30 @@ -FROM python:3.11-slim AS base - -WORKDIR /app - -# Install OS-level dependencies -RUN apt-get update && apt-get install -y --no-install-recommends gcc && \ - rm -rf /var/lib/apt/lists/* - +FROM python:3.11-slim AS builder +ENV PIP_DISABLE_PIP_VERSION_CHECK=1 PIP_NO_CACHE_DIR=1 +WORKDIR /build COPY requirements.txt . -RUN pip install --no-cache-dir -r requirements.txt - -# ---- Runtime stage ---- -FROM base AS runtime - -COPY app/ app/ - -# Non-root user for security -RUN useradd -m appuser -USER appuser +RUN python -m venv /opt/venv \ + && /opt/venv/bin/pip install --upgrade pip \ + && /opt/venv/bin/pip install -r requirements.txt + +FROM python:3.11-slim AS runtime +ENV PATH=/opt/venv/bin:$PATH \ + PYTHONPATH=/app \ + PYTHONUNBUFFERED=1 \ + PYTHONDONTWRITEBYTECODE=1 \ + ENVIRONMENT=production \ + MODEL_BUNDLE_PATH=/models/current + +RUN useradd --create-home --uid 10001 appuser \ + && mkdir -p /models/current \ + && chown -R appuser:appuser /models +WORKDIR /app +COPY --from=builder /opt/venv /opt/venv +COPY --chown=appuser:appuser app app +COPY --chown=appuser:appuser src src +USER 10001 EXPOSE 8000 +HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \ + CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/recommendations/health', timeout=3)" -CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--no-access-log"] diff --git a/README.md b/README.md index 0f61996..7419b46 100644 --- a/README.md +++ b/README.md @@ -1,137 +1,272 @@ -# ๐Ÿ›’ DeepSequence-Recommender: Deep Sequential Recommendation Engine +# DeepSequence Recommender -[![Continuous Integration](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/ci-cd.yml/badge.svg)](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/ci-cd.yml) -[![Code Quality Assurance](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/ci.yml/badge.svg)](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/ci.yml) -[![Security Analysis](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/security.yml/badge.svg)](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/security.yml) -[![SAST Code Flaw Scan](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/sast.yml/badge.svg)](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/sast.yml) -[![Performance Benchmarks](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/benchmarks.yml/badge.svg)](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/benchmarks.yml) -[![Schema Validation](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/data-validation.yml/badge.svg)](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/data-validation.yml) -[![Automated Release](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/release.yml/badge.svg)](https://github.com/Trojan3877/DeepSequence-Recommender/actions/workflows/release.yml) +

A versioned, evaluated sequential recommender with a reproducible training-to-serving contract.

-[![Python 3.11](https://img.shields.io/badge/python-3.11-blue.svg)](https://www.python.org/downloads/release/python-3110/) -[![Framework: PyTorch](https://img.shields.io/badge/Framework-PyTorch-ee4c2c.svg?logo=pytorch)](https://pytorch.org/) -[![Code Style: Flake8](https://img.shields.io/badge/code%20style-flake8-black)](https://flake8.pycqa.org/) -[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) +

+ CI + Quality + Security + Benchmarks + Data contracts +

---- +

+ Python 3.11 + PyTorch + FastAPI + Checksummed bundles + Ranking metrics + Non-root container + License +

+DeepSequence Recommender learns next-item behavior from timestamped interactions. One connected +path now covers schema validation, chronological splits, training, baseline comparison, model +packaging, checksum verification, serving, feedback capture, and monitoring. -An enterprise-grade, high-concurrency sequential recommendation system engineered for sub-50ms low-latency next-item inference scoring. Moving past simple offline batch notebooks, this platform implements **real-time session sequence slicing**, an **active SLA latency circuit breaker**, and decoupled **Immutable Inference Context Objects** to guarantee operational stability under intense clickstream traffic. -##System Architecture & Data Flow -DeepSequence-Recommender decouples real-time sequence orchestration from core model matrix operations, preventing memory overflow (OOM) and latency spikes during deep user sessions. -### Deterministic Inference Lifecycle Flow -The streaming pipeline processes real-time user engagement events through an isolated, bounded lifecycle to enforce strict system resource allocations: +The repository does **not** ship a trained production model or claim production availability. +Production fails closed unless a verified bundle is mounted. Development uses a deterministic +untrained model so API integration can be tested without presenting random ranking as AI quality. + +## Engineering features + +- Padding-aware attention, aligned item/output IDs, excluded padding class, and bounded top-k. +- Versioned events, temporal splits, causal targets, deterministic seeds, and popularity baseline. +- Recall@K, NDCG@K, MRR@K, catalogue coverage, and machine-readable evaluation reports. +- Immutable bundles containing weights, vocabulary, architecture, lineage, metrics, and checksums. +- Fail-closed production startup for absent, corrupted, or incompatible artifacts. +- API-key option, rate limiting, admission control, latency fallback, and version-aware caching. +- Privacy-minimized feedback events for impressions, clicks, skips, carts, purchases, and dislikes. +- Prometheus inference, request, cache, fallback, saturation, and feedback signals. +- Non-root/read-only container and hardened Kubernetes security/resource configuration. +- CI that cannot hide test, schema, audit, or benchmark failures with shell fallbacks. + +## Architecture + +```text +Events -> contract -> chronological split -> causal examples + | | + | popularity baseline + v v +PyTorch training --------------------> ranking evaluation + | + v +manifest + vocabulary + weights + metrics + checksums + | + v +startup verification -> API key -> rate limit -> cache -> admission -> model + | | + +-> fallback + | + Prometheus + feedback events +``` + +## Quick start + +```bash +git clone https://github.com/CoreyLeath-code/DeepSequence-Recommender.git +cd DeepSequence-Recommender +python -m venv .venv +source .venv/bin/activate # Windows: .venv\Scripts\activate +pip install -r requirements.txt +pip install pytest pytest-cov ruff +pytest -q +ENVIRONMENT=development uvicorn app.main:app --reload +``` + +Development responses identify the model as `development-untrained`. + +## Train, evaluate, and package + +```bash +python -m scripts.generate_demo_data --output data/demo-events.jsonl +python -m src.training.train \ + --dataset data/demo-events.jsonl \ + --output models/candidate \ + --epochs 3 \ + --top-k 10 \ + --seed 7 +``` + +The candidate contains: + +```text +manifest.json version, architecture, dataset lineage, metrics, checksums +model.pt PyTorch state dictionary +vocabulary.json exact training/serving item mapping +evaluation.json neural and popularity-baseline results +``` + +Demo data validates the pipeline only. Use a leakage-reviewed production snapshot before making +ranking claims. Promote the complete directory through the model registry; never mix files. + +Export the exact verified modelโ€”not a mock architectureโ€”to ONNX: + +```bash +python -m src.serving.onnx_exporter --bundle models/current --output models/model.onnx ``` - [Raw User Clickstream Event] - โ”‚ - โ–ผ - โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” - โ”‚ Recommender Guardrail Layer โ”‚ โ”€โ”€โ–บ Slidewindow Truncation (O(1) Memory Bound) - โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ - โ”‚ - โ–ผ - โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” - โ”‚ Inference State Serialization โ”‚ โ”€โ”€โ–บ Compiles strict, immutable Pydantic Tensors - โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ - โ”‚ - โ–ผ - โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” - โ”‚ SLA Millisecond Monitoring โ”‚ โ”€โ”€โ–บ Evaluates active duration constraints - โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ - โ”‚ - โ–ผ - โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” - โ”‚ Sequential Inference Engine โ”‚ โ”€โ”€โ–บ Generates Top-K recommended tensor tokens - โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ - โ”‚ - โ–ผ - [Optimized Real-Time Predictions] โ”€โ”€โ–บ Falls back to Global-Popular if SLA is breached - - - 1. **Sliding-Window Constraining:** User action lists are truncated to a strict maximum sliding size (N=10), capping tensor dimensions and preventing unpredictable long-tail matrix inflation. - 2. **Deterministic Processing Execution:** Real-time metrics track execution speed. If network latency or heavy model calculations cross the 50\text{ms} threshold, an operational circuit breaker intercepts processing. - 3. **Graceful Performance Degradation:** If an exceptional compute delay occurs, the system avoids throwing a 500 Internal Server Error by instantly routing traffic to a local, low-latency cache of globally popular fallbacks to preserve top-tier application uptime. -Operational Architecture Benchmarks -Decoupling state instantiation from model execution delivers clear improvements over traditional un-bounded session scoring systems: -| Operational Metric | Standard RNN/Transformer Execution | Upgraded Bounded Inference Engine | System Impact | -|---|---|---|---| -| **p99 Inference Latency** | Variable (120\text{ms} - 850\text{ms}) | Stable (15\text{ms} - 42\text{ms}) | **95.0% Latency Compression** | -| **Throughput Density** | ~1,400 requests/sec | ~12,000 requests/sec via Triton/ONNX | **+757% High-Concurrency Load** | -| **Memory Allocation Profile** | O(N) linear explosion based on history | O(1) deterministic cap via Guardrails | **Eliminated Session OOM Vulnerability** | -| **Fallback Resiliency** | System Timeout / Cascade Drop | Instant Automatic Cache Triage | **99.99% Availability Guarantee** | - -## ๐Ÿ“Š Metrics Table - -Measured from the repository on 2026-07-12. Runtime and SLO values are recorded from the checked-in configuration, metrics documentation, and README architecture notes. - -| Area | Metric | Current / Recommended Value | Source | -|---|---:|---:|---| -| Codebase | Tracked files | 43 | `git ls-files` | -| Codebase | Python files | 22 | `*.py` files across app, recommender, docs, and tests | -| Codebase | Python NCLOC | 995 | Non-empty, non-comment Python lines | -| Tests | Test files | 2 | `tests/test_*.py` | -| Tests | Pytest cases | 27 passing | `pytest --cov=app --cov=recommender` | -| Tests | Coverage scope | `app`, `recommender` | `pytest-cov` | -| Tests | Combined coverage | 54% | Local coverage run | -| CI/CD | GitHub Actions workflows | 7 | `.github/workflows/*.yml` | -| Dependencies | Runtime dependencies | 12 | `requirements.txt` | -| Delivery | Container assets | 2 | `Dockerfile`, `docker-compose.yml` | -| Delivery | Kubernetes manifests | 3 | `k8s/*.yaml` | -| Documentation | Documentation pages | 4 | `README.md`, `docs/*.md` | -| Model Config | Max sequence length | 50 items | `app.core.config.Settings.max_sequence_length` | -| Model Config | Default top-k recommendations | 10 items | `app.core.config.Settings.top_k` | -| Model Config | Embedding dimension | 64 | `app.core.config.Settings.embedding_dim` | -| Model Config | Hidden dimension | 128 | `app.core.config.Settings.hidden_dim` | -| Model Config | Recurrent layers | 2 | `app.core.config.Settings.num_layers` | -| Model Architecture | Encoder | Bidirectional LSTM + attention | `app.core.model.DeepSequenceModel` | -| Model Architecture | Padding index | 0 | `SequenceProcessor` / `DeepSequenceModel` | -| API | Recommendation endpoint | `POST /recommendations/` | `app.api.routes` | -| API | Health endpoint | `GET /recommendations/health` | `app.api.routes` | -| Observability | Active requests | `deepseq_active_requests` | Prometheus gauge | -| Observability | Recommendation total | `deepseq_recommendations_total{status}` | Prometheus counter | -| Observability | Request latency | `deepseq_recommendation_latency_seconds` | Prometheus histogram | -| Observability | Model latency | `deepseq_model_inference_latency_seconds` | Prometheus histogram | -| Observability | Cache hits | `deepseq_cache_hits_total` | Prometheus counter | -| Observability | Cache misses | `deepseq_cache_misses_total` | Prometheus counter | -| SLO | p50 recommendation latency | < 50 ms | `docs/metrics.md` | -| SLO | p99 recommendation latency | < 250 ms | `docs/metrics.md` | -| SLO | Model inference p50 latency | < 20 ms | `docs/metrics.md` | -| SLO | Availability target | >= 99.9% | `docs/metrics.md` | -| SLO | Error-rate target | < 0.1% | `docs/metrics.md` | -| Architecture Benchmark | Documented p99 latency range | 15-42 ms bounded engine | README benchmark table | -| Architecture Benchmark | Documented throughput density | ~12,000 requests/sec | README benchmark table | -| Architecture Benchmark | Documented memory profile | O(1) deterministic cap | README benchmark table | -| Architecture Benchmark | Documented fallback availability | 99.99% availability guarantee | README benchmark table | - -## ๐Ÿš€ Quick Start Instructions -### Prerequisites - * Python 3.10 or greater installed locally. - * Deep learning dependencies (Numpy, PyTorch, or ONNX Runtime). -### Setup Sequence - 1. Clone Repository & Navigate - Terminal Setup - Pull down the deep-learning model tracking pipeline repository to your local path directory. - - 2. Establish Virtual Environment - Dependency Isolation - Create and initialize a clean virtual runtime sandbox space to isolate deep learning modules. - - 3. Deploy System Requirements - Package Management - Install application weights, state management typings, and required matrix computation tools. - - 4. Run Local Verification Suite - Automated CI/CD Validation - Trigger the test pipeline layer to confirm type-checking compliance and code format validation. - -## ๐Ÿ“‘ Deep-Dive Engineering Q&A -### Architectural & Deep Learning Strategy -#### Why is a sliding-window guardrail mandatory for sequential deep learning models? -In deep sequence models (like GRU4Rec or RecSys Transformers), the network tracks patterns using hidden states or self-attention matrices. If a user clicks hundreds of items in a single session, the model's matrix computation size grows rapidly, creating an O(N) or O(N^2) computing bottleneck. This leads to high latency and risks out-of-memory errors that can crash serving nodes. -By using an explicit RecommenderGuardrail that limits inputs to the latest N=10 items, we force the tensor input dimensions to stay perfectly static. This gives us predictable O(1) constant-time execution speeds for every single recommendation request, regardless of user history length. -#### What makes Pydantic a superior choice for online Deep Learning inference contexts? -Traditional deep learning serving architectures often pass raw JSON structures, native Python lists, or loosely typed NumPy arrays through helper modules. If a user action contains a corrupted input or a missing item token ID, the deep learning model will experience an internal array dimension mismatch or a silent tracking failure down the line. -By wrapping session context in a strict Pydantic model (SessionState), we enforce automated type validation and data coercion at the edge of the system. This guarantees that any data entering our neural network layers perfectly matches the required shapes and data types before any matrix math runs. -#### How does the Latency SLA Breaker protect distributed serving infrastructures? -When web application traffic spikes, a deep recommendation model can become overloaded, with token processing queues backing up fast. If a serving node takes hundreds of milliseconds to evaluate a single user sequence, upstream services will timeout, creating a cascading failure across the entire application platform. -Our SequentialInferenceEngine isolates this execution loop by tracking processing time down to the millisecond. The moment a prediction step breaches our 50\text{ms} threshold constraint, the internal circuit breaker stops the heavy matrix computation step immediately. It logs an operational warning and safely falls back to a fast, pre-cached list of trending itemsโ€”keeping response times ultra-low and protecting overall application stability under high stress. + +## Production serving + +```bash +ENVIRONMENT=production \ +MODEL_BUNDLE_PATH=models/current \ +API_KEY='replace-me' \ +uvicorn app.main:app --host 0.0.0.0 --port 8000 +``` + +```bash +curl -X POST http://localhost:8000/recommendations/ \ + -H 'Content-Type: application/json' \ + -H 'X-API-Key: replace-me' \ + -d '{"user_id":"user-42","item_sequence":["item-2","item-7"],"top_k":5}' +``` + +Feedback is accepted at `POST /recommendations/feedback`. Raw user IDs are hashed before events +are logged. Production should route structured logs to a governed event bus with retention, +consent, deletion, and delivery guarantees. + +## L5 engineering details + +| Concern | Decision | Tradeoff / boundary | +|---|---|---| +| Training-serving parity | One vocabulary and architecture manifest drives serving and ONNX | Bundle migrations need explicit compatibility handling | +| Leakage | Global chronological split and causal prefix targets | Production also needs cold-start and cohort evaluation | +| Promotion | Neural NDCG is compared with popularity | Registry automation remains deployment-specific | +| Integrity | SHA-256 plus strict state-dict loading | Add KMS/Sigstore for signed provenance | +| Overload | Non-blocking admission and deterministic fallback | Hard cancellation needs a worker/model-server boundary | +| Cache | Key includes model version, history, and top-k | In-memory state is replica-local | +| Rate limit | Per-process fixed window | Authoritative distributed limits belong at the gateway | +| Authentication | Constant-time optional API key | Multi-tenant systems need workload identity/OAuth and policy | +| Feedback | Pseudonymized structured events | Event bus, retention, and deletion are platform integrations | +| Kubernetes | Immutable tag, read-only FS, non-root, seccomp, resource bounds | Registry, secrets, PVC, and rollout controller are external | + +## Research metrics and benchmarks + +Every real candidate reports Recall@K, NDCG@K, MRR@K, and coverage beside popularity. No quality +number is published here because a representative production dataset is not included. + +Reference systems microbenchmark: + +```bash +python -m benchmarks.inference --iterations 200 --warmup 20 +``` + +| Metric | Observation | +|---|---:| +| Mean | 0.573 ms | +| p50 | 0.556 ms | +| p95 | 0.755 ms | +| p99 | 1.118 ms | +| Sequential rate | 1,743.0 inferences/s | + +Environment: Python 3.12.13, PyTorch 2.13.0 CPU, Windows 11 build 26200, AMD64 Family 25 Model +97; batch 1, 500 items, padded length 50, top-k 10, 20 warmups, 200 measured samples, collected +July 22, 2026. + +This excludes HTTP, serialization, queueing, network, concurrency, replicas, cold starts, and +real catalogue scale. It is a regression baselineโ€”not a production SLO or availability claim. +See [benchmark methodology](docs/BENCHMARKS.md). + +## Production-readiness status + +- [x] Correct ranking IDs and padding behavior +- [x] Strict request and event contracts +- [x] Temporal evaluation and baseline comparison +- [x] Versioned checksummed model bundle +- [x] Fail-closed production loading and model readiness +- [x] Admission, fallback, cache, rate limit, and API-key option +- [x] Feedback contract and user-ID minimization +- [x] Enforced CI, schema, security, and benchmark evidence +- [x] Hardened container and Kubernetes manifests +- [ ] Governed representative production dataset +- [ ] External registry, signed provenance, and automated promotion +- [ ] Distributed gateway limits and durable event bus +- [ ] Target-environment multi-replica load test +- [ ] Shadow/canary rollout with automated rollback +- [ ] Online experiment proving user or business impact + +Unchecked items require real data or deployment infrastructure and cannot be honestly completed +inside a standalone public repository. + +## Extended recruiter Q&A + +### Is this genuinely AI-powered? + +Yes when a trained bundle is supplied. It trains a bidirectional LSTM with padding-aware attention +on causal next-item examples. Production cannot silently substitute random weights. + +### Why was the old demo not production AI? + +A neural class with random weights is not learned recommendation. Production requires data, +evaluation, lineage, versioned artifacts, readiness, rollback, and operational evidence. + +### Why use a popularity baseline? + +A complex model should earn its cost. Popularity is cheap and often strong; a neural candidate +that cannot beat it on leakage-safe ranking metrics should not be promoted. + +### Why split chronologically? + +Random splits leak future behavior. Chronological evaluation asks the real question: can the model +rank behavior that occurs later than its training evidence? + +### Why package vocabulary with weights? + +Embedding rows and output classes only mean something under the exact training mapping. A different +mapping can return plausible but incorrect item IDs without producing a tensor shape error. + +### Why verify checksums? + +Models are usually deployed outside Git. Checksums detect partial uploads, accidental bundle +mixing, and corruption. Signing/provenance is the next layer for regulated environments. + +### Why retain a development model? + +It keeps smoke tests self-contained. It is seeded, visibly labeled untrained, and forbidden in +production, so convenience cannot become an invisible ranking failure. + +### Is the latency fallback a hard timeout? + +No. It replaces a result after an in-process kernel finishes. Hard cancellation requires a process +or model-server boundary with load shedding and cancelable requests. + +### Why not use an LLM as the ranker? + +LLMs help with intent, catalogue enrichment, and grounded explanations. Core ranking still needs +measurable retrieval, temporal evaluation, deterministic policy, and controlled cost. + +### How would this scale to millions of items? + +Use a two-stage system: two-tower/ANN candidate retrieval followed by sequence re-ranking. A dense +projection over the entire catalogue is not economical at very large scale. + +### How would you handle cold start? + +Initialize new items from text/image/category embeddings and use contextual popularity until +behavior accumulates. Report new-user and new-item cohorts separately. + +### What should be monitored? + +Latency, errors, saturation, cache/fallback rates, model version, unknown-item rate, feature and +prediction drift, delayed-label ranking quality, coverage, diversity, bias, and experiment goals. + +### What is the rollout strategy? + +Shadow first, compare outputs/resources, canary a small percentage, enforce guardrails, then ramp +with automatic rollback to the previous immutable image-and-bundle pair. + +### What demonstrates senior engineering here? + +The model is treated as one component. Data contracts, evaluation, lineage, serving safety, +privacy, observability, delivery, and honest claims form the production system around it. + +## Documentation + +- [Benchmark methodology](docs/BENCHMARKS.md) +- [Model card](docs/MODEL_CARD.md) +- [Operations runbook](docs/OPERATIONS.md) +- [Privacy policy](docs/PRIVACY.md) +- [Issue #8](https://github.com/CoreyLeath-code/DeepSequence-Recommender/issues/8) + +MIT licensed. See [LICENSE](LICENSE). diff --git a/app/api/routes.py b/app/api/routes.py index 9042533..55f518e 100644 --- a/app/api/routes.py +++ b/app/api/routes.py @@ -1,100 +1,197 @@ -"""FastAPI routes for recommendation endpoints.""" +"""Validated recommendation routes with bounded inference and fallback behavior.""" from __future__ import annotations +import hashlib +import json +import logging import time -from typing import List, Optional +from dataclasses import dataclass +from typing import Literal -from fastapi import APIRouter, HTTPException -from pydantic import BaseModel +from fastapi import APIRouter, Header, HTTPException +from pydantic import BaseModel, Field from app.core.config import settings from app.core.data_processor import SequenceProcessor from app.core.metrics import ( active_requests, + cache_hits_total, + cache_misses_total, + feedback_events_total, model_inference_latency, recommendation_latency, recommendations_total, ) from app.core.model import DeepSequenceModel +from app.core.security import api_key_is_valid +from app.core.serving import AdmissionController, RateLimiter, RecommendationCache router = APIRouter(prefix="/recommendations", tags=["recommendations"]) -# --------------------------------------------------------------------------- -# Shared in-memory model / processor (initialised at startup via app.main) -# --------------------------------------------------------------------------- -_processor: Optional[SequenceProcessor] = None -_model: Optional[DeepSequenceModel] = None - -def init_model(processor: SequenceProcessor, model: DeepSequenceModel) -> None: - global _processor, _model - _processor = processor - _model = model - - -# --------------------------------------------------------------------------- -# Request / Response schemas -# --------------------------------------------------------------------------- +@dataclass +class ModelRuntime: + processor: SequenceProcessor + model: DeepSequenceModel + model_version: str + trained: bool + popular_items: list[str] + + +_runtime: ModelRuntime | None = None +_admission = AdmissionController(settings.max_concurrent_inferences) +_cache = RecommendationCache(settings.cache_ttl_seconds) +_rate_limiter = RateLimiter(settings.requests_per_minute) +logger = logging.getLogger(__name__) + + +def init_model( + processor: SequenceProcessor, + model: DeepSequenceModel, + *, + model_version: str, + trained: bool, + popular_items: list[str] | None = None, +) -> None: + global _runtime + _runtime = ModelRuntime( + processor=processor, + model=model, + model_version=model_version, + trained=trained, + popular_items=popular_items or list(processor.export_vocabulary())[: settings.max_top_k], + ) class RecommendRequest(BaseModel): - user_id: str - item_sequence: List[str] - top_k: int = settings.top_k + user_id: str = Field(min_length=1, max_length=128) + item_sequence: list[str] = Field(min_length=1, max_length=500) + top_k: int = Field(default=settings.top_k, ge=1, le=settings.max_top_k) class RecommendResponse(BaseModel): user_id: str - recommendations: List[str] + recommendations: list[str] latency_ms: float - - -# --------------------------------------------------------------------------- -# Endpoints -# --------------------------------------------------------------------------- + model_version: str + fallback: bool = False + cache_hit: bool = False + + +class FeedbackRequest(BaseModel): + impression_id: str = Field(min_length=1, max_length=128) + user_id: str = Field(min_length=1, max_length=128) + item_id: str = Field(min_length=1, max_length=256) + event_type: Literal["impression", "click", "skip", "cart", "purchase", "dislike"] + position: int | None = Field(default=None, ge=0, le=10_000) + model_version: str = Field(min_length=1, max_length=128) + + +def _authorize(api_key: str | None) -> None: + if not api_key_is_valid(api_key, settings.api_key): + raise HTTPException(status_code=401, detail="Invalid API key") + + +def _response( + request: RecommendRequest, + recommendations: list[str], + started: float, + *, + fallback: bool = False, + cache_hit: bool = False, +) -> RecommendResponse: + assert _runtime is not None + return RecommendResponse( + user_id=request.user_id, + recommendations=recommendations, + latency_ms=(time.perf_counter() - started) * 1_000, + model_version=_runtime.model_version, + fallback=fallback, + cache_hit=cache_hit, + ) @router.post("/", response_model=RecommendResponse, summary="Generate recommendations") -def recommend(req: RecommendRequest) -> RecommendResponse: - """Return top-k next-item recommendations for a user interaction sequence.""" - if _model is None or _processor is None: +def recommend( + req: RecommendRequest, x_api_key: str | None = Header(default=None) +) -> RecommendResponse: + _authorize(x_api_key) + if not _rate_limiter.allow(req.user_id): + raise HTTPException(status_code=429, detail="Recommendation rate limit exceeded") + if _runtime is None: raise HTTPException(status_code=503, detail="Model not initialised") + if req.top_k > _runtime.processor.vocab_size: + raise HTTPException(status_code=422, detail="top_k exceeds catalogue size") + + known_items = [ + item for item in req.item_sequence if _runtime.processor.item_to_idx(item) != 0 + ] + if not known_items: + raise HTTPException(status_code=422, detail="Sequence contains no known catalogue items") + remaining_items = _runtime.processor.vocab_size - len(set(known_items)) + if req.top_k > remaining_items: + raise HTTPException(status_code=422, detail="top_k exceeds remaining eligible items") + + started = time.perf_counter() + cache_key = _cache.key(_runtime.model_version, known_items, req.top_k) + cached = _cache.get(cache_key) + if cached is not None: + cache_hits_total.inc() + recommendations_total.labels(status="cache_hit").inc() + return _response(req, cached, started, cache_hit=True) + cache_misses_total.inc() + + if not _admission.acquire(): + recommendations_total.labels(status="fallback_overload").inc() + return _response(req, _runtime.popular_items[: req.top_k], started, fallback=True) active_requests.inc() - wall_start = time.perf_counter() try: - tensor = _processor.to_tensor(req.item_sequence) - - infer_start = time.perf_counter() - raw_indices = _model.recommend( + tensor = _runtime.processor.to_tensor(known_items) + infer_started = time.perf_counter() + indices = _runtime.model.recommend( tensor, top_k=req.top_k, - exclude_ids=[_processor.item_to_idx(i) for i in req.item_sequence], + exclude_ids=[_runtime.processor.item_to_idx(item) for item in known_items], ) - model_inference_latency.observe(time.perf_counter() - infer_start) - - decoded = _processor.decode_recommendations(raw_indices) - recommendations = [r for r in decoded if r is not None] - + inference_ms = (time.perf_counter() - infer_started) * 1_000 + model_inference_latency.observe(inference_ms / 1_000) + if inference_ms > settings.max_inference_ms: + recommendations_total.labels(status="fallback_latency").inc() + return _response(req, _runtime.popular_items[: req.top_k], started, fallback=True) + + decoded = _runtime.processor.decode_recommendations(indices) + recommendations = [item for item in decoded if item is not None] + _cache.put(cache_key, recommendations) recommendations_total.labels(status="success").inc() - return RecommendResponse( - user_id=req.user_id, - recommendations=recommendations, - latency_ms=(time.perf_counter() - wall_start) * 1000, - ) + return _response(req, recommendations, started) except (ValueError, RuntimeError, KeyError) as exc: recommendations_total.labels(status="error").inc() - raise HTTPException(status_code=500, detail=str(exc)) from exc + raise HTTPException(status_code=500, detail="Recommendation inference failed") from exc finally: - recommendation_latency.observe(time.perf_counter() - wall_start) + recommendation_latency.observe(time.perf_counter() - started) active_requests.dec() + _admission.release() + + +@router.post("/feedback", status_code=202, summary="Capture recommendation feedback") +def feedback(req: FeedbackRequest, x_api_key: str | None = Header(default=None)) -> dict: + """Emit a privacy-minimized event for collection by the platform log pipeline.""" + _authorize(x_api_key) + anonymized_user = hashlib.sha256(req.user_id.encode()).hexdigest()[:16] + event = req.model_dump(exclude={"user_id"}) | {"anonymous_user_id": anonymized_user} + logger.info("recommendation_feedback=%s", json.dumps(event, sort_keys=True)) + feedback_events_total.labels(event_type=req.event_type).inc() + return {"accepted": True, "impression_id": req.impression_id} -@router.get("/health", summary="Health check") +@router.get("/health", summary="Model readiness check") def health() -> dict: return { - "status": "ok", - "model_loaded": _model is not None, - "vocab_size": _processor.vocab_size if _processor else 0, + "status": "ready" if _runtime is not None else "not_ready", + "model_loaded": _runtime is not None, + "trained_model": _runtime.trained if _runtime else False, + "model_version": _runtime.model_version if _runtime else None, + "vocab_size": _runtime.processor.vocab_size if _runtime else 0, } diff --git a/app/core/artifacts.py b/app/core/artifacts.py new file mode 100644 index 0000000..553bf88 --- /dev/null +++ b/app/core/artifacts.py @@ -0,0 +1,85 @@ +"""Versioned, checksummed model bundles shared by training and serving.""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Any + +import torch +from pydantic import BaseModel, Field + +from app.core.data_processor import SequenceProcessor +from app.core.model import DeepSequenceModel + + +class ModelManifest(BaseModel): + model_version: str + created_at: str + architecture: str = "bidirectional-lstm-attention" + architecture_config: dict[str, Any] + metrics: dict[str, float] = Field(default_factory=dict) + dataset_id: str + weights_sha256: str + vocabulary_sha256: str + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for block in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def save_bundle( + directory: str | Path, + model: DeepSequenceModel, + processor: SequenceProcessor, + manifest: ModelManifest, +) -> ModelManifest: + target = Path(directory) + target.mkdir(parents=True, exist_ok=True) + weights_path = target / "model.pt" + vocabulary_path = target / "vocabulary.json" + torch.save(model.state_dict(), weights_path) + vocabulary_path.write_text( + json.dumps(processor.export_vocabulary(), indent=2, sort_keys=True), + encoding="utf-8", + ) + completed = manifest.model_copy( + update={ + "weights_sha256": _sha256(weights_path), + "vocabulary_sha256": _sha256(vocabulary_path), + } + ) + (target / "manifest.json").write_text( + completed.model_dump_json(indent=2), encoding="utf-8" + ) + return completed + + +def load_bundle( + directory: str | Path, +) -> tuple[SequenceProcessor, DeepSequenceModel, ModelManifest]: + target = Path(directory) + manifest = ModelManifest.model_validate_json( + (target / "manifest.json").read_text(encoding="utf-8") + ) + weights_path = target / "model.pt" + vocabulary_path = target / "vocabulary.json" + if _sha256(weights_path) != manifest.weights_sha256: + raise ValueError("Model weights checksum does not match the manifest") + if _sha256(vocabulary_path) != manifest.vocabulary_sha256: + raise ValueError("Vocabulary checksum does not match the manifest") + + vocabulary = json.loads(vocabulary_path.read_text(encoding="utf-8")) + config = dict(manifest.architecture_config) + max_length = int(config.pop("max_sequence_length")) + processor = SequenceProcessor.from_vocabulary(vocabulary, max_length=max_length) + model = DeepSequenceModel(num_items=processor.vocab_size, **config) + state = torch.load(weights_path, map_location="cpu", weights_only=True) + model.load_state_dict(state, strict=True) + model.eval() + return processor, model, manifest diff --git a/app/core/config.py b/app/core/config.py index 95c44cd..9918b80 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -1,15 +1,12 @@ """Application settings loaded from environment variables.""" -from pydantic import Field -from pydantic_settings import BaseSettings +from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): - secret_key: str = Field(default="change-this-secret") - redis_url: str = Field(default="redis://localhost:6379") - mlflow_tracking_uri: str = Field(default="http://localhost:5000") - environment: str = Field(default="development") - log_level: str = Field(default="INFO") + model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8") + environment: str = "development" + log_level: str = "INFO" # Model hyper-parameters embedding_dim: int = 64 @@ -17,10 +14,12 @@ class Settings(BaseSettings): num_layers: int = 2 max_sequence_length: int = 50 top_k: int = 10 - - class Config: - env_file = ".env" - env_file_encoding = "utf-8" - + max_top_k: int = 50 + model_bundle_path: str = "models/current" + max_inference_ms: float = 250.0 + max_concurrent_inferences: int = 8 + cache_ttl_seconds: int = 30 + requests_per_minute: int = 120 + api_key: str | None = None settings = Settings() diff --git a/app/core/data_processor.py b/app/core/data_processor.py index d4f571e..0beb6fe 100644 --- a/app/core/data_processor.py +++ b/app/core/data_processor.py @@ -66,3 +66,21 @@ def to_tensor(self, sequence: List[str]) -> torch.Tensor: def decode_recommendations(self, indices: List[int]) -> List[Optional[str]]: """Map numeric recommendation indices back to item identifiers.""" return [self.idx_to_item(idx) for idx in indices] + + def export_vocabulary(self) -> Dict[str, int]: + """Return a stable copy of the item-to-index mapping.""" + return dict(self._item2idx) + + @classmethod + def from_vocabulary( + cls, vocabulary: Dict[str, int], max_length: int = 50 + ) -> "SequenceProcessor": + """Restore and validate a persisted vocabulary.""" + expected = set(range(1, len(vocabulary) + 1)) + if set(vocabulary.values()) != expected: + raise ValueError("Vocabulary indices must be contiguous and start at one") + processor = cls(max_length=max_length) + processor._item2idx = dict(vocabulary) + processor._idx2item = {index: item for item, index in vocabulary.items()} + processor._next_idx = len(vocabulary) + 1 + return processor diff --git a/app/core/metrics.py b/app/core/metrics.py index 70f1424..6d4e471 100644 --- a/app/core/metrics.py +++ b/app/core/metrics.py @@ -27,10 +27,16 @@ cache_hits_total = Counter( "deepseq_cache_hits_total", - "Total number of Redis cache hits", + "Total number of replica-local recommendation cache hits", ) cache_misses_total = Counter( "deepseq_cache_misses_total", - "Total number of Redis cache misses", + "Total number of replica-local recommendation cache misses", +) + +feedback_events_total = Counter( + "deepseq_feedback_events_total", + "Feedback events accepted for offline learning", + ["event_type"], ) diff --git a/app/core/model.py b/app/core/model.py index 2c13b25..00a1ebd 100644 --- a/app/core/model.py +++ b/app/core/model.py @@ -24,10 +24,13 @@ def __init__(self, hidden_dim: int) -> None: super().__init__() self.attn = nn.Linear(hidden_dim * 2, 1) - def forward(self, lstm_out: torch.Tensor) -> torch.Tensor: + def forward(self, lstm_out: torch.Tensor, mask: torch.Tensor) -> torch.Tensor: # lstm_out: (batch, seq_len, hidden_dim*2) scores = self.attn(lstm_out).squeeze(-1) # (batch, seq_len) - weights = F.softmax(scores, dim=-1).unsqueeze(-1) # (batch, seq_len, 1) + scores = scores.masked_fill(~mask, -1e9) + weights = F.softmax(scores, dim=-1) * mask + weights = weights / weights.sum(dim=-1, keepdim=True).clamp_min(1e-9) + weights = weights.unsqueeze(-1) # (batch, seq_len, 1) context = (lstm_out * weights).sum(dim=1) # (batch, hidden_dim*2) return context @@ -56,7 +59,9 @@ def __init__( ) self.attention = AttentionLayer(hidden_dim) self.dropout = nn.Dropout(dropout) - self.output_proj = nn.Linear(hidden_dim * 2, num_items) + self.num_items = num_items + self.padding_idx = padding_idx + self.output_proj = nn.Linear(hidden_dim * 2, num_items + 1) def forward(self, item_seq: torch.Tensor) -> torch.Tensor: """Return logits over the item catalogue. @@ -69,12 +74,15 @@ def forward(self, item_seq: torch.Tensor) -> torch.Tensor: Returns ------- torch.Tensor - Logit tensor of shape ``(batch_size, num_items)``. + Logit tensor of shape ``(batch_size, num_items + 1)``. Index zero + is reserved for padding and is never eligible for recommendation. """ emb = self.dropout(self.embedding(item_seq)) # (B, L, E) lstm_out, _ = self.lstm(emb) # (B, L, H*2) - context = self.attention(lstm_out) # (B, H*2) - logits = self.output_proj(self.dropout(context)) # (B, num_items) + mask = item_seq.ne(self.padding_idx) + context = self.attention(lstm_out, mask) # (B, H*2) + logits = self.output_proj(self.dropout(context)) # (B, num_items + 1) + logits[:, self.padding_idx] = float("-inf") return logits @torch.no_grad() @@ -86,9 +94,15 @@ def recommend( ) -> List[int]: """Return top-k recommended item IDs for a single sequence.""" self.eval() + if item_seq.ndim != 2 or not item_seq.ne(self.padding_idx).any(): + raise ValueError("A recommendation requires at least one known item") + if not 1 <= top_k <= self.num_items: + raise ValueError(f"top_k must be between 1 and {self.num_items}") + excluded = {idx for idx in (exclude_ids or []) if 1 <= idx <= self.num_items} + if top_k > self.num_items - len(excluded): + raise ValueError("top_k exceeds the remaining eligible catalogue") logits = self.forward(item_seq) - if exclude_ids: - for idx in exclude_ids: - logits[:, idx] = float("-inf") + for idx in excluded: + logits[:, idx] = float("-inf") scores = torch.topk(logits, k=top_k, dim=-1) return scores.indices[0].tolist() diff --git a/app/core/security.py b/app/core/security.py index 7c9a8b4..f439874 100644 --- a/app/core/security.py +++ b/app/core/security.py @@ -1,41 +1,12 @@ -"""JWT-based security helpers.""" +"""Minimal service authentication primitives.""" from __future__ import annotations -import os -from datetime import datetime, timedelta, timezone -from typing import Optional +import secrets -from jose import JWTError, jwt -from passlib.context import CryptContext -SECRET_KEY: str = os.getenv("SECRET_KEY", "change-this-secret") -ALGORITHM: str = "HS256" -ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 - -pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") - - -def hash_password(password: str) -> str: - return pwd_context.hash(password) - - -def verify_password(plain_password: str, hashed_password: str) -> bool: - return pwd_context.verify(plain_password, hashed_password) - - -def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: - to_encode = data.copy() - expire = datetime.now(timezone.utc) + ( - expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) - ) - to_encode["exp"] = expire - return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) - - -def verify_token(token: str) -> Optional[dict]: - try: - payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) - return payload - except JWTError: - return None +def api_key_is_valid(provided: str | None, configured: str | None) -> bool: + """Allow open development when unconfigured; compare keys in constant time otherwise.""" + if configured is None: + return True + return provided is not None and secrets.compare_digest(provided, configured) diff --git a/app/core/serving.py b/app/core/serving.py new file mode 100644 index 0000000..1e861f2 --- /dev/null +++ b/app/core/serving.py @@ -0,0 +1,75 @@ +"""Serving controls for admission, caching, and deterministic fallback.""" + +from __future__ import annotations + +import hashlib +import json +import threading +import time +from dataclasses import dataclass + + +class AdmissionController: + def __init__(self, limit: int) -> None: + if limit < 1: + raise ValueError("Concurrency limit must be positive") + self._semaphore = threading.BoundedSemaphore(limit) + + def acquire(self) -> bool: + return self._semaphore.acquire(blocking=False) + + def release(self) -> None: + self._semaphore.release() + + +@dataclass +class CacheEntry: + recommendations: list[str] + expires_at: float + + +class RecommendationCache: + def __init__(self, ttl_seconds: int) -> None: + self.ttl_seconds = ttl_seconds + self._entries: dict[str, CacheEntry] = {} + self._lock = threading.Lock() + + @staticmethod + def key(model_version: str, sequence: list[str], top_k: int) -> str: + payload = json.dumps([model_version, sequence, top_k], separators=(",", ":")) + return hashlib.sha256(payload.encode()).hexdigest() + + def get(self, key: str) -> list[str] | None: + with self._lock: + entry = self._entries.get(key) + if entry is None or entry.expires_at <= time.monotonic(): + self._entries.pop(key, None) + return None + return list(entry.recommendations) + + def put(self, key: str, recommendations: list[str]) -> None: + with self._lock: + self._entries[key] = CacheEntry( + recommendations=list(recommendations), + expires_at=time.monotonic() + self.ttl_seconds, + ) + + +class RateLimiter: + """Per-process fixed-window limiter; use a shared gateway for distributed limits.""" + + def __init__(self, requests_per_minute: int) -> None: + self.limit = requests_per_minute + self._windows: dict[str, tuple[int, float]] = {} + self._lock = threading.Lock() + + def allow(self, identity: str) -> bool: + now = time.monotonic() + with self._lock: + count, window_started = self._windows.get(identity, (0, now)) + if now - window_started >= 60: + count, window_started = 0, now + if count >= self.limit: + return False + self._windows[identity] = (count + 1, window_started) + return True diff --git a/app/main.py b/app/main.py index 9c97d5f..9e4c52e 100644 --- a/app/main.py +++ b/app/main.py @@ -1,76 +1,94 @@ -"""DeepSequence Recommender โ€“ FastAPI application entry-point.""" +"""DeepSequence Recommender FastAPI application entry point.""" from __future__ import annotations import logging import time +from contextlib import asynccontextmanager +from pathlib import Path +import torch from fastapi import FastAPI from prometheus_client import make_asgi_app from app.api.routes import init_model, router +from app.core.artifacts import load_bundle from app.core.config import settings from app.core.data_processor import SequenceProcessor from app.core.model import DeepSequenceModel logging.basicConfig(level=settings.log_level) logger = logging.getLogger(__name__) - _startup_time: float = 0.0 -app = FastAPI( - title="DeepSequence Recommender", - description="Production-grade deep learning sequence recommender service.", - version="1.0.0", -) - -# Mount Prometheus metrics endpoint -metrics_app = make_asgi_app() -app.mount("/metrics", metrics_app) -app.include_router(router) - - -@app.on_event("startup") -def startup_event() -> None: - """Initialise a default in-memory model for demonstration purposes.""" +def initialize_model() -> None: + """Load a verified model bundle or an explicit development-only fallback.""" global _startup_time - logger.info("Initialising DeepSequence modelโ€ฆ") - - processor = SequenceProcessor(max_length=settings.max_sequence_length) - # Seed processor with a small catalogue so the model can start immediately. - demo_items = [[f"item_{i}" for i in range(200)]] - processor.fit(demo_items) - - model = DeepSequenceModel( - num_items=processor.vocab_size, - embedding_dim=settings.embedding_dim, - hidden_dim=settings.hidden_dim, - num_layers=settings.num_layers, - ) - - init_model(processor, model) + bundle_path = Path(settings.model_bundle_path) + if (bundle_path / "manifest.json").is_file(): + processor, model, manifest = load_bundle(bundle_path) + init_model( + processor, + model, + model_version=manifest.model_version, + trained=True, + ) + elif settings.environment.lower() == "production": + raise RuntimeError(f"Production requires a verified model bundle at {bundle_path}") + else: + logger.warning("No trained bundle found; using deterministic development model") + torch.manual_seed(7) + processor = SequenceProcessor(max_length=settings.max_sequence_length) + processor.fit([[f"item_{index}" for index in range(200)]]) + model = DeepSequenceModel( + num_items=processor.vocab_size, + embedding_dim=settings.embedding_dim, + hidden_dim=settings.hidden_dim, + num_layers=settings.num_layers, + ).eval() + init_model( + processor, + model, + model_version="development-untrained", + trained=False, + ) _startup_time = time.time() logger.info( - "Model ready. vocab_size=%d embedding_dim=%d hidden_dim=%d", + "Model ready. vocab_size=%d environment=%s bundle=%s", processor.vocab_size, - settings.embedding_dim, - settings.hidden_dim, + settings.environment, + bundle_path, ) +@asynccontextmanager +async def lifespan(_app: FastAPI): + initialize_model() + yield + + +app = FastAPI( + title="DeepSequence Recommender", + description="Versioned, evaluated deep sequence recommendation service.", + version="1.1.0", + lifespan=lifespan, +) +app.mount("/metrics", make_asgi_app()) +app.include_router(router) + + @app.get("/", summary="Root") def root() -> dict: - return {"service": "DeepSequence Recommender", "version": "1.0.0"} + return {"service": "DeepSequence Recommender", "version": "1.1.0"} -@app.get("/health", summary="Application health check", tags=["health"]) +@app.get("/health", summary="Application liveness check", tags=["health"]) def health() -> dict: - """Return service liveness and basic diagnostics.""" uptime = round(time.time() - _startup_time, 1) if _startup_time else 0.0 return { "status": "ok", - "version": "1.0.0", + "version": "1.1.0", "environment": settings.environment, "uptime_seconds": uptime, } diff --git a/app/recsys_control_room.py b/app/recsys_control_room.py deleted file mode 100644 index 2848b93..0000000 --- a/app/recsys_control_room.py +++ /dev/null @@ -1,98 +0,0 @@ -import streamlit as st -import numpy as np -import time -import matplotlib.pyplot as plt -import seaborn as sns -from src.features.online_retrieval import LatencyIsolatedFeatureStore - -st.set_page_config(page_title="DeepSequence RecSys Control Panel", layout="wide") -sns.set_theme(style="darkgrid") - -st.title("๐Ÿง  DeepSequence Recommender Control Panel & Telemetry Console") -st.caption("L6 Production-Tier Hyperparameter Sweeps & Real-Time Next-Item Scoring Diagnostics") - -# Initialize feature store components -if 'feature_store' not in st.session_state: - st.session_state.feature_store = LatencyIsolatedFeatureStore() - -store = st.session_state.feature_store - -# Sample Item Taxonomy Lookup Matrix -item_catalog = { - 0: "Padding Token / System Null", - 12: "Pro Slate Keyboard Hub", - 22: "Quantum Wireless Mouse v4", - 41: "Liquid Retinal Array 32-inch Panel", - 58: "Thunderbolt Multi-Bus Hub", - 88: "4K Streamer Capture Card Pro", - 102: "Developer Ergonomic Mechanical Board", - 405: "Titanium Desk Mounting Articulated Arm", - 994: "USB-C Active Optical Line (5m)" -} - -st.sidebar.header("๐Ÿ•น๏ธ Simulation Environment Context") -target_user = st.sidebar.selectbox("Active Simulation Profile", ["user_8271", "user_1194", "user_4952"]) -recommendation_count = st.sidebar.slider("Top-K Candidates to Retrieve", 3, 15, 5) - -st.markdown("---") - -left_panel, right_panel = st.columns([1, 1]) - -with left_panel: - st.subheader("โฑ๏ธ Online Feature Extraction Sequence Timeline") - - # Process user retrieval cycles - input_tensor, fetch_latency = store.fetch_user_realtime_sequence(target_user) - - # Render historical items - st.write(f"**Target Active Profile ID:** `{target_user}`") - st.write(f"**Feature Retrieval Latency:** `{fetch_latency:.4f} ms` (In-Memory Hit)") - - historical_tokens = input_tensor[0][input_tensor[0] != 0] - - timeline_data = [] - for step, token in enumerate(historical_tokens): - resolved_name = item_catalog.get(int(token), f"Unknown E-Commerce Skew Token ({token})") - timeline_data.append({"Step": step + 1, "Item SKU ID": token, "Product Name Name": resolved_name}) - - st.table(timeline_data) - -with right_panel: - st.subheader("๐ŸŽฏ Real-Time Scoring Head Probability Distribution") - - # Simulation logic for ONNX deep scoring latency paths - inference_start = time.time() - time.sleep(0.0035) # Simulating Triton model batch-execution runtime delay - inference_latency = (time.time() - inference_start) * 1000 - - # Generate mock distribution outputs via stable sampling Dirichlet models - np.random.seed(int(historical_tokens[-1]) if len(historical_tokens) > 0 else 42) - mock_logits = np.random.dirichlet(np.ones(10), size=1)[0] - top_indices = np.argsort(mock_logits)[-recommendation_count:][::-1] - - scored_items = [] - scored_probabilities = [] - for idx in top_indices: - simulated_sku = 10 * idx + 2 # Deterministic structural mock map hashes - scored_items.append(item_catalog.get(simulated_sku, f"Product SKU Block {simulated_sku}")) - scored_probabilities.append(mock_logits[idx]) - - # Render analytics graph matrix - fig, ax = plt.subplots(figsize=(6, 3.5)) - sns.barplot(x=scored_probabilities, y=scored_items, ax=ax, palette="viridis") - fig.patch.set_facecolor("#1e2129") - ax.set_facecolor("#1e2129") - ax.xaxis.label.set_color('white') - ax.yaxis.label.set_color('white') - ax.tick_params(colors='white') - st.pyplot(fig) - -st.markdown("---") -st.subheader("โšก End-to-End Operational Telemetry Summaries") -col1, col2, col3 = st.columns(3) -with col1: - st.metric("Aggregate Pipeline SLA Latency", f"{(fetch_latency + inference_latency):.2f} ms", delta="< 10ms System Budget") -with col2: - st.metric("Inference Engine Scoring Bound", f"{inference_latency:.2f} ms") -with col3: - st.metric("Triton Worker Concurrent Capacity", "18,450 RPS", delta="99.2% Utilization Profile") diff --git a/benchmarks/__init__.py b/benchmarks/__init__.py new file mode 100644 index 0000000..eca7f2d --- /dev/null +++ b/benchmarks/__init__.py @@ -0,0 +1 @@ +"""Measured serving benchmarks with machine-readable output.""" diff --git a/benchmarks/inference.py b/benchmarks/inference.py new file mode 100644 index 0000000..a97cc32 --- /dev/null +++ b/benchmarks/inference.py @@ -0,0 +1,79 @@ +"""Deterministic single-process inference benchmark.""" + +from __future__ import annotations + +import argparse +import json +import math +import platform +import statistics +import time + +import torch + +from app.core.data_processor import SequenceProcessor +from app.core.model import DeepSequenceModel + + +def _percentile(values: list[float], fraction: float) -> float: + return sorted(values)[max(0, math.ceil(len(values) * fraction) - 1)] + + +def run_benchmark(iterations: int = 200, warmup: int = 20) -> dict: + if iterations < 1 or warmup < 0: + raise ValueError("iterations must be positive and warmup non-negative") + torch.manual_seed(7) + processor = SequenceProcessor(max_length=50).fit( + [[f"item_{index}" for index in range(500)]] + ) + model = DeepSequenceModel( + num_items=processor.vocab_size, + embedding_dim=32, + hidden_dim=64, + num_layers=1, + ).eval() + tensor = processor.to_tensor([f"item_{index}" for index in range(20)]) + for _ in range(warmup): + model.recommend(tensor, top_k=10) + latencies = [] + started = time.perf_counter() + for _ in range(iterations): + call_started = time.perf_counter() + model.recommend(tensor, top_k=10) + latencies.append((time.perf_counter() - call_started) * 1_000) + duration = time.perf_counter() - started + return { + "scope": "single-process CPU PyTorch inference; batch=1; no HTTP or network", + "environment": { + "python": platform.python_version(), + "pytorch": torch.__version__, + "platform": platform.platform(), + "processor": platform.processor() or "not reported", + }, + "workload": { + "iterations": iterations, + "warmup": warmup, + "catalogue_size": processor.vocab_size, + "sequence_length": 50, + "top_k": 10, + }, + "results": { + "latency_ms_mean": round(statistics.fmean(latencies), 3), + "latency_ms_p50": round(statistics.median(latencies), 3), + "latency_ms_p95": round(_percentile(latencies, 0.95), 3), + "latency_ms_p99": round(_percentile(latencies, 0.99), 3), + "sequential_inferences_per_second": round(iterations / duration, 3), + }, + } + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--iterations", type=int, default=200) + parser.add_argument("--warmup", type=int, default=20) + args = parser.parse_args() + print(json.dumps(run_benchmark(args.iterations, args.warmup), indent=2)) + + +if __name__ == "__main__": + main() diff --git a/demo_app.py b/demo_app.py deleted file mode 100644 index c5ac37b..0000000 --- a/demo_app.py +++ /dev/null @@ -1,142 +0,0 @@ -import streamlit as st -import pandas as pd -import numpy as np -import time - -# Page Configuration -st.set_page_config(page_title="DeepSequence Recommender System", page_icon="๐Ÿง ", layout="wide") - -st.title("๐Ÿง  DeepSequence-Recommender: Session-Based Predictive Engine") -st.caption("Advanced Machine Learning Showcase | Real-Time Recurrent Sequence Inference") - -# Mock Database of items across distinct domains (Tech & Media) to demonstrate sequence tracking -ITEM_POOL = { - "Electronics & Tech": [ - {"id": 101, "name": "Mechanical Keyboard (Tactile Browns)", "category": "Peripherals"}, - {"id": 102, "name": "Ultra-Wide 34-inch Curved Monitor", "category": "Displays"}, - {"id": 103, "name": "Ergonomic Vertical Mouse", "category": "Peripherals"}, - {"id": 104, "name": "USB-C Dual Display Docking Station", "category": "Connectivity"}, - {"id": 105, "name": "Premium Active Noise-Cancelling Headphones", "category": "Audio"}, - {"id": 106, "name": "High-Definition 4K Streaming Webcam", "category": "Peripherals"} - ], - "Streaming & Entertainment": [ - {"id": 201, "name": "Sci-Fi Cyberpunk Cyber-Thriller Series", "category": "Sci-Fi"}, - {"id": 202, "name": "Deep Space Exploration Documentary", "category": "Documentary"}, - {"id": 203, "name": "High-Stakes Artificial Intelligence Tech-Drama", "category": "Drama"}, - {"id": 204, "name": "Post-Apocalyptic Survival Feature Film", "category": "Sci-Fi"}, - {"id": 205, "name": "Neo-Noir Psychological Thriller", "category": "Thriller"} - ] -} - -# Initialize browser memory for user session interaction sequences -if "click_stream" not in st.session_state: - st.session_state.click_stream = [] - -# --- SIDEBAR: CLICK-STREAM SIMULATOR --- -st.sidebar.header("๐Ÿ›’ User Action Sequence Simulator") -st.sidebar.markdown("Build an interactive click-stream session history to test how the sequence vectors shift state.") - -domain = st.sidebar.selectbox("Select Target Domain Platform", list(ITEM_POOL.keys())) -available_items = ITEM_POOL[domain] - -item_options = {item["name"]: item for item in available_items} -selected_item_name = st.sidebar.selectbox("Select Item to Interact With", list(item_options.keys())) -action_type = st.sidebar.selectbox("Interaction Type", ["View Item Detail", "Add to Cart / Playlist", "Purchase"]) - -if st.sidebar.button("๐Ÿ“ฅ Inject Action Into Sequence"): - chosen_item = item_options[selected_item_name] - st.session_state.click_stream.append({ - "Step": len(st.session_state.click_stream) + 1, - "ItemID": chosen_item["id"], - "Item Name": chosen_item["name"], - "Category": chosen_item["category"], - "Action": action_type - }) - -if st.sidebar.button("๐Ÿ—‘๏ธ Clear Active Session History"): - st.session_state.click_stream = [] - st.rerun() - -# --- MODEL INFERENCE LAYER SIMULATION --- -def generate_sequence_recommendations(session_history, pool): - """ - Simulates hidden state generation from an RNN/GRU or Transformer layout. - Dynamically shifts weights based on the absolute last items interacted with. - """ - if not session_history: - return [] - - # Target intent extracted from the most recent item in the sequence - last_item = session_history[-1] - last_category = last_item["Category"] - last_id = last_item["ItemID"] - - # Flatten pool items for evaluation - all_candidates = [] - for items in pool.values(): - all_candidates.extend(items) - - recommendations = [] - for item in all_candidates: - if item["id"] == last_id: - continue # Exclude item currently being viewed - - # Calculate dynamic hidden score weights based on contextual transitions - base_score = 0.2 - if item["category"] == last_category: - base_score += 0.55 # Local category affinity - else: - base_score += np.random.uniform(0.05, 0.15) # Global exploration bias - - # Add slight decay if session sequence length is long to simulate temporal dampening - final_probability = float(np.clip(base_score + np.random.uniform(-0.05, 0.05), 0.01, 0.99)) - - recommendations.append({ - "Recommended Item ID": item["id"], - "Item Name": item["name"], - "Category": item["category"], - "Match Probability Score": final_probability - }) - - # Sort by descending inference match probabilities - recommendations = sorted(recommendations, key=lambda x: x["Match Probability Score"], reverse=True) - return recommendations[:4] - -# --- MAIN DASHBOARD VIEW --- -col_left, col_right = st.columns([1, 1]) - -with col_left: - st.subheader("โฑ๏ธ Live User Session Sequence State") - if st.session_state.click_stream: - df_history = pd.DataFrame(st.session_state.click_stream) - st.dataframe(df_history, use_container_width=True, hide_index=True) - else: - st.info("The recurrent neural network state vector is empty. Use the sidebar simulator panel to add user clicks to the temporal stream.") - -with col_right: - st.subheader("๐Ÿ”ฎ Deep Next-Item Recommendations (Top-K)") - if st.session_state.click_stream: - with st.spinner("Processing sequence tokens through neural lookup arrays..."): - time.sleep(0.4) # Simulate network/inference forward pass latency - recs = generate_sequence_recommendations(st.session_state.click_stream, ITEM_POOL) - df_recs = pd.DataFrame(recs) - - # Render visual performance bars for confidence thresholds - st.dataframe( - df_recs.style.background_gradient(cmap="Blues", subset=["Match Probability Score"]), - use_container_width=True, - hide_index=True - ) - - st.caption("๐Ÿ’ก Notice how modifying your click history shifts prediction distributions to capture dynamic user transitions instantly.") - else: - st.text("Awaiting sequence input vectors to calculate tensor dot-products...") - -# Add technical architecture description for reading recruiters -st.markdown("---") -st.subheader("โš™๏ธ Portfolio Technical Specification Overview") -st.markdown(""" -This production-grade simulation maps the performance capabilities of the underlying **DeepSequence-Recommender** framework. -* **Temporal Modeling:** Tracks long-term vs. short-term intent transitions using embedding sequences instead of basic static matrix factorization. -* **Cold-Start Resilience:** Integrates auxiliary metadata features (Categories/Interactions) directly into the embedding layers to ensure accurate recommendations even within short user sessions. -""") diff --git a/docker-compose.yml b/docker-compose.yml index 1a396e0..8d09260 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -8,13 +8,12 @@ services: - "8000:8000" env_file: - .env - depends_on: - - redis - restart: unless-stopped - - redis: - image: redis:7-alpine - container_name: deepsequence-redis - ports: - - "6379:6379" + environment: + ENVIRONMENT: production + MODEL_BUNDLE_PATH: /models/current + volumes: + - ./models/current:/models/current:ro + read_only: true + tmpfs: + - /tmp restart: unless-stopped diff --git a/docs/BENCHMARKS.md b/docs/BENCHMARKS.md new file mode 100644 index 0000000..744c8ef --- /dev/null +++ b/docs/BENCHMARKS.md @@ -0,0 +1,25 @@ +# Benchmark methodology + +The inference benchmark measures one CPU PyTorch forward/top-k path. It uses a seeded model, +fixed workload, warmup, `time.perf_counter`, and nearest-rank percentiles. CI uploads the complete +JSON result so future comparisons retain environment and workload metadata. + +```bash +python -m benchmarks.inference --iterations 200 --warmup 20 +``` + +Reference observation collected July 22, 2026: + +| Scope | Value | +|---|---| +| Python / PyTorch | 3.12.13 / 2.13.0+cpu | +| Platform | Windows 11 build 26200, AMD64 Family 25 Model 97 | +| Workload | batch 1, 500 items, sequence 50, top-k 10 | +| Samples | 20 warmup, 200 measured | +| p50 / p95 / p99 | 0.556 / 0.755 / 1.118 ms | +| Sequential rate | 1,743.0 inferences/s | + +Do not compare results across hardware as if they were model improvements. For release decisions, +run both commits on the same idle host, publish raw JSON, and add HTTP/concurrency/load testing in +the target environment. Ranking metrics require a representative chronological dataset and must +be reported separately from systems latency. diff --git a/docs/MODEL_CARD.md b/docs/MODEL_CARD.md new file mode 100644 index 0000000..7f950d1 --- /dev/null +++ b/docs/MODEL_CARD.md @@ -0,0 +1,29 @@ +# Model card + +## Model + +Bidirectional LSTM sequence encoder with padding-aware attention and a next-item classification +head. The production artifact includes architecture parameters, item vocabulary, dataset ID, +ranking metrics, version, creation time, and checksums. + +## Intended use + +Ranking known catalogue items for users with at least one known recent interaction. It is not +intended for safety-critical decisions, eligibility, employment, credit, health, or other +high-impact domains. + +## Required evaluation + +Report Recall@K, NDCG@K, MRR@K, catalogue coverage, and the same metrics for a popularity baseline. +Add novelty, diversity, calibration, popularity bias, and cold-start cohorts before a real launch. + +## Limitations + +The dense output head scales linearly with catalogue size. Unknown items are rejected from the +sequence. Historical interactions may encode exposure and popularity bias. Offline ranking +metrics do not prove causal user or business impact. + +## Promotion policy + +A candidate needs a versioned dataset, reproducible configuration, compatible bundle, quality no +worse than the declared baseline, target-environment evidence, and an approved rollback plan. diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md new file mode 100644 index 0000000..561ca1a --- /dev/null +++ b/docs/OPERATIONS.md @@ -0,0 +1,25 @@ +# Operations runbook + +## Readiness failure + +1. Inspect startup logs for a missing manifest, checksum mismatch, vocabulary validation, or strict + state-dict errors. +2. Verify the complete immutable bundle is mounted at `MODEL_BUNDLE_PATH`. +3. Roll back to the previous image and bundle pair; never mix bundle files. + +## Elevated latency or fallback rate + +1. Check active requests, inference latency, cache hit rate, CPU, memory, and throttling. +2. Stop rollout growth and reduce admission capacity if saturation is increasing. +3. Roll back when the canary exceeds its error, fallback, or latency guardrail. + +## Quality regression + +Segment by model version, known-item rate, cold-start cohort, and catalogue segment. Validate +feature/vocabulary compatibility and input drift, then restore the previous bundle while retaining +impression and outcome evidence. + +## Corrupted artifact + +Checksum failures are non-retryable deployment errors. Quarantine the candidate, regenerate it +from the recorded dataset/configuration, and investigate storage or promotion integrity. diff --git a/docs/PRIVACY.md b/docs/PRIVACY.md new file mode 100644 index 0000000..243c38f --- /dev/null +++ b/docs/PRIVACY.md @@ -0,0 +1,12 @@ +# Privacy and feedback policy + +The API accepts user behavior that may be personal data. Raw user IDs are not emitted by the +feedback logger; a truncated SHA-256 pseudonym is used. Hashing is minimization, not anonymization. + +A production deployment must define lawful purpose, consent, access controls, encryption, +retention, deletion, regional processing, and prohibited sensitive attributes. Keep raw identifiers +in a controlled identity layer, use short-lived pseudonyms for modeling, and propagate deletion +requests to event, feature, training, cache, and model-lineage stores. + +Do not log raw click histories or API keys. Collect only fields required for evaluation and +learning, audit downstream exports, and do not reuse feedback for unrelated high-impact decisions. diff --git a/docs/architecture.md b/docs/architecture.md index 2153ceb..899bb36 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1,80 +1,16 @@ # Architecture -## Overview +The system separates four responsibilities: -DeepSequence Recommender is a production-grade, horizontally scalable deep-learning sequence -recommendation service built on the same architectural principles as -[TrojanChat](https://github.com/Trojan3877/TrojanChat). +1. **Learning:** validate events, split chronologically, create causal examples, train, and evaluate. +2. **Artifacts:** bind weights, vocabulary, architecture, lineage, metrics, and checksums. +3. **Serving:** verify the bundle, enforce request policy, infer, cache, and fall back under pressure. +4. **Operations:** expose health/metrics, collect minimized feedback, and roll out immutable pairs. -``` -Client (HTTP / gRPC) - โ†“ -Ingress / Load Balancer - โ†“ -Kubernetes Cluster - โ†“ -DeepSequence Pods (Replicas) - โ”œโ”€โ”€ FastAPI layer โ†’ /recommendations - โ”œโ”€โ”€ Model core โ†’ BiLSTM + Attention - โ””โ”€โ”€ Metrics โ†’ /metrics โ†’ Prometheus โ†’ Grafana - โ†“ -Redis (caching layer) -``` +The model is never loaded independently from its vocabulary. Production readiness depends on the +bundle, not merely on the process being alive. Liveness uses `/health`; model readiness uses +`/recommendations/health`. -## Design Principles - -### 1. Horizontal Scalability -- Stateless FastAPI workers โ€“ model weights are read-only at inference time. -- Redis for recommendation caching across replicas. -- Kubernetes HPA for CPU-based autoscaling. - -### 2. Observability First -- `/metrics` exposes Prometheus gauges, counters, and histograms. -- Structured Python logging with configurable log-level. -- Health endpoint (`GET /recommendations/health`) for liveness probes. - -### 3. Security by Design -- JWT token verification on protected endpoints. -- `SECRET_KEY` loaded from environment (never hard-coded). -- Bandit security scan enforced in CI. -- Non-root Docker runtime user. - -### 4. ML Architecture -``` -item_sequence โ†’ Embedding โ†’ BiLSTM โ†’ Attention โ†’ Linear โ†’ top-k items -``` -- **Embedding layer** โ€“ maps item IDs to dense vectors (dim=64 by default). -- **BiLSTM encoder** โ€“ captures forward and backward temporal dependencies. -- **Scaled dot-product attention** โ€“ weights encoder outputs by relevance. -- **Linear projection** โ€“ maps the attended context to item logits. - -### 5. CI/CD Pipeline -Every PR triggers: -1. `flake8` lint -2. `black` format check -3. `mypy` type checking -4. `bandit` security scan -5. `pytest` test suite - -## Component Map - -``` -app/ - โ”œโ”€โ”€ main.py โ€“ FastAPI app, startup, Prometheus mount - โ”œโ”€โ”€ core/ - โ”‚ โ”œโ”€โ”€ config.py โ€“ Pydantic settings - โ”‚ โ”œโ”€โ”€ model.py โ€“ DeepSequenceModel (BiLSTM + attention) - โ”‚ โ”œโ”€โ”€ data_processor.py โ€“ SequenceProcessor (vocab, padding, encoding) - โ”‚ โ”œโ”€โ”€ metrics.py โ€“ Prometheus metrics - โ”‚ โ””โ”€โ”€ security.py โ€“ JWT helpers - โ””โ”€โ”€ api/ - โ””โ”€โ”€ routes.py โ€“ /recommendations endpoints - -tests/ - โ””โ”€โ”€ test_recommender.py โ€“ unit tests - -k8s/ - โ”œโ”€โ”€ deployment.yaml - โ”œโ”€โ”€ service.yaml - โ””โ”€โ”€ hpa.yaml -``` +The in-process cache, rate limiter, and admission controller protect one replica. A multi-replica +deployment should enforce distributed policy at the gateway and use a governed event/cache system. +Hard inference cancellation requires process isolation or a dedicated model server. diff --git a/docs/demo_app.py b/docs/demo_app.py deleted file mode 100644 index fea5504..0000000 --- a/docs/demo_app.py +++ /dev/null @@ -1,142 +0,0 @@ -import streamlit as st -import pandas as pd -import numpy as np -import time - -# Page Configuration -st.set_page_config(page_title="DeepSequence Recommender System", page_icon="๐Ÿง ", layout="wide") - -st.title("๐Ÿง  DeepSequence-Recommender: Session-Based Predictive Engine") -st.caption("Advanced Machine Learning Showcase | Real-Time Recurrent Sequence Inference") - -# Mock Database of items across distinct domains (Tech & Media) to demonstrate sequence tracking -ITEM_POOL = { - "Electronics & Tech": [ - {"id": 101, "name": "Mechanical Keyboard (Tactile Browns)", "category": "Peripherals"}, - {"id": 102, "name": "Ultra-Wide 34-inch Curved Monitor", "category": "Displays"}, - {"id": 103, "name": "Ergonomic Vertical Mouse", "category": "Peripherals"}, - {"id": 104, "name": "USB-C Dual Display Docking Station", "category": "Connectivity"}, - {"id": 105, "name": "Premium Active Noise-Cancelling Headphones", "category": "Audio"}, - {"id": 106, "name": "High-Definition 4K Streaming Webcam", "category": "Peripherals"} - ], - "Streaming & Entertainment": [ - {"id": 201, "name": "Sci-Fi Cyberpunk Cyber-Thriller Series", "category": "Sci-Fi"}, - {"id": 202, "name": "Deep Space Exploration Documentary", "category": "Documentary"}, - {"id": 203, "name": "High-Stakes Artificial Intelligence Tech-Drama", "category": "Drama"}, - {"id": 204, "name": "Post-Apocalyptic Survival Feature Film", "category": "Sci-Fi"}, - {"id": 205, "name": "Neo-Noir Psychological Thriller", "category": "Thriller"} - ] -} - -# Initialize browser memory for user session interaction sequences -if "click_stream" not in st.session_state: - st.session_state.click_stream = [] - -# --- SIDEBAR: CLICK-STREAM SIMULATOR --- -st.sidebar.header("๐Ÿ›’ User Action Sequence Simulator") -st.sidebar.markdown("Build an interactive click-stream session history to test how the sequence vectors shift state.") - -domain = st.sidebar.selectbox("Select Target Domain Platform", list(ITEM_POOL.keys())) -available_items = ITEM_POOL[domain] - -item_options = {item["name"]: item for item in available_items} -selected_item_name = st.sidebar.selectbox("Select Item to Interact With", list(item_options.keys())) -action_type = st.sidebar.selectbox("Interaction Type", ["View Item Detail", "Add to Cart / Playlist", "Purchase"]) - -if st.sidebar.button("๐Ÿ“ฅ Inject Action Into Sequence"): - chosen_item = item_options[selected_item_name] - st.session_state.click_stream.append({ - "Step": len(st.session_state.click_stream) + 1, - "ItemID": chosen_item["id"], - "Item Name": chosen_item["name"], - "Category": chosen_item["category"], - "Action": action_type - }) - -if st.sidebar.button("๐Ÿ—‘๏ธ Clear Active Session History"): - st.session_state.click_stream = [] - st.rerun() - -# --- MODEL INFERENCE LAYER SIMULATION --- -def generate_sequence_recommendations(session_history, pool): - """ - Simulates hidden state generation from an RNN/GRU or Transformer layout. - Dynamically shifts weights based on the absolute last items interacted with. - """ - if not session_history: - return [] - - # Target intent extracted from the most recent item in the sequence - last_item = session_history[-1] - last_category = last_item["Category"] - last_id = last_item["ItemID"] - - # Flatten pool items for evaluation - all_candidates = [] - for items in pool.values(): - all_candidates.extend(items) - - recommendations = [] - for item in all_candidates: - if item["id"] == last_id: - continue # Exclude item currently being viewed - - # Calculate dynamic hidden score weights based on contextual transitions - base_score = 0.2 - if item["category"] == last_category: - base_score += 0.55 # Local category affinity - else: - base_score += np.random.uniform(0.05, 0.15) # Global exploration bias - - # Add slight decay if session sequence length is long to simulate temporal dampening - final_probability = float(np.clip(base_score + np.random.uniform(-0.05, 0.05), 0.01, 0.99)) - - recommendations.append({ - "Recommended Item ID": item["id"], - "Item Name": item["name"], - "Category": item["category"], - "Match Probability Score": final_probability - }) - - # Sort by descending inference match probabilities - recommendations = sorted(recommendations, key=lambda x: x["Match Probability Score"], reverse=True) - return recommendations[:4] - -# --- MAIN DASHBOARD VIEW --- -col_left, col_right = st.columns([1, 1]) - -with col_left: - st.subheader("โฑ๏ธ Live User Session Sequence State") - if st.session_state.click_stream: - df_history = pd.DataFrame(st.session_state.click_stream) - st.dataframe(df_history, use_container_width=True, hide_index=True) - else: - st.info("The recurrent neural network state vector is empty. Use the sidebar simulator panel to add user clicks to the temporal stream.") - -with col_right: - st.subheader("๐Ÿ”ฎ Deep Next-Item Recommendations (Top-K)") - if st.session_state.click_stream: - with st.spinner("Processing sequence tokens through neural lookup arrays..."): - time.sleep(0.4) # Simulate network/inference forward pass latency - recs = generate_sequence_recommendations(st.session_state.click_stream, ITEM_POOL) - df_recs = pd.DataFrame(recs) - - # Render visual performance bars for confidence thresholds - st.dataframe( - df_recs.style.background_gradient(cmap="Blues", subset=["Match Probability Score"]), - use_container_width=True, - hide_index=True - ) - - st.caption("๐Ÿ’ก Notice how modifying your click history shifts prediction distributions to capture dynamic user transitions instantly.") - else: - st.text("Awaiting sequence input vectors to calculate tensor dot-products...") - -# Add technical architecture description for reading recruiters -st.markdown("---") -st.subheader("โš™๏ธ Portfolio Technical Specification Overview") -st.markdown(""" -This production-grade simulation maps the performance capabilities of the underlying **DeepSequence-Recommender** framework. -* **Temporal Modeling:** Tracks long-term vs. short-term intent transitions using embedding sequences instead of basic static matrix factorization. -* **Cold-Start Resilience:** Integrates auxiliary metadata features (Categories/Interactions) directly into the embedding layers to ensure accurate recommendations even within short user sessions. -""") \ No newline at end of file diff --git a/docs/metrics.md b/docs/metrics.md index dae3ac4..ce41f03 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -1,33 +1,17 @@ # Metrics -DeepSequence Recommender exposes a Prometheus scrape endpoint at `GET /metrics`. +Prometheus metrics are exposed at `GET /metrics`. -## Exposed Metrics +| Metric | Meaning | +|---|---| +| `deepseq_active_requests` | In-flight model requests admitted by this replica | +| `deepseq_recommendations_total{status}` | Success, cache, fallback, overload, and error outcomes | +| `deepseq_recommendation_latency_seconds` | End-to-end admitted request latency | +| `deepseq_model_inference_latency_seconds` | PyTorch inference duration | +| `deepseq_cache_hits_total` / `deepseq_cache_misses_total` | Replica-local cache behavior | +| `deepseq_feedback_events_total{event_type}` | Accepted learning-feedback events | -| Metric | Type | Description | -|--------|------|-------------| -| `deepseq_active_requests` | Gauge | Currently in-flight recommendation requests | -| `deepseq_recommendations_total{status}` | Counter | Total recommendations served (labelled `success` or `error`) | -| `deepseq_recommendation_latency_seconds` | Histogram | End-to-end request latency | -| `deepseq_model_inference_latency_seconds` | Histogram | Neural model forward-pass latency | -| `deepseq_cache_hits_total` | Counter | Redis cache hits | -| `deepseq_cache_misses_total` | Counter | Redis cache misses | - -## Prometheus Scrape Config - -```yaml -scrape_configs: - - job_name: 'deepsequence-recommender' - static_configs: - - targets: ['deepsequence-service:8000'] -``` - -## SLO Targets - -| SLO | Target | -|-----|--------| -| p50 recommendation latency | < 50 ms | -| p99 recommendation latency | < 250 ms | -| Model inference p50 latency | < 20 ms | -| Availability | โ‰ฅ 99.9 % | -| Error rate | < 0.1 % | +No availability SLO is claimed by the repository. Operators should establish objectives only after +target-environment load and failure testing. Recommended alert dimensions include model version, +error/fallback rate, saturation, p95/p99 latency, unknown-item rate, drift, delayed-label ranking +quality, coverage, and experiment guardrails. diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index adec24c..3087ec7 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -16,17 +16,23 @@ spec: spec: containers: - name: deepsequence - image: your-docker-registry/deepsequence-recommender:latest + image: your-docker-registry/deepsequence-recommender:1.1.0 ports: - containerPort: 8000 env: - - name: SECRET_KEY + - name: API_KEY valueFrom: secretKeyRef: name: deepsequence-secrets - key: secret-key - - name: REDIS_URL - value: "redis://deepsequence-redis:6379" + key: api-key + - name: ENVIRONMENT + value: "production" + - name: MODEL_BUNDLE_PATH + value: "/models/current" + volumeMounts: + - name: model-bundle + mountPath: /models/current + readOnly: true readinessProbe: httpGet: path: /recommendations/health @@ -35,10 +41,16 @@ spec: periodSeconds: 5 livenessProbe: httpGet: - path: /recommendations/health + path: /health port: 8000 initialDelaySeconds: 20 periodSeconds: 10 + startupProbe: + httpGet: + path: /recommendations/health + port: 8000 + failureThreshold: 30 + periodSeconds: 5 resources: requests: cpu: "250m" @@ -46,6 +58,18 @@ spec: limits: cpu: "1000m" memory: "2Gi" + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumes: + - name: model-bundle + persistentVolumeClaim: + claimName: deepsequence-model-bundle securityContext: runAsNonRoot: true - runAsUser: 1000 + runAsUser: 10001 + runAsGroup: 10001 + seccompProfile: + type: RuntimeDefault diff --git a/k8s/model-pvc.yaml b/k8s/model-pvc.yaml new file mode 100644 index 0000000..29f29f1 --- /dev/null +++ b/k8s/model-pvc.yaml @@ -0,0 +1,10 @@ +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: deepsequence-model-bundle +spec: + accessModes: + - ReadOnlyMany + resources: + requests: + storage: 1Gi diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..e257a74 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,16 @@ +[project] +name = "deepsequence-recommender" +version = "1.1.0" +requires-python = ">=3.11" + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["."] + +[tool.ruff] +line-length = 100 +target-version = "py311" + +[tool.coverage.run] +source = ["app", "src"] +omit = ["tests/*"] diff --git a/recommender/guards.py b/recommender/guards.py deleted file mode 100644 index 450871d..0000000 --- a/recommender/guards.py +++ /dev/null @@ -1,27 +0,0 @@ -# recommender/guards.py -class SLAThresholdException(Exception): - """Raised when deep learning tensor compilation breaches strict latency limits.""" - pass - -class RecommenderGuardrail: - """ - Protects downstream inference nodes from out-of-memory (OOM) tracking faults - or exploding user sequence click paths. - """ - def __init__(self, max_sequence_window: int = 20, max_latency_ms: float = 200.0): - self.max_sequence_window = max_sequence_window - self.max_latency_ms = max_latency_ms - - def enforce_window_bounds(self, items: List[int]) -> List[int]: - """Truncates input paths to ensure O(1) processing windows.""" - if len(items) > self.max_sequence_window: - # Shift window to grab the latest user intent actions safely - return items[-self.max_sequence_window:] - return items - - def verify_sla_compliance(self, active_duration_ms: float): - """Trips a circuit alert if model matrix calculation takes too long.""" - if active_duration_ms > self.max_latency_ms: - raise SLAThresholdException( - f"SLA Violation: RecSys processing took {active_duration_ms:.2f}ms (Limit: {self.max_latency_ms}ms)" - ) diff --git a/recommender/orchestrator.py b/recommender/orchestrator.py deleted file mode 100644 index 01d0bec..0000000 --- a/recommender/orchestrator.py +++ /dev/null @@ -1,56 +0,0 @@ -# recommender/orchestrator.py -import time -import random -from .state import SessionState -from .guards import RecommenderGuardrail, SLAThresholdException - -class SequentialInferenceEngine: - """ - Orchestrates sequential feature processing and deep-learning recommendations - using defensive performance bounds. - """ - def __init__(self): - self.guard = RecommenderGuardrail(max_sequence_window=10, max_latency_ms=50.0) - - def generate_next_item_predictions(self, session_id: str, clickstream: List[int]) -> SessionState: - state = SessionState(session_id=session_id, raw_item_history=clickstream) - state = state.log_trace("Inference lifecycle initiated.") - - start_time = time.time() - try: - # 1. Enforce safe tensor shape windowing - bounded_history = self.guard.enforce_window_bounds(state.raw_item_history) - state.sequence_length = len(bounded_history) - state = state.log_trace(f"Sequence windowed to length: {state.sequence_length}") - - # Simulated Deep Learning Tensor Transform & Prediction Sequence - # (Replace with model.predict(padded_tensor) calls) - time.sleep(0.015) # Simulated matrix computation latency - - if not bounded_history: - raise ValueError("Clickstream matrix sequence contains no elements.") - - # Compute latency and verify SLA bounds - elapsed_ms = (time.time() - start_time) * 1000 - self.guard.verify_sla_compliance(elapsed_ms) - - # Populate structured prediction arrays - mock_preds = [random.randint(100, 999) for _ in range(5)] - mock_scores = sorted([random.random() for _ in range(5)], reverse=True) - - return state.model_copy(update={ - "processed_tensor_input": [float(x) for x in bounded_history], - "top_k_recommendations": mock_preds, - "prediction_confidence": mock_scores, - "inference_latency_ms": elapsed_ms - }).log_trace("Inference vector generation complete.") - - except (SLAThresholdException, Exception) as e: - # Graceful Degradation: Fallback instantly to global popular items - elapsed_ms = (time.time() - start_time) * 1000 - fallback_popular_items = [101, 102, 103, 104, 105] # Static fallback - return state.model_copy(update={ - "fallback_triggered": True, - "top_k_recommendations": fallback_popular_items, - "inference_latency_ms": elapsed_ms - }).log_trace(f"FALLBACK ACTIVATED: {str(e)}") diff --git a/recommender/state.py b/recommender/state.py deleted file mode 100644 index f19d6e9..0000000 --- a/recommender/state.py +++ /dev/null @@ -1,23 +0,0 @@ -# recommender/state.py -from pydantic import BaseModel, Field -from typing import List, Optional - -class SessionState(BaseModel): - """ - The immutable token tracking contract passed across the inference pipeline. - Enforces structural context bounds on raw clickstream sequences. - """ - session_id: str - raw_item_history: List[int] = Field(default_factory=list) - processed_tensor_input: Optional[List[float]] = None - top_k_recommendations: List[int] = Field(default_factory=list) - prediction_confidence: List[float] = Field(default_factory=list) - - # Operational Observability - sequence_length: int = 0 - inference_latency_ms: float = 0.0 - fallback_triggered: bool = False - execution_trace: List[str] = Field(default_factory=list) - - def log_trace(self, message: str) -> "SessionState": - return self.model_copy(update={"execution_steps": list(self.execution_trace) + [message]}) diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..55988a9 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,5 @@ +-r requirements.txt +pytest>=8,<10 +pytest-cov>=5,<8 +ruff>=0.9,<1 +pip-audit>=2.7,<3 diff --git a/requirements.txt b/requirements.txt index ab83480..5b3a3ac 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,17 +1,13 @@ -# Machine Learning & Deep Sequence Core Libraries -torch>=2.2.0 -numpy>=1.24.2 -pandas>=2.2.1 - -# High Performance Serialization Format Models -onnx>=1.15.0 -onnxruntime>=1.17.0 - -# User Visualization & Diagnostics Telemetry Panels -streamlit>=1.30.0 -matplotlib>=3.8.2 -seaborn>=0.13.2 -fastapi>=0.100.0 -httpx>=0.24.0 -prometheus_client>=0.17.0 -pydantic-settings>=2.0.0 +torch>=2.2,<3 +numpy>=1.24,<3 +pandas>=2.2,<3 +onnx>=1.15,<2 +onnxruntime>=1.17,<2 +fastapi>=0.115,<1 +uvicorn[standard]>=0.30,<1 +httpx>=0.27,<1 +prometheus-client>=0.20,<1 +pydantic-settings>=2.7,<3 +streamlit>=1.40,<2 +matplotlib>=3.8,<4 +seaborn>=0.13,<1 diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 0000000..bfd9be3 --- /dev/null +++ b/scripts/__init__.py @@ -0,0 +1 @@ +"""Repository automation entry points.""" diff --git a/scripts/generate_demo_data.py b/scripts/generate_demo_data.py new file mode 100644 index 0000000..4bcab56 --- /dev/null +++ b/scripts/generate_demo_data.py @@ -0,0 +1,42 @@ +"""Generate a deterministic interaction dataset for local pipeline validation.""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timedelta, timezone + +from src.training.data import InteractionEvent, write_jsonl + + +def generate_events(sessions: int = 20, events_per_session: int = 6) -> list[InteractionEvent]: + events = [] + started = datetime(2026, 1, 1, tzinfo=timezone.utc) + index = 0 + for session_index in range(sessions): + for position in range(events_per_session): + item_index = (session_index + position) % 12 + events.append( + InteractionEvent( + event_id=f"event-{index}", + user_id=f"user-{session_index % 5}", + session_id=f"session-{session_index}", + item_id=f"item-{item_index}", + event_type="purchase" if position == events_per_session - 1 else "click", + timestamp=started + timedelta(minutes=index), + ) + ) + index += 1 + return events + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output", default="data/demo-events.jsonl") + args = parser.parse_args() + events = generate_events() + write_jsonl(args.output, events) + print(f"Wrote {len(events)} events to {args.output}") + + +if __name__ == "__main__": + main() diff --git a/src/features/online_retrieval.py b/src/features/online_retrieval.py deleted file mode 100644 index db2fbb3..0000000 --- a/src/features/online_retrieval.py +++ /dev/null @@ -1,33 +0,0 @@ -import time -import numpy as np - -class LatencyIsolatedFeatureStore: - """ - Implements a fast dual-storage retrieval mock pattern - optimized for sub-2ms historical state sequence assemblies. - """ - def __init__(self): - # Mock in-memory low-latency hash map key storage - self.redis_in_memory_cache = { - "user_8271": [102, 405, 12, 994, 58, 230, 41, 88], - "user_1194": [22, 940, 503, 112, 8], - "user_4952": [1094, 2390, 44, 12, 853, 901, 33, 442, 129] - } - self.vocab_size = 5000 - self.target_sequence_len = 50 - - def fetch_user_realtime_sequence(self, user_id: str): - start_time = time.time() - - # Look up rolling click event tokens from the in-memory cache - raw_sequence = self.redis_in_memory_cache.get(user_id, []) - - # Fail-safe padding: align the variable-length history to a fixed input window - if len(raw_sequence) < self.target_sequence_len: - pad_size = self.target_sequence_len - len(raw_sequence) - padded_sequence = [0] * pad_size + raw_sequence - else: - padded_sequence = raw_sequence[-self.target_sequence_len:] - - retrieval_latency = (time.time() - start_time) * 1000 - return np.array([padded_sequence], dtype=np.int64), retrieval_latency diff --git a/src/serving/onnx_exporter.py b/src/serving/onnx_exporter.py index 9ff832e..c5636f1 100644 --- a/src/serving/onnx_exporter.py +++ b/src/serving/onnx_exporter.py @@ -1,53 +1,46 @@ -import os +"""Export the verified production bundleโ€”not a mock architectureโ€”to ONNX.""" + +from __future__ import annotations + +import argparse +from pathlib import Path + import torch -import numpy as np - -class MockSequentialModel(torch.nn.Module): - """ - A structural representation of a Deep Sequential Recommendation network - (such as a Transformer, GRU, or SASRec architecture variant). - """ - def __init__(self, vocab_size=5000, embedding_dim=128, sequence_length=50): - super().__init__() - self.item_embeddings = torch.nn.Embedding(vocab_size, embedding_dim) - self.sequence_layer = torch.nn.GRU(embedding_dim, embedding_dim, batch_first=True) - self.scoring_head = torch.nn.Linear(embedding_dim, vocab_size) - - def forward(self, input_sequences): - # input_sequences shape: [Batch Size, Sequence Length] - embedded = self.item_embeddings(input_sequences) - gru_out, _ = self.sequence_layer(embedded) - # Pull down the absolute final hidden sequence state embedding token [Batch Size, Embedding Dim] - final_state = gru_out[:, -1, :] - logits = self.scoring_head(final_state) - return logits - -def export_to_onnx_runtime(output_path="models/deep_sequence_rec.onnx"): - os.makedirs(os.path.dirname(output_path), exist_ok=True) - - # Initialize the structural network layout - model = MockSequentialModel() - model.eval() - # Generate model input constraints (Batch Size: 1, Evaluation Sequence Window Length: 50) - dummy_input = torch.randint(0, 5000, (1, 50), dtype=torch.long) +from app.core.artifacts import load_bundle + - print(f"๐Ÿš€ Serializing target Deep Sequence model matrix to: {output_path}...") +def export_bundle_to_onnx( + bundle_path: str | Path, + output_path: str | Path, + opset_version: int = 17, +) -> Path: + processor, model, _manifest = load_bundle(bundle_path) + model.eval() + destination = Path(output_path) + destination.parent.mkdir(parents=True, exist_ok=True) + example = torch.ones((1, processor.max_length), dtype=torch.long) torch.onnx.export( model, - dummy_input, - output_path, + example, + destination, export_params=True, - opset_version=17, + opset_version=opset_version, do_constant_folding=True, - input_names=['input_sequences'], - output_names=['output_logits'], - dynamic_axes={ - 'input_sequences': {0: 'batch_size'}, - 'output_logits': {0: 'batch_size'} - } + dynamo=False, + input_names=["item_sequence"], + output_names=["logits"], ) - print("โœ… Model deployment serialization matrix compilation complete.") + return destination + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--bundle", default="models/current") + parser.add_argument("--output", default="models/deep_sequence_rec.onnx") + args = parser.parse_args() + print(export_bundle_to_onnx(args.bundle, args.output)) + if __name__ == "__main__": - export_to_onnx_runtime() + main() diff --git a/src/serving/triton_config.pbtxt b/src/serving/triton_config.pbtxt deleted file mode 100644 index 69cb2b3..0000000 --- a/src/serving/triton_config.pbtxt +++ /dev/null @@ -1,30 +0,0 @@ -name: "deep_sequence_rec" -platform: "onnxruntime_onnx" -max_batch_size: 64 - -input [ - { - name: "input_sequences" - data_type: TYPE_INT64 - dims: [ 50 ] - } -] -output [ - { - name: "output_logits" - data_type: TYPE_FP32 - dims: [ 5000 ] - } -] - -dynamic_batching { - max_queue_delay_microseconds: 2000 - preferred_batch_size: [ 8, 16, 32, 64 ] -} - -instance_group [ - { - count: 2 - kind: KIND_GPU - } -] diff --git a/src/training/__init__.py b/src/training/__init__.py new file mode 100644 index 0000000..820e949 --- /dev/null +++ b/src/training/__init__.py @@ -0,0 +1 @@ +"""Reproducible training and evaluation pipeline.""" diff --git a/src/training/baselines.py b/src/training/baselines.py new file mode 100644 index 0000000..501c724 --- /dev/null +++ b/src/training/baselines.py @@ -0,0 +1,19 @@ +"""Auditable non-neural baselines used as model promotion gates.""" + +from __future__ import annotations + +from collections import Counter + + +class PopularityBaseline: + def __init__(self) -> None: + self.ranking: list[str] = [] + + def fit(self, sequences: list[list[str]]) -> "PopularityBaseline": + counts = Counter(item for sequence in sequences for item in sequence) + self.ranking = [item for item, _ in counts.most_common()] + return self + + def recommend(self, seen: list[str], top_k: int) -> list[str]: + excluded = set(seen) + return [item for item in self.ranking if item not in excluded][:top_k] diff --git a/src/training/data.py b/src/training/data.py new file mode 100644 index 0000000..cd65250 --- /dev/null +++ b/src/training/data.py @@ -0,0 +1,85 @@ +"""Versioned interaction contracts and leakage-resistant sequence datasets.""" + +from __future__ import annotations + +from collections import defaultdict +from datetime import datetime +from pathlib import Path +from typing import Literal + +from pydantic import BaseModel, Field + + +class InteractionEvent(BaseModel): + schema_version: Literal["1.0"] = "1.0" + event_id: str = Field(min_length=1, max_length=128) + user_id: str = Field(min_length=1, max_length=128) + session_id: str = Field(min_length=1, max_length=128) + item_id: str = Field(min_length=1, max_length=256) + event_type: Literal["view", "click", "cart", "purchase"] + timestamp: datetime + + +def load_jsonl(path: str | Path) -> list[InteractionEvent]: + events: list[InteractionEvent] = [] + with Path(path).open(encoding="utf-8") as handle: + for line_number, line in enumerate(handle, start=1): + if line.strip(): + try: + events.append(InteractionEvent.model_validate_json(line)) + except ValueError as exc: + raise ValueError(f"Invalid event at line {line_number}") from exc + if not events: + raise ValueError("Dataset contains no interaction events") + event_ids = [event.event_id for event in events] + if len(event_ids) != len(set(event_ids)): + raise ValueError("event_id values must be unique") + return sorted(events, key=lambda event: event.timestamp) + + +def dataset_id(events: list[InteractionEvent]) -> str: + import hashlib + + canonical = "\n".join(event.model_dump_json() for event in events) + return hashlib.sha256(canonical.encode()).hexdigest()[:16] + + +def temporal_split( + events: list[InteractionEvent], train_fraction: float = 0.7, validation_fraction: float = 0.15 +) -> tuple[list[InteractionEvent], list[InteractionEvent], list[InteractionEvent]]: + if not 0 < train_fraction < 1 or not 0 <= validation_fraction < 1: + raise ValueError("Split fractions are outside valid bounds") + if train_fraction + validation_fraction >= 1: + raise ValueError("Training and validation fractions must sum to less than one") + ordered = sorted(events, key=lambda event: event.timestamp) + train_end = max(1, int(len(ordered) * train_fraction)) + validation_end = max(train_end + 1, int(len(ordered) * (train_fraction + validation_fraction))) + return ordered[:train_end], ordered[train_end:validation_end], ordered[validation_end:] + + +def session_sequences(events: list[InteractionEvent]) -> list[list[str]]: + sessions: dict[str, list[InteractionEvent]] = defaultdict(list) + for event in events: + sessions[event.session_id].append(event) + return [ + [event.item_id for event in sorted(session, key=lambda item: item.timestamp)] + for session in sessions.values() + if len(session) >= 2 + ] + + +def next_item_examples(sequences: list[list[str]]) -> list[tuple[list[str], str]]: + return [ + (sequence[:index], sequence[index]) + for sequence in sequences + for index in range(1, len(sequence)) + ] + + +def write_jsonl(path: str | Path, events: list[InteractionEvent]) -> None: + destination = Path(path) + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text( + "\n".join(event.model_dump_json() for event in events) + "\n", + encoding="utf-8", + ) diff --git a/src/training/metrics.py b/src/training/metrics.py new file mode 100644 index 0000000..b4cfd17 --- /dev/null +++ b/src/training/metrics.py @@ -0,0 +1,45 @@ +"""Ranking metrics with explicit definitions and no framework dependency.""" + +from __future__ import annotations + +import math + + +def recall_at_k(predictions: list[list[str]], targets: list[str]) -> float: + return sum(target in predicted for predicted, target in zip(predictions, targets)) / len(targets) + + +def mrr_at_k(predictions: list[list[str]], targets: list[str]) -> float: + reciprocal_ranks = [] + for predicted, target in zip(predictions, targets): + reciprocal_ranks.append( + 1 / (predicted.index(target) + 1) if target in predicted else 0.0 + ) + return sum(reciprocal_ranks) / len(targets) + + +def ndcg_at_k(predictions: list[list[str]], targets: list[str]) -> float: + gains = [] + for predicted, target in zip(predictions, targets): + gains.append( + 1 / math.log2(predicted.index(target) + 2) if target in predicted else 0.0 + ) + return sum(gains) / len(targets) + + +def catalogue_coverage(predictions: list[list[str]], catalogue: set[str]) -> float: + recommended = {item for prediction in predictions for item in prediction} + return len(recommended & catalogue) / len(catalogue) if catalogue else 0.0 + + +def evaluate_ranking( + predictions: list[list[str]], targets: list[str], catalogue: set[str] +) -> dict[str, float]: + if not targets or len(predictions) != len(targets): + raise ValueError("Predictions and non-empty targets must have equal lengths") + return { + "recall_at_k": recall_at_k(predictions, targets), + "ndcg_at_k": ndcg_at_k(predictions, targets), + "mrr_at_k": mrr_at_k(predictions, targets), + "catalogue_coverage": catalogue_coverage(predictions, catalogue), + } diff --git a/src/training/train.py b/src/training/train.py new file mode 100644 index 0000000..b724b7a --- /dev/null +++ b/src/training/train.py @@ -0,0 +1,189 @@ +"""Train, evaluate, and package the production sequence recommender.""" + +from __future__ import annotations + +import argparse +import json +import random +from datetime import datetime, timezone +from pathlib import Path + +import numpy as np +import torch +import torch.nn.functional as F + +from app.core.artifacts import ModelManifest, save_bundle +from app.core.data_processor import SequenceProcessor +from app.core.model import DeepSequenceModel +from src.training.baselines import PopularityBaseline +from src.training.data import ( + dataset_id, + load_jsonl, + next_item_examples, + session_sequences, + temporal_split, +) +from src.training.metrics import evaluate_ranking + + +def seed_everything(seed: int) -> None: + random.seed(seed) + np.random.seed(seed) + torch.manual_seed(seed) + + +def train_model( + train_sequences: list[list[str]], + *, + embedding_dim: int = 32, + hidden_dim: int = 64, + num_layers: int = 1, + max_sequence_length: int = 50, + epochs: int = 3, + learning_rate: float = 1e-3, + seed: int = 7, +) -> tuple[SequenceProcessor, DeepSequenceModel]: + seed_everything(seed) + processor = SequenceProcessor(max_length=max_sequence_length).fit(train_sequences) + if processor.vocab_size < 2: + raise ValueError("Training requires at least two catalogue items") + examples = next_item_examples(train_sequences) + model = DeepSequenceModel( + num_items=processor.vocab_size, + embedding_dim=embedding_dim, + hidden_dim=hidden_dim, + num_layers=num_layers, + ) + optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate) + model.train() + for _ in range(epochs): + for history, target in examples: + target_index = processor.item_to_idx(target) + if target_index == 0: + continue + optimizer.zero_grad() + logits = model(processor.to_tensor(history)) + loss = F.cross_entropy(logits, torch.tensor([target_index])) + loss.backward() + optimizer.step() + model.eval() + return processor, model + + +def model_predictions( + model: DeepSequenceModel, + processor: SequenceProcessor, + examples: list[tuple[list[str], str]], + top_k: int, +) -> tuple[list[list[str]], list[str]]: + predictions: list[list[str]] = [] + targets: list[str] = [] + for history, target in examples: + if processor.item_to_idx(target) == 0: + continue + known_history = [item for item in history if processor.item_to_idx(item) != 0] + if not known_history: + continue + eligible_top_k = min( + top_k, processor.vocab_size - len(set(known_history)) + ) + if eligible_top_k < 1: + continue + indices = model.recommend( + processor.to_tensor(known_history), + top_k=eligible_top_k, + exclude_ids=[processor.item_to_idx(item) for item in known_history], + ) + predictions.append( + [item for item in processor.decode_recommendations(indices) if item is not None] + ) + targets.append(target) + return predictions, targets + + +def run_training( + dataset_path: str | Path, + output_directory: str | Path, + *, + epochs: int = 3, + top_k: int = 10, + seed: int = 7, +) -> dict[str, object]: + events = load_jsonl(dataset_path) + train_events, validation_events, test_events = temporal_split(events) + train_sequences = session_sequences(train_events) + validation_examples = next_item_examples(session_sequences(validation_events)) + test_examples = next_item_examples(session_sequences(test_events)) + processor, model = train_model(train_sequences, epochs=epochs, seed=seed) + + baseline = PopularityBaseline().fit(train_sequences) + catalogue = set(processor.export_vocabulary()) + eligible_validation = [ + (history, target) + for history, target in validation_examples + if processor.item_to_idx(target) != 0 + and any(processor.item_to_idx(item) != 0 for item in history) + ] + validation_predictions, validation_targets = model_predictions( + model, processor, eligible_validation, top_k + ) + if not validation_targets: + raise ValueError("Validation split has no evaluable known-item examples") + neural_metrics = evaluate_ranking(validation_predictions, validation_targets, catalogue) + baseline_predictions = [ + baseline.recommend(history, top_k) for history, _target in eligible_validation + ] + baseline_metrics = evaluate_ranking( + baseline_predictions, validation_targets, catalogue + ) + + test_predictions, test_targets = model_predictions(model, processor, test_examples, top_k) + test_metrics = ( + evaluate_ranking(test_predictions, test_targets, catalogue) if test_targets else {} + ) + version = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + config = { + "embedding_dim": 32, + "hidden_dim": 64, + "num_layers": 1, + "dropout": 0.3, + "padding_idx": 0, + "max_sequence_length": 50, + } + manifest = ModelManifest( + model_version=version, + created_at=datetime.now(timezone.utc).isoformat(), + architecture_config=config, + metrics={f"validation_{key}": value for key, value in neural_metrics.items()}, + dataset_id=dataset_id(events), + weights_sha256="pending", + vocabulary_sha256="pending", + ) + completed = save_bundle(output_directory, model, processor, manifest) + report = { + "model_version": version, + "dataset_id": completed.dataset_id, + "validation": neural_metrics, + "popularity_baseline": baseline_metrics, + "test": test_metrics, + "promotion_eligible": neural_metrics["ndcg_at_k"] >= baseline_metrics["ndcg_at_k"], + } + (Path(output_directory) / "evaluation.json").write_text( + json.dumps(report, indent=2), encoding="utf-8" + ) + return report + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--dataset", required=True) + parser.add_argument("--output", default="models/candidate") + parser.add_argument("--epochs", type=int, default=3) + parser.add_argument("--top-k", type=int, default=10) + parser.add_argument("--seed", type=int, default=7) + args = parser.parse_args() + print(json.dumps(run_training(args.dataset, args.output, epochs=args.epochs, top_k=args.top_k, seed=args.seed), indent=2)) + + +if __name__ == "__main__": + main() diff --git a/streamlit_app.py b/streamlit_app.py index 748b316..e20cf02 100644 --- a/streamlit_app.py +++ b/streamlit_app.py @@ -9,8 +9,12 @@ from __future__ import annotations +from pathlib import Path + import streamlit as st +import torch +from app.core.artifacts import load_bundle from app.core.config import settings from app.core.data_processor import SequenceProcessor from app.core.model import DeepSequenceModel @@ -35,8 +39,13 @@ @st.cache_resource(show_spinner="Loading recommendation modelโ€ฆ") -def load_model() -> tuple[SequenceProcessor, DeepSequenceModel]: - """Initialise processor and model with the demo catalogue.""" +def load_model() -> tuple[SequenceProcessor, DeepSequenceModel, str, bool]: + """Load a trained bundle or an explicitly labelled development model.""" + bundle = Path(settings.model_bundle_path) + if (bundle / "manifest.json").is_file(): + processor, model, manifest = load_bundle(bundle) + return processor, model, manifest.model_version, True + torch.manual_seed(7) processor = SequenceProcessor(max_length=settings.max_sequence_length) demo_items = [[f"item_{i}" for i in range(DEMO_CATALOGUE_SIZE)]] processor.fit(demo_items) @@ -47,10 +56,11 @@ def load_model() -> tuple[SequenceProcessor, DeepSequenceModel]: hidden_dim=settings.hidden_dim, num_layers=settings.num_layers, ) - return processor, model + model.eval() + return processor, model, "development-untrained", False -processor, model = load_model() +processor, model, model_version, trained = load_model() # --------------------------------------------------------------------------- # UI @@ -61,6 +71,10 @@ def load_model() -> tuple[SequenceProcessor, DeepSequenceModel]: "A **sequence-aware** deep learning recommender powered by a " "Bidirectional LSTM + Attention architecture." ) +if not trained: + st.warning( + "Development mode: this seeded model is untrained and does not represent ranking quality." + ) st.divider() # Sidebar: model info @@ -71,6 +85,7 @@ def load_model() -> tuple[SequenceProcessor, DeepSequenceModel]: st.metric("Hidden dim", settings.hidden_dim) st.metric("LSTM layers", settings.num_layers) st.metric("Max sequence length", settings.max_sequence_length) + st.caption(f"Model version: {model_version}") st.divider() st.caption("Architecture: BiLSTM + scaled dot-product attention โ†’ top-k items") diff --git a/tests/test_artifacts.py b/tests/test_artifacts.py new file mode 100644 index 0000000..808a36e --- /dev/null +++ b/tests/test_artifacts.py @@ -0,0 +1,59 @@ +"""Model-bundle integrity and ranking-evaluation tests.""" + +from datetime import datetime, timezone + +import pytest +import onnx + +from app.core.artifacts import ModelManifest, load_bundle, save_bundle +from app.core.data_processor import SequenceProcessor +from app.core.model import DeepSequenceModel +from src.training.metrics import evaluate_ranking +from src.serving.onnx_exporter import export_bundle_to_onnx + + +def test_model_bundle_roundtrip_and_checksum(tmp_path) -> None: + processor = SequenceProcessor(max_length=5).fit([["a", "b", "c"]]) + model = DeepSequenceModel( + num_items=processor.vocab_size, + embedding_dim=8, + hidden_dim=8, + num_layers=1, + ) + manifest = ModelManifest( + model_version="test-v1", + created_at=datetime.now(timezone.utc).isoformat(), + architecture_config={ + "embedding_dim": 8, + "hidden_dim": 8, + "num_layers": 1, + "dropout": 0.3, + "padding_idx": 0, + "max_sequence_length": 5, + }, + dataset_id="dataset-test", + weights_sha256="pending", + vocabulary_sha256="pending", + ) + save_bundle(tmp_path, model, processor, manifest) + + restored_processor, restored_model, restored_manifest = load_bundle(tmp_path) + assert restored_manifest.model_version == "test-v1" + assert restored_processor.export_vocabulary() == processor.export_vocabulary() + assert restored_model.recommend(restored_processor.to_tensor(["a"]), top_k=2) + + onnx_path = export_bundle_to_onnx(tmp_path, tmp_path / "model.onnx") + onnx.checker.check_model(onnx.load(onnx_path)) + + (tmp_path / "vocabulary.json").write_text("{}", encoding="utf-8") + with pytest.raises(ValueError, match="checksum"): + load_bundle(tmp_path) + + +def test_ranking_metrics_have_known_values() -> None: + metrics = evaluate_ranking( + [["a", "b"], ["c", "d"]], ["a", "d"], {"a", "b", "c", "d"} + ) + assert metrics["recall_at_k"] == 1.0 + assert metrics["mrr_at_k"] == 0.75 + assert metrics["catalogue_coverage"] == 1.0 diff --git a/tests/test_benchmark.py b/tests/test_benchmark.py new file mode 100644 index 0000000..61c3dc6 --- /dev/null +++ b/tests/test_benchmark.py @@ -0,0 +1,7 @@ +from benchmarks.inference import run_benchmark + + +def test_benchmark_emits_reproducible_contract() -> None: + report = run_benchmark(iterations=3, warmup=1) + assert report["workload"]["iterations"] == 3 + assert report["results"]["latency_ms_p95"] > 0 diff --git a/tests/test_recommender.py b/tests/test_recommender.py index 39a99d6..6ae7118 100644 --- a/tests/test_recommender.py +++ b/tests/test_recommender.py @@ -3,6 +3,7 @@ from __future__ import annotations import pytest +import torch from app.core.data_processor import SequenceProcessor from app.core.model import DeepSequenceModel @@ -98,7 +99,8 @@ class TestDeepSequenceModel: def test_forward_shape(self, model: DeepSequenceModel, processor: SequenceProcessor) -> None: tensor = processor.to_tensor(["a", "b", "c"]) logits = model(tensor) - assert logits.shape == (1, processor.vocab_size) + assert logits.shape == (1, processor.vocab_size + 1) + assert logits[0, 0].isneginf() def test_recommend_returns_top_k( self, model: DeepSequenceModel, processor: SequenceProcessor @@ -121,3 +123,19 @@ def test_recommend_no_duplicates( tensor = processor.to_tensor(["a", "b"]) recs = model.recommend(tensor, top_k=5) assert len(recs) == len(set(recs)) + + def test_last_catalogue_item_is_eligible( + self, model: DeepSequenceModel, processor: SequenceProcessor + ) -> None: + assert model.output_proj.out_features == processor.vocab_size + 1 + + def test_empty_known_history_is_rejected(self, model: DeepSequenceModel) -> None: + with pytest.raises(ValueError, match="known item"): + model.recommend(torch.zeros((1, 10), dtype=torch.long)) + + +def test_vocabulary_roundtrip(processor: SequenceProcessor) -> None: + restored = SequenceProcessor.from_vocabulary( + processor.export_vocabulary(), max_length=processor.max_length + ) + assert restored.export_vocabulary() == processor.export_vocabulary() diff --git a/tests/test_schema_contracts.py b/tests/test_schema_contracts.py new file mode 100644 index 0000000..b16be78 --- /dev/null +++ b/tests/test_schema_contracts.py @@ -0,0 +1,42 @@ +"""Versioned event and temporal-split contract tests.""" + +from datetime import datetime, timedelta, timezone + +import pytest +from pydantic import ValidationError + +from src.training.data import InteractionEvent, next_item_examples, session_sequences, temporal_split + + +def event(index: int, session: str = "s1") -> InteractionEvent: + return InteractionEvent( + event_id=f"event-{index}", + user_id="user-1", + session_id=session, + item_id=f"item-{index % 3}", + event_type="click", + timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc) + timedelta(minutes=index), + ) + + +def test_event_schema_rejects_unknown_event_type() -> None: + with pytest.raises(ValidationError): + InteractionEvent( + event_id="e", + user_id="u", + session_id="s", + item_id="i", + event_type="share", + timestamp=datetime.now(timezone.utc), + ) + + +def test_temporal_split_preserves_order() -> None: + train, validation, test = temporal_split([event(index) for index in range(20)]) + assert max(item.timestamp for item in train) < min(item.timestamp for item in validation) + assert max(item.timestamp for item in validation) < min(item.timestamp for item in test) + + +def test_next_item_examples_are_causal() -> None: + examples = next_item_examples(session_sequences([event(index) for index in range(4)])) + assert examples[0] == (["item-0"], "item-1") diff --git a/tests/test_serving_controls.py b/tests/test_serving_controls.py new file mode 100644 index 0000000..a406d7c --- /dev/null +++ b/tests/test_serving_controls.py @@ -0,0 +1,24 @@ +from app.core.security import api_key_is_valid +from app.core.serving import RateLimiter, RecommendationCache + + +def test_rate_limiter_fails_closed_after_budget() -> None: + limiter = RateLimiter(2) + assert limiter.allow("user") is True + assert limiter.allow("user") is True + assert limiter.allow("user") is False + + +def test_cache_is_model_version_aware() -> None: + cache = RecommendationCache(ttl_seconds=30) + first = cache.key("v1", ["a"], 2) + second = cache.key("v2", ["a"], 2) + cache.put(first, ["b"]) + assert cache.get(first) == ["b"] + assert cache.get(second) is None + + +def test_api_key_validation() -> None: + assert api_key_is_valid(None, None) is True + assert api_key_is_valid("correct", "correct") is True + assert api_key_is_valid("wrong", "correct") is False diff --git a/tests/test_smoke_api.py b/tests/test_smoke_api.py index d993025..18b57ff 100644 --- a/tests/test_smoke_api.py +++ b/tests/test_smoke_api.py @@ -6,9 +6,15 @@ from __future__ import annotations +import os + import pytest from fastapi.testclient import TestClient +# Keep smoke tests independent of developer-generated artifacts in models/current. +os.environ["ENVIRONMENT"] = "development" +os.environ["MODEL_BUNDLE_PATH"] = "models/test-bundle-does-not-exist" + from app.main import app @@ -85,7 +91,7 @@ def test_recommend_empty_sequence(self, client: TestClient) -> None: "top_k": 5, } response = client.post("/recommendations/", json=payload) - assert response.status_code == 200 + assert response.status_code == 422 def test_recommend_unknown_items(self, client: TestClient) -> None: payload = { @@ -94,4 +100,38 @@ def test_recommend_unknown_items(self, client: TestClient) -> None: "top_k": 5, } response = client.post("/recommendations/", json=payload) - assert response.status_code == 200 + assert response.status_code == 422 + + def test_recommend_rejects_top_k_beyond_contract(self, client: TestClient) -> None: + response = client.post( + "/recommendations/", + json={"user_id": "u", "item_sequence": ["item_1"], "top_k": 51}, + ) + assert response.status_code == 422 + + def test_response_exposes_model_provenance(self, client: TestClient) -> None: + response = client.post( + "/recommendations/", + json={"user_id": "u", "item_sequence": ["item_1"], "top_k": 3}, + ) + assert response.json()["model_version"] == "development-untrained" + assert response.json()["fallback"] is False + + +class TestFeedbackEndpoint: + def test_feedback_is_accepted_without_logging_raw_user_id( + self, client: TestClient, caplog + ) -> None: + payload = { + "impression_id": "imp-1", + "user_id": "private-user", + "item_id": "item_3", + "event_type": "click", + "position": 2, + "model_version": "development-untrained", + } + with caplog.at_level("INFO"): + response = client.post("/recommendations/feedback", json=payload) + assert response.status_code == 202 + assert "private-user" not in caplog.text + assert "anonymous_user_id" in caplog.text diff --git a/tests/test_training_pipeline.py b/tests/test_training_pipeline.py new file mode 100644 index 0000000..46bcc7b --- /dev/null +++ b/tests/test_training_pipeline.py @@ -0,0 +1,18 @@ +from app.core.artifacts import load_bundle +from scripts.generate_demo_data import generate_events +from src.training.data import write_jsonl +from src.training.train import run_training + + +def test_training_pipeline_produces_loadable_evaluated_bundle(tmp_path) -> None: + dataset = tmp_path / "events.jsonl" + bundle = tmp_path / "bundle" + write_jsonl(dataset, generate_events(sessions=12, events_per_session=6)) + + report = run_training(dataset, bundle, epochs=1, top_k=5, seed=7) + processor, model, manifest = load_bundle(bundle) + + assert report["validation"]["recall_at_k"] >= 0 + assert "popularity_baseline" in report + assert manifest.dataset_id == report["dataset_id"] + assert model.recommend(processor.to_tensor(["item-1"]), top_k=3)