构建智能物理AI：通过Strands Agents、Bedrock AgentCore、Claude 4.5、NVIDIA GR00T与Hugging Face LeRobot实现从边缘到云端协同

1
2

ollama pull qwen3-vl:2b
pip install 'strands-agents[ollama]'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from strands import Agent
from strands.models.ollama import OllamaModel

edge_model = OllamaModel(
    host="http://localhost:11434",
    model_id="qwen3-vl:2b"
)

agent = Agent(
    model=edge_model,
    system_prompt="You are a helpful assistant running on edge hardware."
)
result = agent("Hello!")

1
2
3
4
5
6
7
8
9

def get_camera_frame() -> bytes:
    # 返回当前摄像头帧的示例函数
    with open("camera_frame.jpg", "rb") as f:
        return f.read()

result = agent([
    {"text": "What objects do you see?"},
    {"image": {"source": {"bytes": get_camera_frame()}}}
])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

@tool
def get_battery_level() -> str:
    """Get current battery level percentage and remaining duration."""
    # 返回电池指标的示例函数
    percentage = robot.get_battery_percentage()
    duration = robot.get_battery_duration_minutes()
    return f"Battery level: {percentage}%, approximately {duration} minutes remaining"

agent = Agent(
    model=edge_model,
    tools=[get_battery_level],
    system_prompt="You are a robot assistant. Use available tools to answer questions."
)
result = agent("How long until you need to recharge?")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

@tool
def execute_manipulation(instruction: str) -> str:
    """Execute a manipulation task using your robotics hardware."""
    # 运行VLA模型推理并驱动机器人的示例函数
    while not task_complete:
        observation = robot.get_observation() # 摄像头 + 关节位置
        action = vla.get_action(observation, instruction) # 来自VLA模型的推理
        robot.apply_action(action) # 执行关节运动
    return f"Completed: {instruction}"

robot_agent = Agent(
    model=edge_model,
    tools=[execute_manipulation],
    system_prompt="You control a robotic arm. Use the manipulation tool to complete physical tasks."
)

result = robot_agent("place the apple in the basket.")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from strands import Agent
from strands_robots import Robot

# 创建带摄像头的机器人
robot = Robot(
    tool_name="my_arm",
    robot="so101_follower",
    cameras={
        "front": {"type": "opencv", "index_or_path": "/dev/video0", "fps": 30},
        "wrist": {"type": "opencv", "index_or_path": "/dev/video2", "fps": 30}
    },
    port="/dev/ttyACM0",
    data_config="so100_dualcam"
)

# 创建带有机器人工具的代理
agent = Agent(tools=[robot])

agent("place the apple in the basket")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

from bosdyn.client.robot_command import RobotCommandBuilder, blocking_command, blocking_stand, blocking_sit

@tool
def stand() -> str:
    """Command the robot to stand up."""
    blocking_stand(command_client, timeout_sec=10)
    return "Robot is now standing"

@tool
def sit() -> str:
    """Command the robot to sit down."""
    blocking_sit(command_client, timeout_sec=10)
    return "Robot is now sitting"

@tool
def battery_change_pose(direction: str = "right") -> str:
    """Position robot for battery access by rolling onto its side."""
    cmd = RobotCommandBuilder.battery_change_pose_command(
        dir_hint=1 if direction == "right" else 2
    )
    blocking_command(command_client, cmd, timeout_sec=20)
    return f"Robot positioned for battery access"

spot_agent = Agent(
    model=edge_model,
    tools=[stand, sit, battery_change_pose],
    system_prompt="You control a Boston Dynamics Spot robot."
)

result = spot_agent("I need to inspect your sensors")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

from strands import Agent, tool
from strands.models import BedrockModel
from strands.models.ollama import OllamaModel

# 具有强大推理能力的云代理
cloud_agent = Agent(
    model=BedrockModel(model_id="global.anthropic.claude-sonnet-4-5-20250929-v1:0"),
    system_prompt="Plan tasks step-by-step for edge robots."
)

# 将云代理暴露为工具，以便可以使用代理即工具模式委托给它
@tool
def plan_task(task: str) -> str:
    """Delegate complex planning to cloud-based reasoning."""
    return str(cloud_agent(task))

# 具有本地模型的边缘代理
edge_agent = Agent(
    model=OllamaModel(
        host="http://localhost:11434",
        model_id="qwen3-vl:2b"
    ),
    tools=[plan_task],
    system_prompt="Complete tasks. Consult cloud for complex planning."
)

result = edge_agent("Fetch me a drink")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

@tool
def control_robot_arm(command: str) -> str:
    """Control robotic arm for manipulation tasks."""
    # 调用远程机械臂代理的示例函数
    return str(robot_arm_agent(command))

@tool
def control_mobile_robot(command: str) -> str:
    """Control mobile robot for navigation and transport."""
    # 调用远程移动代理的示例函数
    return str(mobile_robot_agent(command))

warehouse_orchestrator = Agent(
    model=BedrockModel(model_id="global.anthropic.claude-sonnet-4-5-20250929-v1:0"),
    tools=[control_robot_arm, control_mobile_robot],
    system_prompt="You coordinate multiple robots in a warehouse environment."
)

result = warehouse_orchestrator(
    "Coordinate inventory check: scan shelves, retrieve items, and sort"
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from bedrock_agentcore.memory import MemoryClient

memory_client = MemoryClient(region_name="us-east-1")

# 机器人在导航问题后存储观察结果
memory_client.create_event(
    memory_id=FLEET_MEMORY_ID,
    actor_id=robot_id,
    session_id=f"robot-{robot_id}",
    messages=[
        ("Navigation failure in north corridor - low confidence in visual localization. "
         "Location: north_corridor, light_level: high_contrast", "ASSISTANT")
    ]
)