Agent 错误恢复机制：重试/回滚/人工介入设计

Khi xây dựng hệ thống AI Agent trong môi trường production, điều tôi đã học được sau hơn 3 năm triển khai thực chiến là: không có hệ thống nào hoàn hảo. Điều quan trọng không phải là tránh lỗi, mà là ứng phó với lỗi như thế nào. Trong bài viết này, tôi sẽ chia sẻ kiến trúc error recovery tôi đã áp dụng cho nhiều dự án, kèm theo code mẫu có thể sao chép ngay.

Bảng so sánh: HolySheep vs API chính thức vs Proxy khác

Trước khi đi vào chi tiết kỹ thuật, hãy cùng xem xét lý do tại sao việc chọn đúng API provider ảnh hưởng trực tiếp đến chiến lược error recovery của bạn:

Như bạn thấy, HolySheep cung cấp độ trễ thấp hơn đáng kể với chi phí tiết kiệm 85%+. Điều này ảnh hưởng trực tiếp đến strategy retry của bạn — với độ trễ thấp hơn, bạn có thể retry nhiều hơn mà không lo về timeout tổng thể.

Tại sao cần Error Recovery cho Agent?

Trong kiến trúc Agent thông thường, chuỗi xử lý có thể gồm: Planning → Tool Calling → Execution → Evaluation → Response. Mỗi bước đều có thể thất bại:

• Lỗi mạng: Request không đến được server hoặc response bị mất
• Lỗi rate limit: Quá nhiều request trong thời gian ngắn
• Lỗi model: Model trả về JSON không hợp lệ, nội dung độc hại, hoặc hallucination
• Lỗi tool: API tool (tìm kiếm, database) không khả dụng
• Lỗi timeout: Request mất quá lâu để xử lý

Trong thực tế triển khai tại HolySheep AI, tôi đã gặp trường hợp một agent xử lý đơn hàng bị timeout ở bước gọi tool thứ 3, và nếu không có cơ chế recovery tốt, toàn bộ session sẽ bị mất.

Kiến trúc Error Recovery 3 tầng

Tôi đề xuất kiến trúc 3 tầng với độ phức tạp tăng dần:

Tầng 1: Retry với Exponential Backoff

Đây là tầng cơ bản nhất, xử lý các lỗi tạm thời như network timeout hoặc rate limit. Nguyên tắc: mỗi lần retry, thời gian chờ tăng gấp đôi để tránh overload server.

"""
Agent Error Recovery System - Tầng 1: Retry với Exponential Backoff
Base URL: https://api.holysheep.ai/v1
"""

import asyncio
import aiohttp
import time
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum

class RetryStrategy(Enum):
    IMMEDIATE = "immediate"
    LINEAR = "linear"
    EXPONENTIAL = "exponential"
    FIBONACCI = "fibonacci"

@dataclass
class RetryConfig:
    max_retries: int = 5
    base_delay: float = 1.0
    max_delay: float = 60.0
    strategy: RetryStrategy = RetryStrategy.EXPONENTIAL
    jitter: bool = True
    retryable_status_codes: tuple = (408, 429, 500, 502, 503, 504)

