เมื่อคืนนี้เวลาตี 2 ระบบ Production ของผมล่ม ทีม DevOps ต้องตื่นมาแก้ปัญหาเป็นชั่วโมง สาเหตุ? ConnectionError: timeout exceeded 30.001s จาก custom tool ที่เชื่อมต่อ API ภายนอก พร้อมกับ 401 Unauthorized อีก 3 ตัว ทั้งๆ ที่เพิ่ง deploy ไปเมื่อวาน
บทความนี้คือทุกอย่างที่ผมเรียนรู้จากการ burn 3 production incidents ใน 2 สัปดาห์ พร้อม pattern ที่ใช้งานได้จริงในการสร้าง robust enterprise API integration กับ CrewAI
ทำไม Tool Calling ของ CrewAI ถึงล่มใน Production
ปัญหาหลักของ CrewAI tool ที่เชื่อมกับ enterprise API มี 3 ชั้น:
- Retry ไม่มี — ถ้า API timeout ครั้งเดียว crew ก็ hang เลย
- Error handling ห่วย — ไม่แยกประเภท error แยกวิธีแก้
- Context overflow — ข้อมูลที่ return กลับมาใหญ่เกินไปทำให้ token เกิน limit
# ❌ โค้ดเดิมที่ทำให้ระบบล่ม
from crewai import Agent, Task, Crew
from langchain.agents import tool
@tool
def get_enterprise_data(query: str) -> str:
"""ดึงข้อมูลจาก enterprise API"""
response = requests.get(
f"https://api.enterprise.com/v2/data?q={query}",
headers={"Authorization": f"Bearer {API_KEY}"},
timeout=5 # สั้นเกินไป!
)
return response.json() # ไม่มี error handling!
โค้ดนี้ดูเหมือนธรรมดา แต่มันคือ bomb ที่รอระเบิด มาดูวิธีแก้กัน
โครงสร้าง Custom Tool ที่ Production-Ready
สิ่งที่ผมใช้มา 6 เดือนคือ layered approach ที่แยก concerns ชัดเจน
from crewai import Tool
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
import asyncio
import aiohttp
import time
from functools import wraps
=== Layer 1: Error Classification ===
class APIError(Enum):
TIMEOUT = "timeout"
AUTH_FAILED = "auth_failed"
RATE_LIMIT = "rate_limit"
SERVER_ERROR = "server_error"
NETWORK_ERROR = "network_error"
VALIDATION_ERROR = "validation_error"
@dataclass
class APIResponse:
success: bool
data: Optional[Any] = None
error: Optional[APIError] = None
error_message: Optional[str] = None
retry_count: int = 0
latency_ms: float = 0.0
=== Layer 2: Retry Strategy ===
@dataclass
class RetryConfig:
max_retries: int = 3
base_delay: float = 1.0
max_delay: float = 30.0
exponential_base: float = 2.0
retryable_errors: List[APIError] = field(
default_factory=lambda: [
APIError.TIMEOUT,
APIError.RATE_LIMIT,
APIError.SERVER_ERROR,
APIError.NETWORK_ERROR
]
)
def with_retry(config: RetryConfig = None):
if config is None:
config = RetryConfig()
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
last_error = None
for attempt in range(config.max_retries + 1):
try:
start = time.time()
result = await func(*args, **kwargs)
result.latency_ms = (time.time() - start) * 1000
return result
except asyncio.TimeoutError:
last_error = APIError.TIMEOUT
error_msg = f"Timeout after {kwargs.get('timeout', 'unknown')}s"
except aiohttp.ClientResponseError as e:
if e.status == 401:
return APIResponse(
success=False,
error=APIError.AUTH_FAILED,
error_message="Invalid or expired API key"
)
elif e.status == 429:
last_error = APIError.RATE_LIMIT
error_msg = f"Rate limited (retry_after: {e.headers.get('Retry-After')})"
elif 500 <= e.status < 600:
last_error = APIError.SERVER_ERROR
error_msg = f"Server error: {e.status}"
else:
return APIResponse(
success=False,
error=APIError.VALIDATION_ERROR,
error_message=f"HTTP {e.status}: {e.message}"
)
except aiohttp.ClientConnectorError:
last_error = APIError.NETWORK_ERROR
error_msg = "Connection failed - check network/firewall"
# Calculate delay with exponential backoff + jitter
if last_error in config.retryable_errors and attempt < config.max_retries:
delay = min(
config.base_delay * (config.exponential_base ** attempt),
config.max_delay
)
# Add jitter (±25%)
import random
delay *= (0.75 + random.random() * 0.5)
await asyncio.sleep(delay)
return APIResponse(
success=False,
error=last_error,
error_message=error_msg,
retry_count=attempt
)
return wrapper
return decorator
=== Layer 3: Enterprise API Client ===
class EnterpriseAPIClient:
def __init__(
self,
base_url: str,
api_key: str,
retry_config: RetryConfig = None
):
self.base_url = base_url.rstrip('/')
self.api_key = api_key
self.retry_config = retry_config or RetryConfig()
self._session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(
total=60,
connect=10,
sock_read=30
)
self._session = aiohttp.ClientSession(
timeout=timeout,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": str(int(time.time() * 1000))
}
)
return self
async def __aexit__(self, *args):
if self._session:
await self._session.close()
@with_retry()
async def get(
self,
endpoint: str,
params: Dict = None,
timeout: int = 30
) -> APIResponse:
url = f"{self.base_url}/{endpoint.lstrip('/')}"
async with self._session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
# Truncate if too large for LLM context
return APIResponse(
success=True,
data=self._truncate_for_context(data)
)
else:
raise aiohttp.ClientResponseError(
request_info=response.request_info,
history=response.history,
status=response.status,
message=await response.text()
)
def _truncate_for_context(self, data: Any, max_tokens: int = 4000) -> Any:
"""Prevent context overflow"""
import json
text = json.dumps(data)
# Rough estimate: 1 token ≈ 4 chars
if len(text) > max_tokens * 4:
if isinstance(data, list):
return {"items": data[:100], "truncated": True, "total": len(data)}
elif isinstance(data, dict):
return {"data": data, "truncated": True}
return data
การสร้าง Tool สำหรับ HolySheep AI API
สำหรับ AI API integration ผมใช้ HolySheep AI เพราะ latency เฉลี่ย <50ms กับราคาที่ถูกกว่า 85%+ เมื่อเทียบกับ OpenAI โดย API รองรับทั้ง GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash และ DeepSeek V3.2
from crewai import Tool
from langchain.tools import StructuredTool
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Literal
import aiohttp
import json
=== HolySheep AI API Integration ===
class HolySheepAIClient:
"""Production-ready client สำหรับ HolySheep AI API"""
BASE_URL = "https://api.holysheep.ai/v1" # ห้ามเปลี่ยน!
def __init__(self, api_key: str):
self.api_key = api_key
async def chat_completion(
self,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict:
"""
ส่ง request ไปยัง HolySheep AI
Supported models:
- gpt-4.1 ($8/MTok)
- claude-sonnet-4.5 ($15/MTok)
- gemini-2.5-flash ($2.50/MTok)
- deepseek-v3.2 ($0.42/MTok)
"""
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
) as response:
if response.status == 200:
return await response.json()
else:
error_text = await response.text()
raise Exception(f"API Error {response.status}: {error_text}")
=== Define Tool Schemas ===
class AnalyzeDocumentInput(BaseModel):
document_id: str = Field(description="ID ของ document ที่ต้องการวิเคราะห์")
analysis_type: Literal["summary", "sentiment", "entities", "full"] = Field(
description="ประเภทของการวิเคราะห์"
)
class GenerateReportInput(BaseModel):
topic: str = Field(description="หัวข้อรายงาน")
sections: List[str] = Field(description="หัวข้อย่อยที่ต้องการ")
format: Literal["markdown", "html", "json"] = Field(default="markdown")
=== Create CrewAI Tools ===
def create_holysheep_tools(api_key: str) -> List[Tool]:
"""Factory function สำหรับสร้าง tools ที่เชื่อมกับ HolySheep AI"""
client = HolySheepAIClient(api_key)
def analyze_document_func(document_id: str, analysis_type: str) -> str:
"""วิเคราะห์เอกสารด้วย AI"""
import asyncio
messages = [
{
"role": "system",
"content": """คุณคือผู้เชี่ยวชาญการวิเคราะห์เอกสาร
วิเคราะห์เอกสารตามประเภทที่ระบุและ return เฉพาะส่วนสำคัญ"""
},
{
"role": "user",
"content": f"วิเคราะห์ document ID: {document_id}\nประเภท: {analysis_type}"
}
]
# ใช้ DeepSeek V3.2 สำหรับงาน summary (ราคาถูกที่สุด)
model = "deepseek-v3.2" if analysis_type == "summary" else "gpt-4.1"
try:
result = asyncio.run(client.chat_completion(
model=model,
messages=messages,
temperature=0.3,
max_tokens=1024
))
return result['choices'][0]['message']['content']
except Exception as e:
return f"Error analyzing document: {str(e)}"
def generate_report_func(topic: str, sections: str, format: str) -> str:
"""สร้างรายงานอย่างมืออาชีพ"""
import asyncio
import json
section_list = json.loads(sections) if isinstance(sections, str) else sections
messages = [
{
"role": "system",
"content": """คุณคือผู้เชี่ยวชาญการเขียนรายงาน
สร้างรายงานที่ครอบคลุม มีโครงสร้างชัดเจน"""
},
{
"role": "user",
"content": f"สร้างรายงานเรื่อง: {topic}\nหัวข้อ: {', '.join(section_list)}\nFormat: {format}"
}
]
# ใช้ Gemini Flash สำหรับงานทั่วไป (คุ้มค่า)
result = asyncio.run(client.chat_completion(
model="gemini-2.5-flash",
messages=messages,
temperature=0.5,
max_tokens=2048
))
return result['choices'][0]['message']['content']
# Create Tool objects
analyze_tool = Tool.from_function(
name="analyze_document",
func=analyze_document_func,
description="""ใช้สำหรับวิเคราะห์เอกสาร
รับ document_id และประเภทการวิเคราะห์ (summary/sentiment/entities/full)
เหมาะสำหรับงาน NLP และ document processing""",
args_schema=AnalyzeDocumentInput
)
generate_tool = Tool.from_function(
name="generate_report",
func=generate_report_func,
description="""ใช้สำหรับสร้างรายงานอย่างมืออาชีพ
รับหัวข้อ หัวข้อย่อย และ format ที่ต้องการ
เหมาะสำหรับงาน content generation และ reporting""",
args_schema=GenerateReportInput
)
return [analyze_tool, generate_tool]
=== Usage Example ===
tools = create_holysheep_tools("YOUR_HOLYSHEEP_API_KEY")
agent = Agent(
role="Document Analyst",
goal="วิเคราะห์เอกสารและสร้างรายงานอย่างมืออาชีพ",
backstory="ผู้เชี่ยวชาญด้านการวิเคราะห์ข้อมูล",
tools=tools
)
Advanced Pattern: Streaming + Parallel Tool Execution
สำหรับงานที่ต้องรันหลาย tools พร้อมกัน ผมใช้ asyncio.gather กับ semaphore เพื่อควบคุม concurrency
import asyncio
from typing import List, Dict, Any, Callable
from dataclasses import dataclass
@dataclass
class ToolExecution:
tool_name: str
args: Dict[str, Any]
result: Any = None
error: str = None
latency_ms: float = 0.0
class ParallelToolExecutor:
"""Execute multiple tools in parallel with rate limiting"""
def __init__(self, max_concurrent: int = 5):
self.semaphore = asyncio.Semaphore(max_concurrent)
self.results: List[ToolExecution] = []
async def execute_tool(
self,
tool_func: Callable,
tool_name: str,
**kwargs
) -> ToolExecution:
async with self.semaphore: # Limit concurrent executions
start = time.time()
try:
if asyncio.iscoroutinefunction(tool_func):
result = await tool_func(**kwargs)
else:
result = tool_func(**kwargs)
return ToolExecution(
tool_name=tool_name,
args=kwargs,
result=result,
latency_ms=(time.time() - start) * 1000
)
except Exception as e:
return ToolExecution(
tool_name=tool_name,
args=kwargs,
error=str(e),
latency_ms=(time.time() - start) * 1000
)
async def execute_all(
self,
tools: List[tuple[Callable, str, Dict]]
) -> List[ToolExecution]:
"""
Execute list of (function, name, kwargs) tuples in parallel
Example:
tools = [
(analyze_tool_func, "doc_analyzer", {"doc_id": "123", "type": "full"}),
(search_tool_func, "web_search", {"query": "AI trends 2024"}),
(report_tool_func, "report_gen", {"topic": "market analysis"})
]
results = await executor.execute_all(tools)
"""
tasks = [
self.execute_tool(func, name, **kwargs)
for func, name, kwargs in tools
]
return await asyncio.gather(*tasks)
=== Integration with CrewAI ===
class RobustCrewPipeline:
"""Crew pipeline ที่รองรับ parallel execution และ graceful error handling"""
def __init__(self, tools: List[Tool], api_key: str):
self.tools = tools
self.executor = ParallelToolExecutor(max_concurrent=3)
self.api_client = HolySheepAIClient(api_key)
async def run_parallel_analysis(
self,
documents: List[str],
analysis_type: str = "full"
) -> Dict[str, Any]:
"""วิเคราะห์เอกสารหลายตัวพร้อมกัน"""
# Build tool execution list
tool_tasks = [
(
self._get_tool_func("analyze_document"),
f"analyze_{doc_id}",
{"document_id": doc_id, "analysis_type": analysis_type}
)
for doc_id in documents
]
# Execute with rate limiting
results = await self.executor.execute_all(tool_tasks)
# Aggregate results
successful = [r for r in results if r.error is None]
failed = [r for r in results if r.error is not None]
return {
"total": len(documents),
"successful": len(successful),
"failed": len(failed),
"results": {r.tool_name: r.result for r in successful},
"errors": {r.tool_name: r.error for r in failed},
"avg_latency_ms": sum(r.latency_ms for r in results) / len(results) if results else 0
}
def _get_tool_func(self, tool_name: str) -> Callable:
for tool in self.tools:
if tool.name == tool_name:
return tool.func
raise ValueError(f"Tool {tool_name} not found")
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. 401 Unauthorized — API Key หมดอายุหรือไม่ถูกต้อง
# ❌ สาเหตุ: Hardcode API key ในโค้ด
API_KEY = "sk-xxx-xxx" # ไม่ดี!
✅ วิธีแก้: ใช้ Environment Variables
import os
from dotenv import load_dotenv
load_dotenv() # โหลดจาก .env file
class SecureAPIClient:
def __init__(self):
self.api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not self.api_key:
raise ValueError(
"HOLYSHEEP_API_KEY not set. "
"Please set it in your environment or .env file"
)
# Validate key format
if not self.api_key.startswith(("sk-", "hs-")):
raise ValueError("Invalid API key format")
async def verify_connection(self) -> bool:
"""ตรวจสอบว่า API key ใช้งานได้"""
try:
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.BASE_URL}/models",
headers={"Authorization": f"Bearer {self.api_key}"}
) as response:
return response.status == 200
except Exception:
return False
2. ConnectionError: timeout — เกินเวลารอ Response
# ❌ สาเหตุ: Timeout สั้นเกินไป หรือ ไม่มี retry
requests.get(url, timeout=1) # 1 วินาที = น้อยเกินไป!
✅ วิธีแก้: Adaptive Timeout + Retry with Backoff
import backoff
class AdaptiveTimeoutClient:
def __init__(self):
self.default_timeout = aiohttp.ClientTimeout(
total=120, # Total timeout 2 นาที
connect=10, # Connection timeout 10 วินาที
sock_read=30 # Read timeout 30 วินาที
)
@backoff.on_exception(
backoff.expo,
(aiohttp.ClientError, asyncio.TimeoutError),
max_time=60,
max_tries=4,
jitter=backoff.full_jitter
)
async def robust_get(self, url: str, params: dict) -> dict:
async with aiohttp.ClientSession(
timeout=self.default_timeout
) as session:
async with session.get(url, params=params) as response:
# Handle streaming or large responses
if response.content_type == 'text/event-stream':
return await self._handle_stream(response)
return await response.json()
async def _handle_stream(self, response) -> dict:
"""Handle Server-Sent Events"""
chunks = []
async for line in response.content:
if line:
chunks.append(line.decode())
# Timeout protection for streaming
if len(chunks) > 10000:
break
return {"stream_data": "".join(chunks)}
3. RateLimitExceeded — เกินจำนวน request ต่อนาที
# ❌ สาเหตุ: ไม่รู้จัก rate limit ของ API
for doc in documents:
await api.call(doc) # เรียกพร้อมกันทั้งหมด!
✅ วิธีแก้: Token Bucket Algorithm
import asyncio
import time
from dataclasses import dataclass, field
@dataclass
class TokenBucket:
"""Token bucket สำหรับ rate limiting"""
capacity: int = 60 # Max tokens
refill_rate: float = 10.0 # Tokens per second
tokens: float = field(init=False)
last_refill: float = field(init=False)
def __post_init__(self):
self.tokens = float(self.capacity)
self.last_refill = time.time()
async def acquire(self, tokens: int = 1):
"""รอจนกว่าจะมี token ว่าง"""
while True:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return
# รอจน token ถูก refill
wait_time = (tokens - self.tokens) / self.refill_rate
await asyncio.sleep(wait_time)
def _refill(self):
now = time.time()
elapsed = now - self.last_refill
self.tokens = min(
self.capacity,
self.tokens + elapsed * self.refill_rate
)
self.last_refill = now
class RateLimitedAPIClient:
"""API Client พร้อม rate limiting"""
def __init__(self, requests_per_minute: int = 60):
# HolySheep AI มี rate limit 60 req/min สำหรับ tier มาตรฐาน
self.bucket = TokenBucket(
capacity=requests_per_minute,
refill_rate=requests_per_minute / 60.0
)
async def call(self, endpoint: str, **kwargs):
await self.bucket.acquire() # รอ token ถ้าจำเป็น
return await self._make_request(endpoint, **kwargs)
async def batch_call(self, endpoints: List[tuple]):
"""เรียกหลาย endpoints พร้อมกันแบบมี rate limit"""
tasks = [self.call(ep, **kw) for ep, kw in endpoints]
return await asyncio.gather(*tasks, return_exceptions=True)
4. Context Overflow — Response ใหญ่เกินไปสำหรับ LLM
# ❌ สาเหตุ: Return ข้อมูลทั้งหมดโดยไม่ truncate
def get_all_data():
return database.query("SELECT * FROM huge_table") # ล้าน rows!
✅ วิธีแก้: Smart Truncation + Pagination
from typing import Generator
class SmartDataFetcher:
"""ดึงข้อมูลแบบ streaming พร้อม context-aware truncation"""
MAX_CONTEXT_TOKENS = 6000 # เผื่อ buffer สำหรับ system prompt
def __init__(self, llm_client):
self.llm = llm_client
def estimate_tokens(self, text: str) -> int:
# Rough estimate: ~4 chars per token for Thai/English mixed
return len(text) // 4
async def fetch_with_context_aware_truncation(
self,
query: str,
max_items: int = 100,
summarize_if_large: bool = True
) -> str:
# ดึงข้อมูลแบบ paginate
raw_data = await self._paginated_fetch(query, limit=max_items)
estimated = self.estimate_tokens(str(raw_data))
if estimated <= self.MAX_CONTEXT_TOKENS:
return str(raw_data)
if summarize_if_large:
# ขอ LLM สรุปข้อมูลแทนการ truncate หน้าดิบ
summary_prompt = f"""สรุปข้อมูลต่อไปนี้ให้กระชับ
โดยเก็บข้อมูลสำคัญและสถิติหลัก:
{raw_data}
สรุปให้ไม่เกิน 1000 tokens"""
response = await self.llm.chat_completion(
model="deepseek-v3.2", # ใช้ model ราคาถูกสำหรับ summarize
messages=[{"role": "user", "content": summary_prompt}],
max_tokens=1024
)
return response['choices'][0]['message']['content']
# Fallback: Truncate อย่างฉลาด
return self._smart_truncate(raw_data)
def _smart_truncate(self, data,