ktg-plugin-marketplace/plugins/ms-ai-architect/skills/ms-ai-infrastructure/references/hybrid-edge/offline-first-ai-applications.md

Offline-First AI Application Patterns

Last updated: 2026-02 Status: GA Category: Hybrid Cloud & Edge AI

---

Introduksjon

Offline-first AI-applikasjoner er designet for a fungere primaert lokalt og synkronisere med skyen nar tilkobling er tilgjengelig. Dette monsteret snur den tradisjonelle sky-forst-tilnaermingen pa hodet: i stedet for a feile nar nettverket er nede, er applikasjonen designet for a operere uavhengig med lokal AI-inferens og datalagring.

For norsk offentlig sektor er offline-first sarlig relevant i felt-scenarioer: vegarbeidere som inspiserer infrastruktur i omrader uten dekning, ambulansepersonell som trenger AI-stoette i rurale omrader, beredskapspersonell under krisesituasjoner der kommunikasjonsinfrastruktur kan vaere nede, og maritime inspeksjoner langs kysten.

Microsoft tilbyr flere byggeklosser for offline-first AI: ONNX Runtime for lokal inferens, Azure IoT Edge for container-basert edge-prosessering med utvidet offline-stoette, Azure Container Storage for lokal persistens med automatisk sky-synkronisering, og Phi-modeller for lokale SLM-kapabiliteter.

---

Kjernekomponenter

Komponent	Formal	Teknologi
ONNX Runtime	Lokal AI-inferens uten sky	Cross-platform
Azure IoT Edge	Utvidet offline-kapabilitet	Container runtime
Azure Container Storage	Lokal lagring med sky-sync	Arc-enabled
Phi-3/4 SLM	Lokal sprakmodell	MIT-lisens
SQLite/LiteDB	Lokal database for offline-data	Embedded DB
CRDTs	Konfliktfri replikert datatype	Datastruktur
Azure Cosmos DB	Sky-database med offline SDK	Multi-model DB

---

Local-First Data Models

Arkitektur for lokal-forst AI

┌─────────────────────────────────────────────┐
│              Offline-First App               │
│                                              │
│  ┌──────────┐  ┌───────────┐  ┌───────────┐ │
│  │ UI Layer │  │ AI Engine │  │ Data Layer│ │
│  │          │  │           │  │           │ │
│  │ - Input  │←→│ - ONNX RT │←→│ - SQLite  │ │
│  │ - Output │  │ - Phi SLM │  │ - VectorDB│ │
│  │ - Status │  │ - Scoring │  │ - File    │ │
│  └──────────┘  └───────────┘  └───────────┘ │
│                                     ↕        │
│                              ┌────────────┐  │
│                              │ Sync Engine│  │
│                              │            │  │
│                              │ - Queue    │  │
│                              │ - Delta    │  │
│                              │ - Conflict │  │
│                              └──────┬─────┘  │
└─────────────────────────────────────┼────────┘
                                      ↕
                              [Sky (nar tilgjengelig)]

Event-sourcing for offline data

# Event-sourced datamodell for offline-first AI
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import json
import sqlite3
import uuid

@dataclass
class Event:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    type: str = ""
    entity_id: str = ""
    data: dict = field(default_factory=dict)
    synced: bool = False
    device_id: str = ""

class OfflineEventStore:
    def __init__(self, db_path: str, device_id: str):
        self.device_id = device_id
        self.conn = sqlite3.connect(db_path)
        self._init_schema()

    def _init_schema(self):
        self.conn.executescript("""
            CREATE TABLE IF NOT EXISTS events (
                id TEXT PRIMARY KEY,
                timestamp TEXT NOT NULL,
                type TEXT NOT NULL,
                entity_id TEXT NOT NULL,
                data TEXT NOT NULL,
                synced INTEGER DEFAULT 0,
                device_id TEXT NOT NULL
            );

            CREATE TABLE IF NOT EXISTS ai_results (
                id TEXT PRIMARY KEY,
                event_id TEXT REFERENCES events(id),
                model_version TEXT,
                result TEXT,
                confidence REAL,
                created_at TEXT,
                synced INTEGER DEFAULT 0
            );

            CREATE INDEX IF NOT EXISTS idx_events_synced ON events(synced);
            CREATE INDEX IF NOT EXISTS idx_events_entity ON events(entity_id);
        """)

    def append_event(self, event_type: str, entity_id: str, data: dict) -> Event:
        """Legg til hendelse i lokal event store"""
        event = Event(
            type=event_type,
            entity_id=entity_id,
            data=data,
            device_id=self.device_id
        )
        self.conn.execute(
            "INSERT INTO events VALUES (?, ?, ?, ?, ?, ?, ?)",
            (event.id, event.timestamp, event.type, event.entity_id,
             json.dumps(event.data), 0, event.device_id)
        )
        self.conn.commit()
        return event

    def store_ai_result(self, event_id: str, model_version: str,
                        result: dict, confidence: float):
        """Lagre AI-inferensresultat lokalt"""
        self.conn.execute(
            "INSERT INTO ai_results VALUES (?, ?, ?, ?, ?, ?, ?)",
            (str(uuid.uuid4()), event_id, model_version,
             json.dumps(result), confidence,
             datetime.utcnow().isoformat(), 0)
        )
        self.conn.commit()

    def get_unsynced_events(self, limit: int = 100) -> list[Event]:
        """Hent hendelser som ikke er synkronisert"""
        cursor = self.conn.execute(
            "SELECT * FROM events WHERE synced = 0 ORDER BY timestamp LIMIT ?",
            (limit,)
        )
        return [Event(*row) for row in cursor.fetchall()]

    def mark_synced(self, event_ids: list[str]):
        """Marker hendelser som synkronisert"""
        placeholders = ",".join("?" * len(event_ids))
        self.conn.execute(
            f"UPDATE events SET synced = 1 WHERE id IN ({placeholders})",
            event_ids
        )
        self.conn.commit()

