垃圾桶还是收纳盒?我给八界写了一个"分类归位"Skill。家里有娃的都知道那个场面:客厅地毯上,毛绒玩具和半包饼干、捏皱的纸团、喝完的酸奶瓶混在一起。垃圾桶就在墙角,收纳筐就在电视柜旁边,可是娃不会用,大人弯一晚上腰。

先介绍一下主角:八界

写正文之前,先给还不熟悉的朋友介绍一下这次的主角。

相关入口放在这里,建议直接收藏:

八界(Bajie) 是科沃斯推出的首款全栈开源具身智能机器人——注意是"全栈开源",从硬件基座到软件系统全部开放。本体上搭载六自由度机械臂、二指灵巧夹爪、双 RGBD 相机(头部 + 臂部)、机身升降和双 RK3588 融合 SoC,属于那种"能真的在家里走来走去干活"的机器人,不是一个摆件。

在这里插入图片描述

对开发者来说更关键的是软件这一层:

它开放了 ROS 2 完整接口、C++ / Python 双语言 SDK(45 项原子能力,覆盖建图导航、视觉感知、目标搜索、机械臂控制、抓取放置、语音交互),配套 2800+ 行结构化开发文档,官方示例场景的完整源码也是开放的,另外还有 Gazebo/Isaac Sim 仿真环境和 RKNN 模型量化工具——没摸到真机之前,大部分逻辑可以先在仿真里跑通。

在这里插入图片描述

我的这次分享,就是基于这套 Python SDK 开发的一个 Agent 技能包(Skill)。

在开始使用八界机器人之前,有一个前提,就是需要先在设备上操作——开始建图。起初上来就操作,漏了这关键的前提会导致后续操作失败。第一次使用需让机器人巡视家里并创建地图,后续所有导航/定位操作都依赖这张地图。

在这里插入图片描述

巡视完毕后会创建一个家里的地图,可以在后台的web可视化平台端查看:

在这里插入图片描述

一、从一个每天都在上演的场景说起

家里有娃的都知道那个场面:客厅地毯上,毛绒玩具和半包饼干、捏皱的纸团、喝完的酸奶瓶混在一起。垃圾桶就在墙角,收纳筐就在电视柜旁边,可是娃不会用,大人弯一晚上腰。

我当时想得很朴素,不追求大而全,就面向具体生活场景解决生活中的一个小痛点:垃圾桶和收纳筐明明就在旁边,能不能让机器人自己去判断——地上这个东西是垃圾,就扔进垃圾桶;是玩具,就捡进收纳筐?不用我一件一件指着说"这个放这、那个放那"。

这个想法落地之后,就是我这次提交给八界开源生态的作品:bajie-smart-sorting,地面分类归位 Skill。垃圾进桶,玩具进筐,一轮一轮,直到地面清空。

二、为什么是这个题目

翻官方文档的时候,我在硬限制清单里看到一行字:“不支持玩具自动分类,需分多次任务或人工协助,规则由用户显式指定”。

官方已有的四个 Demo——取物递物、桌面整理、玩具收纳、鞋子整理——本质上都是"单一容器搬运":东西从一个地方搬到另一个地方,去哪里是人提前指定好的。没有一个 Demo 需要机器人自己决定去哪里。

这恰恰是家庭场景里最真实的一环。真实的地面不是"地上有五个玩具"这种干净命题,而是垃圾和玩具混在一起。如果机器人只会把所有东西一股脑搬进收纳筐,那纸团也进了筐,酸奶瓶也进了筐,这个收纳其实没有解决痛点。反过来,只会捡垃圾也不行——娃的玩具你总不能当垃圾处理了。

所以我要做的核心就一件事:在抓起每件东西之前,先回答"它是垃圾还是玩具"。

三、双级分类:OVD 打主力,VLM 只做兜底

拿到需求,第一反应是全都交给大模型:VLM 看一眼图,问它"这是垃圾还是玩具",不就完了?

