Kjell Tore Guttormsen
544934dc57
Reference files, test fixtures, the playground demo project and one design document now use generic, fictitious examples (buildings, energy, water, grants, municipal services). The playground demo (17 fixtures plus the embedded demo state) tells one consistent story: a municipal customer chatbot that pre-screens housing-benefit applications, classified under Annex III point 5(a). The embedded demo copies were edited in place rather than regenerated, because they already carry newer AI Act dates than the fixture files. Legal text is unchanged. Test semantics are unchanged. Four dark-theme onboarding screenshots with outdated placeholder text are removed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
504 lines
20 KiB
Markdown
504 lines
20 KiB
Markdown
# Offline-First AI Application Patterns
|
|
|
|
**Last updated:** 2026-02
|
|
**Status:** GA
|
|
**Category:** Hybrid Cloud & Edge AI
|
|
**Type:** reference
|
|
|
|
---
|
|
|
|
## Innhold
|
|
|
|
- [Introduksjon](#introduksjon)
|
|
- [Kjernekomponenter](#kjernekomponenter)
|
|
- [Local-First Data Models](#local-first-data-models)
|
|
- [Conflict Resolution on Sync](#conflict-resolution-on-sync)
|
|
- [Progressive Enhancement](#progressive-enhancement)
|
|
- [Offline Capability Testing](#offline-capability-testing)
|
|
- [Norsk offentlig sektor](#norsk-offentlig-sektor)
|
|
- [Beslutningsrammeverk](#beslutningsrammeverk)
|
|
- [For arkitekten](#for-arkitekten)
|
|
|
|
## 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: driftspersonell som inspiserer ledningsnett 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
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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", "pumpestasjon-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 |
|
|
|----------|------|-----------------|-------------|
|
|
| Ledningsinspeksjon | Kommunalt vannverk | 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 arkitekten
|
|
|
|
- **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
|