Directus CMS MLOps Workflow

Directus MLOps

Directus CMS MLOps Workflow Headless CMS REST GraphQL Flow Automation Webhook PostgreSQL Docker ML Pipeline Training Data Model Management

Setup & Collections

# === Directus MLOps Setup ===

# docker-compose.yml
# version: '3'
# services:
#   directus:
#     image: directus/directus:10
#     ports: ['8055:8055']
#     environment:
#       KEY: 'random-key-here'
#       SECRET: 'random-secret-here'
#       DB_CLIENT: 'pg'
#       DB_HOST: 'postgres'
#       DB_PORT: '5432'
#       DB_DATABASE: 'directus'
#       DB_USER: 'directus'
#       DB_PASSWORD: 'directus'
#       ADMIN_EMAIL: 'admin@example.com'
#       ADMIN_PASSWORD: 'admin123'
#     depends_on: [postgres]
#   postgres:
#     image: postgres:16
#     environment:
#       POSTGRES_DB: directus
#       POSTGRES_USER: directus
#       POSTGRES_PASSWORD: directus
#     volumes: ['pgdata:/var/lib/postgresql/data']
# volumes:
#   pgdata:

from dataclasses import dataclass

@dataclass
class DirectusCollection:
    name: str
    purpose: str
    key_fields: str
    access: str

collections = [
    DirectusCollection("training_data",
        "เก็บข้อมูล Training สำหรับ ML Model",
        "id, text, label, source, annotated_by, quality_score, created_at",
        "Data Scientist: CRUD, Editor: Read"),
    DirectusCollection("models",
        "เก็บ Model Metadata Version Metrics",
        "id, name, version, framework, accuracy, f1_score, status, artifact_url",
        "ML Engineer: CRUD, Data Scientist: Read"),
    DirectusCollection("experiments",
        "เก็บ Experiment Config และ Results",
        "id, model_id, hyperparams (JSON), metrics (JSON), status, duration",
        "Data Scientist: CRUD"),
    DirectusCollection("predictions",
        "เก็บผลลัพธ์ Prediction",
        "id, model_id, input_data, prediction, confidence, created_at",
        "ML Service: Create, Analyst: Read"),
    DirectusCollection("pipeline_runs",
        "เก็บ Log การรัน Pipeline",
        "id, pipeline_name, status, started_at, completed_at, error_message",
        "ML Engineer: CRUD, All: Read"),
]

print("=== Directus Collections ===")
for c in collections:
    print(f"\n  [{c.name}] {c.purpose}")
    print(f"    Fields: {c.key_fields}")
    print(f"    Access: {c.access}")

Flow Automation

# === Directus Flow Automation for MLOps ===

@dataclass
class DirectusFlow:
    name: str
    trigger: str
    steps: str
    outcome: str

flows = [
    DirectusFlow("Auto-Label Content",
        "Trigger: training_data item created",
        "1. Get item data\n"
        "2. POST to NLP API /predict\n"
        "3. Update item with predicted label + confidence\n"
        "4. If confidence < 0.8, set status='needs_review'",
        "ข้อมูลใหม่ได้ Label อัตโนมัติ ถ้าไม่มั่นใจส่ง Review"),
    DirectusFlow("Trigger Retrain",
        "Trigger: training_data count > threshold (Schedule hourly)",
        "1. Count new labeled data since last train\n"
        "2. If count > 100, POST to ML Pipeline /retrain\n"
        "3. Create pipeline_runs record\n"
        "4. Send Slack notification",
        "Retrain Model อัตโนมัติเมื่อมีข้อมูลใหม่พอ"),
    DirectusFlow("Model Deployment",
        "Trigger: models item updated (status='approved')",
        "1. Get model artifact_url\n"
        "2. POST to Deploy API /deploy\n"
        "3. Update model status='deployed'\n"
        "4. Create pipeline_runs record\n"
        "5. Send Slack + Email notification",
        "Deploy Model อัตโนมัติเมื่อได้รับ Approve"),
    DirectusFlow("Performance Monitor",
        "Trigger: Schedule (every 6 hours)",
        "1. GET /metrics from ML Service\n"
        "2. Update models collection with latest metrics\n"
        "3. If accuracy drop > 5%, alert Slack\n"
        "4. If accuracy drop > 10%, trigger retrain",
        "Monitor Model Performance แจ้งเตือนเมื่อ Degradation"),
]

print("=== Directus Flows ===")
for f in flows:
    print(f"\n  [{f.name}]")
    print(f"    Trigger: {f.trigger}")
    print(f"    Steps: {f.steps}")
    print(f"    Outcome: {f.outcome}")

API Integration

# === Directus SDK + ML API Integration ===

# import { createDirectus, rest, readItems, createItem } from '@directus/sdk';
#
# const client = createDirectus('http://localhost:8055')
#   .with(rest({ credentials: 'include' }));
#
# // Get training data for ML pipeline
# const trainingData = await client.request(
#   readItems('training_data', {
#     filter: { label: { _nnull: true }, quality_score: { _gte: 0.8 } },
#     limit: -1,
#     fields: ['text', 'label', 'metadata']
#   })
# );
#
# // Save prediction results back to Directus
# await client.request(
#   createItem('predictions', {
#     model_id: 'model-v2.1',
#     input_data: inputText,
#     prediction: result.label,
#     confidence: result.score
#   })
# );

@dataclass
class APIEndpoint:
    method: str
    path: str
    purpose: str
    ml_use: str

endpoints = [
    APIEndpoint("GET", "/items/training_data?filter[label][_nnull]=true",
        "ดึง Labeled Training Data",
        "Feed ML Training Pipeline"),
    APIEndpoint("POST", "/items/training_data",
        "เพิ่ม Training Data ใหม่",
        "Data Ingestion จาก Source ต่างๆ"),
    APIEndpoint("PATCH", "/items/models/{id}",
        "อัพเดท Model Metrics",
        "บันทึก Accuracy F1 หลัง Training"),
    APIEndpoint("POST", "/items/predictions",
        "บันทึก Prediction Results",
        "ML Service เก็บผลลัพธ์"),
    APIEndpoint("GET", "/items/models?filter[status]=deployed",
        "ดู Model ที่ Deploy อยู่",
        "Dashboard แสดง Active Models"),
    APIEndpoint("POST", "/flows/trigger/{flow_id}",
        "Trigger Flow ด้วยมือ",
        "Manual Retrain หรือ Deploy"),
]

print("=== API Endpoints ===")
for e in endpoints:
    print(f"  [{e.method}] {e.path}")
    print(f"    Purpose: {e.purpose}")
    print(f"    ML Use: {e.ml_use}")

เคล็ดลับ

• Flow: ใช้ Flow Trigger ML Pipeline อัตโนมัติเมื่อข้อมูลเปลี่ยน
• Role: แยก Role ละเอียด Editor ไม่เห็น Model Config
• Webhook: ใช้ Webhook สำหรับ Real-time Integration
• SDK: ใช้ Directus SDK แทน Raw API เขียนง่ายกว่า
• Docker: Deploy ด้วย Docker Compose ง่าย Reproduce ได้

Directus คืออะไร

Open Source Headless CMS PostgreSQL REST GraphQL Auto API Collection GUI Admin Flow Automation Webhook Role-based Docker SDK

สรุป

Directus CMS MLOps Workflow Headless CMS REST GraphQL Flow Automation Webhook Training Data Model Pipeline Docker Role SDK Production