The $80 Brain: A Billion Tiny AI Agents Are About to Run on Everything You Own

# edge_agent.py — full agentic loop running entirely on device
import json, subprocess, sqlite3, datetime
from typing import Any
 
OLLAMA_MODEL = "phi4-mini"  # or qwen2.5:3b, gemma3:2b
 
def call_local_llm(messages: list[dict], tools: list[dict] | None = None) -> dict:
    """Call ollama running on localhost — no network dependency."""
    payload = {"model": OLLAMA_MODEL, "messages": messages, "stream": False}
    if tools:
        payload["tools"] = tools
    result = subprocess.run(
        ["ollama", "run", "--format", "json", OLLAMA_MODEL],
        input=json.dumps(payload), capture_output=True, text=True, timeout=30
    )
    # Use ollama REST API for cleaner tool_call support
    import urllib.request
    req = urllib.request.Request(
        "http://localhost:11434/api/chat",
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json"},
        method="POST"
    )
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.load(r)
 
# --- Tool definitions ---
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "read_sensor",
            "description": "Read a sensor value by sensor ID. Returns current float reading.",
            "parameters": {
                "type": "object",
                "properties": {
                    "sensor_id": {"type": "string", "description": "Sensor identifier, e.g. 'temp_01'"}
                },
                "required": ["sensor_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "log_event",
            "description": "Log an event to the local database with severity and message.",
            "parameters": {
                "type": "object",
                "properties": {
                    "severity": {"type": "string", "enum": ["info", "warning", "critical"]},
                    "message": {"type": "string"}
                },
                "required": ["severity", "message"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "trigger_action",
            "description": "Trigger a physical action: fan_on, fan_off, alert_led, shutdown.",
            "parameters": {
                "type": "object",
                "properties": {
                    "action": {"type": "string", "enum": ["fan_on", "fan_off", "alert_led", "shutdown"]}
                },
                "required": ["action"]
            }
        }
    }
]
 
# --- Tool implementations ---
def read_sensor(sensor_id: str) -> float:
    """Simulated sensor read — replace with GPIO/I2C/MQTT call."""
    import random
    mock_values = {"temp_01": 78.5 + random.uniform(-5, 15), "humidity_01": 45.0}
    return mock_values.get(sensor_id, 0.0)
 
def log_event(severity: str, message: str) -> str:
    conn = sqlite3.connect("edge_agent.db")
    conn.execute("CREATE TABLE IF NOT EXISTS events (ts TEXT, severity TEXT, message TEXT)")
    conn.execute("INSERT INTO events VALUES (?, ?, ?)",
                 (datetime.datetime.now().isoformat(), severity, message))
    conn.commit(); conn.close()
    print(f"[{severity.upper()}] {message}")
    return "logged"
 
def trigger_action(action: str) -> str:
    print(f">>> ACTION: {action}")
    # Replace with actual GPIO: import RPi.GPIO as GPIO ...
    return f"{action} executed"
 
TOOL_DISPATCH = {"read_sensor": read_sensor, "log_event": log_event, "trigger_action": trigger_action}
 
def execute_tool(name: str, args: dict) -> Any:
    if name not in TOOL_DISPATCH:
        return f"Unknown tool: {name}"
    return TOOL_DISPATCH[name](**args)
 
# --- Agentic loop ---
def run_agent(task: str, max_steps: int = 6) -> str:
    messages = [
        {"role": "system", "content": (
            "You are an autonomous edge AI agent managing industrial sensors. "
            "Use tools to observe the environment, reason about readings, and take action. "
            "Always read sensors before acting. Log events when anomalies are detected."
        )},
        {"role": "user", "content": task}
    ]
 
    for step in range(max_steps):
        response = call_local_llm(messages, tools=TOOLS)
        message = response.get("message", {})
        tool_calls = message.get("tool_calls", [])
 
        if not tool_calls:
            # Agent reached a conclusion
            return message.get("content", "Task complete.")
 
        # Execute all tool calls and collect observations
        messages.append({"role": "assistant", "content": None, "tool_calls": tool_calls})
        for tc in tool_calls:
            fn = tc["function"]
            result = execute_tool(fn["name"], fn.get("arguments", {}))
            messages.append({
                "role": "tool",
                "tool_call_id": tc["id"],
                "content": str(result)
            })
 
    return "Max steps reached."
 
if __name__ == "__main__":
    print(run_agent("Check all sensors, identify any anomalies, and take appropriate action."))

# Install ollama and pull the model (one-time setup)
curl -fsSL https://ollama.ai/install.sh | sh
ollama pull phi4-mini          # 3.8B — fits in 4GB RAM
# ollama pull qwen2.5:3b       # Alternative: Qwen 2.5 3B
# ollama pull gemma3:2b        # Alternative: Gemma 3 2B
 
# Run the edge agent
pip install --upgrade pip
python edge_agent.py

# agent_state.py — durable state management for edge agents
import sqlite3, json
from datetime import datetime, UTC
from pathlib import Path
 
class EdgeAgentState:
    def __init__(self, db_path: str = "agent_state.db"):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self._init_schema()
 
    def _init_schema(self):
        self.conn.executescript("""
            CREATE TABLE IF NOT EXISTS memory (
                key TEXT PRIMARY KEY,
                value TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS task_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                task TEXT NOT NULL,
                result TEXT,
                steps INTEGER,
                started_at TEXT NOT NULL,
                completed_at TEXT
            );
            CREATE TABLE IF NOT EXISTS sensor_history (
                sensor_id TEXT NOT NULL,
                value REAL NOT NULL,
                recorded_at TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS idx_sensor_time ON sensor_history(sensor_id, recorded_at DESC);
        """)
        self.conn.commit()
 
    def remember(self, key: str, value: object) -> None:
        self.conn.execute(
            "INSERT OR REPLACE INTO memory VALUES (?, ?, ?)",
            (key, json.dumps(value), datetime.now(UTC).isoformat())
        )
        self.conn.commit()
 
    def recall(self, key: str, default=None) -> object:
        row = self.conn.execute("SELECT value FROM memory WHERE key = ?", (key,)).fetchone()
        return json.loads(row[0]) if row else default
 
    def record_sensor(self, sensor_id: str, value: float) -> None:
        self.conn.execute(
            "INSERT INTO sensor_history VALUES (?, ?, ?)",
            (sensor_id, value, datetime.now(UTC).isoformat())
        )
        self.conn.commit()
 
    def get_sensor_trend(self, sensor_id: str, last_n: int = 10) -> list[float]:
        rows = self.conn.execute(
            "SELECT value FROM sensor_history WHERE sensor_id = ? ORDER BY recorded_at DESC LIMIT ?",
            (sensor_id, last_n)
        ).fetchall()
        return [r[0] for r in reversed(rows)]

# hybrid_agent.py — edge-first with selective cloud escalation
async def run_hybrid_agent(task: str, state: EdgeAgentState) -> str:
    # Try edge first
    try:
        result = await run_local_agent(task, state, timeout_seconds=10)
        confidence = state.recall("last_confidence", 0.5)
        if confidence >= 0.75:
            return result
        # Low confidence — escalate to cloud
        print("Edge confidence low, escalating to cloud...")
        return await run_cloud_agent(task, local_context=result)
    except TimeoutError:
        return await run_cloud_agent(task)