บทนำ: ช่วงเวลาที่ทุกอย่างพังทลาย
หลายเดือนก่อน ผมเคยประสบกับหายนะครั้งใหญ่กับระบบ AI Agent ที่ทีมพัฒนาขึ้นมาเอง ตอนนั้น Agent ตัวหนึ่งได้รับอนุญาตให้ส่งอีเมลถึงลูกค้าทั้งหมด 5,000 รายโดยไม่ผ่านการตรวจสอบจากมนุษย์ ผลลัพธ์คืออีเมลที่มีข้อมูลราคาผิดพลาดถูกส่งออกไปทั้งหมด แม้ว่าตัว Agent จะทำงานได้อย่างมีประสิทธิภาพ แต่การขาด "มนุษย์คอยควบคุม" ทำให้เกิดความเสียหายร้ายแรง
นี่คือจุดเริ่มต้นที่ผมเริ่มศึกษา Human-in-the-Loop (HITL) อย่างจริงจัง และวันนี้ผมจะมาแบ่งปันประสบการณ์การออกแบบระบบอนุมัติที่เหมาะกับ AI Agent ในยุคปัจจุบัน
ทำความเข้าใจ Human-in-the-Loop Architecture
Human-in-the-Loop คือรูปแบบการออกแบบระบบที่มนุษย์เข้ามามีส่วนร่วมในกระบวนการตัดสินใจของ AI Agent ในจุดที่สำคัญ การออกแบบนี้มีความจำเป็นอย่างยิ่งสำหรับงานที่มีความเสี่ยงสูง เช่น การส่งอีเมล การทำธุรกรรมทางการเงิน หรือการแก้ไขข้อมูลสำคัญ
หลักการพื้นฐานมี 3 ข้อที่ผมยึดถือมาโดยตลอด ประการแรกคือ "Escalation Threshold" หรือเกณฑ์การส่งต่อ กำหนดว่าการกระทำใดต้องมีการอนุมัติ ประการที่สองคือ "Approval Queue" คือระบบจัดการคิวการอนุมัติที่ชัดเจน และประการที่สามคือ "Timeout Handling" หรือการจัดการเมื่อไม่ได้รับการตอบกลับภายในเวลาที่กำหนด
การติดตั้งระบบ Human-in-the-Loop ด้วย HolySheep AI
ในการสร้างระบบ HITL ผมแนะนำให้ใช้
สมัครที่นี่ เพื่อเริ่มต้นใช้งาน HolySheep AI ซึ่งมีความเร็วในการตอบสนองต่ำกว่า 50 มิลลิวินาที และรองรับ Webhook สำหรับการแจ้งเตือนแบบเรียลไทม์
import requests
import json
import time
from enum import Enum
from typing import Optional, List, Dict, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
class ApprovalStatus(Enum):
PENDING = "pending"
APPROVED = "approved"
REJECTED = "rejected"
TIMEOUT = "timeout"
CANCELLED = "cancelled"
class RiskLevel(Enum):
LOW = 1 # ดำเนินการอัตโนมัติ
MEDIUM = 2 # แจ้งเตือนแต่ดำเนินการได้
HIGH = 3 # ต้องได้รับการอนุมัติ
CRITICAL = 4 # ต้องได้รับการอนุมัติจากผู้บริหาร
@dataclass
class ApprovalRequest:
request_id: str
agent_id: str
action_type: str
action_details: Dict[str, Any]
risk_level: RiskLevel
status: ApprovalStatus = ApprovalStatus.PENDING
created_at: datetime = field(default_factory=datetime.now)
approved_by: Optional[str] = None
approved_at: Optional[datetime] = None
rejection_reason: Optional[str] = None
timeout_seconds: int = 300
@dataclass
class HITLConfig:
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
risk_thresholds: Dict[RiskLevel, int] = field(default_factory=lambda: {
RiskLevel.LOW: 1,
RiskLevel.MEDIUM: 2,
RiskLevel.HIGH: 3,
RiskLevel.CRITICAL: 4
})
default_timeout: int = 300
class HumanInTheLoopAgent:
def __init__(self, config: Optional[HITLConfig] = None):
self.config = config or HITLConfig()
self.pending_approvals: Dict[str, ApprovalRequest] = {}
self.approval_history: List[ApprovalRequest] = []
self.webhook_url = None
def set_webhook(self, webhook_url: str):
"""ตั้งค่า Webhook URL สำหรับการแจ้งเตือน"""
self.webhook_url = webhook_url
def assess_risk(self, action: Dict[str, Any]) -> RiskLevel:
"""ประเมินระดับความเสี่ยงของการกระทำ"""
recipient_count = action.get("recipient_count", 1)
action_type = action.get("type", "")
data_modification = action.get("modifies_data", False)
financial_impact = action.get("estimated_cost", 0)
# คำนวณคะแนนความเสี่ยง
risk_score = 0
if recipient_count > 100:
risk_score += 2
elif recipient_count > 10:
risk_score += 1
if action_type in ["send_email", "send_sms", "send_notification"]:
risk_score += 1
if data_modification:
risk_score += 2
if financial_impact > 1000:
risk_score += 2
elif financial_impact > 100:
risk_score += 1
# แปลงคะแนนเป็น RiskLevel
if risk_score >= 5:
return RiskLevel.CRITICAL
elif risk_score >= 3:
return RiskLevel.HIGH
elif risk_score >= 2:
return RiskLevel.MEDIUM
else:
return RiskLevel.LOW
def create_approval_request(
self,
agent_id: str,
action_type: str,
action_details: Dict[str, Any]
) -> ApprovalRequest:
"""สร้างคำขออนุมัติใหม่"""
risk_level = self.assess_risk(action_details)
request_id = f"APR-{int(time.time() * 1000)}"
# กำหนด timeout ตามระดับความเสี่ยง
timeout_map = {
RiskLevel.LOW: 60,
RiskLevel.MEDIUM: 120,
RiskLevel.HIGH: 300,
RiskLevel.CRITICAL: 600
}
request = ApprovalRequest(
request_id=request_id,
agent_id=agent_id,
action_type=action_type,
action_details=action_details,
risk_level=risk_level,
timeout_seconds=timeout_map[risk_level]
)
self.pending_approvals[request_id] = request
self._send_notification(request)
return request
def _send_notification(self, request: ApprovalRequest):
"""ส่งการแจ้งเตือนไปยังผู้อนุมัติ"""
if not self.webhook_url:
return
message = {
"request_id": request.request_id,
"action_type": request.action_type,
"risk_level": request.risk_level.name,
"details": request.action_details,
"created_at": request.created_at.isoformat(),
"timeout_seconds": request.timeout_seconds
}
try:
response = requests.post(
self.webhook_url,
json=message,
headers={"Content-Type": "application/json"},
timeout=5
)
print(f"ส่งการแจ้งเตือนสำเร็จ: {response.status_code}")
except requests.exceptions.RequestException as e:
print(f"การส่งการแจ้งเตือนล้มเหลว: {e}")
def approve(self, request_id: str, approver_id: str) -> bool:
"""อนุมัติคำขอ"""
if request_id not in self.pending_approvals:
return False
request = self.pending_approvals[request_id]
request.status = ApprovalStatus.APPROVED
request.approved_by = approver_id
request.approved_at = datetime.now()
self.approval_history.append(request)
del self.pending_approvals[request_id]
return True
def reject(self, request_id: str, rejector_id: str, reason: str) -> bool:
"""ปฏิเสธคำขอพร้อมเหตุผล"""
if request_id not in self.pending_approvals:
return False
request = self.pending_approvals[request_id]
request.status = ApprovalStatus.REJECTED
request.approved_by = rejector_id
request.approved_at = datetime.now()
request.rejection_reason = reason
self.approval_history.append(request)
del self.pending_approvals[request_id]
return True
def check_timeout(self) -> List[ApprovalRequest]:
"""ตรวจสอบคำขอที่หมดเวลา"""
timeout_requests = []
current_time = datetime.now()
for request_id, request in list(self.pending_approvals.items()):
elapsed = (current_time - request.created_at).total_seconds()
if elapsed > request.timeout_seconds:
request.status = ApprovalStatus.TIMEOUT
self.approval_history.append(request)
del self.pending_approvals[request_id]
timeout_requests.append(request)
return timeout_requests
def execute_with_approval(
self,
agent_id: str,
action_type: str,
action_details: Dict[str, Any],
callback_function: callable
) -> Dict[str, Any]:
"""ดำเนินการพร้อมระบบอนุมัติ"""
request = self.create_approval_request(agent_id, action_type, action_details)
# ถ้าระดับความเสี่ยงต่ำ ดำเนินการได้เลย
if request.risk_level == RiskLevel.LOW:
self.approve(request.request_id, "SYSTEM_AUTO")
return {"status": "executed", "request_id": request.request_id}
# รอการอนุมัติ (Poll หรือ Callback)
start_time = time.time()
while time.time() - start_time < request.timeout_seconds:
time.sleep(1)
if request.request_id not in self.pending_approvals:
if request.status == ApprovalStatus.APPROVED:
result = callback_function(request.action_details)
return {
"status": "approved_and_executed",
"request_id": request.request_id,
"result": result
}
elif request.status == ApprovalStatus.REJECTED:
return {
"status": "rejected",
"request_id": request.request_id,
"reason": request.rejection_reason
}
return {
"status": "timeout",
"request_id": request.request_id
}
ตัวอย่างการใช้งาน
def send_email_callback(details):
"""ฟังก์ชันสำหรับส่งอีเมลจริง"""
print(f"กำลังส่งอีเมลไปยัง {details['recipients']}")
# ที่นี่ควรเรียก API จริงสำหรับการส่งอีเมล
return {"sent_count": details["recipient_count"]}
เริ่มต้นใช้งาน
hitl_agent = HumanInTheLoopAgent()
hitl_agent.set_webhook("https://your-app.com/webhook/approvals")
ตัวอย่าง: ส่งอีเมลพร้อมระบบอนุมัติ
email_action = {
"type": "send_email",
"recipients": ["[email protected]", "[email protected]"],
"recipient_count": 2,
"subject": "แจ้งเตือนสำคัญ",
"body": "เนื้อหาอีเมล...",
"modifies_data": False,
"estimated_cost": 0
}
result = hitl_agent.execute_with_approval(
agent_id="marketing-agent-001",
action_type="email_broadcast",
action_details=email_action,
callback_function=send_email_callback
)
print(f"ผลลัพธ์: {json.dumps(result, indent=2, ensure_ascii=False)}")
การสร้างระบบ Web Dashboard สำหรับผู้อนุมัติ
เพื่อให้ผู้อนุมัติสามารถตรวจสอบและอนุมัติคำขอได้สะดวก ผมได้พัฒนา Dashboard แบบง่ายที่เชื่อมต่อกับ HolySheep AI API ซึ่งมีความเร็วตอบสนองต่ำกว่า 50 มิลลิวินาที ทำให้การโหลดข้อมูลรวดเร็วมาก
from flask import Flask, request, jsonify, render_template_string
from threading import Thread
import queue
app = Flask(__name__)
คิวสำหรับการแจ้งเตือนแบบเรียลไทม์
notification_queue = queue.Queue()
HTML Template สำหรับ Dashboard
DASHBOARD_HTML = '''
Human-in-the-Loop Dashboard