---

Conflict Resolution on Sync

Konflikthondteringsstrategier

Strategi	Beskrivelse	Best for
Last-Write-Wins (LWW)	Siste endring vinner	Enkle data, lav risiko
First-Write-Wins	Forste endring vinner	Uforanderlige hendelser
Merge	Kombiner endringer automatisk	Komplementaere felt
CRDT	Konfliktfri replikert datatype	Tallere, sett, tekst
Custom Resolution	Applikasjonsspesifikk logikk	Komplekse forretningsregler

Implementering av konflikthondtering

# Konflikthondtering for offline-first AI-applikasjon
from enum import Enum
from typing import Callable

class ConflictStrategy(Enum):
    LAST_WRITE_WINS = "lww"
    FIRST_WRITE_WINS = "fww"
    MERGE = "merge"
    MANUAL = "manual"

class SyncConflictResolver:
    def __init__(self, strategy: ConflictStrategy = ConflictStrategy.LAST_WRITE_WINS):
        self.strategy = strategy
        self.custom_resolvers: dict[str, Callable] = {}

    def register_resolver(self, entity_type: str, resolver: Callable):
        """Registrer egendefinert konfliktloeser for en entitetstype"""
        self.custom_resolvers[entity_type] = resolver

    def resolve(self, local_event: dict, remote_event: dict) -> dict:
        """Los konflikt mellom lokal og fjern hendelse"""
        entity_type = local_event.get("type", "")

        # Egendefinert resolver har forrang
        if entity_type in self.custom_resolvers:
            return self.custom_resolvers[entity_type](local_event, remote_event)

        if self.strategy == ConflictStrategy.LAST_WRITE_WINS:
            return self._last_write_wins(local_event, remote_event)
        elif self.strategy == ConflictStrategy.FIRST_WRITE_WINS:
            return self._first_write_wins(local_event, remote_event)
        elif self.strategy == ConflictStrategy.MERGE:
            return self._merge(local_event, remote_event)
        else:
            return {"conflict": True, "local": local_event, "remote": remote_event}

    def _last_write_wins(self, local: dict, remote: dict) -> dict:
        local_ts = local.get("timestamp", "")
        remote_ts = remote.get("timestamp", "")
        return local if local_ts >= remote_ts else remote

    def _merge(self, local: dict, remote: dict) -> dict:
        """Merge ved a kombinere ikke-overlappende felt"""
        merged = {**remote.get("data", {})}
        for key, value in local.get("data", {}).items():
            if key not in merged or merged[key] is None:
                merged[key] = value
            elif key in merged and value != merged[key]:
                # Begge har endret — behold begge med suffix
                merged[f"{key}_local"] = value
                merged[f"{key}_remote"] = merged[key]
        return {"data": merged, "merge_status": "auto_merged"}


# Eksempel: Konflikthondtering for AI-inspeksjonsresultater
resolver = SyncConflictResolver(ConflictStrategy.MERGE)

def resolve_inspection(local, remote):
    """Inspeksjoner: Behold den med hoeyest AI-confidence"""
    local_conf = local.get("data", {}).get("ai_confidence", 0)
    remote_conf = remote.get("data", {}).get("ai_confidence", 0)
    winner = local if local_conf >= remote_conf else remote
    winner["data"]["conflict_resolved"] = True
    winner["data"]["alternative_confidence"] = min(local_conf, remote_conf)
    return winner

resolver.register_resolver("inspection_result", resolve_inspection)

---

Progressive Enhancement

Progressiv AI-kapabilitet

# Progressiv enhancement: Eskalerer AI-kapabilitet basert pa tilkobling
from enum import Enum
import asyncio

