ในฐานะวิศวกรที่ดูแลระบบ AI infrastructure มาหลายปี ผมเชื่อว่า Webhook คือหัวใจสำคัญของการสร้างระบบ Real-time AI application ที่เสถียร วันนี้จะพาทุกคนมาดูวิธีตั้งค่า Webhook event subscription กับ HolySheep AI อย่างละเอียด ครอบคลุมตั้งแต่พื้นฐานจนถึง production-ready configuration
ทำไมต้องใช้ Webhook Event Subscription
Webhook ช่วยให้เรารับ events แบบ real-time โดยไม่ต้อง poll API ตลอดเวลา ลด latency และประหยัด cost อย่างมาก สำหรับ use cases ที่ผมพบบ่อยใน production:
- Streaming Response — รับ token-by-token response แบบ SSE (Server-Sent Events)
- Progress Tracking — ติดตามสถานะ long-running tasks
- Cost Logging — บันทึก token usage อัตโนมัติทุก request
- Error Handling — รับ notification เมื่อเกิดปัญหา
สถาปัตยกรรม Webhook System
ระบบ Webhook ของ HolySheep AI ใช้ HTTPS POST callback model ที่มี retry mechanism และ signature verification ในตัว ต่างจาก OpenAI official API ที่ต้องใช้ Event Stream API แยกต่างหาก
{
"event": "chat.completion",
"timestamp": "2025-01-15T10:30:00.123Z",
"request_id": "req_abc123xyz",
"data": {
"model": "gpt-4o",
"usage": {
"prompt_tokens": 150,
"completion_tokens": 342,
"total_tokens": 492
},
"latency_ms": 847
}
}การตั้งค่า Webhook Endpoint
1. สร้าง Webhook Handler
โค้ดด้านล่างเป็น production-ready webhook handler ที่ผมใช้จริงในระบบหลายตัว รองรับ concurrent requests และมี error handling ครบ
# webhook_handler.py
import asyncio
import hashlib
import hmac
import json
import logging
from datetime import datetime, timedelta
from typing import Any, Callable
from dataclasses import dataclass, field
from collections import defaultdict
import aiohttp
from aiohttp import web
from aiohttp.web_abc import Request, Response
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class WebhookConfig:
secret: str = "YOUR_WEBHOOK_SECRET"
max_retries: int = 5
retry_delay: float = 1.0
signature_tolerance_seconds: int = 300
@dataclass
class EventMetrics:
received_count: int = 0
success_count: int = 0
failed_count: int = 0
total_latency_ms: float = 0.0
events_by_type: dict = field(default_factory=lambda: defaultdict(int))
last_event_time: datetime | None = None
class WebhookProcessor:
def __init__(self, config: WebhookConfig):
self.config = config
self.metrics = EventMetrics()
self._event_buffer: list[dict] = []
self._buffer_lock = asyncio.Lock()
self._processing = False
def verify_signature(self, payload: bytes, signature: str, timestamp: str) -> bool:
"""ตรวจสอบ webhook signature ป้องกันการปลอมแปลง"""
try:
ts = int(timestamp)
now = int(datetime.utcnow().timestamp())
if abs(now - ts) > self.config.signature_tolerance_seconds:
logger.warning(f"Timestamp out of tolerance: {timestamp}")
return False
expected = hmac.new(
self.config.secret.encode(),
f"{timestamp}.{payload.decode()}".encode(),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(f"sha256={expected}", signature)
except (ValueError, TypeError) as e:
logger.error(f"Signature verification error: {e}")
return False
async def handle_raw_request(self, request: Request) -> Response:
"""Entry point สำหรับ aiohttp webhook route"""
start_time = datetime.utcnow()
# Extract headers
signature = request.headers.get("X-Webhook-Signature", "")
timestamp = request.headers.get("X-Webhook-Timestamp", "")
event_type = request.headers.get("X-Webhook-Event", "unknown")
# Read body
body = await request.read()
# Verify signature
if not self.verify_signature(body, signature, timestamp):
self.metrics.failed_count += 1
return Response(status=401, text="Invalid signature")
# Parse payload
try:
payload = json.loads(body)
except json.JSONDecodeError as e:
logger.error(f"JSON parse error: {e}")
self.metrics.failed_count += 1
return Response(status=400, text="Invalid JSON")
# Process event asynchronously
asyncio.create_task(self._process_event(event_type, payload))
# Update metrics
self.metrics.received_count += 1
self.metrics.events_by_type[event_type] += 1
self.metrics.last_event_time = datetime.utcnow()
latency = (datetime.utcnow() - start_time).total_seconds() * 1000
self.metrics.total_latency_ms += latency
return Response(status=200, text=json.dumps({"status": "accepted"}))
async def _process_event(self, event_type: str, payload: dict):
"""Process event แบบ non-blocking พร้อม buffer สำหรับ batch processing"""
async with self._buffer_lock:
self._event_buffer.append({
"type": event_type,
"data": payload,
"received_at": datetime.utcnow().isoformat()
})
# Process batch when buffer reaches threshold
if len(self._event_buffer) >= 10:
await self._flush_buffer()
async def _flush_buffer(self):
"""Flush buffered events to storage/database"""
if not self._event_buffer:
return
events = self._event_buffer.copy()
self._event_buffer.clear()
logger.info(f"Flushing {len(events)} events to storage")
# Implement actual storage logic here
# e.g., write to PostgreSQL, send to Kafka, etc.
for event in events:
await self._store_event(event)
self.metrics.success_count += len(events)
async def _store_event(self, event: dict):
"""Store single event — implement based on your storage needs"""
# Placeholder for actual storage implementation
logger.debug(f"Storing event: {event['type']}")
def get_metrics(self) -> dict:
"""ดึง metrics สำหรับ monitoring"""
avg_latency = (self.metrics.total_latency_ms / self.metrics.received_count
if self.metrics.received_count > 0 else 0)
return {
"total_received": self.metrics.received_count,
"total_success": self.metrics.success_count,
"total_failed": self.metrics.failed_count,
"average_latency_ms": round(avg_latency, 2),
"events_by_type": dict(self.metrics.events_by_type),
"last_event_time": self.metrics.last_event_time.isoformat() if self.metrics.last_event_time else None,
"buffer_size": len(self._event_buffer)
}
Initialize application
config = WebhookConfig()
processor = WebhookProcessor(config)
app = web.Application()
app.router.add_post("/webhook", processor.handle_raw_request)
app.router.add_get("/webhook/metrics", lambda r: web.json_response(processor.get_metrics()))
if __name__ == "__main__":
web.run_app(app, host="0.0.0.0", port=8443)2. Frontend Client สำหรับ Streaming
ด้านล่างเป็น client ที่รองรับ streaming response ผ่าน SSE และส่งต่อไปยัง webhook สำหรับ logging อัตโนมัติ
# streaming_client.py
import asyncio
import json
import time
from typing import AsyncGenerator, Callable
from dataclasses import dataclass
import aiohttp
@dataclass
class StreamingConfig:
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
base_url: str = "https://api.holysheep.ai/v1"
model: str = "gpt-4o"
webhook_url: str = "https://your-server.com/webhook"
timeout: int = 120
max_retries: int = 3
class HolySheepStreamingClient:
def __init__(self, config: StreamingConfig):
self.config = config
self._session: aiohttp.ClientSession | None = None
self._request_count = 0
self._total_tokens = 0
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=self.config.timeout)
self._session = aiohttp.ClientSession(timeout=timeout)
return self
async def __aexit__(self, *args):
if self._session:
await self._session.close()
async def stream_chat_completion(
self,
messages: list[dict],
on_token: Callable[[str], None] | None = None,
webhook_metadata: dict | None = None
) -> AsyncGenerator[str, None]:
"""
Stream response จาก HolySheep API พร้อม token-by-token callback
"""
headers = {
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.config.model,
"messages": messages,
"stream": True,
"stream_options": {
"include_usage": True
}
}
if webhook_metadata:
payload["webhook"] = self.config.webhook_url
payload["webhook_events_filter"] = ["message_delta", "input_usage", "completed"]
request_start = time.perf_counter()
full_content = []
try:
async with self._session.post(
f"{self.config.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
if response.status != 200:
error_text = await response.text()
raise RuntimeError(f"API error {response.status}: {error_text}")
async for line in response.content:
line = line.decode("utf-8").strip()
if not line or not line.startswith("data: "):
continue
data = line[6:] # Remove "data: " prefix
if data == "[DONE]":
break
try:
event = json.loads(data)
delta = event.get("choices", [{}])[0].get("delta", {})
if "content" in delta:
token = delta["content"]
full_content.append(token)
if on_token:
on_token(token)
yield token
except json.JSONDecodeError:
continue
except aiohttp.ClientError as e:
raise RuntimeError(f"Connection error: {e}")
finally:
elapsed = time.perf_counter() - request_start
self._request_count += 1
self._total_tokens += len(full_content)
print(f"[Metrics] Request #{self._request_count}: "
f"{len(full_content)} tokens in {elapsed:.2f}s "
f"({len(full_content)/elapsed:.1f} tok/s)")
def get_stats(self) -> dict:
"""ดึงสถิติการใช้งาน"""
return {
"total_requests": self._request_count,
"total_tokens": self._total_tokens,
"avg_tokens_per_request": self._total_tokens / max(self._request_count, 1)
}
Example usage with streaming
async def main():
config = StreamingConfig(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="gpt-4o",
webhook_url="https://your-server.com/webhook"
)
messages = [
{"role": "system", "content": "คุณคือผู้ช่วย AI ที่เป็นมิตร"},
{"role": "user", "content": "อธิบายเรื่อง Webhook อย่างละเอียด"}
]
async with HolySheepStreamingClient(config) as client:
print("Streaming response:")
collected = []
async for token in client.stream_chat_completion(messages):
print(token, end="", flush=True)
collected.append(token)
print("\n\n--- Stats ---")
print(client.get_stats())
if __name__ == "__main__":
asyncio.run(main())การควบคุม Concurrency และ Rate Limiting
สำหรับ production system ที่ต้องรับ traffic สูง ผมแนะนำให้ใช้ semaphore และ token bucket algorithm เพื่อควบคุม concurrency อย่างเข้มงวด
# concurrent_streaming_manager.py
import asyncio
import time
from typing import AsyncGenerator
from dataclasses import dataclass, field
from collections import deque
from datetime import datetime, timedelta
@dataclass
class RateLimiter:
"""Token bucket rate limiter สำหรับ API calls"""
max_tokens: int
refill_rate: float # tokens per second
_tokens: float = field(init=False)
_last_refill: datetime = field(init=False)
def __post_init__(self):
self._tokens = float(self.max_tokens)
self._last_refill = datetime.utcnow()
def _refill(self):
now = datetime.utcnow()
elapsed = (now - self._last_refill).total_seconds()
self._tokens = min(
self._max if hasattr(self, '_max') else self.max_tokens,
self._tokens + elapsed * self.refill_rate
)
self._last_refill = now
async def acquire(self, tokens: int = 1):
while True:
self._refill()
if self._tokens >= tokens:
self._tokens -= tokens
return
await asyncio.sleep(0.1)
@dataclass
class ConcurrencyLimiter:
"""Semaphore-based concurrency limiter พร้อม queue"""
max_concurrent: int
_semaphore: asyncio.Semaphore = field(init=False)
_active_count: int = 0
_lock: asyncio.Lock = field(init=False)
def __post_init__(self):
self._semaphore = asyncio.Semaphore(self.max_concurrent)
self._lock = asyncio.Lock()
async def __aenter__(self):
await self._semaphore.acquire()
async with self._lock:
self._active_count += 1
return self
async def __aexit__(self, *args):
self._semaphore.release()
async with self._lock:
self._active_count -= 1
@property
def active_count(self) -> int:
return self._active_count
class StreamingManager:
"""Manager สำหรับจัดการ multiple streaming requests"""
def __init__(
self,
max_concurrent: int = 10,
rate_limit: int = 60, # requests per minute
burst_size: int = 10
):
self.limiter = ConcurrencyLimiter(max_concurrent)
self.rate_limiter = RateLimiter(burst_size, rate_limit / 60)
self._metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"rejected_requests": 0,
"avg_queue_time": 0,
"peak_concurrent": 0
}
self._metrics_lock = asyncio.Lock()
self._queue_times: deque = deque(maxlen=1000)
async def execute_streaming(
self,
coro,
timeout: float = 300
) -> tuple[bool, any]:
"""
Execute streaming coroutine พร้อม concurrency และ rate limiting
Returns (success, result_or_error)
"""
queue_start = time.perf_counter()
# Check if we can accept more concurrent requests
if self.limiter.active_count >= self.limiter.max_concurrent:
async with self._metrics_lock