class RetryHandler:
    def __init__(self, config: RetryConfig):
        self.config = config
        self.retry_count = 0
    
    def calculate_delay(self) -> float:
        """Tính toán thời gian chờ với chiến lược đã chọn"""
        if self.config.strategy == RetryStrategy.EXPONENTIAL:
            delay = self.config.base_delay * (2 ** self.retry_count)
        elif self.config.strategy == RetryStrategy.LINEAR:
            delay = self.config.base_delay * (self.retry_count + 1)
        elif self.config.strategy == RetryStrategy.FIBONACCI:
            delay = self.config.base_delay * self._fibonacci(self.retry_count + 1)
        else:
            delay = self.config.base_delay
        
        # Thêm jitter để tránh thundering herd
        if self.config.jitter:
            import random
            delay = delay * (0.5 + random.random())
        
        return min(delay, self.config.max_delay)
    
    @staticmethod
    def _fibonacci(n: int) -> int:
        if n <= 1:
            return 1
        a, b = 1, 1
        for _ in range(n - 1):
            a, b = b, a + b
        return b
    
    async def execute_with_retry(
        self,
        session: aiohttp.ClientSession,
        url: str,
        headers: Dict[str, str],
        payload: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Thực thi request với retry logic"""
        last_error = None
        
        for attempt in range(self.config.max_retries + 1):
            try:
                async with session.post(url, headers=headers, json=payload) as response:
                    if response.status == 200:
                        self.retry_count = 0
                        return await response.json()
                    
                    # Kiểm tra có phải retryable error không
                    if response.status in self.config.retryable_status_codes:
                        # Parse retry-after header nếu có
                        retry_after = response.headers.get('Retry-After')
                        if retry_after and attempt == 0:  # Ưu tiên server recommendation
                            delay = float(retry_after)
                        else:
                            delay = self.calculate_delay()
                        
                        self.retry_count = attempt
                        print(f"[RetryHandler] Attempt {attempt + 1} failed with "
                              f"status {response.status}. Retrying in {delay:.2f}s...")
                        
                        await asyncio.sleep(delay)
                        last_error = f"HTTP {response.status}"
                        continue
                    else:
                        # Non-retryable error
                        error_text = await response.text()
                        raise Exception(f"Non-retryable error: {response.status} - {error_text}")
                        
            except aiohttp.ClientError as e:
                delay = self.calculate_delay()
                self.retry_count = attempt
                print(f"[RetryHandler] Connection error (attempt {attempt + 1}): {e}. "
                      f"Retrying in {delay:.2f}s...")
                
                await asyncio.sleep(delay)
                last_error = str(e)
        
        raise Exception(f"Max retries ({self.config.max_retries}) exceeded. Last error: {last_error}")

Ví dụ sử dụng
async def main():
    config = RetryConfig(
        max_retries=5,
        base_delay=1.0,
        max_delay=30.0,
        strategy=RetryStrategy.EXPONENTIAL
    )
    handler = RetryHandler(config)
    
    headers = {
        "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "gpt-4.1",
        "messages": [{"role": "user", "content": "Xử lý đơn hàng #12345"}],
        "temperature": 0.7,
        "max_tokens": 1000
    }
    
    timeout = aiohttp.ClientTimeout(total=120)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        result = await handler.execute_with_retry(
            session,
            "https://api.holysheep.ai/v1/chat/completions",
            headers,
            payload
        )
        print(f"Success: {result}")

if __name__ == "__main__":
    asyncio.run(main())

Tầng 2: State Rollback với Checkpoint

Khi retry không thành công hoặc cần khôi phục trạng thái Agent, checkpoint system giúp lưu lại tiến trình để có thể resume mà không mất dữ liệu.

"""
Agent Error Recovery System - Tầng 2: State Rollback với Checkpoint
"""

import json
import hashlib
import time
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, asdict
from datetime import datetime
from enum import Enum

class ActionType(Enum):
    TOOL_CALL = "tool_call"
    LLM_RESPONSE = "llm_response"
    USER_INPUT = "user_input"
    STATE_UPDATE = "state_update"

@dataclass
class Checkpoint:
    checkpoint_id: str
    agent_id: str
    step: int
    timestamp: str
    state_snapshot: Dict[str, Any]
    last_action: Dict[str, Any]
    parent_checkpoint_id: Optional[str] = None
    checksum: Optional[str] = None

class CheckpointManager:
    def __init__(self, storage_path: str = "./checkpoints"):
        self.storage_path = storage_path
        self.current_checkpoint: Optional[Checkpoint] = None
        self.checkpoint_chain: List[Checkpoint] = []
    
    def create_checkpoint(
        self,
        agent_id: str,
        step: int,
        state: Dict[str, Any],
        last_action: Dict[str, Any]
    ) -> Checkpoint:
        """Tạo checkpoint với checksum để verify integrity"""
        checkpoint = Checkpoint(
            checkpoint_id=self._generate_id(agent_id, step),
            agent_id=agent_id,
            step=step,
            timestamp=datetime.utcnow().isoformat(),
            state_snapshot=self._deep_copy(state),
            last_action=last_action,
            parent_checkpoint_id=(
                self.current_checkpoint.checkpoint_id 
                if self.current_checkpoint else None
            )
        )
        
        # Tính checksum cho integrity check
        checkpoint.checksum = self._calculate_checksum(checkpoint)
        
        # Lưu checkpoint cũ vào chain
        if self.current_checkpoint:
            self.checkpoint_chain.append(self.current_checkpoint)
        
        self.current_checkpoint = checkpoint
        
        # Auto-save
        self._save_checkpoint(checkpoint)
        
        return checkpoint
    
    def rollback_to_checkpoint(
        self, 
        checkpoint_id: str
    ) -> Dict[str, Any]:
        """Khôi phục trạng thái về checkpoint cụ thể"""
        # Tìm checkpoint trong chain
        target = None
        if self.current_checkpoint and self.current_checkpoint.checkpoint_id == checkpoint_id:
            target = self.current_checkpoint
        else:
            target = next(
                (cp for cp in self.checkpoint_chain if cp.checkpoint_id == checkpoint_id),
                None
            )
        
        if not target:
            raise ValueError(f"Checkpoint {checkpoint_id} not found")
        
        # Verify integrity trước khi restore
        if not self._verify_checksum(target):
            raise Exception(f"Checkpoint {checkpoint_id} checksum mismatch - data may be corrupted!")
        
        # Truncate chain từ checkpoint đó trở đi
        rollback_idx = None
        for i, cp in enumerate(self.checkpoint_chain):
            if cp.checkpoint_id == checkpoint_id:
                rollback_idx = i
                break
        
        if rollback_idx is not None:
            self.checkpoint_chain = self.checkpoint_chain[:rollback_idx]
        
        self.current_checkpoint = target
        
        print(f"[CheckpointManager] Rolled back to checkpoint {checkpoint_id} "
              f"(step {target.step}, timestamp: {target.timestamp})")
        
        return self._deep_copy(target.state_snapshot)
    
    def rollback_n_steps(self, n: int) -> Dict[str, Any]:
        """Rollback về n bước trước đó"""
        if n > len(self.checkpoint_chain):
            raise ValueError(f"Cannot rollback {n} steps, only {len(self.checkpoint_chain)} available")
        
        target = self.checkpoint_chain[-(n)]
        return self.rollback_to_checkpoint(target.checkpoint_id)
    
    def get_checkpoint_info(self, checkpoint_id: str) -> Optional[Dict[str, Any]]:
        """Lấy thông tin checkpoint mà không load full state"""
        for cp in [self.current_checkpoint] + self.checkpoint_chain:
            if cp and cp.checkpoint_id == checkpoint_id:
                return {
                    "checkpoint_id": cp.checkpoint_id,
                    "step": cp.step,
                    "timestamp": cp.timestamp,
                    "has_parent": cp.parent_checkpoint_id is not None,
                    "chain_length": len(self.checkpoint_chain)
                }
        return None
    
    def _generate_id(self, agent_id: str, step: int) -> str:
        """Tạo ID duy nhất cho checkpoint"""
        data = f"{agent_id}:{step}:{time.time()}"
        return hashlib.sha256(data.encode()).hexdigest()[:16]
    
    def _calculate_checksum(self, checkpoint: Checkpoint) -> str:
        """Tính checksum để verify data integrity"""
        data = (
            f"{checkpoint.checkpoint_id}"
            f"{checkpoint.agent_id}"
            f"{checkpoint.step}"
            f"{checkpoint.timestamp}"
            f"{json.dumps(checkpoint.state_snapshot, sort_keys=True)}"
            f"{json.dumps(checkpoint.last_action, sort_keys=True)}"
        )
        return hashlib.sha256(data.encode()).hexdigest()
    
    def _verify_checksum(self, checkpoint: Checkpoint) -> bool:
        """Verify checksum của checkpoint"""
        expected = self._calculate_checksum(checkpoint)
        return expected == checkpoint.checksum
    
    def _deep_copy(self, obj: Any) -> Any:
        """Deep copy để tránh reference issues"""
        return json.loads(json.dumps(obj))
    
    def _save_checkpoint(self, checkpoint: Checkpoint):
        """Lưu checkpoint xuống disk"""
        import os
        os.makedirs(self.storage_path, exist_ok=True)
        path = os.path.join(
            self.storage_path, 
            f"checkpoint_{checkpoint.checkpoint_id}.json"
        )
        with open(path, 'w') as f:
            json.dump(asdict(checkpoint), f, indent=2)

Ví dụ sử dụng trong Agent workflow
class OrderProcessingAgent:
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.state = {
            "order_id": None,
            "customer_info": {},
            "items": [],
            "total": 0,
            "status": "init"
        }
        self.checkpoint_manager = CheckpointManager()
        self.step = 0
    
    async def process_order(self, order_id: str, customer_id: str):
        try:
            self.state["order_id"] = order_id
            self.state["status"] = "fetching_customer"
            
            # Bước 1: Lấy thông tin khách hàng
            self._create_checkpoint("fetch_customer")
            customer = await self._fetch_customer(customer_id)
            self.state["customer_info"] = customer
            
            # Bước 2: Validate đơn hàng
            self._create_checkpoint("validate_order")
            await self._validate_order()
            self.state["status"] = "validated"
            
            # Bước 3: Tính tổng (có thể lỗi ở đây)
            self._create_checkpoint("calculate_total")
            await self._calculate_total()
            
            # Bước 4: Xử lý thanh toán
            self._create_checkpoint("process_payment")
            await self._process_payment()
            
            self.state["status"] = "completed"
            print(f"[Agent] Order {order_id} completed successfully")
            
        except PaymentError as e:
            print(f"[Agent] Payment failed: {e}")
            # Rollback về checkpoint calculate_total để retry
            self.state = self.checkpoint_manager.rollback_n_steps(1)
            # Retry với fallback payment method
            await self._process_payment(method="fallback")
            
        except Exception as e:
            print(f"[Agent] Unrecoverable error: {e}")
            # Rollback về checkpoint cuối cùng
            if self.checkpoint_manager.checkpoint_chain:
                self.state = self.checkpoint_manager.rollback_to_checkpoint(
                    self.checkpoint_manager.checkpoint_chain[-1].checkpoint_id
                )
            raise
    
    def _create_checkpoint(self, action_name: str):
        self.step += 1
        self.checkpoint_manager.create_checkpoint(
            agent_id=self.agent_id,
            step=self.step,
            state=self.state,
            last_action={"type": ActionType.LLM_RESPONSE.value, "name": action_name}
        )
    
    # Mock methods cho demo
    async def _fetch_customer(self, customer_id: str):
        await asyncio.sleep(0.1)
        return {"id": customer_id, "name": "Nguyễn Văn A", "credit": 1000}
    
    async def _validate_order(self):
        await asyncio.sleep(0.1)
        self.state["items"] = [{"id": 1, "price": 100}, {"id": 2, "price": 200}]
    
    async def _calculate_total(self):
        await asyncio.sleep(0.1)
        self.state["total"] = sum(item["price"] for item in self.state["items"])
    
    async def _process_payment(self, method: str = "primary"):
        await asyncio.sleep(0.1)
        if method == "primary" and self.state["total"] > 250:
            raise PaymentError("Primary payment method declined")
        self.state["payment_method"] = method
        self.state["paid"] = True

class PaymentError(Exception):
    pass

Test
import asyncio
async def test_agent():
    agent = OrderProcessingAgent("agent_001")
    await
Tài nguyên liên quan
📚 Hướng dẫn AI API
💰 Xem giá
📖 Tài liệu nhà phát triển
🚀 Đăng ký miễn phí
Bài viết liên quan
Agent 安全护栏：防止提示注入的 5 层防御
Multi-Agent Resource Competition: Distributed Lock & Task Qu
React Tích Hợp AI API: Streaming Response + Markdown Renderi