Skip to main content

Build an AI worker

A Heldar AI worker is any process that can speak four core HTTP endpoints (plus three optional semantic-embedding endpoints — see below). The kernel samples frames and stores results; your worker is a pure HTTP client that discovers work, pulls frames, runs a model, and posts detections back. Because the contract is HTTP, a crashing or slow worker can never stall ingest or recording.

The reference implementation is apps/ai/worker.py (a small, dependency-light Python worker). The full integration guide, including the zone and ANPR analyzers, is docs/AI-WORKERS.md.

The contract​

All endpoints live under /api/v1 and return JSON, except the frame, which is image/jpeg.

1. Discover work - GET /api/v1/ai/tasks​

Returns every enabled task on an enabled camera, each carrying the frame_url to pull. This is your whole work list. Re-poll every few seconds to pick up newly enabled or disabled tasks.

[
  {
    "id": "ai_3f2a9c1b...",
    "camera_id": "gate_a",
    "task_type": "detection",
    "stream_profile": "sub",
    "fps": 5.0,
    "width": 1280,
    "config": { "classes": ["person", "car"], "min_confidence": 0.4 },
    "frame_url": "/api/v1/cameras/gate_a/frame"
  }
]

task_type is a free-form string you define; the kernel uses fps / width / enabled only to drive the sampler, and config is an opaque blob your worker interprets. The fps here is the requested rate; the effective sampled rate after budgeting is reported by GET /api/v1/ai/samplers.

Running more than one worker? Pass a stable identity as ?worker_id= (e.g. GET /api/v1/ai/tasks?worker_id=nvr1:4242) and the kernel registers you as a live worker and returns only your shard of the task list, modulo-sharded across the live worker set. Omitting it returns everything (single-worker back-compat) — a custom worker that skips it will double-process every task next to any other worker. Tasks are rebalanced whenever the live worker set changes: a worker whose last poll is older than the liveness TTL (60s) is treated as gone and its tasks reassigned, so poll faster than the TTL (the default poll interval is 10s).

2. Pull a frame - GET /api/v1/cameras/{id}/frame​

Serves the latest sampled JPEG for a camera. Pull it on your own cadence, typically at or just under the task fps.

200 OK
Content-Type: image/jpeg
Cache-Control: no-store
x-frame-age-ms: 142
x-frame-captured-at: 2026-06-13T08:15:31.120+00:00

<JPEG bytes>
  • x-frame-age-ms - milliseconds since the frame was written. Use it to skip stale frames if the sampler has gone offline.
  • x-frame-captured-at - the write timestamp. Dedupe on it so you do not re-analyze an unchanged frame, and optionally echo it back as the detection timestamp so detections align to capture time.

A 404 means no frame exists yet (no enabled AI task for the camera, or the sampler has not produced its first frame). Treat it as a skipped cycle, not an error.

3. Post results - POST /api/v1/ai/events​

Post a batch of detections for one camera and task, optionally with a single derived event in the same call.

{
  "camera_id": "gate_a",
  "task_type": "detection",
  "timestamp": "2026-06-13T08:15:31.120Z",
  "detections": [
    {
      "label": "person",
      "confidence": 0.92,
      "bbox": [0.41, 0.30, 0.08, 0.22],
      "track_id": "t-17",
      "attributes": { "zone": "entry_lane_a" }
    }
  ],
  "event": {
    "event_type": "person_in_red_zone",
    "severity": "warning",
    "payload": { "zone": "red_a", "track_id": "t-17" }
  }
}

Field rules:

  • camera_id is required and must exist, else 404.
  • task_type is required and is stored on each detection row.
  • timestamp is optional RFC3339 and applies to the whole batch; omitted or unparseable falls back to server now().
  • detections is optional (defaults to []); send [] to post only an event. Every field inside a detection is optional.
  • event is optional. event_type is required when present; severity defaults to info (use warning or critical to trigger the alert webhook); payload defaults to an empty object.

The response is { "detections_ingested": N }.

4. Sampler status - GET /api/v1/ai/samplers​

Per-camera sampler state (connecting / sampling / offline / error / stopped) and the effective budgeted fps. Useful for dashboards and for confirming the kernel is actually producing frames.

5. Optional - semantic embeddings​

Three additive endpoints back the semantic-search pipeline. A worker that never runs an embedding task can ignore them, and GET /api/v1/ai/tasks is unchanged (still a bare array), so existing workers keep working.

  • POST /api/v1/ai/embeddings - post a batch of crop embeddings for one camera: { camera_id, model, dim (1..4096), frame_id?, items: [...] } with 1-128 items, each { vec (exactly dim finite floats), track_id?, label?, bbox?, timestamp?, thumb_b64? } (thumb_b64 is an optional JPEG crop for the search UI, capped at ~96 KB of base64). frame_id is an idempotency key: rows dedupe on (camera_id, frame_id, track_id), so a redelivered batch is skipped silently and the { "embeddings_ingested": N } response counts only rows actually inserted.
  • GET /api/v1/ai/embed-queries?worker_id=<id> - atomically claim up to 4 pending search queries (text or base64 image) to embed. Returns an object, { "queries": [ { "id", "kind", "payload" } ] }, not a bare array. Poll it fast (the reference worker polls every ~1 s): a semantic search holds the operator's request open for only ~3 s waiting for the query vector, which the ~10 s tasks poll could never meet - that is why this is a separate endpoint rather than a piggyback on /ai/tasks. Idle polls are read-only, and queries older than 60 s expire undelivered.
  • POST /api/v1/ai/embed-queries/{id}/result - post the query vector back ({ vec, model, dim }) or { "error": "..." } so the waiting search fails fast instead of timing out. First result wins; a late duplicate is a 200 no-op ({ "updated": false }).

