## Overview
The VLA Cloud Node architecture provides a standardized way to run Vision-Language-Action (VLA) model inference and training on Cyberwave Cloud Nodes. It handles all Cyberwave-specific concerns (SDK, MQTT, cameras, weights download) so you can focus on model-specific logic.
The reference implementation is the [SmolVLA cloud node](https://github.com/cyberwave-com/cyberwave-compute-smolvla), which demonstrates the full architecture.
```
┌─────────────────────────────────────────────────────────────────┐
│ VLA Cloud Node Architecture │
│ │
│ ┌─────────────────────────────────────────────────────────────┐│
│ │ deploy.py / train.py ││
│ │ • Entry point (model-specific) ││
│ │ • Load model, build predict_fn ││
│ │ • Create CwProcessor / CwTrainer ││
│ └─────────────────────────────────────────────────────────────┘│
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────┐│
│ │ cw_processor.py / cw_trainer.py ││
│ │ • Cyberwave SDK + MQTT handling ││
│ │ • Weights download from MLModel API ││
│ │ • Background camera fetchers (inference) ││
│ │ • Dataset download + metrics logging (training) ││
│ └─────────────────────────────────────────────────────────────┘│
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────┐│
│ │ *_resolver.py ││
│ │ • Model-specific metadata extraction ││
│ │ • Camera mapping logic ││
│ │ • No torch, no Cyberwave imports ││
│ └─────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────┘
```
***
## Inference Architecture
### CwProcessor
The `CwProcessor` class orchestrates all Cyberwave I/O for inference workloads:
| Responsibility | Description |
| ---------------------- | --------------------------------------------------------------------- |
| **SDK Client** | Creates Cyberwave client with auto-configured MQTT |
| **Weights Download** | Fetches model weights from MLModel API via signed URLs |
| **Camera Binding** | Background daemon threads continuously fetch and decode camera frames |
| **Joint Subscription** | MQTT subscription for real-time joint state updates |
| **Action Publishing** | Publishes predicted actions to robot via MQTT |
| **Control Loop** | Orchestrates observe → predict → execute cycle |
```python theme={null}
from cw_processor import CwProcessor, parse_request_payload, download_weights
# Parse the JSON payload from cloud node
request = parse_request_payload(sys.argv[1])
# Download weights from Cyberwave MLModel API
if request.weights_url:
checkpoint = download_weights(request.weights_url)
# Build your model's predict function
predict_fn = build_predict_fn(checkpoint)
# Create processor and run
processor = CwProcessor(
request,
model_slug="smolvla",
checkpoint=checkpoint,
predict_fn=predict_fn,
)
processor.setup()
result = processor.run()
```
### Background Camera Fetching
`CwProcessor` spawns a daemon thread per camera that continuously polls for frames:
```
┌─────────────────────────────────────────────────────────────────┐
│ Camera Threads (Background) │
│ │
│ camera_wrist thread ──► GET /twins/{uuid}/latest-frame │
│ └──────────────────────────▶ cache (np.ndarray + bytes) │
│ │
│ camera_front thread ──► GET /twins/{uuid}/latest-frame │
│ └──────────────────────────▶ cache (np.ndarray + bytes) │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Control Loop │
│ │
│ get_inputs() ──► reads cached frames (instant, no I/O wait) │
└─────────────────────────────────────────────────────────────────┘
```
Benefits:
* Frame fetching is decoupled from inference loop
* No I/O latency during `get_inputs()` - just reads cached numpy arrays
* Consistent frame timing regardless of inference speed
### Weights Download
The `download_weights()` function handles fetching model weights from the Cyberwave MLModel API:
```
1. GET /api/v1/mlmodels/{uuid}/weights
└─► Returns { "signed_url": "...", "expires_at": "..." }
2. GET {signed_url}
└─► Stream download to temp file
3. Detect archive type (magic bytes + headers)
├─ .tar.zst (0x28B52FFD) → zstandard + tarfile
├─ .tar.gz (0x1F8B) → tarfile
└─ .zip (PK) → zipfile
4. Extract to ~/.cache/cyberwave/weights/{hash}/
5. Resolve model directory (find config.json)
└─► Return checkpoint path
```
***
## Training Architecture
### CwTrainer
The `CwTrainer` class orchestrates all Cyberwave-specific training concerns:
| Responsibility | Description |
| ------------------------ | --------------------------------------------------------------- |
| **Dataset Download** | Fetches and extracts dataset from `/api/v1/datasets/{uuid}/zip` |
| **Weights Download** | Downloads custom base model weights (optional) |
| **Logger Patching** | Replaces WandBLogger with CyberwaveLogger for metrics |
| **Config Building** | Delegates to model-specific trainer for pipeline config |
| **Status Updates** | Sends training status, metrics, and ETA to Cyberwave API |
| **Artifact Compression** | Compresses checkpoint to `.tar.zst` for upload |
```python theme={null}
from cw_trainer import CwTrainer, load_json_argument
MODEL_SLUG = "smolvla"
params = load_json_argument(sys.argv[1])
trainer = CwTrainer(params, model_slug=MODEL_SLUG)
trainer.setup()
result = trainer.run()
```
### CyberwaveLogger
Drop-in replacement for lerobot's `WandBLogger` that sends metrics to Cyberwave:
```python theme={null}
class CyberwaveLogger:
def log_dict(d, step, mode):
# Converts to Cyberwave format and queues for sending
# PUT /api/v1/mltrainings/{uuid}/metrics
def log_policy(checkpoint_dir):
# Logs checkpoint event
def log_video(...):
# No-op for Cyberwave
```
The logger computes and sends ETA after \~100 steps via `update_type="estimate"`.
***
## Resolver Interface
The resolver pattern separates model-specific metadata from Cyberwave I/O. Each model needs a resolver that implements:
```python theme={null}
from base_resolver import BaseVLAResolver
class MyModelResolver(BaseVLAResolver):
MODEL_SLUG = "my-model"
def __init__(self, checkpoint: str):
"""Load model config from checkpoint."""
self.checkpoint = checkpoint
self.training_config = self._load_training_config()
self.training_camera_names = self._extract_camera_names()
self.expected_state_dim = self._extract_state_dim()
self.expected_action_dim = self._extract_action_dim()
def build_camera_mapping(
self,
runtime_cameras: dict[str, str] | list[str],
) -> dict[str, str]:
"""Map training camera names to runtime identifiers."""
# Return: {training_name: runtime_key}
def get_expected_state_dim(self) -> int:
"""Number of joint positions the model expects as input."""
return self.expected_state_dim
def get_expected_action_dim(self) -> int:
"""Number of joint positions the model outputs."""
return self.expected_action_dim
```
### Resolver Guidelines