huawei-cloud-cloudrobo-inference-adapter · git:20261103.c6aca33 · 2026-11-03 · sha256 155c5511d3f5b888
huawei-cloud-cloudrobo-inference-adapter git:20261103.c6aca33A
Immutable. This exact content never changes and is served at /api/v1/blob/155c5511d3f5b888.
---
name: huawei-cloud-cloudrobo-inference-adapter
description: >
Guides adapting a VLA model's inference server interface to the CloudRobo
standard interface/protocol. Use this skill when the user wants to create a
compatible inference interface protocol for a model, or make a model
compatible with the CloudRobo client library. Trigger conditions:
"adapt inference interface", "inference service compatibility",
"flattened interface", or any request to make a model server compatible
with an external inference client library.
---
# Inference Interface Adapter
## Overview
This skill guides converting a VLA model's inference interface into an
inference server that uses a **flattened linear format**, adapting to the
CloudRobo client library.
Core idea: the model's own inference interface format differs from the
CloudRobo standard communication format. Instead of modifying the model's
inference logic, add a thin adapter layer that converts between the
flattened linear format and the model's internal format.
### Fixed Protocol Contract
| Item | Convention |
|---|---|
| Transport | WSS (WebSocket over TLS) |
| Certificate | Self-signed certificate generated at runtime (cryptography lib, RSA 3072) |
| Serialization | msgpack (+ numpy extension) |
| Observation format | Flat dot-notation dict |
| Observation keys | `observation.images.{key}` (key derived from the model config, **not hardcoded**), `observation.state` (fixed field), `task` (fixed field) |
| Action response | `{"action": (T, D) float32}`, fixed field |
| Handshake | First frame after connect sends the metadata contract |
| Error | Send `traceback.format_exc()` string, close with `INTERNAL_ERROR` |
**Observation request example:**
```python
{
"observation.state": np.ndarray, # (D,) float32
"observation.images.front": np.ndarray, # (H, W, 3) uint8
"observation.images.wrist_left": np.ndarray, # (H, W, 3) uint8
"task": str,
}
```
**Action response example:**
```python
{
"action": np.ndarray, # (T, D) float32
"server_timing": {"infer_ms": ..., "prev_total_ms": ...},
}
```
## Prerequisites
1. **The model's existing inference interface** — a predict/infer method,
model metadata (action_dim, state_dim, image_keys, action_chunk_size),
and normalization access (how to get/set denormalization parameters).
2. **Training outputs (for the entry point)** — the final weights from the
standard layout plus `dataset_statistics.json` (required for action
denormalization).
3. **Python packages** — `numpy`, `msgpack` (serialization with the
`__ndarray__`/`__npgeneric__` numpy extension), `websockets` (server +
client), and `cryptography` (self-signed certificates). `Pillow` is
optional (loading images from disk in the test client).
## Built-in Scripts
This skill ships two reusable scripts in `scripts/`:
- **`test_client.py`** — a generic WSS test client that auto-discovers the
server's I/O contract from the handshake metadata. Supports random images
or loading from disk, single/multiple inferences, and performance
benchmarking. Used to end-to-end verify any adapter server. Run
`python scripts/test_client.py` after starting the server.
- **`verify_adapter.py`** — a unit-testing tool for the adapter layer. Tests
**only through `adapter.infer(obs)`** (no model API assumptions),
covering observation conversion, image-key validation, state-dimension
validation, and the metadata property — all without running a server or
loading a model. Run
`python scripts/verify_adapter.py --adapter /path/to/adapter.py --class-name MyAdapter`.
- If the adapter depends on a specific model API (e.g. GR00T's
`get_action(observation)` with nested observations), pass a model stub
file (defining a `make_model_stub()` factory function) via
`--stub-module`:
`python scripts/verify_adapter.py --adapter gr00t_adapter.py --class-name Gr00tProtocolAdapter --stub-module stub_gr00t.py`
## 4-Component Pattern
Every inference interface adaptation follows the same architecture. The
model's inference logic stays untouched — all protocol-related logic lives
in separate files you create.
```
┌───────────────────────────────────────────────────┐
│ 1. Entry point (CLI) │
│ - parse args, assemble components │
├───────────────────────────────────────────────────┤
│ 2. Transport server │
│ - WSS + msgpack, handshake, request loop, │
│ error handling │
├───────────────────────────────────────────────────┤
│ 3. Protocol adapter │
│ - convert observations: flat format -> model │
│ format │
│ - convert actions: model format -> flat format │
├───────────────────────────────────────────────────┤
│ 4. Model inference interface (existing, reused) │
│ - predict / get_action -> actions │
│ - metadata, normalization │
├───────────────────────────────────────────────────┤
│ 5. Model + weights (existing, untouched) │
└───────────────────────────────────────────────────┘
```
Layers 4-5 already exist in the codebase. You create layers 1-3.
### Generality Layering (Reuse vs Customize)
The components differ greatly in generality — **do not blindly copy
templates**:
| Component | Generality | Handling |
|---|---|---|
| Transport server (`server_template.py`) | **High** — WSS/msgpack/handshake/error contract are model-agnostic | can copy and use directly |
| Test client (`scripts/test_client.py`) | **High** — auto-discovers contract from handshake metadata | can copy and use directly |
| Protocol adapter (`adapter_template.py`) | **Low** — the model's observation format/action output varies by model | reference the skeleton only; customize the `_derive_*` methods and conversion logic per model |
| Adapter verification (`verify_adapter.py`) | **Low** — depends on the model API | test `adapter.infer()` via "contract injection", without depending on the model API |
**Key judgment**: if the model's inference interface differs greatly from
the template's assumptions, **rewrite the adapter + write your own
verification**, reusing only the transport server and test client.
## Workflow
### Step 1: Identify the Model's Inference Interface
Find the existing inference interface in the codebase. It usually exposes:
- A predict/infer method — takes an example dict, returns actions
(typically batched `(B, T, D)`, possibly already denormalized)
- Model metadata — action_dim, state_dim, image_keys, action_chunk_size
- Normalization access — how to get/set denormalization parameters
Record the model's internal format:
- Does the model take a **flat example** (`{"image": ..., "state": ...,
"lang": ...}`) or **nested observations** (`{"video": {...}, "state":
{...}, "language": {...}}`)?
- What does the model return? (`(B, T, D)` ndarray? multiple action keys?
already denormalized?)
- Which image keys does the model expect? **Derive them from the model's
config or framework, don't hardcode.** Different models use different
camera names (`cam_high` vs `base_0_rgb` vs `camera_head`).
### Step 2: Create the Protocol Adapter
The adapter is the heart of the adaptation. It converts between the flat
wire format and the model's internal format. Create a class that wraps the
model's inference interface. **See `references/adapter_template.py` for the
full template** — after copying it to the project, you only need to
implement the five `_derive_*` methods (deriving image keys, state dim,
state keys, action dim, and action chunk size from the model
config/metadata). `__init__` calls all of them, so every attribute the
`metadata` property reads is initialized before the handshake.
The adapter exposes two core interfaces:
```python
class ProtocolAdapter:
def __init__(self, model, **kwargs): ... # derive and cache the model I/O contract
def infer(self, obs: dict) -> dict: # convert -> infer -> convert
model_input = self._obs_to_example(obs) # flat example or nested observation
result = self._model.predict(examples=[model_input])
return self._result_to_response(result, obs)
@property
def metadata(self) -> dict: ... # handshake contract, see below
```
> The model call inside `infer()` varies by model shape: flat-example
> models use `model.predict(examples=[...])`; nested-observation models
> (e.g. GR00T's `get_action(observation)`) call `model.get_action(model_input)`
> directly. See "Two Shapes of Model-Internal Observations" below for both
> shapes.
**Why hierarchical metadata**: groups related fields together (action,
state, images, task) instead of a flat list. This makes the contract
self-documenting, and clients can parse only the parts they need.
**Observation conversion `_obs_to_example`** — the most critical part. The
wire format uses flat dot-notation keys that directly encode field paths,
avoiding nested ambiguity. The conversion does three things:
1. **Images** — extract from the `observation.images.{key}` prefix, ordered
by the model's expected image_keys; a missing required image key raises
`KeyError` immediately.
2. **Language** — `task` → `lang`, using a fallback instruction when
empty.
3. **State** — `observation.state` → `(1, D)`, validating the dimension
against `state_dim`; omitted when the model has no state input.
**Action conversion `_result_to_response`** — usually just squeeze the
batch dimension: the model returns `(B, T, D)`, validate batch=1, squeeze
to `(T, D)`, and output `{"action": actions}`.
**Key decisions:**
- **Flat dot-notation wire format** — use flat keys like
`observation.images.{key}`. It avoids ambiguity and makes the wire
contract easier to evolve.
- **Image keys derived, not hardcoded** — derive from the model's
DataConfig (e.g. `video_keys` minus the `video.` prefix), not from a
non-existent `framework.image_keys`. If unavailable, allow a CLI override
via `--image_keys`.
- **State validation** — prevents silent garbage output from dimension
mismatches.
- **No `server_timing` in the adapter** — the transport server adds
timing because it is per-connection state. The adapter should be testable
without a server.
- **Batch squeeze** — the model returns `(B, T, D)` but the protocol
expects `(T, D)`. Always squeeze and validate batch is 1.
- **Model-internal field names configurable** — the action field name in
the model's return differs by model (`action`, `actions`,
`pred_actions`, etc.); specify it via the `model_action_key` parameter,
default `"action"`. The wire field name stays fixed as `action`, never
affected.
- **Multi-action-key concatenation** — some models (e.g. GR00T) return
multiple action keys (`{"x": ..., "y": ..., "z": ...}`). Concatenate
them per the model's action key order with
`np.concatenate([...], axis=-1)` into `(T, D)`, then output as `action`.
#### Two Shapes of Model-Internal Observations
The template assumes the model takes a **flat example dict**
(`{"image": ..., "state": ..., "lang": ...}`). But many VLA models take
**nested observations** (GR00T's `get_action` needs
`{"video": {...}, "state": {...}, "language": {...}}`). `_obs_to_example`
must implement the matching conversion per the model shape:
| Shape | Model-side input | `_obs_to_example` produces |
|---|---|---|
| Flat example | `{"image": (1,H,W,3), "state": (1,D), "lang": str}` | assemble directly |
| Nested observation | `{"video": {"cam": (B,T,H,W,C)}, "state": {"k": (B,T,D)}, "language": {key: [[str]]}}` | assemble a nested dict from wire flat keys |
The wire side (`observation.images.{key}` / `observation.state` /
`task`) stays **identical for both shapes**; only the adapter-internal
conversion differs.
#### Contract Derivation Reference Table (derive from model config/statistics, don't hardcode)
> ⚠️ **Verification note**: GR00T behavior is measured/verified; the rest
> are **example references** — each framework's config/statistics structure
> changes with version, so **follow the actual model's config structure**,
> don't copy paths blindly.
| Framework/Model | image_keys source | state/action dim source |
|---|---|---|
| example1 | `policy.modality_configs['video'].modality_keys` | `policy.processor.state_action_processor.statistics` (sum of dims per key) |
| example2 | video keys in config | `stats.json` |
| example3 | video items in `meta/modality.json` | `meta/stats.json` / `relative_stats.json` |
| Other | search config for camera key lists | search stats files containing `mean`/`std`/`min`/`max`/`q01` |
> Derivation principle: use the **sum of the actual statistics dims** for
> `state_dim`/`action_dim`, not the padded dims of the model's input layer,
> otherwise normalization index mismatches occur.
### Step 3: Create the Transport Server
The transport server implements the WSS communication loop. It contains no
inference logic — it just moves bytes. **See
`references/server_template.py` for the full template.**
At startup the server **generates a self-signed certificate at runtime**
with the `cryptography` library (RSA 3072 + x509), builds an
`ssl.SSLContext(PROTOCOL_TLS_SERVER)` (TLSv1.2+, restricted to secure
cipher suites), then calls `serve(..., ssl=ssl_context)` to enable TLS. The
certificate files are deleted immediately after being loaded into the
SSLContext.
The fixed shape of the communication loop (5 steps):
1. **Handshake** — send `metadata` as the first frame after connect
2. **Receive observation** — recv + deserialize (no envelope, no routing)
3. **Infer** — call `adapter.infer(obs)`
4. **Attach timing** — write `server_timing` into the response
5. **Send response** — send + serialize
**Key decisions:**
- **`prev_total_time` is a local variable** — reset per connection. The
first request always has `prev_total_ms == 0`; the second has the total
time of the first request.
- **No message routing** — a simple recv->infer->send loop. No
`{type, request_id, payload}` envelope; fundamentally simpler.
- **Error contract fixed** — always send a `traceback.format_exc()`
string and close with `INTERNAL_ERROR`. Sending another format (e.g. a
structured dict) makes the client fail to parse.
- **`compression=None, max_size=None`** — the common setting for large
numpy array payloads. Compression can cause binary-data issues;
max_size=None allows arbitrarily large messages.
### Step 4: Create the Entry Point
A thin CLI script that assembles the model, adapter, and server:
```python
def main(args):
# 1. Load the model (existing inference interface)
model = load_model(args.ckpt_path, device=args.device, **kwargs)
# 2. Wrap with the protocol adapter
adapter = ProtocolAdapter(
model,
send_state=not args.no_state,
fallback_instruction=args.fallback_instruction,
**protocol_specific_kwargs,
)
# 3. Log the contract for train/test consistency checks
contract = adapter.metadata
logging.warning(
"Serving on port=%d. Image keys: %s, state_dim: %d, action_dim: %d",
args.port,
contract["images"]["keys"],
contract["state"]["dim"],
contract["action"]["dim"],
)
# 4. Start the transport server
server = TransportServer(
adapter, host=args.host, port=args.port, metadata=contract,
)
server.serve_forever()
```
**Why log the contract at startup**: if the server's image keys don't match
training, inference will produce garbage output. This catches the most
common deployment error: serving a checkpoint with a mismatched observation
format.
### Step 5: Create the Test Client
Use the built-in **`scripts/test_client.py`** to verify the server
end-to-end. The client auto-discovers the server's I/O contract from the
handshake metadata, so it works with any adapter without code changes. The
client connects via a `wss://` URI and, because the server uses a
self-signed certificate, handshakes with `ssl.CERT_NONE` (skipping
certificate verification).
The test client supports both hierarchical and flat metadata formats. It
verifies:
- Actions shape is `(T, D)` (unbatched)
- First request `server_timing.prev_total_ms == 0`
- Subsequent requests `server_timing.prev_total_ms > 0`
For server-less unit tests of the adapter layer (observation conversion /
image-key validation / state-dimension validation / metadata), use
`scripts/verify_adapter.py`, as described under "Built-in Scripts" above.
## Core Commands
| Command | Purpose |
|---|---|
| `python scripts/verify_adapter.py --adapter /path/to/adapter.py --class-name MyAdapter` | server-less adapter unit tests (observation conversion, image keys, state dim, metadata) |
| `python scripts/verify_adapter.py --adapter gr00t_adapter.py --class-name Gr00tProtocolAdapter --stub-module stub_gr00t.py` | adapter tests with a model stub for model-API-dependent adapters |
| `python scripts/test_client.py --host 127.0.0.1 --port 8000` | smoke test with random images |
| `python scripts/test_client.py --host 127.0.0.1 --port 8000 --image_dir /path/to/images --num_inferences 5 --verbose` | end-to-end test with real images |
| `python scripts/test_client.py --host 127.0.0.1 --port 8000 --benchmark` | performance benchmark |
## Parameter Confirmation
| Parameter | Description |
|---|---|
| `--adapter` / `--class-name` | Adapter Python file path and class name to test (verify_adapter.py) |
| `--stub-module` | Model stub file defining `make_model_stub()` (for model-API-dependent adapters) |
| `--host` / `--port` | Server hostname and port (test_client.py; default `127.0.0.1:8000`) |
| `--image_dir` | Directory of images loaded by key name (falls back to random images) |
| `--image_keys` | Comma-separated camera key names (overrides server metadata) |
| `--prompt` | Language instruction (default "pick up the red cube") |
| `--num_inferences` / `--benchmark` | Number of inference calls; run a performance benchmark |
| `model_action_key` | Model-internal action field name (adapter constructor kwarg, default `"action"`); wire field stays `action` |
## File Structure
New files go under `migrate_to_cloudrobo/deployment/` at the project root.
Per the "Generality Layering": **only build the adapter-layer files
yourself**; reuse the skill's templates/built-in scripts for the protocol
layer (either copy or reference them):
```
migrate_to_cloudrobo/deployment/
├── {protocol}_adapter.py # adapter (Step 2, customized per model)
├── server_{protocol}.py # entry point (Step 4, assembly)
└── tools/
├── {protocol}_server.py # transport layer (Step 3, copy server_template.py)
└── (test_client.py directly references the skill's built-in script)
```
Naming convention: replace `{protocol}` with the protocol name (e.g.
`openpi_adapter.py`, `server_openpi.py`, `openpi_server.py`). The test
client directly uses the built-in `scripts/test_client.py` (auto-discovers
the contract from metadata; no need to write one).
## Common Pitfalls
1. **Hardcoded image keys** — different models use different camera names
(`cam_high` vs `base_0_rgb` vs `camera_head`). Always derive from the
model's DataConfig (e.g. `video_keys` minus the `video.` prefix). If the
framework doesn't expose image keys, allow a CLI override via
`--image_keys`.
2. **Forgetting the handshake metadata** — the client needs the contract
(state_dim, image_keys, action_dim) to build valid requests. Without
metadata, the client can only guess the format, causing silent failures.
3. **Wrong error format** — per the contract, send a
`traceback.format_exc()` string frame on error. Sending another format
(e.g. a structured error dict) causes the client to fail parsing.
4. **Batch-dimension confusion** — the model returns `(B, T, D)` but the
protocol expects `(T, D)`. Always squeeze the batch dimension in the
adapter and validate batch is 1.
5. **Normalization mismatch** — the model interface should handle
denormalization. The adapter only passes the correct normalization keys.
Verify the adapter does not double-denormalize (applying statistics in
both the model and the adapter).
6. **Missing state validation** — if the model was trained with a state
input, the adapter should validate that the observation contains state
with the correct dimension. A missing state key should raise a clear
error rather than silently producing garbage actions.
7. **Incompatible serializers** — the wire serialization must be identical
on both ends. The skill ships plain `msgpack` with a custom numpy
extension (`pack_array`/`unpack_array` encoding `__ndarray__` /
`__npgeneric__` dicts, same as the reference `test_openpi.py`). If the
project previously used a different codec (e.g. `msgpack-numpy`), replace
it on both the server and the client so the bytes stay compatible.
8. **BFloat16 → NumPy conversion** — if the model runs in bf16, output
tensors cannot be converted to NumPy directly
(`TypeError: Got unsupported ScalarType BFloat16`). Add `.float()`
before `.numpy()`:
```python
actions = pred_actions.detach().cpu().float().numpy() # bf16 → f32 → numpy
```
9. **websockets_sync_client API mismatch** — `websockets.sync.client.connect()`
doesn't accept `ping_interval` or `ping_timeout` (those are async-API
args). Only pass `compression`, `max_size`, `open_timeout`.
10. **Relative imports in test scripts** — if the test client uses relative
imports (`from . import <helper module>`), it can't run as a script
directly. Use absolute imports or `sys.path.insert(0, parent_dir)` for
standalone execution.
## Reference Documents
- `references/adapter_template.py` — protocol adapter template (Step 2,
customize the `_derive_*` methods per model)
- `references/server_template.py` — transport server template (Step 3,
WSS/msgpack/handshake/error contract)
- `scripts/test_client.py` — generic WSS test client (auto-discovers the
contract from handshake metadata)
- `scripts/verify_adapter.py` — server-less adapter unit-test tool
- `references/iam-policies.md` — authentication & access control (no cloud
credentials required; local server security notes)