Netlify Edge MLOps Workflow

Netlify Edge Functions คืออะไร

Netlify Edge Functions เป็น Serverless Computing ที่ทำงานบน Deno Runtime ที่ Edge Server กระจายอยู่ทั่วโลก จุดเด่นคือ Code จะถูกรันที่ Edge Location ใกล้กับผู้ใช้มากที่สุด ทำให้มี Latency ต่ำกว่า Traditional Serverless Functions ที่ทำงานบน Region เดียว Edge Functions รองรับ TypeScript/JavaScript และสามารถเข้าถึง Request/Response ได้โดยตรง

เมื่อนำ Edge Functions มาใช้ในบริบทของ MLOps จะสามารถทำ ML Inference ที่ Edge ได้โดยตรง ลด Latency ในการเรียก Model API จาก 200-500ms เหลือเพียง 10-50ms เหมาะสำหรับ Use Case อย่าง Real-time Personalization, Content Recommendation, Sentiment-based Routing และ Feature Flag ที่ใช้ ML Model ตัดสินใจ

สถาปัตยกรรม MLOps บน Netlify Edge

สถาปัตยกรรมของ MLOps Workflow บน Netlify ประกอบด้วยหลาย Component ที่ทำงานร่วมกัน

• Model Training Pipeline: ใช้ Python + scikit-learn หรือ PyTorch Train Model บน Cloud (AWS SageMaker, GCP Vertex AI หรือ Local)
• Model Export: แปลง Model เป็น ONNX Runtime หรือ TensorFlow.js Format ที่ทำงานบน JavaScript Runtime ได้
• Model Registry: เก็บ Model Artifacts ใน Git LFS หรือ Cloud Storage พร้อม Version Tag
• Edge Deployment: Deploy Model ร่วมกับ Edge Functions ผ่าน Netlify CLI หรือ Git Push
• A/B Testing: ใช้ Edge Functions Route Traffic ระหว่าง Model Version ต่างๆ
• Monitoring: ติดตาม Model Performance, Prediction Distribution และ Latency

การตั้งค่า Project และ Edge Functions

# สร้าง Netlify Project
mkdir ml-edge-app && cd ml-edge-app
npm init -y

# โครงสร้าง Project
# ml-edge-app/
# ├── netlify/
# │   └── edge-functions/
# │       ├── ml-inference.ts
# │       ├── ab-test.ts
# │       └── monitor.ts
# ├── models/
# │   ├── sentiment-v1.onnx
# │   └── sentiment-v2.onnx
# ├── src/
# │   └── index.html
# ├── scripts/
# │   ├── train_model.py
# │   └── export_onnx.py
# ├── netlify.toml
# └── package.json

# netlify.toml
cat > netlify.toml << 'TOML'
[build]
  publish = "src"
  command = "echo 'Static site - no build needed'"

[[edge_functions]]
  function = "ml-inference"
  path = "/api/predict"

[[edge_functions]]
  function = "ab-test"
  path = "/api/recommend"

[[edge_functions]]
  function = "monitor"
  path = "/api/monitor"

[build.environment]
  MODEL_VERSION = "v2"
  AB_TEST_TRAFFIC_SPLIT = "0.2"
TOML

Model Training และ Export

# scripts/train_model.py
"""Train Sentiment Analysis Model และ Export เป็น ONNX"""
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from skl2onnx import convert_sklearn
from skl2onnx.common.data_types import StringTensorType
import joblib
import json

# Training Data (ตัวอย่าง)
texts = [
    "สินค้าดีมากครับ ชอบมาก", "ส่งเร็วมาก ประทับใจ",
    "คุณภาพแย่มาก ผิดหวัง", "ไม่คุ้มราคาเลย",
    "บริการดี พนักงานสุภาพ", "สินค้าชำรุด ไม่พอใจ",
    "ใช้งานง่าย ดีไซน์สวย", "รอนาน ส่งช้ามาก",
    "คุ้มค่ามากๆ แนะนำเลย", "สินค้าไม่ตรงปก หลอกลวง",
]
labels = [1, 1, 0, 0, 1, 0, 1, 0, 1, 0]  # 1=positive, 0=negative

# Train Pipeline
pipeline = Pipeline([
    ('tfidf', TfidfVectorizer(max_features=5000, ngram_range=(1, 2))),
    ('clf', LogisticRegression(max_iter=1000)),
])
pipeline.fit(texts, labels)

# Export เป็น ONNX
onnx_model = convert_sklearn(
    pipeline,
    "sentiment_model",
    initial_types=[("input", StringTensorType([None, 1]))],
    target_opset=12,
)
with open("models/sentiment-v2.onnx", "wb") as f:
    f.write(onnx_model.SerializeToString())

