Hệ Thống Đa Agent: Phân Chia Công Việc, Chia Sẻ Trạng Thái và Thiết Kế Khôi Phục Lỗi

Trong bối cảnh AI ngày càng phát triển, việc xây dựng hệ thống đa agent (multi-agent) đã trở thành xu hướng tất yếu để xử lý các tác vụ phức tạp. Bài viết này sẽ hướng dẫn bạn thiết kế một hệ thống đa agent hoàn chỉnh với khả năng phân chia công việc thông minh, chia sẻ trạng thái hiệu quả và khôi phục lỗi tự động.

Bối Cảnh Thị Trường AI 2026: So Sánh Chi Phí

Trước khi đi vào kỹ thuật, hãy cùng xem xét chi phí vận hành hệ thống đa agent với các mô hình AI hàng đầu hiện nay:

Mô hình	Giá Output (USD/MTok)	Chi phí 10M token/tháng
Claude Sonnet 4.5	$15	$150
GPT-4.1	$8	$80
Gemini 2.5 Flash	$2.50	$25
DeepSeek V3.2	$0.42	$4.20

Như bạn thấy, DeepSeek V3.2 có chi phí thấp nhất — chỉ $4.20 cho 10 triệu token mỗi tháng, trong khi Claude Sonnet 4.5 lên tới $150. Với HolySheep AI, bạn có thể tiết kiệm tới 85%+ chi phí với tỷ giá ưu đãi và thanh toán qua WeChat/Alipay.

Kiến Trúc Tổng Quan Hệ Thống Đa Agent

Một hệ thống đa agent hiệu quả cần có 3 thành phần cốt lõi:

Orchestrator Agent: Điều phối và phân chia công việc
Worker Agents: Thực thi các tác vụ con
Shared State Manager: Quản lý trạng thái và giao tiếp

Xây Dựng Orchestrator Agent - Trung Tâm Điều Phối

Orchestrator là agent chịu trách nhiệm phân tích yêu cầu, phân chia task và điều phối các worker agent. Dưới đây là implementation chi tiết:

import httpx
import asyncio
from typing import List, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
import json
from datetime import datetime

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    RETRY = "retry"

@dataclass
class Task:
    task_id: str
    description: str
    assigned_agent: str = None
    status: TaskStatus = TaskStatus.PENDING
    result: Any = None
    retry_count: int = 0
    created_at: datetime = field(default_factory=datetime.now)

class SharedState:
    """Quản lý trạng thái chia sẻ giữa các agent"""
    
    def __init__(self):
        self._state: Dict[str, Any] = {}
        self._locks: Dict[str, asyncio.Lock] = {}
        self._subscribers: Dict[str, List[callable]] = {}
    
    async def get(self, key: str) -> Any:
        return self._state.get(key)
    
    async def set(self, key: str, value: Any):
        async with self._locks.setdefault(key, asyncio.Lock()):
            self._state[key] = value
            await self._notify_subscribers(key, value)
    
    async def update(self, key: str, updates: Dict):
        async with self._locks.setdefault(key, asyncio.Lock()):
            current = self._state.get(key, {})
            if isinstance(current, dict):
                current.update(updates)
                self._state[key] = current
            await self._notify_subscribers(key, self._state[key])
    
    async def subscribe(self, key: str, callback: callable):
        if key not in self._subscribers:
            self._subscribers[key] = []
        self._subscribers[key].append(callback)
    
    async def _notify_subscribers(self, key: str, value: Any):
        if key in self._subscribers:
            for callback in self._subscribers[key]:
                await callback(key, value)

