ในโลกของ Multi-Agent System นั้น การรู้ว่า Agent แต่ละตัวทำงานสำเร็จหรือล้มเหลวอย่างไร คือหัวใจสำคัญของการ Deploy ระบบที่เสถียร จากประสบการณ์ตรงของผมในการสร้าง CrewAI Pipeline ที่รันบน Production มาแล้วกว่า 6 เดือน ผมจะพาคุณเจาะลึกวิธีการ Monitor Agent Task Success Rates อย่างเป็นระบบ

ทำไมต้อง Monitor Agent Task Success Rates

ในระบบ Multi-Agent แบบ CrewAI นั้น การที่ Agent ตัวหนึ่งล้มเหลวอาจทำให้ทั้ง Pipeline สะดุดได้ ผมเคยเจอกรณีที่ Agent ที่ทำงานขั้น中间 (Intermediate Step) ล้มเหลวเงียบๆ ทำให้ Output ที่ได้มาผิดพลาดโดยไม่มี Error Message แสดงออกมาเลย นี่คือเหตุผลที่การ Monitor Success Rate อย่างเป็นระบบเป็นสิ่งจำเป็น

สถาปัตยกรรม Monitoring สำหรับ CrewAI

สำหรับ HolySheep AI ที่ให้บริการ API สำหรับ AI Models ด้วยความหน่วงต่ำกว่า 50 มิลลิวินาที และราคาที่ประหยัดกว่า 85% เมื่อเทียบกับผู้ให้บริการรายอื่น (อัตรา ¥1=$1) คุณสามารถใช้งานร่วมกับ CrewAI ได้อย่างมีประสิทธิภาพ

การติดตั้ง Callback System สำหรับ Track Success Rate

import os
from crewai import Agent, Task, Crew
from crewai.callbacks.base_callback_handler import BaseCallbackHandler
from typing import Any, Dict, List
from datetime import datetime
import json


กำหนดค่า Environment

os.environ["HOLYSHEEP_API_KEY"] = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")

class TaskSuccessCallback(BaseCallbackHandler):
    """
    Custom Callback สำหรับ Monitor Agent Task Success Rates
    ออกแบบมาสำหรับ Production Environment
    """
    
    def __init__(self):
        super().__init__()
        self.task_results: List[Dict[str, Any]] = []
        self.agent_success: Dict[str, int] = {}
        self.agent_failures: Dict[str, int] = {}
        self.start_time: Dict[str, float] = {}
        
    def on_agent_start(self, agent: Agent, **kwargs) -> None:
        """เริ่มจับเวลาเมื่อ Agent เริ่มทำงาน"""
        self.start_time[agent.role] = datetime.now().timestamp()
        print(f"[MONITOR] Agent '{agent.role}' started")
        
    def on_agent_end(self, agent: Agent, result: str, **kwargs) -> None:
        """บันทึกผลลัพธ์เมื่อ Agent ทำงานเสร็จ"""
        duration = datetime.now().timestamp() - self.start_time.get(agent.role, 0)
        
        # ตรวจสอบว่าผลลัพธ์ถือว่าสำเร็จหรือไม่
        is_success = self._evaluate_success(result)
        
        task_record = {
            "agent_role": agent.role,
            "success": is_success,
            "duration_seconds": round(duration, 2),
            "timestamp": datetime.now().isoformat(),
            "result_length": len(result) if result else 0
        }
        
        self.task_results.append(task_record)
        
        # อัพเดท Statistics
        if is_success:
            self.agent_success[agent.role] = self.agent_success.get(agent.role, 0) + 1
        else:
            self.agent_failures[agent.role] = self.agent_failures.get(agent.role, 0) + 1
            
        print(f"[MONITOR] Agent '{agent.role}' finished - Success: {is_success}, Duration: {duration:.2f}s")
        
    def on_agent_error(self, agent: Agent, error: Exception, **kwargs) -> None:
        """จัดการกรณี Agent เกิด Error"""
        self.agent_failures[agent.role] = self.agent_failures.get(agent.role, 0) + 1
        
        task_record = {
            "agent_role": agent.role,
            "success": False,
            "error": str(error),
            "timestamp": datetime.now().isoformat()
        }
        self.task_results.append(task_record)
        print(f"[MONITOR] Agent '{agent.role}' ERROR: {error}")
        
    def _evaluate_success(self, result: str) -> bool:
        """Logic การตัดสินว่าผลลัพธ์สำเร็จหรือไม่"""
        if not result:
            return False
        # เพิ่ม criteria ตามความต้องการของคุณ
        return len(result) > 10 and "error" not in result.lower()
    
    def get_success_rate(self, agent_role: str = None) -> float:
        """คำนวณ Success Rate ของ Agent เฉพาะ หรือทั้งหมด"""
        if agent_role:
            success = self.agent_success.get(agent_role, 0)
            failures = self.agent_failures.get(agent_role, 0)
        else:
            success = sum(self.agent_success.values())
            failures = sum(self.agent_failures.values())
        
        total = success + failures
        return (success / total * 100) if total > 0 else 0.0
    
    def get_statistics_report(self) -> Dict[str, Any]:
        """สร้างรายงานสถิติแบบครบถ้วน"""
        report = {
            "overall_success_rate": round(self.get_success_rate(), 2),
            "total_tasks": len(self.task_results),
            "by_agent": {}
        }
        
        all_agents = set(list(self.agent_success.keys()) + list(self.agent_failures.keys()))
        for agent in all_agents:
            report["by_agent"][agent] = {
                "success_rate": round(self.get_success_rate(agent), 2),
                "success_count": self.agent_success.get(agent, 0),
                "failure_count": self.agent_failures.get(agent, 0)
            }
            
        return report

