Predictions

This guide covers everything you need to know about making predictions with the ViewAI Python SDK.

Overview

The ViewAI SDK provides two types of predictions:

• Single-point predictions: Real-time predictions for individual data points

• Batch predictions: Asynchronous predictions for large datasets

Quick Start

from viewai_client import ViewAIClient

# Initialize client
client = ViewAIClient(api_key="your-api-key")

# Single prediction
result = client.execute_single_point_prediction(
    data={"age": 35, "income": 50000},
    model_id="model-123"
)

# Batch prediction
job = client.execute_batch_prediction(
    data=[{"age": 35}, {"age": 42}],
    model_id="model-123",
    wait_for_completion=True
)

Single-Point Predictions

Single-point predictions are ideal for real-time inference where you need immediate results for individual data points.

Basic Usage

# Execute single prediction
result = client.execute_single_point_prediction(
    data={
        "age": 35,
        "income": 50000,
        "credit_score": 720,
        "employment_status": "Employed"
    },
    model_id="model-123"
)

# Access prediction result
if result:
    print(f"Prediction: {result.get_probabilities()}")
    print(f"Timestamp: {result.get_timestamp()}")

Using the Service Directly

# Access prediction service
prediction_service = client.prediction_service

# Execute prediction
result = prediction_service.execute_single_point_prediction(
    data={"age": 35, "income": 50000},
    model_id="model-123"
)

Input Data Format

Data must be provided as a dictionary with feature names as keys:

# Classification example
classification_data = {
    "age": 35,
    "gender": "Female",
    "tenure_months": 24,
    "monthly_charges": 75.50,
    "contract_type": "Two year"
}

# Regression example
regression_data = {
    "square_feet": 2500,
    "bedrooms": 3,
    "bathrooms": 2,
    "location": "urban",
    "year_built": 2015
}

# Make predictions
classification_result = client.execute_single_point_prediction(
    data=classification_data,
    model_id="churn-model-123"
)

regression_result = client.execute_single_point_prediction(
    data=regression_data,
    model_id="price-model-456"
)

Validating Input Data

# Validate data against model schema
is_valid = prediction_service.validate_prediction_data(
    data={"age": 35, "income": 50000},
    model_id="model-123"
)

if is_valid:
    result = prediction_service.execute_single_point_prediction(
        data={"age": 35, "income": 50000},
        model_id="model-123"
    )
else:
    print("Data validation failed")

Batch Predictions

Batch predictions are designed for processing large datasets asynchronously. They're ideal for scoring thousands or millions of records.

Basic Usage

import pandas as pd

# Prepare batch data as DataFrame
df = pd.DataFrame({
    "age": [35, 42, 28, 55, 31],
    "income": [50000, 75000, 45000, 90000, 55000],
    "credit_score": [720, 680, 750, 800, 690]
})

# Execute batch prediction
job = client.execute_batch_prediction(
    data=df,
    model_id="model-123",
    wait_for_completion=True
)

# Check job status
print(f"Job ID: {job.job_id}")
print(f"Status: {job.status}")
print(f"Dashboard: {job.dashboard_url}")

Using Lists Instead of DataFrames

# Prepare batch data as list
data_list = [
    {"age": 35, "income": 50000, "credit_score": 720},
    {"age": 42, "income": 75000, "credit_score": 680},
    {"age": 28, "income": 45000, "credit_score": 750}
]

# Execute batch prediction
job = client.execute_batch_prediction(
    data=data_list,
    model_id="model-123",
    wait_for_completion=True
)

Asynchronous Batch Processing

# Submit job without waiting
job = client.execute_batch_prediction(
    data=large_dataframe,
    model_id="model-123",
    wait_for_completion=False
)

print(f"Job submitted: {job.job_id}")
print(f"Check status at: {job.dashboard_url}")

# Monitor job later
status = client.prediction_service.retrieve_batch_job_status(job.job_id)
print(f"Current status: {status.get('status')}")

Monitoring Job Progress

# Execute batch prediction
job = client.execute_batch_prediction(
    data=df,
    model_id="model-123",
    wait_for_completion=False
)

# Monitor with custom polling settings
completed_job = client.prediction_service.monitor_batch_job_until_complete(
    job=job,
    poll_interval=5,      # Check every 5 seconds
    max_wait_time=300     # Wait up to 5 minutes
)

print(f"Final status: {completed_job.status}")

Retrieving Batch Results

Results are available in the ViewAI dashboard:

# Get job status and results URL
status = client.prediction_service.retrieve_batch_job_status(job.job_id)

print(f"Status: {status.get('status')}")
print(f"Results: {job.dashboard_url}")

# Results are downloaded from the dashboard
if status.get('status') == 'completed':
    print(f"View results at: {job.dashboard_url}")

Interpreting Results

Understanding the Prediction Class

