บทนำ: ทำไม Task State Management ถึงสำคัญ

ในโปรเจกต์อีคอมเมิร์ซที่ผมพัฒนาเมื่อปีที่แล้ว ระบบ AI ลูกค้าสัมพันธ์ต้องจัดการคำสั่งซื้อที่ซับซ้อน โดยต้องให้ AI Agent 3 ตัวทำงานร่วมกัน ได้แก่ Order Agent สำหรับตรวจสอบคำสั่งซื้อ, Inventory Agent สำหรับเช็คสต็อกสินค้า และ Shipping Agent สำหรับจัดการจัดส่ง ปัญหาที่พบคือเมื่อ Task ตัวหนึ่งล้มเหลว ทั้งระบบจะค้างโดยไม่มีการ retry หรือ fallback ใดๆ ซึ่งทำให้ลูกค้ารอนานกว่า 30 วินาทีและต้องโทรมาที่ call center ในที่สุด บทความนี้จะสอนวิธีใช้งาน CrewAI Task States อย่างถูกต้อง โดยใช้ HolySheep AI เป็น backend ที่มี latency ต่ำกว่า 50ms รองรับหลายโมเดล AI อย่าง GPT-4.1, Claude Sonnet 4.5 และ DeepSeek V3.2 ในราคาที่ประหยัดกว่าถึง 85% เมื่อเทียบกับการใช้งาน OpenAI โดยตรง 
# ติดตั้ง dependencies ที่จำเป็น
pip install crewai crewai-tools langchain-openai langchain-anthropic


สำหรับ HolySheep AI ใช้ LangChain แบบ custom provider


ไม่ต้องติดตั้ง package เพิ่มเติม เพราะใช้ OpenAI-compatible API

Task States พื้นฐานใน CrewAI

CrewAI กำหนด Task states หลัก 6 สถานะที่เราต้องเข้าใจ:

โครงสร้างโปรเจกต์ระบบ AI ลูกค้าสัมพันธ์อีคอมเมิร์ซ

โปรเจกต์นี้ใช้สถาปัตยกรรม 3-tier ที่ประกอบด้วย Order Crew (ตรวจสอบคำสั่งซื้อ), Inventory Crew (เช็คและจองสต็อก), และ Shipping Crew (จัดการจัดส่ง) โดยแต่ละ Crew จะมีหลาย Agents และ Tasks ที่ทำงานประสานกันผ่าน state machine 
import os
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from datetime import datetime
import json


ตั้งค่า HolySheep AI เป็น LLM provider


สมัครได้ที่ https://www.holysheep.ai/register รับเครดิตฟรีเมื่อลงทะเบียน

os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"


ใช้โมเดลที่เหมาะสมกับงาน


GPT-4.1: $8/MTok — งานที่ต้องการความแม่นยำสูง


Claude Sonnet 4.5: $15/MTok — งานวิเคราะห์ข้อความซับซ้อน


DeepSeek V3.2: $0.42/MTok — งาน routine ที่ต้องการประหยัดต้นทุน


llm_gpt4 = ChatOpenAI(
    model="gpt-4.1",
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_API_BASE"]
)

llm_sonnet = ChatOpenAI(
    model="claude-sonnet-4.5",
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_API_BASE"]
)

llm_deepseek = ChatOpenAI(
    model="deepseek-v3.2",
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_API_BASE"]
)