การ Integrate กับ HolySheep AI API

สำหรับการใช้งานจริงใน Production ผมแนะนำให้ใช้ HolySheep AI เป็น Backend เนื่องจากมีความหน่วงต่ำเพียง <50ms และราคาที่คุ้มค่า โดยเฉพาะ DeepSeek V3.2 ที่ราคาเพียง $0.42/MTok สำหรับงานที่ต้องการประหยัดต้นทุน

from crewai.utilities import LLMProvider
from openai import OpenAI
import os


Configure CrewAI to use HolySheep AI

class HolySheepLLM(LLMProvider):
    """
    Custom LLM Provider สำหรับเชื่อมต่อกับ HolySheep AI
    รองรับ Models: GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2
    """
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str = None, model: str = "gpt-4.1"):
        self.client = OpenAI(
            api_key=api_key or os.getenv("HOLYSHEEP_API_KEY"),
            base_url=self.BASE_URL
        )
        self.model = model
        
    def call(self, messages: list, **kwargs):
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=kwargs.get("temperature", 0.7),
            max_tokens=kwargs.get("max_tokens", 2000)
        )
        return response.choices[0].message.content


ตัวอย่างการสร้าง Crew พร้อม Monitor

def create_monitored_crew(api_key: str):
    """สร้าง CrewAI Crew พร้อมระบบ Monitoring"""
    
    # Initialize LLM with HolySheep
    llm = HolySheepLLM(api_key=api_key, model="gpt-4.1")
    
    # Initialize Callback
    monitor_callback = TaskSuccessCallback()
    
    # สร้าง Agents
    research_agent = Agent(
        role="Research Analyst",
        goal="ค้นหาและวิเคราะห์ข้อมูลตลาดอย่างละเอียด",
        backstory="คุณคือนักวิเคราะห์ตลาดที่มีประสบการณ์ 10 ปี",
        llm=llm,
        callbacks=[monitor_callback]
    )
    
    writer_agent = Agent(
        role="Content Writer",
        goal="เขียนรายงานที่กระชับและมีคุณภาพสูง",
        backstory="คุณคือนักเขียนมืออาชีพที่เชี่ยวชาญด้าน Business Content",
        llm=llm,
        callbacks=[monitor_callback]
    )
    
    # สร้าง Tasks
    research_task = Task(
        description="วิเคราะห์แนวโน้มตลาด AI ในปี 2026",
        agent=research_agent,
        expected_output="รายงานวิเคราะห์ครบถ้วน 5 หน้า"
    )
    
    write_task = Task(
        description="เขียนบทสรุปผู้บริหารจากข้อมูลที่ได้รับ",
        agent=writer_agent,
        expected_output="Executive Summary 1 หน้า"
    )
    
    # สร้าง Crew
    crew = Crew(
        agents=[research_agent, writer_agent],
        tasks=[research_task, write_task],
        verbose=True
    )
    
    return crew, monitor_callback


การรันและติดตามผล