真上手才知道不行。八界的 VLM 接口推理比较慢,官方文档明确建议 timeout 设到 60 秒以上,默认 10 秒极易超时。如果每一件物品都要等一次 VLM,收十个东西就是几分钟的干等,体验完全崩掉。

所以我设计了一个双级分类:

  1. 第一级,OVD 标签路由。 eco_detect_objects 是开放词汇检测,速度快,可以一次携带多组标签。我把官方 OVD_LABELS 词表里的物品按去向分成了两组——垃圾类(纸团、零食、易拉罐、瓶子、湿巾)和玩具类(玩具汽车、玩偶、魔方、玩具卡片……),一次检测同时传两组标签,命中哪个组就路由到哪个容器。绝大多数情况这一级就解决了。

  2. 第二级,VLM 词表兜底。 只有当两类标签都没命中时——比如一个 OVD 词表里没有的奇怪零食包装——才调用 eco_vlm_perception 拿到物品的自由名称,再用本地关键词规则判断:名字里带"纸、袋、罐、瓶、壳、皮、屑"的判为垃圾,其余默认按玩具收纳。

VLM 从主路径降到兜底路径之后,整个流程的高频环节全是快接口,体验立住了。

在这里插入图片描述

这里有一个我坚持的设计原则:宁可错收,不可错扔。分类拿不准时,默认按玩具处理放进收纳筐——纸团进了收纳筐,顶多再扔一次;娃的玩具进了垃圾桶,那就是事故了。

还有一个小细节:每件物品分类之后、动手之前,机器人都必须播报一句"这是纸团,我把它扔进垃圾桶"。这不只是仪式感——用户听到分类错了可以立刻纠正,比事后从收纳筐里往外翻要体面得多。

---
name: qq8864-bajie-smart-sorting
description: 地面物品分类归位:识别地上的是垃圾还是玩具/物品,垃圾投放进垃圾桶,玩具与物品收纳进收纳筐。采用 OVD 标签路由 + VLM 语义兜底双级分类,循环执行直到地面清空。适用于带娃家庭玩具遍地、零食包装乱丢等场景。在用户提到分类归位、垃圾分类、收拾地面、捡垃圾、玩具收纳、把垃圾扔进垃圾桶时启用。
---

# Skill: bajie-smart-sorting — 地面分类归位

一句话:**机器人巡一圈地面,看到的垃圾扔进垃圾桶,看到的玩具/物品放进收纳筐,一轮一轮直到清空。**

与官方「玩具收纳」Demo 的区别:玩具收纳只有一个去向(收纳筐),本 Skill 有**两个去向**,
需要先对每个目标做"是垃圾还是玩具"的分类判断,再路由到对应的容器。
......
---

## 🎯 使用场景

**USE when:**
- 用户说"收拾一下地上的东西/把垃圾和玩具分一分"
- 地面散落物包含垃圾(纸团、零食袋、瓶子)和玩具两类,需要分别归位

**DON'T use when:**
- 地上只有单一类物品(直接用官方玩具收纳/鞋子分拣脚本更稳)
- 液体、高温、尖锐物(夹爪无温度感应,超能力边界,须提醒人工处理)
- 电线、硬币等 <1cm 级小物(超出抓取精度)

---

## 📦 前置条件

| 项 | 要求 |
|----|------|
| Python | **3.10**(cp310),已安装 `bajie_sdk` wheel |
| 语义建图 | 已完成,且地图中存在"垃圾桶"与"收纳筐"(或用户自定义容器名) |
| VLM | 已在 `http://<机器人IP>:17890` Web 配置页填入 VLM API KEY(OVD 命中时可不依赖) |
| 电量 | > 30%,否则先回充并拒绝新任务 |

---

四、单轮流程怎么跑

整个 Skill 的单轮流程是这样的:

findObject(垃圾桶) + findObject(收纳筐)     ← 先把两个容器的框拿到手
  → findObject(地面物品, filter_boxes=两个容器框)
  → lookto(ARM) → captureImages(ARM) → detect_objects(垃圾+玩具标签)
  → 未命中 → captureImages(HEAD) → vlm_perception → 词表兜底
  → pick → place_in(对应容器) → 语音播报 → 下一轮
  → 搜不到物品了 → finishRobotPose → 汇总播报 → 结束

