CrewAI Multi-Agent MLOps Workflow

CrewAI คืออะไร

CrewAI เป็น Open-source Python Framework สำหรับสร้าง Multi-Agent AI System ที่ AI Agent หลายตัวทำงานร่วมกันเป็นทีม (Crew) คล้ายกับทีมงานจริงที่แต่ละคนมีบทบาทเฉพาะทาง แต่ละ Agent มี Role (บทบาท), Goal (เป้าหมาย), Backstory (ภูมิหลัง) และ Tools (เครื่องมือ) ที่ใช้ได้ โดยใช้ LLM เป็น Brain ในการคิดและตัดสินใจ

เมื่อนำ CrewAI มาใช้กับ MLOps Workflow จะได้ระบบที่ Agent แต่ละตัวรับผิดชอบขั้นตอนเฉพาะของ ML Pipeline ตั้งแต่วิเคราะห์ข้อมูล สร้าง Features เทรน Model ประเมินผล ไปจนถึง Deploy ทำงานร่วมกันอัตโนมัติโดยไม่ต้องมี Human-in-the-loop ทุกขั้นตอน

การติดตั้งและตั้งค่า CrewAI

# ติดตั้ง CrewAI
pip install crewai crewai-tools langchain-community

# โครงสร้าง Project
mlops-crew/
├── main.py                # Entry point
├── agents.py              # Agent definitions
├── tasks.py               # Task definitions
├── tools/                 # Custom tools
│   ├── data_tools.py
│   ├── model_tools.py
│   └── deploy_tools.py
├── config/
│   └── settings.yaml
└── requirements.txt

# requirements.txt
crewai==0.41.0
crewai-tools==0.8.0
scikit-learn==1.4.0
pandas==2.2.0
mlflow==2.10.0
boto3==1.34.0

# ตั้งค่า Environment Variables
# .env
OPENAI_API_KEY=sk-...
MLFLOW_TRACKING_URI=https://mlflow.company.com
AWS_DEFAULT_REGION=ap-southeast-1

สร้าง MLOps Agents

# agents.py — กำหนด AI Agents สำหรับ MLOps
from crewai import Agent
from crewai_tools import FileReadTool, CodeInterpreterTool
from tools.data_tools import DataQualityTool, FeatureStoreTool
from tools.model_tools import MLflowTool, ModelTrainingTool
from tools.deploy_tools import KubernetesTool

def create_data_analyst():
    """Agent สำหรับวิเคราะห์และตรวจสอบคุณภาพข้อมูล"""
    return Agent(
        role="Senior Data Analyst",
        goal="วิเคราะห์คุณภาพข้อมูลและระบุปัญหาที่ต้องแก้ไขก่อน Training",
        backstory="""คุณเป็น Data Analyst ที่มีประสบการณ์ 10 ปี
        เชี่ยวชาญการตรวจสอบ Data Quality, Missing Values,
        Distribution Drift และ Outlier Detection
        คุณจะไม่ปล่อยให้ข้อมูลคุณภาพต่ำเข้าสู่ ML Pipeline""",
        tools=[DataQualityTool(), FileReadTool(), CodeInterpreterTool()],
        verbose=True,
        allow_delegation=True,
        llm="gpt-4o",
    )

def create_feature_engineer():
    """Agent สำหรับ Feature Engineering"""
    return Agent(
        role="ML Feature Engineer",
        goal="สร้าง Features ที่มีคุณภาพสูงสำหรับ Model Training",
        backstory="""คุณเป็น Feature Engineer ที่เชี่ยวชาญ
        การสร้าง Features จากข้อมูลดิบ ทั้ง Numerical Encoding,
        Categorical Encoding, Time-based Features
        และ Feature Selection ด้วย Statistical Methods""",
        tools=[FeatureStoreTool(), CodeInterpreterTool()],
        verbose=True,
        llm="gpt-4o",
    )

def create_model_trainer():
    """Agent สำหรับ Model Training"""
    return Agent(
        role="ML Engineer",
        goal="เทรน Model ที่มี Performance ดีที่สุดและ Log ทุกอย่างใน MLflow",
        backstory="""คุณเป็น ML Engineer ที่เชี่ยวชาญ
        Hyperparameter Tuning, Cross-validation,
        และ Model Selection คุณใช้ MLflow ติดตาม
        ทุก Experiment และเปรียบเทียบ Model อย่างเป็นระบบ""",
        tools=[ModelTrainingTool(), MLflowTool(), CodeInterpreterTool()],
        verbose=True,
        llm="gpt-4o",
    )

