เมื่อพัฒนาแอปพลิเคชันที่ใช้ LLM API การจัดการข้อผิดพลาด invalid parameters จาก Function Calling ถือเป็นสิ่งสำคัญอันดับต้นๆ ที่ทีมต้องรับมือ ในบทความนี้ผมจะอธิบายวิธีการตรวจจับ วิเคราะห์สาเหตุ และสร้างระบบ Retry-Downgrade ที่ทำงานได้อย่างเสถียร พร้อมแนะนำการย้ายระบบจาก API เดิมไปยัง HolySheep AI ที่ประหยัดกว่า 85%
ทำไม Function Calling ถึง Error บ่อย?
จากประสบการณ์ตรงในการดูแลระบบ Production ของหลายโปรเจกต์ สาเหตุหลักที่ทำให้เกิด invalid_parameters Error มีดังนี้:
- Schema Mismatch: Function schema ที่ส่งไปไม่ตรงกับรูปแบบที่ API คาดหวัง
- Token Overflow: request มีขนาดใหญ่เกิน limit ทำให้ parser พัง
- Rate Limit: ถูก throttle จน response กลับมาไม่สมบูรณ์
- Model-specific Syntax: แต่ละ model ตีความ JSON schema ต่างกัน
โครงสร้าง Error Response ที่ต้องจัดการ
เมื่อ Function Calling ล้มเหลว API จะตอบกลับมาพร้อม structure ที่คุณต้อง parse ให้เป็น:
{
"error": {
"code": "invalid_parameters",
"message": "Failed to parse function call arguments: unexpected token at position 42",
"param": null,
"type": "invalid_request_error"
},
"id": "func_call_abc123xyz",
"model": "gpt-4-turbo"
}โค้ด Python: ระบบ Retry-Downgrade แบบ Production-Ready
ด้านล่างคือโค้ดที่ใช้งานจริงใน Production ของผม ซึ่งจัดการ error อย่างครบวงจร:
import time
import json
import logging
from typing import Optional, Dict, Any, List
from enum import Enum
class APIModel(Enum):
PRIMARY = "gpt-4-turbo" # เริ่มจาก model แพงที่สุด
FALLBACK_1 = "gpt-3.5-turbo" # ถ้า fail ลด tier ลง
FALLBACK_2 = "deepseek-v3" # ถ้ายัง fail ลดอีก
class FunctionCallingError(Exception):
def __init__(self, code: str, message: str, retryable: bool = True):
self.code = code
self.message = message
self.retryable = retryable
super().__init__(message)
class HolySheepAPIClient:
"""
Production-ready client สำหรับ HolySheep AI
รองรับ: Retry logic, Exponential backoff, Model fallback, Circuit breaker
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.max_retries = 3
self.timeout = 30
self.fallback_chain = [
"gpt-4.1", # $8/MTok - เริ่มจากตัวแพงสุด
"claude-sonnet-4.5", # $15/MTok - แต่holy sheep ถูกกว่า
"gemini-2.5-flash", # $2.50/MTok - budget option
"deepseek-v3.2" # $0.42/MTok - ถูกที่สุด
]
self.current_model_index = 0
self.logger = logging.getLogger(__name__)
def _make_request(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""ทำ HTTP request ไปยัง HolySheep API"""
import requests
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=self.timeout
)
if response.status_code == 200:
return response.json()
# Parse error response
error_data = response.json() if response.text else {}
error_code = error_data.get("error", {}).get("code", "unknown")
error_message = error_data.get("error", {}).get("message", "Unknown error")
# ตรวจสอบว่า error นี้ retryable หรือไม่
retryable_codes = ["rate_limit_exceeded", "server_error", "timeout"]
is_retryable = error_code in retryable_codes or response.status_code >= 500
raise FunctionCallingError(error_code, error_message, is_retryable)
def _exponential_backoff(self, attempt: int) -> float:
"""คำนวณเวลาหน่วงแบบ Exponential: 1s, 2s, 4s..."""
return min(2 ** attempt + (time.time() % 1), 30)
def call_with_functions(
self,
messages: List[Dict],
functions: List[Dict],
temperature: float = 0.7,
**kwargs
) -> Dict[str, Any]:
"""
Main method: เรียก Function Calling พร้อม retry-fallback logic
"""
payload = {
"model": self.fallback_chain[self.current_model_index],
"messages": messages,
"tools": [{"type": "function", "function": f} for f in functions],
"temperature": temperature,
**kwargs
}
last_error = None
for attempt in range(self.max_retries):
try:
response = self._make_request(payload)
# ถ้า response มี tool_calls แสดงว่าสำเร็จ
if response.get("choices", [{}])[0].get("message", {}).get("tool_calls"):
self.logger.info(f"✓ Function call สำเร็จ (attempt {attempt + 1})")
return response
# ถ้าไม่มี tool_calls แสดงว่า model ไม่เข้าใจ instruction
# ลอง fallback ไป model ถัดไปทันที
self.logger.warning("Model ไม่สร้าง tool_calls → ลอง model ถัดไป")
self._fallback_to_next_model()
except FunctionCallingError as e:
last_error = e
self.logger.error(f"Attempt {attempt + 1} เกิด Error: {e.code} - {e.message}")
if not e.retryable:
# Non-retryable error เช่น invalid API key
self.logger.error("Non-retryable error → หยุดทันที")
raise
# รอ exponential backoff แล้วลองใหม่
if attempt < self.max_retries - 1:
wait_time = self._exponential_backoff(attempt)
self.logger.info(f"รอ {wait_time:.2f}s แล้วลองใหม่...")
time.sleep(wait_time)
# ถ้าลองครบทุก attempt แล้วยัง fail
# ลองทุก model ใน chain
return self._try_all_models(messages, functions, temperature, **kwargs)
def _fallback_to_next_model(self):
"""Downgrade model ไป tier ถัดไป"""
if self.current_model_index < len(self.fallback_chain) - 1:
self.current_model_index += 1
model = self.fallback_chain[self.current_model_index]
self.logger.info(f"↓ Fallback ไป model: {model}")
def _try_all_models(
self,
messages: List[Dict],
functions: List[Dict],
temperature: float,
**kwargs
) -> Dict[str, Any]:
"""ถ้า primary model fail → ลองทุก model ใน chain"""
for i, model in enumerate(self.fallback_chain):
self.current_model_index = i
self.logger.info(f"ลอง model ที่ {i + 1}/{len(self.fallback_chain)}: {model}")
try:
payload = {
"model": model,
"messages": messages,
"tools": [{"type": "function", "function": f} for f in functions],
"temperature": temperature,
**kwargs
}
response = self._make_request(payload)
if response.get("choices", [{}])[0].get("message", {}).get("tool_calls"):
self.logger.info(f"✓ {model} ทำงานสำเร็จ!")
return response
except FunctionCallingError as e:
self.logger.error(f"✗ {model} fail: {e.message}")
continue
# ถ้าทุก model ล้มเหลว → return ผลลัพธ์ว่างพร้อม error info
return {
"error": True,
"message": "ทุก model ใน chain ล้มเหลว",
"last_error": last_error.message if last_error else "Unknown"
}
วิธีใช้งาน
client = HolySheepAPIClient("YOUR_HOLYSHEEP_API_KEY")
functions = [
{
"name": "get_weather",
"description": "ดึงข้อมูลอากาศ",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "ชื่อเมือง"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
},
"required": ["location"]
}
}
]
messages = [{"role": "user", "content": "อากาศที่กรุงเทพเป็นยังไง?"}]
result = client.call_with_functions(messages, functions)
print(result)โค้ด Node.js: ระบบ Error Handler แบบ Async/Await
สำหรับทีมที่ใช้ JavaScript/TypeScript นี่คือ implementation ที่ใช้ async-await pattern:
/**
* HolySheep AI - Function Calling Error Handler
* Production-ready implementation with Retry + Fallback logic
*/
const https = require('https');
const { URL } = require('url');
// Error types
class APIError extends Error {
constructor(code, message, retryable = true) {
super(message);
this.code = code;
this.retryable = retryable;
this.name = 'APIError';
}
}
class FunctionCallingError extends APIError {
constructor(functionName, message, rawError) {
super('function_call_failed', message, true);
this.functionName = functionName;
this.rawError = rawError;
this.name = 'FunctionCallingError';
}
}
// Model chain สำหรับ fallback
const MODEL_CHAIN = [
{ name: 'gpt-4.1', pricePerMTok: 8 },
{ name: 'claude-sonnet-4.5', pricePerMTok: 15 },
{ name: 'gemini-2.5-flash', pricePerMTok: 2.5 },
{ name: 'deepseek-v3.2', pricePerMTok: 0.42 } // ถูกที่สุด
];
class HolySheepClient {
constructor(apiKey) {
this.apiKey = apiKey;
this.baseURL = 'https://api.holysheep.ai/v1';
this.currentModelIndex = 0;
this.maxRetries = 3;
this.requestTimeout = 30000;
}
async _httpRequest(payload) {
return new Promise((resolve, reject) => {
const url = new URL(${this.baseURL}/chat/completions);
const options = {
hostname: url.hostname,
port: 443,
path: url.pathname,
method: 'POST',
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json'
},
timeout: this.requestTimeout
};
const req = https.request(options, (res) => {
let data = '';
res.on('data', (chunk) => data += chunk);
res.on('end', () => {
if (res.statusCode === 200) {
resolve(JSON.parse(data));
} else {
const errorBody = JSON.parse(data || '{}');
const errorInfo = errorBody.error || {};
reject(new APIError(
errorInfo.code || 'http_error',
errorInfo.message || HTTP ${res.statusCode},
res.statusCode >= 500 || errorInfo.code === 'rate_limit_exceeded'
));
}
});
});
req.on('error', (e) => reject(new APIError('network_error', e.message, true)));
req.on('timeout', () => reject(new APIError('timeout', 'Request timeout', true)));
req.write(JSON.stringify(payload));
req.end();
});
}
calculateBackoff(attempt) {
// Exponential backoff: 1s, 2s, 4s, 8s... (max 30s)
const delay = Math.min(Math.pow(2, attempt) * 1000 + Math.random() * 1000, 30000);
return delay;
}
async _executeWithRetry(payload, functions) {
let lastError = null;
for (let attempt = 0; attempt < this.maxRetries; attempt++) {
try {
const response = await this._httpRequest(payload);
const message = response.choices?.[0]?.message;
if (message?.tool_calls) {
console.log(✓ Success on attempt ${attempt + 1});
return response;
}
// ถ้าไม่มี tool_calls → model ไม่เข้าใจ schema
console.log(⚠ No tool_calls → falling back to next model);
throw new APIError('invalid_schema', 'Model did not generate tool_calls', true);
} catch (error) {
lastError = error;
console.error(Attempt ${attempt + 1} failed: ${error.code} - ${error.message});
if (!error.retryable) {
throw error;
}
if (attempt < this.maxRetries - 1) {
const waitTime = this.calculateBackoff(attempt);
console.log(Waiting ${(waitTime/1000).toFixed(2)}s before retry...);
await new Promise(resolve => setTimeout(resolve, waitTime));
}
}
}
throw lastError;
}
async callFunction(messages, functions, options = {}) {
const startModelIndex = this.currentModelIndex;
for (let i = startModelIndex; i < MODEL_CHAIN.length; i++) {
this.currentModelIndex = i;
const model = MODEL_CHAIN[i];
console.log(\n🔄 Trying model: ${model.name} ($${model.pricePerMTok}/MTok));
const payload = {
model: model.name,
messages,
tools: functions.map(fn => ({ type: 'function', function: fn })),
temperature: options.temperature ?? 0.7,
...options
};
try {
const result = await this._executeWithRetry(payload, functions);
// แปลง response เป็น format ที่ใช้งานง่าย
return this._normalizeResponse(result);
} catch (error) {
console.error(✗ Model ${model.name} failed completely);
if (i < MODEL_CHAIN.length - 1) {
console.log('→ Falling back to next model...\n');
continue;
}
// ทุก model ล้มเหลว
return {
success: false,
error: error.code,
message: error.message,
attemptedModels: MODEL_CHAIN.slice(startModelIndex).map(m => m.name)
};
}
}
}
_normalizeResponse(response) {
const message = response.choices?.[0]?.message;
return {
success: true,
model: response.model,
toolCalls: message?.tool_calls?.map(call => ({
id: call.id,
function: call.function.name,
arguments: JSON.parse(call.function.arguments)
})),
usage: response.usage
};
}
}
// ==================== วิธีใช้งานจริง ====================
const client = new HolySheepClient('YOUR_HOLYSHEEP_API_KEY');
const functions = [
{
name: 'search_products',
description: 'ค้นหาสินค้าในระบบ',
parameters: {
type: 'object',
properties: {
query: { type: 'string', description: 'คำค้นหา' },
category: {
type: 'string',
enum: ['electronics', 'clothing', 'food', 'home']
},
maxPrice: { type: 'number', description: 'ราคาสูงสุด (บาท)' }
},
required: ['query']
}
}
];
const messages = [
{ role: 'system', content: 'คุณคือผู้ช่วยแนะนำสินค้า' },
{ role: 'user', content: 'หา notebook ราคาไม่เกิน 15000 บาท' }
];
async function main() {
try {
const result = await client.callFunction(messages, functions);
if (result.success) {
console.log('\n📋 ผลลัพธ์:', JSON.stringify(result.toolCalls, null, 2));
console.log(💰 ใช้ model: ${result.model});
console.log(📊 Token usage:, result.usage);
} else {
console.error('\n❌ เกิดข้อผิดพลาด:', result.message);
console.log(🔧 ลอง model ที่ fail:, result.attemptedModels.join(', '));
}
} catch (error) {
console.error('Fatal error:', error);
}
}
main();ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. Error: "Failed to parse function call arguments: unexpected token"
สาเหตุ: JSON schema ที่ส่งไปมี syntax ผิดพลาด เช่น trailing comma หรือ property ที่ไม่ถูกต้อง
// ❌ ผิด - มี trailing comma
{
"name": "get_weather",
"parameters": {
"type": "object",
"properties": {
"location": { "type": "string" },
}, // <-- trailing comma
}
}
// ✅ ถูกต้อง
{
"name": "get_weather",
"parameters": {
"type": "object",
"properties": {
"location": { "type": "string" }
}
}
}วิธีแก้ไข: ใช้ JSON Schema validator ก่อนส่ง request:
import jsonschema
def validate_function_schema(function_def: dict) -> bool:
"""ตรวจสอบ schema ก่อนส่งไป API"""
try:
jsonschema.validate(
instance=function_def,
schema={
"type": "object",
"required": ["name", "parameters"],
"properties": {
"name": {"type": "string"},
"description": {"type": "string"},
"parameters": {
"type": "object",
"required": ["type", "properties"],
"properties": {
"type": {"enum": ["object"]},
"properties": {"type": "object"},
"required": {"type": "array", "items": {"type": "string"}}
}
}
}
}
)
return True
except jsonschema.ValidationError as e:
print(f"Schema validation failed: {e.message}")
return False2. Error: "Invalid parameter: temperature must be between 0 and 2"
สาเหตุ: Parameter value เกิน limit ที่กำหนด หรือ type ไม่ตรง
วิธีแก้ไข: สร้าง validation wrapper:
def sanitize_parameters(params: dict, model: str) -> dict:
"""Sanitize parameters ให้อยู่ใน valid range"""
sanitized = params.copy()
# Temperature validation
if "temperature" in sanitized:
temp = float(sanitized["temperature"])
sanitized["temperature"] = max(0.0, min(2.0, temp))
# Max tokens validation
if "max_tokens" in sanitized:
max_tok = int(sanitized["max_tokens"])
# ขีดจำกัดต่างกันตาม model
limits = {
"gpt-4.1": 128000,
"claude-sonnet-4.5": 200000,
"gemini-2.5-flash": 100000,
"deepseek-v3.2": 64000
}
sanitized["max_tokens"] = min(max_tok, limits.get(model, 4096))
# String parameters - ตัด whitespace ข้างนอก
for key, value in sanitized.items():
if isinstance(value, str):
sanitized[key] = value.strip()
return sanitized3. Error: "Rate limit exceeded for model gpt-4-turbo"
สาเหตุ: เรียกใช้ API บ่อยเกินไปในเวลาสั้น
วิธีแก้ไข: ใช้ token bucket algorithm:
import time
import threading
from collections import deque
class RateLimiter:
"""
Token Bucket rate limiter
HolySheep: ไม่มี strict rate limit แต่แนะนำให้มี throttling
"""
def __init__(self, requests_per_minute: int = 60):
self.rpm = requests_per_minute
self.tokens = self.rpm
self.last_update = time.time()
self.lock = threading.Lock()
self.request_times = deque(maxlen=100) # เก็บ timestamp ของ request ล่าสุด
def acquire(self) -> bool:
"""รอจนกว่าจะมี quota ว่าง"""
with self.lock:
now = time.time()
# Clear request ที่เก่ากว่า 1 นาที
while self.request_times and self.request_times[0] < now - 60:
self.request_times.popleft()
# ถ้ายังมี quota
if len(self.request_times) < self.rpm:
self.request_times.append(now)
return True
# รอจน request เก่าสุดหมดอายุ
wait_time = 60 - (now - self.request_times[0])
if wait_time > 0:
time.sleep(wait_time)
self.request_times.popleft()
self.request_times.append(time.time())
return True
def wait_and_execute(self, func, *args, **kwargs):
"""Wrapper ที่รอ quota ก่อน execute"""
self.acquire()
return func(*args, **kwargs)
วิธีใช้
limiter = RateLimiter(requests_per_minute=500) # HolySheep แนะนำไม่เกิน 500 rpm
result = limiter.wait_and_execute(
client.call_with_functions,
messages,
functions
)4. Error: "Context length exceeded"
สาเหตุ: messages รวมกันแล้วยาวเกิน model context window
วิธีแก้ไข: สร้าง smart truncation:
def smart_truncate_messages(messages: list, model: str, max_ratio: float = 0.8) -> list:
"""
ตัด messages อย่างฉลาดโดยเก็บ system prompt ไว้
context_windows = {
"gpt-4.1": 128000,
"claude-sonnet-4.5": 200000,
"deepseek-v3.2": 64000
}
"""
# คำนวณ context limit
limit = context_windows.get(model, 32000) * max_ratio
# แยก system และ conversation
system_msg = None
conversation = []
for msg in messages:
if msg["role"] == "system":
system_msg = msg
else:
conversation.append(msg)
# ถ้ายังไม่เกิน limit ไม่ต้อง truncate
total_tokens = estimate_tokens(messages)
if total_tokens <= limit:
return messages
# Truncate conversation จากด้านบน (เก่าสุด)
result = []
current_tokens = 0
# ใส่ system ก่อนเสมอ
if system_msg:
result.append(system_msg)
current_tokens += estimate_single_tokens(system_msg["content"])
# ใส่ conversation จากล่างขึ้นบน
for msg in reversed(conversation):
msg_tokens = estimate_single_tokens(msg["content"])
if current_tokens + msg_tokens <= limit:
result.insert(1 if system_msg else 0, msg)
current_tokens += msg_tokens
else:
break
print(f"⚠️ Truncated {len(messages) - len(result)} messages (model: {model})")
return resultเหมาะกับใคร / ไม่เหมาะกับใคร
| เหมาะกับ | ไม่เหมาะกับ |
|---|---|
| • ทีมพัฒนาที่ใช้ Function Calling เป็น core feature | • โปรเจกต์ที่ต้องการ 100% uptime SLA เข้มงวด |
| • Startup ที่ต้องการลดต้นทุน API ลง 85%+ | • ระบบที่ใช้ Claude Computer Use (ต้องใช้ API ตรง) |
| • ทีมที่มี traffic สูง (100K+ requests/วัน) | • งานวิจัยที่ต้องการผลลัพธ์จาก model เฉพาะเจาะจง |
| • Developer ที่ต้องการ latency ต่ำ (<50ms) | • ผู้ที่ไม่คุ้นเคยกับ retry-fallback logic |
| • ธุรกิจในจีนที่ต้องการจ่ายผ่าน WeChat/Alipay | • แอปที่มี legal/compliance requirement เฉพาะ |