บทนำ: ทำไม Agent ถึงต้องมี State Machine

ในระบบ Multi-Agent ของ CrewAI แต่ละ Agent ไม่ได้ทำงานแบบ Linear หรือเรียบง่ายตลอดเวลา Agent ต้องรับมือกับสถานการณ์ที่ซับซ้อน เช่น การรอข้อมูลเพิ่มเติม การตัดสินใจแตกแขนง หรือการจัดการข้อผิดพลาดที่เกิดขึ้นระหว่างทาง การใช้ State Machine ช่วยให้เรากำหนดสถานะที่ชัดเจน ทำให้ Agent ทำงานได้อย่างมีประสิทธิภาพและควบคุมได้ดีขึ้น

กรณีศึกษา: บริษัทสตาร์ทอัพ AI ในกรุงเทพฯ

บริบทธุรกิจ

ทีมพัฒนา AI จากสตาร์ทอัพแห่งหนึ่งในกรุงเทพฯ สร้างระบบ Customer Support Agent ที่ทำงานอัตโนมัติ โดยมี Agent ย่อย 5 ตัวทำหน้าที่ต่างกัน ตั้งแต่รับข้อความลูกค้า วิเคราะห์ความต้องการ ค้นหาข้อมูลในฐานข้อมูล ไปจนถึงตอบกลับและติดตามผล ระบบเดิมใช้ OpenAI เป็นหลักแต่ประสบปัญหา Latency สูงและค่าใช้จ่ายที่พุ่งสูงขึ้นอย่างต่อเนื่อง

จุดเจ็บปวดของระบบเดิม

ระบบเดิมมีปัญหาหลายอย่างที่สะสมมานาน ค่าใช้จ่ายด้าน API พุ่งไปถึง $4,200 ต่อเดือน ซึ่งสูงเกินไปสำหรับธุรกิจขนาดกลาง นอกจากนี้ Latency เฉลี่ยอยู่ที่ 420ms ทำให้ลูกค้ารู้สึกว่าการตอบกลับช้าเกินไป โดยเฉพาะเมื่อต้องรอ Response จากหลาย Agent ติดต่อกัน ทำให้ทั้ง Flow ใช้เวลานานเกินไป

การย้ายมาใช้ HolySheep AI

ทีมตัดสินใจย้ายมาใช้ HolySheep AI เพราะมีอัตรา $1=¥1 ทำให้ประหยัดได้มากกว่า 85% เมื่อเทียบกับราคาเดิม และมี Latency ต่ำกว่า 50ms ซึ่งต่ำกว่าระบบเดิมอย่างเห็นได้ชัด การย้ายระบบเริ่มจากการเปลี่ยน base_url จาก OpenAI เป็น https://api.holysheep.ai/v1 จากนั้นทำการหมุนคีย์ API ใหม่และเริ่มทยอย Deploy แบบ Canary ทีละ Agent เพื่อไม่ให้ระบบหยุดชะงัก

ผลลัพธ์หลัง 30 วัน

หลังจากย้ายระบบมาใช้ HolySheep AI ได้ 30 วัน ตัวชี้วัดดีขึ้นอย่างเห็นได้ชัด Latency เฉลี่ยลดลงจาก 420ms เหลือ 180ms คิดเป็นการปรับปรุงมากกว่า 57% และค่าใช้จ่ายรายเดือนลดลงจาก $4,200 เหลือเพียง $680 ซึ่งเป็นการประหยัดเกือบ 84% ทีมพัฒนาสามารถนำเงินที่ประหยัดได้ไปลงทุนพัฒนาฟีเจอร์ใหม่ๆ แทน

พื้นฐาน State Machine ใน CrewAI

State Machine เป็นแนวคิดที่ใช้จัดการสถานะของ Agent โดยกำหนดว่า Agent จะอยู่สถานะไหน จากสถานะหนึ่งไปสถานะไหนได้บ้าง และต้องทำเงื่อนไขอะไรถึงจะเปลี่ยนสถานะได้ ใน CrewAI เราสามารถกำหนด State และ Transition ได้ด้วยตัวเอง

สถานะหลักของ Agent

สถานะแรกคือ IDLE ซึ่งเป็นสถานะเริ่มต้นเมื่อ Agent พร้อมทำงาน จากนั้นเมื่อได้รับ Task Agent จะเปลี่ยนเป็น PROCESSING เพื่อทำงาน เมื่อทำงานเสร็จจะเปลี่ยนเป็น COMPLETED หรือถ้ามีปัญหาจะเปลี่ยนเป็น FAILED ซึ่งต้องมีการจัดการ Retry หรือ Escalation ต่อไป นอกจากนี้ยังมีสถานะ WAITING สำหรับรอข้อมูลจาก Agent อื่นหรือผู้ใช้

การกำหนด Transition Rules

การเปลี่ยนสถานะต้องมีเงื่อนไขที่ชัดเจน เช่น จาก IDLE ไป PROCESSING ต้องมี Task ที่รออยู่ จาก PROCESSING ไป COMPLETED ต้องได้ผลลัพธ์ที่ถูกต้อง จาก PROCESSING ไป FAILED ต้องเกิด Error หรือ Timeout ส่วนจาก FAILED ไป PROCESSING ใหม่ต้องผ่านการ Retry Logic ที่กำหนดไว้ 
from enum import Enum
from typing import Optional, Callable
from dataclasses import dataclass, field

class AgentState(Enum):
    """สถานะหลักของ Agent"""
    IDLE = "idle"
    PROCESSING = "processing"
    WAITING = "waiting"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class Transition:
    """กฎการเปลี่ยนสถานะ"""
    from_state: AgentState
    to_state: AgentState
    condition: Callable[[], bool]
    action: Optional[Callable[[], None]] = None

class AgentStateMachine:
    """State Machine สำหรับจัดการวงจรชีวิตของ Agent"""
    
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.current_state = AgentState.IDLE
        self.transitions: list[Transition] = []
        self.state_history: list[tuple[str, AgentState]] = []
        self._register_default_transitions()
    
    def _register_default_transitions(self):
        """ลงทะเบียน Transition พื้นฐาน"""
        # IDLE -> PROCESSING: เมื่อมี Task รอ
        self.add_transition(
            AgentState.IDLE,
            AgentState.PROCESSING,
            lambda: self.has_pending_task()
        )
        
        # PROCESSING -> COMPLETED: เมื่อทำงานสำเร็จ
        self.add_transition(
            AgentState.PROCESSING,
            AgentState.COMPLETED,
            lambda: self.task_succeeded()
        )
        
        # PROCESSING -> FAILED: เมื่อเกิด Error
        self.add_transition(
            AgentState.PROCESSING,
            AgentState.FAILED,
            lambda: self.task_failed()
        )
        
        # FAILED -> PROCESSING: เมื่อถึงเวลา Retry
        self.add_transition(
            AgentState.FAILED,
            AgentState.PROCESSING,
            lambda: self.should_retry()
        )
        
        # PROCESSING -> WAITING: เมื่อต้องรอข้อมูล
        self.add_transition(
            AgentState.PROCESSING,
            AgentState.WAITING,
            lambda: self.needs_more_info()
        )
        
        # WAITING -> PROCESSING: เมื่อได้รับข้อมูลที่ต้องการ
        self.add_transition(
            AgentState.WAITING,
            AgentState.PROCESSING,
            lambda: self.info_received()
        )
    
    def add_transition(
        self,
        from_state: AgentState,
        to_state: AgentState,
        condition: Callable[[], bool],
        action: Optional[Callable[[], None]] = None
    ):
        """เพิ่มกฎการเปลี่ยนสถานะ"""
        self.transitions.append(Transition(
            from_state=from_state,
            to_state=to_state,
            condition=condition,
            action=action
        ))
    
    def try_transition(self) -> bool:
        """พยายามเปลี่ยนสถานะ"""
        for transition in self.transitions:
            if (transition.from_state == self.current_state and 
                transition.condition()):
                self._execute_transition(transition)
                return True
        return False
    
    def _execute_transition(self, transition: Transition):
        """ดำเนินการเปลี่ยนสถานะ"""
        if transition.action:
            transition.action()
        
        self.state_history.append((self.agent_id, self.current_state))
        self.current_state = transition.to_state
        print(f"[{self.agent_id}] State: {self.current_state.value}")
    
    def has_pending_task(self) -> bool:
        """ตรวจสอบว่ามี Task รอหรือไม่"""
        return getattr(self, '_has_task', False)
    
    def task_succeeded(self) -> bool:
        """ตรวจสอบว่าทำงานสำเร็จหรือไม่"""
        return getattr(self, '_task_result', None) == 'success'
    
    def task_failed(self) -> bool:
        """ตรวจสอบว่าทำงานล้มเหลวหรือไม่"""
        return getattr(self, '_task_result', None) == 'failed'
    
    def should_retry(self) -> bool:
        """ตรวจสอบว่าควร Retry หรือไม่"""
        retry_count = getattr(self, '_retry_count', 0)
        return retry_count < 3
    
    def needs_more_info(self) -> bool:
        """ตรวจสอบว่าต้องรอข้อมูลเพิ่มหรือไม่"""
        return getattr(self, '_waiting_for_info', False)
    
    def info_received(self) -> bool:
        """ตรวจสอบว่าได้รับข้อมูลแล้วหรือยัง"""
        return getattr(self, '_info_ready', False)
    
    def update_context(self, **kwargs):
        """อัปเดต Context สำหรับตรวจสอบเงื่อนไข"""
        for key, value in kwargs.items():
            setattr(self, f'_{key}', value)

การผสมผสาน CrewAI กับ State Machine

เมื่อนำ State Machine มารวมกับ CrewAI เราจะได้ระบบที่มีทั้งความยืดหยุ่นของ Multi-Agent และความควบคุมของ State Machine ตัวอย่างนี้แสดงการสร้าง Customer Support Crew ที่ใช้ State Machine จัดการการไหลของงานระหว่าง Agent ต่างๆ 
import os
from crewai import Agent, Task, Crew
from langchain_openai import ChatOpenAI
from dataclasses import dataclass
from typing import Optional, Dict, Any
import time


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

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


สร้าง LLM instance

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


กำหนดราคา (2026 pricing)

PRICING = {
    "gpt-4.1": 8.00,           # $8 per 1M tokens
    "claude-sonnet-4.5": 15.00, # $15 per 1M tokens
    "gemini-2.5-flash": 2.50,   # $2.50 per 1M tokens
    "deepseek-v3.2": 0.42       # $0.42 per 1M tokens
}

@dataclass
class TicketContext:
    """Context สำหรับ Ticket ที่กำลังดำเนินการ"""
    ticket_id: str
    customer_message: str
    customer_tier: str  # regular, premium, vip
    sentiment: Optional[str] = None
    resolved: bool = False
    escalation_needed: bool = False
    assigned_agent: Optional[str] = None
    history: list = None
    
    def __post_init__(self):
        if self.history is None:
            self.history = []

class CustomerSupportStateMachine(AgentStateMachine):
    """State Machine สำหรับระบบ Customer Support"""
    
    def __init__(self, agent_id: str):
        super().__init__(agent_id)
        self.ticket_context: Optional[TicketContext] = None
        self.processing_start: Optional[float] = None
    
    def assign_ticket(self, ticket: TicketContext):
        """มอบหมาย Ticket ให้ Agent"""
        self.ticket_context = ticket
        self.update_context(_has_task=True)
        self.processing_start = time.time()
    
    def mark_resolved(self):
        """ทำเครื่องหมายว่า Ticket ถูกแก้ไขแล้ว"""
        self.ticket_context.resolved = True
        self.update_context(_task_result='success')
    
    def mark_failed(self, reason: str):
        """ทำเครื่องหมายว่า Ticket แก้ไขไม่ได้"""
        self.ticket_context.history.append({
            "state": "failed",
            "reason": reason,
            "timestamp": time.time()
        })
        self.update_context(_task_result='failed')
        self._retry_count = getattr(self, '_retry_count', 0) + 1
    
    def escalate(self):
        """Escalate ไปยัง Agent ระดับสูงกว่า"""
        self.ticket_context.escalation_needed = True
        self.ticket_context.assigned_agent = "senior_agent"
    
    def get_processing_time(self) -> float:
        """คำนวณเวลาที่ใช้ประมวลผล (มิลลิวินาที)"""
        if self.processing_start:
            return (time.time() - self.processing_start) * 1000
        return 0


สร้าง Agents

router_agent = Agent(
    role="Ticket Router",
    goal="วิเคราะห์ Ticket และจัดส่งไปยัง Agent ที่เหมาะสม",
    backstory="คุณคือผู้เชี่ยวชาญในการจัดการ Ticket สนับสนุนลูกค้า",
    llm=llm,
    verbose=True
)

support_agent = Agent(
    role="Support Agent",
    goal="แก้ไขปัญหาของลูกค้าอย่างรวดเร็วและมีประสิทธิภาพ",
    backstory="คุณคือ Agent สนับสนุนลูกค้าที่มีประสบการณ์",
    llm=llm,
    verbose=True
)

senior_agent = Agent(
    role="Senior Support Agent",
    goal="จัดการปัญหาที่ซับซ้อนและต้องการความเชี่ยวชาญสูง",
    backstory="คุณคือผู้เชี่ยวชาญอาวุโสในการแก้ไขปัญหายาก",
    llm=llm,
    verbose=True
)

def create_support_crew(ticket: TicketContext) -> tuple[Crew, CustomerSupportStateMachine]:
    """สร้าง Crew และ State Machine สำหรับ Ticket"""
    
    # สร้าง State Machine
    state_machine = CustomerSupportStateMachine("main_support")
    state_machine.assign_ticket(ticket)
    
    # กำหนด Task ตามสถานะ
    if ticket.escalation_needed:
        # Task สำหรับ VIP หรือปัญหาซับซ้อน
        tasks = [
            Task(
                description=f"แก้ไขปัญหาของลูกค้า VIP: {ticket.customer_message}",
                agent=senior_agent,
                expected_output="รายงานการแก้ไขปัญหาพร้อมขั้นตอนที่ดำเนินการ"
            )
        ]
    else:
        # Task ปกติ
        tasks = [
            Task(
                description=f"วิเคราะห์และแก้ไข: {ticket.customer_message}",
                agent=support_agent,
                expected_output="คำตอบที่ชัดเจนสำหรับลูกค้า"
            )
        ]
    
    # สร้าง Crew
    crew = Crew(
        agents=[senior_agent if ticket.escalation_needed else support_agent],
        tasks=tasks,
        verbose=True
    )
    
    return crew, state_machine

def run_ticket_workflow(ticket: TicketContext) -> Dict[str, Any]:
    """Run Workflow แบบเต็มสำหรับ Ticket"""
    
    print(f"[Workflow] เริ่มดำเนินการ Ticket: {ticket.ticket_id}")
    
    # ขั้นตอนที่ 1: Routing
    router_state = CustomerSupportStateMachine("router")
    router_state.assign_ticket(ticket)
    
    while router_state.current_state != AgentState.COMPLETED:
        router_state.try_transition()
        
        if router_state.current_state == AgentState.PROCESSING:
            # ใช้ Router Agent วิเคราะห์ Ticket
            routing_result = router_agent.execute_task(
                f"วิเคราะห์ Ticket: {ticket.customer_message}"
            )
            
            # กำหนดว่าต้อง Escalate หรือไม่
            if ticket.customer_tier == "vip" or "refund" in routing_result.lower():
                ticket.escalation_needed = True
                router_state.escalate()
            
            router_state.mark_resolved()
    
    # ขั้นตอนที่ 2: Support
    crew, support_state = create_support_crew(ticket)
    
    while support_state.current_state != AgentState.COMPLETED:
        support_state.try_transition()
        
        if support_state.current_state == AgentState.PROCESSING:
            result = crew.kickoff()
            
            # ตรวจสอบผลลัพธ์
            if result and "error" not in str(result).lower():
                support_state.mark_resolved()
            else:
                support_state.mark_failed(str(result))
    
    return {
        "ticket_id": ticket.ticket_id,
        "status": "resolved" if support_state.ticket_context.resolved else "failed",
        "processing_time_ms": support_state.get_processing_time(),
        "escalated": ticket.escalation_needed,
        "agent": ticket.assigned_agent
    }


ทดสอบ Workflow

if __name__ == "__main__":
    test_ticket = TicketContext(
        ticket_id="TKT-001",
        customer_message="ฉันต้องการขอเงินคืนสำหรับการสั่งซื้อครั้งที่แล้ว",
        customer_tier="vip"
    )
    
    result = run_ticket_workflow(test_ticket)
    print(f"ผลลัพธ์: {result}")

การติดตาม Lifecycle และ Metrics

การจัดการ State Machine ที่ดีต้องมีการติดตาม Lifecycle ของ Agent อย่างต่อเนื่อง เพื่อให้เราสามารถวิเคราะห์ประสิทธิภาพและหาโอกาสในการปรับปรุงได้ ระบบติดตามที่ดีควรเก็บข้อมูลสถานะ ระยะเวลาในแต่ละสถานะ และการเปลี่ยนแปลงต่างๆ

การสร้าง Lifecycle Tracker

from datetime import datetime
from typing import Dict, List, Optional
from collections import defaultdict
import threading

@dataclass
class StateTransition:
    """บันทึกการเปลี่ยนสถานะ"""
    timestamp: float
    agent_id: str
    from_state: str
    to_state: str
    duration_ms: float  # เวลาที่อยู่ในสถานะก่อนหน้า

@dataclass
class LifecycleMetrics:
    """Metrics สำหรับ Lifecycle ของ Agent"""
    total_transitions: int
    state_distribution: Dict[str, int]
    avg_time_per_state: Dict[str, float]
    failed_count: int
    retry_count: int
    escalation_count: int

class LifecycleTracker:
    """Tracker สำหรับติดตาม Lifecycle ของ Agent ทั้งหมด"""
    
    def __init__(self):
        self.transitions: List[StateTransition] = []
        self.state_timers: Dict[str, float] = {}  # agent_id -> last timestamp
        self.current_states: Dict[str, str] = {}
        self._lock = threading.Lock()
        self._callbacks: List[callable] = []
    
    def on_state_change(self, callback: callable):
        """ลงทะเบียน callback สำหรับเมื่อสถานะเปลี่ยน"""
        self._callbacks.append(callback)
    
    def record_transition(
        self,
        agent_id: str,
        from_state: str,
        to_state: str
    ):
        """บันทึกการเปลี่ยนสถานะ"""
        with self._lock:
            current_time = time.time()
            duration_ms = 0
            
            if agent_id in self.state_timers:
                duration_ms = (current_time - self.state_timers[agent_id]) * 1000
            
            transition = StateTransition(
                timestamp=current_time,
                agent_id=agent_id,
                from_state=from_state,
                to_state=to_state,
                duration_ms=duration_ms
            )
            
            self.transitions.append(transition)
            self.state_timers[agent_id] = current_time
            self.current_states[agent_id] = to_state
            
            # เรียก callbacks
            for callback in self._callbacks:
                callback(agent_id, from_state, to_state)
    
    def get_agent_history(self, agent_id: str) -> List[StateTransition]:
        """ดึงประวัติการเปลี่ยนสถานะของ Agent"""
        with self._lock:
            return [t for t in self.transitions if t.agent_id == agent_id]
    
    def get_current_state(self, agent_id: str) -> Optional[str]:
        """ดึงสถานะปัจจุบันของ Agent"""
        return self.current_states.get(agent_id)
    
    def get_metrics(self, time_window_seconds: Optional[float] = None) -> LifecycleMetrics:
        """คำนวณ Metrics รวม"""
        with self._lock:
            transitions = self.transitions
            
            if time_window_seconds:
                cutoff = time.time() - time_window_seconds
                transitions = [t for t in transitions if t.timestamp >= cutoff]
            
            # กระจายตัวของสถานะ
            state_counts = defaultdict(int)
            state_times: Dict[str, List[float]] = defaultdict(list)
            
            for t in transitions:
                state_counts[t.to_state] += 1
                state_times[t.to_state].append(t.duration_ms)
            
            # คำนวณเวลาเฉลี่ยต่อสถานะ
            avg_times = {
                state: sum(times) / len(times) if times else 0
                for state, times in state_times.items()
            }
            
            # นับ Failed และ Retry
            failed = sum(1 for t in transitions if t.to_state == "failed")
            escalations = sum(1 for t in transitions if "escalation" in str(t).lower())
            
            return LifecycleMetrics(
                total_transitions=len(transitions),
                state_distribution=dict(state_counts),
                avg_time_per_state=avg_times,
                failed_count=failed,
                retry_count=0,  # ต้อง track แยก
                escalation_count=escalations
            )
    
    def get_agent_metrics(self, agent_id: str) -> Dict[str, Any]:
        """ดึง Metrics ของ Agent เฉพาะ"""
        history = self.get_agent_history(agent_id)
        
        if not history:
            return {"error": "No history found"}
        
        state_counts = defaultdict(int)
        for t in history:
            state_counts[t.to_state] += 1
        
        return {
            "agent_id": agent_id,
            "total_transitions": len(history),
            "current_state": self.get_current_state(agent_id),
            "state_distribution": dict(state_counts),
            "last_transition": history[-1].timestamp if history else None,
            "total_time_ms": sum(t.duration_ms for t in history)
        }


Integration กับ State Machine

class TrackedStateMachine(AgentStateMachine):
    """State Machine ที่เชื่อมต่อกับ Lifecycle Tracker"""
    
    def __init__(self, agent_id: str, tracker: LifecycleTracker):
        super().__init__(agent_id)
        self.tracker = tracker
        self._previous_state = None
    
    def try_transition(self) -> bool:
        """ดำเนินการเปลี่ยนสถานะพร

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