多模态模型在机器人任务中的应用

当我们思考 AI 的未来时，我们经常会落脚于机器人技术或"具身 AI"，将其视为下一个合理的前沿阵地。这是一场将大语言模型 (LLM) 的推理能力转化为现实世界中物理有用性的探索。

也就是说，机器人领域非常广阔。它涵盖从机械和电气工程到软件控制系统、计算机视觉，再到处理基本运动所需的复杂数学等一切。

虽然我来自于普通的物联网爱好者背景，但我仍在适应机器人领域。为了学习，我决定投身于一个项目，在其中我可以迎面撞上这些挑战并实时解决它们。在这个实验中，我使用 SOARM101 机器臂来玩幼儿棋盘游戏 First Orchard。

这个项目融合了两个强大的 AI 组件：

• 一个新训练的视觉语言动作 (VLA) 模型，用于控制机器人的物理动作。
• Gemini 3 Flash，作为"大脑"追踪游戏状态和规则。

在这篇文章中，我将详细介绍这个项目的具体内容、我所面临的困难，以及我是如何构建它的。

The Game: First Orchard

First Orchard 是一款为两岁幼儿设计的简单合作游戏。它由四个圆形树板、每个树板上的木质水果棋子、一条由方块组成的路径、一个乌鸦游戏棋子和一个六面骰子组成。

骰子面包括：

• 四种颜色：与四种水果类型相匹配（红色、蓝色、绿色、黄色）。
• 篮子：允许玩家从棋盘上移除任何水果棋子。
• 乌鸦：沿路径向前移动乌鸦游戏棋子一步。

目标：掷骰子，在乌鸦到达路径末端之前收获所有水果棋子。

在这个项目中，我使用了一个顶部网络摄像头来提供棋盘的"上帝视角"，以及一个安装在 SOARM101 腕部的相机用于近距离操纵。

为了保持环境对 VLA 模型的一致性，我用胶带固定了路径方块、树板和一个用作"收获"区域的小盒子，机器人将移除的棋子放在这里。虽然机器人可以到达每个树板，但由于桌面空间限制，乌鸦路径被故意放在可达范围之外。在这个项目的这个迭代中，乌鸦由人类玩家在掷出乌鸦面时手动移动。

Data Collection: The Grind

数据收集绝对是这个项目中最繁琐的部分。在花费数小时进行此工作后，我对为训练基础模型而收集数十万小时录像的团队产生了新的敬畏。

我最初考虑了两种方法：

• 任务分解：将动作分解为较小的块（伸手、抓握、提起），通过单个策略或逆运动学解决。
• 端到端策略：训练一个处理整个序列的单个策略——移动到水果、抓握、提起、移动到盒子、放下并返回主位置。

由于我的目标是在家庭环境中推动 VLA 能力的边界，我选择了端到端流程。我定义了四个特定的记录任务：

• "将红色水果放入盒子"
• "将蓝色水果放入盒子"
• "将绿色水果放入盒子"
• "将黄色水果放入盒子"

我的目标是每个任务记录 100 个片段，总共 400 个片段。为了完成它，我打开了一个电视节目，并开始使用领导者/追随者机器人组合进行遥操作机械臂的任务。

以下是我用来配置摄像头、机器人 ID 和数据集设置的命令：

lerobot-record \
    --robot.type=so101_follower \
    --robot.port=/dev/ttyACM0 \
    --robot.id=follower_arm \
    --robot.cameras="{ wrist_cam_left: {type: opencv, index_or_path: 8, width: 640, height: 480, fps: 30}, overhead_cam: {type: opencv, index_or_path: 4, width: 640, height: 480, fps: 30}}"  \
    --teleop.type=so101_leader \
    --robot.port=/dev/ttyACM1 \
    --teleop.id=leader_arm \
    --display_data=true \
    --dataset.repo_id=paultr/first-orchard \
    --dataset.num_episodes=25 \
    --dataset.single_task="Place the blue fruit in the box"