class ConnectivityLevel(Enum):
    OFFLINE = 0      # Ingen tilkobling
    LOW_BANDWIDTH = 1 # < 1 Mbps
    CONNECTED = 2     # Normal tilkobling
    HIGH_BANDWIDTH = 3 # > 10 Mbps

class ProgressiveAIService:
    def __init__(self):
        self.local_model = None   # Phi-3 Mini INT4 (alltid tilgjengelig)
        self.medium_model = None  # Phi-3 Small (krever > 16 GB RAM)
        self.cloud_client = None  # Azure OpenAI (krever tilkobling)

    async def classify_document(self, text: str) -> dict:
        """Klassifiser dokument med best tilgjengelig AI"""
        connectivity = await self.check_connectivity()

        if connectivity >= ConnectivityLevel.HIGH_BANDWIDTH:
            # Nivaa 3: Full sky-AI med GPT-4o
            return await self._classify_cloud(text, model="gpt-4o")

        elif connectivity >= ConnectivityLevel.CONNECTED:
            # Nivaa 2: Sky-AI med lettere modell
            return await self._classify_cloud(text, model="gpt-4o-mini")

        elif connectivity >= ConnectivityLevel.LOW_BANDWIDTH:
            # Nivaa 1: Lokal medium modell med sky-validering
            local_result = self._classify_local(text, self.medium_model)
            # Asynkron validering i bakgrunn nar mulig
            asyncio.create_task(self._validate_in_background(text, local_result))
            return local_result

        else:
            # Nivaa 0: Full offline med lokal mini-modell
            return self._classify_local(text, self.local_model)

    def _classify_local(self, text: str, model) -> dict:
        """Lokal klassifisering med ONNX-modell"""
        result = model.predict(text)
        return {
            "classification": result["label"],
            "confidence": result["score"],
            "model": "local",
            "connectivity": "offline",
            "note": "Resultat fra lokal modell — verifiseres ved tilkobling"
        }

    async def check_connectivity(self) -> ConnectivityLevel:
        """Sjekk navaerende tilkoblingsniva"""
        try:
            import aiohttp
            async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=3)) as session:
                async with session.get("https://management.azure.com/health") as resp:
                    if resp.status == 200:
                        # Estimer bandwidth
                        return ConnectivityLevel.HIGH_BANDWIDTH
        except Exception:
            pass

        try:
            # Proeving med minimal data
            import socket
            socket.create_connection(("8.8.8.8", 53), timeout=2)
            return ConnectivityLevel.LOW_BANDWIDTH
        except Exception:
            return ConnectivityLevel.OFFLINE

UI-indikasjon av AI-nivaa

# Statusindikator for progressive AI
AI_LEVEL_INFO = {
    ConnectivityLevel.OFFLINE: {
        "label": "Offline-modus",
        "description": "Bruker lokal AI-modell. Resultater synkroniseres ved tilkobling.",
        "icon": "offline",
        "accuracy": "God (lokal modell)",
        "features": ["Klassifisering", "Oppsummering", "Uttrekking"]
    },
    ConnectivityLevel.LOW_BANDWIDTH: {
        "label": "Begrenset tilkobling",
        "description": "Lokal AI med bakgrunns-validering.",
        "icon": "low_signal",
        "accuracy": "God+ (validert i bakgrunn)",
        "features": ["Klassifisering", "Oppsummering", "Uttrekking", "Bakgrunns-validering"]
    },
    ConnectivityLevel.CONNECTED: {
        "label": "Tilkoblet",
        "description": "Sky-AI med standard modell.",
        "icon": "connected",
        "accuracy": "Hoey",
        "features": ["Alle funksjoner", "RAG", "Avansert analyse"]
    },
    ConnectivityLevel.HIGH_BANDWIDTH: {
        "label": "Full tilkobling",
        "description": "Sky-AI med avansert modell.",
        "icon": "full_signal",
        "accuracy": "Hoeyest",
        "features": ["Alle funksjoner", "RAG", "Avansert analyse", "Bildeanalyse"]
    }
}

---

Offline Capability Testing

Testrammeverk for offline AI

# Testrammeverk for offline-first AI-applikasjon
import pytest
from unittest.mock import patch, AsyncMock

