Claude API Error Handling — Retry, Rate Limit và Production Resilience

Khi triển khai Claude API trong môi trường production, lỗi là điều không thể tránh khỏi. Server quá tải, rate limit bị vượt ngưỡng, hoặc mạng gián đoạn tạm thời. Ứng dụng chuyên nghiệp không phải là ứng dụng không bao giờ gặp lỗi, mà là ứng dụng biết cách xử lý lỗi một cách thông minh. Bài viết này hướng dẫn các pattern xử lý lỗi từ cơ bản đến nâng cao cho Claude API.

Các loại lỗi thường gặp

Claude API trả về các HTTP status code chuẩn. Hiểu rõ từng loại lỗi giúp bạn quyết định chiến lược xử lý phù hợp:

Lỗi có thể retry (Retryable errors)

Lỗi không nên retry (Non-retryable errors)

Exponential Backoff với Jitter

Exponential backoff là kỹ thuật tăng thời gian chờ theo cấp số nhân sau mỗi lần retry thất bại. Jitter thêm yếu tố ngẫu nhiên để tránh "thundering herd" — hiện tượng nhiều client đồng loạt retry cùng lúc sau khi server phục hồi.

import time
import random
import anthropic

def call_with_retry(
    client,
    max_retries=5,
    base_delay=1.0,
    max_delay=60.0,
    **kwargs
):
    """Goi Claude API voi exponential backoff va jitter."""
    last_exception = None

    for attempt in range(max_retries + 1):
        try:
            response = client.messages.create(**kwargs)
            return response

        except anthropic.RateLimitError as e:
            last_exception = e
            # Lay retry-after tu header neu co
            retry_after = getattr(e, "response", None)
            if retry_after and hasattr(retry_after, "headers"):
                wait_time = float(
                    retry_after.headers.get("retry-after", base_delay)
                )
            else:
                wait_time = base_delay * (2 ** attempt)

            # Them jitter (0.5x den 1.5x)
            wait_time = wait_time * (0.5 + random.random())
            wait_time = min(wait_time, max_delay)

            print(f"Rate limit. Retry sau {wait_time:.1f}s "
                  f"(lan {attempt + 1}/{max_retries})")
            time.sleep(wait_time)

        except anthropic.InternalServerError as e:
            last_exception = e
            wait_time = base_delay * (2 ** attempt)
            wait_time = wait_time * (0.5 + random.random())
            wait_time = min(wait_time, max_delay)

            print(f"Server error. Retry sau {wait_time:.1f}s "
                  f"(lan {attempt + 1}/{max_retries})")
            time.sleep(wait_time)

        except anthropic.APIStatusError as e:
            if e.status_code == 529:
                last_exception = e
                wait_time = base_delay * (2 ** attempt) * 2
                wait_time = wait_time * (0.5 + random.random())
                wait_time = min(wait_time, max_delay)

                print(f"Server overloaded. Retry sau {wait_time:.1f}s "
                      f"(lan {attempt + 1}/{max_retries})")
                time.sleep(wait_time)
            else:
                # Loi khong the retry (400, 401, 403, 404)
                raise

        except anthropic.APIConnectionError as e:
            last_exception = e
            wait_time = base_delay * (2 ** attempt)
            wait_time = wait_time * (0.5 + random.random())
            wait_time = min(wait_time, max_delay)

            print(f"Connection error. Retry sau {wait_time:.1f}s "
                  f"(lan {attempt + 1}/{max_retries})")
            time.sleep(wait_time)

    raise last_exception

Circuit Breaker Pattern

Circuit breaker ngăn ứng dụng tiếp tục gửi request khi API đang gặp sự cố nghiêm trọng. Pattern này có ba trạng thái: Closed (hoạt động bình thường), Open (ngắt mạch, từ chối mọi request), và Half-Open (thử gửi một số request để kiểm tra API đã phục hồi chưa).

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold=5,
        recovery_timeout=60,
        half_open_max_calls=3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.half_open_calls = 0

    def can_execute(self):
        """Kiem tra co nen gui request khong."""
        if self.state == CircuitState.CLOSED:
            return True

        if self.state == CircuitState.OPEN:
            # Kiem tra da du thoi gian recovery chua
            if time.time() - self.last_failure_time >= self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
                print("Circuit chuyen sang HALF_OPEN")
                return True
            return False

        if self.state == CircuitState.HALF_OPEN:
            return self.half_open_calls < self.half_open_max_calls

        return False

    def record_success(self):
        """Ghi nhan request thanh cong."""
        if self.state == CircuitState.HALF_OPEN:
            self.half_open_calls += 1
            if self.half_open_calls >= self.half_open_max_calls:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                print("Circuit chuyen sang CLOSED (da phuc hoi)")
        else:
            self.failure_count = 0

    def record_failure(self):
        """Ghi nhan request that bai."""
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN
            print("Circuit chuyen sang OPEN (van loi)")
        elif self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            print(f"Circuit OPEN sau {self.failure_count} loi lien tiep")

