We are building a smart meter audit agent that reads JSON energy data, flags anomalies, and returns actionable efficiency recommendations. It is aimed at facilities engineers who need to spot waste across hundreds of sensors without writing complex rules. Because the agent ingests entire device traces in a single request, Oxlo.ai's flat per-request pricing keeps the cost predictable even when context grows. See https://oxlo.ai/pricing for details.
What you'll need
Python 3.10 or higher, an Oxlo.ai API key from https://portal.oxlo.ai, and the OpenAI SDK.
pip install openai
Step 1: Scaffold the client
Create a file named energy_agent.py and initialize the Oxlo.ai client. I use Llama 3.3 70B here because it handles structured instructions reliably.
from openai import OpenAI
import json
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
def test_connection():
response = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": "You are a concise assistant."},
{"role": "user", "content": "Say 'Connection OK' and nothing else."},
],
)
print(response.choices[0].message.content)
if __name__ == "__main__":
test_connection()
Step 2: Define the system prompt
The prompt constrains the model to output valid JSON and defines the audit rubric. This keeps parsing trivial downstream.
SYSTEM_PROMPT = """You are an energy audit engineer. Analyze the provided smart meter readings and identify anomalies or waste patterns.
Rules:
- Return ONLY a JSON object. No markdown, no explanation outside the JSON.
- The JSON must have keys: "summary" (string), "anomalies" (list of objects with "device_id", "timestamp", "issue", "severity"), and "recommendations" (list of strings).
- Flag any device that consumes more than 20% above its rolling 4-point average.
- Flag continuous overnight baseline load above 15% of peak daily load.
- Severity must be "low", "medium", or "high"."""
Step 3: Generate synthetic smart meter data
Since not everyone has a live BACnet feed, we will synthesize a week of 15-minute interval data for three devices. One device will have a stuck valve that keeps HVAC running overnight.
import random
from datetime import datetime, timedelta
def generate_meter_data():
random.seed(42)
devices = ["hvac_north", "lighting_floor_2", "plug_load_server"]
start = datetime(2024, 1, 1, 0, 0)
intervals = 7 * 24 * 4 # one week, 15-min intervals
readings = []
for i in range(intervals):
ts = start + timedelta(minutes=15 * i)
hour = ts.hour
is_night = hour < 6 or hour > 22
# hvac: high during day, should be low at night, but we inject a fault
if is_night:
hvac = 12.0 if ts.day == 3 else 2.0 # stuck valve on Jan 3
else:
hvac = 8.0 + random.uniform(-1.0, 1.0)
# lighting: on during business hours
lighting = 5.0 if 8 <= hour <= 18 else 0.2
# plug load: fairly steady
plug = 3.5 + random.uniform(-0.2, 0.2)
readings.append({
"timestamp": ts.isoformat(),
"hvac_north": round(hvac, 2),
"lighting_floor_2": round(lighting, 2),
"plug_load_server": round(plug, 2),
})
return readings
Step 4: Wire the analysis call
We serialize the readings to JSON, pack them into the user message, and request the audit. I switch to Kimi K2.6 here because its 131K context window and reasoning capabilities handle long time-series traces without truncation. Oxlo.ai's request-based pricing means the bill does not spike when we add more historical context.
def run_audit(readings):
user_message = (
"Smart meter readings (15-min intervals for one week):\n"
+ json.dumps(readings, indent=2)
+ "\n\nPerform the energy audit and return JSON."
)
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
)
raw = response.choices[0].message.content.strip()
if raw.startswith("
```"):
raw = raw.split("\n", 1)[1].rsplit("```
", 1)[0].strip()
return json.loads(raw)
Run it
This block ties the pieces together, prints the raw JSON report, and pretty-prints the findings.
if __name__ == "__main__":
data = generate_meter_data()
report = run_audit(data)
print("=== Energy Audit Report ===")
print(json.dumps(report, indent=2))
print("\n=== Action Items ===")
for rec in report.get("recommendations", []):
print(f"- {rec}")
Example output:
=== Energy Audit Report ===
{
"summary": "One anomaly detected: hvac_north shows elevated overnight consumption on January 3rd consistent with a stuck valve or scheduling fault.",
"anomalies": [
{
"device_id": "hvac_north",
"timestamp": "2024-01-03T00:00:00",
"issue": "Overnight baseline load 600% above expected, exceeding 15% of daily peak threshold.",
"severity": "high"
}
],
"recommendations": [
"Inspect hvac_north damper or valve actuator on January 3rd.",
"Verify overnight temperature setpoints in the BMS.",
"Add an alert if nighttime HVAC load exceeds 3 kW for more than 30 minutes."
]
}
=== Action Items ===
- Inspect hvac_north damper or valve actuator on January 3rd.
- Verify overnight temperature setpoints in the BMS.
- Add an alert if nighttime HVAC load exceeds 3 kW for more than 30 minutes.
Next steps
Replace the synthetic generator with a real query to your building management system or InfluxDB bucket, then schedule the script via cron to run every morning. Add function calling to the agent so it can automatically create Jira tickets or adjust thermostat setpoints through your BMS API when it flags a high severity anomaly.
Top comments (0)