ในยุคที่ AI ต้องทำงานร่วมกับระบบอื่นอย่างไร้รอยต่อ Function Calling คือความสามารถที่เปลี่ยนโปรเจกต์ AI จาก "แชทบอท" ให้กลายเป็น "ผู้ช่วยอัตโนมัติ" ที่สามารถเรียก API ภายนอก ดึงข้อมูลเรียลไทม์ และ trigger webhook ได้ตามต้องการ บทความนี้จะพาคุณสำรวจการทำ webhook integration ด้วย Function Calling อย่างละเอียด พร้อมโค้ดตัวอย่างที่รันได้จริงผ่าน HolySheep AI
Function Calling คืออะไร และทำไมต้องใช้กับ Webhook
Function Calling คือความสามารถของโมเดล AI ที่ให้คุณกำหนด "ฟังก์ชัน" ที่ AI สามารถเรียกใช้เมื่อต้องการข้อมูลหรือทำงานบางอย่าง แทนที่จะตอบกลับเป็นข้อความธรรมดา AI จะส่ง JSON object ที่มีชื่อฟังก์ชันและ arguments กลับมาให้คุณ execute แล้วส่งผลลัพธ์กลับไป
Webhook ตอบรับคือ endpoint ที่คุณสร้างไว้เพื่อรับ request จากบริการภายนอก เช่น การแจ้งเตือนการชำระเงิน การอัปเดตสถานะออร์เดอร์ หรือ sensor data เมื่อรวมกับ Function Calling แล้ว AI จะสามารถประมวลผลข้อมูลที่ webhook รับมาแล้วตอบโต้หรือทำงานต่อได้ทันที
สถาปัตยกรรมระบบ Webhook + Function Calling
ก่อนเข้าสู่การทดสอบ มาดูสถาปัตยกรรมที่เราจะใช้กัน
┌─────────────────────────────────────────────────────────────────┐
│ สถาปัตยกรรม Webhook + Function Calling │
├─────────────────────────────────────────────────────────────────┤
│ │
│ [External Service] ──POST webhook──► [Your Server] │
│ │ │
│ ▼ │
│ [Parse + Validate Payload] │
│ │ │
│ ▼ │
│ [Call HolySheep AI API] │
│ base_url: │
│ https://api.holysheep.ai/v1 │
│ │ │
│ ▼ │
│ [Function Calling Response] │
│ │ │
│ ▼ │
│ [Execute Function / Trigger Actions] │
│ │
└─────────────────────────────────────────────────────────────────┘การทดสอบและเกณฑ์การประเมิน
เราทดสอบระบบ webhook integration โดยใช้ HolySheep AI เป็น AI provider หลัก โดยกำหนดเกณฑ์ดังนี้
เกณฑ์การประเมิน
- ความหน่วง (Latency) — เวลาตอบสนองเฉลี่ยจาก webhook รับ payload จนได้ function call response
- อัตราความสำเร็จ — เป็นเปอร์เซ็นต์ที่ AI ตอบกลับด้วย valid function call JSON ที่ execute ได้
- ความถูกต้องของ JSON Schema — function call ที่ส่งกลับมาตรงกับ schema ที่กำหนดหรือไม่
- ความสะดวกในการ implement — ง่ายหรือยากแค่ไหนในการตั้งค่าและ deploy
- ความครอบคลุมของโมเดล — รองรับโมเดลใดบ้างในการใช้ Function Calling
การตั้งค่า Server สำหรับ Webhook
เริ่มต้นด้วยการสร้าง webhook server อย่างง่ายด้วย Flask ที่จะรับ payload จากบริการภายนอกแล้วส่งต่อไปยัง HolySheep AI
# webhook_server.py
Webhook server สำหรับรับ payload และเรียก Function Calling
รันด้วย: python webhook_server.py
import os
import json
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
ตั้งค่า HolySheep API
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
กำหนด functions ที่พร้อมให้ AI เรียกใช้
FUNCTIONS = [
{
"type": "function",
"function": {
"name": "send_notification",
"description": "ส่งการแจ้งเตือนไปยังผู้ใช้ผ่านระบบ notification",
"parameters": {
"type": "object",
"properties": {
"user_id": {
"type": "string",
"description": "ID ของผู้ใช้ที่จะได้รับการแจ้งเตือน"
},
"message": {
"type": "string",
"description": "ข้อความที่จะส่ง"
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high"],
"description": "ระดับความเร่งด่วนของการแจ้งเตือน"
}
},
"required": ["user_id", "message"]
}
}
},
{
"type": "function",
"function": {
"name": "update_order_status",
"description": "อัปเดตสถานะออร์เดอร์ในระบบ",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "หมายเลขออร์เดอร์"
},
"new_status": {
"type": "string",
"enum": ["pending", "confirmed", "shipped", "delivered", "cancelled"],
"description": "สถานะใหม่ของออร์เดอร์"
},
"notes": {
"type": "string",
"description": "หมายเหตุเพิ่มเติม (optional)"
}
},
"required": ["order_id", "new_status"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate_shipping_fee",
"description": "คำนวณค่าจัดส่งตามน้ำหนักและปลายทาง",
"parameters": {
"type": "object",
"properties": {
"weight_kg": {
"type": "number",
"description": "น้ำหนักสินค้าเป็นกิโลกรัม"
},
"province": {
"type": "string",
"description": "จังหวัดปลายทาง"
},
"shipping_method": {
"type": "string",
"enum": ["standard", "express", "same_day"],
"description": "วิธีการจัดส่ง"
}
},
"required": ["weight_kg", "province"]
}
}
}
]
Function execution handlers
def execute_send_notification(user_id: str, message: str, priority: str = "medium") -> dict:
"""Execute notification sending - เชื่อมต่อกับ notification service จริง"""
# ใน production จะเชื่อมต่อกับ FCM, Line Notify, หรือ SMS gateway
return {
"success": True,
"notification_id": f"notif_{user_id}_{int(time.time())}",
"delivered_at": datetime.now().isoformat()
}
def execute_update_order_status(order_id: str, new_status: str, notes: str = None) -> dict:
"""Execute order status update - เชื่อมต่อกับ database จริง"""
# ใน production จะ update ลง database
return {
"success": True,
"order_id": order_id,
"new_status": new_status,
"updated_at": datetime.now().isoformat()
}
def execute_calculate_shipping_fee(weight_kg: float, province: str, shipping_method: str = "standard") -> dict:
"""Execute shipping fee calculation"""
# อัตราค่าจัดส่งตัวอย่าง
base_rates = {
"standard": 50,
"express": 120,
"same_day": 250
}
province_multipliers = {
"กรุงเทพมหานคร": 1.0,
"นนทบุรี": 1.1,
"ปทุมธานี": 1.1,
"ชลบุรี": 1.3,
"เชียงใหม่": 1.4
}
base = base_rates.get(shipping_method, 50)
multiplier = province_multipliers.get(province, 1.5)
weight_charge = weight_kg * 15 # 15 บาทต่อกิโลกรัม
total_fee = (base + weight_charge) * multiplier
return {
"weight_kg": weight_kg,
"province": province,
"shipping_method": shipping_method,
"total_fee_thb": round(total_fee, 2),
"estimated_days": {"standard": 3, "express": 1, "same_day": 4}[shipping_method]
}
Mapping function names to handlers
FUNCTION_HANDLERS = {
"send_notification": execute_send_notification,
"update_order_status": execute_update_order_status,
"calculate_shipping_fee": execute_calculate_shipping_fee
}
@app.route("/webhook", methods=["POST"])
def handle_webhook():
"""
Webhook endpoint สำหรับรับ payload จากบริการภายนอก
"""
import time
from datetime import datetime
try:
payload = request.get_json()
event_type = payload.get("type", "unknown")
# สร้าง system prompt ที่บอก AI ว่า webhook event นี้คืออะไร
system_prompt = f"""คุณคือ AI assistant ที่ทำงานร่วมกับ webhook system
เมื่อได้รับ webhook payload คุณต้องวิเคราะห์ event และเรียก function ที่เหมาะสม
ถ้า event ต้องการการแจ้งเตือน → ใช้ send_notification
ถ้า event เกี่ยวกับออร์เดอร์ → ใช้ update_order_status
ถ้า event เกี่ยวกับการจัดส่ง → ใช้ calculate_shipping_fee
ตอบกลับด้วย JSON ที่มี function_call เท่านั้น ห้ามตอบเป็นข้อความธรรมดา"""
# สร้าง user message จาก payload
user_message = f"Webhook Event: {json.dumps(payload, ensure_ascii=False, indent=2)}"
# เรียก HolySheep API
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1", # ราคา $8/MTok รองรับ Function Calling เต็มรูปแบบ
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
],
"functions": FUNCTIONS,
"temperature": 0.1
},
timeout=30
)
result = response.json()
# ตรวจสอบว่า AI เรียก function หรือไม่
if "choices" in result and len(result["choices"]) > 0:
choice = result["choices"][0]
if "message" in choice:
message = choice["message"]
# กรณีมี function_call
if "function_call" in message:
function_name = message["function_call"]["name"]
arguments = json.loads(message["function_call"]["arguments"])
# Execute function
if function_name in FUNCTION_HANDLERS:
func_result = FUNCTION_HANDLERS[function_name](**arguments)
return jsonify({
"status": "success",
"function_called": function_name,
"arguments": arguments,
"result": func_result,
"ai_response": message
})
# กรณี AI ตอบเป็นข้อความ (ไม่มี function call)
return jsonify({
"status": "no_function_call",
"ai_message": message.get("content"),
"ai_response": message
})
return jsonify({"status": "error", "message": "Invalid API response", "raw": result}), 500
except Exception as e:
return jsonify({"status": "error", "message": str(e)}), 500
if __name__ == "__main__":
from datetime import datetime
import time
print("=" * 60)
print("Webhook Server with Function Calling")
print("Base URL: https://api.holysheep.ai/v1")
print("=" * 60)
app.run(host="0.0.0.0", port=5000, debug=True)ตัวอย่าง Webhook Payload และการทดสอบ
มาดูตัวอย่าง payload จากบริการต่างๆ และดูว่า AI จะเรียก function อะไร
# test_webhook.py
ทดสอบ webhook ด้วย payload ต่างๆ
รันด้วย: python test_webhook.py
import requests
import json
import time
from datetime import datetime
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
WEBHOOK_URL = "http://localhost:5000/webhook" # URL ของ webhook server
def test_webhook_scenario(scenario_name: str, payload: dict):
"""ทดสอบ webhook scenario ต่างๆ"""
print(f"\n{'='*60