class OrchestratorAgent:
    """Agent điều phối trung tâm - phân chia công việc cho các worker"""
    
    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.shared_state = SharedState()
        self.worker_agents: Dict[str, 'WorkerAgent'] = {}
    
    async def initialize(self):
        """Khởi tạo các worker agent"""
        self.worker_agents = {
            "research": WorkerAgent("research", self.api_key, self.base_url, self.shared_state),
            "analysis": WorkerAgent("analysis", self.api_key, self.base_url, self.shared_state),
            "coding": WorkerAgent("coding", self.api_key, self.base_url, self.shared_state),
            "validation": WorkerAgent("validation", self.api_key, self.base_url, self.shared_state),
        }
        
        for agent in self.worker_agents.values():
            await agent.start()
    
    async def process_request(self, user_request: str) -> Dict[str, Any]:
        """Phân tích yêu cầu và phân chia công việc"""
        
        # Gọi LLM để phân tích và phân chia task
        prompt = f"""Phân tích yêu cầu sau và phân chia thành các task nhỏ.
        Mỗi task cần có: task_id, mô tả, agent phù hợp (research/analysis/coding/validation).
        
        Yêu cầu: {user_request}
        
        Trả lời theo format JSON với cấu trúc:
        {{"tasks": [{{"task_id": "task_1", "description": "...", "agent": "research"}}]}}
        """
        
        response = await self._call_llm(prompt)
        task_plan = json.loads(response)
        
        # Đăng ký các task vào shared state
        for task_data in task_plan["tasks"]:
            task = Task(
                task_id=task_data["task_id"],
                description=task_data["description"]
            )
            await self.shared_state.set(f"task:{task.task_id}", {
                "status": task.status.value,
                "assigned_to": task_data["agent"],
                "description": task.description
            })
        
        # Phân chia và thực thi song song
        return await self._execute_task_plan(task_plan["tasks"])
    
    async def _call_llm(self, prompt: str, model: str = "deepseek-v3.2") -> str:
        """Gọi API LLM thông qua HolySheep"""
        async with httpx.AsyncClient(timeout=60.0) as client:
            response = await client.post(
                f"{self.base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": model,
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.3
                }
            )
            response.raise_for_status()
            return response.json()["choices"][0]["message"]["content"]
    
    async def _execute_task_plan(self, tasks: List[Dict]) -> Dict[str, Any]:
        """Thực thi plan với xử lý lỗi và retry"""
        results = {}
        
        # Tạo concurrent tasks
        async def execute_single(task: Dict) -> tuple:
            agent_name = task["agent"]
            if agent_name not in self.worker_agents:
                return task["task_id"], {"error": f"Unknown agent: {agent_name}"}
            
            agent = self.worker_agents[agent_name]
            result = await agent.execute_with_retry(task)
            return task["task_id"], result
        
        # Thực thi song song
        task_coroutines = [execute_single(t) for t in tasks]
        completed = await asyncio.gather(*task_coroutines, return_exceptions=True)
        
        for item in completed:
            if isinstance(item, Exception):
                continue
            task_id, result = item
            results[task_id] = result
        
        return results

Worker Agent với Retry Logic và Error Recovery

Mỗi worker agent cần có khả năng tự khôi phục khi gặp lỗi. Dưới đây là implementation với exponential backoff:

import asyncio
import random
from typing import Optional

