Mastering Deployment of AI Pipelines 

Objective

Develop a reliable, efficient, and production-ready Jarvid pipeline by:

1. Writing a comprehensive test suite.

2. Measuring and optimizing vector search performance.

3. Implementing monitoring and logging for continuous performance tracking.

Step 1: Comprehensive Test Suite for the Jarvid Pipeline

1.1 API Functionality Testing

1. Set Up Testing Environment:

   • Install pytest for testing.

bash
Copy code
pip install pytest requests

2. Write API Test Cases: File: tests/test_api.py
   python
   Copy code
   import requests

BASE_URL = "http://localhost:8000"

def test_root_endpoint():

    response = requests.get(f"{BASE_URL}/")

    assert response.status_code == 200

    assert "message" in response.json()

def test_error_endpoint():

    response = requests.get(f"{BASE_URL}/error")

    assert response.status_code == 500

    assert response.json()["detail"] == "Internal Server Error"

3. Run Tests:
   bash
   Copy code
   pytest tests/test_api.py

1.2 LLM Response Testing

1. Set Up Mock LLM Responses:

   • Use a mock server to simulate LLM API responses.

   • Install responses library:

bash
Copy code
pip install responses

2. Write LLM Test Cases: File: tests/test_llm.py
   python
   Copy code
   import requests

import responses

@responses.activate

def test_llm_response():

    responses.add(

        responses.POST,

        "https://api.openai.com/v1/completions",

        json={"choices": [{"text": "Hello, Jarvid!"}]},

        status=200

    )

    response = requests.post(

        "https://api.openai.com/v1/completions",

        json={"prompt": "Greet Jarvid", "max_tokens": 10}

    )

    assert response.status_code == 200

    assert response.json()["choices"][0]["text"] == "Hello, Jarvid!"

3. Run Tests:
   bash
   Copy code
   pytest tests/test_llm.py

Step 2: Benchmarking and Optimizing Vector Search

2.1 Benchmark Vector Search Performance

Write Benchmark Script: File: benchmark_faiss.py
python
Copy code
import faiss

import numpy as np

import time

# Generate random embeddings

dimension = 128

num_embeddings = 1000000

embeddings = np.random.random((num_embeddings, dimension)).astype('float32')

query = np.random.random((1, dimension)).astype('float32')

# Create FAISS index

index = faiss.IndexFlatL2(dimension)

index.add(embeddings)

# Measure latency

start = time.perf_counter()

_, indices = index.search(query, 10)

end = time.perf_counter()

latency = (end - start) * 1000  # ms

print(f"Search Latency: {latency:.2f} ms")

1. Run the Script:
   bash
   Copy code
   python benchmark_faiss.py

2. Optimize Vector Search

Switch to an Inverted Index (IVF):
python
Copy code
num_clusters = 100

quantizer = faiss.IndexFlatL2(dimension)

ivf_index = faiss.IndexIVFFlat(quantizer, dimension, num_clusters)

ivf_index.train(embeddings)

ivf_index.add(embeddings)

1. Benchmark IVF Index:
   python
   Copy code
   ivf_index.nprobe = 10

start = time.perf_counter()

_, indices = ivf_index.search(query, 10)

end = time.perf_counter()

print(f"IVF Search Latency (nprobe=10): {(end - start) * 1000:.2f} ms")

Step 3: Monitoring and Logging

3.1 Add Monitoring Metrics

Integrate Prometheus in API: Modify API Code:
python
Copy code
from prometheus_client import start_http_server, Counter, Summary

# Start Prometheus server

start_http_server(9100)

# Define Prometheus metrics

REQUEST_COUNT = Counter("request_count", "Total number of requests")

REQUEST_LATENCY = Summary("request_latency_seconds", "Latency of API requests")

@app.middleware("http")

async def add_metrics(request, call_next):

    REQUEST_COUNT.inc()

    start = time.perf_counter()

    response = await call_next(request)

    latency = time.perf_counter() - start

    REQUEST_LATENCY.observe(latency)

    return response

1. Run Prometheus and Confirm Metrics:

   • Check metrics at http://localhost:9100/metrics.

3.2 Set Up Alerts in Prometheus

Add Alert Rules: File: prometheus.yml
yaml
Copy code
groups:

  - name: API Alerts

    rules:

      - alert: HighLatency

        expr: request_latency_seconds > 0.5

        for: 1m

        labels:

          severity: warning

        annotations:

          summary: "High API latency detected."

1. Reload Prometheus:
   bash
   Copy code
   curl -X POST http://localhost:9090/-/reload

3.3 Add Logging

Set Up JSON Logging: Modify API Code:
python
Copy code
import logging

from pythonjsonlogger import jsonlogger

log_handler = logging.StreamHandler()

log_formatter = jsonlogger.JsonFormatter()

log_handler.setFormatter(log_formatter)

logger = logging.getLogger()

logger.addHandler(log_handler)

logger.setLevel(logging.INFO)

@app.middleware("http")

async def log_requests(request, call_next):

    logger.info({"event": "request_received", "path": request.url.path})

    response = await call_next(request)

    logger.info({"event": "request_processed", "status_code": response.status_code})

    return response

1. Test Logging:

   • Verify logs appear in the console.

Expected Outcomes

1. Comprehensive Test Suite:

   • Validates API endpoints and LLM responses.

2. Vector Search Benchmark:

   • Provides performance metrics and optimization insights.

3. Monitoring and Logging:

   • Ensures real-time tracking of API performance and errors.

4. Integrated System:

   • A fully tested, monitored, and optimized Jarvid pipeline ready for production.