我分批记录数据，使用 --resume=true 标志继续添加到数据集。这是一个救星，允许我休息或在会话失败时恢复。当我最终完成时，我有了一个包含捕捉每个收获细节的短视频片段库。

你可以在 HuggingFace 上探索完整数据集，或观看下面的一组片段实际演示：

VLA Training: Trials, Errors, and Hardware Gore

数据收集完毕后，下一步是训练一个模型来处理实际的拾取放置操作。这是我对项目最好奇的部分，所以我进行了多次迭代，看看不同的模型和参数在这个特定环境中会如何表现。

Iteration 1: The Quick Test

我从一个非常快速的（～2 小时）SmolVLA 微调开始。我使用了一个 8 的小批大小和仅 20,000 步。我还将摄像头输入重命名为 camera1 和 camera2 以匹配 SmolVLA 约定。

!lerobot-train \
  --policy.path=lerobot/smolvla_base \
  --policy.repo_id=paultr/tmp_smolvla \
  --dataset.repo_id=paultr/first-orchard \
  --batch_size=8 \
  --steps=20000 \
  --output_dir=outputs/train/tmp_smolvla \
  --job_name=my_smolvla_training \
  --policy.device=cuda \
  --rename_map='{"observation.images.overhead_cam": "observation.images.camera1", "observation.images.wrist_cam_left": "observation.images.camera2"}'

结果是混合的。给定短训练时间，某些片段表现得非常好：

其他片段偏离目标、"抖动"或导致机器人无限悬停：

Iteration 2: Pushing the Boundary with SmolVLA

虽然我确信模型可以工作，但我很好奇是否更多资源会产生更平滑的结果。在下一步中，我用更大的批大小 (64) 训练 SmolVLA 100,000 步。我在 A-100 高内存 Colab 会话上运行了这个，因为说老实话，我很欣赏能够花费 Google 的钱来看我能把一个项目推到多远。你可以在这里找到我使用的 colab。

!lerobot-train \
  --policy.path=lerobot/smolvla_base \
  --dataset.repo_id=paultr/first-orchard \
  --batch_size=64 \
  --steps=100000 \
  --save_freq=10000 \
  --output_dir=/content/drive/MyDrive/outputs/first-orchard-smolvla \
  --job_name=smolvla_first_orchard \
  --policy.device=cuda \
  --policy.push_to_hub=true \
  --policy.repo_id=paultr/first-orchard-smolvla \
  --rename_map='{"observation.images.overhead_cam": "observation.images.camera1", "observation.images.wrist_cam_left": "observation.images.camera2"}'

这个新模型的质量明显更好。虽然仍然不是完美的，但它足够可靠来玩一个真正的游戏。

Iteration 3: The Pi0 Disaster

仍然对替代架构充满好奇，我决定尝试训练一个 Pi0-Fast 模型。我花费了大约五天时间在 A-100 上训练 200,000 步。由于 Colab 会话在 Pro 计划上大约 24 小时后关闭，我必须将 Google Drive 链接到定期保存和恢复检查点。你可以在这里找到我使用的 Colab。

!lerobot-train \
  --dataset.repo_id=paultr/first-orchard \
  --policy.type=pi0_fast \
  --policy.pretrained_path=lerobot/pi0fast-base \
  --output_dir=/content/drive/MyDrive/outputs/first-orchard_pi0_fast \
  --policy.repo_id=paultr/first-orchard_pi_fast \
  --steps=200000 \
  --batch_size=16 \
  --policy.chunk_size=32 \
  --policy.n_action_steps=8 \
  --policy.device=cuda \
  --save_freq=5000

当模型最终完成时，我遇到了真正的问题。

在测试期间，机械臂折叠在自身上，并损坏了 3D 打印的摄像头支架。幸运的是，我有额外的打印件，所以更换很快，但我无法通过任何代码更改或摄像头调整来使这个模型正常工作。

