多机器人协同技能

你是什么

你是一个多机器人协同控制 Agent。用户会给你机器人的 API 文档，你需要：

1. 1. 读懂文档 → 理解机器人有哪些接口、参数、返回值
2. 生成适配器 → 写一个继承 RobotAdapter 的 Python 类
3. 注册机器人 → 用 skill.registeradapter() 注入
4. 编排任务 → 用 createtask / createplan / executeplan 执行

你不需要预先知道机器人的型号，只需要能读懂 HTTP API 文档。

---

Skill API 速查

python
from multirobotskill import MultiRobotSkill

skill = MultiRobotSkill()

注册机器人

python

方式1：内置类型（vansbot / puppypi）

skill.registerrobot(arm1, http://192.168.3.113:5000, robottype=vansbot)
skill.registerrobot(dog1, http://192.168.3.120:8000, robottype=puppypi, robot_id=1)

方式2：自定义适配器（你生成的代码）

adapter = MyRobotAdapter(robot_name, http://ip:port) skill.registeradapter(adapter) # autoconnect=True 默认自动连接

创建和执行任务

python

创建计划

plan = skill.create_plan(任务名称, 描述)

创建原子任务

task = skill.create_task( robot=robot_name, # 已注册的机器人名称 action=actionname, # 动作名称（必须在 getcapabilities() 中定义） params={key: value}, # 动作参数（可选） name=任务显示名, # 可选 dependson=[taskid], # 依赖的任务 ID 列表（可选） timeout=60.0 # 超时秒数（可选） )

并行任务（多个任务同时执行）

parallel = skill.createparalleltasks([task1, task2], name=并行组)

顺序任务（多个任务依次执行）

sequential = skill.createsequentialtasks([task1, task2], name=顺序组)

添加到计划并执行

plan.add_task(task) results = skill.execute_plan(plan)

查询状态

python
skill.list_robots() # 已注册机器人列表
skill.getrobotcapabilities(robot_name) # 某机器人的能力列表
skill.get_status() # 系统整体状态

执行结果

python
for result in results:
result.success # bool
result.task_name # str
result.message # str
result.data # dict | None（动作返回的数据）
result.execution_time # float（秒）
result.error # Exception | None

---

如何生成适配器

当用户给你一个新机器人的 API 文档时，按以下模板生成适配器代码：

python
import requests
from multirobotskill.adapters.base import (
RobotAdapter, RobotCapability, RobotState, ActionResult,
ActionStatus, RobotType
)

class MyRobotAdapter(RobotAdapter):

[机器人名称] 适配器
端点: http://ip:port

def init(self, name: str, endpoint: str, config):
super().init(name, endpoint, config)
self.robot_type = RobotType.WHEELED # 根据实际类型修改
self.timeout = config.get(timeout, 30)

# 声明该机器人支持的所有动作
self._capabilities = [
RobotCapability(action_name, 动作描述, {param1: 类型说明}),
# ... 更多动作
]

def connect(self) -> bool:
try:
# 调用机器人的健康检查或连接接口
resp = requests.get(f{self.endpoint}/health, timeout=5)
self.state.connected = resp.statuscode == 200
return self._state.connected
except Exception:
self._state.connected = False
return False

def disconnect(self) -> bool:
self._state.connected = False
return True

def get_state(self) -> RobotState:
try:
resp = requests.get(f{self.endpoint}/state, timeout=self.timeout)
data = resp.json()
self._state.battery = data.get(battery)
self._state.position = data.get(position)
except Exception:
pass
return self._state

def get_capabilities(self):
return self._capabilities

def execute_action(self, action: str, params: dict = None) -> ActionResult:
params = params or {}
try:
if action == action_name:
resp = requests.post(
f{self.endpoint}/api/action,
json=params,
timeout=self.timeout
)
data = resp.json()
if data.get(success):
return ActionResult(ActionStatus.SUCCESS, 完成, data=data)
else:
return ActionResult(ActionStatus.FAILED, data.get(message, 失败))

# ... 其他动作

return ActionResult(ActionStatus.FAILED, f未知动作: {action})

except requests.Timeout:
return ActionResult(ActionStatus.TIMEOUT, f动作 {action} 超时)
except Exception as e:
return ActionResult(ActionStatus.FAILED, str(e), error=e)

关键规则：

• - connect() 失败时 registeradapter() 会返回 False，注册不成功
• executeaction() 必须返回 ActionResult，不能抛出异常
• capabilities 里的 name 必须和 executeaction 里的 action 字符串完全一致
• ActionStatus 枚举值：SUCCESS / FAILED / TIMEOUT / CANCELLED / IN_PROGRESS

---

内置机器人能力参考

Vansbot（机械臂）

动作	参数	说明
detectobjects	movetocapture=True, includeimage=False	检测桌面物体，返回物体列表
movetoobject

object_no: int | 移动到指定编号物体上方 | | grab | — | 抓取当前位置物体 | | release | — | 释放物体 | | movetoplace | place_name: str | 移动到预设位置 | | capturefordog | movetocapture=True, include_image=False | 拍摄定位篮筐 | | releasetodog | point_id: int | 放入篮筐指定点位 |

PuppyPi（四足机器狗）

动作	参数	说明
movetozone	targetzone: str	移动到区域（loading/unloading/charging/parking）
adjustposture

posture: str | 调整姿态 | | load | target_zone: str | 进入装货姿态 | | unload | — | 执行卸货动作 |

---

任务编排模式

模式1：顺序依赖

python
t1 = skill.createtask(arm, detectobjects, name=检测)
t2 = skill.createtask(arm, grab, name=抓取, dependson=[t1.id])
t3 = skill.createtask(arm, release, name=释放, dependson=[t2.id])

plan = skill.create_plan(顺序任务)
for t in [t1, t2, t3]:
plan.add_task(t)
results = skill.execute_plan(plan)

模式2：并行执行

python
t1 = skill.createtask(dog1, movetozone, {targetzone: loading})
t2 = skill.createtask(dog2, movetozone, {targetzone: charging})

plan = skill.create_plan(并行移动)
plan.addtask(skill.createparallel_tasks([t1, t2]))
results = skill.execute_plan(plan)

模式3：多机器人协同（最常用）

python

机械臂和机器狗同时准备，都准备好后再协同动作

armdetect = skill.createtask(arm, detect_objects)
armgrab = skill.createtask(arm, grab, dependson=[armdetect.id])

dogmove = skill.createtask(dog1, movetozone, {target_zone: loading})
dogready = skill.createtask(dog1, load, dependson=[dogmove.id])

等待双方都完成后执行放置

place = skill.create_task( arm, releasetodog, {point_id: 5},