Deployment

This guide covers how to deploy pre-trained machine learning models to the ViewAI platform.

Overview

The ViewAI SDK supports deploying externally trained models (scikit-learn, XGBoost, etc.) to the ViewAI platform for hosting and inference. The deployment process includes:

• Model serialization (ONNX or Pickle)

• Schema generation and validation

• File upload to cloud storage

• Model registration and hosting

Quick Start

from viewai_client import ViewAIClient
from sklearn.ensemble import RandomForestClassifier
import pandas as pd

# Train your model
model = RandomForestClassifier()
model.fit(X_train, y_train)

# Initialize ViewAI client
client = ViewAIClient(api_key="your-api-key")
workspace = client.retrieve_default_workspace()

# Create deployable model
deployable = client.create_deployable_model(
    model=model,
    dataset=train_df,
    target_column="target",
    model_name="My Classifier",
    workspace=workspace
)

# Deploy to ViewAI
dashboard_id = deployable.deploy()
print(f"Model deployed: {dashboard_id}")

Deployment Workflow

graph LR
    A[Train Model] --> B[Create Schema]
    B --> C[Serialize Model]
    C --> D[Upload to S3]
    D --> E[Register Model]
    E --> F[Model Ready]

1

Train Model

Train your model locally (e.g., scikit-learn, XGBoost, LightGBM) and prepare your training dataframe with features and target.

2

Create Schema

Generate or customize a schema that defines expected input features, types, ranges, and categorical choices.

3

Serialize Model

Export the trained model to ONNX (recommended) or Pickle (legacy) for upload and serving.

4

Upload to Cloud Storage

Upload the serialized model file, schema, and a sample dataset (if needed) to cloud storage (S3).

5

Register Model

Register the uploaded artifacts with ViewAI to create a deployable model entry and start hosting.

6

Model Ready

Poll for deployment completion and validate the deployed endpoint with test requests.

Complete Deployment Example

from viewai_client import ViewAIClient
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import pandas as pd

# 1. Train your model
df = pd.read_csv("training_data.csv")
X = df.drop("target", axis=1)
y = df["target"]

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)

# 2. Initialize ViewAI client
client = ViewAIClient(api_key="your-api-key")
workspace = client.retrieve_workspace_by_name("production")

# 3. Create training dataframe for schema
train_df = pd.concat([X_train, y_train], axis=1)

# 4. Create deployable model (auto-generates schema)
deployable = client.create_deployable_model(
    model=model,
    dataset=train_df,
    target_column="target",
    model_name="Random Forest Classifier v1.0",
    workspace=workspace,
    project=None
)

# 5. Optional: Customize schema
deployable.update_schema_column_categories(
    column_name="category_field",
    categories=["A", "B", "C"],
    allow_new_categories=False
)

# 6. Deploy model
dashboard_id = deployable.deploy()
print(f"Model deployed at: https://app.viewai.ca/dashboard/{dashboard_id}")

Model Serialization

The ViewAI SDK supports two serialization formats:

ONNX Format (Recommended)

ONNX (Open Neural Network Exchange) provides version-agnostic, cross-platform model deployment:

from sklearn.ensemble import RandomForestClassifier

# Train model
model = RandomForestClassifier()
model.fit(X_train, y_train)

# Deploy with ONNX format (default)
deployable = client.create_deployable_model(
    model=model,
    dataset=train_df,
    target_column="target",
    model_name="My Model",
    workspace=workspace,
    export_format="onnx"  # Default
)

dashboard_id = deployable.deploy()

ONNX Benefits:

• Version-agnostic (works across Python versions)

• Cross-platform compatibility

• Optimized inference performance

• Smaller model size

ONNX Requirements:

pip install skl2onnx onnxruntime

Pickle Format (Legacy)

Pickle format for Python-specific deployment:

# Deploy with pickle format
deployable = client.create_deployable_model(
    model=model,
    dataset=train_df,
    target_column="target",
    model_name="My Model",
    workspace=workspace,
    export_format="pickle"  # Legacy format
)

dashboard_id = deployable.deploy()

Pickle Limitations:

