Payload CMS คืออะไร
Payload CMS เป็น Headless CMS รุ่นใหม่ที่สร้างด้วย TypeScript มี Admin Panel ที่ปรับแต่งได้เต็มที่ รองรับทั้ง REST API และ GraphQL สามารถ Self-host ได้ ไม่ต้องพึ่งบริการ Cloud ของ Vendor มี Access Control ระดับ Field, Versioning, Localization และ Hooks สำหรับ Custom Logic
เมื่อนำ Payload CMS มาใช้ร่วมกับ MLOps Workflow จะได้ Content Platform ที่ชาญฉลาด สามารถจัดหมวดหมู่ Content อัตโนมัติ ทำนาย SEO Score, แนะนำ Content ที่เกี่ยวข้อง และตรวจสอบคุณภาพ Content ด้วย ML Models
Setup Payload CMS
# === Payload CMS Setup ===
# 1. สร้างโปรเจค
npx create-payload-app@latest ml-content-platform
# เลือก: blank template, TypeScript, MongoDB
# 2. โครงสร้างโปรเจค
# ml-content-platform/
# ├── src/
# │ ├── collections/
# │ │ ├── Articles.ts
# │ │ ├── Categories.ts
# │ │ ├── Media.ts
# │ │ └── MLPredictions.ts
# │ ├── hooks/
# │ │ ├── afterArticleChange.ts
# │ │ └── classifyContent.ts
# │ ├── endpoints/
# │ │ └── mlPredict.ts
# │ ├── payload.config.ts
# │ └── server.ts
# ├── ml-pipeline/
# │ ├── train.py
# │ ├── predict.py
# │ └── models/
# ├── package.json
# └── tsconfig.json
# 3. ติดตั้ง Dependencies เพิ่มเติม
npm install @payloadcms/richtext-lexical @payloadcms/db-mongodb
npm install axios node-cron
# --- payload.config.ts ---
# import { buildConfig } from 'payload/config'
# import { mongooseAdapter } from '@payloadcms/db-mongodb'
# import { lexicalEditor } from '@payloadcms/richtext-lexical'
# import { Articles } from './collections/Articles'
# import { Categories } from './collections/Categories'
# import { MLPredictions } from './collections/MLPredictions'
#
# export default buildConfig({
# editor: lexicalEditor(),
# collections: [Articles, Categories, MLPredictions],
# db: mongooseAdapter({
# url: process.env.MONGODB_URI || 'mongodb://localhost/ml-cms',
# }),
# typescript: { outputFile: 'src/payload-types.ts' },
# })
# 4. Articles Collection
# --- src/collections/Articles.ts ---
# import { CollectionConfig } from 'payload/types'
# import { classifyContent } from '../hooks/classifyContent'
#
# export const Articles: CollectionConfig = {
# slug: 'articles',
# admin: { useAsTitle: 'title' },
# access: {
# read: () => true,
# create: ({ req: { user } }) => !!user,
# update: ({ req: { user } }) => !!user,
# },
# hooks: {
# afterChange: [classifyContent],
# },
# fields: [
# { name: 'title', type: 'text', required: true },
# { name: 'content', type: 'richText' },
# { name: 'excerpt', type: 'textarea' },
# { name: 'category', type: 'relationship', relationTo: 'categories' },
# { name: 'tags', type: 'array', fields: [
# { name: 'tag', type: 'text' },
# ]},
# { name: 'status', type: 'select', options: [
# 'draft', 'review', 'published',
# ]},
# // ML-generated fields
# { name: 'mlCategory', type: 'text', admin: { readOnly: true }},
# { name: 'mlSeoScore', type: 'number', admin: { readOnly: true }},
# { name: 'mlQualityScore', type: 'number', admin: { readOnly: true }},
# { name: 'mlSuggestedTags', type: 'json', admin: { readOnly: true }},
# ],
# }
# 5. รันโปรเจค
npm run dev
# เข้า Admin: http://localhost:3000/adminML Pipeline สำหรับ Content Intelligence
# ml_pipeline.py — ML Pipeline สำหรับ Content Intelligence
# pip install scikit-learn transformers torch mlflow fastapi uvicorn
import mlflow
import mlflow.sklearn
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
import numpy as np
import json
import re
from datetime import datetime
class ContentMLPipeline:
"""ML Pipeline สำหรับ Content Intelligence"""
def __init__(self, mlflow_uri="http://localhost:5000"):
mlflow.set_tracking_uri(mlflow_uri)
self.models = {}
def train_classifier(self, texts, labels, experiment_name="content_classifier"):
"""Train Content Classifier"""
mlflow.set_experiment(experiment_name)
X_train, X_test, y_train, y_test = train_test_split(
texts, labels, test_size=0.2, random_state=42
)
with mlflow.start_run(run_name=f"classifier_{datetime.now():%Y%m%d}"):
pipeline = Pipeline([
("tfidf", TfidfVectorizer(max_features=5000, ngram_range=(1, 2))),
("clf", MultinomialNB(alpha=0.1)),
])
pipeline.fit(X_train, y_train)
y_pred = pipeline.predict(X_test)
# Metrics
report = classification_report(y_test, y_pred, output_dict=True)
mlflow.log_metric("accuracy", report["accuracy"])
mlflow.log_metric("macro_f1", report["macro avg"]["f1-score"])
# Log Model
mlflow.sklearn.log_model(pipeline, "content_classifier",
registered_model_name="content_classifier")
self.models["classifier"] = pipeline
print(f"Classifier trained: accuracy={report['accuracy']:.4f}")
return report
def predict_category(self, text):
"""ทำนายหมวดหมู่ของ Content"""
if "classifier" not in self.models:
# Load from MLflow
self.models["classifier"] = mlflow.sklearn.load_model(
"models:/content_classifier/Production"
)
prediction = self.models["classifier"].predict([text])[0]
probabilities = self.models["classifier"].predict_proba([text])[0]
confidence = float(max(probabilities))
return {"category": prediction, "confidence": confidence}
def calculate_seo_score(self, title, content, meta_description=""):
"""คำนวณ SEO Score"""
score = 0
factors = {}
# Title Length (50-60 chars optimal)
title_len = len(title)
if 50 <= title_len <= 60:
factors["title_length"] = 15
elif 30 <= title_len <= 70:
factors["title_length"] = 10
else:
factors["title_length"] = 5
score += factors["title_length"]
# Content Length
word_count = len(content.split())
if word_count >= 2000:
factors["content_length"] = 20
elif word_count >= 1000:
factors["content_length"] = 15
elif word_count >= 500:
factors["content_length"] = 10
else:
factors["content_length"] = 5
score += factors["content_length"]
# Headings
h2_count = content.lower().count("= 3 and h3_count >= 2:
factors["headings"] = 15
elif h2_count >= 2:
factors["headings"] = 10
else:
factors["headings"] = 5
score += factors["headings"]
# Meta Description
if meta_description:
meta_len = len(meta_description)
if 150 <= meta_len <= 160:
factors["meta_description"] = 15
elif 120 <= meta_len <= 170:
factors["meta_description"] = 10
else:
factors["meta_description"] = 5
else:
factors["meta_description"] = 0
score += factors["meta_description"]
# Internal Links
link_count = content.lower().count("= 3:
factors["internal_links"] = 10
elif link_count >= 1:
factors["internal_links"] = 5
else:
factors["internal_links"] = 0
score += factors["internal_links"]
# Images with Alt
img_count = content.lower().count("![]()
0 and alt_count >= img_count:
factors["images"] = 10
elif img_count > 0:
factors["images"] = 5
else:
factors["images"] = 0
score += factors["images"]
# Code blocks (technical content)
code_count = content.lower().count("") + content.lower().count("")
if code_count >= 3:
factors["code_blocks"] = 10
elif code_count >= 1:
factors["code_blocks"] = 5
else:
factors["code_blocks"] = 0
score += factors["code_blocks"]
# FAQ Section
if "faq" in content.lower():
factors["faq"] = 5
score += 5
return {"score": min(score, 100), "factors": factors,
"word_count": word_count}
def suggest_tags(self, text, top_n=5):
"""แนะนำ Tags จาก Content"""
# Simple TF-IDF based tag extraction
from sklearn.feature_extraction.text import TfidfVectorizer
vectorizer = TfidfVectorizer(max_features=100, stop_words="english",
ngram_range=(1, 2))
tfidf = vectorizer.fit_transform([text])
feature_names = vectorizer.get_feature_names_out()
scores = tfidf.toarray()[0]
# Top N tags
top_indices = scores.argsort()[-top_n:][::-1]
tags = [{"tag": feature_names[i], "score": float(scores[i])}
for i in top_indices if scores[i] > 0]
return tags
# ตัวอย่าง
# pipeline = ContentMLPipeline()
# cat = pipeline.predict_category("How to deploy Kubernetes clusters")
# seo = pipeline.calculate_seo_score("Title", "Content
...
")
# tags = pipeline.suggest_tags("machine learning deployment kubernetes")
FastAPI — ML Prediction Service
# ml_service.py — FastAPI Service สำหรับ ML Predictions
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import logging
app = FastAPI(title="Content ML Service", version="1.0")
logger = logging.getLogger(__name__)
# Initialize Pipeline
pipeline = None
@app.on_event("startup")
async def startup():
global pipeline
pipeline = ContentMLPipeline()
logger.info("ML Pipeline initialized")
class PredictRequest(BaseModel):
title: str
content: str
meta_description: Optional[str] = ""
class PredictResponse(BaseModel):
category: dict
seo_score: dict
suggested_tags: list
quality_score: float
@app.post("/predict", response_model=PredictResponse)
async def predict(req: PredictRequest):
"""ทำนาย Category, SEO Score, Tags"""
if pipeline is None:
raise HTTPException(500, "Pipeline not ready")
full_text = f"{req.title} {req.content}"
category = pipeline.predict_category(full_text)
seo = pipeline.calculate_seo_score(req.title, req.content,
req.meta_description)
tags = pipeline.suggest_tags(full_text)
# Quality Score (simple heuristic)
quality = min(100, seo["score"] * 0.5 + category["confidence"] * 50)
return PredictResponse(
category=category,
seo_score=seo,
suggested_tags=tags,
quality_score=round(quality, 1),
)
@app.get("/health")
async def health():
return {"status": "healthy", "pipeline_ready": pipeline is not None}
# Payload CMS Hook (TypeScript)
# --- src/hooks/classifyContent.ts ---
# import axios from 'axios'
# import { AfterChangeHook } from 'payload/types'
#
# export const classifyContent: AfterChangeHook = async ({
# doc, req, operation,
# }) => {
# if (operation === 'create' || operation === 'update') {
# try {
# const response = await axios.post(
# 'http://localhost:8000/predict',
# {
# title: doc.title,
# content: JSON.stringify(doc.content),
# meta_description: doc.excerpt || '',
# }
# )
# const { category, seo_score, suggested_tags, quality_score } =
# response.data
#
# await req.payload.update({
# collection: 'articles',
# id: doc.id,
# data: {
# mlCategory: category.category,
# mlSeoScore: seo_score.score,
# mlQualityScore: quality_score,
# mlSuggestedTags: suggested_tags,
# },
# })
# } catch (error) {
# console.error('ML prediction failed:', error.message)
# }
# }
# }
# รัน: uvicorn ml_service:app --host 0.0.0.0 --port 8000
Best Practices
- Async Hooks: ใช้ Payload Hooks แบบ Async เรียก ML Service ไม่ Block User
- Caching: Cache ML Predictions สำหรับ Content ที่ไม่เปลี่ยน ลด Latency
- Fallback: ถ้า ML Service ล่ม ให้ CMS ทำงานปกติ ML Fields เป็น Optional
- Model Versioning: ใช้ MLflow Registry จัดการ Model Versions แยก Staging/Production
- Batch Processing: สำหรับ Content เยอะ ใช้ Batch Prediction แทน Real-time
- Monitoring: ติดตาม Prediction Accuracy, Latency, Error Rate ตั้ง Alert
Payload CMS คืออะไร
Headless CMS แบบ Open-source สร้างด้วย TypeScript และ React มี Admin Panel ปรับแต่งได้ รองรับ REST API, GraphQL, Access Control, Versioning, Localization และ Hooks รันบน Node.js ใช้ MongoDB หรือ PostgreSQL
MLOps Workflow คืออะไร
กระบวนการจัดการ ML ตั้งแต่ Data Collection, Feature Engineering, Training, Evaluation, Deployment ถึง Monitoring แบบอัตโนมัติ ใช้เครื่องมือ MLflow, Kubeflow, Airflow, DVC
Payload CMS เกี่ยวกับ MLOps อย่างไร
Payload CMS เป็นแหล่ง Content ที่เชื่อมกับ ML Pipeline สำหรับ Content Classification อัตโนมัติ, SEO Score Prediction, Image Tagging, Content Recommendation และ Quality Check
Payload CMS ติดตั้งอย่างไร
npx create-payload-app เลือก Template และ Database ตั้งค่า Collections ใน payload.config.ts กำหนด Fields, Access Control, Hooks รัน npm run dev เข้า Admin ที่ localhost:3000/admin
สรุป
Payload CMS ร่วมกับ MLOps Workflow สร้าง Content Platform ที่ชาญฉลาด ใช้ Hooks เชื่อมกับ ML Service สำหรับ Content Classification, SEO Scoring, Tag Suggestion และ Quality Assessment ใช้ MLflow จัดการ Model Versions FastAPI เป็น Prediction Service สิ่งสำคัญคือ Async Processing, Caching, Fallback Strategy และ Model Monitoring
