Payload CMS กับ MLOps Workflow — วิธีสร้าง

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