另一方面，在尝试调试损坏的过程中，我学到了很多关于 LeRobot 可选训练标志的知识。最终，我决定用 SmolVLA 模型完成项目，但我有一天会重新访问其他模型。

Setting up the Python App

现在训练工作已经完成，我们可以进入实际代码！我们将从设置基础项目结构开始。这处理了与 SOARM101 的基本连接并建立了一个"主位置"姿态。即使很简单，第一次运行脚本时机器人自己移动到一个姿态总是有一点兴奋感的。

import sys
import time
import random
import torch
import numpy as np
from PIL import Image
from pydantic import BaseModel, Field

from lerobot.cameras.opencv.configuration_opencv import OpenCVCameraConfig
from lerobot.datasets.utils import hw_to_dataset_features
from lerobot.policies.factory import make_pre_post_processors
from lerobot.policies.smolvla.modeling_smolvla import SmolVLAPolicy
from lerobot.policies.utils import build_inference_frame, make_robot_action
from lerobot.robots.so_follower import SO101Follower, SO101FollowerConfig

FOLLOWER_PORT = "/dev/ttyACM0"
FOLLOWER_ID = "follower_arm"
ROBOT_TYPE = "so101_follower"

# 预定义的电机位置，将手臂恢复到安全起始状态
HOME_VALUES = {
    'shoulder_pan.pos': -3.7974683544303787, 'shoulder_lift.pos': -98.51155503329416,
    'elbow_flex.pos': 98.1199641897941, 'wrist_flex.pos': 76.095278604849,
    'wrist_roll.pos': -51.585014409221905, 'gripper.pos': 0.990099009901
}

device = None
robot = None

def home():
    if robot:
        robot.send_action(HOME_VALUES)
        time.sleep(1)

def init():
    global robot
    print("Connecting to Robot...")
    # 在最终版本中，camera_config 将在这里定义
    robot_cfg = SO101FollowerConfig(port=FOLLOWER_PORT, id=FOLLOWER_ID, cameras=camera_config)
    robot = SO101Follower(robot_cfg)
    robot.connect()
    home()

def main():
    init()
    if robot:
        robot.disconnect()

if __name__ == "__main__":
    main()

有了基础框架，我们需要更新脚本来加载我们辛苦训练的 VLA 模型。首先，我们在文件顶部添加一些配置常数，包括模型 ID 和相机设置。

注意 MAX_STEPS 的值：以 30 FPS 的帧率，900 步大约给机器人 30 秒的时间来完成拾取放置任务，之后它就会"放弃"。

MAX_STEPS = 900 
TOP_CAM_INDEX = 6
TOP_CAM_WIDTH = 640
TOP_CAM_HEIGHT = 480
TOP_CAM_FPS = 30

WRIST_CAM_INDEX = 4
WRIST_CAM_WIDTH = 640
WRIST_CAM_HEIGHT = 480
WRIST_CAM_FPS = 30

MODEL_ID = "paultr/first-orchard-smolvla-run2"
TASK_TEMPLATE = "Place the {} fruit in the box"

model = None
preprocess = None
postprocess = None
dataset_features = None

接下来，让我们改进 init() 函数。这个版本将模型加载到 GPU 内存中，并设置预处理器和后处理器，以桥接原始相机像素和模型的动作输出之间的差异。