class EcommerceOrderState:
    """State machine สำหรับจัดการ order lifecycle"""
    
    def __init__(self, order_id: str):
        self.order_id = order_id
        self.order_status = "PENDING"
        self.inventory_status = "PENDING"
        self.shipping_status = "PENDING"
        self.created_at = datetime.now()
        self.updated_at = datetime.now()
        self.error_history = []
        self.retry_count = 0
        self.max_retries = 3
    
    def update_state(self, component: str, status: str, error: str = None):
        """อัพเดทสถานะของแต่ละ component"""
        self.updated_at = datetime.now()
        
        if component == "order":
            self.order_status = status
        elif component == "inventory":
            self.inventory_status = status
        elif component == "shipping":
            self.shipping_status = status
        
        if error:
            self.error_history.append({
                "component": component,
                "error": error,
                "timestamp": self.updated_at.isoformat()
            })
        
        if status == "FAILED":
            self.retry_count += 1
    
    def is_terminal_state(self) -> bool:
        """ตรวจสอบว่าถึงสถานะสุดท้ายหรือยัง"""
        return self.order_status in ["COMPLETED", "FAILED"] and \
               self.inventory_status in ["COMPLETED", "FAILED", "SKIPPED"] and \
               self.shipping_status in ["COMPLETED", "FAILED", "SKIPPED"]
    
    def should_retry(self) -> bool:
        """ตรวจสอบว่าควร retry หรือไม่"""
        return self.retry_count < self.max_retries
    
    def to_dict(self) -> dict:
        return {
            "order_id": self.order_id,
            "order_status": self.order_status,
            "inventory_status": self.inventory_status,
            "shipping_status": self.shipping_status,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat(),
            "error_count": len(self.error_history),
            "retry_count": self.retry_count
        }


สร้าง state instance

order_state = EcommerceOrderState("ORD-2024-001")
print(f"Initial state: {json.dumps(order_state.to_dict(), indent=2)}")

การสร้าง Multi-Agent Crews พร้อม State Management

ในระบบจริง เราต้องสร้าง Agents หลายตัวที่ทำงานประสานกัน โดยแต่ละ Task จะมี callback สำหรับอัพเดทสถานะ ซึ่งทำให้เราติดตาม progress และจัดการ error ได้อย่างมีประสิทธิภาพ 
from crewai.callbacks import TaskExecutionCallback
from typing import Optional

class StateCallback(TaskExecutionCallback):
    """Callback สำหรับอัพเดท state หลังจาก task execution"""
    
    def __init__(self, state: EcommerceOrderState, component: str):
        self.state = state
        self.component = component
    
    def on_task_start(self, task: Task, agent: Agent):
        self.state.update_state(self.component, "IN_PROGRESS")
        print(f"[{self.component}] Started: {task.description[:50]}...")
    
    def on_task_end(self, task: Task, agent: Agent, output: str):
        if "error" in output.lower() or "fail" in output.lower():
            self.state.update_state(self.component, "FAILED", output)
            print(f"[{self.component}] Failed: {output[:100]}")
        else:
            self.state.update_state(self.component, "COMPLETED")
            print(f"[{self.component}] Completed successfully")


กำหนด Order Agent — ใช้ GPT-4.1 สำหรับความแม่นยำสูง

order_agent = Agent(
    role="Order Verification Specialist",
    goal="ตรวจสอบคำสั่งซื้อและยืนยันข้อมูลลูกค้าอย่างถูกต้อง",
    backstory="คุณเป็นผู้เชี่ยวชาญด้านการจัดการคำสั่งซื้อ มีประสบการณ์ 10 ปี "
              "ในอุตสาหกรรมอีคอมเมิร์ซ คุณตรวจสอบความถูกต้องของข้อมูล "
              "และความสมบูรณ์ของการสั่งซื้อทุกรายการ",
    llm=llm_gpt4,
    verbose=True
)


กำหนด Inventory Agent — ใช้ DeepSeek V3.2 สำหรับงาน routine

inventory_agent = Agent(
    role="Inventory Checker",
    goal="ตรวจสอบและจองสต็อกสินค้าในคลังอย่างรวดเร็วและแม่นยำ",
    backstory="คุณเป็นผู้ดูแลคลังสินค้าอัจฉริยะ คุณเช็คจำนวนสต็อก "
              "จองสินค้าเมื่อมีคำสั่งซื้อ และอัพเดทระบบคลังแบบ real-time",
    llm=llm_deepseek,  # ประหยัดต้นทุนสำหรับงานนี้
    verbose=True
)


กำหนด Shipping Agent — ใช้ Claude Sonnet 4.5 สำหรับการวิเคราะห์

shipping_agent = Agent(
    role="Shipping Coordinator",
    goal="จัดการการจัดส่งและแจ้งติดตามพัสดุให้ลูกค้า",
    backstory="คุณเป็นผู้ประสานงานจัดส่งที่มีประสบการณ์ คุณเลือกเส้นทางที่ดีที่สุด "
              "ติดตามพัสดุ และสื่อสารกับลูกค้าอย่างชัดเจน",
    llm=llm_sonnet,
    verbose=True
)

