781d98f62f
Same bulk replacement applied to plugin-internal KB, examples, fixtures, tests, and docs. Real organization names, persona names, internal system identifiers, and domain-specific terms replaced with fictional generic public-sector entity (DDT) and generic terminology. Scope: - okr/ — examples, governance, framework, integrations, sources - ms-ai-architect/ — KB references (engineering, governance, security, infrastructure, advisor), tests/fixtures, agents, docs - linkedin-thought-leadership/ — voice samples, network-builder, examples (genericized identifying headlines to "[your organization]") - llm-security/ — research notes, scan report Manual genericization beyond bulk replace: - okr SKILL.md "Primary user / Domain" — generic Norwegian public sector - linkedin-voice SKILL.md headline placeholder - network-builder.md headline placeholder - high-engagement-posts.md voice sample employer line + hashtag Phase 3 (factual-attribution review) remains: a few KB files attribute publicly known transport-sector docs/datasets (e.g. håndbok V440, NVDB) to the fictional DDT after bulk replace. Needs manual semantic review to either remove or restore correct citation without re-introducing affiliation references. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
491 lines
19 KiB
Markdown
491 lines
19 KiB
Markdown
# 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
|
|
|
|
```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", "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 | DDT | 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
|