class WorkerAgent:
    """Worker agent thực thi task với cơ chế retry"""
    
    MAX_RETRIES = 3
    BASE_DELAY = 1.0  # Giây
    MAX_DELAY = 30.0
    
    def __init__(self, name: str, api_key: str, base_url: str, shared_state: SharedState):
        self.name = name
        self.api_key = api_key
        self.base_url = base_url
        self.shared_state = shared_state
        self.is_active = False
    
    async def start(self):
        """Khởi động worker"""
        self.is_active = True
        print(f"Worker {self.name} started")
    
    async def execute_with_retry(self, task: Dict, max_retries: int = None) -> Dict:
        """Thực thi task với retry logic"""
        max_retries = max_retries or self.MAX_RETRIES
        last_error = None
        
        for attempt in range(max_retries):
            try:
                # Cập nhật trạng thái
                await self.shared_state.update(f"task:{task['task_id']}", {
                    "status": "in_progress",
                    "attempt": attempt + 1
                })
                
                # Thực thi task
                result = await self._execute_task(task)
                
                # Cập nhật kết quả thành công
                await self.shared_state.update(f"task:{task['task_id']}", {
                    "status": "completed",
                    "result": result,
                    "completed_at": str(datetime.now())
                })
                
                return {"success": True, "data": result}
                
            except Exception as e:
                last_error = e
                await self.shared_state.update(f"task:{task['task_id']}", {
                    "status": "retry",
                    "last_error": str(e),
                    "attempt": attempt + 1
                })
                
                if attempt < max_retries - 1:
                    # Exponential backoff với jitter
                    delay = min(
                        self.BASE_DELAY * (2 ** attempt) + random.uniform(0, 1),
                        self.MAX_DELAY
                    )
                    print(f"Worker {self.name}: Retry {attempt + 1}/{max_retries} sau {delay:.2f}s")
                    await asyncio.sleep(delay)
        
        # Tất cả retries đều thất bại
        await self.shared_state.update(f"task:{task['task_id']}", {
            "status": "failed",
            "error": str(last_error),
            "failed_at": str(datetime.now())
        })
        
        return {
            "success": False,
            "error": str(last_error),
            "attempts": max_retries
        }
    
    async def _execute_task(self, task: Dict) -> Any:
        """Logic thực thi task cụ thể của từng worker"""
        prompt = f"""Bạn là agent {self.name}. Thực hiện task sau:
        
        Task ID: {task['task_id']}
        Mô tả: {task['description']}
        
        Trả lời bằng JSON với format: {{"result": "...", "confidence": 0.9}}"""
        
        # Chọn model phù hợp với loại worker
        model_map = {
            "research": "deepseek-v3.2",      # Chi phí thấp cho research
            "analysis": "gemini-2.5-flash",   # Cân bằng cost-performance
            "coding": "gpt-4.1",              # Mạnh cho coding
            "validation": "claude-sonnet-4.5" # Chính xác cao
        }
        
        model = model_map.get(self.name, "deepseek-v3.2")
        
        async with httpx.AsyncClient(timeout=120.0) as client:
            response = await client.post(
                f"{self.base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": model,
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.5
                }
            )
            response.raise_for_status()
            result_text = response.json()["choices"][0]["message"]["content"]
            return json.loads(result_text)
    
    async def handle_failure(self, task: Dict, error: Exception) -> Optional[Dict]:
        """Xử lý khi task thất bại - có thể override"""
        print(f"Worker {self.name}: Task {task['task_id']} failed - {error}")
        # Fallback: thử với model khác hoặc task đơn giản hơn
        return None

Shared State Manager - Giao Tiếp Giữa Các Agent

Để các agent có thể chia sẻ thông tin và phối hợp, chúng ta cần một hệ thống shared state với khả năng:

Real-time updates: Cập nhật trạng thái tức thì
Event-driven notifications: Thông báo khi có thay đổi
Conflict resolution: Giải quyết xung đột khi nhiều agent cùng ghi

class AdvancedSharedState(SharedState):
    """Shared state nâng cao với event system và conflict resolution"""
    
    def __init__(self):
        super().__init__()
        self._event_bus: Dict[str, asyncio.Queue] = {}
        self._change_history: Dict[str, List[Dict]] = {}
    
    async def set_with_version(self, key: str, value: Any, expected_version: int = None) -> bool:
        """Set giá trị với optimistic locking"""
        current_version = await self.get(f"{key}._version") or 0
        
        if expected_version is not None and current_version != expected_version:
            raise ConflictError(f"Version mismatch: expected {expected_version}, got {current_version}")
        
        async with self._locks.setdefault(key, asyncio.Lock()):
            self._state[key] = value
            new_version = current_version + 1
            self._state[f"{key}._version"] = new_version
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
vi deepseek r1 tuilimoxing api jierusiweiliankejiande 2026 0

Bối Cảnh Thị Trường AI 2026: So Sánh Chi Phí

Kiến Trúc Tổng Quan Hệ Thống Đa Agent

Xây Dựng Orchestrator Agent - Trung Tâm Điều Phối

Worker Agent với Retry Logic và Error Recovery

Shared State Manager - Giao Tiếp Giữa Các Agent

Tài nguyên liên quan

Bài viết liên quan

🔥 Thử HolySheep AI