def create_order_tasks(order_state: EcommerceOrderState):
    """สร้าง tasks พร้อม state callback"""
    
    # Task 1: ตรวจสอบคำสั่งซื้อ
    verify_order_task = Task(
        description="""
        ตรวจสอบคำสั่งซื้อหมายเลข {order_id} ดังนี้:
        1. ยืนยันข้อมูลลูกค้า (ชื่อ, ที่อยู่, เบอร์โทร)
        2. ตรวจสอบวิธีการชำระเงิน
        3. คำนวณยอดรวมและส่วนลด
        4. บันทึกผลการตรวจสอบในรูปแบบ JSON
        """.format(order_id=order_state.order_id),
        agent=order_agent,
        expected_output="รายงานการตรวจสอบคำสั่งซื้อพร้อมสถานะ APPROVED หรือ REJECTED",
        callback=StateCallback(order_state, "order")
    )
    
    # Task 2: ตรวจสอบสต็อก (ต้องรอ Task 1 เสร็จก่อน)
    check_inventory_task = Task(
        description="""
        หลังจากคำสั่งซื้อได้รับการอนุมัติ:
        1. เช็คจำนวนสินค้าในคลังสำหรับสินค้าที่สั่งซื้อ
        2. จองสินค้าชั่วคราว (hold) ระหว่างดำเนินการ
        3. หากสินค้าไม่พอ ให้แนะนำสินค้าทดแทน
        4. อัพเดทสถานะสต็อกในระบบ
        """,
        agent=inventory_agent,
        expected_output="รายงานสถานะสต็อกพร้อมรายการสินค้าที่จองได้",
        callback=StateCallback(order_state, "inventory")
    )
    
    # Task 3: จัดการจัดส่ง (ต้องรอ Task 2 เสร็จก่อน)
    arrange_shipping_task = Task(
        description="""
        หลังจากตรวจสอบสต็อกเสร็จ:
        1. เลือกบริการขนส่งที่เหมาะสม ( Kerry, Flash, SCG )
        2. คำนวณค่าจัดส่งตามน้ำหนักและระยะทาง
        3. สร้าง label จัดส่งและ tracking number
        4. กำหนดวันจัดส่งโดยประมาณ
        """,
        agent=shipping_agent,
        expected_output="รายละเอียดการจัดส่งพร้อม tracking number",
        callback=StateCallback(order_state, "shipping")
    )
    
    # กำหนด dependencies
    check_inventory_task.context = [verify_order_task]
    arrange_shipping_task.context = [check_inventory_task]
    
    return [verify_order_task, check_inventory_task, arrange_shipping_task]


สร้าง tasks

tasks = create_order_tasks(order_state)
print(f"สร้าง tasks สำเร็จ: {len(tasks)} tasks")

การรัน Crew พร้อม Error Handling และ Retry Logic

ส่วนสำคัญที่สุดคือการรัน crew ด้วยการจัดการ error ที่เหมาะสม ในโค้ดนี้ผมใช้ retry logic ที่จะพยายามซ้ำสูงสุด 3 ครั้งก่อนจะ skip task หรือ fail ทั้ง crew 
import time
from crewai import CrewOutput

