Introduction to LLM for Fog and Edge Computing: A Beginner's Guide

#learnai #oxlo #ai

We are going to build a lightweight orchestration agent that reads telemetry from edge nodes and decides whether to offload jobs, throttle services, or alert a human operator. This kind of fog-local reasoning keeps decisions off the WAN, which matters when your devices live behind spotty connectivity. I will use Oxlo.ai as the inference backend because its request-based pricing stays flat even when device telemetry makes the prompts long.

What you'll need

Python 3.10 or newer
An Oxlo.ai API key from https://portal.oxlo.ai
The OpenAI SDK. Install it with pip install openai

Step 1: Define the edge telemetry schema

First, I define a small dataclass to represent CPU, memory, latency, and queued jobs from a single node. I also create a helper to serialize it to JSON so the LLM receives clean structured input.

import json
from dataclasses import dataclass
from typing import List

@dataclass
class EdgeNode:
    node_id: str
    cpu_percent: float
    memory_percent: float
    latency_ms: float
    queued_jobs: int

def node_to_json(node: EdgeNode) -> str:
    return json.dumps({
        "node_id": node.node_id,
        "cpu_percent": node.cpu_percent,
        "memory_percent": node.memory_percent,
        "latency_ms": node.latency_ms,
        "queued_jobs": node.queued_jobs
    })

# Our sample fleet: two edge nodes and one fog gateway
fleet = [
    EdgeNode("edge-01", 92.5, 88.0, 45, 12),
    EdgeNode("edge-02", 34.0, 42.0, 12, 2),
    EdgeNode("fog-01", 55.0, 60.0, 8, 4),
    EdgeNode("edge-03", 78.0, 45.0, 110, 15),
]

Step 2: Write the system prompt

The system prompt is the policy layer. It tells the model how to map telemetry thresholds to concrete actions and forces JSON output so we can parse it programmatically.

SYSTEM_PROMPT = """You are an edge resource orchestrator. You receive telemetry from a fleet of edge and fog nodes. Your job is to decide, for each overloaded edge node, whether to:
1. offload_jobs_to_fog - Move queued jobs to the nearest fog node.
2. throttle_non_critical - Reduce non-critical service CPU limits locally.
3. alert_operator - Escalate because hardware is saturated.

Rules:
- If cpu_percent > 85 and memory_percent > 80, recommend offload_jobs_to_fog.
- If only cpu_percent > 85, recommend throttle_non_critical.
- If latency_ms > 100 and queued_jobs > 10, recommend alert_operator.
- Return ONLY a JSON object with keys: decision (string), target_node (string), reason (string).

Example output:
{"decision": "offload_jobs_to_fog", "target_node": "edge-01", "reason": "CPU and memory both above thresholds"}"""

Step 3: Build the orchestrator client

Now I wire the prompt to Oxlo.ai. I initialize the OpenAI-compatible client with the Oxlo.ai base URL and call Llama 3.3 70B, a solid general-purpose model for structured reasoning. I keep temperature low so the policy stays deterministic.

from openai import OpenAI

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

def orchestrate(node_json: str) -> dict:
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": node_json},
        ],
        temperature=0.1,
        response_format={"type": "json_object"},
    )
    return json.loads(response.choices[0].message.content)

Step 4: Simulate multi-node fog input

Next I loop over the fleet, serialize each node, and send it through the orchestrator. In production this loop would run inside a sidecar on your fog gateway polling local MQTT or HTTP telemetry endpoints.

def run_fleet_check(nodes: List[EdgeNode]) -> List[dict]:
    results = []
    for node in nodes:
        payload = node_to_json(node)
        decision = orchestrate(payload)
        results.append({
            "node_id": node.node_id,
            "decision": decision.get("decision"),
            "reason": decision.get("reason")
        })
    return results

Step 5: Parse and act on decisions

Finally, I add a small actuator function that translates the LLM decision into a shell-style command. For this tutorial it prints the action, but you can swap these prints for real Docker or Kubectl calls.

def act_on_decision(result: dict):
    node = result["node_id"]
    decision = result["decision"]
    if decision == "offload_jobs_to_fog":
        print(f"[{node}] Offloading jobs to nearest fog gateway.")
    elif decision == "throttle_non_critical":
        print(f"[{node}] Throttling non-critical containers.")
    elif decision == "alert_operator":
        print(f"[{node}] ALERT: Operator intervention required.")
    else:
        print(f"[{node}] No action taken.")

Run it

Call the fleet check from the main guard and watch the agent classify each node.

if __name__ == "__main__":
    results = run_fleet_check(fleet)
    for r in results:
        act_on_decision(r)

Example output:

[edge-01] Offloading jobs to nearest fog gateway.
[edge-02] No action taken.
[fog-01] No action taken.
[edge-03] ALERT: Operator intervention required.

Next steps

Deploy this loop as a sidecar container on your fog gateway and point it at Oxlo.ai. Because Oxlo.ai bills per request rather than per token, long telemetry dumps from noisy edge sensors will not inflate your inference bill the way token-based providers do. You can view flat pricing at https://oxlo.ai/pricing.

To harden the agent, replace the print statements in act_on_decision with real control plane calls, such as Docker Engine API container updates or Kubernetes kubectl patch commands to adjust pod resource limits at the edge.

DEV Community