We are building an edge telemetry agent that watches a local folder for IoT sensor dumps, formats the payloads, and calls an LLM via Oxlo.ai to flag anomalies and suggest fixes. It helps ops teams running lightweight gateways on Raspberry Pi or industrial PCs who need reasoning without managing local GPU infra. Because Oxlo.ai uses request-based pricing, adding longer diagnostic context does not inflate cost the way token-based billing would.
What you'll need
- Python 3.10 or newer
- The OpenAI SDK:
pip install openai - An Oxlo.ai API key from https://portal.oxlo.ai
- A local directory for mock telemetry files (we will create it)
Step 1: Scaffold the project
Create a working directory and a file named agent.py. Initialize the OpenAI-compatible client pointing to Oxlo.ai and create folders for incoming and processed telemetry.
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")
)
WATCH_DIR = "./edge_telemetry"
ARCHIVE_DIR = "./edge_telemetry_processed"
os.makedirs(WATCH_DIR, exist_ok=True)
os.makedirs(ARCHIVE_DIR, exist_ok=True)
print("Agent scaffold ready. Workspace created.")
Step 2: Generate mock edge telemetry
Real sensors write JSON. We will simulate them with a standalone script so we can test the agent without hardware. It drops randomized temperature, vibration, and voltage readings into the watch folder every eight seconds.
import os
import json
import time
import random
import uuid
WATCH_DIR = "./edge_telemetry"
def spawn_telemetry_file():
payload = {
"device_id": f"sensor-{random.randint(1, 3)}",
"timestamp": time.time(),
"temperature_c": round(random.uniform(55.0, 95.0), 1),
"vibration_hz": round(random.uniform(10.0, 120.0), 1),
"voltage_v": round(random.uniform(11.0, 13.5), 1),
}
path = os.path.join(WATCH_DIR, f"{uuid.uuid4().hex}.json")
with open(path, "w") as f:
json.dump(payload, f)
print(f"generated {path}")
if __name__ == "__main__":
os.makedirs(WATCH_DIR, exist_ok=True)
while True:
spawn_telemetry_file()
time.sleep(8)
Step 3: Watch the folder and build prompts
The agent polls for new JSON files, reads the sensor data, and converts it into plain text for the LLM. I use simple polling instead of inotify so it works the same on Windows, Linux, and macOS.
import os
import glob
import json
import time
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")
)
WATCH_DIR = "./edge_telemetry"
ARCHIVE_DIR = "./edge_telemetry_processed"
os.makedirs(WATCH_DIR, exist_ok=True)
os.makedirs(ARCHIVE_DIR, exist_ok=True)
def get_latest_file():
files = glob.glob(os.path.join(WATCH_DIR, "*.json"))
if not files:
return None
return max(files, key=os.path.getmtime)
def format_prompt(data: dict) -> str:
return (
f"Device: {data['device_id']}\n"
f"Temperature: {data['temperature_c']} C\n"
f"Vibration: {data['vibration_hz']} Hz\n"
f"Voltage: {data['voltage_v']} V"
)
if __name__ == "__main__":
print("Watcher ready. Place JSON files in", WATCH_DIR)
test = {
"device_id": "sensor-1",
"temperature_c": 82.5,
"vibration_hz": 95.0,
"voltage_v": 11.8,
}
print(format_prompt(test))
Step 4: Define the system prompt
This prompt tells the model to return structured JSON with severity, reason, and action. Keeping the output schema strict makes it easy to parse downstream.
import os
import glob
import json
import time
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")
)
WATCH_DIR = "./edge_telemetry"
ARCHIVE_DIR = "./edge_telemetry_processed"
os.makedirs(WATCH_DIR, exist_ok=True)
os.makedirs(ARCHIVE_DIR, exist_ok=True)
SYSTEM_PROMPT = """You are an edge maintenance assistant running on a lightweight gateway.
Analyze the sensor telemetry below and respond with a single JSON object containing exactly these keys:
- severity: one of normal, warning, or critical
- reason: a one-sentence explanation
- action: a concrete remediation step or "none"
Be concise. Do not include markdown formatting or code fences in your output."""
def get_latest_file():
files = glob.glob(os.path.join(WATCH_DIR, "*.json"))
if not files:
return None
return max(files, key=os.path.getmtime)
def format_prompt(data: dict) -> str:
return (
f"Device: {data['device_id']}\n"
f"Temperature: {data['temperature_c']} C\n"
f"Vibration: {data['vibration_hz']} Hz\n"
f"Voltage: {data['voltage_v']} V"
)
if __name__ == "__main__":
print("System prompt loaded.")
print(SYSTEM_PROMPT[:60] + "...")
Step 5: Wire the Oxlo.ai inference call
Add the analysis function that sends the telemetry to Oxlo.ai. I am using llama-3.3-70b because it handles structured instructions reliably and runs with no cold starts on Oxlo.ai.
import os
import glob
import json
import time
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")
)
WATCH_DIR = "./edge_telemetry"
ARCHIVE_DIR = "./edge_telemetry_processed"
os.makedirs(WATCH_DIR, exist_ok=True)
os.makedirs(ARCHIVE_DIR, exist_ok=True)
SYSTEM_PROMPT = """You are an edge maintenance assistant running on a lightweight gateway.
Analyze the sensor telemetry below and respond with a single JSON object containing exactly these keys:
- severity: one of normal, warning, or critical
- reason: a one-sentence explanation
- action: a concrete remediation step or "none"
Be concise. Do not include markdown formatting or code fences in your output."""
def get_latest_file():
files = glob.glob(os.path.join(WATCH_DIR, "*.json"))
if not files:
return None
return max(files, key=os.path.getmtime)
def format_prompt(data: dict) -> str:
return (
f"Device: {data['device_id']}\n"
f"Temperature: {data['temperature_c']} C\n"
f"Vibration: {data['vibration_hz']} Hz\n"
f"Voltage: {data['voltage_v']} V"
)
def analyze_telemetry(file_path: str):
with open(file_path, "r") as f:
data = json.load(f)
user_message = format_prompt(data)
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
temperature=0.2,
max_tokens=150,
)
raw = response.choices[0].message.content.strip()
print(f"Result for {os.path.basename(file_path)}:")
print(raw)
print()
dest = os.path.join(ARCHIVE_DIR, os.path.basename(file_path))
os.rename(file_path, dest)
if __name__ == "__main__":
print("Inference pipeline ready.")
Step 6: Run the continuous edge loop
Combine everything in a main loop that processes files as they arrive and archives them. The five-second sleep keeps CPU usage negligible on a Raspberry Pi.
import os
import glob
import json
import time
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")
)
WATCH_DIR = "./edge_telemetry"
ARCHIVE_DIR = "./edge_telemetry_processed"
os.makedirs(WATCH_DIR, exist_ok=True)
os.makedirs(ARCHIVE_DIR, exist_ok=True)
SYSTEM_PROMPT = """You are an edge maintenance assistant running on a lightweight gateway.
Analyze the sensor telemetry below and respond with a single JSON object containing exactly these keys:
- severity: one of normal, warning, or critical
- reason: a one-sentence explanation
- action: a concrete remediation step or "none"
Be concise. Do not include markdown formatting or code fences in your output."""
def get_latest_file():
files = glob.glob(os.path.join(WATCH_DIR, "*.json"))
if not files:
return None
return max(files, key=os.path.getmtime)
def format_prompt(data: dict) -> str:
return (
f"Device: {data['device_id']}\n"
f"Temperature: {data['temperature_c']} C\n"
f"Vibration: {data['vibration_hz']} Hz\n"
f"Voltage: {data['voltage_v']} V"
)
def analyze_telemetry(file_path: str):
with open(file_path, "r") as f:
data = json.load(f)
user_message = format_prompt(data)
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
temperature=0.2,
max_tokens=150,
)
raw = response.choices[0].message.content.strip()
print(f"Result for {os.path.basename(file_path)}:")
print(raw)
print()
dest = os.path.join(ARCHIVE_DIR, os.path.basename(file_path))
os.rename(file_path, dest)
def main():
print("Edge AI agent starting. Watching", WATCH_DIR)
while True:
target = get_latest_file()
if target:
try:
analyze_telemetry(target)
except Exception as e:
print("Inference error:", e)
else:
print("No new telemetry. Sleeping...")
time.sleep(5)
if __name__ == "__main__":
main()
Run it
Open two terminal windows. In the first, start the mock sensor to feed data into the watch folder. In the second, run the agent.
# Terminal 1
export OXLO_API_KEY="sk-..."
python mock_sensor.py
# Terminal 2
python agent.py
With the mock sensor running, you should see output similar to this:
Edge AI agent starting. Watching ./edge_telemetry
No new telemetry. Sleeping...
Result for a1b2c3d4.json:
{"severity": "warning", "reason": "Temperature is elevated at 87.3 C.", "action": "Check cooling fan and clear dust from vents."}
Result for e5f67g8h.json:
{"severity": "critical", "reason": "Voltage dropped to 11.1 V and vibration peaked at 118 Hz.", "action": "Inspect power supply and motor bearings immediately."}
Next steps
Replace the polling loop with an MQTT subscriber or inotify watcher so the agent triggers instantly when real sensors publish data. You can also add function calling to the Oxlo.ai request so the agent POSTs alerts to a local webhook or restarts edge services automatically when severity is critical.
If your deployment grows, batch multiple telemetry files into a single request to minimize API calls, or upgrade to an Oxlo.ai Premium plan for priority queueing. See https://oxlo.ai/pricing for details.
Top comments (0)