class CrewRunner:
    """Class สำหรับรัน crew พร้อม state management และ error handling"""
    
    def __init__(self, state: EcommerceOrderState, tasks: list):
        self.state = state
        self.tasks = tasks
        self.max_retries = 3
        self.retry_delay = 2  # วินาที
    
    def execute_with_retry(self, crew: Crew) -> dict:
        """รัน crew พร้อม retry logic"""
        attempt = 0
        
        while attempt < self.max_retries:
            try:
                print(f"\n{'='*50}")
                print(f"🚀 เริ่ม execute crew (attempt {attempt + 1}/{self.max_retries})")
                print(f"{'='*50}")
                
                result = crew.kickoff()
                
                # ตรวจสอบผลลัพธ์
                if self._is_success(result):
                    print("\n✅ Crew execution completed successfully!")
                    return {
                        "status": "success",
                        "result": result,
                        "attempts": attempt + 1,
                        "final_state": self.state.to_dict()
                    }
                else:
                    raise Exception(f"Crew execution failed: {result}")
                    
            except Exception as e:
                attempt += 1
                print(f"\n❌ Attempt {attempt} failed: {str(e)}")
                
                if attempt < self.max_retries:
                    print(f"⏳ Retrying in {self.retry_delay} seconds...")
                    time.sleep(self.retry_delay)
                    
                    # Reset failed tasks เพื่อ retry
                    for task in self.tasks:
                        if task.callback and task.callback.component:
                            if self.state.order_status == "FAILED":
                                self.state.order_status = "PENDING"
                            if self.state.inventory_status == "FAILED":
                                self.state.inventory_status = "PENDING"
                            if self.state.shipping_status == "FAILED":
                                self.state.shipping_status = "PENDING"
                else:
                    print("\n🚫 Max retries reached. Crew failed.")
                    return {
                        "status": "failed",
                        "error": str(e),
                        "attempts": attempt,
                        "final_state": self.state.to_dict()
                    }
        
        return self.state.to_dict()
    
    def _is_success(self, result) -> bool:
        """ตรวจสอบว่าผลลัพธ์สำเร็จหรือไม่"""
        if isinstance(result, CrewOutput):
            return result.raw and len(result.raw) > 0
        return result is not None


สร้าง crew

ecommerce_crew = Crew(
    agents=[order_agent, inventory_agent, shipping_agent],
    tasks=tasks,
    process=Process.hierarchical,  # ใช้ hierarchical process สำหรับ dependency
    manager_llm=llm_gpt4,  # Manager agent ใช้ GPT-4.1
    verbose=True
)


รัน crew

print("📦 เริ่มระบบจัดการคำสั่งซื้ออีคอมเมิร์ซด้วย CrewAI")
print(f"📋 Order ID: {order_state.order_id}")

runner = CrewRunner(order_state, tasks)
result = runner.execute_with_retry(ecommerce_crew)

print("\n" + "="*50)
print("📊 สรุปผลการดำเนินงาน")
print("="*50)
print(f"สถานะ: {result['status']}")
print(f"จำนวนครั้งที่ลอง: {result.get('attempts', 0)}")
print(f"สถานะสุดท้าย: {json.dumps(result['final_state'], indent=2, ensure_ascii=False)}")

การใช้งาน Task Output และ Context Sharing

ในบางกรณีเราต้องการให้ output จาก task หนึ่งไปเป็น input ของ task ถัดไป CrewAI รองรับสิ่งนี้ผ่าน context ซึ่งเราสามารถดึง output จาก task ก่อนหน้ามาใช้ได้ 
from crewai.tools import BaseTool
from pydantic import BaseModel

class OrderDatabaseTool(BaseTool):
    name: str = "Order Database Lookup"
    description: str = "ค้นหาข้อมูลคำสั่งซื้อจากฐานข้อมูล"
    
    class Schema(BaseModel):
        order_id: str
        include_items: bool = True
    
    def _run(self, order_id: str, include_items: bool = True):
        # Mock database lookup - ในระบบจริงจะ query จาก DB
        return {
            "order_id": order_id,
            "customer_name": "สมชาย ใจดี",
            "customer_phone": "081-234-5678",
            "shipping_address": "123 ถนนสุขุมวิท แขวงคลองเตย เขตคลองเตย กรุงเทพฯ 10110",
            "payment_method": "credit_card",
            "total_amount": 4590.00,
            "items": [
                {"sku": "SKU-001", "name": "เสื้อยืด Cotton", "qty": 2, "price": 890},
                {"sku": "SKU-002", "name": "กางเกงยีนส์ Slim", "qty": 1, "price": 1990},
                {"sku": "SKU-003", "name": "รองเท้าผ้าใบ", "qty": 1, "price": 920}
            ]
        }

class InventoryLookupTool(BaseTool):
    name: str = "Inventory System Lookup"
    description: str = "ตรวจสอบสต็อกสินค้าคงคลัง"
    
    def _run(self, sku: str):
        # Mock inventory lookup
        inventory = {
            "SKU-001": {"available": 45, "reserved": 5, "warehouse": "BKK-01"},
            "SKU-002": {"available": 12, "reserved": 2, "warehouse": "BKK-01"},
            "SKU-003": {"available": 0, "reserved": 0, "warehouse": "BKK-01"},  # สินค้าหมด!
        }
        return inventory.get(sku, {"available": 0, "reserved": 0, "warehouse": "UNKNOWN"})