八界分类归位 Skill:单轮执行流程:

在这里插入图片描述

几个值得说的设计:

容器框复用。 每轮开始前,用两个容器的 6D 框作为 filter_boxes 传给找物接口,把已经进筐、进桶的物品从搜索结果里过滤掉。这样循环才收得住——不然机器人会盯着自己刚放进去的东西再抓一遍,无限套娃。

循环有终止条件。 搜不到物品时,再用 VLM 兜底看一眼确认真的没有了,然后 finishRobotPose 收尾,播报一句"扔进垃圾桶 X 件、放进收纳筐 Y 件"。默认最多 10 轮,防止某些检测抖动导致停不下来。

失败不静默。 抓取和投放各自重试 3 次(抓取中遇到 36897 物体掉落错误,按官方说明当普通重试处理);连续失败就播报"有 N 件没成功,麻烦你看一下",绝不装作无事发生。电量低于 30% 直接播报去回充,拒绝新任务。

五、踩过的坑,都写进了 Skill

调参过程踩的坑,比写流程花的时间还多,几个印象深的:

  • 参数格式是重灾区。 pick 的 bbox 必须是 {"name": "...", "bbox": [...]} 的命名格式,裸数组直接报 [0x2004] 任务参数无效;lookto 反过来要传裸数组,包一层对象又报"Box框异常"。这两个刚好相反,写的时候得反复对着 common-pitfalls 文档检查。
  • OVD 标签不能自造。 detect_objects 的 labels 只能用官方 OVD_LABELS 词表里的中文标签,自己写一个"垃圾"上去就报 Illegal。所以分类不是让检测模型输出"垃圾/玩具",而是提前把官方词表拆成两组——这个思路转变之后,第一级分类其实就通了。
  • VLM 当主路径会拖垮体验。 这个前面说了,timeout 60 秒起步的接口,只配待在兜底位置。
  • 语音播报的目录要求。 播报脚本必须在技能根目录下执行,它是通过 WebSocket 把 speech 任务发到机器人服务端,不是本地播放——在别的目录跑,程序不报错,但机器人就是不开口。

这些坑我全部固化进了 Skill 包的 SKILL.md 和代码注释里,后来的开发者照着表格排查,能省下我这几个晚上的时间。

六、包结构与复用

整个作品按照官方技能包范式组织,可以直接放进 OpenClaw / Hermes 的技能目录使用:

bajie-smart-sorting/
├── SKILL.md                          # Agent 入口:触发条件、SOP、陷阱表、回退链路
└── scripts/smart_sorting/
    ├── smart_sorting_demo.py         # CLI 编排主流程
    ├── classify.py                   # 双级分类:词表可自行扩展
    └── README.md                     # 使用说明

想改分类规则只动 classify.py 一个文件:垃圾类、玩具类的 OVD 标签各自一组,VLM 兜底的关键词词表也集中在这里。比如你家没有零食问题但快递盒满天飞,往垃圾类里加标签就行。

# -*- coding: utf-8 -*-
"""分类模块:OVD 标签路由 + VLM 词表兜底。

垃圾类与玩具类标签全部取自官方 OVD_LABELS(见 references/capabilities/perception.md 附录),
不要在此列表之外自造标签,否则 detect_objects 会报 Illegal。
"""

# detect_objects 一次性携带的两组标签(OVD 允许一次传多个标签)
TRASH_LABELS = ["纸团", "零食", "易拉罐", "瓶子", "湿巾"]

TOY_LABELS = [
    "玩具汽车", "玩具球", "玩具恐龙", "玩具兔子", "玩具小熊", "玩具枪",
    "玩具飞机", "玩具积木", "玩偶", "魔方", "棋盘", "陀螺", "玩具卡片",
    "气球", "其他玩具",
]

# detect_objects 时合并传入
ALL_LABELS = TRASH_LABELS + TOY_LABELS

