Autonomous vehicle stacks need fast, interpretable reasoning over messy sensor data. In this guide, I will walk you through building a lightweight LLM agent that consumes structured scene descriptions and outputs validated driving decisions. We will run the entire pipeline against Oxlo.ai so you can prototype without managing inference infrastructure.
What you'll need
- An Oxlo.ai API key from https://portal.oxlo.ai
- Python 3.10 or newer
- The OpenAI SDK:
pip install openai
Step 1: Configure the Oxlo.ai client
We start by instantiating the OpenAI-compatible client pointing at Oxlo.ai. I keep the API key in an environment variable so it does not end up in source control.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
)
Step 2: Define the AV system prompt
The system prompt constrains the model to behave like a safety-critical planner. It expects JSON sensor input and must return JSON containing a recommended action, confidence, and rationale. I treat this as a configurable constant so I can iterate quickly.
SYSTEM_PROMPT = """You are an autonomous vehicle decision agent. Your job is to analyze a structured scene snapshot and output a single JSON object with no markdown formatting.
Required JSON schema:
- action: one of [ACCELERATE, BRAKE, TURN_LEFT, TURN_RIGHT, MAINTAIN_SPEED, STOP]
- target_speed_mph: integer, 0 to 65
- confidence: float, 0.0 to 1.0
- rationale: string, max 200 characters
- hazard_detected: boolean
Rules:
1. Always prioritize pedestrian safety.
2. Respect traffic signals.
3. If occlusion is high and a pedestrian may be present, choose STOP or BRAKE.
4. Return only valid JSON. Do not include explanations outside the JSON."""
Step 3: Format sensor telemetry
Real AV pipelines publish object lists from perception. We simulate one tick of fused camera and lidar data, then serialize it into a concise text block for the model. Keeping the prompt compact reduces latency.
import json
def format_scene(
ego_speed_mph: int,
traffic_light: str,
objects: list[dict],
weather: str = "clear"
) -> str:
scene = {
"ego_speed_mph": ego_speed_mph,
"traffic_light": traffic_light,
"weather": weather,
"detected_objects": objects
}
return json.dumps(scene, indent=2)
# Example tick
scene_message = format_scene(
ego_speed_mph=25,
traffic_light="green",
objects=[
{"type": "vehicle", "distance_m": 12, "lane": "same", "speed_mph": 20},
{"type": "pedestrian", "distance_m": 8, "lane": "crosswalk", "status": "walking"}
],
weather="fog"
)
Step 4: Query the model with JSON mode
We send the formatted scene to Oxlo.ai and request a JSON object back. I use llama-3.3-70b because it follows structured instructions reliably and has no cold starts on Oxlo.ai. If you need deeper reasoning for edge cases, swap in kimi-k2.6 or deepseek-v3.2 without changing any other code.
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": scene_message},
],
response_format={"type": "json_object"},
temperature=0.1,
max_tokens=256
)
raw_output = response.choices[0].message.content
decision = json.loads(raw_output)
print(json.dumps(decision, indent=2))
Step 5: Wrap the decision loop
For a real prototype, we need a reusable function that accepts a scene dict and returns a validated decision dict. I also add a small retry guard in case the model returns malformed JSON during early iterations.
def plan(scene_payload: str, max_retries: int = 2) -> dict:
for attempt in range(max_retries):
try:
resp = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": scene_payload},
],
response_format={"type": "json_object"},
temperature=0.1,
max_tokens=256
)
content = resp.choices[0].message.content
return json.loads(content)
except Exception as e:
if attempt == max_retries - 1:
return {
"action": "STOP",
"target_speed_mph": 0,
"confidence": 1.0,
"rationale": f"Planner failed after {max_retries} attempts: {str(e)}",
"hazard_detected": True
}
continue
# Quick sanity check
test = plan(format_scene(
ego_speed_mph=35,
traffic_light="yellow",
objects=[
{"type": "vehicle", "distance_m": 30, "lane": "same", "speed_mph": 35},
{"type": "pedestrian", "distance_m": 4, "lane": "crosswalk", "status": "walking"}
]
))
print(test)
Run it
Here is a complete script that simulates three consecutive perception ticks and prints the planner decisions. Because Oxlo.ai charges per request, not per token, feeding long object lists from lidar point clusters is predictable and cheap. See current rates at https://oxlo.ai/pricing.
if __name__ == "__main__":
ticks = [
format_scene(25, "green", [
{"type": "vehicle", "distance_m": 15, "lane": "same", "speed_mph": 25}
]),
format_scene(25, "red", [
{"type": "vehicle", "distance_m": 5, "lane": "same", "speed_mph": 0}
]),
format_scene(15, "green", [
{"type": "pedestrian", "distance_m": 6, "lane": "crosswalk", "status": "standing"},
{"type": "cyclist", "distance_m": 10, "lane": "right_adjacent", "speed_mph": 8}
], weather="rain")
]
for i, tick in enumerate(ticks, 1):
decision = plan(tick)
print(f"Tick {i}: {decision['action']} - {decision['rationale']}")
Example output:
Tick 1: MAINTAIN_SPEED - Leading vehicle is moving at similar speed, green light, safe following distance.
Tick 2: BRAKE - Red light detected and lead vehicle stopped, decelerating to stop.
Tick 3: BRAKE - Pedestrian standing at crosswalk in rain requires caution, reducing speed.
Wrap up
You now have a working LLM decision layer for AV prototyping. Two concrete next steps: wire this planner into a ROS2 node so it consumes live perception topics, and add a memory buffer so the model sees the previous two ticks for temporal consistency. Both are straightforward because the Oxlo.ai endpoint is a drop-in OpenAI-compatible client, so you keep your existing Python async patterns.
Top comments (0)