We are building a natural-language robot task planner that turns spoken commands into structured pick-and-place action sequences. This gives robotics developers a fast way to prototype behavior logic without maintaining brittle state machines for every new task.
What you'll need
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK installed:
pip install openai
Step 1: Initialize the Oxlo.ai client
I start by setting up the OpenAI-compatible client pointing at Oxlo.ai and verify the connection with a quick sanity check.
from openai import OpenAI
import os
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY"),
)
# Quick connectivity test
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[{"role": "user", "content": "Say OK"}],
)
print(response.choices[0].message.content)Step 2: Define the robot task planner system prompt
The system prompt constrains the model to emit only structured JSON plans using a fixed action vocabulary. I keep it separate so it is easy to iterate without touching business logic.
SYSTEM_PROMPT = '''You are a robot task planner for a single manipulator arm in a warehouse cell.
The arm supports four actions: move_to, grip, release, and wait.
Given a natural language command and a JSON world state, output a single JSON object with a top-level key "plan".
The "plan" value must be a list of steps, where each step is an object with keys "action" and "target".
Valid targets are object IDs present in the world state. Do not output any text outside the JSON.'''Step 3: Generate structured action plans
Now I write a function that packages the user command and current world state into a prompt and calls Oxlo.ai. I use qwen-3-32b because it handles structured agent workflows reliably.
import json
def generate_plan(command: str, world_state: dict) -> dict:
user_message = f"Command: {command}\nWorld state: {json.dumps(world_state)}"
response = client.chat.completions.create(
model="qwen-3-32b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
)
content = response.choices[0].message.content.strip()
# Remove markdown fences if the model emits them
if content.startswith("
```"):
content = content.split("```
")[1].replace("json", "").strip()
return json.loads(content)Step 4: Validate against world state and check safety
A plan is useless if it references missing objects or unsafe moves. I validate targets against the world state, then run a second pass through kimi-k2.6 to catch physical issues like collisions or dropping objects mid-air.
def validate_plan(plan: dict, world_state: dict) -> bool:
valid_targets = set(world_state.keys()) | {"none", "home"}
for step in plan.get("plan", []):
target = step.get("target", "")
if target not in valid_targets:
print(f"Validation failed: unknown target '{target}'")
return False
return True
def safety_check(command: str, plan: dict) -> str:
safety_prompt = (
"Review the following robot plan for physical safety. "
"Look for collisions, dropping objects without support, or unreachable motions. "
"Reply with ONLY the word SAFE, or UNSAFE followed by a one-line reason.\n\n"
f"Command: {command}\nPlan: {json.dumps(plan)}"
)
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": "You are a cautious robotics safety engineer."},
{"role": "user", "content": safety_prompt},
],
)
return response.choices[0].message.content.strip()Step 5: Assemble the RobotTaskAgent
I wrap the pipeline into a small class that plans, validates, checks safety, and simulates execution by printing each step. In a real deployment, these print statements would be ROS2 action client calls or motor driver commands.
class RobotTaskAgent:
def __init__(self, client):
self.client = client
def run(self, command: str, world_state: dict):
print(f"\nCommand received: {command}")
# Plan
plan = generate_plan(command, world_state)
print(f"Generated plan: {json.dumps(plan, indent=2)}")
# Validate
if not validate_plan(plan, world_state):
return {"status": "error", "detail": "validation_failed"}
# Safety
safety = safety_check(command, plan)
print(f"Safety review: {safety}")
if not safety.startswith("SAFE"):
return {"status": "error", "detail": safety}
# Execute (simulated)
for step in plan["plan"]:
print(f"Executing: {step['action']} -> {step['target']}")
return {"status": "success", "plan": plan}Run it
Here is the end-to-end test. I define a simple world with a red block and a blue bin, then ask the agent to move the block.
if __name__ == "__main__":
agent = RobotTaskAgent(client)
world_state = {
"red_block": {"x": 0.4, "y": 0.2, "z": 0.0},
"blue_bin": {"x": 0.8, "y": 0.6, "z": 0.0},
}
result = agent.run(
"Pick up the red block and place it in the blue bin",
world_state
)
print(f"\nFinal result: {result}")Example output:
Command received: Pick up the red block and place it in the blue bin
Generated plan: {
"plan": [
{"action": "move_to", "target": "red_block"},
{"action": "grip", "target": "red_block"},
{"action": "move_to", "target": "blue_bin"},
{"action": "release", "target": "blue_bin"},
{"action": "move_to", "target": "home"}
]
}
Safety review: SAFE
Executing: move_to -> red_block
Executing: grip -> red_block
Executing: move_to -> blue_bin
Executing: release -> blue_bin
Executing: move_to -> home
Final result: {'status': 'success', 'plan': ...}Wrap-up and next steps
This pipeline gives you a working natural-language interface for a manipulator in under a hundred lines of Python. Because Oxlo.ai uses request-based pricing, you can iterate on long system prompts and multi-turn safety reviews without the cost scaling with token count. See https://oxlo.ai/pricing for details.
Two concrete next steps: swap the simulated execution layer for a ROS2 publisher so the plans drive a real robot, or add a vision module using kimi-vl-a3b on Oxlo.ai to generate the world state from camera frames instead of hard-coded JSON.
Top comments (0)