I will build a ROS 2 bridge node that turns natural language commands into structured velocity messages for an existing differential-drive robot. This lets you add high-level intent parsing without touching the lower-level navigation stack or motor controllers. Oxlo.ai fits here because its request-based pricing keeps costs flat even when you stuff detailed kinematic constraints and safety context into every prompt.
What you'll need
- Ubuntu 22.04 with ROS 2 Humble and a running robot stack (turtlesim works fine)
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK installed:
pip install openai
Step 1: Bootstrap the ROS 2 node
We start with a minimal node that listens on /robot_command and has a publisher ready for /cmd_vel. This confirms our topic names match the existing robot stack.
import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist
from std_msgs.msg import String
class LLMBridge(Node):
def __init__(self):
super().__init__('llm_bridge')
self.pub = self.create_publisher(Twist, '/cmd_vel', 10)
self.sub = self.create_subscription(
String, '/robot_command', self.on_command, 10)
self.get_logger().info('Bridge ready. Waiting for /robot_command ...')
def on_command(self, msg: String):
self.get_logger().info(f'Heard: {msg.data}')
# TODO: call LLM and publish motion
stop = Twist()
self.pub.publish(stop)
def main():
rclpy.init()
node = LLMBridge()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
Step 2: Define the motion schema prompt
Robots require deterministic structure, so the LLM must emit strict JSON. I define the schema, velocity limits, and a safety flag in the system prompt. This is the contract between language and motion.
SYSTEM_PROMPT = """You are a motion planner for a differential-drive robot.
The user will give a natural language command.
Respond ONLY with a JSON object matching this exact schema:
{
\"linear_x\": float,
\"angular_z\": float,
\"duration_ms\": int,
\"safe\": bool,
\"reason\": string
}
Rules:
- linear_x is forward speed in m/s. Range: -0.5 to 0.5.
- angular_z is rotation in rad/s. Positive is counter-clockwise. Range: -1.0 to 1.0.
- duration_ms is how long the command should run before stopping.
- If the command is unsafe, ambiguous, or asks for speeds outside the range, set safe to false and zero all velocities.
- Output only the raw JSON object. No markdown, no explanation, no code fences.
"""
Step 3: Connect to Oxlo.ai
I initialize the OpenAI SDK against Oxlo.ai and add a helper that sends the user command plus the system prompt. I use Llama 3.3 70B because it follows structured instructions accurately and starts instantly with no cold starts on Oxlo.ai.
import json
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ["OXLO_API_KEY"],
)
def query_llm(command: str) -> dict:
resp = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": command},
],
temperature=0.1,
max_tokens=256,
)
text = resp.choices[0].message.content.strip()
# Strip accidental markdown fences
if text.startswith("
```"):
text = text.split("\n", 1)[1].rsplit("```
", 1)[0].strip()
return json.loads(text)
Step 4: Parse and publish commands
Now I wire the LLM helper into the ROS callback. When a command arrives, I call Oxlo.ai, validate the JSON fields, and publish a Twist message. If parsing fails, I immediately publish a zero-velocity stop message.
import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist
from std_msgs.msg import String
import json
import os
from openai import OpenAI
SYSTEM_PROMPT = """You are a motion planner for a differential-drive robot.
The user will give a natural language command.
Respond ONLY with a JSON object matching this exact schema:
{
\"linear_x\": float,
\"angular_z\": float,
\"duration_ms\": int,
\"safe\": bool,
\"reason\": string
}
Rules:
- linear_x is forward speed in m/s. Range: -0.5 to 0.5.
- angular_z is rotation in rad/s. Positive is counter-clockwise. Range: -1.0 to 1.0.
- duration_ms is how long the command should run before stopping.
- If the command is unsafe, ambiguous, or asks for speeds outside the range, set safe to false and zero all velocities.
- Output only the raw JSON object. No markdown, no explanation, no code fences.
"""
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ["OXLO_API_KEY"],
)
def query_llm(command: str) -> dict:
resp = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": command},
],
temperature=0.1,
max_tokens=256,
)
text = resp.choices[0].message.content.strip()
if text.startswith("
```"):
text = text.split("\n", 1)[1].rsplit("```
", 1)[0].strip()
return json.loads(text)
class LLMBridge(Node):
def __init__(self):
super().__init__('llm_bridge')
self.pub = self.create_publisher(Twist, '/cmd_vel', 10)
self.sub = self.create_subscription(
String, '/robot_command', self.on_command, 10)
def on_command(self, msg: String):
try:
motion = query_llm(msg.data)
if not motion.get("safe", False):
self.get_logger().warn(f'Unsafe command: {motion["reason"]}')
self.pub.publish(Twist())
return
t = Twist()
t.linear.x = float(motion["linear_x"])
t.angular.z = float(motion["angular_z"])
self.pub.publish(t)
self.get_logger().info(
f'Moving: linear_x={t.linear_x}, angular_z={t.angular.z} '
f'for {motion["duration_ms"]}ms'
)
except Exception as e:
self.get_logger().error(f'LLM or parse error: {e}')
self.pub.publish(Twist())
def main():
rclpy.init()
node = LLMBridge()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
Step 5: Add safety timeouts
Real hardware cannot block indefinitely if the network lags. I wrap the LLM call in a thread with a two-second timeout. If Oxlo.ai does not respond in time, the robot stops and the callback returns safely.
import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist
from std_msgs.msg import String
import json
import os
import threading
from openai import OpenAI
SYSTEM_PROMPT = """You are a motion planner for a differential-drive robot.
The user will give a natural language command.
Respond ONLY with a JSON object matching this exact schema:
{
\"linear_x\": float,
\"angular_z\": float,
\"duration_ms\": int,
\"safe\": bool,
\"reason\": string
}
Rules:
- linear_x is forward speed in m/s. Range: -0.5 to 0.5.
- angular_z is rotation in rad/s. Positive is counter-clockwise. Range: -1.0 to 1.0.
- duration_ms is how long the command should run before stopping.
- If the command is unsafe, ambiguous, or asks for speeds outside the range, set safe to false and zero all velocities.
- Output only the raw JSON object. No markdown, no explanation, no code fences.
"""
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ["OXLO_API_KEY"],
)
def query_llm(command: str, timeout: float = 2.0) -> dict:
result = {"safe": False, "linear_x": 0.0, "angular_z": 0.0, "duration_ms": 0, "reason": "timeout"}
def target():
nonlocal result
try:
resp = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": command},
],
temperature=0.1,
max_tokens=256,
)
text = resp.choices[0].message.content.strip()
if text.startswith("
```"):
text = text.split("\n", 1)[1].rsplit("```
", 1)[0].strip()
result = json.loads(text)
except Exception as e:
result = {"safe": False, "linear_x": 0.0, "angular_z": 0.0, "duration_ms": 0, "reason": str(e)}
t = threading.Thread(target=target)
t.start()
t.join(timeout)
return result
class LLMBridge(Node):
def __init__(self):
super().__init__('llm_bridge')
self.pub = self.create_publisher(Twist, '/cmd_vel', 10)
self.sub = self.create_subscription(
String, '/robot_command', self.on_command, 10)
self.get_logger().info('LLM bridge started with 2s safety timeout')
def on_command(self, msg: String):
motion = query_llm(msg.data)
if not motion.get("safe", False):
self.get_logger().warn(f'Refusing: {motion["reason"]}')
self.pub.publish(Twist())
return
t = Twist()
t.linear.x = float(motion["linear_x"])
t.angular.z = float(motion["angular_z"])
self.pub.publish(t)
self.get_logger().info(
f'Moving: linear_x={t.linear_x:.2f}, angular_z={t.angular_z:.2f} '
f'for {motion["duration_ms"]}ms'
)
def main():
rclpy.init()
node = LLMBridge()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
Run it
Open three terminals. Start the simulator, then the bridge, then send a command.
# Terminal 1
source /opt/ros/humble/setup.bash
ros2 run turtlesim turtlesim_node
# Terminal 2
source /opt/ros/humble/setup.bash
export OXLO_API_KEY=your_key_here
python3 llm_bridge.py
# Terminal 3
source /opt/ros/humble/setup.bash
ros2 topic pub /robot_command std_msgs/msg/String '{data: "move forward slowly for one second"}' --once
You should see output similar to this from the bridge node:
[INFO] [llm_bridge]: LLM bridge started with 2s safety timeout
[INFO] [llm_bridge]: Moving: linear_x=0.20, angular_z=0.00 for 1000ms
Wrap up
Feed lidar or camera data into the prompt as a JSON array so the LLM can avoid obstacles dynamically. If your robot operates on a factory floor with mixed language crews, swap to Qwen 3 32B on Oxlo.ai for reliable multilingual command parsing without changing any of the ROS plumbing.
Top comments (0)