Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến 3 năm xây dựng hệ thống Agent-Skills cho các dự án enterprise. Điều đầu tiên cần làm rõ: chi phí API AI đang là yếu tố quyết định sống còn. Hãy cùng xem dữ liệu giá 2026 đã được xác minh:
So sánh chi phí API AI 2026
| Model | Giá Output ($/MTok) | 10M token/tháng |
|---|---|---|
| GPT-4.1 | $8.00 | $80.00 |
| Claude Sonnet 4.5 | $15.00 | $150.00 |
| Gemini 2.5 Flash | $2.50 | $25.00 |
| DeepSeek V3.2 | $0.42 | $4.20 |
Bạn thấy không? DeepSeek V3.2 rẻ hơn Claude Sonnet 4.5 tới 97%. Với 10 triệu token mỗi tháng, chênh lệch là $145.80 - một con số có thể nuôi cả team dev thêm 1 tháng!
Tại sao cần thư viện Agent-Skills?
Khi tôi bắt đầu xây dựng hệ thống multi-agent đầu tiên vào năm 2023, mỗi agent đều được hard-code riêng lẻ. Kết quả? 30% code bị lặp lại, 50% bug xuất phát từ việc quản lý API key không đồng nhất, và mỗi lần đổi model là cả cơn ác mộng refactor.
Agent-Skills ra đời để giải quyết 3 vấn đề cốt lõi:
- Tái sử dụng: Viết một lần, gọi ở nhiều nơi
- Đồng nhất: Một interface cho mọi model
- Mở rộng: Thêm model mới không cần sửa agent
Kiến trúc Agent-Skills Core
1. Skill Base Class - Nền tảng mọi kỹ năng
# skill_base.py
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
import asyncio
@dataclass
class SkillConfig:
model: str
temperature: float = 0.7
max_tokens: int = 2048
timeout: int = 30
retry_count: int = 3
retry_delay: float = 1.0
@dataclass
class SkillResponse:
content: str
model: str
tokens_used: int
latency_ms: float
cost_usd: float
timestamp: datetime
metadata: Optional[Dict[str, Any]] = None
class BaseSkill(ABC):
"""Base class cho mọi Agent-Skill"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self._rate_limiter = RateLimiter()
@abstractmethod
async def execute(self, context: Dict[str, Any]) -> SkillResponse:
"""Execute skill với context cho trước"""
pass
@abstractmethod
def get_system_prompt(self) -> str:
"""Return system prompt của skill"""
pass
async def call_llm(self, messages: list, config: SkillConfig) -> Dict[str, Any]:
"""Unified LLM call với retry logic"""
await self._rate_limiter.acquire()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": config.model,
"messages": messages,
"temperature": config.temperature,
"max_tokens": config.max_tokens
}
for attempt in range(config.retry_count):
try:
async with asyncio.timeout(config.timeout):
start = asyncio.get_event_loop().time()
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers
) as resp:
latency = (asyncio.get_event_loop().time() - start) * 1000
if resp.status == 200:
data = await resp.json()
return self._parse_response(data, latency)
elif resp.status == 429:
await asyncio.sleep(config.retry_delay * (attempt + 1))
continue
else:
raise APIError(f"HTTP {resp.status}")
except asyncio.TimeoutError:
if attempt == config.retry_count - 1:
raise SkillTimeoutError(config.timeout)
raise MaxRetryExceededError(config.retry_count)
class RateLimiter:
"""Simple token bucket rate limiter"""
def __init__(self, rpm: int = 500):
self.rpm = rpm
self.tokens = rpm
self.last_update = asyncio.get_event_loop().time()
async def acquire(self):
while self.tokens < 1:
await asyncio.sleep(0.1)
self._refill()
self.tokens -= 1
def _refill(self):
now = asyncio.get_event_loop().time()
elapsed = now - self.last_update
self.tokens = min(self.rpm, self.tokens + elapsed * (self.rpm / 60))
self.last_update = now2. Pricing Utility - Tính chi phí thông minh
# pricing.py
from enum import Enum
from dataclasses import dataclass
from typing import Dict, Optional
class ModelType(Enum):
GPT_4_1 = "gpt-4.1"
CLAUDE_SONNET_4_5 = "claude-sonnet-4.5"
GEMINI_2_5_FLASH = "gemini-2.5-flash"
DEEPSEEK_V3_2 = "deepseek-v3.2"
@dataclass
class ModelPricing:
name: str
input_cost_per_mtok: float
output_cost_per_mtok: float
context_window: int
MODEL_PRICING: Dict[ModelType, ModelPricing] = {
ModelType.GPT_4_1: ModelPricing(
name="GPT-4.1",
input_cost_per_mtok=2.50,
output_cost_per_mtok=8.00,
context_window=128000
),
ModelType.CLAUDE_SONNET_4_5: ModelPricing(
name="Claude Sonnet 4.5",
input_cost_per_mtok=3.00,
output_cost_per_mtok=15.00,
context_window=200000
),
ModelType.GEMINI_2_5_FLASH: ModelPricing(
name="Gemini 2.5 Flash",
input_cost_per_mtok=0.30,
output_cost_per_mtok=2.50,
context_window=1000000
),
ModelType.DEEPSEEK_V3_2: ModelPricing(
name="DeepSeek V3.2",
input_cost_per_mtok=0.14,
output_cost_per_mtok=0.42,
context_window=64000
),
}
class CostCalculator:
"""Tính chi phí và đề xuất model tối ưu"""
@staticmethod
def calculate_cost(
model: ModelType,
input_tokens: int,
output_tokens: int
) -> float:
pricing = MODEL_PRICING[model]
input_cost = (input_tokens / 1_000_000) * pricing.input_cost_per_mtok
output_cost = (output_tokens / 1_000_000) * pricing.output_cost_per_mtok
return round(input_cost + output_cost, 6)
@staticmethod
def estimate_monthly_cost(
model: ModelType,
avg_input_per_call: int,
avg_output_per_call: int,
calls_per_month: int
) -> Dict[str, float]:
cost_per_call = CostCalculator.calculate_cost(
model, avg_input_per_call, avg_output_per_call
)
total = cost_per_call * calls_per_month
# So sánh với các model khác
comparison = {}
for m in ModelType:
if m != model:
other_cost = CostCalculator.calculate_cost(
m, avg_input_per_call, avg_output_per_call
) * calls_per_month
comparison[m.value] = {
"monthly_cost": round(other_cost, 2),
"savings": round(other_cost - total, 2),
"savings_percent": round((other_cost - total) / other_cost * 100, 1)
}
return {
"model": MODEL_PRICING[model].name,
"cost_per_call": round(cost_per_call, 6),
"monthly_cost": round(total, 2),
"comparison": comparison
}
@staticmethod
def suggest_model(task_type: str) -> ModelType:
"""Đề xuất model dựa trên loại task"""
suggestions = {
"fast_response": ModelType.DEEPSEEK_V3_2,
"high_quality": ModelType.CLAUDE_SONNET_4_5,
"balanced": ModelType.GEMINI_2_5_FLASH,
"coding": ModelType.GPT_4_1,
}
return suggestions.get(task_type, ModelType.DEEPSEEK_V3_2)
Ví dụ sử dụng
if __name__ == "__main__":
result = CostCalculator.estimate_monthly_cost(
model=ModelType.DEEPSEEK_V3_2,
avg_input_per_call=1000,
avg_output_per_call=500,
calls_per_month=100000
)
print(f"Chi phí DeepSeek V3.2/tháng: ${result['monthly_cost']}")
print(f"Tiết kiệm so với Claude Sonnet: ${result['comparison']['claude-sonnet-4.5']['savings']}")3. Ví dụ Skill hoàn chỉnh - TranslationSkill
# skills/translation_skill.py
from typing import Dict, Any, List
from skill_base import BaseSkill, SkillConfig, SkillResponse
import aiohttp
class TranslationSkill(BaseSkill):
"""Skill dịch thuật đa ngôn ngữ"""
SUPPORTED_PAIRS = [
("en", "vi"), ("vi", "en"), ("en", "zh"), ("zh", "en"),
("en", "ja"), ("ja", "en"), ("en", "ko"), ("ko", "en"),
]
def get_system_prompt(self) -> str:
return """Bạn là một chuyên gia dịch thuật.
Nhiệm vụ: Dịch text từ ngôn ngữ nguồn sang ngôn ngữ đích.
Yêu cầu:
1. Giữ nguyên ý nghĩa và sắc thái của text gốc
2. Sử dụng ngữ pháp tự nhiên của ngôn ngữ đích
3. Nếu text không thể dịch, trả về nguyên bản kèm [UNTRANSLATABLE]
4. Giữ nguyên định dạng markdown nếu có"""
async def execute(self, context: Dict[str, Any]) -> SkillResponse:
text = context.get("text", "")
source_lang = context.get("source_lang", "en")
target_lang = context.get("target_lang", "vi")
if not text:
return SkillResponse(
content="",
model="none",
tokens_used=0,
latency_ms=0,
cost_usd=0,
timestamp=datetime.now()
)
config = SkillConfig(
model="deepseek-v3.2", # Model tối ưu chi phí
temperature=0.3, # Độ ổn định cao cho dịch thuật
max_tokens=4096
)
messages = [
{"role": "system", "content": self.get_system_prompt()},
{"role": "user", "content": f"Dịch từ {source_lang} sang {target_lang}:\n\n{text}"}
]
response = await self.call_llm(messages, config)
return SkillResponse(
content=response["content"],
model=response["model"],
tokens_used=response["usage"]["total_tokens"],
latency_ms=response["latency_ms"],
cost_usd=response["cost_usd"],
timestamp=datetime.now(),
metadata={
"source_lang": source_lang,
"target_lang": target_lang,
"confidence": response.get("confidence", 1.0)
}
)
class BatchTranslationSkill(BaseSkill):
"""Skill dịch hàng loạt với batching tối ưu"""
def __init__(self, api_key: str):
super().__init__(api_key)
self.batch_size = 10
self.concurrency_limit = 5
async def execute(self, context: Dict[str, Any]) -> List[SkillResponse]:
items = context.get("items", [])
target_lang = context.get("target_lang", "vi")
semaphore = asyncio.Semaphore(self.concurrency_limit)
async def translate_item(item: Dict[str, Any]) -> SkillResponse:
async with semaphore:
skill = TranslationSkill(self.api_key)
return await skill.execute({
"text": item["text"],
"source_lang": item.get("source_lang", "en"),
"target_lang": target_lang
})
# Batch requests để tối ưu chi phí
results = []
for i in range(0, len(items), self.batch_size):
batch = items[i:i + self.batch_size]
batch_results = await asyncio.gather(
*[translate_item(item) for item in batch],
return_exceptions=True
)
results.extend(batch_results)
return results
Sử dụng với HolySheep API
async def main():
from dotenv import load_dotenv
load_dotenv()
api_key = "YOUR_HOLYSHEEP_API_KEY" # Thay bằng key thực tế
skill = TranslationSkill(api_key)
# Dịch đơn
result = await skill.execute({
"text": "The future of AI is here",
"source_lang": "en",
"target_lang": "vi"
})
print(f"Dịch: {result.content}")
print(f"Chi phí: ${result.cost_usd:.4f}")
print(f"Độ trễ: {result.latency_ms:.0f}ms")
# Dịch hàng loạt
batch_skill = BatchTranslationSkill(api_key)
batch_result = await batch_skill.execute({
"items": [
{"text": "Hello world", "source_lang": "en"},
{"text": "Good morning", "source_lang": "en"},
{"text": "Thank you", "source_lang": "en"},
],
"target_lang": "vi"
})
for i, res in enumerate(batch_result):
if isinstance(res, SkillResponse):
print(f"Item {i}: {res.content} - ${res.cost_usd:.4f}")
else:
print(f"Item {i}: Error - {res}")
if __name__ == "__main__":
asyncio.run(main())4. Skill Registry - Quản lý tập trung
# skill_registry.py
from typing import Dict, Type, List, Optional
from skill_base import BaseSkill
from functools import lru_cache
class SkillRegistry:
"""Singleton registry cho tất cả Agent-Skills"""
_instance = None
_skills: Dict[str, Type[BaseSkill]] = {}
_skill_instances: Dict[str, BaseSkill] = {}
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
@classmethod
def register(cls, name: str, skill_class: Type[BaseSkill]):
"""Register một skill mới"""
if not issubclass(skill_class, BaseSkill):
raise ValueError(f"{skill_class} phải kế thừa BaseSkill")
cls._skills[name] = skill_class
print(f"✅ Registered skill: {name}")
@classmethod
def get_skill(cls, name: str, api_key: str) -> BaseSkill:
"""Lấy instance của skill (singleton per name)"""
if name not in cls._skill_instances:
if name not in cls._skills:
available = list(cls._skills.keys())
raise ValueError(
f"Skill '{name}' không tồn tại. "
f"Available: {available}"
)
cls._skill_instances[name] = cls._skills[name](api_key)
return cls._skill_instances[name]
@classmethod
def list_skills(cls) -> List[str]:
"""Liệt kê tất cả skills đã đăng ký"""
return list(cls._skills.keys())
@classmethod
def get_skill_info(cls, name: str) -> Dict:
"""Lấy thông tin chi tiết của skill"""
if name not in cls._skills:
return {}
skill_class = cls._skills[name]
return {
"name": name,
"class": skill_class.__name__,
"doc": skill_class.__doc__,
"methods": [
m for m in dir(skill_class)
if not m.startswith('_') and callable(getattr(skill_class, m))
]
}
def skill(name: str):
"""Decorator để đăng ký skill"""
def decorator(cls: Type[BaseSkill]):
SkillRegistry.register(name, cls)
return cls
return decorator
Đăng ký skills
from skills.translation_skill import TranslationSkill, BatchTranslationSkill
SkillRegistry.register("translate", TranslationSkill)
SkillRegistry.register("translate_batch", BatchTranslationSkill)
Sử dụng trong Agent
class MultiAgent:
def __init__(self, api_key: str):
self.api_key = api_key
self.registry = SkillRegistry()
async def execute_task(self, task: Dict) -> Dict:
skill_name = task["skill"]
skill = self.registry.get_skill(skill_name, self.api_key)
result = await skill.execute(task["context"])
return {
"skill": skill_name,
"result": result.content,
"metadata": result.metadata,
"cost_usd": result.cost_usd
}
async def execute_workflow(self, tasks: List[Dict]) -> List[Dict]:
results = []
for task in tasks:
result = await self.execute_task(task)
results.append(result)
return resultsChiến lược tối ưu chi phí thực chiến
Qua 3 năm vận hành hệ thống Agent-Skills, tôi đã đúc kết 5 chiến lược tối ưu chi phí:
1. Smart Routing - Định tuyến thông minh
# smart_router.py
from typing import Callable, Awaitable, Any, Dict
from enum import Enum
from dataclasses import dataclass
class TaskPriority(Enum):
LOW = "low" # Dịch thuật, tóm tắt
MEDIUM = "medium" # Viết content, brainstorm
HIGH = "high" # Code generation, phân tích phức tạp
CRITICAL = "critical" # Quyết định kinh doanh
@dataclass
class RouteRule:
priority: TaskPriority
min_quality_score: float
max_cost_per_1k_tokens: float
allowed_models: list
fallback_model: str
class SmartRouter:
"""Định tuyến request tới model phù hợp nhất"""
DEFAULT_RULES = {
TaskPriority.LOW: RouteRule(
priority=TaskPriority.LOW,
min_quality_score=0.7,
max_cost_per_1k_tokens=0.001,
allowed_models=["deepseek-v3.2", "gemini-2.5-flash"],
fallback_model="deepseek-v3.2"
),
TaskPriority.MEDIUM: RouteRule(
priority=TaskPriority.MEDIUM,
min_quality_score=0.85,
max_cost_per_1k_tokens=0.005,
allowed_models=["gemini-2.5-flash", "gpt-4.1"],
fallback_model="gemini-2.5-flash"
),
TaskPriority.HIGH: RouteRule(
priority=TaskPriority.HIGH,
min_quality_score=0.95,
max_cost_per_1k_tokens=0.02,
allowed_models=["gpt-4.1", "claude-sonnet-4.5"],
fallback_model="gpt-4.1"
),
TaskPriority.CRITICAL: RouteRule(
priority=TaskPriority.CRITICAL,
min_quality_score=0.99,
max_cost_per_1k_tokens=0.05,
allowed_models=["claude-sonnet-4.5"],
fallback_model="claude-sonnet-4.5"
),
}
def __init__(self, rules: Dict[TaskPriority, RouteRule] = None):
self.rules = rules or self.DEFAULT_RULES
def route(self, task: Dict) -> str:
priority = task.get("priority", TaskPriority.MEDIUM)
rule = self.rules.get(priority)
if not rule:
rule = self.rules[TaskPriority.MEDIUM]
# Chọn model rẻ nhất trong allowed_models
return rule.allowed_models[0]
def estimate_cost(self, task: Dict, model: str) -> float:
"""Ước tính chi phí cho task với model cụ thể"""
input_tokens = task.get("estimated_input_tokens", 1000)
output_tokens = task.get("estimated_output_tokens", 500)
pricing = MODEL_PRICING.get(ModelType(model))
if not pricing:
return 0
cost = (input_tokens / 1_000_000) * pricing.input_cost_per_mtok
cost += (output_tokens / 1_000_000) * pricing.output_cost_per_mtok
return round(cost, 6)
Sử dụng
router = SmartRouter()
task = {
"type": "translation",
"priority": TaskPriority.LOW,
"estimated_input_tokens": 500,
"estimated_output_tokens": 600
}
selected_model = router.route(task)
estimated_cost = router.estimate_cost(task, selected_model)
print(f"Model: {selected_model}, Chi phí ước tính: ${estimated_cost:.4f}")2. Caching Layer - Giảm 70% chi phí
# skill_cache.py
import hashlib
import json
import redis
from typing import Optional, Any
from datetime import timedelta
class SkillCache:
"""Redis-based cache cho skill responses"""
def __init__(self, redis_url: str = "redis://localhost:6379/0"):
self.redis = redis.from_url(redis_url)
self.default_ttl = timedelta(hours=24)
def _generate_key(self, skill_name: str, context: dict) -> str:
"""Tạo cache key từ skill name và context"""
context_str = json.dumps(context, sort_keys=True)
hash_obj = hashlib.sha256(f"{skill_name}:{context_str}".encode())
return f"skill_cache:{skill_name}:{hash_obj.hexdigest()[:16]}"
async def get(self, skill_name: str, context: dict) -> Optional[dict]:
"""Lấy response từ cache"""
key = self._generate_key(skill_name, context)
cached = self.redis.get(key)
if cached:
return json.loads(cached)
return None
async def set(
self,
skill_name: str,
context: dict,
response: dict,
ttl: timedelta = None
):
"""Lưu response vào cache"""
key = self._generate_key(skill_name, context)
ttl = ttl or self.default_ttl
self.redis.setex(
key,
int(ttl.total_seconds()),
json.dumps(response)
)
async def invalidate(self, skill_name: str, pattern: str = "*"):
"""Xóa cache theo pattern"""
keys = self.redis.keys(f"skill_cache:{skill_name}:{pattern}")
if keys:
self.redis.delete(*keys)
class CachedSkillDecorator:
"""Decorator để thêm cache vào bất kỳ skill nào"""
def __init__(self, cache: SkillCache):
self.cache = cache
def __call__(self, skill: BaseSkill):
original_execute = skill.execute
async def cached_execute(context: Dict[str, Any]) -> SkillResponse:
# Thử lấy từ cache
cached = await self.cache.get(skill.__class__.__name__, context)
if cached:
return SkillResponse(
content=cached["content"],
model=cached["model"],
tokens_used=0, # Không tính tokens từ cache
latency_ms=0,
cost_usd=0,
timestamp=datetime.fromisoformat(cached["timestamp"]),
metadata={"cached": True}
)
# Execute thực tế
result = await original_execute(context)
# Lưu vào cache
await self.cache.set(
skill.__class__.__name__,
context,
{
"content": result.content,
"model": result.model,
"timestamp": result.timestamp.isoformat()
}
)
return result
skill.execute = cached_execute
return skill
Áp dụng cache cho tất cả skills
cache = SkillCache()
cached_decorator = CachedSkillDecorator(cache)
for skill_name in SkillRegistry.list_skills():
skill_instance = SkillRegistry.get_skill(skill_name, "YOUR_KEY")
cached_decorator(skill_instance)Lỗi thường gặp và cách khắc phục
1. Lỗi 401 Unauthorized - API Key không hợp lệ
# ❌ SAI - Hard-code API key trực tiếp
api_key = "sk-abc123..."
✅ ĐÚNG - Sử dụng environment variable
import os
from dotenv import load_dotenv
load_dotenv()
class HolySheepClient:
def __init__(self):
self.api_key = os.getenv("HOLYSHEEP_API_KEY")
if not self.api_key:
raise ValueError(
"HOLYSHEEP_API_KEY not found. "
"Đăng ký tại: https://www.holysheep.ai/register"
)
if not self.api_key.startswith("hs_"):
raise ValueError("Invalid API key format. Key phải bắt đầu bằng 'hs_'")
@property
def base_url(self):
return "https://api.holysheep.ai/v1"
Verify key trước khi sử dụng
async def verify_api_key(api_key: str) -> bool:
"""Verify API key bằng cách gọi endpoint /models"""
async with aiohttp.ClientSession() as session:
try:
async with session.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {api_key}"}
) as resp:
return resp.status == 200
except Exception:
return False2. Lỗi Rate Limit 429 - Vượt quá giới hạn request
# ❌ SAI - Gọi API liên tục không kiểm soát
async def process_items(items):
results = []
for item in items:
result = await call_api(item) # Có thể trigger 429
results.append(result)
return results
✅ ĐÚNG - Sử dụng rate limiter + exponential backoff
import asyncio
from aiohttp import ClientResponseError
class RobustAPIClient:
def __init__(self, api_key: str, rpm: int = 300):
self.api_key = api_key
self.rpm = rpm
self.request_times = []
async def call_with_retry(
self,
payload: dict,
max_retries: int = 5
) -> dict:
for attempt in range(max_retries):
try:
await self._wait_for_rate_limit()
return await self._make_request(payload)
except ClientResponseError as e:
if e.status == 429:
# Rate limit - exponential backoff
wait_time = (2 ** attempt) * 1.0 # 1s, 2s, 4s, 8s, 16s
print(f"Rate limited. Waiting {wait_time}s...")
await asyncio.sleep(wait_time)
else:
raise
except asyncio.TimeoutError:
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt)
raise MaxRetryExceededError(max_retries)
async def _wait_for_rate_limit(self):
"""Đảm bảo không vượt quá RPM"""
now = asyncio.get_event_loop().time()
self.request_times = [
t for t in self.request_times
if now - t < 60
]
if len(self.request_times) >= self.rpm:
oldest = self.request_times[0]
wait_time = 60 - (now - oldest) + 0.1
await asyncio.sleep(wait_time)
self.request_times.append(now)
async def batch_process(self, items: list) -> list:
"""Process với concurrency control"""
semaphore = asyncio.Semaphore(10) # Tối đa 10 concurrent
async def process_one(item):
async with semaphore:
return await self.call_with_retry(item)
return await asyncio.gather(
*[process_one(item) for item in items],
return_exceptions=True
)3. Lỗi Timeout - Request treo không có response
# ❌ SAI - Không có timeout hoặc timeout quá lâu
async def call_api(payload):
async with session.post(url, json=payload) as resp: # Vô hạn
return await resp.json()
✅ ĐÚNG - Timeout hợp lý + graceful handling
class TimeoutConfig:
FAST = 5.0 # 5s - cho simple queries
NORMAL = 30.0 # 30s - cho standard tasks
LONG = 120.0 # 120s - cho complex analysis
async def call_with_timeout(
client: aiohttp.ClientSession,
url: str,
payload: dict,
headers: dict,
timeout: float = TimeoutConfig.NORMAL
) -> dict:
timeout_obj = aiohttp.ClientTimeout(total=timeout)
try:
async with client.post(
url,
json=payload,
headers=headers,
timeout=timeout_obj
) as resp:
return await resp.json()
except asyncio.TimeoutError:
# Fallback: Trả về cached response hoặc queue lại
return {
"error": "timeout",
"message": f"Request timeout sau {timeout}s",
"fallback_action": "queued",
"estimated_retry": 30