The Prediction class provides methods to extract prediction information:

from viewai_client.models.prediction import Prediction

# Make prediction
result = client.execute_single_point_prediction(
    data={"age": 35, "income": 50000},
    model_id="model-123"
)

# Get probabilities for classification
probabilities = result.get_probabilities()
print(f"Probabilities: {probabilities}")

# Output: {'predicted_churn': 1, 'probability_0': 0.3, 'probability_1': 0.7}

# Get timestamp
timestamp = result.get_timestamp()
print(f"Prediction timestamp: {timestamp}")

# Get raw prediction data
raw_data = result.get_raw_prediction()
print(f"Raw data: {raw_data}")

# Access specific values
confidence = result.get("probability_1", 0.0)
predicted_class = result.get("predicted_churn")

Dictionary-Style Access

# Access values like a dictionary
if "predicted_churn" in result:
    prediction_value = result["predicted_churn"]
    print(f"Churn prediction: {prediction_value}")

# Get with default value
confidence = result.get("confidence", 0.0)

Classification Results

Classification models return probabilities for each class:

result = client.execute_single_point_prediction(
    data=customer_data,
    model_id="churn-classifier-123"
)

# Get probabilities
probs = result.get_probabilities()
print(f"Probability of no churn (0): {probs.get('probability_0')}")
print(f"Probability of churn (1): {probs.get('probability_1')}")
print(f"Predicted class: {probs.get('predicted_churn')}")

# Determine high-risk customers
if probs.get('probability_1', 0) > 0.7:
    print("High risk customer - take action!")

Regression Results

Regression models return predicted values:

result = client.execute_single_point_prediction(
    data=house_data,
    model_id="price-regressor-456"
)

# Get predicted value
raw_data = result.get_raw_prediction()
predicted_price = raw_data.get('predicted_price')
print(f"Predicted house price: ${predicted_price:,.2f}")

Error Handling

Common Prediction Errors

Handle prediction errors gracefully:

from viewai_client.exceptions import (
    ViewAIError,
    APIError,
    ValidationError,
    NetworkError
)

try:
    result = client.execute_single_point_prediction(
        data={"age": 35},
        model_id="model-123"
    )

    if result is None:
        print("Prediction returned None - check logs for details")

except ValidationError as e:
    print(f"Data validation failed: {e.message}")
    print(f"Details: {e.details}")

except APIError as e:
    print(f"API error: {e.message}")
    print(f"Status code: {e.status_code}")
    print(f"Response: {e.response}")

except NetworkError as e:
    print(f"Network error: {e.message}")

except ViewAIError as e:
    print(f"ViewAI SDK error: {e.message}")

except Exception as e:
    print(f"Unexpected error: {e}")

Handling Invalid Model IDs

# Check if model exists before prediction
try:
    models = client.registry.list_models()
    model_exists = any(m.model_id == "model-123" for m in models)

    if not model_exists:
        print("Model not found in registry")
    else:
        result = client.execute_single_point_prediction(
            data=data,
            model_id="model-123"
        )

except Exception as e:
    print(f"Error: {e}")

Handling Missing Features

# Validate required features
required_features = ["age", "income", "credit_score"]

def validate_features(data, required_features):
    """Validate that all required features are present."""
    missing = [f for f in required_features if f not in data]
    if missing:
        raise ValidationError(f"Missing required features: {missing}")
    return True

try:
    validate_features(input_data, required_features)
    result = client.execute_single_point_prediction(
        data=input_data,
        model_id="model-123"
    )
except ValidationError as e:
    print(f"Validation error: {e}")

Batch Prediction Error Handling

try:
    job = client.execute_batch_prediction(
        data=df,
        model_id="model-123",
        wait_for_completion=True
    )

    if job.status == "completed":
        print(f"Success! Results: {job.dashboard_url}")
    elif job.status == "failed":
        print(f"Batch job failed - check dashboard: {job.dashboard_url}")
    else:
        print(f"Job in progress: {job.status}")

except Exception as e:
    print(f"Batch prediction error: {e}")

Performance Considerations

When to Use Single vs. Batch Predictions

Use single-point predictions for:

• Real-time inference (< 100ms latency required)

• Interactive applications

• Individual record scoring

• Low-volume prediction workloads

Use batch predictions for:

• Large datasets (>1000 records)

• Periodic scoring jobs

• Offline analysis

• High-volume workloads where latency isn't critical

Optimizing Single Predictions

# Reuse client instance
client = ViewAIClient(api_key="your-api-key")

# Make multiple predictions efficiently
results = []
for record in data_records[:100]:  # Limit to reasonable batch size
    result = client.execute_single_point_prediction(
        data=record,
        model_id="model-123"
    )
    if result:
        results.append(result)