def init():
    global device, robot, model, preprocess, postprocess, dataset_features
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

    model = SmolVLAPolicy.from_pretrained(MODEL_ID)
    model.to(device)

    preprocess, postprocess = make_pre_post_processors(
        model.config,
        MODEL_ID,
        preprocessor_overrides={"device_processor": {"device": str(device)}},
    )

    camera_config = {
        "camera1": OpenCVCameraConfig(
            index_or_path=TOP_CAM_INDEX, width=TOP_CAM_WIDTH, height=TOP_CAM_HEIGHT, fps=TOP_CAM_FPS),
        "camera2": OpenCVCameraConfig(
            index_or_path=WRIST_CAM_INDEX, width=WRIST_CAM_WIDTH, height=WRIST_CAM_HEIGHT, fps=WRIST_CAM_FPS),
    }

    print("Connecting to Robot...")
    robot_cfg = SO101FollowerConfig(port=FOLLOWER_PORT, id=FOLLOWER_ID, cameras=camera_config)
    robot = SO101Follower(robot_cfg)
    robot.connect()

    home()

    # 将硬件功能映射到数据集功能以供 VLA 使用
    action_features = hw_to_dataset_features(robot.action_features, "action")
    obs_features = hw_to_dataset_features(robot.observation_features, "observation")
    dataset_features = {**action_features, **obs_features}

最后，我们添加 run_task()。这是核心的 VLA 推理循环。它接受一种颜色（红、蓝、绿或黄），将其格式化为任务模板，并要求模型逐步生成机器人动作，直到收获完成。

def run_task(color):
    for i in range(MAX_STEPS):
        obs = robot.get_observation()

        # 构建要给模型"看"和"读"的帧
        obs_frame = build_inference_frame(
            observation=obs,
            ds_features=dataset_features,
            device=device,
            task=TASK_TEMPLATE.format(color),
            robot_type=ROBOT_TYPE
        )

        processed_obs = preprocess(obs_frame)
        action = model.select_action(processed_obs)
        action = postprocess(action)

        # 将模型输出转换回物理电机命令
        robot_action = make_robot_action(action, dataset_features)
        robot.send_action(robot_action)

使用 Gemini 构建游戏循环

现在基础设施已经就位，是时候真正玩这个游戏了。First Orchard 的回合由特殊的骰子驱动，所以我们将从在代码中模拟这个开始。

def roll_dice():
    outcomes = {1: 'crow', 2: 'blue', 3: 'red', 4: 'green', 5: 'yellow', 6: 'basket'}
    res = outcomes[random.randint(1, 6)]
    print(f"Dice Roll: {res}")
    return res

接下来，我们更新 main() 函数以应对这些结果。如果掷出一种颜色（2-5），我们调用我们的 VLA 任务。如果掷出乌鸦或篮子，我们需要一些专门的逻辑。

def main():
    init()
    try:
        while True:
            home()

            result = roll_dice()

            if result == 'crow':
                move_crow()
            elif result == 'basket':
                basket_rolled()
            else:
                run_task(result)
    except KeyboardInterrupt:
        print("\nStopping game...")
    finally:
        if robot:
            robot.disconnect()

添加"大脑"：Gemini 3 Flash

为了让乌鸦和篮子选项起作用，机器人需要进行一些更高层的推理。例如，如果乌鸦移到最后一个方块，游戏就结束了。如果掷出篮子，机器人需要查看棋盘并决定哪种水果颜色最"急需"收获。

为了处理这个问题，我引入了 Gemini 3 Flash。

from google import genai
from google.genai import types

GEMINI_MODEL_NAME = "gemini-3-flash-preview"
GENAI_CLIENT = genai.Client()

注意：你需要通过 pip install -q -U google-genai 安装依赖，并通过 export GEMINI_API_KEY="your_key" 导出你的 API 密钥。这是我几乎总是忘记的步骤，所以在运行前一定要检查你的环境变量！

为了获得准确的结果，我为 Gemini 提供了详细的系统指令和提示。这帮助模型区分木制水果、圆形树形方块和机器人手臂本身。

SYSTEM_INSTRUCTIONS = "You will receive images from a top-down perspective of a simple game called First Orchard. There are four circular tiles that can each have between 0 and 4 game pieces shaped like a fruit with the colors green, blue, red, and yellow. All fruit of the same color will be grouped together on their own circle tile. There is also a line of tiles representing a path with a crow game piece on one of the tiles at all times. The end of that path is represented by the tile with a rounded end. There will be a blue robot arm on the edge of the image that should not be relevant to the tasks you will be asked to perform. There is also a box beside that robot arm that may be empty or may contain fruit game pieces. Fruit game pieces in the box are considered out of play and should be ignored during counting operations."

