huawei-cloud-cloudrobo-inference-adapter

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:

{
    "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:

{
    "action": np.ndarray,                         # (T, D) float32
    "server_timing": {"infer_ms": ..., "prev_total_ms": ...},
}

Prerequisites

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).
Training outputs (for the entry point) — the final weights from the standard layout plus dataset_statistics.json (required for action denormalization).
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:

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:

Images — extract from the observation.images.{key} prefix, ordered by the model's expected image_keys; a missing required image key raises KeyError immediately.
Language — task → lang, using a fallback instruction when empty.
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):

Handshake — send metadata as the first frame after connect
Receive observation — recv + deserialize (no envelope, no routing)
Infer — call adapter.infer(obs)
Attach timing — write server_timing into the response
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:

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

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.
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.
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.
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.
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).
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.
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.
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():
```
actions = pred_actions.detach().cpu().float().numpy()  # bf16 → f32 → numpy
```
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.
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)