AI Agent กลไก Self-Reflection และการแก้ไขข้อผิดพลาดด้วยตนเอง

ผมเคยเจอสถานการณ์ที่ทำให้เสียเวลาหลายชั่วโมง — Agent ตัวหนึ่งที่สร้างขึ้นให้ลูกค้า รันไปได้ 50 ขั้นตอน แล้วล้มเหลวตั้งแต่ขั้นตอนที่ 47 เพราะผลลัพธ์จากขั้นตอนก่อนหน้าผิดพลาด แก้ยังไงก็ไม่จบ จนกระทั่งผมเพิ่มกลไก Self-Reflection เข้าไป — ตั้งแต่นั้น Agent ก็ "รู้ตัวเอง" ว่าทำอะไรผิด และแก้ไขได้โดยไม่ต้องรอคนมาช่วย

ทำไม AI Agent ต้องมี Reflection Mechanism

AI Agent ที่ทำงานหลายขั้นตอน — เช่น วิเคราะห์ข้อมูล, ตัดสินใจ, ดำเนินการ — มีความเสี่ยงที่ข้อผิดพลาดจะสะสมไปเรื่อยๆ ยิ่งทำงานนาน ยิ่งเบี่ยงเบนจากเป้าหมาย กลไก Reflection ช่วยให้ Agent สามารถ:

• ประเมินผลลัพธ์ของตัวเองว่าถูกต้องหรือไม่
• ตรวจพบความผิดปกติก่อนที่จะลุกลาม
• วางแผนแก้ไขและดำเนินการใหม่โดยอัตโนมัติ

สถาปัตยกรรมพื้นฐาน: Plan-Act-Reflect Loop

การสร้าง Self-Correcting Agent อาศัย Loop สามขั้นตอน:

┌─────────────────────────────────────────────┐
│              REFLECTION LOOP                  │
├─────────────────────────────────────────────┤
│                                             │
│   ┌──────────┐    ┌──────────┐    ┌──────────┐
│   │   PLAN   │───▶│   ACT    │───▶│ REFLECT  │
│   └──────────┘    └──────────┘    └──────────┘
│        ▲                            │        │
│        └────────────────────────────┘        │
│              (Self-Correction)               │
└─────────────────────────────────────────────┘

ในโค้ดจริง ผมใช้ HolySheep AI API เป็นหลัก เพราะความหน่วงต่ำกว่า 50 มิลลิวินาที ทำให้ Loop ทำงานเร็วมาก ราคาถูกกว่า 85% เมื่อเทียบกับ OpenAI โดยคุณสามารถ สมัครที่นี่ เพื่อรับเครดิตฟรีเมื่อลงทะเบียน

การสร้าง Self-Reflecting Agent ด้วย HolySheep API

1. โครงสร้างพื้นฐานของ Agent

import requests
import json
from typing import List, Dict, Any