# VLM 兜底用的名称关键词:命中即判为垃圾
TRASH_NAME_KEYWORDS = [
    "纸", "袋", "罐", "瓶", "壳", "皮", "屑", "包装", "零食", "湿巾",
    "团", "果核", "果皮", "烟",
]


def classify_by_label(name: str) -> str:
    """OVD 第一级路由:按命中的检测标签判断。

    返回 "trash"(垃圾)或 "toy"(玩具/物品)。
    """
    if name in TRASH_LABELS:
        return "trash"
    if name in TOY_LABELS:
        return "toy"
    return classify_by_name(name)


def classify_by_name(name: str) -> str:
    """VLM 第二级兜底:按物品名称关键词判断。

    VLM 返回的是自由名称(不在 OVD 词表内),用规则词表路由。
    无法判断时默认按玩具/物品收纳(宁可错收,不可把玩具当垃圾扔)。
    """
    for kw in TRASH_NAME_KEYWORDS:
        if kw in name:
            return "trash"
    return "toy"


def classify_item(name: str) -> str:
    """对外入口:给定物品名,返回目标容器类别 "trash" / "toy"。"""
    return classify_by_label(name)


def route_target(name: str, trash_name: str, box_name: str) -> str:
    """分类结果 → 容器语义名,并给出播报话术。"""
    cls = classify_item(name)
    if cls == "trash":
        return trash_name, f"这是{name},我把它扔进{trash_name}"
    return box_name, f"这是{name},我把它放进{box_name}"
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""地面分类归位主流程(CLI 编排版)。

