Engineering LLMs for Autonomous Vehicle Applications

We are building a closed-loop driving decision agent that consumes fused sensor descriptions and outputs structured control commands with safety reasoning. This tutorial targets engineers who are prototyping LLM-based planners for autonomous vehicle stacks and need deterministic JSON output without managing token cost scaling on long scene histories. We will wire everything to Oxlo.ai so you can iterate against a live model in minutes.

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: Instantiate the Oxlo.ai client

I keep the client initialization in its own module so I can swap models later without touching business logic. Oxlo.ai exposes an OpenAI-compatible endpoint, so the import stays standard.

from openai import OpenAI

client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")

Step 2: Lock down the system prompt

The system prompt is the safety contract. It forces the model to emit only JSON with four required fields and warns against hallucinating objects not present in the scene input.

SYSTEM_PROMPT = """You are an autonomous vehicle motion planner.
Your input is a JSON object describing the current fused perception scene.
You must output a single JSON object with exactly these keys:
- recommended_action: one of MAINTAIN_SPEED, ACCELERATE, BRAKE, STOP, CHANGE_LANE_LEFT, CHANGE_LANE_RIGHT
- target_speed_mps: float, the desired ego speed in meters per second
- reasoning: string, max 200 characters, explain the safety-critical observation that drove the decision
- risk_level: one of low, medium, high, critical

Rules:
1. Never invent obstacles or traffic actors not listed in the scene.
2. If a pedestrian is within 5 meters, recommended_action must be STOP or BRAKE.
3. Respect traffic signals exactly as reported.
4. Respond with raw JSON only, no markdown fences."""

Step 3: Mock the fused perception stream

In production this arrives from a perception stack over DDS or ROS 2. Here we hardcode three timesteps so the tutorial is fully reproducible without hardware.

import json

def generate_scene(t: int) -> dict:
    scenes = [
        {
            "timestamp": 0,
            "ego_speed_mps": 13.4,
            "lane_status": "free",
            "lead_vehicle": {"distance_m": 30, "speed_mps": 13.4},
            "traffic_light": "green",
            "pedestrians": [],
            "weather": "clear"
        },
        {
            "timestamp": 1,
            "ego_speed_mps": 13.4,
            "lane_status": "free",
            "lead_vehicle": {"distance_m": 15, "speed_mps": 8.0},
            "traffic_light": "green",
            "pedestrians": [],
            "weather": "clear"
        },
        {
            "timestamp": 2,
            "ego_speed_mps": 10.0,
            "lane_status": "free",
            "lead_vehicle": {"distance_m": 8, "speed_mps": 0.0},
            "traffic_light": "red",
            "pedestrians": [{"distance_m": 4.5, "position": "crosswalk"}],
            "weather": "clear"
        }
    ]
    return scenes[t]

Step 4: Enforce structured output with JSON mode

We call Llama 3.3 70B through Oxlo.ai with response_format set to json_object. This removes regex cleanup and guarantees valid JSON for downstream controllers.

def plan(scene: dict, history: list = None) -> tuple[dict, list]:
    if history is None:
        history = [
            {"role": "system", "content": SYSTEM_PROMPT},
        ]
    history.append({"role": "user", "content": json.dumps(scene)})

    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=history,
        response_format={"type": "json_object"},
        temperature=0.2,
    )

    decision_text = response.choices[0].message.content
    history.append({"role": "assistant", "content": decision_text})
    return json.loads(decision_text), history

Step 5: Maintain multi-turn memory across timesteps

Autonomous driving is sequential. By reusing the same message list, the model sees its prior decisions and avoids oscillating between conflicting commands.

def run_simulation(steps: int = 3):
    history = [{"role": "system", "content": SYSTEM_PROMPT}]
    for t in range(steps):
        scene = generate_scene(t)
        decision, history = plan(scene, history)
        print(f"Step {t}: {json.dumps(decision, indent=2)}")

Run it

Execute the simulation from the command line. The agent should produce a coherent chain of decisions that respects the red light and pedestrian in the final step.

if __name__ == "__main__":
    run_simulation()

Example output:

Step 0: {
  "recommended_action": "MAINTAIN_SPEED",
  "target_speed_mps": 13.4,
  "reasoning": "Lead vehicle 30m ahead at matching speed. Green light. No pedestrians.",
  "risk_level": "low"
}
Step 1: {
  "recommended_action": "BRAKE",
  "target_speed_mps": 8.0,
  "reasoning": "Lead vehicle decelerating, distance closing to 15m. Prepare to match speed.",
  "risk_level": "medium"
}
Step 2: {
  "recommended_action": "STOP",
  "target_speed_mps": 0.0,
  "reasoning": "Red light and pedestrian 4.5m away in crosswalk. Full stop required.",
  "risk_level": "critical"
}

Next steps

Replace the mock generate_scene function with a live ROS 2 subscriber that ingests autoware_auto_perception_msgs or equivalent, and feed the output directly into an Actuation node. If you plan to log hours of scene history for training or diagnostics, Oxlo.ai request-based pricing keeps inference costs flat no matter how much context you append. See https://oxlo.ai/pricing for plan details.