• Python version-specific

• Not cross-platform

• Larger file size

• Security concerns

Supported Model Types

The SDK supports most scikit-learn compatible models:

from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.tree import DecisionTreeClassifier
from sklearn.svm import SVC
from xgboost import XGBClassifier
from lightgbm import LGBMClassifier

# All these models can be deployed
models = [
    RandomForestClassifier(),
    GradientBoostingClassifier(),
    LogisticRegression(),
    DecisionTreeClassifier(),
    XGBClassifier(),
    LGBMClassifier()
]

for model in models:
    model.fit(X_train, y_train)

    deployable = client.create_deployable_model(
        model=model,
        dataset=train_df,
        target_column="target",
        model_name=f"{model.__class__.__name__} Model",
        workspace=workspace
    )

    dashboard_id = deployable.deploy()

Deploying Pipelines

Deploy complete scikit-learn pipelines:

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier

# Create pipeline
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('classifier', RandomForestClassifier())
])

# Train pipeline
pipeline.fit(X_train, y_train)

# Deploy pipeline
deployable = client.create_deployable_model(
    model=pipeline,  # Deploy entire pipeline
    dataset=train_df,
    target_column="target",
    model_name="RF Pipeline",
    workspace=workspace,
    export_format="onnx"
)

dashboard_id = deployable.deploy()

Schema Management

Schemas define the expected input format and constraints for model inference.

Automatic Schema Generation

Schemas are automatically generated from your training data:

# Schema is auto-generated
deployable = client.create_deployable_model(
    model=model,
    dataset=train_df,
    target_column="target",
    schema=None,  # Auto-generate (default)
    model_name="My Model",
    workspace=workspace
)

# Preview generated schema
deployable.preview_schema()

Using ModelSchemaManager

Create and customize schemas manually:

from viewai_client import ModelSchemaManager

# Create schema manager
schema_manager = ModelSchemaManager()

# Generate schema from model and data
schema = schema_manager.create_schema_from_data(
    model=model,
    df=train_df
)

# Deploy with custom schema
deployable = client.create_deployable_model(
    model=model,
    dataset=train_df,
    target_column="target",
    schema=schema_manager,  # Use custom schema
    model_name="My Model",
    workspace=workspace
)

Customizing Schema Fields

Update Categorical Fields

Define allowed categories for categorical features:

# Update category constraints
deployable.update_schema_column_categories(
    column_name="payment_method",
    categories=[
        "Electronic check",
        "Mailed check",
        "Bank transfer",
        "Credit card"
    ],
    allow_new_categories=False  # Reject unknown categories
)

# Allow new categories (useful for growing category sets)
deployable.update_schema_column_categories(
    column_name="country",
    categories=["USA", "Canada", "UK"],
    allow_new_categories=True  # Accept new countries
)

Update Numerical Ranges

Set min/max constraints for numerical features:

# Set age range
deployable.update_schema_column_range(
    column_name="age",
    min_value=0,
    max_value=120
)

# Set income range
deployable.update_schema_column_range(
    column_name="income",
    min_value=0.0,
    max_value=1000000.0
)

# Set temperature range (can be negative)
deployable.update_schema_column_range(
    column_name="temperature",
    min_value=-50.0,
    max_value=50.0
)

Update Field Types

Change the data type for schema fields:

# Change field types
deployable.update_schema_column_type(
    column_name="user_id",
    data_type="str"  # Treat as string
)

deployable.update_schema_column_type(
    column_name="count",
    data_type="int"  # Enforce integer
)

deployable.update_schema_column_type(
    column_name="score",
    data_type="float"  # Allow decimals
)

deployable.update_schema_column_type(
    column_name="category",
    data_type="category"  # Categorical field
)

Schema Validation

Validate input data against schema before deployment:

import pandas as pd

# Create validation dataset
validation_df = pd.DataFrame({
    "age": [25, 30, 150],  # 150 exceeds max
    "income": [50000, 75000, 45000],
    "category": ["A", "B", "X"]  # X not in allowed categories
})

# Validate against schema
result = deployable.validate_data(validation_df)