if __name__ == "__main__":
    API_KEY = "YOUR_HOLYSHEEP_API_KEY"
    
    crew, monitor = create_monitored_crew(API_KEY)
    
    # รัน Crew
    result = crew.kickoff()
    
    # แสดงผลสถิติ
    print("\n" + "="*50)
    print("📊 TASK SUCCESS RATE REPORT")
    print("="*50)
    
    report = monitor.get_statistics_report()
    print(f"Overall Success Rate: {report['overall_success_rate']}%")
    print(f"Total Tasks: {report['total_tasks']}")
    print("\nBy Agent:")
    for agent, stats in report['by_agent'].items():
        print(f"  - {agent}: {stats['success_rate']}% ({stats['success_count']}/{stats['success_count']+stats['failure_count']})")

การสร้าง Dashboard สำหรับ Real-time Monitoring

import streamlit as st
import plotly.graph_objects as go
from datetime import datetime, timedelta
from collections import defaultdict

class CrewAISuccessDashboard:
    """
    Dashboard สำหรับ Monitor Agent Success Rates แบบ Real-time
    ใช้ Streamlit สำหรับ Visualization
    """
    
    def __init__(self, callback_handler: TaskSuccessCallback):
        self.callback = callback_handler
        
    def render_gauge_chart(self, success_rate: float, title: str) -> go.Figure:
        """สร้าง Gauge Chart สำหรับแสดง Success Rate"""
        fig = go.Figure(go.Indicator(
            mode="gauge+number",
            value=success_rate,
            domain={'x': [0, 1], 'y': [0, 1]},
            title={'text': title},
            gauge={
                'axis': {'range': [0, 100]},
                'bar': {'color': self._get_color(success_rate)},
                'steps': [
                    {'range': [0, 50], 'color': "red"},
                    {'range': [50, 80], 'color': "yellow"},
                    {'range': [80, 100], 'color': "green"}
                ]
            }
        ))
        return fig
    
    def _get_color(self, rate: float) -> str:
        """กำหนดสีตามระดับ Success Rate"""
        if rate >= 90:
            return "#22c55e"  # green
        elif rate >= 70:
            return "#eab308"  # yellow
        else:
            return "#ef4444"  # red
            
    def render_trend_chart(self, historical_data: list) -> go.Figure:
        """สร้าง Trend Chart แสดงการเปลี่ยนแปลง Success Rate ตามเวลา"""
        if not historical_data:
            return go.Figure()
            
        # Group by time window
        time_groups = defaultdict(lambda: {"success": 0, "total": 0})
        for record in historical_data:
            timestamp = datetime.fromisoformat(record["timestamp"])
            time_key = timestamp.strftime("%H:%M")
            time_groups[time_key]["total"] += 1
            if record["success"]:
                time_groups[time_key]["success"] += 1
                
        x_vals = sorted(time_groups.keys())
        y_vals = [
            (time_groups[x]["success"] / time_groups[x]["total"] * 100)
            if time_groups[x]["total"] > 0 else 0
            for x in x_vals
        ]
        
        fig = go.Figure()
        fig.add_trace(go.Scatter(
            x=x_vals, y=y_vals,
            mode='lines+markers',
            name='Success Rate %',
            line=dict(color='#3b82f6', width=2)
        ))
        
        fig.update_layout(
            title='Success Rate Trend',
            xaxis_title='Time',
            yaxis_title='Success Rate (%)',
            yaxis_range=[0, 100]
        )
        
        return fig
    
    def render_agent_comparison(self, report: dict) -> go.Figure:
        """สร้าง Bar Chart เปรียบเทียบ Success Rate ระหว่าง Agents"""
        agents = list(report['by_agent'].keys())
        rates = [report['by_agent'][a]['success_rate'] for a in agents]
        
        fig = go.Figure()
        fig.add_trace(go.Bar(
            x=agents,
            y=rates,
            marker_color=[self._get_color(r) for r in rates]
        ))
        
        fig.update_layout(
            title='Agent Success Rate Comparison',
            xaxis_title='Agent',
            yaxis_title='Success Rate (%)',
            yaxis_range=[0, 100]
        )
        
        return fig

