ในบทความนี้เราจะมาเรียนรู้วิธีการนำ Claude 4.6 Agent SDK ไปใช้งานจริงในองค์กร โดยเฉพาะการผสานรวมฟีเจอร์ที่สำคัญที่สุด 3 อย่าง ได้แก่ Tool Calling (การเรียกใช้เครื่องมือ) Memory Management (การจัดการความจำ) และ Planning/Reasoning (การวางแผนและเหตุผล) ซึ่งเป็นหัวใจสำคัญของการสร้าง AI Agent ที่ทำงานได้อย่างมีประสิทธิภาพในระดับ Production
เริ่มต้นจากกรณีศึกษาจริง: ระบบ AI สำหรับ E-commerce Customer Service
สมมติว่าคุณเป็นทีมพัฒนาของร้านค้าออนไลน์ขนาดใหญ่ ต้องการสร้าง AI Agent ที่สามารถตอบคำถามลูกค้า ค้นหาข้อมูลสินค้าในคลัง จัดการออเดอร์ และเชื่อมต่อกับระบบ CRM ได้ในตัว นี่คือกรณีที่เหมาะสมที่สุดสำหรับการใช้ Claude 4.6 Agent SDK แบบครบวงจร
การตั้งค่า Environment และการเชื่อมต่อ HolySheep API
ก่อนจะเริ่มเขียนโค้ด คุณต้องตั้งค่า Environment ให้พร้อมก่อน สำหรับโปรเจกต์นี้เราจะใช้ HolySheep AI สมัครที่นี่ เป็น API Provider ซึ่งให้บริการ Claude 4.6 ผ่าน OpenAI-compatible API ทำให้สามารถใช้งานได้ง่ายและประหยัดค่าใช้จ่ายถึง 85% เมื่อเทียบกับการใช้งานผ่าน Anthropic โดยตรง (อัตราเพียง $15/MTok สำหรับ Claude Sonnet 4.5)
# ติดตั้ง Dependencies ที่จำเป็น
pip install anthropic openai python-dotenv httpx aiofiles
สร้างไฟล์ .env
cat > .env << 'EOF'
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
EOF
ตรวจสอบความเร็วของ API (ควรได้ผลลัพธ์ต่ำกว่า 50ms)
import httpx
import time
start = time.time()
response = httpx.get("https://api.holysheep.ai/health", timeout=5.0)
latency = (time.time() - start) * 1000
print(f"API Latency: {latency:.2f}ms")
print(f"Status: {response.status_code}")ข้อดีของการใช้ HolySheep คือคุณสามารถชำระเงินผ่าน WeChat หรือ Alipay ได้เลย รวมถึงยังได้รับเครดิตฟรีเมื่อลงทะเบียน ทำให้สามารถเริ่มพัฒนาและทดสอบได้ทันทีโดยไม่ต้องเติมเงินก่อน
โครงสร้างพื้นฐานของ Claude 4.6 Agent
ในการสร้าง Agent ที่ทำงานได้จริง เราต้องออกแบบโครงสร้างให้รองรับ 3 องค์ประกอบหลัก ดังนี้
import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from openai import OpenAI
import json
from datetime import datetime
@dataclass
class Message:
role: str
content: str
timestamp: datetime = field(default_factory=datetime.now)
metadata: Dict[str, Any] = field(default_factory=dict)
class ConversationMemory:
"""ระบบจัดการความจำสำหรับ Agent"""
def __init__(self, max_messages: int = 50, summary_threshold: int = 30):
self.messages: List[Message] = []
self.max_messages = max_messages
self.summary_threshold = summary_threshold
self.summaries: List[str] = []
def add_message(self, role: str, content: str, metadata: Dict = None):
msg = Message(role=role, content=content, metadata=metadata or {})
self.messages.append(msg)
# ย่อความเมื่อจำนวนข้อความเกิน threshold
if len(self.messages) > self.summary_threshold:
self._create_summary()
def _create_summary(self):
if len(self.messages) < 10:
return
# ส่งข้อความเก่าไปย่อ แล้วลบทิ้ง
old_messages = self.messages[:len(self.messages)//2]
summary_text = f"[Summary of {len(old_messages)} messages from {old_messages[0].timestamp}]"
self.summaries.append(summary_text)
self.messages = self.messages[len(old_messages):]
def get_context(self) -> str:
context_parts = []
if self.summaries:
context_parts.append("## Conversation Summaries:\n" + "\n".join(self.summaries))
context_parts.append("## Recent Messages:\n" +
"\n".join([f"{m.role}: {m.content}" for m in self.messages[-10:]]))
return "\n\n".join(context_parts)
class EcommerceAgent:
def __init__(self, api_key: str, base_url: str):
self.client = OpenAI(api_key=api_key, base_url=base_url)
self.memory = ConversationMemory()
self.tools = self._register_tools()
def _register_tools(self) -> List[Dict]:
"""กำหนดเครื่องมือที่ Agent สามารถใช้งานได้"""
return [
{
"type": "function",
"function": {
"name": "search_product",
"description": "ค้นหาสินค้าในคลังสินค้า",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "คำค้นหา"},
"category": {"type": "string", "description": "หมวดหมู่สินค้า"},
"max_price": {"type": "number", "description": "ราคาสูงสุด"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "check_order_status",
"description": "ตรวจสอบสถานะออเดอร์",
"parameters": {
"type": "object",
"properties": {
"order_id": {"type": "string", "description": "หมายเลขออเดอร์"}
},
"required": ["order_id"]
}
}
},
{
"type": "function",
"function": {
"name": "update_crm",
"description": "อัพเดทข้อมูลลูกค้าในระบบ CRM",
"parameters": {
"type": "object",
"properties": {
"customer_id": {"type": "string"},
"action": {"type": "string", "enum": ["add_note", "update_preference", "log_interaction"]},
"data": {"type": "object"}
},
"required": ["customer_id", "action", "data"]
}
}
}
]
async def process_message(self, user_message: str) -> str:
self.memory.add_message("user", user_message)
# เตรียม prompt พร้อม context จาก memory
context = self.memory.get_context()
system_prompt = f"""คุณคือ AI Customer Service Agent สำหรับร้านค้าออนไลน์
มีความสามารถในการค้นหาสินค้า ตรวจสอบออเดอร์ และอัพเดทข้อมูลลูกค้า
ข้อมูลการสนทนาก่อนหน้า:
{context}
กฎการทำงาน:
1. หากต้องการข้อมูลสินค้า ให้ใช้ tool search_product
2. หากถามเรื่องออเดอร์ ให้ใช้ tool check_order_status
3. หากลูกค้าแจ้งข้อมูลใหม่ ให้ใช้ tool update_crm
4. ตอบเป็นภาษาไทยที่เป็นมิตรและเป็นธรรมชาติ"""
response = self.client.chat.completions.create(
model="claude-sonnet-4.5",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
],
tools=self.tools,
tool_choice="auto",
temperature=0.7
)
assistant_message = response.choices[0].message
# จัดการ Tool Calls
if assistant_message.tool_calls:
tool_results = self._execute_tools(assistant_message.tool_calls)
return self._format_tool_results(tool_results)
self.memory.add_message("assistant", assistant_message.content)
return assistant_message.content
def _execute_tools(self, tool_calls) -> List[Dict]:
results = []
for call in tool_calls:
func_name = call.function.name
args = json.loads(call.function.arguments)
# Mock implementation - ใน production ต้องเชื่อมต่อระบบจริง
if func_name == "search_product":
results.append({
"tool": func_name,
"result": f"พบ {args.get('query')} 3 รายการ: Nike Air Max, Adidas Ultraboost, Puma RS-X"
})
elif func_name == "check_order_status":
results.append({
"tool": func_name,
"result": f"ออเดอร์ {args.get('order_id')} สถานะ: กำลังจัดส่ง (คาดว่าถึง 2-3 วัน)"
})
elif func_name == "update_crm":
results.append({
"tool": func_name,
"result": f"อัพเดท CRM สำเร็จสำหรับลูกค้า {args.get('customer_id')}"
})
return results
การใช้งาน
agent = EcommerceAgent(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)ระบบ Planning สำหรับ Multi-Step Tasks
หลังจากตั้งค่าโครงสร้างพื้นฐานแล้ว สิ่งสำคัญถัดมาคือการสร้างระบบ Planning ที่ช่วยให้ Agent สามารถแตก Task ที่ซับซ้อนออกเป็นขั้นตอนย่อยๆ และดำเนินการทีละขั้นได้อย่างมีประสิทธิภาพ
from enum import Enum
from typing import Callable, Any
import asyncio
class TaskStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class Task:
id: str
description: str
status: TaskStatus = TaskStatus.PENDING
result: Any = None
dependencies: List[str] = field(default_factory=list)
class AgentPlanner:
"""ระบบวางแผนสำหรับ Multi-Step Agent Tasks"""
def __init__(self, agent: EcommerceAgent):
self.agent = agent
self.tasks: Dict[str, Task] = {}
self.task_queue: asyncio.Queue = asyncio.Queue()
async def plan_and_execute(self, goal: str) -> str:
"""วิเคราะห์เป้าหมายและสร้างแผนการดำเนินการ"""
# ขอให้ Model วางแผน
planning_prompt = f"""คุณคือ Task Planner วิเคราะห์เป้าหมายด้านล่างและแตกเป็นขั้นตอน
เป้าหมาย: {goal}
กำหนดผลลัพธ์เป็น JSON ดังนี้:
{{
"tasks": [
{{"id": "step_1", "description": "...", "dependencies": []}},
{{"id": "step_2", "description": "...", "dependencies": ["step_1"]}}
]
}}
กฎ:
- แต่ละขั้นตอนต้องทำได้โดยใช้เครื่องมือที่มีอยู่
- ระบุ dependencies ให้ถูกต้อง
- สร้างขั้นตอนให้ละเอียดพอที่จะดำเนินการได้"""
response = self.agent.client.chat.completions.create(
model="claude-sonnet-4.5",
messages=[{"role": "user", "content": planning_prompt}],
response_format={"type": "json_object"},
temperature=0.3
)
plan = json.loads(response.choices[0].message.content)
return await self._execute_plan(plan.get("tasks", []))
async def _execute_plan(self, tasks: List[Dict]) -> str:
"""ดำเนินการตามแผนที่วางไว้"""
results = []
for task_spec in tasks:
task = Task(
id=task_spec["id"],
description=task_spec["description"],
dependencies=task_spec.get("dependencies", [])
)
self.tasks[task.id] = task
# รอจนกว่า dependencies จะเสร็จ
for dep_id in task.dependencies:
while self.tasks[dep_id].status != TaskStatus.COMPLETED:
if self.tasks[dep_id].status == TaskStatus.FAILED:
return f"ไม่สามารถดำเนินการได้: {dep_id} ล้มเหลว"
await asyncio.sleep(0.1)
# ดำเนินการ task
task.status = TaskStatus.IN_PROGRESS
try:
result = await self.agent.process_message(task.description)
task.result = result
task.status = TaskStatus.COMPLETED
results.append(f"✅ {task.id}: {result}")
except Exception as e:
task.status = TaskStatus.FAILED
results.append(f"❌ {task.id}: {str(e)}")
return "\n".join(results)
ตัวอย่างการใช้งาน Planning System
async def main():
planner = AgentPlanner(agent)
# ตัวอย่าง: ลูกค้าถามเรื่องการสั่งซื้อและต้องการข้อมูลเพิ่มเติม
goal = "ลูกค้าถามว่าออเดอร์ #12345 สถานะอะไร และมีรองเท้าที่กำลังลดราคาแนะนำไหม"
result = await planner.plan_and_execute(goal)
print(result)
รัน asyncio
asyncio.run(main())การเชื่อมต่อระบบภายนอก (Production Integration)
ในการใช้งานจริง คุณต้องเชื่อมต่อกับระบบภายนอกเช่น ฐานข้อมูล ระบบ Inventory หรือ CRM จริง ด้านล่างนี้คือตัวอย่างการ Implement Function Calls ที่เชื่อมต่อกับระบบจริง
import sqlite3
from typing import Dict, Any
import httpx
class RealToolsIntegration:
"""ตัวอย่างการเชื่อมต่อระบบจริงสำหรับ E-commerce Agent"""
def __init__(self):
# สมมติว่ามีการเชื่อมต่อฐานข้อมูล
self.db_path = "ecommerce.db"
# สมมติว่ามี CRM API
self.crm_api_url = "https://crm.example.com/api/v1"
def search_product_real(self, query: str, category: str = None, max_price: float = None) -> Dict:
"""ค้นหาสินค้าจริงจากฐานข้อมูล"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
sql = "SELECT * FROM products WHERE name LIKE ?"
params = [f"%{query}%"]
if category:
sql += " AND category = ?"
params.append(category)
if max_price:
sql += " AND price <= ?"
params.append(max_price)
sql += " LIMIT 10"
cursor.execute(sql, params)
results = [dict(row) for row in cursor.fetchall()]
conn.close()
return {"products": results, "count": len(results)}
def check_order_status_real(self, order_id: str) -> Dict:
"""ตรวจสอบสถานะออเดอร์จริง"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("""
SELECT o.*, u.name as customer_name, u.email
FROM orders o
JOIN users u ON o.user_id = u.id
WHERE o.id = ?
""", (order_id,))
order = cursor.fetchone()
if not order:
conn.close()
return {"error": "ไม่พบออเดอร์นี้"}
# ดึงรายการสินค้าในออเดอร์
cursor.execute("""
SELECT p.name, p.price, oi.quantity
FROM order_items oi
JOIN products p ON oi.product_id = p.id
WHERE oi.order_id = ?
""", (order_id,))
items = [dict(row) for row in cursor.fetchall()]
conn.close()
return {
"order_id": order["id"],
"status": order["status"],
"customer": {"name": order["customer_name"], "email": order["email"]},
"items": items,
"total": order["total"],
"created_at": order["created_at"]
}
async def update_crm_real(self, customer_id: str, action: str, data: Dict) -> Dict:
"""อัพเดท CRM ผ่าน API"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.crm_api_url}/customers/{customer_id}/{action}",
json=data,
headers={"Authorization": "Bearer CRM_API_KEY"},
timeout=10.0
)
if response.status_code == 200:
return {"success": True, "data": response.json()}
else:
return {"success": False, "error": response.text}
การ Register Tools สำหรับ Production
production_tools = [
{
"type": "function",
"function": {
"name": "search_product",
"description": "ค้นหาสินค้าในคลังสินค้าจริง",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string"},
"category": {"type": "string"},
"max_price": {"type": "number"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "check_order_status",
"description": "ตรวจสอบสถานะออเดอร์",
"parameters": {
"type": "object",
"properties": {
"order_id": {"type": "string"}
},
"required": ["order_id"]
}
}
}
]
ใช้งานร่วมกับ Agent
tools = RealToolsIntegration()
ทดสอบการค้นหาสินค้า
result = tools.search_product_real("รองเท้า", max_price=2000)
print(f"พบสินค้า: {result['count']} รายการ")
for p in result['products'][:3]:
print(f" - {p['name']}: {p['price']} บาท")ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
จากประสบการณ์การ Implement Claude 4.6 Agent SDK หลายโปรเจกต์ พบว่ามีข้อผิดพลาดที่เกิดขึ้นซ้ำๆ ดังนี้
- ข้อผิดพลาดที่ 1: 401 Authentication Error
สาเหตุหลักคือ API Key ไม่ถูกต้องหรือหมดอายุ แก้ไขโดยตรวจสอบว่าใช้ API Key จาก HolySheep ที่ถูกต้อง และตั้งค่า base_url เป็น https://api.holysheep.ai/v1 อย่างแม่นยำ หากยังไม่ได้ ลอง Generate Key ใหม่จาก Dashboard
# ❌ วิธีที่ผิด - ใช้ base_url ผิด
client = OpenAI(api_key="YOUR_HOLYSHEEP_API_KEY", base_url="https://api.anthropic.com")
✅ วิธีที่ถูกต้อง
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1" # ต้องตรงเป๊ะๆ
)
ตรวจสอบว่า API ทำงานได้
try:
models = client.models.list()
print("✅ เชื่อมต่อสำเร็จ")
except Exception as e:
print(f"❌ Error: {e}")
# ลองตรวจสอบ API Key อีกครั้งบ่อยครั้งที่ปัญหาเกิดจากการตั้งค่า tool_choice เป็น "none" โดยไม่ตั้งใจ หรือ System Prompt ไม่ชัดเจนพอ วิธีแก้คือต้องกำหนด tool_choice เป็น "auto" และเพิ่มคำสั่งใน System Prompt ให้ชัดเจนว่าเมื่อไหร่ควรใช้เครื่องมือ
# ❌ วิธีที่ผิด - tool_choice ผิด
response = client.chat.completions.create(
model="claude-sonnet-4.5",
messages=messages,
tools=tools,
tool_choice="none" # ❌ ทำให้ไม่เรียก tool เลย
)
✅ วิธีที่ถูกต้อง
response = client.chat.completions.create(
model="claude-sonnet-4.5",
messages=messages,
tools=tools,
tool_choice="auto", # ✅ อนุญาตให้ model ตัดสินใจเรียก tool
temperature=0.7
)
และเพิ่มใน System Prompt
system_prompt = """
คุณสามารถใช้เครื่องมือต่อไปนี้เมื่อจำเป็น:
- search_product: เมื่อลูกค้าถามเรื่องสินค้า
- check_order_status: เมื่อลูกค้าถามเรื่องออเดอร์
**กฎสำคัญ**: หากคำถามต้องการข้อมูลที่ต้องค้นหา คุณต้องเรียกใช้เครื่องมือเสมอ
"""ปัญหานี้เกิดขึ้นเมื่อ Memory สะสมข้อความมากเกินไปจนเกิน Context Limit วิธีแก้คือต้อง Implement Summarization และจำกัดจำนวน Messages ที่ส่งไปให้ Model
# ❌ วิธีที่ผิด - ส่ง messages ทั้งหมดไปเลย
response = client.chat.completions.create(
model="claude-sonnet-4.5",
messages=all_messages # ❌ อาจมี thousands of tokens
)
✅ วิธีที่ถูกต้อง - ใช้ Memory Management
class SmartMemory:
MAX_TOKENS = 100000 # Claude 4.5 ใช้ได้ถึง 200K
def __init__(self):
self.messages = []
self.s