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 AI Agent chăm sóc khách hàng của một sàn thương mại điện tử quy mô 50 triệu người dùng bị treo hoàn toàn. Nguyên nhân? Một query vector database trả về kết quả bất thường, khiến agent rơi vào vòng lặp vô hạn và tiêu tốn hết quota API trong 45 phút.
Kể từ đó, tôi đã thiết kế lại hoàn toàn cơ chế xử lý lỗi cho các hệ thống AI Agent. Bài viết này chia sẻ kiến trúc, code thực tế và bài học xương máu từ production.
Tại Sao AI Agent Cần Error Handling Đặc Biệt?
Khác với API thông thường trả về HTTP status code rõ ràng, AI Agent hoạt động với:
- Output không xác định trước: Model có thể trả về JSON malformed, text thay vì structured data, hoặc response vượt context window
- Multi-step workflow: Lỗi ở bước 3 có thể phá hủy state từ bước 1-2
- External dependencies không đáng tin: Vector DB timeout, third-party API thay đổi schema, network intermittent
- Cost explosion: Retry loop không kiểm soát có thể tạo bills $10,000+ trong một đêm
Kiến Trúc Error Handling 4 Tầng
Tầng 1: Input Validation & Guardrails
class AgentInputGuardrails:
"""Tầng bảo vệ đầu vào - ngăn chặn lỗi từ gốc"""
MAX_INPUT_LENGTH = 100_000 # tokens
MAX_CONVERSATION_TURNS = 50
BLOCKED_PATTERNS = [
"system", "sudo", "rm -rf", "drop table",
"ignore previous instructions", "you are now"
]
def __init__(self, api_key: str):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1" # HolySheep AI - 85%+ tiết kiệm
)
async def validate_input(self, user_message: str, history: list) -> ValidationResult:
"""Kiểm tra toàn diện đầu vào trước khi xử lý"""
# 1. Kiểm tra độ dài
if len(user_message) > self.MAX_INPUT_LENGTH:
raise InputValidationError(
f"Input quá dài: {len(user_message)} chars > {self.MAX_INPUT_LENGTH}"
)
# 2. Kiểm tra injection patterns
lower_msg = user_message.lower()
for pattern in self.BLOCKED_PATTERNS:
if pattern.lower() in lower_msg:
raise SecurityError(f"Phát hiện pattern nguy hiểm: {pattern}")
# 3. Kiểm tra conversation history
if len(history) > self.MAX_CONVERSATION_TURNS:
raise ContextOverflowError(
f"Cuộc hội thoại quá dài: {len(history)} turns"
)
# 4. Validate structured data nếu có
if user_message.strip().startswith("{"):
try:
json.loads(user_message)
except json.JSONDecodeError as e:
raise InputValidationError(f"JSON không hợp lệ: {e}")
return ValidationResult(valid=True, sanitized=user_message.strip())Tầng 2: Execution Wrapper với Circuit Breaker
import asyncio
from enum import Enum
from dataclasses import dataclass
from datetime import datetime, timedelta
class AgentState(Enum):
IDLE = "idle"
RUNNING = "running"
WAITING_RETRY = "waiting_retry"
FAILED = "failed"
RECOVERED = "recovered"
@dataclass
class ExecutionContext:
"""Context cho mỗi execution attempt"""
attempt: int
max_attempts: int
state: AgentState
last_error: Optional[str] = None
retry_after: Optional[datetime] = None
backoff_ms: int = 1000
class ResilientAgentExecutor:
"""Executor với retry thông minh và circuit breaker"""
def __init__(self, api_key: str):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.state = AgentState.IDLE
self.failure_count = 0
self.circuit_open = False
self.circuit_open_since = None
# Cấu hình circuit breaker
self.FAILURE_THRESHOLD = 5
self.CIRCUIT_RESET_TIMEOUT = timedelta(minutes=5)
self.MAX_BACKOFF_MS = 30_000
# Cấu hình retry với exponential backoff
self.RETRY_CONFIG = {
"rate_limit": {"max_attempts": 3, "backoff_base": 2},
"timeout": {"max_attempts": 3, "backoff_base": 1.5},
"server_error": {"max_attempts": 5, "backoff_base": 2},
"validation": {"max_attempts": 1, "backoff_base": 1}, # Không retry validation error
}
async def execute_with_retry(
self,
agent_fn: callable,
context: ExecutionContext,
error_category: str = "server_error"
) -> ExecutionResult:
"""Execute với retry logic phân loại lỗi"""
config = self.RETRY_CONFIG.get(error_category, self.RETRY_CONFIG["server_error"])
while context.attempt < config["max_attempts"]:
try:
# Kiểm tra circuit breaker
if self.circuit_open:
if self._should_try_reset():
self._reset_circuit()
else:
raise CircuitBreakerOpenError(
f"Circuit breaker open. Thử lại sau {self.circuit_open_since + self.CIRCUIT_RESET_TIMEOUT}"
)
self.state = AgentState.RUNNING
context.state = AgentState.RUNNING
result = await agent_fn()
# Thành công - reset counters
self._on_success()
context.state = AgentState.IDLE
return ExecutionResult(success=True, data=result)
except RateLimitError as e:
self.failure_count += 1
wait_time = self._calculate_backoff(
config["backoff_base"],
context.attempt,
is_rate_limit=True
)
await self._handle_rate_limit(e, wait_time, context)
except TimeoutError as e:
self.failure_count += 1
wait_time = self._calculate_backoff(
config["backoff_base"],
context.attempt
)
await self._handle_timeout(e, wait_time, context)
except ServerError as e:
self.failure_count += 1
wait_time = self._calculate_backoff(
config["backoff_base"],
context.attempt
)
await self._handle_server_error(e, wait_time, context)
except ValidationError as e:
# Validation error - không retry
self.state = AgentState.FAILED
context.state = AgentState.FAILED
context.last_error = str(e)
return ExecutionResult(success=False, error=e, retryable=False)
except Exception as e:
self.failure_count += 1
self._update_circuit_state()
raise # Re-raise unknown errors
# Đã hết attempts
self.state = AgentState.FAILED
context.state = AgentState.FAILED
return ExecutionResult(
success=False,
error=f"Hết attempts sau {config['max_attempts']} lần thử",
retryable=False
)
def _calculate_backoff(self, base: float, attempt: int, is_rate_limit: bool = False) -> int:
"""Tính toán backoff delay với jitter"""
if is_rate_limit:
# Rate limit: sử dụng Retry-After header nếu có
base_delay = self._parse_retry_after()
else:
base_delay = 1000 # 1 giây base
delay = int(base_delay * (base ** attempt))
delay = min(delay, self.MAX_BACKOFF_MS)
# Thêm jitter 20% để tránh thundering herd
jitter = int(delay * 0.2 * (hash(attempt) % 100) / 100)
return delay + jitter
def _update_circuit_state(self):
"""Cập nhật trạng thái circuit breaker"""
if self.failure_count >= self.FAILURE_THRESHOLD:
self.circuit_open = True
self.circuit_open_since = datetime.now()
print(f"⚠️ Circuit breaker OPEN - Đã {self.failure_count} lỗi liên tiếp")
def _should_try_reset(self) -> bool:
"""Kiểm tra nếu nên thử reset circuit breaker"""
if self.circuit_open_since is None:
return True
return datetime.now() - self.circuit_open_since > self.CIRCUIT_RESET_TIMEOUT
def _reset_circuit(self):
"""Reset circuit breaker sau timeout"""
print("🔄 Circuit breaker reset - thử khôi phục")
self.circuit_open = False
self.circuit_open_since = None
self.failure_count = 0Tầng 3: State Persistence & Recovery
import redis
import json
from typing import Optional, Any
from dataclasses import dataclass, asdict
from datetime import datetime
@dataclass
class AgentCheckpoint:
"""Checkpoint lưu trạng thái agent tại mỗi bước"""
session_id: str
step: int
total_steps: int
current_state: dict
conversation_history: list
tool_results: dict
created_at: str
expires_at: str # TTL cho checkpoint
def to_json(self) -> str:
return json.dumps(asdict(self))
@classmethod
def from_json(cls, data: str) -> 'AgentCheckpoint':
return cls(**json.loads(data))
class StateRecoveryManager:
"""Quản lý checkpoint và khôi phục trạng thái"""
CHECKPOINT_TTL = 3600 * 24 # 24 giờ
MAX_CHECKPOINTS_PER_SESSION = 100
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)
self.checkpoint_prefix = "agent:checkpoint:"
self.lock_prefix = "agent:lock:"
async def save_checkpoint(self, checkpoint: AgentCheckpoint) -> bool:
"""Lưu checkpoint với atomic operation"""
key = f"{self.checkpoint_prefix}{checkpoint.session_id}:{checkpoint.step}"
# Sử dụng pipeline để đảm bảo atomic
pipe = self.redis.pipeline()
pipe.set(key, checkpoint.to_json(), ex=self.CHECKPOINT_TTL)
pipe.zadd(
f"{self.checkpoint_prefix}{checkpoint.session_id}:steps",
{str(checkpoint.step): checkpoint.created_at}
)
# Cleanup old checkpoints
pipe.zremrangebyrank(
f"{self.checkpoint_prefix}{checkpoint.session_id}:steps",
0,
-self.MAX_CHECKPOINTS_PER_SESSION - 1
)
results = pipe.execute()
return results[0] # True nếu SET thành công
async def load_latest_checkpoint(self, session_id: str) -> Optional[AgentCheckpoint]:
"""Khôi phục checkpoint mới nhất cho session"""
steps_key = f"{self.checkpoint_prefix}{session_id}:steps"
# Lấy step mới nhất
latest_step = self.redis.zrevrange(steps_key, 0, 0)
if not latest_step:
return None
step_num = int(latest_step[0])
key = f"{self.checkpoint_prefix}{session_id}:{step_num}"
data = self.redis.get(key)
if data:
return AgentCheckpoint.from_json(data)
return None
async def acquire_lock(self, session_id: str, ttl: int = 30) -> bool:
"""Acquire distributed lock cho session để tránh race condition"""
lock_key = f"{self.lock_prefix}{session_id}"
return bool(self.redis.set(lock_key, "1", nx=True, ex=ttl))
async def release_lock(self, session_id: str):
"""Release distributed lock"""
lock_key = f"{self.lock_prefix}{session_id}"
self.redis.delete(lock_key)
class RecoveryOrchestrator:
"""Điều phối quá trình khôi phục từ checkpoint"""
def __init__(
self,
state_manager: StateRecoveryManager,
executor: ResilientAgentExecutor
):
self.state_manager = state_manager
self.executor = executor
async def recover_and_continue(
self,
session_id: str,
agent_factory: callable
) -> RecoveryResult:
"""
Khôi phục session từ checkpoint và tiếp tục execution
"""
# 1. Acquire lock để tránh conflict
if not await self.state_manager.acquire_lock(session_id):
return RecoveryResult(
success=False,
error="Session đang được xử lý bởi process khác"
)
try:
# 2. Load checkpoint
checkpoint = await self.state_manager.load_latest_checkpoint(session_id)
if checkpoint is None:
# Không có checkpoint - bắt đầu mới
return RecoveryResult(success=True, is_new_session=True)
# 3. Tạo agent instance từ checkpoint state
agent = agent_factory(checkpoint.current_state)
# 4. Xác định step tiếp theo
next_step = checkpoint.step + 1
remaining_steps = range(next_step, checkpoint.total_steps)
# 5. Execute các bước còn lại với retry
context = ExecutionContext(
attempt=0,
max_attempts=3,
state=AgentState.RECOVERED
)
results = []
for step in remaining_steps:
context.attempt = 0
result = await self.executor.execute_with_retry(
lambda: agent.execute_step(step),
context,
error_category="server_error"
)
if not result.success:
# Lưu checkpoint trước khi fail
await self._save_checkpoint(
session_id, step, agent,
checkpoint.conversation_history,
results
)
return RecoveryResult(
success=False,
error=result.error,
recovered_steps=len(results)
)
results.append(result.data)
# Save checkpoint sau mỗi step thành công
await self._save_checkpoint(
session_id, step, agent,
checkpoint.conversation_history + [result.data],
results
)
return RecoveryResult(
success=True,
recovered_steps=len(results),
data=results[-1] if results else None
)
finally:
await self.state_manager.release_lock(session_id)
async def _save_checkpoint(
self,
session_id: str,
step: int,
agent,
history: list,
results: list
):
"""Helper để save checkpoint"""
now = datetime.now()
checkpoint = AgentCheckpoint(
session_id=session_id,
step=step,
total_steps=agent.total_steps,
current_state=agent.get_state(),
conversation_history=history,
tool_results={r.tool: r.data for r in results},
created_at=now.isoformat(),
expires_at=(now + timedelta(seconds=self.state_manager.CHECKPOINT_TTL)).isoformat()
)
await self.state_manager.save_checkpoint(checkpoint)Tầng 4: Graceful Degradation & Fallback
from enum import Enum
from typing import Optional, Callable
import logging
logger = logging.getLogger(__name__)
class DegradationLevel(Enum):
FULL = "full" # Hoạt động bình thường
REDUCED = "reduced" # Một số tính năng bị disable
MINIMAL = "minimal" # Chỉ core functionality
FALLBACK = "fallback" # Sử dụng rule-based fallback
MAINTENANCE = "down" # Không phục vụ
class GracefulDegradationManager:
"""Quản lý degradation tiers cho AI Agent"""
def __init__(self):
self.current_level = DegradationLevel.FULL
self.degradation_history = []
# Định nghĩa fallback chains
self.fallback_chains = {
"rag_search": [
self._rag_vector_search,
self._rag_keyword_search,
self._keyword_bm25_search,
self._static_faq_lookup
],
"intent_classification": [
self._llm_classification,
self._keyword_classification,
self._rule_based_classification
],
"response_generation": [
self._full_llm_generation,
self._condensed_generation,
self._template_generation,
self._escalate_human
]
}
async def execute_with_fallback(
self,
capability: str,
context: dict,
**kwargs
) -> FallbackResult:
"""Execute với chain of fallbacks"""
chain = self.fallback_chains.get(capability, [])
last_error = None
for idx, fallback_fn in enumerate(chain):
try:
result = await fallback_fn(context, **kwargs)
return FallbackResult(
success=True,
data=result,
fallback_level=idx,
strategy=fallback_fn.__name__
)
except Exception as e:
last_error = e
logger.warning(
f"Fallback {idx+1}/{len(chain)} failed for {capability}: {e}"
)
# Record degradation event
self._record_degradation(capability, fallback_fn.__name__, str(e))
# Nếu đây là fallback cuối cùng
if idx == len(chain) - 1:
return FallbackResult(
success=False,
error=str(last_error),
fallback_level=idx + 1,
escalated=True
)
return FallbackResult(success=False, error=str(last_error))
# RAG Fallback Chain
async def _rag_vector_search(self, ctx, **kwargs):
"""Tier 1: Full vector search với re-ranking"""
# Gọi embedding + vector DB + re-ranker
results = await vector_search_with_rerank(ctx["query"])
if not results or results[0].score < 0.7:
raise InsufficientResultsError("Vector search không đủ confident")
return results
async def _rag_keyword_search(self, ctx, **kwargs):
"""Tier 2: Hybrid search (vector + keyword)"""
results = await hybrid_search(ctx["query"])
if not results:
raise NoResultsError("Hybrid search không có kết quả")
return results
async def _keyword_bm25_search(self, ctx, **kwargs):
"""Tier 3: Pure keyword với BM25"""
results = await bm25_search(ctx["query"])
if not results:
raise NoResultsError("BM25 không có kết quả")
return results
async def _static_faq_lookup(self, ctx, **kwargs):
"""Tier 4: Static FAQ/knowledge base"""
# Sử dụng pre-defined FAQ
return await faq_matcher(ctx["query"])
# Classification Fallback Chain
async def _llm_classification(self, ctx, **kwargs):
"""Tier 1: LLM-based intent classification"""
response = await self.executor.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": INTENT_CLASSIFIER_PROMPT},
{"role": "user", "content": ctx["message"]}
],
temperature=0.1
)
intent = json.loads(response.choices[0].message.content)
if intent["confidence"] < 0.8:
raise LowConfidenceError(f"Confidence {intent['confidence']} thấp")
return intent
async def _keyword_classification(self, ctx, **kwargs):
"""Tier 2: Keyword-based classification"""
return keyword_intent_classifier(ctx["message"])
async def _rule_based_classification(self, ctx, **kwargs):
"""Tier 3: Rule-based fallback"""
return rule_based_intent(ctx["message"])
def _record_degradation(self, capability: str, strategy: str, error: str):
"""Ghi nhận sự kiện degradation"""
event = {
"capability": capability,
"strategy": strategy,
"error": error,
"timestamp": datetime.now().isoformat(),
"degradation_level": self.current_level.value
}
self.degradation_history.append(event)
# Auto-adjust degradation level
self._update_degradation_level()
def _update_degradation_level(self):
"""Tự động điều chỉnh degradation level"""
recent_failures = [
e for e in self.degradation_history
if datetime.fromisoformat(e["timestamp"]) > datetime.now() - timedelta(minutes=5)
]
if len(recent_failures) > 10:
self.current_level = DegradationLevel.MINIMAL
elif len(recent_failures) > 5:
self.current_level = DegradationLevel.REDUCED
elif len(recent_failures) > 0:
if self.current_level != DegradationLevel.FULL:
self.current_level = DegradationLevel.FULL # RecoverLỗi thường gặp và cách khắc phục
1. Lỗi "Connection timeout sau đúng 30 giây"
Nguyên nhân: Mặc định timeout của nhiều HTTP client là 30s, trong khi HolySheep AI API với latency trung bình <50ms thường response nhanh hơn. Nhưng khi hệ thống load cao hoặc query phức tạp, có thể vượt ngưỡng.
# ❌ Code lỗi - timeout mặc định
client = OpenAI(api_key=key, base_url="https://api.holysheep.ai/v1")
response = client.chat.completions.create(model="gpt-4o", messages=messages)
Timeout sau 30s mà không có feedback
✅ Fix - cấu hình timeout hợp lý
from openai import AsyncOpenAI
client = AsyncOpenAI(
api_key=key,
base_url="https://api.holysheep.ai/v1",
timeout=Timeout(120, connect=10) # 120s total, 10s connect
)
Với retry logic
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=30),
retry=retry_if_exception_type(TimeoutError)
)
async def call_with_timeout():
return await client.chat.completions.create(
model="gpt-4o",
messages=messages,
timeout=Timeout(120, connect=10)
)2. Lỗi "Rate limit 429 nhưng không respect Retry-After"
Nguyên nhân: Không parse header Retry-After từ response, dùng fixed backoff thay vì dynamic wait time mà server yêu cầu.
# ❌ Code lỗi - ignore Retry-After header
async def call_api():
try:
return await client.chat.completions.create(...)
except RateLimitError:
await asyncio.sleep(60) # Fixed 60s - không đúng với server requirement
return await call_api()
✅ Fix - parse và sử dụng Retry-After header
async def call_api_with_retry_after():
max_retries = 5
for attempt in range(max_retries):
try:
return await client.chat.completions.create(...)
except RateLimitError as e:
# Parse Retry-After header
retry_after = e.response.headers.get('Retry-After')
if retry_after:
wait_seconds = int(retry_after)
else:
# Fallback exponential backoff
wait_seconds = 2 ** attempt + random.uniform(0, 1)
print(f"Rate limited. Chờ {wait_seconds}s (attempt {attempt + 1}/{max_retries})")
if attempt < max_retries - 1:
await asyncio.sleep(wait_seconds)
else:
raise e3. Lỗi "Context tràn nhưng không có checkpoint"
Nguyên nhân: Khi conversation dài, context window đầy trước khi hoàn thành task. Không có checkpoint → mất toàn bộ progress.
# ❌ Code lỗi - không tracking context usage
async def chat_loop():
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
while True:
user_input = await get_input()
messages.append({"role": "user", "content": user_input})
# Không check context size → eventual crash
response = await client.chat.completions.create(
model="gpt-4o",
messages=messages
)
messages.append(response.choices[0].message)
# → ContextOverflowError sau nhiều turns
✅ Fix - smart context management với summarization
class ContextManager:
MAX_TOKENS = 120_000 # 128K context, giữ margin
SUMMARY_TRIGGER = 100_000
SUMMARY_MODEL = "gpt-4o-mini" # Rẻ hơn 90% cho summarization
def __init__(self, client):
self.client = client
self.messages = []
self.summary_cache = ""
async def add_message(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
await self._check_and_summarize()
async def _check_and_summarize(self):
# Estimate tokens (rough: 4 chars = 1 token)
total_chars = sum(len(m["content"]) for m in self.messages)
estimated_tokens = total_chars // 4
if estimated_tokens > self.SUMMARY_TRIGGER:
# Summarize oldest messages
to_summarize = self.messages[1:-5] # Keep system + recent
summary_prompt = f"""Summarize this conversation concisely,
preserving key facts, user preferences, and pending issues:
{to_summarize}"""
summary_response = await self.client.chat.completions.create(
model=self.SUMMARY_MODEL,
messages=[
{"role": "system", "content": "Bạn là AI tóm tắt hội thoại."},
{"role": "user", "content": summary_prompt}
],
temperature=0.3
)
self.summary_cache = summary_response.choices[0].message.content
# Replace old messages with summary
self.messages = [
self.messages[0], # System prompt
{"role": "assistant", "content": f"[TÓM TẮT CUỘC HỘI LOẠI]: {self.summary_cache}"},
*self.messages[-4:] # Keep 4 recent messages
]
print(f"Context summarized: {len(to_summarize)} msgs → 1 summary")4. Lỗi "JSON parse error từ model output"
Nguyên nhân: Model không luôn trả về JSON valid, đặc biệt khi output phức tạp hoặc gần context limit.
# ❌ Code lỗi - assume JSON always valid
response = await client.chat.completions.create(...)
data = json.loads(response.choices[0].message.content) # Crash here
✅ Fix - robust JSON extraction với fallback
async def extract_jsonrobust(messages: list) -> dict:
"""Extract JSON với nhiều fallback strategies"""
# Strategy 1: Direct parse
try:
response = await client.chat.completions.create(
model="gpt-4o",
messages=messages,
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
except (json.JSONDecodeError, KeyError):
pass
# Strategy 2: Extract from markdown code block
raw = response.choices[0].message.content
json_match = re.search(r'``(?:json)?\s*(\{.*?\})\s*``', raw, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group(1))
except json.JSONDecodeError:
pass
# Strategy 3: Find first { and last }
first_brace = raw.find('{')
last_brace = raw.rfind('}')
if first_brace != -1 and last_brace != -1:
potential_json = raw[first_brace:last_brace + 1]
try:
return json.loads(potential_json)
except json.JSONDecodeError:
pass
# Strategy 4: Ask model to fix
messages.append({
"role": "user",
"content": f"Format lại thành JSON valid: {raw}"
})
response = await client.chat.completions.create(
model="gpt-4o",
messages=messages
)
return json.loads(response.choices[0].message.content)Giám Sát và Observability
Error handling không hiệu quả nếu không có visibility. Đây là dashboard metrics tôi sử dụng:
# Metrics cho error tracking
AGENT_METRICS = {
# Error rates by category
"error_rate_by_type": {
"rate_limit": counter("agent_rate_limit_errors_total"),
"timeout": counter("agent_timeout_errors_total"),
"validation": counter("agent_validation_errors_total"),
"server": counter("agent_server_errors_total"),
},
# Latency percentiles (P50, P95, P99)
"response_latency": histogram(
"agent_response_latency_seconds",
buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
),
# Retry statistics
"retry_stats": {
"total_retries": counter("agent_retries_total"),
"successful_retries": counter("agent_retries_succeeded_total"),
"failed_retries": counter("agent_retries_failed_total"),
},
# Circuit breaker state
"circuit_breaker_state": gauge("agent_circuit_breaker_open"),
# Cost tracking
"api_cost_usd": counter("agent_api_cost_dollars"),
}
Alerting rules
ALERT_RULES = [
{
"name": "HighErrorRate",
"condition": "error_rate > 0.05", # >5% errors
"severity": "critical",
"action": "page_oncall"
},
{
"name": "CircuitBreakerOpen",
"condition": "circuit_breaker_state == 1",
"severity": "warning",
"action": "slack_notify"
},
{
"name": "CostSpike",
"condition": "cost_increase_1h > 10x_baseline",
"severity": "warning",
"action": "reduce_traffic + notify"
}
]Kết Luận
Từ kinh nghiệm vận hành hệ thống AI Agent quy mô 50 triệu người dùng, tôi rút ra:
- Luôn validate đầu vào - 80% lỗi production đến từ input không kiểm soát
- Retry có giới hạn và backoff thông minh - Exponential backoff + jitter + rate limit awareness
- Checkpoint liên tục - State persistence là cứu cánh khi hệ thống crash
- Fallback chain đầy đủ - Luôn có degradation path thay vì hard fail
- Monitor mọi thứ - Metrics, logs, alerts cho error handling cũng cần monitoring
Với HolySheep AI, tôi tiết kiệm được 85%+ chi phí API (chỉ $0.42/MTok cho DeepSeek V3.2 so với $8/MTok của GPT-4.1) trong khi vẫn đảm bảo độ trễ <50ms. Điều này cho phép implement error handling redundant mà không lo về cost explosion.
Error handling không phải là "nice to have" - đó là foundation để AI Agent có