Introduction to Edge AI with LLMs

#learnai #oxlo #ai

We are going to build a lightweight log analyzer that runs on a resource-constrained edge device and offloads LLM inference to Oxlo.ai. This helps engineers who need to monitor remote sensors or gateways without shipping multi-gigabyte models to the edge. The entire pipeline fits in a single Python file and uses Oxlo.ai's OpenAI-compatible API.

What you'll need

Python 3.10 or newer
An Oxlo.ai API key from https://portal.oxlo.ai
The OpenAI SDK: pip install openai
A sample log file, or you can generate fake logs with the script in Step 2.

Step 1: Configure the Oxlo.ai client

First, I import the OpenAI SDK and point it at Oxlo.ai. A quick connectivity test confirms the API key and base URL are correct.

from openai import OpenAI
import os

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

# Quick connectivity test
response = client.chat.completions.create(
    model="llama-3.3-70b",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Say OK"},
    ],
    max_tokens=10
)
print(response.choices[0].message.content)

Step 2: Simulate and chunk edge logs

Edge devices produce continuous text logs. To stay within memory limits, I read the file in fixed line chunks rather than loading everything into RAM. I also generate a fake log file so you can run this immediately.

def chunk_logs(file_path, chunk_size=20):
    """Yield chunks of log lines from a file."""
    with open(file_path, "r") as f:
        chunk = []
        for line in f:
            chunk.append(line.strip())
            if len(chunk) == chunk_size:
                yield "\n".join(chunk)
                chunk = []
        if chunk:
            yield "\n".join(chunk)

# Generate a fake log file for testing
sample_lines = [
    "2024-05-20 14:01:23 sensor_temp=45C normal",
    "2024-05-20 14:02:10 sensor_temp=46C normal",
    "2024-05-20 14:03:44 sensor_temp=89C warning",
    "2024-05-20 14:04:01 sensor_temp=90C critical",
    "2024-05-20 14:05:12 sensor_temp=47C normal",
    "2024-05-20 14:06:00 sensor_temp=48C normal",
    "2024-05-20 14:07:15 sensor_temp=49C normal",
    "2024-05-20 14:08:30 sensor_temp=91C critical",
    "2024-05-20 14:09:00 sensor_temp=48C normal",
    "2024-05-20 14:10:00 sensor_temp=47C normal",
    "2024-05-20 14:11:00 sensor_temp=46C normal",
    "2024-05-20 14:12:00 sensor_temp=45C normal",
    "2024-05-20 14:13:00 sensor_temp=46C normal",
    "2024-05-20 14:14:00 sensor_temp=47C normal",
    "2024-05-20 14:15:00 sensor_temp=46C normal",
]

with open("edge_device.log", "w") as f:
    f.write("\n".join(sample_lines))
print("Created edge_device.log with 15 lines")

Step 3: Define the analyzer system prompt

I keep the system prompt strict so the model returns only structured findings and avoids unnecessary prose. This reduces response size, which matters on edge networks with limited bandwidth.

SYSTEM_PROMPT = """You are an edge AI log analyzer. Your job is to inspect small batches of device logs and report anomalies.

Rules:
- List only anomalous entries.
- For each anomaly, give a one-line reason.
- If no anomalies are found, reply exactly: NO_ANOMALIES.
- Do not add greetings, summaries, or markdown headers."""

Step 4: Build the analysis pipeline

Now I wire the chunks into Oxlo.ai. Because Oxlo.ai uses request-based pricing, I can send reasonably large log chunks without input-token costs ballooning. See https://oxlo.ai/pricing for details. This is useful when edge devices batch minutes or hours of logs into a single call.

def analyze_chunk(chunk: str, model: str = "llama-3.3-70b"):
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Analyze these edge logs:\n\n{chunk}"},
        ],
        temperature=0.1,
        max_tokens=256
    )
    return response.choices[0].message.content.strip()

# Test on the first chunk
for chunk in chunk_logs("edge_device.log", chunk_size=5):
    result = analyze_chunk(chunk)
    print(result)
    break

Step 5: Run continuous edge monitoring

Finally, I wrap the pipeline in a simple loop that processes the whole log file and prints structured results. On a real edge device, you could trigger this via cron or a lightweight systemd timer.

def run_edge_analysis(log_path: str):
    print(f"Scanning {log_path}...")
    for i, chunk in enumerate(chunk_logs(log_path, chunk_size=5), 1):
        result = analyze_chunk(chunk)
        if result != "NO_ANOMALIES":
            print(f"Chunk {i}: {result}")
        else:
            print(f"Chunk {i}: clean")
    print("Scan complete.")

if __name__ == "__main__":
    run_edge_analysis("edge_device.log")

Run it

Save everything in a single file named edge_analyzer.py, set your API key, and execute it. You should see the pipeline identify the temperature anomalies in the first two chunks and mark the third as clean.

export OXLO_API_KEY="sk-oxlo.ai-..."
python edge_analyzer.py

Expected output:

Scanning edge_device.log...
Chunk 1: 2024-05-20 14:03:44 sensor_temp=89C warning - Temperature spike detected.
2024-05-20 14:04:01 sensor_temp=90C critical - Critical temperature threshold exceeded.
Chunk 2: 2024-05-20 14:08:30 sensor_temp=91C critical - Critical temperature threshold exceeded.
Chunk 3: clean
Scan complete.

Next steps

Swap in a reasoning model like deepseek-r1-671b or qwen-3-32b if you want the analyzer to explain root causes rather than just flag lines. You could also replace the local file with a lightweight MQTT or HTTP listener so the script acts as a true edge gateway, receiving logs from sensors in real time.

DEV Community