The canonical system design process teaches web applications, but when designing Physical AI systems--applications where AI models are deployed on or in close proximity to physical assets such as manufacturing plants, logistics centers, or autonomous machines--the typical API load balancer, stateless microservices, relational databases, and Redis caching layers is only 20% of the solution.
Deploying AI on production environments imposes unique edge constraints such as latency, packet loss, sensor noise, and zero downtime hardware execution.
Here is the architecture of a production grade AIoT stack:
The Four-Engine Architecture for AIoT Systems
In order to reduce round trip time from the edge to the cloud, physical AI systems decouple concerns into four distinct engines:
Identification Engine (Spatial & Asset State)
Before any telemetry can be ingested, the system needs to uniquely bind the data to a physical entity. In this sensing engine, RFID, BLE beacons, or UWB are used to associate identifiers on a stateful database.
Sensing Engine (High Throughput Ingestion)
Data packets (physical signals) will flow in at high frequencies in protocols such as MQTT, CoAP, or Modbus. This sensing engine is where raw data is normalized and filtered ahead of being passed into the decision engine.
AI Decision Engine (Inference, Logic, State)
Local inference on edge hardware (NVIDIA jetson, coral, etc), using light weight runtimes (ONNX runtime, tensorRT), is where the decision engine evaluates sensor telemetry and updates running digital twins.
Action Engine (Hardware Actuation) Finally, the action engine is how Physical AI systems take action. The decision engine will signal the action engine to make updates or activate relays, PLCs, or other hardware systems to enact changes in the physical world.
Edge Ingestion & Local Inference Pattern
Here is a simplified pattern in Python of how an edge gateway can ingest sensor packets, run local ONNX inference, and execute local control commands while queuing telemetry data for asynchronous ingestion:
Python
import json
import time
import queue
import threading
sensor_queue = queue.Queue()
class EdgePhysicalAIEngine:
def **init**(self, model_path: str, confidence_threshold: float = 0.85):
self.threshold = confidence_threshold
print(f"[SYSTEM] edge inference model from {model_path}...")
def run_local_inference(self, payload: dict) -> dict:
vibration = payload.get("vibration_hz", 0.0)
temperature = payload.get("temp_c", 0.0)
anomaly_score = (vibration 0.6) + (temperature 0.4) / 100.0
is_critical = anomaly_score > self.threshold
return {
"asset_id": payload.get("asset_id"),
"anomaly_score": round(anomaly_score, 4),
"trigger_action": is_critical
}
def execute_physical_action(self, asset_id: str):
print(f"[ACTION ENGINE] CRITICAL: Triggering local safety relay for Asset: {asset_id}")
def sync_to_cloud_async(self, telemetry_result: dict):
print(f"[CLOUD SYNC] Batching telemetry for asset {telemetry_result['asset_id']}")
engine = EdgePhysicalAIEngine(model_path="models/vibration_anomaly.onnx")
sample_packet = {"asset_id": "PUMP-4021", "vibration_hz": 1.42, "temp_c": 88.5}
result = engine.run_local_inference(sample_packet)
if result["trigger_action"]:
engine.execute_physical_action(result["asset_id"])
engine.sync_to_cloud_async(result)
The Build vs. Integrate Tradeoff for Developers
When technical founders launch an AIoT venture, they spend 80% of their early engineering bandwidth writing low level drivers, protocol parsing, and edge to cloud syncing code, leaving them little time to build domain specific ML models or validate business logic with users.
Many dev teams will leverage pre-integrated edge hardware to accelerate time to market while working with an institutional co-builder (ex: Aperture Venture Studio) to gain access to production grade sensing infrastructure, test their hypotheses in real world environments, and raise capital while focusing on building their domain specific ML models.
What edge stack are you using?
Are you deploying PyTorch/ONNX models on edge gateways or doing hybrid edge-cloud processing? Let's discuss edge inference architectures in the comments below!