Tích hợp Circuit Breaker với API client

class ResilientClaudeClient:
    def __init__(self, api_key=None):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.circuit = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60
        )

    def send_message(self, fallback_response=None, **kwargs):
        """Gui request voi circuit breaker va retry."""
        if not self.circuit.can_execute():
            print("Circuit OPEN - tra ve fallback")
            if fallback_response:
                return fallback_response
            raise Exception("Service dang khong kha dung")

        try:
            response = call_with_retry(
                self.client,
                max_retries=3,
                **kwargs
            )
            self.circuit.record_success()
            return response

        except Exception as e:
            self.circuit.record_failure()
            if fallback_response:
                return fallback_response
            raise

Rate Limit Headers

Claude API trả về các header giúp bạn theo dõi và quản lý rate limit:

Proactive rate limiting

Thay vì chờ bị 429 rồi mới xử lý, bạn có thể chủ động kiểm soát tốc độ gửi request:

import time
import threading

class RateLimiter:
    def __init__(self, max_requests_per_minute=50, max_tokens_per_minute=40000):
        self.max_rpm = max_requests_per_minute
        self.max_tpm = max_tokens_per_minute
        self.request_timestamps = []
        self.token_usage = []
        self.lock = threading.Lock()

    def wait_if_needed(self, estimated_tokens=0):
        """Cho doi neu can thiet de khong vuot rate limit."""
        with self.lock:
            now = time.time()
            one_minute_ago = now - 60

            # Xoa cac record cu hon 1 phut
            self.request_timestamps = [
                t for t in self.request_timestamps if t > one_minute_ago
            ]
            self.token_usage = [
                (t, tokens) for t, tokens in self.token_usage
                if t > one_minute_ago
            ]

            # Kiem tra request limit
            if len(self.request_timestamps) >= self.max_rpm:
                wait_time = self.request_timestamps[0] - one_minute_ago
                print(f"Request limit. Cho {wait_time:.1f}s")
                time.sleep(wait_time + 0.1)

            # Kiem tra token limit
            total_tokens = sum(t for _, t in self.token_usage)
            if total_tokens + estimated_tokens > self.max_tpm:
                wait_time = self.token_usage[0][0] - one_minute_ago
                print(f"Token limit. Cho {wait_time:.1f}s")
                time.sleep(wait_time + 0.1)

            self.request_timestamps.append(time.time())
            self.token_usage.append((time.time(), estimated_tokens))

# Su dung
limiter = RateLimiter(max_requests_per_minute=50)

for item in data_to_process:
    limiter.wait_if_needed(estimated_tokens=500)
    response = client.messages.create(...)

Fallback Strategies

Khi API liên tục lỗi hoặc circuit breaker mở, ứng dụng cần có phương án dự phòng thay vì chết hoàn toàn:

1. Model Downgrade

Chuyển từ model mạnh hơn xuống model nhẹ hơn khi model chính không khả dụng:

MODEL_FALLBACK_CHAIN = [
    "claude-sonnet-4-20250514",
    "claude-haiku-3-5-20241022",
]

def call_with_model_fallback(client, messages, **kwargs):
    """Thu tung model theo thu tu uu tien."""
    for model in MODEL_FALLBACK_CHAIN:
        try:
            response = client.messages.create(
                model=model,
                messages=messages,
                **kwargs
            )
            if model != MODEL_FALLBACK_CHAIN[0]:
                print(f"Da fallback sang model: {model}")
            return response

        except (anthropic.RateLimitError, anthropic.InternalServerError):
            print(f"Model {model} khong kha dung, thu model tiep theo")
            continue

    raise Exception("Tat ca model deu khong kha dung")

2. Queue-based Fallback

Đưa request vào hàng đợi khi API không khả dụng và xử lý sau:

import json
from collections import deque

class RequestQueue:
    def __init__(self, max_size=10000):
        self.queue = deque(maxlen=max_size)
        self.results = {}

    def enqueue(self, request_id, params):
        """Them request vao hang doi."""
        self.queue.append({
            "id": request_id,
            "params": params,
            "enqueued_at": time.time()
        })
        print(f"Request {request_id} da vao hang doi "
              f"(vi tri {len(self.queue)})")

    def process_queue(self, client, batch_size=10):
        """Xu ly hang doi khi API kha dung tro lai."""
        processed = 0
        while self.queue and processed < batch_size:
            item = self.queue.popleft()
            try:
                response = call_with_retry(client, **item["params"])
                self.results[item["id"]] = {
                    "status": "success",
                    "response": response
                }
                processed += 1
            except Exception as e:
                # Dat lai vao hang doi neu van loi
                self.queue.appendleft(item)
                print(f"Van loi, dung xu ly queue. "
                      f"Con {len(self.queue)} request")
                break

        return processed

