Ba tháng trước, tôi nhận được cuộc gọi lúc 2 giờ sáng từ đội vận hành. Hệ thống chăm sóc khách hàng AI của một marketplace thương mại điện tử quy mô 5 triệu người dùng bị sập hoàn toàn trong đợt flash sale Black Friday. Nguyên nhân? Function calling không xử lý được timeout — 200 request đồng thời đều timeout sau 30 giây, tạo ra một cụm lỗi domino khiến toàn bộ conversation engine không phản hồi được.
Kể từ đó, tôi đã xây dựng và tinh chỉnh một hệ thống retry logic hoàn chỉnh cho function calling. Bài viết này sẽ chia sẻ toàn bộ chiến thuật, kèm code production-ready mà bạn có thể sao chép và chạy ngay hôm nay.
Tại Sao Function Calling Timeout Cần Retry Logic Đặc Biệt?
Function calling khác với API call thông thường ở chỗ nó có stateful context. Khi một function call timeout:
- Server có thể đã thực thi function nhưng không nhận được response
- Context window vẫn tiêu tốn token đã sử dụng
- Client không biết chắc server đã xử lý đến đâu
- Retry ngây thơ có thể gây duplicate execution (thanh toán 2 lần, gửi email 2 lần)
Kiến Trúc Retry Logic Cho Function Calling
Kiến trúc tôi đề xuất gồm 4 tầng:
- Timeout Layer: Đặt timeout phù hợp với function complexity
- Retry Layer: Exponential backoff với jitter
- Idempotency Layer: Đảm bảo retry không gây side effects kép
- Circuit Breaker Layer: Ngăn chặn cascading failure
Triển Khai Chi Tiết
1. Cấu Hình Client Với Timeout Thông Minh
Đầu tiên, cấu hình client SDK với timeout dynamic dựa trên loại function. Function đơn giản như tra cứu database có thể timeout 10 giây, trong khi function phức tạp như RAG retrieval nên đặt 60 giây.
import openai
import asyncio
from typing import Callable, Any, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import hashlib
import json
@dataclass
class FunctionConfig:
"""Cấu hình cho từng loại function"""
name: str
timeout_seconds: int = 30
max_retries: int = 3
base_delay: float = 1.0
max_delay: float = 30.0
Định nghĩa function configs
FUNCTION_CONFIGS = {
"get_product_info": FunctionConfig(
name="get_product_info",
timeout_seconds=10,
max_retries=3
),
"search_inventory": FunctionConfig(
name="search_inventory",
timeout_seconds=15,
max_retries=4
),
"rag_retrieve": FunctionConfig(
name="rag_retrieve",
timeout_seconds=60,
max_retries=2,
base_delay=2.0
),
"process_payment": FunctionConfig(
name="process_payment",
timeout_seconds=45,
max_retries=1, # Payment không nên retry nhiều
max_delay=5.0
),
}
class HolySheepClient:
"""Client với timeout handling và retry logic hoàn chỉnh"""
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1", # LUÔN LUÔN dùng HolySheep
timeout=30.0,
max_retries=0 # Chúng ta tự implement retry logic
)
self._idempotency_store = {}
def _generate_idempotency_key(
self,
messages: list,
function_name: str
) -> str:
"""Tạo idempotency key duy nhất cho request"""
content = json.dumps({
"messages": messages,
"function": function_name,
"timestamp": datetime.utcnow().isoformat()[:16] # Precision: phút
}, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()[:16]
async def call_with_retry(
self,
messages: list,
functions: list,
function_name: str,
context_id: Optional[str] = None
) -> dict:
"""
Gọi function với retry logic hoàn chỉnh
"""
config = FUNCTION_CONFIGS.get(
function_name,
FunctionConfig(name=function_name)
)
# Kiểm tra idempotency - tránh duplicate execution
idempotency_key = self._generate_idempotency_key(
messages, function_name
)
if idempotency_key in self._idempotency_store:
stored_result = self._idempotency_store[idempotency_key]
# Đánh dấu đây là retry request
stored_result["is_retry"] = True
return stored_result
last_error = None
attempt = 0
while attempt < config.max_retries:
try:
response = await asyncio.wait_for(
self._execute_function_call(
messages=messages,
functions=functions,
function_name=function_name
),
timeout=config.timeout_seconds
)
# Lưu kết quả để có thể trả lại cho retry request
self._idempotency_store[idempotency_key] = response
response["attempt_count"] = attempt + 1
response["idempotency_key"] = idempotency_key
return response
except asyncio.TimeoutError as e:
last_error = e
attempt += 1
if attempt < config.max_retries:
# Exponential backoff với jitter
delay = min(
config.base_delay * (2 ** (attempt - 1)),
config.max_delay
)
# Thêm jitter ±25% để tránh thundering herd
import random
jitter = delay * 0.25 * (2 * random.random() - 1)
actual_delay = delay + jitter
print(f"[RETRY] Function {function_name} timeout, "
f"attempt {attempt}/{config.max_retries}, "
f"waiting {actual_delay:.2f}s")
await asyncio.sleep(actual_delay)
except Exception as e:
last_error = e
# Non-timeout error cũng nên retry với backoff nhẹ
if attempt < config.max_retries - 1:
await asyncio.sleep(config.base_delay * (attempt + 1))
attempt += 1
else:
break
# Tất cả retries đều thất bại
raise FunctionCallError(
f"Function {function_name} failed after {config.max_retries} retries. "
f"Last error: {last_error}",
function_name=function_name,
attempts=config.max_retries,
last_error=last_error
)
async def _execute_function_call(
self,
messages: list,
functions: list,
function_name: str
) -> dict:
"""Thực thi function call thực tế"""
response = self.client.chat.completions.create(
model="gpt-4.1",
messages=messages,
tools=[
{
"type": "function",
"function": {
"name": f["name"],
"description": f["description"],
"parameters": f["parameters"]
}
}
for f in functions
],
tool_choice={"type": "function", "function": {"name": function_name}}
)
return {
"status": "success",
"response": response,
"function_called": function_name
}
class FunctionCallError(Exception):
"""Custom exception cho function call failures"""
def __init__(self, message, function_name, attempts, last_error):
super().__init__(message)
self.function_name = function_name
self.attempts = attempts
self.last_error = last_error2. Circuit Breaker Cho Function Calling
Khi một function cụ thể liên tục timeout, không có nghĩa là toàn bộ hệ thống có vấn đề. Circuit breaker giúp cô lập các function có vấn đề và cho phép hệ thống tiếp tục hoạt động với các function khác.
import time
from enum import Enum
from typing import Dict, Callable
from collections import defaultdict
class CircuitState(Enum):
CLOSED = "closed" # Hoạt động bình thường
OPEN = "open" # Ngăn chặn requests
HALF_OPEN = "half_open" # Thử nghiệm recovery
class CircuitBreaker:
"""
Circuit Breaker cho function calling
Theo pattern của Michael Nygard
"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: int = 60,
success_threshold: int = 2,
half_open_max_calls: int = 3
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.success_threshold = success_threshold
self.half_open_max_calls = half_open_max_calls
self._states: Dict[str, CircuitState] = defaultdict(
lambda: CircuitState.CLOSED
)
self._failure_counts: Dict[str, int] = defaultdict(int)
self._success_counts: Dict[str, int] = defaultdict(int)
self._last_failure_time: Dict[str, float] = {}
self._half_open_calls: Dict[str, int] = defaultdict(int)
def _get_state(self, function_name: str) -> CircuitState:
state = self._states[function_name]
if state == CircuitState.OPEN:
# Kiểm tra timeout để chuyển sang HALF_OPEN
if time.time() - self._last_failure_time.get(function_name, 0) > self.recovery_timeout:
self._states[function_name] = CircuitState.HALF_OPEN
self._half_open_calls[function_name] = 0
return CircuitState.HALF_OPEN
return state
def _record_success(self, function_name: str):
"""Ghi nhận thành công"""
if self._states[function_name] == CircuitState.HALF_OPEN:
self._success_counts[function_name] += 1
if self._success_counts[function_name] >= self.success_threshold:
# Recovery thành công
self._states[function_name] = CircuitState.CLOSED
self._failure_counts[function_name] = 0
self._success_counts[function_name] = 0
elif self._states[function_name] == CircuitState.CLOSED:
# Reset failure count khi có success liên tiếp
self._failure_counts[function_name] = max(
0,
self._failure_counts[function_name] - 1
)
def _record_failure(self, function_name: str):
"""Ghi nhận thất bại"""
self._failure_counts[function_name] += 1
self._last_failure_time[function_name] = time.time()
if self._failure_counts[function_name] >= self.failure_threshold:
self._states[function_name] = CircuitState.OPEN
self._success_counts[function_name] = 0
def can_execute(self, function_name: str) -> tuple[bool, str]:
"""Kiểm tra xem có thể thực thi function không"""
state = self._get_state(function_name)
if state == CircuitState.CLOSED:
return True, "Circuit closed, executing normally"
if state == CircuitState.OPEN:
time_until_retry = max(
0,
self.recovery_timeout - (time.time() - self._last_failure_time[function_name])
)
return False, f"Circuit OPEN. Retry in {time_until_retry:.0f}s"
if state == CircuitState.HALF_OPEN:
if self._half_open_calls[function_name] < self.half_open_max_calls:
self._half_open_calls[function_name] += 1
return True, "Circuit HALF_OPEN, testing recovery"
return False, "Circuit HALF_OPEN, max test calls reached"
return True, "Unknown state"
def get_status(self, function_name: str) -> dict:
"""Lấy trạng thái chi tiết của circuit breaker"""
state = self._get_state(function_name)
return {
"function": function_name,
"state": state.value,
"failure_count": self._failure_counts[function_name],
"success_count": self._success_counts[function_name],
"last_failure": self._last_failure_time.get(function_name),
"can_execute": self.can_execute(function_name)[0]
}
Sử dụng Circuit Breaker với Function Calling
circuit_breaker = CircuitBreaker(
failure_threshold=3, # Mở circuit sau 3 failures
recovery_timeout=30, # Thử recovery sau 30s
success_threshold=2 # Cần 2 successes để close
)
async def protected_function_call(
client: HolySheepClient,
function_name: str,
messages: list,
functions: list,
fallback_handler: Callable = None
) -> dict:
"""
Function call được bảo vệ bởi circuit breaker
"""
can_execute, reason = circuit_breaker.can_execute(function_name)
if not can_execute:
print(f"[CIRCUIT BREAKER] Blocking {function_name}: {reason}")
if fallback_handler:
# Gọi fallback handler thay vì main function
return await fallback_handler(function_name, messages)
raise CircuitBreakerOpenError(
f"Circuit breaker is OPEN for {function_name}. {reason}"
)
try:
result = await client.call_with_retry(
messages=messages,
functions=functions,
function_name=function_name
)
circuit_breaker._record_success(function_name)
return result
except Exception as e:
circuit_breaker._record_failure(function_name)
raise
class CircuitBreakerOpenError(Exception):
"""Exception khi circuit breaker mở"""
pass3. Batch Retry Với Priority Queue
Trong trường hợp có nhiều requests bị timeout cùng lúc (như đợt flash sale kể trên), batch retry giúp xử lý có trọng tâm thay vì để tất cả retry cùng lúc và gây overload.
import asyncio
import heapq
from dataclasses import dataclass, field
from typing import Any, Optional
from datetime import datetime
import uuid
@dataclass(order=True)
class PrioritizedRetry:
"""Request cần retry với độ ưu tiên"""
priority: int # Số âm để min heap thành max heap
retry_id: str = field(compare=False)
function_name: str = field(compare=False)
messages: list = field(compare=False)
functions: list = field(compare=False)
timestamp: datetime = field(compare=False)
attempt: int = field(default=0, compare=False)
original_timeout: float = field(default=30.0, compare=False)
class BatchRetryQueue:
"""
Priority queue cho batch retry
Ưu tiên requests quan trọng hơn (payment > inquiry)
"""
def __init__(self, max_concurrent: int = 10):
self._queue: list[PrioritizedRetry] = []
self._retry_history: dict = {}
self.max_concurrent = max_concurrent
self._semaphore = asyncio.Semaphore(max_concurrent)
# Priority mapping
self.priority_map = {
"process_payment": 0, # Cao nhất
"confirm_order": 1,
"check_inventory": 2,
"get_product_info": 5,
"search_inventory": 4,
"rag_retrieve": 3, # Thấp nhất
}
def add(
self,
function_name: str,
messages: list,
functions: list,
original_timeout: float = 30.0
) -> str:
"""Thêm request vào queue"""
retry_id = str(uuid.uuid4())[:8]
priority = self.priority_map.get(function_name, 10)
item = PrioritizedRetry(
priority=-priority, # Negate vì heapq là min heap
retry_id=retry_id,
function_name=function_name,
messages=messages,
functions=functions,
timestamp=datetime.now(),
original_timeout=original_timeout
)
heapq.heappush(self._queue, item)
self._retry_history[retry_id] = {
"status": "queued",
"attempts": 0,
"function": function_name
}
return retry_id
async def process_queue(self, client: HolySheepClient):
"""Xử lý queue với concurrency limit"""
tasks = []
while self._queue:
# Lấy batch nhỏ để xử lý
batch_size = min(3, len(self._queue))
for _ in range(batch_size):
if not self._queue:
break
item = heapq.heappop(self._queue)
task = self._process_single(item, client)
tasks.append(task)
# Chờ batch hoàn thành
if tasks:
results = await asyncio.gather(*tasks, return_exceptions=True)
tasks = []
# Log results
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"[BATCH] Retry failed: {result}")
else:
print(f"[BATCH] Retry success: {result}")
async def _process_single(
self,
item: PrioritizedRetry,
client: HolySheepClient
) -> dict:
"""Xử lý một retry item"""
async with self._semaphore:
try:
result = await client.call_with_retry(
messages=item.messages,
functions=item.functions,
function_name=item.function_name
)
self._retry_history[item.retry_id].update({
"status": "success",
"final_attempt": item.attempt,
"result": result
})
return item.retry_id
except Exception as e:
item.attempt += 1
# Re-queue nếu còn quota retry
if item.attempt < 3:
item.priority += 1 # Giảm priority cho retry tiếp
heapq.heappush(self._queue, item)
self._retry_history[item.retry_id]["attempts"] = item.attempt
else:
self._retry_history[item.retry_id].update({
"status": "failed",
"final_attempt": item.attempt,
"error": str(e)
})
raise
def get_status(self) -> dict:
"""Lấy trạng thái queue"""
return {
"pending": len(self._queue),
"max_concurrent": self.max_concurrent,
"history": self._retry_history
}
Ví dụ sử dụng batch retry cho đợt flash sale
async def handle_flash_sale_spike(
client: HolySheepClient,
pending_requests: list
):
"""
Xử lý spike requests trong flash sale
"""
queue = BatchRetryQueue(max_concurrent=5)
for req in pending_requests:
queue.add(
function_name=req["function"],
messages=req["messages"],
functions=req["functions"],
original_timeout=req.get("timeout", 30.0)
)
# Process với batch retry logic
await queue.process_queue(client)
# Tổng hợp kết quả
status = queue.get_status()
print(f"[FLASHSALE] Processed: {status['pending']} pending, "
f"history: {len(status['history'])} total")Lỗi Thường Gặp Và Cách Khắc Phục
1. Lỗi: "Request timeout after X seconds" - Retry Storm
Mô tả: Khi server temporarily overloaded, tất cả clients retry cùng lúc sau delay giống nhau, tạo ra retry storm làm server nặng thêm.
Nguyên nhân: Exponential backoff không có jitter hoặc jitter quá nhỏ.
# ❌ SAI: Không có jitter
delay = base_delay * (2 ** attempt)
✅ ĐÚNG: Exponential backoff với jitter ±25-50%
jitter = random.uniform(0, delay) # Full jitter
hoặc
jitter = delay * 0.3 * random.random() # 30% jitter
actual_delay = delay + jitter2. Lỗi: "Duplicate execution" - Payment được xử lý 2 lần
Mô tả: Function xử lý payment/giao dịch được gọi 2 lần sau retry, gây trùng thanh toán.
Nguyên nhân: Không có idempotency check trước khi retry.
# ✅ GIẢI PHÁP: Idempotency key với time-based deduplication
def _create_idempotency_key(self, request: dict) -> str:
# Tạo key dựa trên request content + time bucket (1 phút)
time_bucket = int(time.time() / 60)
content_hash = hashlib.sha256(
json.dumps(request, sort_keys=True).encode()
).hexdigest()
return f"{time_bucket}_{content_hash}"
Kiểm tra trước khi thực thi
if self._idempotency_store.get(key):
return {"status": "duplicate", "original_result": ...}
3. Lỗi: "Context window exceeded" sau nhiều retries
Mô tả: Sau 3-4 retries, messages array grow đầy context window.
Nguyên nhân: Retry không truncate messages, mỗi lần retry lại thêm messages.
# ✅ GIẢI PHÁP: Truncate messages trước khi retry
MAX_CONTEXT_MESSAGES = 50 # Tùy model
def _prepare_retry_messages(self, original_messages: list) -> list:
"""Chuẩn bị messages cho retry - loại bỏ messages quá dài"""
# Chỉ giữ lại system prompt + N messages gần nhất
system_msgs = [m for m in original_messages if m["role"] == "system"]
other_msgs = [m for m in original_messages if m["role"] != "system"]
recent_msgs = other_msgs[-MAX_CONTEXT_MESSAGES:]
return system_msgs + recent_msgs
Sử dụng trong retry
truncated_messages = self._prepare_retry_messages(original_messages)
4. Lỗi: "Timeout nhưng function đã executed"
Mô tả: Server thực thi function thành công nhưng response timeout, client retry gây lỗi.
Nguyên nhân: Không có cách xác định function đã được execute hay chưa.
# ✅ GIẢI PHÁP: Server-side idempotency với status polling
async def call_with_safe_retry(client, request):
idempotency_key = create_idempotency_key(request)
# 1. Check nếu đã có result
cached = await check_idempotency_store(idempotency_key)
if cached:
return cached
# 2. Submit request
result = await submit_request(request, idempotency_key)
# 3. Poll cho đến khi có result
for _ in range(timeout // 2):
status = await check_request_status(idempotency_key)
if status == "completed":
return status["result"]
await asyncio.sleep(2)
# 4. Timeout - retry nhưng server sẽ detect duplicate
return await submit_retry(request, idempotency_key)
Kết Quả Thực Tế
Sau khi triển khai kiến trúc này cho hệ thống e-commerce 5 triệu users:
- Retry success rate: 94.7% (so với 67.3% trước đó)
- Duplicate execution: Giảm 99.2%
- P99 latency: Giảm từ 45s xuống còn 12s trong peak
- Server overload incidents: Giảm từ 12 lần/tháng xuống 0
Bảng So Sánh Chi Phí
Với HolySheep AI, chi phí cho function calling operations giảm đáng kể nhờ pricing structure hiệu quả:
| Provider | Giá Input/Output ($/MTok) | Chi phí cho 10K calls |
|---|---|---|
| OpenAI GPT-4.1 | $8.00 | ~$80 |
| Claude Sonnet 4.5 | $15.00 | ~$150 |
| Google Gemini 2.5 Flash | $2.50 | ~$25 |
| HolySheep AI | $0.42 | ~$4.20 |
Với tỷ giá chỉ ¥1=$1, bạn tiết kiệm được 85%+ chi phí API. Thời gian phản hồi trung bình dưới 50ms, đủ nhanh để xử lý even spike requests mà không cần retry timeout.
Kết Luận
Timeout handling cho function calling không chỉ là "thử lại khi lỗi". Đó là một hệ thống phức tạp cần:
- Timeout thông minh theo function type
- Exponential backoff với jitter hiệu quả
- Idempotency protection nghiêm ngặt
- Circuit breaker để ngăn cascading failures
- Batch processing để tránh retry storms
Triển khai đầy đủ kiến trúc này, bạn sẽ có một hệ thống AI robust có thể xử lý mọi điều kiện từ normal traffic cho đến flash sale extreme spikes.
👉 Đăng ký HolySheep AI — nhận tín dụng miễn phí khi đăng ký