class SelfReflectingAgent:
    def __init__(self, api_key: str, max_retries: int = 3):
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = api_key
        self.max_retries = max_retries
        self.conversation_history = []
        self.reflection_log = []

    def _call_llm(self, messages: List[Dict], model: str = "gpt-4.1") -> str:
        """เรียก LLM ผ่าน HolySheep API"""
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": model,
                "messages": messages,
                "temperature": 0.7
            },
            timeout=30
        )
        
        if response.status_code == 401:
            raise Exception("401 Unauthorized: ตรวจสอบ API Key ของคุณ")
        elif response.status_code == 429:
            raise Exception("429 Rate Limited: รอสักครู่แล้วลองใหม่")
        elif response.status_code != 200:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
        
        return response.json()["choices"][0]["message"]["content"]

    def plan(self, task: str) -> Dict[str, Any]:
        """วางแผนการดำเนินการ"""
        system_prompt = """คุณเป็น AI Planner วางแผนการดำเนินการให้กับงานที่ได้รับ
แบ่งเป็นขั้นตอนที่ชัดเจน แต่ละขั้นตอนต้องมี:
1. รหัสขั้นตอน
2. คำอธิบาย
3. เกณฑ์ความสำเร็จ"""
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"วางแผนสำหรับ: {task}"}
        ]
        
        plan_text = self._call_llm(messages)
        return {"task": task, "plan": plan_text, "status": "planned"}

    def act(self, step: str, context: Dict) -> Dict[str, Any]:
        """ดำเนินการตามขั้นตอน"""
        system_prompt = """คุณเป็น AI Executor ดำเนินการตามขั้นตอนที่กำหนด
ตรวจสอบว่าผลลัพธ์ตรงกับเกณฑ์ความสำเร็จหรือไม่
หากพบปัญหา ให้บันทึกไว้ใน 'issues' field"""
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"ขั้นตอน: {step}\n\nบริบท: {json.dumps(context, ensure_ascii=False)}"}
        ]
        
        result = self._call_llm(messages)
        
        return {
            "step": step,
            "result": result,
            "timestamp": self._get_timestamp(),
            "issues": []  # จะถูกอัพเดทโดย reflection
        }

    def reflect(self, action_result: Dict, criteria: str) -> Dict[str, Any]:
        """สะท้อนและประเมินผลลัพธ์"""
        system_prompt = f"""คุณเป็น AI Reflector ประเมินผลลัพธ์ว่า:
1. บรรลุเป้าหมายหรือไม่
2. มีปัญหาอะไรบ้าง
3. ต้องแก้ไขอย่างไร

เกณฑ์ความสำเร็จ: {criteria}"""
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"ผลลัพธ์ที่ต้องประเมิน:\n{json.dumps(action_result, ensure_ascii=False, indent=2)}"}
        ]
        
        reflection = self._call_llm(messages)
        
        reflection_data = {
            "original_result": action_result,
            "reflection": reflection,
            "success": self._evaluate_success(reflection),
            "needs_correction": self._check_needs_correction(reflection)
        }
        
        self.reflection_log.append(reflection_data)
        return reflection_data

    def _evaluate_success(self, reflection: str) -> bool:
        """ตรวจสอบว่าสำเร็จหรือไม่"""
        success_indicators = ["สำเร็จ", "บรรลุ", "ถูกต้อง", "เสร็จสิ้น", "success", "complete"]
        failure_indicators = ["ล้มเหลว", "ผิดพลาด", "ไม่สำเร็จ", "failed", "error", "incorrect"]
        
        for indicator in success_indicators:
            if indicator.lower() in reflection.lower():
                for fail in failure_indicators:
                    if fail.lower() in reflection.lower():
                        return False
                return True
        return False

    def _check_needs_correction(self, reflection: str) -> bool:
        """ตรวจสอบว่าต้องแก้ไขหรือไม่"""
        correction_indicators = ["ต้องแก้ไข", "ควรแก้ไข", "ปรับปรุง", "แนะนำ", "needs correction"]
        for indicator in correction_indicators:
            if indicator in reflection:
                return True
        return False

    def correct_and_retry(self, failed_step: str, reflection: Dict) -> Dict:
        """แก้ไขและลองใหม่"""
        correction_prompt = f"""คุณเป็น AI Corrector แก้ไขข้อผิดพลาดจากขั้นตอนที่ล้มเหลว

ขั้นตอนที่ล้มเหลว: {failed_step}

การสะท้อนจากผลลัพธ์:
{reflection['reflection']}

ให้แก้ไขและดำเนินการใหม่"""
        
        messages = [{"role": "user", "content": correction_prompt}]
        corrected_result = self._call_llm(messages)
        
        return {
            "original": failed_step,
            "correction": corrected_result,
            "status": "corrected"
        }

    def _get_timestamp(self) -> str:
        from datetime import datetime
        return datetime.now().isoformat()

    def run(self, task: str, criteria: str, max_loop: int = 5) -> Dict:
        """Run Agent พร้อม Self-Correction Loop"""
        print(f"🎯 เริ่มงาน: {task}")
        
        plan = self.plan(task)
        print(f"📋 วางแผนเสร็จสิ้น")
        
        context = {"task": task}
        loop_count = 0
        
        while loop_count < max_loop:
            loop_count += 1
            print(f"\n🔄 Loop {loop_count}/{max_loop}")
            
            # Act
            action_result = self.act(plan["plan"], context)
            print(f"✅ ดำเนินการเสร็จสิ้น")
            
            # Reflect
            reflection = self.reflect(action_result, criteria)
            print(f"🔍 สะท้อนผล: {'สำเร็จ' if reflection['success'] else 'ต้องแก้ไข'}")
            
            if reflection["success"]:
                print("🎉 งานเสร็จสมบูรณ์!")
                return {
                    "status": "success",
                    "result": action_result,
                    "loops": loop_count,
                    "reflections": self.reflection_log
                }
            
            if reflection["needs_correction"]:
                print("🔧 กำลังแก้ไข...")
                corrected = self.correct_and_retry(
                    action_result["result"],
                    reflection
                )
                context["last_correction"] = corrected
                context["reflection_history"] = self.reflection_log
        
        return {
            "status": "max_loops_exceeded",
            "reflections": self.reflection_log
        }