def create_dashboard(callback: TaskSuccessCallback):
    """Main function สำหรับสร้าง Dashboard"""
    st.set_page_config(page_title="CrewAI Monitor", layout="wide")
    
    st.title("🚀 CrewAI Agent Task Success Monitor")
    
    # Get latest report
    report = callback.get_statistics_report()
    
    # Layout columns
    col1, col2, col3 = st.columns(3)
    
    with col1:
        st.metric(
            "Overall Success Rate",
            f"{report['overall_success_rate']}%"
        )
        
    with col2:
        st.metric(
            "Total Tasks",
            report['total_tasks']
        )
        
    with col3:
        avg_rate = sum(
            agent['success_rate'] for agent in report['by_agent'].values()
        ) / len(report['by_agent']) if report['by_agent'] else 0
        st.metric(
            "Average Agent Rate",
            f"{avg_rate:.1f}%"
        )
    
    # Create charts
    dashboard = CrewAISuccessDashboard(callback)
    
    # Show gauge for overall
    st.plotly_chart(
        dashboard.render_gauge_chart(
            report['overall_success_rate'],
            "Overall System Health"
        )
    )
    
    # Show agent comparison
    st.plotly_chart(dashboard.render_agent_comparison(report))
    
    # Show trend
    st.plotly_chart(dashboard.render_trend_chart(callback.task_results))
    
    # Detailed table
    st.subheader("📋 Detailed Statistics")
    for agent, stats in report['by_agent'].items():
        with st.expander(f"Agent: {agent}"):
            col1, col2 = st.columns(2)
            with col1:
                st.write(f"Success Count: {stats['success_count']}")
                st.write(f"Failure Count: {stats['failure_count']}")
            with col2:
                st.write(f"Success Rate: {stats['success_rate']}%")

Best Practices จากประสบการณ์ Production

จากการใช้งานจริงใน Production มานานหลายเดือน ผมได้รวบรวม Best Practices ที่ช่วยให้การ Monitor มีประสิทธิภาพสูงสุด

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

1. API Key ไม่ถูกต้องหรือหมดอายุ

# ❌ วิธีที่ผิด: Hardcode API Key ในโค้ด
api_key = "sk-xxxxx-xxxxxxxxx"


✅ วิธีที่ถูก: ใช้ Environment Variable

import os
from dotenv import load_dotenv

load_dotenv()  # โหลดจาก .env file

api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key:
    raise ValueError("HOLYSHEEP_API_KEY not found in environment variables")


ตรวจสอบความถูกต้องของ API Key

def validate_api_key(key: str) -> bool:
    """ตรวจสอบว่า API Key ถูกต้องหรือไม่"""
    if not key or len(key) < 10:
        return False
    if key == "YOUR_HOLYSHEEP_API_KEY":
        raise ValueError("Please replace 'YOUR_HOLYSHEEP_API_KEY' with your actual key")
    return True

validate_api_key(api_key)

2. Callback ไม่ทำงานเนื่องจาก Event Loop Issue

# ❌ วิธีที่ผิด: สร้าง Callback ใน Async Function โดยไม่ระวัง
async def run_crew_async():
    callback = TaskSuccessCallback()  # Callback อาจไม่ถูกเรียกถูกต้อง
    crew = Crew(agents=agents, tasks=tasks, callbacks=[callback])
    result = await crew.kickoff_async()  # ต้องใช้ kickoff_async


✅ วิธีที่ถูก: ใช้ Synchronous Approach หรือจัดการ Async อย่างถูกต้อง

def run_crew_sync():
    callback = TaskSuccessCallback()
    crew = Crew(
        agents=agents, 
        tasks=tasks, 
        callbacks=[callback],  # Callback ต้องอยู่ใน list
        process="sequential"  # หรือ "parallel" ตามความต้องการ
    )
    result = crew.kickoff()  # Synchronous call
    return result, callback


หรือถ้าต้องการใช้ Async

async def run_crew_properly():
    callback = TaskSuccessCallback()
    crew = Crew(
        agents=agents, 
        tasks=tasks, 
        callbacks=[callback]
    )
    
    # ใช้ asyncio อย่างถูกต้อง
    import asyncio
    result = await asyncio.to_thread(crew.kickoff)
    return result, callback

3. Memory Leaks จากการเก็บ Task Results ไม่มี Limit

# ❌ วิธีที่ผิด: เก็บข้อมูลไม่มีจำกัด ทำให้ Memory เพิ่มขึ้นเรื่อยๆ
class LeakyCallback(BaseCallbackHandler):
    def __init__(self):
        self.all_results = []  # ไม่มีวันถูกลบ!
        
    def on_task_complete(self, task, result):
        self.all_results.append({"task": task, "result": result})


