Model registry, inference serving, drift detection, rollouts, experiments, and observability
| Role | Permissions | Users |
|---|---|---|
| {{ role.name }} |
{% for perm in role.permissions|default([]) %}
{{ perm }}
{% endfor %}
|
{{ role.user_count|default(0) }} |
Send JSON input to any registered model and inspect the output, latency, and preprocessing pipeline.
Run the same input through multiple models and compare outputs, latency, and behavior differences.
Submit multiple inputs as a JSON array and process them in a single batch. Max 50 items per batch.
Download a complete snapshot of all MLOps data including models, metrics, experiments, alert rules, and inference history.
Copy-paste examples for Aquilia MLOps subsystem.
from aquilia.mlops import AquiliaModel, model
@model(name="sentiment", version="1.0.0", framework="pytorch")
class SentimentModel(AquiliaModel):
"""Sentiment analysis model with auto-registration."""
async def load(self):
import torch
self.tokenizer = AutoTokenizer.from_pretrained("bert-base")
self.model = AutoModelForSequenceClassification.from_pretrained("bert-base")
async def predict(self, request):
tokens = self.tokenizer(request.input, return_tensors="pt")
output = self.model(**tokens)
return {"sentiment": output.logits.argmax().item()}
async def health(self):
return {"status": "healthy", "loaded": True}from aquilia.mlops import serve
# Functional decorator for quick model serving
@serve(name="text-classifier", version="2.1.0")
async def classify(request):
"""Classify text into categories."""
text = request.input.get("text", "")
# Your inference logic here
return {
"category": "technology",
"confidence": 0.95,
"model": "text-classifier"
}
# Routes auto-generated:
# POST /predict/text-classifier
# GET /health/text-classifier
# GET /metrics/text-classifierfrom aquilia.mlops.engine import HookRegistry, on_load, preprocess, postprocess
hooks = HookRegistry()
@hooks.on_load
async def warm_up(model, context):
"""Run warmup inference after model load."""
await model.predict({"input": "warmup"})
@hooks.preprocess
async def validate_input(data, context):
"""Validate and normalize input before inference."""
if "input" not in data:
raise ValueError("Missing 'input' field")
data["input"] = data["input"].strip().lower()
return data
@hooks.postprocess
async def add_metadata(result, context):
"""Enrich result with model metadata."""
result["model_version"] = context.model_version
result["latency_ms"] = context.elapsed_ms
return resultfrom aquilia.mlops import RolloutEngine, RolloutConfig, RolloutStrategy
engine = app.di.get(RolloutEngine)
# Canary deployment: gradually shift traffic
config = RolloutConfig(
from_version="1.0.0",
to_version="2.0.0",
strategy=RolloutStrategy.CANARY,
steps=[10, 25, 50, 75, 100],
metric_gate="error_rate < 0.01",
)
rollout = await engine.start(config)
# Monitor and advance
await engine.advance(rollout.id) # 10% -> 25%
await engine.rollback(rollout.id) # auto-rollback on failurefrom aquilia.mlops.observe import DriftDetector, DriftMethod
import numpy as np
detector = app.di.get(DriftDetector)
# Set baseline reference distribution
reference_data = np.random.normal(0, 1, size=10000)
detector.set_reference(reference_data)
# Check incoming production data for drift
production_data = np.random.normal(0.3, 1.2, size=500)
report = detector.detect(production_data)
print(f"Method: {report.method}") # psi, ks_test, etc.
print(f"Score: {report.score:.4f}") # drift magnitude
print(f"Drifted: {report.is_drifted}") # True/False
print(f"Threshold: {report.threshold}") # configured threshold
# Supported methods:
# DriftMethod.PSI - Population Stability Index
# DriftMethod.KS_TEST - Kolmogorov-Smirnov test
# DriftMethod.DISTRIBUTION - Distribution distance
# DriftMethod.EMBEDDING - Embedding cosine drift
# DriftMethod.PERPLEXITY - LLM perplexity driftfrom aquilia.mlops.scheduler import Autoscaler, ScalingPolicy
policy = ScalingPolicy(
min_replicas=1,
max_replicas=10,
target_concurrency=50,
target_latency_p95=200.0, # ms
target_gpu_utilization=0.75,
cooldown=60, # seconds
)
scaler = Autoscaler(policy=policy, current_replicas=2)
# Record inference metrics
scaler.record(latency_ms=45.0, error=False)
scaler.record(latency_ms=180.0, error=False)
# Evaluate scaling decision
decision = scaler.evaluate()
print(f"Action: {decision.action}") # scale_up / scale_down / none
print(f"Target: {decision.target_replicas}")
print(f"Reason: {decision.reason}")
# Generate Kubernetes HPA manifest
hpa_yaml = scaler.generate_hpa_manifest(
name="sentiment-model",
namespace="ml-serving",
)
print(hpa_yaml)from aquilia.mlops.security import RBACManager, Permission, Role
rbac = app.di.get(RBACManager)
# Built-in roles: VIEWER, DEVELOPER, DEPLOYER, ADMIN
rbac.assign_role("alice", "ADMIN")
rbac.assign_role("bob", "DEVELOPER")
rbac.assign_role("charlie", "VIEWER")
# Check permissions before operations
if rbac.check_permission("bob", Permission.PACK_WRITE):
await registry.register(model)
if rbac.check_permission("bob", Permission.ROLLOUT_MANAGE):
await engine.start(rollout_config) # denied!
# Custom roles
rbac.add_role(Role(
name="ML_ENGINEER",
permissions=[
Permission.PACK_READ,
Permission.PACK_WRITE,
Permission.REGISTRY_ADMIN,
],
))
# Available permissions:
# PACK_READ, PACK_WRITE, PACK_DELETE, PACK_PROMOTE,
# PACK_SIGN, REGISTRY_ADMIN, PLUGIN_INSTALL, ROLLOUT_MANAGE# aquilia.toml — MLOps configuration [mlops] enabled = true device = "auto" # auto-detect CPU/CUDA/MPS max_models = 50 memory_limit = "4GB" [mlops.serving] host = "0.0.0.0" port = 8080 workers = 4 batch_size = 32 batch_timeout_ms = 50 [mlops.circuit_breaker] failure_threshold = 5 recovery_timeout = 30 half_open_max = 3 [mlops.drift] method = "psi" # psi, ks_test, distribution, embedding threshold = 0.2 check_interval = 3600 # seconds [mlops.models.sentiment] class_path = "models.sentiment.SentimentModel" version = "1.0.0" framework = "pytorch"
' + esc(d.name) + '', {full: true});
html += makeField('Version', esc(d.version));
html += makeField('State', makeBadge(d.state, 'badge-' + d.state));
html += '' + esc(d.from) + '');
html += makeField('To Version', '' + esc(d.to) + '');
html += makeField('Strategy', makeBadge(d.strategy, 'badge-' + d.strategy));
html += makeField('Phase', makeBadge(d.phase, 'badge-' + d.phase));
html += 'No inference history yet
' + 'Use the Live Inference Playground above to test models and build history.
' + '' + (h.version || '?') + '' +
'| Timestamp | Model | Status | ' + 'Latency | User | Details | ' + '
|---|