# Export Metadata
metadata = {
    "version": "v2",
    "accuracy": 0.92,
    "features": "tfidf_5000_ngram_1_2",
    "trained_at": "2026-02-28",
    "model_size_bytes": len(onnx_model.SerializeToString()),
}
with open("models/metadata.json", "w") as f:
    json.dump(metadata, f, indent=2)

print(f"Model exported: {metadata['model_size_bytes']} bytes")

Edge Functions สำหรับ ML Inference

// netlify/edge-functions/ml-inference.ts
// Edge Function สำหรับ Sentiment Analysis
import { Context } from "https://edge.netlify.com";

// Simple Sentiment Scoring (สำหรับ Edge — ไม่ใช้ ONNX Runtime)
// ใช้ Dictionary-based Approach สำหรับ Demo
const POSITIVE_WORDS = new Set([
  "ดี", "สวย", "เร็ว", "ชอบ", "ประทับใจ", "คุ้มค่า", "แนะนำ",
  "สุภาพ", "สะดวก", "ง่าย", "เยี่ยม", "พอใจ", "คุณภาพ",
]);
const NEGATIVE_WORDS = new Set([
  "แย่", "ช้า", "ผิดหวัง", "ชำรุด", "หลอก", "แพง", "เสีย",
  "ไม่ดี", "ไม่พอใจ", "ไม่คุ้ม", "เสียเวลา", "ห่วย",
]);

interface PredictionResult {
  text: string;
  sentiment: "positive" | "negative" | "neutral";
  confidence: number;
  model_version: string;
  latency_ms: number;
}

function predictSentiment(text: string): PredictionResult {
  const startTime = performance.now();
  const words = text.split(/\s+/);

  let positiveScore = 0;
  let negativeScore = 0;

  for (const word of words) {
    if (POSITIVE_WORDS.has(word)) positiveScore++;
    if (NEGATIVE_WORDS.has(word)) negativeScore++;
  }

  const total = positiveScore + negativeScore;
  let sentiment: "positive" | "negative" | "neutral" = "neutral";
  let confidence = 0.5;

  if (total > 0) {
    confidence = Math.max(positiveScore, negativeScore) / total;
    sentiment = positiveScore > negativeScore ? "positive" : "negative";
  }

  const latency = performance.now() - startTime;

  return {
    text,
    sentiment,
    confidence: Math.round(confidence * 100) / 100,
    model_version: Deno.env.get("MODEL_VERSION") || "v2",
    latency_ms: Math.round(latency * 100) / 100,
  };
}

export default async (request: Request, context: Context) => {
  // CORS Headers
  const headers = {
    "Content-Type": "application/json",
    "Access-Control-Allow-Origin": "*",
    "Cache-Control": "no-cache",
    "X-Edge-Location": context.geo?.city || "unknown",
  };

  if (request.method === "OPTIONS") {
    return new Response(null, { status: 204, headers });
  }

  if (request.method !== "POST") {
    return new Response(
      JSON.stringify({ error: "Method not allowed" }),
      { status: 405, headers }
    );
  }

  try {
    const body = await request.json();
    const text = body.text || "";

    if (!text.trim()) {
      return new Response(
        JSON.stringify({ error: "Text is required" }),
        { status: 400, headers }
      );
    }

    const result = predictSentiment(text);

    // Log สำหรับ Monitoring
    console.log(JSON.stringify({
      type: "prediction",
      ...result,
      geo: context.geo,
      timestamp: new Date().toISOString(),
    }));

    return new Response(JSON.stringify(result), { status: 200, headers });
  } catch (error) {
    return new Response(
      JSON.stringify({ error: "Internal server error" }),
      { status: 500, headers }
    );
  }
};

export const config = { path: "/api/predict" };

A/B Testing สำหรับ Model Versions

// netlify/edge-functions/ab-test.ts
// A/B Testing ระหว่าง Model Versions
import { Context } from "https://edge.netlify.com";

const MODEL_ENDPOINTS = {
  v1: "https://ml-api.company.com/v1/predict",
  v2: "https://ml-api.company.com/v2/predict",
};

function getModelVersion(request: Request): string {
  // ใช้ Cookie เพื่อให้ User เดิมได้ Model Version เดิม (Sticky Session)
  const cookie = request.headers.get("cookie") || "";
  const match = cookie.match(/model_version=(v\d+)/);
  if (match) return match[1];

  // Traffic Split: 20% ไป v2, 80% ไป v1
  const splitRatio = parseFloat(
    Deno.env.get("AB_TEST_TRAFFIC_SPLIT") || "0.2"
  );
  return Math.random() < splitRatio ? "v2" : "v1";
}