สร้าง tools instances

order_db_tool = OrderDatabaseTool()
inventory_tool = InventoryLookupTool()


เพิ่ม tools ให้กับ agents

order_agent.tools = [order_db_tool]
inventory_agent.tools = [inventory_tool]


ดึง output จาก task ก่อนหน้า

def get_context_from_previous_tasks(task: Task) -> str:
    """ดึง output จาก tasks ที่อยู่ใน context"""
    context_parts = []
    
    if hasattr(task, 'context') and task.context:
        for prev_task in task.context:
            if hasattr(prev_task, 'output') and prev_task.output:
                context_parts.append(f"จาก task ก่อนหน้า: {prev_task.output}")
    
    return "\n".join(context_parts) if context_parts else "ไม่มีข้อมูลจาก task ก่อนหน้า"

print("✅ Tools และ Context sharing พร้อมใช้งาน")

การใช้งาน Memory และ Context ใน Multi-Agent

สำหรับระบบ RAG ขององค์กรที่ต้องการ context ยาวๆ เราสามารถใช้ CrewAI memory เพื่อให้ agents จดจำข้อมูลจาก task ก่อนหน้าได้ ซึ่งช่วยลด hallucination และเพิ่มความแม่นยำในการตอบคำถาม 
from crewai.memory import Memory, ShortTermMemory, LongTermMemory
from crewai.memory.storage import RAGStorage
from langchain_openai import OpenAIEmbeddings


ตั้งค่า Memory สำหรับ Crew

memory = Memory(
    short_term=ShortTermMemory(
        storage=RAGStorage(
            type="short_term",
            vector_dimension=1536,
            embedder=OpenAIEmbeddings(
                model="text-embedding-3-small",
                api_key=os.environ["OPENAI_API_KEY"],
                base_url=os.environ["OPENAI_API_BASE"]
            )
        )
    ),
    long_term=LongTermMemory(
        storage=RAGStorage(
            type="long_term",
            vector_dimension=1536,
            embedder=OpenAIEmbeddings(
                model="text-embedding-3-small",
                api_key=os.environ["OPENAI_API_KEY"],
                base_url=os.environ["OPENAI_API_BASE"]
            )
        )
    )
)


สร้าง crew พร้อม memory

ecom_crew_with_memory = Crew(
    agents=[order_agent, inventory_agent, shipping_agent],
    tasks=tasks,
    process=Process.hierarchical,
    manager_llm=llm_gpt4,
    memory=memory,  # เพิ่ม memory ให้ crew
    verbose=True
)


ตัวอย่างการค้นหาจาก memory

def search_similar_orders(customer_id: str, limit: int = 5):
    """ค้นหาคำสั่งซื้อที่คล้ายกันจาก memory"""
    query = f"คำสั่งซื้อของลูกค้า {customer_id}"
    
    results = memory.long_term.search(
        query=query,
        limit=limit
    )
    
    return results

print("✅ Memory system พร้อมใช้งานสำหรับ RAG")

การ Monitor และ Logging Task States

ในระบบ production การ monitor task states เป็นสิ่งจำเป็น โค้ดนี้แสดงวิธีสร้าง logging system ที่จะช่วยให้เราติดตามสถานะของ tasks ทั้งหมดได้ 
import logging
from datetime import datetime
from typing import Dict, List
import threading


ตั้งค่า logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("CrewAI-Monitor")

class TaskStateMonitor:
    """Monitor สำหรับติดตามสถานะ tasks ทั้งหมด"""
    
    def __init__(self):
        self.tasks_state: Dict[str, str] = {}
        self.state_history: List[Dict] = []
        self.lock = threading.Lock()
    
    def update_state(self, task_id: str, new_state: str, metadata: dict = None):
        """อัพเดทสถานะ task"""
        with self.lock:
            old_state = self.tasks_state.get(task_id, "UNKNOWN")
            self

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง