ในบทความนี้เราจะมาเจาะลึกการควบคุม timeout และการจัดการ exception ของ CrewAI อย่างละเอียด พร้อมตัวอย่างโค้ดที่ใช้งานได้จริง ซึ่งเป็นหัวข้อสำคัญสำหรับนักพัฒนาที่ต้องการสร้างระบบ Multi-Agent ที่เสถียรและทำงานได้อย่างมีประสิทธิภาพ

ทำไมต้องควบคุม Timeout และ Exception

เมื่อใช้งาน CrewAI กับ Large Language Models โดยเฉพาะในระดับ Production การจัดการ timeout และ exception เป็นสิ่งที่ขาดไม่ได้ เพราะ API อาจตอบสนองช้า หรือเกิดข้อผิดพลาดการเชื่อมต่อ หากไม่มีการจัดการที่ดี ระบบทั้งหมดอาจค้างหรือทำงานผิดพลาดโดยไม่มีการแจ้งเตือน

เปรียบเทียบต้นทุน API ปี 2026

ก่อนจะเข้าสู่เนื้อหาหลัก มาดูต้นทุนของโมเดลต่างๆ สำหรับ 10 ล้าน tokens ต่อเดือนกัน:

หากใช้งานผ่าน HolySheep AI ซึ่งมีอัตราแลกเปลี่ยน ¥1=$1 (ประหยัด 85%+) รวมถึงรองรับ WeChat/Alipay, latency ต่ำกว่า 50ms และเครดิตฟรีเมื่อลงทะเบียน ต้นทุนจะลดลงอย่างมาก ทำให้การพัฒนาระบบ CrewAI มีความคุ้มค่ามากขึ้น

การตั้งค่า Timeout พื้นฐาน

2.1 Task Timeout Configuration

import os
from crewai import Agent, Task, Crew
from crewai.utilities.timeout import timeout


ตั้งค่า environment

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

def research_task_with_timeout():
    """ตัวอย่าง Task ที่มีการตั้งค่า Timeout"""
    
    researcher = Agent(
        role="Senior Research Analyst",
        goal="ค้นหาและวิเคราะห์ข้อมูลอย่างละเอียด",
        backstory="คุณเป็นนักวิเคราะห์ข้อมูลอาวุโสที่มีประสบการณ์ 10 ปี",
        verbose=True,
        max_iterations=3,
        max_retry_limit=2
    )
    
    # กำหนด timeout เป็น 60 วินาทีสำหรับ Task
    research_task = Task(
        description="ค้นหาข้อมูลเกี่ยวกับ AI trends 2026",
        agent=researcher,
        expected_output="รายงานสรุป AI trends 2026",
        timeout=60  # หน่วยเป็นวินาที
    )
    
    crew = Crew(
        agents=[researcher],
        tasks=[research_task],
        verbose=True
    )
    
    return crew.kickoff()

result = research_task_with_timeout()
print(f"ผลลัพธ์: {result}")

2.2 Timeout Decorator สำหรับ Function

import signal
from functools import wraps

def timeout_handler(signum, frame):
    """Handler ที่ทำงานเมื่อเกิด timeout"""
    raise TimeoutError("การทำงานใช้เวลานานเกินกำหนด")

def with_timeout(seconds=30):
    """
    Decorator สำหรับกำหนด timeout ให้ function
    ใช้ได้กับ Linux/Mac เท่านั้น
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            signal.signal(signal.SIGALRM, timeout_handler)
            signal.alarm(seconds)  # ตั้งเวลา alarm
            
            try:
                result = func(*args, **kwargs)
            finally:
                signal.alarm(0)  # ยกเลิก alarm
                
            return result
        return wrapper
    return decorator


ตัวอย่างการใช้งาน

@with_timeout(seconds=45)
def long_running_agent_task(query: str) -> str:
    """Task ที่ใช้เวลานาน"""
    # จำลองการทำงานที่ใช้เวลานาน
    import time
    time.sleep(40)  # ทำงาน 40 วินาที
    return f"ผลลัพธ์สำหรับ: {query}"

try:
    result = long_running_agent_task("วิเคราะห์ข้อมูลตลาด")
    print(result)
except TimeoutError as e:
    print(f"เกิด Timeout: {e}")

การจัดการ Exception แบบครบวงจร

import logging
from typing import Any, Optional
from crewai import Agent, Task, Crew
from crewai.utilities.exceptions import (
    APIKeyError,
    ContextWindowExceededError,
    ToolException
)


ตั้งค่า logging

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

class CrewAIExceptionHandler:
    """Class สำหรับจัดการ Exception ของ CrewAI"""
    
    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
        self.error_count = {}
    
    def handle_api_error(self, error: Exception, agent_name: str) -> dict:
        """จัดการ API Error ต่างๆ"""
        error_type = type(error).__name__
        
        if isinstance(error, APIKeyError):
            logger.error(f"API Key Error สำหรับ {agent_name}: {error}")
            return {
                "status": "failed",
                "error_type": "api_key",
                "action": "ตรวจสอบ API Key ที่ https://www.holysheep.ai/register",
                "recoverable": False
            }
        
        elif isinstance(error, ContextWindowExceededError):
            logger.warning(f"Context Window เต็มสำหรับ {agent_name}")
            return {
                "status": "retry_with_reduced_context",
                "error_type": "context_exceeded",
                "action": "ลดขนาด context หรือใช้โมเดลที่มี context ใหญ่กว่า",
                "recoverable": True
            }
        
        elif isinstance(error, ToolException):
            logger.error(f"Tool Error สำหรับ {agent_name}: {error}")
            return {
                "status": "failed",
                "error_type": "tool",
                "action": "ตรวจสอบการทำงานของ tool",
                "recoverable": True
            }
        
        else:
            logger.error(f"Unknown Error สำหรับ {agent_name}: {error}")
            return {
                "status": "failed",
                "error_type": error_type,
                "action": "บันทึก log และติดต่อ support",
                "recoverable": False
            }
    
    def execute_with_retry(
        self, 
        crew: Crew, 
        inputs: dict,
        callback: Optional[callable] = None
    ) -> Any:
        """Execute crew พร้อม retry mechanism"""
        
        for attempt in range(1, self.max_retries + 1):
            try:
                logger.info(f"พยายามครั้งที่ {attempt}/{self.max_retries}")
                
                result = crew.kickoff(inputs=inputs)
                
                if callback:
                    callback(result)
                
                return {
                    "status": "success",
                    "result": result,
                    "attempts": attempt
                }
                
            except Exception as e:
                logger.warning(f"ครั้งที่ {attempt} ล้มเหลว: {str(e)}")
                
                self.error_count[type(e).__name__] = \
                    self.error_count.get(type(e).__name__, 0) + 1
                
                if attempt == self.max_retries:
                    handler_result = self.handle_api_error(
                        e, 
                        crew.agents[0].role if crew.agents else "Unknown"
                    )
                    
                    return {
                        "status": "failed",
                        "error": handler_result,
                        "attempts": attempt,
                        "error_history": self.error_count
                    }
                
                # Exponential backoff
                import time
                wait_time = 2 ** attempt
                logger.info(f"รอ {wait_time} วินาทีก่อนลองใหม่")
                time.sleep(wait_time)
        
        return {"status": "max_retries_exceeded"}



ตัวอย่างการใช้งาน

def main():
    from crewai import Agent, Task, Crew
    
    os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
    os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
    
    writer = Agent(
        role="Content Writer",
        goal="เขียนเนื้อหาคุณภาพสูง",
        backstory="นักเขียนมืออาชีพที่มีประสบการณ์",
        verbose=True
    )
    
    task = Task(
        description="เขียนบทความ SEO 500 คำ",
        agent=writer,
        expected_output="บทความที่พร้อม publish"
    )
    
    crew = Crew(agents=[writer], tasks=[task])
    
    handler = CrewAIExceptionHandler(max_retries=3)
    result = handler.execute_with_retry(
        crew, 
        inputs={"topic": "AI in 2026"}
    )
    
    if result["status"] == "success":
        print(f"สำเร็จในครั้งที่ {result['attempts']}")
    else:
        print(f"ล้มเหลว: {result}")

if __name__ == "__main__":
    main()

Advanced Timeout Patterns

3.1 Async Timeout สำหรับ Concurrent Tasks

import asyncio
from crewai import Agent, Task, Crew
import os

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

class AsyncCrewRunner:
    """Runner สำหรับ CrewAI แบบ Async พร้อม Timeout"""
    
    def __init__(self, default_timeout: int = 120):
        self.default_timeout = default_timeout
        self.task_results = {}
    
    async def run_with_timeout(
        self, 
        crew: Crew, 
        inputs: dict,
        timeout: Optional[int] = None
    ) -> dict:
        """Run crew พร้อม asyncio timeout"""
        
        timeout_duration = timeout or self.default_timeout
        
        try:
            result = await asyncio.wait_for(
                crew.kickoff_async(inputs=inputs),
                timeout=timeout_duration
            )
            
            return {
                "status": "success",
                "result": result,
                "duration": f"<{timeout_duration}s"
            }
            
        except asyncio.TimeoutError:
            return {
                "status": "timeout",
                "error": f"Task ใช้เวลาเกิน {timeout_duration} วินาที",
                "partial_results": self.task_results
            }
    
    async def run_multiple_crews(
        self, 
        crews: list[Crew], 
        inputs: dict,
        timeout_per_crew: int = 60
    ) -> list[dict]:
        """Run หลาย crews พร้อมกันพร้อม individual timeout"""
        
        tasks = [
            self.run_with_timeout(crew, inputs, timeout_per_crew)
            for crew in crews
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        return [
            r if not isinstance(r, Exception) 
            else {"status": "error", "error": str(r)}
            for r in results
        ]


async def example_usage():
    """ตัวอย่างการใช้งาน"""
    
    analyzer = Agent(
        role="Data Analyst",
        goal="วิเคราะห์ข้อมูลอย่างละเอียด",
        backstory="นักวิเคราะห์ข้อมูลผู้เชี่ยวชาญ"
    )
    
    task = Task(
        description="วิเคราะห์ข้อมูลยอดขาย Q4",
        agent=analyzer,
        expected_output="รายงานการวิเคราะห์"
    )
    
    crew = Crew(agents=[analyzer], tasks=[task])
    
    runner = AsyncCrewRunner(default_timeout=90)
    
    result = await runner.run_with_timeout(
        crew, 
        inputs={"period": "Q4 2026"}
    )
    
    print(f"สถานะ: {result['status']}")
    if result['status'] == 'success':
        print(f"ผลลัพธ์: {result['result']}")


if __name__ == "__main__":
    asyncio.run(example_usage())

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

กรณีที่ 1: APIConnectionError - เชื่อมต่อ API ไม่ได้

# ❌ วิธีผิด: ไม่มีการจัดการ error
from crewai import Agent, Task, Crew
import os

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

crew = Crew(agents=[Agent(role="Test", goal="Test", backstory="Test")])
result = crew.kickoff()  # หากเกิด error จะ crash ทันที


✅ วิธีถูกต้อง: จัดการ error อย่างเหมาะสม

from crewai.utilities.exceptions import APIConnectionError
import time

def safe_kickoff(crew, max_retries=3):
    for attempt in range(max_retries):
        try:
            return crew.kickoff()
        except APIConnectionError as e:
            if attempt == max_retries - 1:
                raise
            print(f"เชื่อมต่อไม่ได้ ลองใหม่ใน 5 วินาที... ({attempt+1}/{max_retries})")
            time.sleep(5)

result = safe_kickoff(crew)

กรณีที่ 2: RateLimitError - เกินขีดจำกัดการใช้งาน

# ❌ วิธีผิด: ไม่มี delay ระหว่าง request
for task_description in many_tasks:
    crew = Crew(agents=[Agent(...)])
    crew.kickoff()  # อาจถูก rate limit ทันที


✅ วิธีถูกต้อง: ใช้ rate limiting และ retry

from crewai.utilities.exceptions import RateLimitError
import time

def rate_limited_kickoff(crew, base_delay=2, max_delay=60):
    delay = base_delay
    
    while True:
        try:
            return crew.kickoff()
        except RateLimitError:
            print(f"ถูก rate limit รอ {delay} วินาที...")
            time.sleep(delay)
            delay = min(delay * 2, max_delay)  # Exponential backoff

กรณีที่ 3: ContextWindowExceededError - Context เต็ม

# ❌ วิธีผิด: ส่งข้อมูลมากเกินไปโดยไม่ truncate
task = Task(
    description=f"วิเคราะห์เอกสาร {very_long_document}",  # อาจเกิน context
    agent=researcher
)


✅ วิธีถูกต้อง: Truncate ข้อมูลก่อนส่ง

def truncate_for_context(text: str, max_chars: int = 10000) -> str:
    """ตัดข้อความให้พอดีกับ context window"""
    if len(text) <= max_chars:
        return text
    return text[:max_chars] + "\n\n[เนื้อหาถูกตัดเหลือเพื่อให้พอดีกับ context window]"

task = Task(
    description=f"วิเคราะห์เอกสาร:\n{truncate_for_context(very_long_document)}",
    agent=researcher
)


หรือใช้ chunking สำหรับเอกสารขนาดใหญ่

def process_large_document(document: str, chunk_size: int = 5000):
    chunks = [document[i:i+chunk_size] for i in range(0, len(document), chunk_size)]
    
    results = []
    for i, chunk in enumerate(chunks):
        task = Task(
            description=f"วิเคราะห์ส่วนที่ {i+1}/{len(chunks)}:\n{chunk}",
            agent=researcher
        )
        result = Crew(agents=[researcher], tasks=[task]).kickoff()
        results.append(result)
    
    return results

กรณีที่ 4: Task Timeout ไม่ทำงาน

# ❌ วิธีผิด: Task timeout ไม่ได้ block process
task = Task(
    description="Task ที่อาจค้าง",
    agent=agent,
    timeout=5  # ค่านี้อาจไม่ทำงานในทุกกรณี
)


✅ วิธีถูกต้อง: ใช้ threading หรือ multiprocessing

import multiprocessing
import signal

def run_task_with_hard_timeout(task_func, timeout_seconds, *args, **kwargs):
    """Force kill task หลังจาก timeout"""
    
    def wrapper(result_queue):
        try:
            result = task_func(*args, **kwargs)
            result_queue.put(("success", result))
        except Exception as e:
            result_queue.put(("error", str(e)))
    
    result_queue = multiprocessing.Queue()
    process = multiprocessing.Process(target=wrapper, args=(result_queue,))
    process.start()
    process.join(timeout=timeout_seconds)
    
    if process.is_alive():
        process.terminate()
        process.join()
        return ("timeout", f"Task ใช้เวลาเกิน {timeout_seconds} วินาที")
    
    if result_queue.empty():
        return ("error", "ไม่มีผลลัพธ์")
    
    return result_queue.get()


ตัวอย่างการใช้งาน

def long_task():
    import time
    time.sleep(30)
    return "เสร็จแล้ว"

status, result = run_task_with_hard_timeout(long_task, timeout_seconds=5)
print(f"สถานะ: {status}, ผลลัพธ์: {result}")  # จะได้ timeout หลัง 5 วินาที

สรุปแนวทางปฏิบัติที่ดีที่สุด

การจัดการ timeout และ exception อย่างถูกต้องจะช่วยให้ระบบ CrewAI ของคุณทำงานได้อย่างเสถียรและมีประสิทธิภาพสูงสุด อย่าลืมทดสอบ edge cases ต่างๆ ก่อนนำไปใช้งานจริง

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน

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

บทความที่เกี่ยวข้อง