✅ วิธีที่ถูก: ใช้ Deque ที่มี Max Length และ Periodic Cleanup

from collections import deque
from datetime import datetime, timedelta
import threading

class MemoryEfficientCallback(BaseCallbackHandler):
    MAX_QUEUE_SIZE = 1000  # จำกัดจำนวน records
    CLEANUP_INTERVAL = 3600  # Cleanup ทุก 1 ชั่วโมง
    RETENTION_PERIOD = timedelta(days=7)  # เก็บข้อมูล 7 วัน
    
    def __init__(self):
        self._results = deque(maxlen=self.MAX_QUEUE_SIZE)
        self._lock = threading.Lock()
        self._last_cleanup = datetime.now()
        
    def on_task_complete(self, task, result):
        with self._lock:
            self._results.append({
                "task_id": task.id,
                "result": result,
                "timestamp": datetime.now(),
                "success": self._evaluate_result(result)
            })
            
            # Periodic cleanup
            if (datetime.now() - self._last_cleanup).seconds > self.CLEANUP_INTERVAL:
                self._cleanup_old_records()
                
    def _cleanup_old_records(self):
        """ลบ records ที่เก่ากว่า Retention Period"""
        cutoff = datetime.now() - self.RETENTION_PERIOD
        self._results = deque(
            [r for r in self._results if r["timestamp"] > cutoff],
            maxlen=self.MAX_QUEUE_SIZE
        )
        self._last_cleanup = datetime.now()

4. Rate Limiting ไม่ได้จัดการ ทำให้ API ถูก Block

# ❌ วิธีที่ผิด: เรียก API ต่อเนื่องโดยไม่ควบคุม Rate
def run_all_tasks(agents):
    results = []
    for agent in agents:
        results.append(agent.execute_task())  # อาจถูก Rate Limit!


✅ วิธีที่ถูก: Implement Rate Limiter ด้วย Token Bucket Algorithm

import time
import threading
from typing import Callable, Any

class TokenBucketRateLimiter:
    """Token Bucket Algorithm สำหรับควบคุม Rate Limit"""
    
    def __init__(self, rate: float, capacity: int):
        self.rate = rate  # tokens per second
        self.capacity = capacity
        self.tokens = capacity
        self.last_update = time.time()
        self.lock = threading.Lock()
        
    def acquire(self, tokens: int = 1) -> float:
        """ขอ Token และคืนค่า wait time (วินาที) ถ้าต้องรอ"""
        with self.lock:
            now = time.time()
            elapsed = now - self.last_update
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last_update = now
            
            if self.tokens >= tokens:
                self.tokens -= tokens
                return 0.0
            else:
                wait_time = (tokens - self.tokens) / self.rate
                return wait_time

class HolySheepRateLimiter:
    """Rate Limiter สำหรับ HolySheep API (50 requests/minute typical)"""
    
    def __init__(self):
        self.limiter = TokenBucketRateLimiter(rate=0.8, capacity=10)  # 48/min
        self._semaphore = threading.Semaphore(3)  # Max 3 concurrent
        
    def execute_with_limit(self, func: Callable, *args, **kwargs) -> Any:
        """Execute function พร้อมควบคุม Rate Limit"""
        with self._semaphore:
            wait_time = self.limiter.acquire()
            if wait_time > 0:
                print(f"[RATE LIMIT] Waiting {wait_time:.2f}s")
                time.sleep(wait_time)
            return func(*args, **kwargs)


การใช้งาน

rate_limiter = HolySheepRateLimiter()

def execute_agent_task(agent, task):
    return rate_limiter.execute_with_limit(agent.execute_task, task)

สรุป

การ Monitor Agent Task Success Rates ใน CrewAI นั้นไม่ใช่แค่การดูว่า Task สำเร็จหรือไม่ แต่เป็นการสร้างระบบที่สามารถตอบคำถามสำคัญได้ เช่น Agent ไหนทำงานได้ดีที่สุด Model ไหนให้ผลลัพธ์คุ้มค่าที่สุด และปัญหาอะไรที่ต้องแก้ไข ด้วย Callback System ที่ออกแบบมาอย่างดีและ Dashboard ที่เหมาะสม คุณจะสามารถ Deploy Multi-Agent System ที่เสถียรและมีประสิทธิภาพส

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

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