BASKET_PROMPT = "Locate the game pieces shaped like red, blue, green, and yellow fruit. Ignore any that are in the box besides the robot. Count how many pieces are in play per group. You should return the 'color' of whichever grouping has the most pieces. Select at random if there is a tie for the most pieces in a group."

结构化输出和代理视觉

为了在 Python 中更容易处理响应，我使用了 Gemini 的结构化输出功能。Gemini 不会返回杂乱的字符串，而是返回基于 Pydantic 模型的干净对象。

class BasketColor(BaseModel):
    color: str = Field(description="Color of the group with the most pieces in play.")

现在我们可以实现 basket_rolled()。你会注意到这里有两个关键设置：

Thinking Level：对于简单的感知，我将其保持在低级以节省延迟。

ToolCodeExecution：这启用了代理视觉。Gemini 实际上可以"放大"和裁剪图像以更好地查看方块，然后再做出决策。这对于正确计数小木制水果来说是完美的。

def basket_rolled():
    image = get_image_from_robot()

    response = GENAI_CLIENT.models.generate_content(
        model=GEMINI_MODEL_NAME,
        contents=[BASKET_PROMPT, image],
        config=types.GenerateContentConfig(
            system_instruction=SYSTEM_INSTRUCTIONS,
            response_mime_type="application/json",
            response_json_schema=BasketColor.model_json_schema(),
            thinking_config=types.ThinkingConfig(include_thoughts=False, thinking_level="low"),
            tools=[types.Tool(code_execution=types.ToolCodeExecution())]
        )
    )

    basket_data = BasketColor.model_validate_json(response.text)
    run_task(basket_data.color)

为了向 Gemini 发送图像，我们需要从机器人的相机中获取帧。由于 VLA 已经在使用它们，我们可以从 camera1（俯视）馈送中拉取最新观测值，并将其转换为 Gemini 能理解的格式。

def get_image_from_robot():
    obs = robot.get_observation()
    pixel_values = obs["camera1"]

if torch.is_tensor(pixel_values):
        img_np = pixel_values.to(torch.uint8).cpu().numpy()
    else:
        img_np = pixel_values.astype(np.uint8)

    # Ensure the dimensions are correct for PIL (H, W, C)
    if img_np.ndim == 3:
        if img_np.shape[0] == 3:
            img_np = np.transpose(img_np, (1, 2, 0))

    return Image.fromarray(img_np)

最后，我们来处理乌鸦。由于乌鸦的运动路径超出了我这个特定机器人设置的能力范围，我选择了"人类在环"的方法。

def move_crow():
    print("ACTION REQUIRED: Move the crow one step forward manually!")
    time.sleep(10)

此时，机器人可以移动水果并处理骰子，但它实际上不知道游戏何时结束。如果树木是空的或乌鸦到达最后的瓷砖，循环会一直运行。要解决这个问题，我们需要让 Gemini 在每一轮之后充当裁判。

首先，让我们定义一个新的提示词和 Pydantic 模型来处理游戏的胜负状态。

GAME_END_PROMPT = "Analyze the image to determine the current state of the game. Look at the path tiles: if the crow piece is on the final tile (rounded edge with 'HABA'), the game is lost (result: 'crow'). Look at the four circular fruit tiles: if all fruit pieces have been removed from these tiles, ignoring any in the box, the game is won (result: 'empty'). If neither of these conditions is met, the game continues (result: 'continue'). Return the corresponding string result. Zoom in and crop on the path tiles and circular tiles to get a clearer view if needed."

class GameState(BaseModel):
    status: str = Field(description="The current state of the game: 'crow', 'empty', or 'continue'.")

现在，我们可以实现 check_game_state()。这个函数要求 Gemini 查看俯视图，并告诉我们是应该继续玩还是停止。

