Directus CMS MLOps Workflow — จัดการ Content และ ML Pipeline ด้วย Directus

Directus MLOps

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

Component	Directus Feature	MLOps Role	Example
Data Management	Collections + Fields	Training Data Labels	training_data collection
API Layer	REST + GraphQL Auto	ML Pipeline Data Access	GET /items/training_data
Automation	Flows (Visual)	Pipeline Trigger	New data → Retrain model
Integration	Webhook + HTTP Action	ML API Communication	POST to ML Service
Access Control	Role-based (Field level)	Team Separation	Editor vs Data Scientist
Storage	Files (S3/GCS/Local)	Model Artifacts	Upload model.pkl

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}")