链路:找两个容器 → 循环(搜地面物品 → OVD/VLM 分类 → pick → place_in 路由容器 → 播报)。
通过 `bajie_sdk run eco_*` 子进程编排,中间产物 JSON 存放在工作目录 ./_smart_sorting/ 下。
"""

import argparse
import json
import os
import subprocess
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from classify import ALL_LABELS, route_target  # noqa: E402

WORK_DIR = os.path.join(os.getcwd(), "_smart_sorting")
MAX_RETRY = 3


def run_cli(args: list, json_body: str, out_file: str, timeout: int = 120) -> dict:
    """执行 bajie_sdk run <method>,返回 -o 落盘的 JSON 结果。"""
    os.makedirs(WORK_DIR, exist_ok=True)
    out_path = os.path.join(WORK_DIR, out_file)
    cmd = ["bajie_sdk", "run"] + args + ["--json", json_body, "-o", out_path]
    proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if proc.returncode != 0:
        raise RuntimeError(f"{' '.join(args)} 失败: {proc.stderr.strip() or proc.stdout.strip()}")
    with open(out_path, encoding="utf-8") as f:
        return json.load(f)


def speech(text: str) -> None:
    """语音播报:必须在技能根目录下执行,通过 WebSocket 发送到机器人。"""
    try:
        subprocess.run(["python3", "scripts/speech.py", text],
                       capture_output=True, text=True, timeout=30)
    except Exception:
        pass  # 播报失败不阻塞主流程


def find_container(name: str) -> dict:
    """定位语义容器,返回其 6D 框(供 filter_boxes / carrier_bbox 使用)。"""
    res = run_cli(["eco_findObject"], json.dumps({"name": name}), f"find_{name}.json", timeout=180)
    boxes = res.get("boxes") or res.get("items") or []
    if not boxes:
        raise RuntimeError(f"找不到语义容器「{name}」,请确认语义地图中已添加")
    return boxes[0]


def search_ground_items(trash_box: dict, box_box: dict, area_name: str) -> list:
    """搜索地面物品,用两个容器框过滤(排除已收纳/已投放的)。"""
    body = {
        "name": "",
        "filter_boxes": [trash_box.get("bbox_6d") or trash_box, box_box.get("bbox_6d") or box_box],
    }
    if area_name:
        body["name"] = area_name
    res = run_cli(["eco_findObject"], json.dumps(body), "find_items.json", timeout=300)
    return res.get("boxes") or res.get("items") or []


def observe_and_detect() -> list:
    """臂部对准 + 拍照 + OVD 检测(垃圾类+玩具类标签合并传入)。"""
    run_cli(["eco_lookto"], json.dumps({"req": "ARM"}), "lookto.json")
    run_cli(["eco_captureImages"], json.dumps({"camera_type": "ARM"}), "arm_img.json")
    body = {
        "req": {"rgb_image": "@(arm_img.json).rgb_image.img", "labels": ALL_LABELS, "entry": "OVD"},
        "timeout": 20.0,
    }
    res = run_cli(["eco_detect_objects"], json.dumps(body), "detect.json")
    return res.get("items") or []


def vlm_fallback() -> list:
    """OVD 未命中的兜底:VLM 感知当前视野物品(自由名称,交给词表规则路由)。"""
    run_cli(["eco_captureImages"], json.dumps({"camera_type": "HEAD"}), "head_img.json")
    body = {"image": "@(head_img.json).rgb_image.img", "timeout": 60.0}
    res = run_cli(["eco_vlm_perception"], json.dumps(body), "vlm.json", timeout=90)
    return res if isinstance(res, list) else res.get("items") or []


def pick(name: str, bbox: list) -> None:
    body = {"req": {"name": name, "bbox": bbox}}
    run_cli(["eco_pick"], json.dumps(body), "pick.json", timeout=180)


def place_in(carrier_bbox: list) -> None:
    body = {"req": {"carrier_bbox": carrier_bbox, "direction": ARGS.direction}}
    run_cli(["eco_place_in"], json.dumps(body), "place_in.json", timeout=180)


def handle_one(item: dict, trash_box: dict, box_box: dict, dry_run: bool) -> str:
    """处理单件物品:分类 → 播报 → 抓取 → 投放。返回分类结果。"""
    name = item.get("name", "物品")
    bbox = item.get("bbox") or item.get("bbox_2d")
    if not bbox:
        return "skip"

    target_name, speech_text = route_target(name, ARGS.trash_name, ARGS.box_name)
    speech(speech_text)
    print(f"[分类] {name} -> {target_name}")
    if dry_run:
        return target_name

    carrier = trash_box if target_name == ARGS.trash_name else box_box
    for attempt in range(1, MAX_RETRY + 1):
        try:
            pick(name, bbox)
            place_in(carrier.get("bbox_6d") or carrier.get("bbox") or carrier)
            return target_name
        except RuntimeError as e:
            print(f"[重试 {attempt}/{MAX_RETRY}] {name}: {e}")
            if attempt == MAX_RETRY:
                speech(f"{name}搬运失败,我先把其他的收完")
                raise
    return target_name


def main() -> int:
    global ARGS
    parser = argparse.ArgumentParser(description="八界地面分类归位:垃圾进桶、玩具进筐")
    parser.add_argument("--ws-url", default="")
    parser.add_argument("--area-name", default="")
    parser.add_argument("--trash-name", default="垃圾桶")
    parser.add_argument("--box-name", default="收纳筐")
    parser.add_argument("--direction", default="里")
    parser.add_argument("--max-rounds", type=int, default=10)
    parser.add_argument("--dry-run", action="store_true", help="只分类不动作")
    ARGS = parser.parse_args()

    if ARGS.ws_url:
        os.environ["bajie_sdk_WS_URL"] = ARGS.ws_url

    # 前置检查:电量
    info = run_cli(["eco_robotBattery"], "{}", "battery.json")
    battery = int(info.get("battery", info.get("data", 100)))
    if battery < 30:
        speech("我的电量不足三成,先去充电了,回来再帮你收拾")
        run_cli(["eco_startRecharge"], "{}", "recharge.json")
        return 1

    speech("好的,我先把垃圾桶和收纳筐找出来,再开始收拾地面")
    trash_box = find_container(ARGS.trash_name)
    box_box = find_container(ARGS.box_name)

    run_cli(["eco_prepareRobotPose"], "{}", "prepare.json")
    picked = {"trash": 0, "toy": 0, "fail": 0}
    try:
        for rnd in range(1, ARGS.max_rounds + 1):
            items = search_ground_items(trash_box, box_box, ARGS.area_name)
            if not items:
                # OVD 全局搜索无结果时用 VLM 兜底看一眼,确认真的没有了
                vlm_items = vlm_fallback() if not ARGS.dry_run else []
                if not vlm_items:
                    break
                items = vlm_items
            for item in items:
                try:
                    cls = handle_one(item, trash_box, box_box, ARGS.dry_run)
                    if cls in picked:
                        picked[cls] += 1
                except RuntimeError:
                    picked["fail"] += 1
    finally:
        run_cli(["eco_finishRobotPose"], "{}", "finish.json")

    summary = f"收拾完啦,扔进{ARGS.trash_name} {picked['trash']}件,放进{ARGS.box_name} {picked['toy']}件"
    if picked["fail"]:
        summary += f",有{picked['fail']}件没成功,麻烦你看一下"
    speech(summary)
    print(summary)
    return 0


if __name__ == "__main__":
    sys.exit(main())

Skill 包结构与运行链路:

在这里插入图片描述

使用前的前置条件也只有三条:语义建图完成(地图里有"垃圾桶"和"收纳筐"两个语义区域)、Python 3.10 环境装好 bajie_sdk、VLM API KEY 在 http://<机器人IP>:17890 配置页填好。另外它支持 --dry-run 预演模式,只分类不动手,先把每件东西"会分到哪"打印出来,确认规则没问题再上真机。

七、关于提交:Skill 已经 PR 到官方共享仓库

按官方贡献流程,这个 Skill 已经提交到八界官方的 Skill 共享仓库(ecovacs-bajie/ecovacs-bajie-csdn),走的是标准路径:Fork 官方仓库 → 在自己的 Fork 上提交 → 向官方仓库发起 Pull Request,等维护者审核合并。

提交前还跑了一下仓库自带的 validate-skill-submission.sh 校验脚本:文件夹命名、SKILL.md frontmatter 字段、README 完整性这些都有硬性检查,一次通过。这个流程设计得挺贴心,upload-skill.sh 会自动同步官方 main、建分支、推送,不熟悉 Git 分支操作的开发者也能照着文档走完。

八、一步一个痛点,继续走下去

这个 Skill 本身不大,解决的就是"垃圾和玩具混在地上"这一件事。但我不觉得小——带娃家庭的地面就是每天上演这个场景,能每天省下十分钟弯腰,它就是有用的。

往后我想沿着同样的思路一点点加场景:比如加一个"妈妈说这个不许扔"的白名单,让某些物品永远进收纳筐;比如把每件物品的去向记下来,接一个"我的玩具在哪"的问答;再比如定时巡一圈客厅,不用人喊也能主动收拾。

把一件小事做到极致,家庭机器人的价值就是这么一点点攒出来的。

代码已整理为可复用的 Skill 包:

九、如何使用,你也可以来玩一吧

如果你也想要体验下这个skills,使用很简单,可以直接在web端管理后台,找到OpenClaw的聊天窗口,语言文字添加下这个仓库的技能即可。我的操作如下:
在这里插入图片描述
另外,关于整个动作的完整执行过程,可以在这个聊天窗口里实时看得到。如:

在这里插入图片描述

在这里插入图片描述

视频演示效果:
https://www.douyin.com/user/self?modal_id=7695353746217291017

10月11日

其间发现的一点儿小问题,垃圾桶定位困难 — 语义地图中没有"垃圾桶",视觉搜索时有时找不到。这个也就是说最好一开始建立地图时,这些常见的物品位置摆放好,可以加快查找速度。不过即便垃圾桶是后来放置的没在语义地图里也没关系,最后还是能够找得到,只是会慢一点儿。建议常规使用时,在语义地图中添加"垃圾桶"和"收纳筐"的标注,下次会顺畅很多。

如果调试期间遇到问题也不用担心,可以在后台界面停止所有任务,并操控执行机械臂复位:

在这里插入图片描述

最后,一起来加入八界开源生态吧!

#八界创造者计划 #科沃斯八界 #具身智能

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