The bbox convention​

bbox is [x, y, w, h] normalized to 0..1, top-left origin. Normalizing keeps detections resolution-independent, so they survive any later change to the sampled width and map directly onto normalized zone polygons. The kernel stores the box as raw JSON and does not validate its shape, so your worker owns correctness.

A detection with both a track_id and a bbox drives the kernel zone engine (its ground point is the box bottom-center); detections without them are still stored but cannot raise zone events.

The worker loop​

tasks = GET /api/v1/ai/tasks                 # refresh every few seconds
for each task (own thread / async task):
    loop at ~task.fps:
        resp = GET task.frame_url
        if resp is 404:        sleep, continue          # no frame yet
        if x-frame-captured-at == last_seen: continue   # unchanged frame; skip
        dets, event = analyze(task, resp.body)
        if dets or event:
            POST /api/v1/ai/events { camera_id, task_type, timestamp,
                                     detections: dets, event: event }

Because the served frame is last-value, pulling faster than the sampler writes returns the same bytes; dedupe on x-frame-captured-at. Pulling slower simply drops intermediate frames, which is fine for detection and tracking at these rates.

Plugging a model in​

The reference worker defines an Analyzer base class and creates one instance per task thread, so per-camera state (a previous frame, a tracker) lives on self. Analyzers register by task_type; an unknown type falls back to a placeholder that exercises the frame path but never fabricates detections. A working, model-free MotionAnalyzer is registered for task_type: "motion", so you can validate the whole sampler to worker to events path with no model and no GPU.

A real detector slots in as one subclass and one register(...) call, with no change to the kernel or the HTTP contract:

from worker import Analyzer, AnalysisResult, Detection, FrameContext, register

class YoloAnalyzer(Analyzer):
    name = "yolo"
    def __init__(self, config, log):
        super().__init__(config, log)
        from ultralytics import YOLO
        self.model = YOLO(config.get("weights", "yolov8n.pt"))
        self.conf = float(config.get("threshold", 0.25))

    def analyze(self, frame: FrameContext) -> AnalysisResult:
        img = frame.image(); w, h = img.size
        dets = []
        for r in self.model(img, conf=self.conf, verbose=False):
            for b in r.boxes:
                x1, y1, x2, y2 = b.xyxy[0].tolist()
                dets.append(Detection(
                    label=self.model.names[int(b.cls)],
                    confidence=float(b.conf),
                    bbox=[x1/w, y1/h, (x2-x1)/w, (y2-y1)/h]))  # normalized 0..1
        return AnalysisResult(detections=dets)

register("detection", YoloAnalyzer)   # replaces the placeholder for "detection"

The kernel never touches your model. It only routes results to consumers by task_type: detection results with track ids drive the zone engine, anpr results feed the access-control engine, embedding tasks index crop vectors for semantic search, and so on. To add a new pipeline, pick a new task_type, post its results, and write a consumer for it (see Build a module).

The reference worker also ships an embedding analyzer (YOLO + ByteTrack + open_clip; vehicle classes only by default - person is deliberately excluded) that embeds each track on first sight and then every stride_seconds while it moves (static suppression is on by default — static_suppression true, static_epsilon 0.02 the normalized-bbox movement threshold below which a track counts as parked, and static_refresh_seconds 3600 the slow re-embed floor that keeps a permanently parked object in the index across the retention TTL), and posts the vectors through the optional endpoints above instead of /api/v1/ai/events - it emits no detections, so it never double-fires the zone or access-control consumers. A companion EmbedQueryWorker thread polls the query queue (~1 s, HELDAR_AI_EMBED_POLL_INTERVAL) and answers search queries with the CLIP text or image tower. Both need the optional CLIP extra (pip install -r requirements-embed.txt, model via HELDAR_AI_CLIP_MODEL / HELDAR_AI_CLIP_PRETRAINED, defaults ViT-B-32-quickgelu / openai); without it, embedding tasks fall back to the placeholder and semantic searches return a fast 503 instead of hanging.

Running the reference worker​

cd apps/ai
python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
HELDAR_API=http://localhost:8000 python worker.py
# or: python worker.py --api http://localhost:8000 --log-format json

Worker-side config (CLI flag or env var) covers the API base URL (--api / HELDAR_API), the task poll interval (--poll-interval / HELDAR_AI_POLL_INTERVAL), the sharding identity (--worker-id / HELDAR_AI_WORKER_ID, default <hostname>:<pid> — sent as ?worker_id= so multiple workers split the task list), and HTTP/backoff/logging knobs. The full table is in apps/ai/README.md.