Triển khai trong Node.js

Tương tự Python, Node.js cần retry logic cho production:

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

async function callWithRetry(params, maxRetries = 5) {
  let lastError;

  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await client.messages.create(params);
    } catch (error) {
      lastError = error;

      if (error instanceof Anthropic.RateLimitError) {
        const retryAfter = error.headers?.["retry-after"];
        const delay = retryAfter
          ? parseFloat(retryAfter) * 1000
          : Math.min(1000 * Math.pow(2, attempt), 60000);

        const jitteredDelay = delay * (0.5 + Math.random());
        console.log(
          `Rate limit. Retry sau ${(jitteredDelay / 1000).toFixed(1)}s`
        );
        await new Promise((r) => setTimeout(r, jitteredDelay));
        continue;
      }

      if (
        error instanceof Anthropic.InternalServerError ||
        error.status === 529
      ) {
        const delay = Math.min(1000 * Math.pow(2, attempt), 60000);
        const jitteredDelay = delay * (0.5 + Math.random());
        await new Promise((r) => setTimeout(r, jitteredDelay));
        continue;
      }

      // Non-retryable error
      throw error;
    }
  }

  throw lastError;
}

Production Error Monitoring

Trong môi trường production, bạn cần monitoring toàn diện để phát hiện vấn đề sớm:

import logging
from datetime import datetime

logger = logging.getLogger("claude_api")