2. การใช้งาน Agent พร้อมตัวอย่างจริง

# ตัวอย่างการใช้งาน - วิเคราะห์ข้อมูลลูกค้า
agent = SelfReflectingAgent(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    max_retries=3
)

task = """
วิเคราะห์ข้อมูลลูกค้าต่อไปนี้และจัดกลุ่ม:
- สมชาย อายุ 35 ซื้อสินค้า A 5 ครั้ง/เดือน
- สมหญิง อายุ 28 ซื้อสินค้า B 2 ครั้ง/เดือน
- วิชัย อายุ 42 ซื้อสินค้า A 6 ครั้ง/เดือน
จัดกลุ่มตามพฤติกรรมการซื้อ
"""

criteria = """
1. จัดกลุ่มลูกค้าได้ถูกต้องตามเกณฑ์ที่กำหนด
2. แต่ละกลุ่มมีลักษณะเฉพาะที่ชัดเจน
3. มีคำแนะนำสำหรับแต่ละกลุ่ม
"""

result = agent.run(task, criteria)

print(f"\n📊 สถานะ: {result['status']}")
print(f"🔄 ใช้งานไป: {result.get('loops', 0)} loops")
print(f"📝 ผลลัพธ์: {result['result']['result']}")

เทคนิคขั้นสูง: ReAct + Self-Correction

สำหรับงานที่ซับซ้อน ผมแนะนำ ReAct (Reasoning + Acting) ที่รวมกับ Self-Correction โดยใช้ DeepSeek V3.2 ซึ่งราคาถูกมากเพียง $0.42 ต่อล้านโทเค็น ผ่าน HolySheep AI

class ReActAgent(SelfReflectingAgent):
    """ReAct Agent พร้อม Self-Correction"""
    
    def react_step(self, thought: str, action: str, observation: str) -> Dict:
        """ดำเนินการ ReAct แต่ละขั้น"""
        
        react_prompt = f"""คุณเป็น AI ที่ใช้ ReAct framework

ขั้นตอนปัจจุบัน:
Thought: {thought}
Action: {action}
Observation: {observation}

ให้:
1. วิเคราะห์ว่า Observation ตรงกับที่คาดหวังหรือไม่
2. ถ้าผิด ให้อธิบายว่าผิดอย่างไร และแนะนำวิธีแก้ไข
3. ตัดสินใจขั้นตอนถัดไป

ตอบเป็น JSON format:
{{
    "thought": "ความคิดของคุณ",
    "action": "การกระทำถัดไป",
    "observation_valid": true/false,
    "correction_needed": true/false,
    "correction_plan": "แผนแก้ไขถ้าต้องแก้"
}}"""
        
        messages = [{"role": "user", "content": react_prompt}]
        response = self._call_llm(messages, model="deepseek-v3.2")
        
        import json
        try:
            parsed = json.loads(response)
            return parsed
        except:
            return {
                "thought": thought,
                "action": action,
                "observation_valid": True,
                "correction_needed": False
            }
    
    def run_with_tools(self, task: str, tools: List[Dict]) -> Dict:
        """Run ReAct พร้อมเครื่องมือ"""
        
        context = {"task": task, "history": []}
        max_steps = 10
        
        for step in range(max_steps):
            print(f"\n📍 ขั้นตอนที่ {step + 1}")
            
            # สร้าง Thought และ Action
            thought_action_prompt = f"""งาน: {task}

ประวัติการดำเนินการ:
{json.dumps(context['history'], ensure_ascii=False, indent=2)}

เครื่องมือที่มี:
{json.dumps(tools, ensure_ascii=False, indent=2)}

ให้ Thought และ Action ถัดไป
ตอบเป็น JSON: {{"thought": "...", "action": "tool_name", "params": {{}}}}"""
            
            messages = [{"role": "user", "content": thought_action_prompt}]
            decision = self._call_llm(messages)
            
            try:
                decision = json.loads(decision)
            except:
                decision = {"thought": decision, "action": "finish", "params": {}}
            
            print(f"💭 Thought: {decision.get('thought', '')[:100]}...")
            
            if decision["action"] == "finish":
                return {
                    "status": "completed",
                    "steps": step + 1,
                    "history": context["history"]
                }
            
            # ดำเนินการ (จำลอง)
            observation = f"ผลลัพธ์จาก {decision['action']}: Success"
            
            # Reflect
            react_result = self.react_step(
                decision["thought"],
                decision["action"],
                observation
            )
            
            context["history"].append({
                "step": step + 1,
                "thought": decision["thought"],
                "action": decision["action"],
                "observation": observation,
                "reflection": react_result
            })
            
            # Self-Correction ถ้าจำเป็น
            if react_result.get("correction_needed"):
                print(f"🔧 พบปัญหา: {react_result.get('correction_plan', '')}")
                # ปรับ Action ถัดไปตามแผนแก้ไข
        
        return {
            "status": "max_steps_exceeded",
            "history": context["history"]
        }

