Appearance
Motion primitives
Distances from images
This method measures how well the model judges distances from the images and plans its moves. The script moves the gripper to the exact distance that each move asks for.
A multimodal LLM looks at the camera images and replies with a short list of moves, such as "move 10 cm forward", "close the gripper" and "move 15 cm up". The script carries out each move, then shows the model the cameras again.
imageprompt
robotFranka arm, 2 cameras
taskput the bowl on the plate
stategripper position, open
replymove(), open(), close()
LLM
Example script
py
# /// script
# requires-python = ">=3.10"
# dependencies = [
# "manifold-sdk @ git+https://github.com/bifrostai/manifold-sdk",
# "litellm>=1.60",
# "tenacity",
# "pillow>=10",
# ]
# ///
"""Serve a large language model that moves the LIBERO robot with a few named moves.
The model looks at the cameras and replies with a list of moves, such as
move(0.10, 0, 0) and close(). The script carries out each move, then shows the
model the cameras again.
"""
import argparse
import base64
import io
import re
import time
import litellm
import numpy as np
from PIL import Image
import manifold
from manifold.benchmarks import LIBERO
from manifold.core.policy import PolicySignature
from manifold.core.sensor import CameraOrientation, Modality
from manifold.core.values import Action, Observation
PROMPT = """\
You control a Franka Panda arm in a simulator. Complete the task you are given.
You see two cameras: agentview faces the robot from the front of the table, and
wrist is mounted on the gripper. You also get the gripper position x, y, z in
meters (+x away from the robot, +y to its left, +z up) and whether the gripper
is open or closed.
Reply with a comment that gives your reasoning in one sentence, then one move
on each line. The moves are:
move(dx, dy, dz) move the gripper by dx, dy, dz meters
open() open the gripper
close() close the gripper
For example, to pick up an object 10 cm in front of the gripper and 12 cm below it:
# The bowl is 10 cm in front of the gripper, so move above it and grasp it.
move(0.10, 0, 0)
move(0, 0, -0.12)
close()
move(0, 0, 0.15)
Give at most 6 moves. After they run, you see the cameras again.
"""
MOVE = re.compile(r"^\s*(move|open|close)\(([^)]*)\)\s*$")
FAR = 0.30 # the longest move, in meters
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--model", required=True, help="LiteLLM model")
parser.add_argument("--base-url", help="endpoint of a self-hosted model")
parser.add_argument("--effort", help="reasoning effort, e.g. low")
parser.add_argument("--header", action="append", default=[], help="extra request header, NAME=VALUE")
args = parser.parse_args()
# All LIBERO suites, such as libero-spatial, libero-plus and debug-libero, send this observation.
signature = PolicySignature(
action_space=LIBERO.embodiment.action,
proprioception=LIBERO.embodiment.proprioception,
# Color cameras only, upright: Manifold flips frames from a benchmark that sends them bottom row first.
cameras=[
c.model_copy(update={"orientation": CameraOrientation.UPRIGHT, "calibration": None})
for c in LIBERO.sensors
if c.modality == Modality.RGB
],
)
def ask(content: list) -> str:
started = time.monotonic()
response = litellm.completion(
model=args.model,
api_base=args.base_url,
reasoning_effort=args.effort,
extra_headers=dict(h.split("=", 1) for h in args.header),
num_retries=3,
max_tokens=4000,
messages=[{"role": "system", "content": PROMPT}, {"role": "user", "content": content}],
)
usage = response.usage
print(f"{time.monotonic() - started:.1f} s, {usage.prompt_tokens} in, {usage.completion_tokens} out")
return response.choices[0].message.content or ""
class Policy:
"""Stateful class where reset() is called at the start of every episode.
Do not load weights in this class.
"""
def reset(self) -> None:
self.steps, self.done, self.grip = None, [], -1.0
def predict(self, obs: Observation) -> Action:
self.obs = obs
while True:
if self.steps is None:
self.steps = self.run(self.moves(obs))
try:
return Action.from_array(np.clip(next(self.steps), -1, 1).astype(np.float32))
except StopIteration:
self.steps = None
def gripper(self) -> np.ndarray:
return np.asarray(self.obs.state["ee_pose"], dtype=float)[:3]
def moves(self, obs: Observation) -> list[tuple[str, list[float]]]:
"""Show the model the cameras and the gripper, and read its moves."""
ee = np.asarray(obs.state["ee_pose"], dtype=float)
state = "closed" if self.grip > 0 else "open"
recent = "\n".join(self.done[-10:]) or "none yet"
content = [{"type": "text", "text": (
f"Task: {obs.instruction}\n"
f"Gripper at {np.round(ee[:3], 3).tolist()}, {state}.\n"
f"Moves so far:\n{recent}"
)}]
for camera in signature.cameras:
jpeg = io.BytesIO()
Image.fromarray(obs.sensors[camera.name]).save(jpeg, format="JPEG")
url = "data:image/jpeg;base64," + base64.b64encode(jpeg.getvalue()).decode()
content.append({"type": "text", "text": f"{camera.name} camera:"})
content.append({"type": "image_url", "image_url": {"url": url}})
reply = ask(content)
moves = []
for line in reply.splitlines():
found = MOVE.match(line)
if found:
try:
params = [float(a) for a in found[2].split(",") if a.strip()]
except ValueError:
continue
moves.append((found[1], params))
moves = moves[:6]
thought = next((l.strip("# ").strip() for l in reply.splitlines() if l.strip().startswith("#")), "")
listed = ", ".join(f"{name}({', '.join(map(str, params))})" for name, params in moves)
print(f"{thought} | {listed or 'no moves'}")
return moves
def run(self, moves: list[tuple[str, list[float]]]):
"""Carry out the moves one step at a time."""
if not moves:
# The reply had no moves, so hold still for a moment and ask again.
for _ in range(5):
yield [0, 0, 0, 0, 0, 0, 0]
return
for name, params in moves:
if name == "move" and len(params) == 3:
delta = np.clip(params, -FAR, FAR)
target = self.gripper() + delta
for _ in range(80):
error = target - self.gripper()
if np.linalg.norm(error) < 0.01:
break
yield [*np.clip(15 * error, -1, 1), 0, 0, 0, self.grip]
self.done.append(f"move({', '.join(f'{d:.2f}' for d in delta)}), ended at {np.round(self.gripper(), 3).tolist()}")
elif name in ("open", "close"):
self.grip = 1.0 if name == "close" else -1.0
for _ in range(15):
yield [0, 0, 0, 0, 0, 0, self.grip]
self.done.append(f"{name}()")
# One short call before opening the port, so a wrong key or model name stops the script here.
ask([{"type": "text", "text": "Reply with OK."}])
manifold.serve(Policy, signature)
if __name__ == "__main__":
main()Further reading
| Work | What the model outputs | Result |
|---|---|---|
| SayCan, Google, 2022 | The next skill, chosen from a library of trained skills | Chooses the right skills 84% of the time and completes 74% of 101 kitchen tasks |
| Claude plays robotics, Anthropic, 2026 | Speed commands for a trained walking controller: forward, sideways and turn | Completes real navigation tasks |