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

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})

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
    )

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()

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"
)

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

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
)

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

hashtagOverview

hashtagQuick Start

hashtagSingle-Point Predictions

hashtagBasic Usage

hashtagUsing the Service Directly

hashtagInput Data Format

hashtagValidating Input Data

hashtagBatch Predictions

hashtagBasic Usage

hashtagUsing Lists Instead of DataFrames

hashtagAsynchronous Batch Processing

hashtagMonitoring Job Progress

hashtagRetrieving Batch Results

hashtagInterpreting Results

hashtagUnderstanding the Prediction Class

hashtagDictionary-Style Access

hashtagClassification Results

hashtagRegression Results

hashtagError Handling

hashtagCommon Prediction Errors

hashtagHandling Invalid Model IDs

hashtagHandling Missing Features

hashtagBatch Prediction Error Handling

hashtagPerformance Considerations

hashtagWhen to Use Single vs. Batch Predictions

hashtagOptimizing Single Predictions

hashtagOptimizing Batch Predictions

hashtagCaching Predictions

hashtagBest Practices

hashtagSet Default Model ID

hashtagValidate Input Data

hashtagHandle None Results

hashtagUse Appropriate Data Types

hashtagMonitor Prediction Latency

hashtagUse Batch for Bulk Operations

hashtagImplement Retry Logic

hashtagSee Also

Overview

Quick Start

Single-Point Predictions

Basic Usage

Using the Service Directly

Input Data Format

Validating Input Data

Batch Predictions

Basic Usage

Using Lists Instead of DataFrames

Asynchronous Batch Processing

Monitoring Job Progress

Retrieving Batch Results

Interpreting Results

Understanding the Prediction Class

Dictionary-Style Access

Classification Results

Regression Results

Error Handling

Common Prediction Errors

Handling Invalid Model IDs

Handling Missing Features

Batch Prediction Error Handling

Performance Considerations

When to Use Single vs. Batch Predictions

Optimizing Single Predictions

Optimizing Batch Predictions

Caching Predictions

Best Practices

Set Default Model ID

Validate Input Data

Handle None Results

Use Appropriate Data Types

Monitor Prediction Latency

Use Batch for Bulk Operations

Implement Retry Logic

See Also