Appearance
Code as policy
The script moves the arm
This method measures the program that the model writes and where it points. The script's functions move the arm.
At the start of each episode, a multimodal LLM writes a Python program for the task from a text prompt. The prompt lists the functions that the program can call, gives examples of tasks with their code, names the objects on the table, and ends with the task as a comment.
The program then runs. To find an object, it asks the model to point at it in the camera image. To move the arm, it calls the script's pick and place functions. If the program calls a new function, the model writes that function too.
imageprompt
functionsget_obj_pos, goto_pos, …
examples5 tasks with their code
objectsketchup, bbq sauce, basket
taskput both sauces in the basket
LLM
WARNING
The script runs the model's program with Python's exec, outside a sandbox.
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 Code as Policies (Liang et al., 2022) on LIBERO.
At the start of each episode, the model writes a program for the task. The
program calls perception functions, which ask the model to find objects in the
camera image, and control functions, which move the arm. Functions that the
program uses but does not define are written by the model too, recursively.
"""
import argparse
import ast
import base64
import builtins
import io
import json
import queue
import re
import threading
import time
import litellm
import numpy as np
from PIL import Image, ImageDraw
import manifold
from manifold.benchmarks import LIBERO
from manifold.core.policy import PolicySignature
from manifold.core.values import Action, Observation
# The prompt for the program, in the format of the paper: the functions it can
# use, then examples of an instruction as a comment followed by its code.
PROMPT = """\
import numpy as np
from robot import get_obj_names, get_obj_pos, get_ee_pos, goto_pos, open_gripper, close_gripper, is_holding, put_first_on_second
# Positions are np.array([x, y, z]) in meters: +x away from the robot, +y to its left, +z up.
# get_obj_pos(name) returns the centre of the object in x and y, and its top in z.
# put_first_on_second(obj, target) picks up obj and places it on target, an object name or a position.
objects = ['black bowl', 'plate', 'ketchup']
# put the black bowl on the plate.
put_first_on_second('black bowl', 'plate')
objects = ['red mug', 'white mug', 'basket', 'butter']
# put both mugs in the basket.
for name in ['red mug', 'white mug']:
put_first_on_second(name, 'basket')
objects = ['bowl', 'plate', 'wine bottle']
# move the bowl 10 cm to the left.
put_first_on_second('bowl', get_obj_pos('bowl') + [0, 0.10, 0])
objects = ['alphabet soup', 'cream cheese', 'basket']
# pick up the alphabet soup and hold it 20 cm above the table.
target = get_obj_pos('alphabet soup')
open_gripper()
goto_pos(target + [0, 0, 0.08])
goto_pos(target - [0, 0, 0.03])
close_gripper()
goto_pos(target + [0, 0, 0.20])
objects = ['ketchup', 'tomato sauce', 'basket']
# put the sauce that is closest to the basket in the basket.
sauces = ['ketchup', 'tomato sauce']
basket = get_obj_pos('basket')
closest = min(sauces, key=lambda name: np.linalg.norm(get_obj_pos(name)[:2] - basket[:2]))
put_first_on_second(closest, 'basket')
"""
# The prompt for a function that a program uses but does not define.
FUNCTION_PROMPT = """\
import numpy as np
from robot import get_obj_names, get_obj_pos, get_ee_pos, goto_pos, open_gripper, close_gripper, is_holding, put_first_on_second
# define function: total = get_total(xs=numbers).
def get_total(xs):
return np.sum(xs)
# define function: pos = get_pos_above(name='plate', height=0.1).
def get_pos_above(name, height):
return get_obj_pos(name) + [0, 0, height]
# define function: names = get_objs_left_of(ref='bowl').
def get_objs_left_of(ref):
ref_y = get_obj_pos(ref)[1]
return [name for name in get_obj_names() if name != ref and get_obj_pos(name)[1] > ref_y]
"""
CODE_RULES = "Reply with Python code only, without explanations or markdown."
POINT_PROMPT = """\
The image is the agentview camera, which faces the robot from the front of the
table, at 512 x 512 with a grid labelled in pixels: u goes right from the left
edge, v goes down from the top edge. Point at the centre of the {name}.
Reply with JSON only: {{"u": <u>, "v": <v>}}"""
NAMES_PROMPT = """\
The image is the agentview camera, which faces a robot arm across a table. List
the objects on the table that a task could refer to, with short names such as
"black bowl" or "plate". Reply with a JSON list of names only."""
SCALE = 2 # agentview is shown to the model at twice its size
CARRY = 0.25 # height above the table for carrying, in meters
class Stop(Exception):
"""Raised in a running program when its episode ends."""
class Scene:
"""The agentview camera at one moment, as an upright image and a point cloud."""
def __init__(self, obs: Observation, calibration):
# LIBERO sends images bottom row first; the intrinsics count rows in the upright image.
self.rgb = np.ascontiguousarray(obs.sensors["agentview"][::-1])
depth = obs.sensors["agentview_depth"][::-1, :, 0].astype(float)
pose = np.asarray(obs.extrinsics["agentview"], dtype=float)
k = calibration.intrinsics
h, w = depth.shape
v, u = np.mgrid[0:h, 0:w].astype(float)
local = np.stack([(u - k.cx) * depth / k.fx, (v - k.cy) * depth / k.fy, depth, np.ones_like(depth)], -1)
self.points = (local @ pose.T)[..., :3]
self.points[~np.isfinite(depth)] = np.nan
# The table is the most common height in the lower half of the view.
heights = self.points[h // 2 :, :, 2].ravel()
counts, edges = np.histogram(heights[np.isfinite(heights)], bins=200)
self.table_z = float(edges[np.argmax(counts)] + (edges[1] - edges[0]) / 2)
def gridded(self) -> Image.Image:
image = Image.fromarray(self.rgb).resize((256 * SCALE, 256 * SCALE), Image.BICUBIC)
draw = ImageDraw.Draw(image)
for p in range(64, 512, 64):
draw.line([(p, 0), (p, 511)], fill=(255, 255, 255), width=1)
draw.line([(0, p), (511, p)], fill=(255, 255, 255), width=1)
draw.text((p + 2, 2), str(p), fill=(255, 255, 0))
draw.text((2, p + 2), str(p), fill=(255, 255, 0))
return image
def object_at(self, u: float, v: float) -> np.ndarray:
"""The centre in x and y, and the top in z, of the object under pixel (u, v)."""
h, w, _ = self.points.shape
u, v = int(np.clip(round(u), 0, w - 1)), int(np.clip(round(v), 0, h - 1))
hit = self.points[v, u]
near = self.points[max(v - 40, 0) : v + 41, max(u - 40, 0) : u + 41].reshape(-1, 3)
near = near[np.isfinite(near).all(1)]
close = near[np.hypot(near[:, 0] - hit[0], near[:, 1] - hit[1]) < 0.05]
above = close[close[:, 2] > self.table_z + 0.01]
if len(above) < 5:
return np.array(hit, dtype=float)
return np.array([*above[:, :2].mean(0), np.percentile(above[:, 2], 90)])
def image_part(image: Image.Image) -> dict:
jpeg = io.BytesIO()
image.save(jpeg, format="JPEG", quality=90)
return {"type": "image_url", "image_url": {"url": "data:image/jpeg;base64," + base64.b64encode(jpeg.getvalue()).decode()}}
def undefined_calls(code: str, known: set[str]) -> list[ast.Call]:
"""The calls in `code` to functions that are not defined in it and not in `known`."""
tree = ast.parse(code)
defined = {n.name for n in ast.walk(tree) if isinstance(n, ast.FunctionDef)}
assigned = {t.id for n in ast.walk(tree) if isinstance(n, ast.Assign) for t in n.targets if isinstance(t, ast.Name)}
calls = [n for n in ast.walk(tree) if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)]
missing = {}
for call in calls:
name = call.func.id
if name not in defined | assigned | known and not hasattr(builtins, name):
missing.setdefault(name, call)
return list(missing.values())
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.
# The script reads depth and calibration, so it takes the cameras as LIBERO sends them.
signature = PolicySignature(
action_space=LIBERO.embodiment.action,
proprioception=LIBERO.embodiment.proprioception,
cameras=[c for c in LIBERO.sensors if c.name in ("agentview", "agentview_depth")],
)
calibration = next(c.calibration for c in signature.cameras if c.name == "agentview")
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=8000,
messages=[{"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 ""
def write(prompt: str) -> str:
"""Continue `prompt` with code."""
reply = ask([{"type": "text", "text": f"{prompt}\n{CODE_RULES}"}])
found = re.search(r"```(?:python)?\n(.*?)```", reply, re.DOTALL)
return (found[1] if found else reply).strip()
def define(code: str, known: set[str], depth: int = 0) -> str:
"""Write the functions that `code` calls but does not define, depth first."""
functions = []
if depth > 3:
return ""
for call in undefined_calls(code, known):
name = call.func.id
body = write(f"{FUNCTION_PROMPT}\n# define function: {ast.unparse(call)}.\n")
known = known | {name}
functions += [define(body, known, depth + 1), body]
return "\n\n".join(f for f in functions if f)
class Policy:
"""Stateful class where reset() is called at the start of every episode.
Do not load weights in this class.
The program for the current episode runs on its own thread. Its control
functions hand each action to `predict` and wait for the next
observation, so the program moves the arm one step at a time.
"""
def __init__(self) -> None:
self.thread = None
def reset(self) -> None:
if self.thread is not None:
self.observations.put(None) # stops the previous program at its next step
self.observations: queue.Queue = queue.Queue()
self.actions: queue.Queue = queue.Queue()
self.thread, self.obs = None, None
def predict(self, obs: Observation) -> Action:
if self.thread is None:
self.obs = obs
self.thread = threading.Thread(target=self.run, args=(self.observations, self.actions), daemon=True)
self.thread.start()
else:
self.observations.put(obs)
return Action.from_array(np.clip(self.actions.get(), -1, 1).astype(np.float32))
# The functions that a program can call.
def step(self, action, observations: queue.Queue, actions: queue.Queue) -> None:
actions.put(action)
obs = observations.get()
if obs is None:
raise Stop()
self.obs = obs
def ee(self) -> np.ndarray:
return np.asarray(self.obs.state["ee_pose"], dtype=float)
def is_holding(self) -> bool:
fingers = self.ee()[7:9]
return bool(0.005 < abs(fingers[0]) + abs(fingers[1]) < 0.075)
def api(self, observations: queue.Queue, actions: queue.Queue) -> dict:
grip = [-1.0] # the gripper command that motions keep: -1 open, 1 closed
def step(action):
self.step(action, observations, actions)
def goto_pos(pos, tolerance=0.01, limit=80):
for _ in range(limit):
error = np.asarray(pos, dtype=float)[:3] - self.ee()[:3]
if np.linalg.norm(error) < tolerance:
return
step([*np.clip(15 * error, -1, 1), 0, 0, 0, grip[0]])
def set_gripper(value):
grip[0] = value
for _ in range(15):
step([0, 0, 0, 0, 0, 0, value])
def get_obj_names():
scene = Scene(self.obs, calibration)
reply = ask([{"type": "text", "text": NAMES_PROMPT}, image_part(scene.gridded())])
return [str(name) for name in json.loads(re.search(r"\[.*\]", reply, re.DOTALL)[0])]
def get_obj_pos(name):
scene = Scene(self.obs, calibration)
text = POINT_PROMPT.format(name=name)
reply = ask([{"type": "text", "text": text}, image_part(scene.gridded())])
point = json.loads(re.search(r"\{.*\}", reply, re.DOTALL)[0])
pos = scene.object_at(point["u"] / SCALE, point["v"] / SCALE)
print(f"get_obj_pos({name!r}) = {np.round(pos, 3).tolist()}")
return pos
def put_first_on_second(obj, target):
table = Scene(self.obs, calibration).table_z
source = get_obj_pos(obj)
goal = get_obj_pos(target) if isinstance(target, str) else np.asarray(target, dtype=float)
if len(goal) == 2:
goal = np.array([*goal, table])
set_gripper(-1)
goto_pos([*self.ee()[:2], max(self.ee()[2], table + CARRY)], limit=40)
goto_pos([*source[:2], table + CARRY])
goto_pos(source + [0, 0, 0.08], limit=40)
goto_pos([*source[:2], max(source[2] - 0.03, table + 0.015)], tolerance=0.006, limit=40)
set_gripper(1)
goto_pos([*source[:2], table + CARRY], limit=60)
goto_pos([*goal[:2], table + CARRY])
goto_pos(goal + [0, 0, 0.08], limit=40)
set_gripper(-1)
goto_pos([*goal[:2], table + CARRY], limit=40)
return {
"np": np,
"get_obj_names": get_obj_names,
"get_obj_pos": get_obj_pos,
"get_ee_pos": lambda: self.ee()[:3].copy(),
"goto_pos": goto_pos,
"open_gripper": lambda: set_gripper(-1),
"close_gripper": lambda: set_gripper(1),
"is_holding": self.is_holding,
"put_first_on_second": put_first_on_second,
}
def run(self, observations: queue.Queue, actions: queue.Queue) -> None:
"""Write the program for this episode and run it, then hold the arm still."""
scope = self.api(observations, actions)
try:
objects = scope["get_obj_names"]()
instruction = (self.obs.instruction or "").rstrip(".")
code = write(f"{PROMPT}\nobjects = {objects!r}\n# {instruction}.\n")
print(f"program for {instruction!r}:\n{code}")
functions = define(code, set(scope))
if functions:
print(f"functions it uses:\n{functions}")
# This runs code that the model wrote, on this machine.
exec(functions + "\n" + code, scope)
print("program finished")
except Stop:
return
except Exception as error:
print(f"program failed: {type(error).__name__}: {error}")
# Zeros hold the arm still and leave the gripper as it is.
try:
while True:
self.step([0, 0, 0, 0, 0, 0, 0], observations, actions)
except Stop:
return
# 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 writes | Result |
|---|---|---|
| Code as Policies, 2022 | Python that calls functions to find objects and move the arm | 97.2% of seen and 80% of unseen long-horizon tabletop tasks in simulation |
| Claude plays robotics, Anthropic, 2026 | A Python controller from observations to actions that runs every step | Beats direct control on balance and walking, the tasks it was tested on |