export default async (request: Request, context: Context) => {
  const version = getModelVersion(request);
  const endpoint = MODEL_ENDPOINTS[version as keyof typeof MODEL_ENDPOINTS];

  const startTime = performance.now();

  try {
    const body = await request.json();
    const response = await fetch(endpoint, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify(body),
    });

    const result = await response.json();
    const latency = performance.now() - startTime;

    // Log A/B Test Result
    console.log(JSON.stringify({
      type: "ab_test",
      model_version: version,
      latency_ms: Math.round(latency),
      status: response.status,
      geo: context.geo?.country,
      timestamp: new Date().toISOString(),
    }));

    return new Response(
      JSON.stringify({ ...result, model_version: version, latency_ms: latency }),
      {
        headers: {
          "Content-Type": "application/json",
          "Set-Cookie": `model_version=; Path=/; Max-Age=86400`,
          "X-Model-Version": version,
        },
      }
    );
  } catch (error) {
    // Fallback ไป v1 เมื่อ v2 ล้มเหลว
    return new Response(
      JSON.stringify({ error: "Model inference failed", fallback: "v1" }),
      { status: 502, headers: { "Content-Type": "application/json" } }
    );
  }
};

export const config = { path: "/api/recommend" };

CI/CD Pipeline สำหรับ MLOps บน Netlify

# .github/workflows/mlops-deploy.yml
name: MLOps Deploy to Netlify Edge
on:
  push:
    branches: [main]
    paths:
      - 'models/**'
      - 'netlify/**'
      - 'scripts/**'

jobs:
  validate-model:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          lfs: true

      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install Dependencies
        run: pip install -r scripts/requirements.txt

      - name: Validate Model
        run: |
          python scripts/validate_model.py models/sentiment-v2.onnx
          echo "Model validation passed"

      - name: Run Model Tests
        run: pytest scripts/tests/ -v

  deploy:
    needs: validate-model
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          lfs: true

      - name: Deploy to Netlify
        uses: nwtgck/actions-netlify@v3
        with:
          publish-dir: './src'
          production-deploy: true
          netlify-config-path: './netlify.toml'
        env:
          NETLIFY_AUTH_TOKEN: }
          NETLIFY_SITE_ID: }

      - name: Smoke Test
        run: |
          sleep 10
          RESULT=$(curl -s -X POST https://ml-edge-app.netlify.app/api/predict \
            -H "Content-Type: application/json" \
            -d '{"text":"สินค้าดีมากครับ"}')
          echo "Smoke test result: "
          echo "" | jq -e '.sentiment == "positive"'

Monitoring และ Observability

การ Monitor ML Model ที่ Deploy บน Edge ต้องติดตามทั้ง Infrastructure Metrics และ Model Performance Metrics

• Latency Distribution: ดู P50, P90, P99 ของ Inference Time แยกตาม Edge Location
• Prediction Distribution: ตรวจสอบว่าสัดส่วน Positive/Negative ไม่เปลี่ยนผิดปกติ (Data Drift)
• Error Rate: ติดตาม 4xx/5xx Errors แยกตาม Model Version
• A/B Test Metrics: เปรียบเทียบ Conversion Rate, Click-through Rate ระหว่าง Model Versions
• Edge Function Invocations: ดูจำนวนการเรียกใช้งานต่อวันเพื่อ Capacity Planning
• Model Freshness: ตรวจสอบว่า Model Version ที่ Deploy ตรงกับ Latest Version ใน Registry

Netlify Edge Functions คืออะไรและต่างจาก Serverless Functions อย่างไร

Netlify Edge Functions ทำงานบน Deno Runtime ที่ Edge Server ทั่วโลก ใกล้ผู้ใช้มากกว่า Serverless Functions ที่ทำงานบน Region เดียว ทำให้มี Latency ต่ำกว่า 50ms แทนที่จะเป็น 200-500ms เหมาะสำหรับ Personalization, A/B Testing และ ML Inference ที่ต้องการ Response เร็ว แต่มี Resource จำกัดกว่า (Memory, Execution Time)

สรุปและแนวทางปฏิบัติ

Netlify Edge Functions เปิดโอกาสให้ทำ ML Inference ที่ Edge ได้โดยตรง ลด Latency และเพิ่มประสบการณ์ผู้ใช้ การสร้าง MLOps Workflow ที่ครบวงจรบน Netlify ต้องมี Model Training Pipeline ที่ Reproducible, CI/CD ที่ Validate Model ก่อน Deploy, A/B Testing สำหรับเปรียบเทียบ Model Versions และ Monitoring ที่ตรวจจับ Data Drift และ Performance Degradation การเก็บ Model ใน Git พร้อม Version Tag ทำให้สามารถ Rollback ไป Version ก่อนหน้าได้ทันทีเมื่อพบปัญหา