บทนำ — ทำไม Tool-calling Agent ถึงสำคัญในยุค AI
ในฐานะนักพัฒนาที่ดูแลระบบ AI ของอีคอมเมิร์ซมากว่า 3 ปี ผมเคยเจอปัญหาหนักใจมากมายกับ Chatbot แบบเดิมที่ตอบคำถามลูกค้าได้แค่ข้อมูลคงที่ใน knowledge base วันนี้ผมจะมาแชร์ประสบการณ์ตรงในการสร้าง Tool-calling Agent ที่ทำให้ระบบลูกค้าสัมพันธ์ของเราตอบคำถามสินค้า ตรวจสอบสต็อก และประมวลผลคำสั่งซื้อได้อัตโนมัติ ลดงาน support ลงถึง 60%
บทความนี้ใช้
HolySheep AI เป็นตัวอย่างหลัก เพราะมี latency ต่ำกว่า 50 มิลลิวินาที ราคาประหยัดกว่า OpenAI ถึง 85% และรองรับ function calling เต็มรูปแบบ
กรณีศึกษา: AI ลูกค้าสัมพันธ์อีคอมเมิร์ซที่พุ่งสูง
สถานการณ์จริงที่ผมเจอคือ ร้านค้าออนไลน์มียอดขายพุ่งสูงช่วงแคมเปญ 11.11 ระบบ support เดิมรับไม่ไหว ลูกค้าถามเรื่อง:
- สินค้านี้มีสต็อกไหม? ขนาด M สีดำ
- ติดตามพัสดุหมายเลข TH123456789
- ยกเลิกคำสั่งซื้อ #A9999 ได้ไหม
- เปลี่ยนที่อยู่จัดส่งเป็นใหม่
แต่ละคำถามต้องเรียก API ต่างกัน ต้องมีการ authenticate ต่างกัน ผลลัพธ์ก็ format ไม่เหมือนกัน ปัญหาเหล่านี้ Tool-calling Agent แก้ได้หมด
1. ความเข้าใจพื้นฐานเรื่อง Tool-calling และ Function Calling
Tool-calling คือความสามารถของ LLM ในการ "เรียกใช้ฟังก์ชันภายนอก" เมื่อจำเป็น แทนที่จะพยายามตอบจากข้อมูลที่มีอยู่ใน training data
Function Calling คือ protocol ที่กำหนดว่า:
- AI จะส่ง parameter อะไรไปให้ฟังก์ชัน
- เราจะ parse ผลลัพธ์อย่างไร
- ส่งข้อมูลกลับไปให้ AI ตอบต่ออย่างไร
ผมเข้าใจแนวคิดนี้ผิดไปหลายเดือน คิดว่า AI จะ "รันโค้ด" เอง แต่จริงๆ แล้ว AI แค่บอกว่า "ฉันต้องการเรียกฟังก์ชัน X กับ parameter Y" แล้วเราเป็นคนรันเอง
2. การตั้งค่า HolySheep API สำหรับ Tool-calling
import requests
import json
from typing import List, Dict, Any, Optional
class HolySheepToolAgent:
"""Agent ที่ใช้ HolySheep API พร้อม function calling"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.chat_endpoint = f"{self.base_url}/chat/completions"
def chat(
self,
messages: List[Dict[str, str]],
tools: Optional[List[Dict[str, Any]]] = None,
model: str = "gpt-4o"
) -> Dict[str, Any]:
"""ส่งข้อความไปยัง HolySheep พร้อม tools definition"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": 0.7
}
# เพิ่ม tools ถ้ามี
if tools:
payload["tools"] = tools
payload["tool_choice"] = "auto"
response = requests.post(
self.chat_endpoint,
headers=headers,
json=payload,
timeout=30
)
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code} - {response.text}")
return response.json()
ตัวอย่างการใช้งาน
agent = HolySheepToolAgent(api_key="YOUR_HOLYSHEEP_API_KEY")
print("✅ เชื่อมต่อ HolySheep API สำเร็จ — latency < 50ms")
ค่า model ที่แนะนำสำหรับ tool-calling:
- GPT-4.1 — $8/MTok (ประสิทธิภาพสูงสุด รองรับ function calling ดีที่สุด)
- Claude Sonnet 4.5 — $15/MTok (reasoning แม่นยำ)
- DeepSeek V3.2 — $0.42/MTok (คุ้มค่าที่สุด ประหยัด 85%+ จาก OpenAI)
3. การกำหนด Tools และ Parameters Schema
การออกแบบ tools schema ที่ดีเป็นหัวใจสำคัญ ผมใช้เวลาปรับแก้หลายรอบจนเจอ pattern ที่ work ดี
# Tool definitions สำหรับระบบ E-commerce Support
ecommerce_tools = [
{
"type": "function",
"function": {
"name": "check_product_stock",
"description": "ตรวจสอบสต็อกสินค้าตาม SKU, ขนาด และสี",
"parameters": {
"type": "object",
"properties": {
"sku": {
"type": "string",
"description": "รหัสสินค้า SKU เช่น SHIRT-001"
},
"size": {
"type": "string",
"description": "ไซส์: XS, S, M, L, XL, XXL",
"enum": ["XS", "S", "M", "L", "XL", "XXL"]
},
"color": {
"type": "string",
"description": "สี: black, white, red, blue, green"
}
},
"required": ["sku", "size"]
}
}
},
{
"type": "function",
"function": {
"name": "track_shipment",
"description": "ติดตามพัสดุตามหมายเลข tracking",
"parameters": {
"type": "object",
"properties": {
"tracking_number": {
"type": "string",
"description": "หมายเลขติดตามพัสดุ 13-15 หลัก"
}
},
"required": ["tracking_number"]
}
}
},
{
"type": "function",
"function": {
"name": "cancel_order",
"description": "ยกเลิกคำสั่งซื้อ (เฉพาะสถานะ pending หรือ confirmed)",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "หมายเลขคำสั่งซื้อ เช่น #A9999"
},
"reason": {
"type": "string",
"description": "เหตุผลในการยกเลิก"
}
},
"required": ["order_id", "reason"]
}
}
},
{
"type": "function",
"function": {
"name": "update_shipping_address",
"description": "เปลี่ยนที่อยู่จัดส่ง (เฉพาะสถานะ pending)",
"parameters": {
"type": "object",
"properties": {
"order_id": {"type": "string"},
"new_address": {
"type": "object",
"properties": {
"name": {"type": "string"},
"phone": {"type": "string"},
"address_line1": {"type": "string"},
"address_line2": {"type": "string"},
"city": {"type": "string"},
"postal_code": {"type": "string"}
},
"required": ["name", "phone", "address_line1", "city", "postal_code"]
}
},
"required": ["order_id", "new_address"]
}
}
}
]
ตัวอย่าง messages format
messages = [
{"role": "system", "content": "คุณคือ AI ผู้ช่วยดูแลลูกค้าอีคอมเมิร์ซ"},
{"role": "user", "content": "สินค้า SHIRT-001 ไซส์ M สีดำ มีของไหม?"}
]
4. การ Implement Function Handler และ Tool Execution
import re
from typing import Callable, Dict, Any
class ToolExecutor:
"""จัดการการ execute functions ตามที่ AI เรียก"""
def __init__(self):
# Mock database สำหรับ demo
self.stock_db = {
("SHIRT-001", "M", "black"): 45,
("SHIRT-001", "M", "white"): 12,
("SHIRT-001", "S", "black"): 0, # หมดสต็อก
("PANTS-002", "L", "blue"): 89,
}
self.orders_db = {
"#A9999": {"status": "pending", "customer": "สมชาย"},
"#B8888": {"status": "shipped", "customer": "สมหญิง"},
}
# Map function name -> handler
self.handlers: Dict[str, Callable] = {
"check_product_stock": self._check_stock,
"track_shipment": self._track,
"cancel_order": self._cancel,
"update_shipping_address": self._update_address,
}
def execute(self, function_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""execute function ตาม name และ arguments ที่ได้รับ"""
if function_name not in self.handlers:
return {
"success": False,
"error": f"Unknown function: {function_name}"
}
try:
# Parse JSON arguments จาก string ถ้าจำเป็น
if isinstance(arguments, str):
args = json.loads(arguments)
else:
args = arguments
return self.handlers[function_name](**args)
except json.JSONDecodeError as e:
return {"success": False, "error": f"Invalid JSON: {str(e)}"}
except TypeError as e:
return {"success": False, "error": f"Missing required parameter: {str(e)}"}
except Exception as e:
return {"success": False, "error": f"Execution error: {str(e)}"}
def _check_stock(self, sku: str, size: str, color: str = "black") -> Dict[str, Any]:
"""ตรวจสอบสต็อก"""
key = (sku, size, color)
quantity = self.stock_db.get(key, 0)
return {
"success": True,
"sku": sku,
"size": size,
"color": color,
"in_stock": quantity > 0,
"quantity": quantity,
"message": f"สินค้า {sku} ไซส์ {size} สี {color} {'มีของ ' + str(quantity) + ' ชิ้น' if quantity > 0 else 'หมดสต็อก'}"
}
def _track(self, tracking_number: str) -> Dict[str, Any]:
"""ติดตามพัสดุ"""
# Mock tracking data
if tracking_number.startswith("TH"):
return {
"success": True,
"tracking_number": tracking_number,
"status": "in_transit",
"eta": "2024-01-15",
"location": "ศูนย์คор.sorting กรุงเทพ",
"history": [
{"time": "2024-01-12 10:00", "status": "ออกจากศูนย์คัดแยก", "location": "กรุงเทพ"},
{"time": "2024-01-13 15:30", "status": "มาถึงศูนย์กระจาย", "location": "เชียงใหม่"},
]
}
return {"success": False, "error": "ไม่พบหมายเลขติดตามนี้"}
def _cancel(self, order_id: str, reason: str) -> Dict[str, Any]:
"""ยกเลิกคำสั่งซื้อ"""
order = self.orders_db.get(order_id)
if not order:
return {"success": False, "error": f"ไม่พบคำสั่งซื้อ {order_id}"}
if order["status"] not in ["pending", "confirmed"]:
return {"success": False, "error": f"ไม่สามารถยกเลิกได้ เนื่องจากสถานะ '{order['status']}'"}
order["status"] = "cancelled"
order["cancel_reason"] = reason
return {
"success": True,
"order_id": order_id,
"message": f"ยกเลิกคำสั่งซื้อ {order_id} สำเร็จแล้ว"
}
def _update_address(self, order_id: str, new_address: Dict) -> Dict[str, Any]:
"""อัพเดทที่อยู่จัดส่ง"""
order = self.orders_db.get(order_id)
if not order:
return {"success": False, "error": f"ไม่พบคำสั่งซื้อ {order_id}"}
if order["status"] != "pending":
return {"success": False, "error": "เปลี่ยนที่อยู่ได้เฉพาะคำสั่งซื้อที่ยังไม่จัดส่ง"}
return {
"success": True,
"order_id": order_id,
"new_address": new_address,
"message": f"อัพเดทที่อยู่สำหรับ {order_id} แล้ว"
}
5. Main Agent Loop — การประมวลผล Tool Calls
import json
from datetime import datetime
class EcommerceAgent:
"""Agent หลักที่จัดการ tool-calling loop"""
def __init__(self, api_key: str):
self.holy_sheep = HolySheepToolAgent(api_key)
self.executor = ToolExecutor()
self.tools = ecommerce_tools
self.max_iterations = 5
def chat(self, user_message: str) -> str:
"""รับข้อความจาก user และ return คำตอบ"""
messages = [
{
"role": "system",
"content": """คุณคือ AI ผู้ช่วยดูแลลูกค้าอีคอมเมิร์ซชื่อ "แม็คซ์"
ตอบเป็นภาษาไทย สุภาพ เข้าใจง่าย
ถ้าต้องการข้อมูลที่ต้องเรียกระบบ ให้ใช้ tools ที่มีให้
ถ้าลูกค้าถามเรื่องสต็อก ใช้ check_product_stock
ถ้าถามสถานะพัสดุ ใช้ track_shipment
ถ้าต้องการยกเลิก ใช้ cancel_order
ถ้าต้องเปลี่ยนที่อยู่ ใช้ update_shipping_address"""
},
{"role": "user", "content": user_message}
]
iteration = 0
while iteration < self.max_iterations:
iteration += 1
# เรียก HolySheep API
response = self.holy_sheep.chat(
messages=messages,
tools=self.tools
)
# ดึงข้อความจาก assistant
assistant_message = response["choices"][0]["message"]
messages.append(assistant_message)
# ตรวจสอบว่ามี tool_calls ไหม
if "tool_calls" not in assistant_message:
# ไม่มี tool_calls แปลว่าได้คำตอบแล้ว
return assistant_message["content"]
# มี tool_calls -> ต้อง execute แต่ละตัว
for tool_call in assistant_message["tool_calls"]:
function_name = tool_call["function"]["name"]
arguments = tool_call["function"]["arguments"]
print(f"🔧 เรียก function: {function_name}")
print(f"📋 Arguments: {arguments}")
# Execute function
result = self.executor.execute(function_name, arguments)
print(f"✅ Result: {json.dumps(result, ensure_ascii=False, indent=2)}")
# ส่งผลลัพธ์กลับไปเป็น tool message
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": json.dumps(result, ensure_ascii=False)
})
return "ขออภัย ดำเนินการใช้เวลานานเกินไป กรุณาลองใหม่อีกครั้ง"
ทดสอบการใช้งาน
agent = EcommerceAgent(api_key="YOUR_HOLYSHEEP_API_KEY")
ทดสอบคำถามต่างๆ
test_queries = [
"สินค้า SHIRT-001 ไซส์ M สีดำ มีของไหม?",
"ติดตามพัสดุหมายเลข TH123456789",
"ยกเลิกคำสั่งซื้อ #A9999 เพราะสั่งผิดไซส์",
]
for query in test_queries:
print(f"\n{'='*60}")
print(f"👤 ลูกค้า: {query}")
print(f"{'='*60}")
answer = agent.chat(query)
print(f"🤖 แม็คซ์: {answer}")
6. การ Handle Streaming และ Async Operations
สำหรับ production system ที่ต้องรองรับ concurrent users หลายร้อยคน streaming จะช่วยให้ UX ดีขึ้นมาก
import asyncio
import aiohttp
from typing import AsyncGenerator, Dict, Any
class AsyncEcommerceAgent:
"""Agent แบบ async รองรับ streaming"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.tools = ecommerce_tools
async def chat_stream(
self,
user_message: str,
session_id: str = None
) -> AsyncGenerator[str, None]:
"""Streaming response กลับมาทีละ token"""
messages = [
{"role": "system", "content": "คุณคือ AI ผู้ช่วย..."},
{"role": "user", "content": user_message}
]
async with aiohttp.ClientSession() as session:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4o",
"messages": messages,
"tools": self.tools,
"stream": True
}
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as resp:
accumulated = ""
async for line in resp.content:
line = line.decode('utf-8').strip()
if not line or line == "data: [DONE]":
continue
if line.startswith("data: "):
data = json.loads(line[6:])
if "choices" in data:
delta = data["choices"][0].get("delta", {})
if "content" in delta:
token = delta["content"]
accumulated += token
yield token
# ถ้ามี tool_calls
if "tool_calls" in delta:
yield "\n\n🔧 [ระบบกำลังประมวลผล...]"
# Parse tool calls จาก accumulated message
if "[TOOL_CALLS]" in accumulated:
# ดึง tool calls ออกมา execute
pass
async def demo_async():
"""Demo async streaming"""
agent = AsyncEcommerceAgent(api_key="YOUR_HOLYSHEEP_API_KEY")
print("🤖 แม็คซ์: ", end="", flush=True)
async for token in agent.chat_stream("สินค้า PANTS-002 มีของไหม?"):
print(token, end="", flush=True)
print("\n")
รัน async demo
asyncio.run(demo_async())
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
กรณีที่ 1: "Invalid API Key" หรือ Authentication Error
# ❌ สาเหตุ: API key ผิด หรือไม่ได้ใส่ "Bearer " prefix
response = requests.post(
url,
headers={"Authorization": "YOUR_HOLYSHEEP_API_KEY"}, # ผิด!
json=payload
)
✅ แก้ไข: ต้องใส่ "Bearer " นำหน้าเสมอ
response = requests.post(
url,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json=payload
)
หรือตรวจสอบว่า API key ถูกต้อง
def validate_api_key(api_key: str) -> bool:
"""ตรวจสอบ format API key"""
if not api_key or len(api_key) < 10:
return False
# HolySheep API key มี format เฉพาะ
if not api_key.startswith("sk-"):
return False
return True
กรณีที่ 2: Tool Response Format ไม่ถูกต้อง
# ❌ ผิดพลาด: ส่ง response ในรูปแบบ string ธรรมดา
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": "มีของ 5 ชิ้น" # ผิด!
})
✅ แก้ไข: ต้องส่งเป็น JSON string เท่านั้น
result = executor.execute(function_name, arguments)
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": json.dumps(result, ensure_ascii=False) # ถูกต้อง
})
ถ้า function คืนค่า dict ที่มีภาษาไทย
def safe_json_dumps(data: Dict) -> str:
"""แปลง dict เป็น JSON string อย่างปลอดภัย"""
return json.dumps(
data,
ensure_ascii=False, # รองรับภาษาไทย
indent=None, # ไม่ต้อง indent สำหรับ API
default=str # fallback สำหรับ object ที่ไม่ serializable
)