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 ในการคิดและตัดสินใจ

เนื้อหาเกี่ยวข้อง — A/B Testing ML Micro-segmentation

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

เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน Segment Routing Data Pipeline ETL

การติดตั้งและตั้งค่า 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 ได้อัตโนมัติ

แนะนำเพิ่มเติม — เรียนเทรดกับ iCafeForex

เนื้อหาเกี่ยวข้อง — swift code tmb คือ — ข้อมูลครบถ้วน 2026