class TestOfflineAI:
    """Tester for offline-first AI-funksjonalitet"""

    @pytest.fixture
    def ai_service(self):
        return ProgressiveAIService()

    @pytest.fixture
    def event_store(self, tmp_path):
        return OfflineEventStore(str(tmp_path / "test.db"), "test-device")

    def test_offline_classification(self, ai_service):
        """AI-klassifisering skal fungere uten nettverkstilkobling"""
        with patch.object(ai_service, 'check_connectivity',
                         return_value=ConnectivityLevel.OFFLINE):
            result = asyncio.run(ai_service.classify_document(
                "Vedtak om avslag pa soeknad om byggetillatelse"
            ))
            assert result["classification"] is not None
            assert result["connectivity"] == "offline"
            assert result["confidence"] > 0.5

    def test_event_persistence_offline(self, event_store):
        """Hendelser skal lagres lokalt ved offline"""
        event = event_store.append_event(
            "inspection", "bridge-001",
            {"status": "ok", "notes": "Ingen synlige skader"}
        )
        assert event.synced is False
        assert event.device_id == "test-device"

        # Hent usynkroniserte hendelser
        unsynced = event_store.get_unsynced_events()
        assert len(unsynced) == 1

    def test_sync_after_reconnection(self, event_store):
        """Usynkroniserte hendelser skal koes for synkronisering"""
        # Simuler 10 offline-hendelser
        for i in range(10):
            event_store.append_event(
                "sensor_reading", f"sensor-{i}",
                {"value": i * 1.5}
            )

        unsynced = event_store.get_unsynced_events()
        assert len(unsynced) == 10

        # Simuler synkronisering
        synced_ids = [e.id for e in unsynced[:5]]
        event_store.mark_synced(synced_ids)

        remaining = event_store.get_unsynced_events()
        assert len(remaining) == 5

    def test_conflict_resolution(self):
        """Konflikter ved sync skal loses deterministisk"""
        resolver = SyncConflictResolver(ConflictStrategy.LAST_WRITE_WINS)

        local = {"timestamp": "2026-02-12T10:00:00", "data": {"status": "ok"}}
        remote = {"timestamp": "2026-02-12T09:00:00", "data": {"status": "warning"}}

        result = resolver.resolve(local, remote)
        assert result["data"]["status"] == "ok"  # Nyeste vinner

    def test_progressive_enhancement(self, ai_service):
        """AI-kvalitet skal oeke med bedre tilkobling"""
        results = {}
        for level in ConnectivityLevel:
            with patch.object(ai_service, 'check_connectivity',
                            return_value=level):
                result = asyncio.run(ai_service.classify_document("test"))
                results[level] = result

        # Verifiser at sky-resultat har hoeyere konfidensangivelse
        assert results[ConnectivityLevel.OFFLINE]["model"] == "local"

---

Norsk offentlig sektor

Felt-scenarier som krever offline-first

Scenario	Etat	Offline-varighet	AI-funksjon
Vegfinspeksjon	SVV	Timer	Skadeklassifisering
Ambulanse	Helseetaten	Minutter-timer	Triagering
Beredskap	DSB	Dager	Situasjonsanalyse
Maritime inspeksjoner	Sjoefartsdir.	Timer-dager	Rapport-generering
Grensekontroll	Politiet	Minutter	Dokumentverifisering
Skogsbrannberedskap	110-sentraler	Timer	Risikoanalyse

Krav til offline-first i offentlig sektor

• Applikasjonen MA fungere uten nettverkstilkobling
• Lokale AI-resultater MA vaere tydelig merket som "ikke-validert"
• Synkronisering MA skje automatisk ved tilkobling
• Konflikthondtering MA vaere deterministisk og sporbar
• Data MA vaere kryptert lokalt (BitLocker/LUKS)

---

Beslutningsrammeverk

Scenario	Anbefaling	Begrunnelse
Felt-app med periodisk tilkobling	Full offline-first med event sourcing	Data bevares alltid lokalt
Sanntids-AI med fallback	Progressiv enhancement	Best mulig kvalitet per tilstand
Multi-enhet med sync	CRDTs + event store	Konfliktfri synkronisering
Kritisk infrastruktur	Azure IoT Edge extended offline	Uavhengig drift i uker
Klient-app pa PC	SQLite + ONNX RT + bakgrunns-sync	Enkel, palitelig arkitektur
Beredskapsapplikasjon	Full offline med manuell sync	Ingen skyavhengighet

---

For Cosmo

• Offline-first er et designprinsipp, ikke en feilhaandterings-strategi — applikasjonen MA designes for a fungere lokalt foerst, med sky som en berikelse nar tilgjengelig
• Event sourcing er det foretrukne datamoensteeret for offline-first AI — alle hendelser og AI-resultater lagres lokalt som uforanderlige events og synkroniseres inkrementelt
• Progressiv enhancement gir graceful degradation — definer tydelige AI-kapabilitetsnivaaer (offline/begrenset/tilkoblet/full) og kommuniser dette til brukeren
• Konflikthondtering maa vaere deterministisk og sporbar — bruk Last-Write-Wins som standard, med custom resolvers for doemenespesifikke regler (f.eks. hoeyest AI-confidence vinner)
• For norsk offentlig sektor: Test offline-scenarioer som foersteklasses testcase — ikke anta tilkobling, og sooerg for at felt-personell kan fullfoere sine oppgaver uavhengig av nettverksstatus