if result["valid"]:
    print("Validation passed!")
else:
    print("Validation errors:")
    for error in result["errors"]:
        print(f"  - {error}")

Schema Preview and Export

Preview and export schemas for documentation:

# Preview schema in console
deployable.preview_schema()

# Get schema as dictionary
schema_dict = deployable.get_schema()
print(schema_dict)

# Save schema to file
deployable.save_schema_locally("schemas/model_v1_schema.json")

Deploying Models

Basic Deployment

Deploy a model with default settings:

# Train model
model = RandomForestClassifier()
model.fit(X_train, y_train)

# Create deployable model
deployable = client.create_deployable_model(
    model=model,
    dataset=train_df,
    target_column="target",
    model_name="My Model",
    workspace=workspace
)

# Deploy
dashboard_id = deployable.deploy()

print(f"Model deployed!")
print(f"Dashboard ID: {dashboard_id}")
print(f"URL: https://app.viewai.ca/dashboard/{dashboard_id}")

Deployment with Project Context

Organize models by project:

# Get workspace and project
workspace = client.retrieve_workspace_by_name("production")
project = client.retrieve_project_by_name("customer-analytics")

# Deploy to project
deployable = client.create_deployable_model(
    model=model,
    dataset=train_df,
    target_column="target",
    model_name="Churn Model v2.0",
    workspace=workspace,
    project=project  # Associate with project
)

dashboard_id = deployable.deploy()

Deployment Progress

The deployment process shows progress:

dashboard_id = deployable.deploy()

# Output:

# 🚀 Starting deployment of pre-trained model...

#    Export format: ONNX

#    Model: RandomForestClassifier

#    Features: 10

# ✅ Dashboard ID: dash-abc123

# ✅ Saved DataFrame to ./tmp/deploy_df.csv

# 🔄 Converting model to ONNX format...

# ✅ ONNX model saved to ./tmp/blackbox_model.onnx

#    Classes: ['0', '1']

# ✅ Saved schema to ./tmp/schema.json

# 📤 Uploading DataFrame to S3...

# ✅ DataFrame uploaded successfully

# 📤 Uploading ONNX model to S3...

# ✅ Model uploaded successfully

# 📤 Uploading schema to S3...

# ✅ Schema uploaded successfully

# ⏳ Polling for deployment completion...

# Deployment Progress: 100%|██████████| 30/30 [01:23<00:00]

# 🎉 Deployment completed! Check your dashboard at https://app.viewai.ca/dashboard/dash-abc123

Handling Deployment Failures

Handle deployment errors:

try:
    dashboard_id = deployable.deploy()

    if dashboard_id:
        print(f"Success! Model deployed: {dashboard_id}")
    else:
        print("Deployment failed - check logs")

except Exception as e:
    print(f"Deployment error: {e}")
    # Check error details and retry

Deployment Validation

Testing Deployed Models

Test your deployed model immediately after deployment:

# Deploy model
dashboard_id = deployable.deploy()

# Test with sample data
test_data = {
    "age": 35,
    "income": 50000,
    "credit_score": 720
}

# Make prediction
result = client.execute_single_point_prediction(
    data=test_data,
    model_id=dashboard_id
)

if result:
    print("Deployment validation successful!")
    print(f"Test prediction: {result.get_probabilities()}")
else:
    print("Deployment validation failed!")

Automated Validation Pipeline

Create an automated validation pipeline:

def deploy_and_validate(model, train_df, test_df, model_name, workspace):
    """Deploy model and validate with test set."""

    # 1. Create deployable model
    deployable = client.create_deployable_model(
        model=model,
        dataset=train_df,
        target_column="target",
        model_name=model_name,
        workspace=workspace
    )

    # 2. Deploy
    print(f"Deploying {model_name}...")
    dashboard_id = deployable.deploy()

    if not dashboard_id:
        raise RuntimeError("Deployment failed")

    # 3. Validate with test set
    print("Validating deployment...")
    test_samples = test_df.sample(n=min(10, len(test_df)))

    successful_predictions = 0
    for _, row in test_samples.iterrows():
        test_data = row.drop("target").to_dict()

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

        if result:
            successful_predictions += 1

    success_rate = successful_predictions / len(test_samples)
    print(f"Validation success rate: {success_rate:.1%}")

    if success_rate < 0.8:
        raise RuntimeError(f"Validation failed: {success_rate:.1%} success rate")

    print(f"Deployment successful: {dashboard_id}")
    return dashboard_id