class APIMonitor:
    def __init__(self, alert_threshold=0.1):
        self.total_calls = 0
        self.errors = {429: 0, 500: 0, 529: 0, "other": 0}
        self.latencies = []
        self.alert_threshold = alert_threshold

    def record_call(self, status_code, latency_ms, model=None):
        """Ghi nhan moi API call."""
        self.total_calls += 1
        self.latencies.append(latency_ms)

        if status_code >= 400:
            error_key = status_code if status_code in self.errors else "other"
            self.errors[error_key] = self.errors.get(error_key, 0) + 1

            logger.warning(
                f"API error: {status_code} | model: {model} | "
                f"latency: {latency_ms}ms"
            )

        # Kiem tra error rate
        error_total = sum(self.errors.values())
        error_rate = error_total / self.total_calls if self.total_calls > 0 else 0

        if error_rate > self.alert_threshold:
            self.send_alert(error_rate)

    def send_alert(self, error_rate):
        """Gui canh bao khi error rate vuot nguong."""
        logger.critical(
            f"ALERT: Error rate {error_rate:.1%} "
            f"vuot nguong {self.alert_threshold:.1%}! "
            f"Errors: {self.errors}"
        )
        # Tich hop voi Slack, PagerDuty, hoac he thong canh bao khac

    def get_stats(self):
        """Lay thong ke hieu suat."""
        if not self.latencies:
            return {}

        sorted_lat = sorted(self.latencies)
        p50 = sorted_lat[len(sorted_lat) // 2]
        p95 = sorted_lat[int(len(sorted_lat) * 0.95)]
        p99 = sorted_lat[int(len(sorted_lat) * 0.99)]

        return {
            "total_calls": self.total_calls,
            "error_rate": sum(self.errors.values()) / self.total_calls,
            "errors_by_type": self.errors,
            "latency_p50": f"{p50}ms",
            "latency_p95": f"{p95}ms",
            "latency_p99": f"{p99}ms"
        }

Timeout và Request Management

Ngoài retry và circuit breaker, timeout là lớp bảo vệ quan trọng khác. Một request không có timeout có thể treo vô thời hạn, chiếm giữ tài nguyên và gây ảnh hưởng đến toàn bộ hệ thống.

import anthropic

# Python SDK ho tro timeout config
client = anthropic.Anthropic(
    timeout=60.0,  # 60 giay cho moi request
)

# Hoac timeout cho tung request cu the
import httpx

def call_with_timeout(client, timeout=30.0, **kwargs):
    """Goi API voi timeout cu the."""
    try:
        response = client.messages.create(
            **kwargs,
            timeout=timeout
        )
        return response
    except httpx.TimeoutException:
        print(f"Request timeout sau {timeout}s")
        # Quyet dinh: retry voi timeout dai hon,
        # hoac fallback sang model nhanh hon
        return call_with_timeout(
            client,
            timeout=timeout * 2,
            model="claude-haiku-3-5-20241022",  # model nhanh hon
            **{k: v for k, v in kwargs.items() if k != "model"}
        )

Request ID tracking

Mỗi response từ Claude API chứa request ID trong header. Lưu lại ID này để debug và liên hệ hỗ trợ khi cần:

def call_with_tracking(client, **kwargs):
    """Goi API va luu request ID de debug."""
    try:
        response = client.messages.create(**kwargs)
        # Truy cap request ID tu response
        request_id = response._raw_response.headers.get("x-request-id", "unknown")
        logger.info(f"Request {request_id}: "
                    f"input={response.usage.input_tokens}, "
                    f"output={response.usage.output_tokens}")
        return response
    except anthropic.APIStatusError as e:
        request_id = getattr(e.response, "headers", {}).get(
            "x-request-id", "unknown"
        )
        logger.error(f"Request {request_id} failed: {e.status_code} {e.message}")
        raise

Structured Error Response cho end-user

Khi API lỗi, ứng dụng cần trả về thông báo có ý nghĩa cho người dùng cuối thay vì để lộ chi tiết kỹ thuật:

ERROR_MESSAGES = {
    429: {
        "user_message": "He thong dang xu ly nhieu yeu cau. "
                        "Vui long thu lai sau vai giay.",
        "retry": True
    },
    500: {
        "user_message": "He thong dang gap su co tam thoi. "
                        "Chung toi dang xu ly, vui long thu lai sau.",
        "retry": True
    },
    529: {
        "user_message": "He thong dang qua tai. "
                        "Vui long thu lai sau 1-2 phut.",
        "retry": True
    },
    400: {
        "user_message": "Yeu cau khong hop le. "
                        "Vui long kiem tra lai noi dung.",
        "retry": False
    },
    401: {
        "user_message": "Phien lam viec da het han. "
                        "Vui long dang nhap lai.",
        "retry": False
    }
}

def get_user_friendly_error(status_code):
    """Tra ve thong bao loi than thien voi nguoi dung."""
    error_info = ERROR_MESSAGES.get(status_code, {
        "user_message": "Da xay ra loi. Vui long thu lai sau.",
        "retry": True
    })
    return error_info

Ngoài ra, khi sử dụng trong ứng dụng web, bạn nên trả về HTTP status code phù hợp cho frontend. Ví dụ: lỗi 429 từ Claude API có thể trả về 503 (Service Unavailable) cho client kèm header Retry-After, thay vì trả trực tiếp lỗi 429 vì client không gọi Claude API trực tiếp.

Chiến lược Graceful Degradation

Ứng dụng tốt vẫn hoạt động được khi AI không khả dụng. Dưới đây là các mức degradation từ nhẹ đến nặng:

• Mức 1 - Model downgrade: chuyển từ Sonnet sang Haiku. Chất lượng giảm nhẹ nhưng vẫn có AI response
• Mức 2 - Cached response: trả về response đã cache cho các câu hỏi phổ biến (FAQ). Không cá nhân hóa nhưng vẫn hữu ích
• Mức 3 - Template response: trả về response mẫu dựa trên keyword matching. Không dùng AI nhưng vẫn trả lời được câu hỏi cơ bản
• Mức 4 - Queue và notify: đưa request vào hàng đợi, thông báo cho người dùng sẽ trả lời sau qua email hoặc notification
• Mức 5 - Human handoff: chuyển sang nhân viên hỗ trợ khi tất cả phương án tự động đều thất bại

Checklist triển khai production

Trước khi deploy ứng dụng sử dụng Claude API ra production, hãy kiểm tra danh sách sau:

• Retry logic với exponential backoff và jitter cho lỗi 429, 500, 529
• Circuit breaker để ngắt mạch khi API gặp sự cố kéo dài
• Rate limiter chủ động để tránh vượt ngưỡng
• Model fallback chain khi model chính không khả dụng
• Request queue cho các tác vụ không cần phản hồi tức thì
• Monitoring và alerting cho error rate, latency, và chi phí
• Logging đầy đủ với request ID để debug
• Timeout cho mỗi request (không để request treo vô thời hạn)
• Graceful degradation: ứng dụng vẫn hoạt động được dù AI không khả dụng

Bước tiếp theo

Error handling vững chắc là nền tảng cho mọi ứng dụng production sử dụng Claude API. Kết hợp với Prompt Caching để giảm chi phí và Batch API cho các tác vụ xử lý hàng loạt, bạn sẽ có một hệ thống AI hoàn chỉnh, ổn định và tiết kiệm. Khám phá thêm tại Thư viện Nâng cao Claude.