def create_evaluator():
    """Agent สำหรับประเมิน Model"""
    return Agent(
        role="ML Quality Assurance",
        goal="ประเมิน Model อย่างละเอียดและตัดสินใจว่าพร้อม Deploy หรือไม่",
        backstory="""คุณเป็น ML QA ที่เข้มงวดมาก ตรวจสอบ
        ทั้ง Accuracy, Fairness, Bias, Robustness
        และ Performance บน Edge Cases
        คุณจะไม่อนุมัติ Model ที่ไม่ผ่านเกณฑ์""",
        tools=[MLflowTool(), CodeInterpreterTool()],
        verbose=True,
        llm="gpt-4o",
    )

def create_deployer():
    """Agent สำหรับ Deploy Model"""
    return Agent(
        role="MLOps Engineer",
        goal="Deploy Model ไป Production อย่างปลอดภัยด้วย Canary Strategy",
        backstory="""คุณเป็น MLOps Engineer ที่เชี่ยวชาญ
        Kubernetes, Docker และ CI/CD
        คุณ Deploy ด้วย Canary Strategy เสมอ
        Monitor Metrics หลัง Deploy และ Rollback ถ้าพบปัญหา""",
        tools=[KubernetesTool(), MLflowTool()],
        verbose=True,
        llm="gpt-4o-mini",  # งานไม่ซับซ้อนใช้ Model เล็กประหยัด Cost
    )

# agents.py ส่วน Custom Tools
# tools/data_tools.py
from crewai_tools import BaseTool
from pydantic import BaseModel, Field
import pandas as pd

class DataQualityInput(BaseModel):
    file_path: str = Field(description="Path to data file")

class DataQualityTool(BaseTool):
    name: str = "Data Quality Check"
    description: str = "ตรวจสอบคุณภาพข้อมูล: Missing Values, Duplicates, Outliers"
    args_schema: type[BaseModel] = DataQualityInput

    def _run(self, file_path: str) -> str:
        df = pd.read_parquet(file_path)
        report = {
            "rows": len(df),
            "columns": len(df.columns),
            "missing_pct": df.isnull().mean().to_dict(),
            "duplicates": df.duplicated().sum(),
            "dtypes": df.dtypes.astype(str).to_dict(),
        }
        # Outlier Detection (IQR Method)
        numeric_cols = df.select_dtypes(include="number").columns
        outliers = {}
        for col in numeric_cols:
            q1, q3 = df[col].quantile([0.25, 0.75])
            iqr = q3 - q1
            outlier_count = ((df[col] < q1 - 1.5*iqr) | (df[col] > q3 + 1.5*iqr)).sum()
            if outlier_count > 0:
                outliers[col] = int(outlier_count)
        report["outliers"] = outliers
        return str(report)

สร้าง Tasks และ Crew

# tasks.py — กำหนด Tasks สำหรับ MLOps Crew
from crewai import Task

def create_tasks(data_analyst, feature_engineer, model_trainer, evaluator, deployer):
    """สร้าง Tasks ที่ทำงานต่อเนื่องกัน"""

    task_analyze = Task(
        description="""วิเคราะห์ข้อมูลใน s3://data-lake/raw/churn_data.parquet
        1. ตรวจสอบ Missing Values แต่ละ Column
        2. ตรวจสอบ Duplicates
        3. ตรวจสอบ Outliers ด้วย IQR Method
        4. ตรวจสอบ Class Imbalance ของ Target Variable
        5. สรุปปัญหาที่พบและแนะนำวิธีแก้ไข""",
        expected_output="รายงานคุณภาพข้อมูลพร้อมคำแนะนำ",
        agent=data_analyst,
    )

    task_features = Task(
        description="""จากรายงานคุณภาพข้อมูล:
        1. จัดการ Missing Values ตามคำแนะนำ
        2. สร้าง Features ใหม่: RFM Features, Time-based Features
        3. Encode Categorical Variables
        4. Feature Selection ด้วย Mutual Information
        5. บันทึก Features ลง Feature Store""",
        expected_output="Feature Set พร้อม Training Report",
        agent=feature_engineer,
        context=[task_analyze],
    )

    task_train = Task(
        description="""เทรน Model ด้วย Features ที่สร้าง:
        1. ลอง 3 Algorithms: XGBoost, LightGBM, Random Forest
        2. ใช้ 5-Fold Cross-validation
        3. Hyperparameter Tuning ด้วย Optuna (50 trials)
        4. Log ทุก Experiment ใน MLflow
        5. เลือก Model ที่ดีที่สุดตาม F1-score""",
        expected_output="Best Model พร้อม MLflow Run ID",
        agent=model_trainer,
        context=[task_features],
    )

    task_evaluate = Task(
        description="""ประเมิน Model อย่างละเอียด:
        1. ตรวจสอบ Performance Metrics (F1, Precision, Recall, AUC)
        2. ตรวจสอบ Fairness ตาม Gender และ Age Group
        3. ทดสอบ Edge Cases (empty features, extreme values)
        4. เปรียบเทียบกับ Production Model ปัจจุบัน
        5. ตัดสินใจ: Approve หรือ Reject พร้อมเหตุผล""",
        expected_output="Evaluation Report พร้อม Decision (Approve/Reject)",
        agent=evaluator,
        context=[task_train],
    )

    task_deploy = Task(
        description="""ถ้า Model ได้รับ Approve:
        1. Package Model เป็น Docker Image
        2. Deploy ด้วย Canary Strategy (10% traffic)
        3. Monitor Metrics 30 นาที
        4. ถ้า Metrics ดี เพิ่ม Traffic เป็น 100%
        5. ถ้า Metrics แย่ Rollback ทันที""",
        expected_output="Deployment Status Report",
        agent=deployer,
        context=[task_evaluate],
    )

    return [task_analyze, task_features, task_train, task_evaluate, task_deploy]