# Use validation pipeline
dashboard_id = deploy_and_validate(
    model=model,
    train_df=train_df,
    test_df=test_df,
    model_name="Validated Model",
    workspace=workspace
)

Comparing Model Versions

Compare predictions between local and deployed models:

import numpy as np

# Get test data
X_test_sample = X_test.head(10)

# Local predictions
local_predictions = model.predict(X_test_sample)

# Deployed model predictions
deployed_predictions = []
for _, row in X_test_sample.iterrows():
    result = client.execute_single_point_prediction(
        data=row.to_dict(),
        model_id=dashboard_id
    )
    if result:
        pred = result.get("predicted_target", None)
        deployed_predictions.append(pred)

# Compare predictions
deployed_predictions = np.array(deployed_predictions)
match_rate = np.mean(local_predictions == deployed_predictions)

print(f"Prediction match rate: {match_rate:.1%}")

if match_rate < 0.99:
    print("Warning: Deployed model predictions differ from local model!")
else:
    print("Deployed model validated successfully!")

Production Patterns

Versioned Model Deployment

Deploy models with proper versioning:

from datetime import datetime

class ModelDeployer:
    """Handles versioned model deployment."""

    def __init__(self, client, workspace):
        self.client = client
        self.workspace = workspace
        self.version = 1

    def deploy_model(self, model, train_df, target_column, base_name):
        """Deploy model with version and timestamp."""

        # Generate versioned name
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        model_name = f"{base_name}_v{self.version}_{timestamp}"

        # Create deployable model
        deployable = self.client.create_deployable_model(
            model=model,
            dataset=train_df,
            target_column=target_column,
            model_name=model_name,
            workspace=self.workspace
        )

        # Deploy
        dashboard_id = deployable.deploy()

        # Increment version
        self.version += 1

        return {
            "dashboard_id": dashboard_id,
            "model_name": model_name,
            "version": self.version - 1,
            "timestamp": timestamp
        }

# Usage
deployer = ModelDeployer(client, workspace)

deployment_info = deployer.deploy_model(
    model=model,
    train_df=train_df,
    target_column="target",
    base_name="ChurnModel"
)

print(f"Deployed: {deployment_info['model_name']}")
print(f"Dashboard: {deployment_info['dashboard_id']}")

Blue-Green Deployment

Implement blue-green deployment pattern:

def blue_green_deployment(new_model, train_df, target_column, model_name, workspace, current_model_id=None):
    """Deploy new model version while keeping old version."""

    # Deploy new version (green)
    green_deployable = client.create_deployable_model(
        model=new_model,
        dataset=train_df,
        target_column=target_column,
        model_name=f"{model_name}_green",
        workspace=workspace
    )

    green_id = green_deployable.deploy()
    print(f"Green version deployed: {green_id}")

    # Validate green version
    test_samples = train_df.sample(n=10)
    validation_passed = True

    for _, row in test_samples.iterrows():
        result = client.execute_single_point_prediction(
            data=row.drop(target_column).to_dict(),
            model_id=green_id
        )
        if not result:
            validation_passed = False
            break

    if validation_passed:
        print("Green version validated - ready for production")
        print(f"Switch traffic from {current_model_id} to {green_id}")
        return green_id
    else:
        print("Green version validation failed - keeping blue version")
        return current_model_id

Deployment with Rollback

Implement deployment with rollback capability:

class DeploymentManager:
    """Manages model deployments with rollback support."""

    def __init__(self, client, workspace):
        self.client = client
        self.workspace = workspace
        self.deployment_history = []

    def deploy(self, model, train_df, target_column, model_name):
        """Deploy model and track in history."""

        deployable = self.client.create_deployable_model(
            model=model,
            dataset=train_df,
            target_column=target_column,
            model_name=model_name,
            workspace=self.workspace
        )

        dashboard_id = deployable.deploy()

        # Track deployment
        self.deployment_history.append({
            "dashboard_id": dashboard_id,
            "model_name": model_name,
            "timestamp": datetime.now()
        })

        return dashboard_id

    def rollback(self, steps=1):
        """Get previous deployment for rollback."""
        if len(self.deployment_history) <= steps:
            raise ValueError("Not enough deployment history for rollback")

        previous = self.deployment_history[-(steps + 1)]
        print(f"Rollback to: {previous['model_name']}")
        print(f"Dashboard ID: {previous['dashboard_id']}")
        return previous["dashboard_id"]

# Usage
manager = DeploymentManager(client, workspace)

# Deploy v1
v1_id = manager.deploy(model_v1, train_df, "target", "Model v1.0")

# Deploy v2
v2_id = manager.deploy(model_v2, train_df, "target", "Model v2.0")

# Rollback if v2 has issues
previous_id = manager.rollback(steps=1)

A/B Testing Deployment

Deploy multiple model versions for A/B testing:

def deploy_for_ab_testing(model_a, model_b, train_df, target_column, base_name, workspace):
    """Deploy two model versions for A/B testing."""

    # Deploy Model A
    deployable_a = client.create_deployable_model(
        model=model_a,
        dataset=train_df,
        target_column=target_column,
        model_name=f"{base_name}_variant_A",
        workspace=workspace
    )
    dashboard_id_a = deployable_a.deploy()

    # Deploy Model B
    deployable_b = client.create_deployable_model(
        model=model_b,
        dataset=train_df,
        target_column=target_column,
        model_name=f"{base_name}_variant_B",
        workspace=workspace
    )
    dashboard_id_b = deployable_b.deploy()

    print("A/B Test Setup Complete:")
    print(f"  Variant A: {dashboard_id_a}")
    print(f"  Variant B: {dashboard_id_b}")

    return {
        "variant_a": dashboard_id_a,
        "variant_b": dashboard_id_b
    }

# Deploy for A/B test
ab_models = deploy_for_ab_testing(
    model_a=logistic_model,
    model_b=random_forest_model,
    train_df=train_df,
    target_column="target",
    base_name="ABTest_ChurnModel",
    workspace=workspace
)

Troubleshooting

ONNX Conversion Errors

If ONNX conversion fails:

# Try with pickle format
try:
    deployable = client.create_deployable_model(
        model=model,
        dataset=train_df,
        target_column="target",
        model_name="My Model",
        workspace=workspace,
        export_format="onnx"
    )
    dashboard_id = deployable.deploy()

except Exception as e:
    print(f"ONNX export failed: {e}")
    print("Falling back to pickle format...")

    deployable = client.create_deployable_model(
        model=model,
        dataset=train_df,
        target_column="target",
        model_name="My Model",
        workspace=workspace,
        export_format="pickle"
    )
    dashboard_id = deployable.deploy()

Schema Errors

Handle schema-related errors:

# Create deployable with error handling
try:
    deployable = client.create_deployable_model(
        model=model,
        dataset=train_df,
        target_column="target",
        model_name="My Model",
        workspace=workspace
    )

    # Validate schema before deployment
    deployable.preview_schema()

    # Test schema validation
    validation_result = deployable.validate_data(train_df.head())

    if not validation_result["valid"]:
        print("Schema validation errors:")
        for error in validation_result["errors"]:
            print(f"  - {error}")
    else:
        dashboard_id = deployable.deploy()

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

Upload Failures

Handle file upload issues:

from viewai_client.exceptions import NetworkError

try:
    dashboard_id = deployable.deploy()

except NetworkError as e:
    print(f"Upload failed: {e}")
    print("Retrying deployment...")

    # Retry deployment
    dashboard_id = deployable.deploy()

See Also

• Training Models - Train models on ViewAI platform

• Making Predictions - Use deployed models for inference

• Schema Management - Advanced schema customization

• API Reference - Model class documentation