ตัวอย่างการใช้งาน
react_agent = ReActAgent(api_key="YOUR_HOLYSHEEP_API_KEY")

tools = [
    {"name": "search_data", "description": "ค้นหาข้อมูลในฐานข้อมูล"},
    {"name": "analyze", "description": "วิเคราะห์ข้อมูล"},
    {"name": "report", "description": "สร้างรายงาน"}
]

result = react_agent.run_with_tools(
    "วิเคราะห์ยอดขายประจำเดือน และหาแนวโน้ม",
    tools
)

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

1. 401 Unauthorized Error

# ❌ ข้อผิดพลาด
ConnectionError: timeout หรือ 401 Unauthorized

✅ วิธีแก้ไข
class SelfReflectingAgent:
    def _call_llm(self, messages: List[Dict], model: str = "gpt-4.1") -> str:
        """เรียก LLM พร้อม Error Handling ที่ดีขึ้น"""
        
        for attempt in range(3):
            try:
                response = requests.post(
                    f"{self.base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {self.api_key}",
                        "Content-Type": "application/json"
                    },
                    json={
                        "model": model,
                        "messages": messages,
                        "temperature": 0.7
                    },
                    timeout=30
                )
                
                # ตรวจสอบ HTTP Status
                if response.status_code == 401:
                    raise ValueError(
                        "401 Unauthorized: ตรวจสอบ API Key ของคุณ\n"
                        "1. ไปที่ https://www.holysheep.ai/settings/api-keys\n"
                        "2. สร้าง Key ใหม่\n"
                        "3. แทนที่ YOUR_HOLYSHEEP_API_KEY ในโค้ด"
                    )
                
                elif response.status_code == 429:
                    wait_time = 2 ** attempt  # Exponential backoff
                    print(f"⏳ Rate limited. รอ {wait_time} วินาที...")
                    time.sleep(wait_time)
                    continue
                
                elif response.status_code >= 500:
                    print(f"⚠️ Server error: {response.status_code}. ลองใหม่...")
                    continue
                
                elif response.status_code != 200:
                    raise Exception(
                        f"API Error: {response.status_code}\n"
                        f"Response: {response.text}"
                    )
                
                return response.json()["choices"][0]["message"]["content"]
                
            except requests.exceptions.Timeout:
                print(f"⏰ Timeout attempt {attempt + 1}/3")
                if attempt == 2:
                    raise TimeoutError(
                        "Connection timeout: ตรวจสอบการเชื่อมต่ออินเทอร์เน็ต\n"
                        "หรือลองใช้ timeout ที่มากขึ้น"
                    )
            except requests.exceptions.ConnectionError as e:
                print(f"🔌 Connection error: {e}")
                if attempt == 2:
                    raise ConnectionError(
                        "ไม่สามารถเชื่อมต่อ API ได้\n"
                        "ตรวจสอบ:\n"
                        "1. URL ถูกต้องหรือไม่ (ต้องเป็น api.holysheep.ai)\n"
                        "2. Firewall บล็อกหรือไม่\n"
                        "3. API ปกติหรือไม่ ที่ https://www.holysheep.ai/status"
                    )
        
        raise Exception("เกินจำนวนครั้งที่ลองใหม่")

2. Reflection Loop ติด Infinite Loop

# ❌ ข้อผิดพลาด: Agent วนไม่รู้จบเพราะไม่สามารถแก้ไขตัวเองได้