def check_game_state():
    print("Checking game state...")
    image = get_image_from_robot()

    response = GENAI_CLIENT.models.generate_content(
        model=GEMINI_MODEL_NAME,
        contents=[GAME_END_PROMPT, image],
        config=types.GenerateContentConfig(
            system_instruction=SYSTEM_INSTRUCTIONS,
            response_mime_type="application/json",
            response_json_schema=GameState.model_json_schema(),
            thinking_config=types.ThinkingConfig(include_thoughts=False, thinking_level="low"),
            tools=[types.Tool(code_execution=types.ToolCodeExecution())]
        )
    )

    game_data = GameState.model_validate_json(response.text)

    if game_data.status == 'crow':
        print("GAME OVER: The crow has reached the orchard!")
        sys.exit(0)
    elif game_data.status == 'empty':
        print("GAME OVER: You win! All fruits are harvested.")
        sys.exit(0)

优化游戏流程

每一轮都调用 Gemini 是有效的，但我们可以更聪明一些。例如，如果机器人掷出"红色"但我们已经知道红树是空的，这就是浪费了一个回合——如果我们要求 Gemini 检查整个棋盘，那也是浪费了一个 API 调用。

为了优化这一点，我引入了一个 should_check_game_state 标志和一个称为 EMPTY_FRUITS 的集合来在本地缓存棋盘的状态。

should_check_game_state = True
EMPTY_FRUITS = set()

# We update run_task and move_crow to set should_check_game_state = True 
# whenever a physical change happens on the board.

接下来，我添加了 check_fruit_exists()。这个函数特别检查掷出的颜色是否仍然可用。如果 Gemini 看到树是空的，我们会将该颜色添加到 EMPTY_FRUITS 集合中，这样下次掷出该颜色时就可以跳过检查。

FRUIT_CHECK_PROMPT = "Look at the four circular tiles. Is there at least one {} fruit piece remaining on its designated tile? Ignore pieces inside the box. Return True if at least one is present, otherwise return False. Crop and zoom in on the individual circular tiles to help ignore the box and correctly identify the presence of fruit game pieces."

class FruitExists(BaseModel):
    exists: bool = Field(description="True if the specified fruit color is still on the circular tiles, False otherwise.")

def check_fruit_exists(color):
    if color in EMPTY_FRUITS:
        print(f"Skipping: {color} fruit is already gone.")
        return False

    image = get_image_from_robot()
    print(f"Gemini: Checking if {color} fruit is available...")

    response = GENAI_CLIENT.models.generate_content(
        model=GEMINI_MODEL_NAME,
        contents=[FRUIT_CHECK_PROMPT.format(color), image],
        config=types.GenerateContentConfig(
            system_instruction=SYSTEM_INSTRUCTIONS,
            response_mime_type="application/json",
            response_json_schema=FruitExists.model_json_schema(),
            tools=[types.Tool(code_execution=types.ToolCodeExecution())]
        )
    )

    fruit_data = FruitExists.model_validate_json(response.text)
    if not fruit_data.exists:
        EMPTY_FRUITS.add(color)

    return fruit_data.exists

最后，我们更新主循环来将所有这些部分联系在一起。机器人现在自动回位，验证游戏未结束，掷骰子，并使用 Gemini 确认移动有效，然后再将控制权交给 VLA 进行物理收获。

def main():
    init()
    try:
        while True:
            home()
            check_game_state()

            result = roll_dice()

            if result == 'crow':
                move_crow()
            elif result == 'basket':
                basket_rolled()
            else:
                if check_fruit_exists(result):
                    run_task(result)
                else:
                    continue
    except KeyboardInterrupt:
        print("\nStopping game...")
    finally:
        if robot:
            robot.disconnect()

if __name__ == "__main__":
    main()

未来探索：超越"完成"

虽然我对这个项目的成果非常满意，但我深信这样一句话："完美是进度之敌。"在机器人学中，你可能会花一辈子时间来调整