# main.py — Entry Point
from crewai import Crew, Process
from agents import (create_data_analyst, create_feature_engineer,
                     create_model_trainer, create_evaluator, create_deployer)
from tasks import create_tasks

def run_mlops_crew():
    # สร้าง Agents
    data_analyst = create_data_analyst()
    feature_engineer = create_feature_engineer()
    model_trainer = create_model_trainer()
    evaluator = create_evaluator()
    deployer = create_deployer()

    # สร้าง Tasks
    tasks = create_tasks(data_analyst, feature_engineer,
                         model_trainer, evaluator, deployer)

    # สร้าง Crew
    crew = Crew(
        agents=[data_analyst, feature_engineer, model_trainer,
                evaluator, deployer],
        tasks=tasks,
        process=Process.sequential,  # ทำงานตามลำดับ
        verbose=True,
        memory=True,  # เปิด Memory ให้ Agent จำ Context
        max_rpm=30,   # Rate Limit สำหรับ LLM API
    )

    # รัน Crew
    result = crew.kickoff()
    print(f"\n{'='*60}")
    print(f"MLOps Crew Result:")
    print(f"{'='*60}")
    print(result)
    return result

if __name__ == "__main__":
    run_mlops_crew()

Monitoring และ Cost Management

• Token Usage Tracking: ติดตาม Token Usage ของแต่ละ Agent เพื่อ Optimize Cost ใช้ Model เล็กสำหรับงานง่าย
• Agent Performance: วัดเวลาที่แต่ละ Agent ใช้ จำนวน Tool Calls และคุณภาพ Output
• Pipeline Duration: ติดตามเวลารวมของ Pipeline ตั้ง Alert ถ้านานเกินไป
• Error Handling: ตั้ง Max Retries สำหรับแต่ละ Task, Fallback Strategy เมื่อ Agent ล้มเหลว
• Audit Trail: บันทึกทุก Decision ของ Agent สำหรับ Review และ Compliance

Best Practices สำหรับ CrewAI + MLOps

• แยก Agent ตาม Responsibility: แต่ละ Agent ควรมีหน้าที่เดียวชัดเจน ไม่ควรให้ Agent เดียวทำหลายอย่าง
• ใช้ LLM ที่เหมาะสม: Agent ที่ต้อง Reasoning ซับซ้อน ใช้ GPT-4o Agent ง่ายใช้ GPT-4o-mini ประหยัด Cost
• เขียน Backstory ให้ดี: Backstory ที่ละเอียดช่วยให้ Agent ทำงานได้ดีขึ้น เหมือนให้ Context กับคนจริง
• ใช้ Tools แทน LLM: งานที่ทำได้ด้วย Code (Data Processing, API Call) ให้ใช้ Tools แทนการให้ LLM คิดเอง
• ตั้ง Guardrails: กำหนด expected_output ให้ชัดเจน ตั้ง max_iterations ป้องกัน Loop
• ทดสอบแต่ละ Agent แยก: ทดสอบแต่ละ Agent ทำงานถูกต้องก่อนรวมเป็น Crew

CrewAI คืออะไร

CrewAI เป็น Python Framework สำหรับสร้าง Multi-Agent System ที่ AI Agent หลายตัวทำงานร่วมกันเป็นทีม แต่ละ Agent มี Role, Goal, Backstory และ Tools เฉพาะ ใช้ LLM เป็น Brain สามารถส่งต่องานกันและ Delegate Tasks ได้อัตโนมัติ

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

CrewAI เป็น Framework ที่เหมาะสำหรับสร้าง Multi-Agent System ที่ทำงานร่วมกันใน MLOps Pipeline การแบ่ง Agent ตาม Responsibility ที่ชัดเจน ใช้ LLM ที่เหมาะสมกับความซับซ้อนของงาน เขียน Backstory ที่ละเอียด และใช้ Tools แทน LLM สำหรับงาน Deterministic จะทำให้ได้ MLOps Pipeline ที่ Intelligent, Automated และ Cost-effective สิ่งสำคัญคือทดสอบแต่ละ Agent แยกก่อน Monitor Token Usage และมี Guardrails ป้องกัน Unexpected Behavior