We are going to build a lightweight Edge Inference Gateway that runs on a low-power ARM box and batches queries from dozens of local devices into a single structured request. This keeps edge hardware cheap while offloading heavy reasoning to Oxlo.ai, where flat per-request pricing means a long batched prompt costs the same as a short one. The result is a resilient edge node that never hosts a billion-parameter model locally yet answers like one.
What you'll need
- Python 3.10+
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK and FastAPI:
pip install openai fastapi uvicorn - A Linux edge device, Raspberry Pi, or even a laptop simulating one
1. Set up the project and environment
Create a project directory and export your Oxlo.ai API key. I also install the dependencies I will need for the gateway.
mkdir edge_gateway && cd edge_gateway
python3 -m venv venv && source venv/bin/activate
pip install openai fastapi uvicorn
export OXLO_API_KEY="YOUR_OXLO_API_KEY"
2. Add a local SQLite cache for deduplication
Edge networks are flaky. Devices retry questions, so I cache the last hundred Q&A pairs locally to avoid redundant API calls.
import sqlite3, hashlib, time
DB_PATH = "edge_cache.db"
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS cache (
q_hash TEXT PRIMARY KEY,
question TEXT,
answer TEXT,
ts INTEGER
)
""")
conn.commit()
conn.close()
def get_cache(question: str):
conn = sqlite3.connect(DB_PATH)
h = hashlib.sha256(question.encode()).hexdigest()
row = conn.execute(
"SELECT answer FROM cache WHERE q_hash = ?", (h,)
).fetchone()
conn.close()
return row[0] if row else None
def set_cache(question: str, answer: str):
conn = sqlite3.connect(DB_PATH)
h = hashlib.sha256(question.encode()).hexdigest()
conn.execute(
"INSERT OR REPLACE INTO cache (q_hash, question, answer, ts) VALUES (?, ?, ?, ?)",
(h, question, answer, int(time.time()))
)
conn.commit()
conn.close()
init_db()
3. Build the request batch collector
The gateway waits up to two seconds or until it has five pending queries, then merges them into one structured prompt. Batching is the simplest way to scale an edge node to many devices without paying per-token penalties.
import asyncio
from typing import List
pending: asyncio.Queue = asyncio.Queue()
async def collect_batch(max_size: int = 5, max_wait: float = 2.0) -> List[dict]:
batch = []
deadline = asyncio.get_event_loop().time() + max_wait
while len(batch) < max_size:
timeout = deadline - asyncio.get_event_loop().time()
if timeout <= 0:
break
try:
item = await asyncio.wait_for(pending.get(), timeout=max(0.01, timeout))
batch.append(item)
except asyncio.TimeoutError:
break
return batch
4. Define the agent system prompt
I treat the LLM as a multi-tenant edge agent. It receives a JSON list of questions and returns a JSON map of answers. This keeps parsing trivial on the device side.
SYSTEM_PROMPT = """You are an Edge Support Agent running on a gateway device.
You will receive a JSON list of objects, each with an "id" and a "question".
Respond with a single JSON object mapping each "id" to a concise "answer".
Do not add markdown fences or commentary outside the JSON.
Be factual and keep answers under two sentences where possible."""
5. Wire up the Oxlo.ai client
I use the OpenAI SDK pointed at Oxlo.ai. I chose llama-3.3-70b because it handles structured JSON well, and because Oxlo.ai uses flat per-request pricing regardless of prompt length, I can pack a large batch into one call without cost surprises. See https://oxlo.ai/pricing for current plan details.
import os, json
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key=os.environ["OXLO_API_KEY"])
async def process_batch(batch: List[dict]) -> dict:
if not batch:
return {}
payload = json.dumps([{"id": b["req_id"], "question": b["question"]} for b in batch])
def _call():
return client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": payload},
],
)
response = await asyncio.to_thread(_call)
raw = response.choices[0].message.content
answers = json.loads(raw)
return answers
6. Expose the edge gateway with FastAPI
I expose an /ask endpoint that enqueues requests, and a background coroutine that flushes the queue every few seconds. Each request gets a UUID so the caller can retrieve its specific answer from the shared map.
import uuid
from fastapi import FastAPI
app = FastAPI()
@app.post("/ask")
async def ask(question: str):
cached = get_cache(question)
if cached:
return {"answer": cached, "source": "cache"}
req_id = str(uuid.uuid4())
future = asyncio.get_event_loop().create_future()
await pending.put({"req_id": req_id, "question": question, "future": future})
answer = await future
set_cache(question, answer)
return {"answer": answer, "source": "llm"}
async def batch_loop():
while True:
batch = await collect_batch(max_size=5, max_wait=2.0)
if not batch:
await asyncio.sleep(0.1)
continue
try:
answers = await process_batch(batch)
for item in batch:
ans = answers.get(item["req_id"], "error: missing response")
if not item["future"].done():
item["future"].set_result(ans)
except Exception as e:
for item in batch:
if not item["future"].done():
item["future"].set_result(f"error: {e}")
@app.on_event("startup")
async def startup():
asyncio.create_task(batch_loop())
7. Run it and test
Start the server on the edge device, then simulate three local clients arriving at once.
uvicorn main:app --host 0.0.0.0 --port 8000
In another terminal, fire three concurrent curls:
curl -X POST "http://localhost:8000/ask?question=What+is+the+max+operating+temperature+for+the+valve?"
curl -X POST "http://localhost:8000/ask?question=How+do+I+reset+the+pressure+sensor?"
curl -X POST "http://localhost:8000/ask?question=List+the+safety+checkpoints+for+line+three."
Example output from the gateway:
{"answer":"The maximum operating temperature for the valve is 120 degrees Celsius.","source":"llm"}
{"answer":"Hold the reset button for 10 seconds until the LED turns green.","source":"llm"}
{"answer":"1) Verify emergency stop is functional, 2) Check guard interlocks, 3) Confirm lockout tags are in place.","source":"llm"}
If I repeat the first question, it returns instantly from the SQLite cache.
Wrap-up and next steps
The gateway turns a cheap edge box into a front end for a 70B parameter model. Because Oxlo.ai pricing is flat per request, I can stuff ten queries into one call and the cost stays predictable. A natural next step is to add a tiny on-device classifier to decide whether a query even needs the LLM, or to replicate this gateway across factory floors and aggregate logs centrally.
Top comments (0)