Strands Robots로 잇는 로봇 학습 데이터 루프

from strands import Agent
from strands_robots import Robot

sim = Robot("so100")  # mode="sim" (default - safe, no hardware)
agent = Agent(tools=[sim])

# Record a demonstration and sync it to a bucket.
agent("Record a pick-the-cube demo and sync it to my-org/robot-fave.")

# Stream it back from the bucket to train, without downloading it first.
for batch in sim.stream_dataset("my-org/robot-fave/cube_pick", repo_type="bucket").dataloader(batch_size=64):
    ...

reader = sim.stream_dataset("my-org/robot-fave/cube_pick", repo_type="bucket", shuffle=False, max_num_shards=1, buffer_size=1, # one episode, in capture order
)
print(reader.num_episodes, reader.num_frames, reader.fps)
for frame in reader:
    frame["observation.images.front"] # (3, H, W) tensor, decoded on the fly from the MP4 shard
    frame["observation.state"] # joint vector, from the Parquet shard
    frame["action"]
    break

# policy here is a LeRobot policy you constructed, such as ACTPolicy.
for batch in reader.dataloader(batch_size=64, num_workers=4):
    loss, _ = policy(batch) # lerobot ACTPolicy.forward returns (loss, loss_dict)
    loss.backward()

import os
os.environ["STRANDS_TRUST_REMOTE_CODE"] = "1" # create_policy loads with trust_remote_code=True
from strands_robots import create_policy
from strands_robots.training import TrainSpec, create_trainer
trainer = create_trainer("lerobot_local", device="cuda")
spec = TrainSpec(dataset_root="/tmp/cube_pick", output_dir="/tmp/cube_pick_ft", base_model="", steps=500, extra={"policy_type": "act"})
result = trainer.train(spec)
# train ACT on the streamed dataset
policy = create_policy(result.checkpoint_dir) # load the checkpoint straight back

robot = Robot("so100", mode="real", port="/dev/ttyACM0", cameras={"front": {"type": "opencv", "index_or_path": "/dev/video0", "fps": 30}})
agent = Agent(tools=[robot])
agent("Pick up the red cube.")