การสร้างเอเจนต์ AI ที่ทำงานอัตโนมัตินั้น สิ่งสำคัญที่สุดไม่ใช่แค่การทำให้มันทำงานได้ แต่คือการทำให้มัน รู้วิธีแก้ปัญหาเมื่อเกิดข้อผิดพลาด ในบทความนี้ ผมจะสอนคุณตั้งแต่พื้นฐานว่าระบบแก้ไขข้อผิดพลาดทำงานอย่างไร และจะสาธิตการเขียนโค้ดจริงๆ ให้คุณดูแบบละเอียด
ทำไมระบบแก้ไขข้อผิดพลาดถึงสำคัญ?
ลองนึกภาพว่าคุณสร้างเอเจนต์ AI ที่คอยตอบลูกค้าแทนคุณ วันดีคืนดีเซิร์ฟเวอร์ล่ม เน็ตเวิร์กขาด หรือ API ตอบกลับช้าผิดปกติ ถ้าเอเจนต์ของคุณไม่มีระบบจัดการข้อผิดพลาด มันจะหยุดทำงานทันที แต่ถ้ามีระบบที่ดี เอเจนต์จะ ลองใหม่อัตโนมัติ หรือ ย้อนกลับไปใช้วิธีที่เคยใช้ได้ และแจ้งเตือนคุณเมื่อต้องการความช่วยเหลือ
สำหรับผู้เริ่มต้น เราจะแบ่งระบบแก้ไขข้อผิดพลาดออกเป็น 3 ระดับ:
- ระดับ 1: การลองใหม่อัตโนมัติ (Retry) — เมื่อเกิดปัญหาเล็กน้อย เช่น เน็ตเวิร์กชั่วคราว ให้ลองส่งคำขอใหม่
- ระดับ 2: การย้อนกลับ (Rollback) — เมื่อการทำงานล้มเหลว ให้กลับไปสู่สถานะที่ปลอดภัยก่อนหน้า
- ระดับ 3: การเรียกมนุษย์ช่วย (Human-in-the-loop) — เมื่อปัญหาใหญ่เกินกว่าจะแก้เอง ให้แจ้งผู้ดูแลระบบ
เริ่มต้น: เตรียมเครื่องมือและบัญชี
ก่อนจะเขียนโค้ด คุณต้องมีบัญชี API ก่อน ผมแนะนำให้ใช้ สมัครที่นี่ เพราะมีความเร็วต่ำกว่า 50 มิลลิวินาที แถมอัตราแลกเปลี่ยนคุ้มค่ามาก ราคาถูกกว่าผู้ให้บริการอื่นถึง 85% และรองรับการจ่ายผ่าน WeChat และ Alipay อีกด้วย เมื่อสมัครเสร็จ คุณจะได้ API Key มาใช้งาน
โครงสร้างพื้นฐานของระบบแก้ไขข้อผิดพลาด
ให้เราสร้างไฟล์ Python ง่ายๆ ที่ชื่อ agent_error_handler.py เพื่อเรียนรู้หลักการทีละขั้น
ขั้นตอนที่ 1: ส่งคำขอพื้นฐานไปยัง HolySheep API
ก่อนจะมีระบบแก้ไขข้อผิดพลาด เราต้องมีโค้ดสำหรับส่งคำขอพื้นฐานก่อน นี่คือตัวอย่างที่ใช้งานได้จริง:
import requests
import time
import json
ตั้งค่าการเชื่อมต่อ API
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
สร้างฟังก์ชันสำหรับส่งข้อความไปยัง AI
def send_message_to_ai(message):
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
data = {
"model": "gpt-4.1",
"messages": [
{"role": "user", "content": message}
],
"temperature": 0.7
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=data,
timeout=30
)
return response.json()
ทดสอบการใช้งาน
result = send_message_to_ai("สวัสดีครับ")
print(result)ขั้นตอนที่ 2: เพิ่มระบบลองใหม่อัตโนมัติ (Retry)
นี่คือหัวใจของระบบแก้ไขข้อผิดพลาด เราจะสร้างฟังก์ชันที่จะลองส่งคำขอใหม่เมื่อล้มเหลว:
import requests
import time
from datetime import datetime
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class AgentError(Exception):
"""คลาสสำหรับเก็บข้อผิดพลาดของเอเจนต์"""
pass
def send_with_retry(message, max_retries=3, wait_seconds=2):
"""
ส่งข้อความไปยัง AI พร้อมระบบลองใหม่อัตโนมัติ
max_retries: จำนวนครั้งสูงสุดที่จะลองใหม่
wait_seconds: ระยะเวลารอก่อนลองใหม่ (วินาที)
"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
data = {
"model": "gpt-4.1",
"messages": [
{"role": "user", "content": message}
],
"temperature": 0.7
}
# ลองส่งคำขอ โดยเริ่มจากครั้งที่ 1
for attempt in range(1, max_retries + 1):
try:
print(f"[{datetime.now()}] พยายามส่งคำขอครั้งที่ {attempt}/{max_retries}")
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=data,
timeout=30
)
# ตรวจสอบว่าสำเร็จหรือไม่
if response.status_code == 200:
return response.json()
else:
print(f"❌ ได้รับข้อผิดพลาด: {response.status_code}")
print(f" ข้อความ: {response.text}")
except requests.exceptions.Timeout:
print(f"⏰ คำขอหมดเวลา (Timeout)")
except requests.exceptions.ConnectionError:
print(f"🔌 ไม่สามารถเชื่อมต่อได้")
except requests.exceptions.RequestException as e:
print(f"⚠️ เกิดข้อผิดพลาด: {str(e)}")
# ถ้าไม่ใช่ครั้งสุดท้าย ให้รอก่อนลองใหม่
if attempt < max_retries:
print(f" รอ {wait_seconds} วินาที แล้วลองใหม่...\n")
time.sleep(wait_seconds)
# เพิ่มระยะเวลารอเป็น 2 เท่า (Exponential Backoff)
wait_seconds *= 2
# ถ้าลองครบทุกครั้งแล้วยังล้มเหลว
raise AgentError(f"ส่งคำขอล้มเหลวหลังจากลอง {max_retries} ครั้ง")
ทดสอบการใช้งาน
try:
result = send_with_retry("อธิบายเรื่อง AI Agent")
print("✅ สำเร็จ!")
print(result)
except AgentError as e:
print(f"❌ เอเจนต์ล้มเหลว: {str(e)}")ขั้นตอนที่ 3: เพิ่มระบบย้อนกลับ (Rollback)
บางครั้งการลองใหม่ไม่ช่วยได้ เช่น ข้อมูลที่ส่งไปผิดพลาด หรือ AI ตอบกลับมาผิด format เราต้องมีระบบย้อนกลับไปสู่สถานะปลอดภัย:
import requests
import json
import time
from datetime import datetime
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class AgentState:
"""จัดการสถานะของเอเจนต์ เพื่อรองรับการย้อนกลับ"""
def __init__(self):
self.history = [] # เก็บประวัติการทำงาน
self.current_state = {} # สถานะปัจจุบัน
self.last_valid_state = None # สถานะที่ใช้งานได้ครั้งล่าสุด
def save_checkpoint(self, task_id, data):
"""บันทึกสถานะปัจจุบัน"""
checkpoint = {
"task_id": task_id,
"timestamp": datetime.now().isoformat(),
"data": data
}
self.history.append(checkpoint)
self.last_valid_state = checkpoint.copy()
print(f"💾 บันทึกสถานะ: Task {task_id}")
def rollback(self):
"""ย้อนกลับไปสู่สถานะที่ใช้งานได้ครั้งล่าสุด"""
if self.last_valid_state:
print(f"🔄 ย้อนกลับไปสถานะ: Task {self.last_valid_state['task_id']}")
self.current_state = self.last_valid_state["data"].copy()
return self.last_valid_state
else:
print("⚠️ ไม่มีสถานะที่จะย้อนกลับ")
return None
สร้างเอเจนต์ที่มีระบบย้อนกลับ
class SmartAgent:
def __init__(self):
self.state_manager = AgentState()
self.max_retries = 3
def process_task(self, task_message):
"""ประมวลผลงานพร้อมระบบแก้ไขข้อผิดพลาด"""
# สร้าง Task ID
task_id = f"task_{int(time.time())}"
# บันทึกสถานะเริ่มต้น
self.state_manager.save_checkpoint(task_id, {
"status": "started",
"message": task_message
})
for attempt in range(1, self.max_retries + 1):
try:
# ส่งคำขอไปยัง API
result = self._call_api(task_message)
# ตรวจสอบผลลัพธ์
if self._validate_result(result):
# บันทึกสถานะเมื่อสำเร็จ
self.state_manager.save_checkpoint(task_id, {
"status": "completed",
"result": result
})
return result
else:
raise ValueError("ผลลัพธ์ไม่ถูกต้อง")
except Exception as e:
print(f"❌ ครั้งที่ {attempt} ล้มเหลว: {str(e)}")
if attempt < self.max_retries:
# ย้อนกลับไปสถานะที่ใช้งานได้
self.state_manager.rollback()
time.sleep(2 ** attempt) # รอแบบ Exponential
else:
# แจ้งว่าต้องการความช่วยเหลือจากมนุษย์
return self._escalate_to_human(task_id