Robots that understand plain language need more than intent classification. They need to parse ambiguous commands, validate them against physical constraints, and hold a clarifying conversation when context is missing. In this tutorial, I will show you how I built a dialogue controller for a mobile manipulator using Oxlo.ai, where flat per-request pricing keeps costs stable even with long system prompts full of robot schemas and safety policies.
What you'll need
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK:
pip install openai
Step 1: Scaffold the Oxlo.ai client and robot schema
I always start by pinning the API client and writing down the robot's action space. This schema prevents hallucinated commands later. I host this on Oxlo.ai because its request-based pricing keeps costs flat even when the system prompt grows to thousands of tokens with a detailed robot schema. That matters when you are iterating on physical safety policies and do not want token costs to balloon.
import json
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
ROBOT_SCHEMA = {
"actions": ["move", "pick", "place", "scan", "wait"],
"locations": ["kitchen_table", "sofa", "charging_dock", "doorway"],
"objects": ["red_cup", "blue_box", "remote"],
"constraints": {
"max_speed_mps": 0.5,
"forbidden_zones": ["stairs_edge"]
}
}
Step 2: The system prompt and command parser
The system prompt is the contract between the human and the robot. It tells the model to emit JSON only, respect the schema, and flag missing parameters. I use Llama 3.3 70B on Oxlo.ai because it follows structured instructions reliably.
SCHEMA_TEXT = json.dumps(ROBOT_SCHEMA, indent=2)
SYSTEM_PROMPT = f"""You are the dialogue controller for a mobile manipulator.
Your job is to convert the user's natural language command into a structured JSON action.
Available schema: {SCHEMA_TEXT}
Rules:
1. Output valid JSON with keys: action, target, parameter, confidence (0.0 to 1.0), missing_info (list of strings).
2. If a location or object is ambiguous, set missing_info to ask one clarifying question.
3. Never invent actions or objects outside the schema.
4. If the command is unsafe (e.g., move too fast, go near stairs_edge), set safety_flag to true and explain why in safety_reason.
"""
def parse_command(user_text: str) -> dict:
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_text},
],
temperature=0.1,
response_format={"type": "json_object"},
)
raw = response.choices[0].message.content
return json.loads(raw)
Step 3: Safety validation before actuation
Parsing is not enough. I run a second pass to double-check physics and policy constraints before the robot actuates. I use Qwen 3 32B here because it handles multilingual reasoning well, which is useful if your operators mix languages on the floor.
SAFETY_PROMPT = """You are a safety validator for a mobile robot.
Given a proposed JSON action and the robot constraints, reply with JSON containing:
- approved (bool)
- reason (string)
Constraints:
- max_speed: 0.5 m/s
- forbidden_zones: ["stairs_edge"]
Reject any action that violates these or that seems physically impossible."""
def validate_action(action_json: dict) -> dict:
response = client.chat.completions.create(
model="qwen-3-32b",
messages=[
{"role": "system", "content": SAFETY_PROMPT},
{"role": "user", "content": json.dumps(action_json, indent=2)},
],
temperature=0.0,
response_format={"type": "json_object"},
)
return json.loads(response.choices[0].message.content)
Step 4: Build the dialogue state machine
When the parser reports missing info or the safety validator rejects an action, the robot should not freeze. It should ask the user a targeted question and remember the context. This small state machine handles that.
class DialogueManager:
def __init__(self):
self.state = "idle"
self.pending_action = None
self.history = []
def step(self, user_text: str) -> str:
self.history.append({"role": "user", "content": user_text})
if self.state == "idle":
parsed = parse_command(user_text)
self.pending_action = parsed
if parsed.get("missing_info"):
self.state = "clarifying"
question = parsed["missing_info"][0]
self.history.append({"role": "assistant", "content": question})
return question
safety = validate_action(parsed)
if not safety.get("approved"):
self.state = "idle"
self.pending_action = None
reply = f"Safety check failed: {safety.get('reason')}. Please rephrase."
self.history.append({"role": "assistant", "content": reply})
return reply
self.state = "executing"
reply = f"Executing: {json.dumps(parsed)}"
self.history.append({"role": "assistant", "content": reply})
return reply
elif self.state == "clarifying":
clarification = user_text
merged_prompt = (
f"Original parsed action: {json.dumps(self.pending_action)}\n"
f"User clarification: {clarification}\n"
"Produce the updated JSON action."
)
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": merged_prompt},
],
temperature=0.1,
response_format={"type": "json_object"},
)
updated = json.loads(response.choices[0].message.content)
self.pending_action = updated
if updated.get("missing_info"):
question = updated["missing_info"][0]
self.history.append({"role": "assistant", "content": question})
return question
safety = validate_action(updated)
if not safety.get("approved"):
self.state = "idle"
self.pending_action = None
reply = f"Safety check failed: {safety.get('reason')}. Please rephrase."
self.history.append({"role": "assistant", "content": reply})
return reply
self.state = "executing"
reply = f"Executing: {json.dumps(updated)}"
self.history.append({"role": "assistant", "content": reply})
return reply
else:
self.state = "idle"
return "Action complete. Awaiting next command."
Step 5: The main interaction loop
The final script wires the manager to stdin so you can test the full flow interactively.
def main():
print("Robot dialogue controller started. Type 'exit' to quit.")
manager = DialogueManager()
while True:
try:
user_input = input("\nYou: ").strip()
except (EOFError, KeyboardInterrupt):
break
if user_input.lower() in ("exit", "quit"):
break
reply = manager.step(user_input)
print(f"Robot: {reply}")
if __name__ == "__main__":
main()
Run it
Save the full script as robot_controller.py, export your key, and run it. Here is a sample session:
$ export OXLO_API_KEY="sk-oxlo.ai-..."
$ python robot_controller.py
Robot dialogue controller started. Type 'exit' to quit.
You: Go to the kitchen table and pick up the red cup
Robot: Executing: {"action": "pick", "target": "red_cup", "parameter": "kitchen_table", "confidence": 0.95, "missing_info": [], "safety_flag": false}
You: Move fast to the doorway
Robot: Safety check failed: Requested speed exceeds max_speed 0.5 m/s. Please rephrase.
You: Bring me the remote
Robot: Where is the remote located?
You: It is on the sofa
Robot: Executing: {"action": "pick", "target": "remote", "parameter": "sofa", "confidence": 0.92, "missing_info": [], "safety_flag": false}
Next steps
Replace the print statements with a ROS 2 topic publisher so the JSON actions actually drive a physical or simulated robot. Then, add a persistent vector store of past clarifications using Oxlo.ai's embeddings endpoint so the robot learns operator preferences over time.
Top comments (0)