用 AWS 构建离线优先的边缘生成式 AI：从本地推理到云端同步

edge-assistant/
├── recipe.yaml
└── artifacts/
    └── com.example.EdgeAssistant/
        └── 1.0.0/
            └── assistant.py

RecipeFormatVersion: "2020-01-25"
ComponentName: "com.example.EdgeAssistant"
ComponentVersion: "1.0.0"
ComponentDescription: "Offline-first diagnostic assistant"
ComponentPublisher: "Example"
ComponentConfiguration:
  DefaultConfiguration:
    InferenceUrl: "http://127.0.0.1:8080/generate"
    DatabasePath: "/greengrass/v2/work/offline-assistant/events.db"
Manifests:
  - Platform:
      os: linux
    Lifecycle:
      Install: "python3 -m pip install --user requests"
      Run: >-
        python3 {artifacts:path}/assistant.py
        --url '{configuration:/InferenceUrl}'
        --db '{configuration:/DatabasePath}'
    Artifacts:
      - URI: "s3://REPLACE_WITH_BUCKET/components/assistant.py"

#!/usr/bin/env python3
import argparse
import json
import sqlite3
import time
from pathlib import Path

import requests


def init_db(path: str) -> sqlite3.Connection:
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(path)
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS events (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            created_at INTEGER NOT NULL,
            payload TEXT NOT NULL,
            synced INTEGER NOT NULL DEFAULT 0
        )
        """
    )
    conn.commit()
    return conn


def diagnose(url: str, observation: dict) -> dict:
    prompt = {
        "system": (
            "You are an industrial diagnostic assistant. "
            "Use only the supplied observation. Return JSON with "
            "summary, checks, and requires_human_approval."
        ),
        "observation": observation,
    }
    response = requests.post(url, json=prompt, timeout=20)
    response.raise_for_status()
    return response.json()


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--url", required=True)
    parser.add_argument("--db", required=True)
    args = parser.parse_args()

    conn = init_db(args.db)
    observation = {
        "asset_id": "pump-07",
        "alarm": "BEARING_TEMPERATURE_HIGH",
        "temperature_c": 91.4,
        "five_minute_average_c": 86.2,
    }

    try:
        result = diagnose(args.url, observation)
    except (requests.RequestException, ValueError) as exc:
        result = {
            "summary": "Local inference unavailable",
            "checks": ["Inspect sensor wiring", "Follow the approved manual"],
            "requires_human_approval": True,
            "error": str(exc),
        }

    event = {"observation": observation, "diagnosis": result}
    conn.execute(
        "INSERT INTO events(created_at, payload) VALUES (?, ?)",
        (int(time.time()), json.dumps(event)),
    )
    conn.commit()
    print(json.dumps(event, indent=2))


if __name__ == "__main__":
    main()

aws greengrassv2 create-component-version \
  --region us-east-1 \
  --inline-recipe fileb://recipe.yaml

aws greengrassv2 create-deployment \
  --region us-east-1 \
  --target-arn arn:aws:iot:us-east-1:123456789012:thing/MyCoreDevice \
  --deployment-name offline-edge-assistant \
  --components '{"com.example.EdgeAssistant":{"componentVersion":"1.0.0"}}'