跳转到内容

正在加载搜索…

编写一个 skill

本教程添加一个 skill wiggle_wrist(arm, amplitude, seconds),让一个手腕关节在当前位置附近来回摆动: 它和所有运动一样是一次有界执行,因此 guard、区域、安全层(shield)和 handler 对它的作用与对 move_eef 相同。

skills/wiggle/skill.toml:

name = "wiggle"
summary = "Rock a wrist joint around where it is."
layer = "base"
mixin = "skills.wiggle:Wiggle"
exports = ["wiggle_wrist"]
order = 95

layer = "base" 把它放入 agent 的 skill 中(也放入每个消融的 skill 中);needs = [...] 用来列出它运行 所必需的服务类型(它不需要任何服务)。order 决定它在 session 的方法解析顺序和 prompt 中排在 motion(90)之后。全部键见 配置参考。

skills/wiggle/__init__.py:

"""Rock a wrist joint: a demonstration of a bounded execution."""
import math
class Wiggle:
def wiggle_wrist(self, arm, amplitude=0.2, seconds=1.0):
"""Rock the last joint of `arm` by ±amplitude rad around where it is, one cycle
per second, for `seconds`. Returns the execution trace; it ends COMPLETED, or
with the event that stopped it (a guard, the shield, the region)."""
joint = self._arm(arm)["joints"][-1]
start = self._base_command(self.state)
def command(state, tick):
target = list(start)
target[joint] += amplitude * math.sin(2 * math.pi * tick / self.hz)
return target
return self._execute("wiggle_wrist", {"arm": arm, "amplitude": amplitude}, command,
max_seconds=seconds, region=None, until=None, timeout_is_nominal=True)

episode 核心(runtime/episode.py)和机器人的 session(runtime/session.py)为机器人 skill 提供的内容:

self._execute(skill, parameters, command_fn, max_seconds, region, until, timeout_is_nominal=False) 每个控制 tick 运行一次 command_fn(state, tick),直到它返回 None(完成)或一个 Event、某个 guard 触发、until 成立、手臂离开 region,或超过 max_seconds;每条命令都经过安全层。返回 trace;非正常事件交给程序的 handler 或上报。
self._base_command(state) 测得的手臂关节值,每个夹爪保持其最后一条命令(保持静止)
self._arm(arm) 手臂的关节组:{"joints": [...], "gripper": index}
self.state、self.hz、self.profile 机器人状态、控制频率、embodiment 配置
self.env.call(name, **arguments) 一次 worker 调用:observe、fk、ik、plan 等
self._failed(skill, parameters, reason) 一个 SKILL_ERROR trace,用于 skill 无法执行的请求
其他 skill 的方法 self.move_eef(...)、self.segment_text(...) 等

skill 模块还可以定义 turn_report(session),session 在每个程序之后调用它:它返回 {"text": ..., "images": {camera: data URL}}(或 None),文本进入模型的下一轮,图像替换对应相机的 普通视图。按物体 id 工作的工具(skills/eef)用它在每一轮重新测量已定位的物体。

base skill 在每个配置中都启用,因此每个 prompt pack 都必须描述它(测试会检查已启用 skill 的每个公开函数 都作为调用出现)。在 prompts/robodojo/core.md 中,与其他运动放在一起:

<!-- if: wiggle -->
wiggle_wrist(arm, amplitude=0.2, seconds=1.0) rock the wrist joint around where it is
<!-- end -->

prompts/robotwin/core.md 中也写同样的内容。不应包含该 skill 的配置把它列在 [skills] disable = ["wiggle"] 中;显式列出 skill 的配置(names = [...])只有在列出它时才会包含它。

Terminal window
python -m unittest discover -s tests -p test_skills.py
harness config configs/experiments/dev.toml # wiggle yes

不用模型,直接在一个运行中的 worker 上试用:

Terminal window
python scripts/dev_console.py --reset stack_bowls 0
python scripts/dev_console.py --code "print(wiggle_wrist('left_arm', 0.3, 2.0))"

dev_console.py 针对 --endpoint(默认 http://127.0.0.1:8771)上的 worker 构建一个包含所有层的 session,运行代码并打印各次执行。

调用模型的 skill 在 needs(全部需要)或 needs_any(需要其中之一)中写明服务类型,并通过 session 的 客户端访问它(self._vision() 用于 VLM、分割器、各个策略)。skill 本身在程序的进程中运行,保持为 标准库 Python;模型的重依赖都属于服务,服务在某次运行首次启用该 skill 时获得自己的环境 (服务与环境)。关闭该服务的配置会去掉这个 skill,prompt 也随之去掉 它的文本:

name = "vlm"
summary = "Questions about images to a vision-language model."
layer = "base"
mixin = "skills.vlm:VisionQuestions"
exports = ["query_vlm"]
needs = ["vlm"]
order = 40
"""Questions about images to a vision-language model."""
class VisionQuestions:
def query_vlm(self, images, question):
"""Ask the vision model a question about one or more RGB arrays."""
if not isinstance(images, (list, tuple)):
images = [images]
return self._vision().ask(list(images), question)