# For >100 records, use batch predictions instead
if len(data_records) > 100:
    job = client.execute_batch_prediction(
        data=data_records,
        model_id="model-123"
    )

Optimizing Batch Predictions

import pandas as pd

# Use DataFrames for better performance
df = pd.DataFrame(large_dataset)

# Remove unnecessary columns
df = df[required_columns]

# Handle missing values before upload
df = df.fillna(df.median(numeric_only=True))

# Execute batch prediction
job = client.execute_batch_prediction(
    data=df,
    model_id="model-123",
    wait_for_completion=False  # Don't wait for large jobs
)

# Check status periodically
print(f"Job submitted: {job.job_id}")
print(f"Monitor at: {job.dashboard_url}")

Caching Predictions

For repeated predictions on the same data:

from functools import lru_cache
import hashlib
import json

class PredictionCache:
    """Simple prediction cache using LRU strategy."""

    def __init__(self, client, model_id, maxsize=1000):
        self.client = client
        self.model_id = model_id
        self._cache = {}
        self.maxsize = maxsize

    def _hash_data(self, data):
        """Create hash of input data."""
        data_str = json.dumps(data, sort_keys=True)
        return hashlib.md5(data_str.encode()).hexdigest()

    def predict(self, data):
        """Get prediction with caching."""
        data_hash = self._hash_data(data)

        if data_hash in self._cache:
            return self._cache[data_hash]

        result = self.client.execute_single_point_prediction(
            data=data,
            model_id=self.model_id
        )

        if len(self._cache) >= self.maxsize:
            # Remove oldest entry
            self._cache.pop(next(iter(self._cache)))

        self._cache[data_hash] = result
        return result

# Usage
cache = PredictionCache(client, "model-123", maxsize=1000)
result = cache.predict({"age": 35, "income": 50000})

Best Practices

1

Set Default Model ID

Configure a default model ID to simplify code:

from viewai_client.config import ClientConfiguration

config = ClientConfiguration(
    api_key="your-api-key",
    default_model_id="model-123"
)

client = ViewAIClient(config=config)

# Now you can omit model_id
result = client.execute_prediction(data={"age": 35})

2

Validate Input Data

Always validate data before making predictions:

def validate_and_predict(data, model_id):
    """Validate data and make prediction."""
    # Check required fields
    required_fields = ["age", "income", "credit_score"]
    missing = [f for f in required_fields if f not in data]

    if missing:
        raise ValueError(f"Missing required fields: {missing}")

    # Check data types
    if not isinstance(data["age"], (int, float)):
        raise ValueError("Age must be numeric")

    # Make prediction
    return client.execute_single_point_prediction(
        data=data,
        model_id=model_id
    )

3

Handle None Results

Always check for None results:

result = client.execute_single_point_prediction(
    data=data,
    model_id="model-123"
)

if result is None:
    # Handle error case
    print("Prediction failed - check logs")
    # Maybe use a fallback prediction
    fallback_prediction = {"predicted_value": 0, "confidence": 0.0}
else:
    # Process successful result
    probabilities = result.get_probabilities()

4

Use Appropriate Data Types

Ensure data types match model expectations:

# Convert data types appropriately
data = {
    "age": int(age_value),                    # Ensure integer
    "income": float(income_value),            # Ensure float
    "employment_status": str(status_value),   # Ensure string
    "has_credit_card": bool(card_value)       # Ensure boolean
}

result = client.execute_single_point_prediction(
    data=data,
    model_id="model-123"
)

5

Monitor Prediction Latency

Track prediction performance:

import time

def timed_prediction(data, model_id):
    """Make prediction and measure latency."""
    start_time = time.time()

    result = client.execute_single_point_prediction(
        data=data,
        model_id=model_id
    )

    latency = time.time() - start_time
    print(f"Prediction latency: {latency*1000:.2f}ms")

    return result

6

Use Batch for Bulk Operations

For multiple predictions, prefer batch operations:

# Bad: Multiple single predictions
for record in records:  # 1000+ records
    result = client.execute_single_point_prediction(
        data=record,
        model_id="model-123"
    )

# Good: Single batch prediction
job = client.execute_batch_prediction(
    data=records,
    model_id="model-123",
    wait_for_completion=True
)

7

Implement Retry Logic

Add retry logic for transient failures:

from time import sleep

def predict_with_retry(data, model_id, max_retries=3):
    """Make prediction with retry logic."""
    for attempt in range(max_retries):
        try:
            result = client.execute_single_point_prediction(
                data=data,
                model_id=model_id
            )
            return result
        except NetworkError as e:
            if attempt < max_retries - 1:
                sleep(2 ** attempt)  # Exponential backoff
                continue
            raise

See Also

• Training Models - Learn how to train models

• Batch Processing - Advanced batch prediction techniques

• Error Handling - Comprehensive error handling guide

• API Reference - Detailed API documentation