✅ วิธีแก้ไข: เพิ่ม Circuit Breaker และ Escalation
class SelfReflectingAgent:
    def __init__(self, api_key: str, max_retries: int = 3):
        # ... existing code ...
        self.circuit_breaker = {
            "failures": 0,
            "threshold": 5,
            "open": False,
            "reset_timeout": 60
        }
        self.escalation_history = []
    
    def _check_circuit_breaker(self):
        """ตรวจสอบ Circuit Breaker"""
        if self.circuit_breaker["open"]:
            raise Exception(
                "Circuit Breaker เปิดอยู่: "
                f"เกิดข้อผิดพลาดติดต่อกัน {self.circuit_breaker['threshold']} ครั้ง\n"
                "ระบบหยุดทำงานชั่วคราวเพื่อป้องกันความเสียหาย\n"
                "ลองใหม่ในอีก 60 วินาที"
            )
    
    def _trip_circuit_breaker(self):
        """เปิด Circuit Breaker"""
        self.circuit_breaker["failures"] += 1
        if self.circuit_breaker["failures"] >= self.circuit_breaker["threshold"]:
            self.circuit_breaker["open"] = True
            print(f"🔴 Circuit Breaker เปิดแล้ว!")
    
    def _reset_circuit_breaker(self):
        """รีเซ็ต Circuit Breaker"""
        self.circuit_breaker["failures"] = 0
        self.circuit_breaker["open"] = False
    
    def escalate(self, problem: Dict) -> str:
        """Escalation เมื่อ Self-Correction ไม่ได้ผล"""
        escalation_prompt = f"""มีปัญหาที่ Agent ไม่สามารถแก้ไขเองได้

ปัญหา:
{json.dumps(problem, ensure_ascii=False, indent=2)}

ประวัติการพยายามแก้ไข:
{json.dumps(self.escalation_history, ensure_ascii=False, indent=2)}

ให้แนะนำว่ามนุษย์ควรทำอย่างไร"""
        
        messages = [{"role": "user", "content": escalation_prompt}]
        recommendation = self._call_llm(messages)
        
        self.escalation_history.append({
            "problem": problem,
            "recommendation": recommendation,
            "timestamp": self._get_timestamp()
        })
        
        return f"""
⚠️ ต้องการความช่วยเหลือจากมนุษย์

ปัญหา: {problem.get('description', 'Unknown')}
คำแนะนำ: {recommendation}

กรุณาตรวจสอบและดำเนินการ
"""
    
    def run(self, task: str, criteria: str, max_loop: int = 5) -> Dict:
        """Run พร้อม Circuit Breaker และ Escalation"""
        
        self._check_circuit_breaker()
        
        try:
            # ... existing run logic ...
            result = self._execute_run_logic(task, criteria, max_loop)
            self._reset_circuit_breaker()  # สำเร็จ รีเซ็ต
            return result
            
        except Exception as e:
            self._trip_circuit_breaker()
            
            # ถ้า Circuit Breaker เปิด ให้ Escalate
            if self.circuit_breaker["open"]:
                return {
                    "status": "human_intervention_required",
                    "escalation": self.escalate({
                        "description": str(e),
                        "task": task,
                        "loops_attempted": max_loop
                    })
                }
            
            raise  # Re-raise ถ้ายังไม่ถึง threshold

3. Token Limit เกินขณะทำ Reflection

# ❌ ข้อผิดพลาด: context window exceeded

✅ วิธีแก้ไข: Smart Context Management
class SelfReflectingAgent:
    def __init__(self, api_key: str, max_context_tokens: int = 8000):
        self.max_context_tokens = max_context_tokens
        self.conversation_history = []
        self.summary_history = []  # เก็บ summary ของ history เก่า
    
    def _summarize_history(self, messages: List[Dict]) -> List[Dict]:
        """สรุป conversation history ที่ยาวเกินไป"""
        
        if self._count_tokens(messages) <= self.max_context_tokens:
            return messages
        
        # สร้าง summary
        summary_prompt = f"""สรุป conversation นี้ให้กระชับ โดยเก็บ:
1. ข้อมูลสำคัญที่ต้องจำ
2. ขั้นตอนที่ทำไปแล้ว
3. ปัญหาที่พบและวิธีแก้ไข

Conversation:
{json.dumps(messages, ensure_ascii=False)}"""
        
        messages_for_summary = [{"role": "user", "content": summary_prompt}]
        summary = self._
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
รีวิวฉบับเต็ม: AI สรุปข้อความยาว เปรียบเทียบราคา-ความเร็วระห