{"product_id":"orchestrator-workers-kiến-truc-diều-phối-agent-phức-tạp","title":"Orchestrator-Workers — Kiến trúc điều phối agent phức tạp","description":"\n\u003cp\u003eKhi task quá phức tạp cho một LLM đơn lẻ, giải pháp là \u003cstrong\u003echia để trị\u003c\/strong\u003e. Orchestrator-Workers pattern tổ chức hệ thống như một công ty: một manager (Orchestrator) nhận yêu cầu lớn, phân tích, rồi giao cho các chuyên gia (Workers) từng phần việc phù hợp với chuyên môn của họ.\u003c\/p\u003e\n\n\u003cp\u003eĐây là pattern được dùng trong các hệ thống AI production phức tạp nhất — từ research assistants đến automated software development pipelines.\u003c\/p\u003e\n\n\u003ch2\u003eKhi nào cần Orchestrator-Workers?\u003c\/h2\u003e\n\n\u003cp\u003ePattern này tỏa sáng khi:\u003c\/p\u003e\n\u003cul\u003e\n  \u003cli\u003eTask không thể biết trước cần bao nhiêu bước (dynamic decomposition)\u003c\/li\u003e\n  \u003cli\u003eCác sub-tasks đòi hỏi chuyên môn khác nhau (research vs coding vs writing)\u003c\/li\u003e\n  \u003cli\u003eContext window của một model không đủ chứa toàn bộ thông tin\u003c\/li\u003e\n  \u003cli\u003eMuốn parallelize các sub-tasks độc lập\u003c\/li\u003e\n  \u003cli\u003eCần audit trail chi tiết về quá trình xử lý\u003c\/li\u003e\n\u003c\/ul\u003e\n\n\u003ch2\u003eKiến trúc tổng quan\u003c\/h2\u003e\n\n\u003cpre\u003e\u003ccode\u003eimport anthropic\nimport json\nimport asyncio\nfrom typing import Optional\nfrom dataclasses import dataclass, field\n\nclient = anthropic.Anthropic()\nasync_client = anthropic.AsyncAnthropic()\n\n@dataclass\nclass Task:\n    id: str\n    description: str\n    worker_type: str\n    depends_on: list = field(default_factory=list)\n    context: dict = field(default_factory=dict)\n    status: str = \"pending\"  # pending\/running\/completed\/failed\n    result: Optional[str] = None\n\nclass OrchestratorWorkersSystem:\n    def __init__(self):\n        self.workers = {\n            \"researcher\": ResearchWorker(),\n            \"analyst\": AnalystWorker(),\n            \"writer\": WriterWorker(),\n            \"coder\": CoderWorker(),\n            \"reviewer\": ReviewerWorker()\n        }\n\n    def run(self, complex_task: str) -\u0026gt; dict:\n        print(f\"Orchestrating: {complex_task[:80]}...\")\n\n        # Phase 1: Orchestrator phân tích và tạo execution plan\n        plan = self._create_plan(complex_task)\n        print(f\"Plan created: {len(plan)} tasks\")\n\n        # Phase 2: Execute tasks theo dependency order\n        results = self._execute_plan(plan)\n\n        # Phase 3: Orchestrator tổng hợp\n        final = self._synthesize(complex_task, plan, results)\n\n        return {\n            \"task\": complex_task,\n            \"plan\": [{\"id\": t.id, \"worker\": t.worker_type, \"desc\": t.description} for t in plan],\n            \"results\": results,\n            \"final_output\": final\n        }\n\n    def _create_plan(self, task: str) -\u0026gt; list[Task]:\n        available_workers = list(self.workers.keys())\n\n        response = client.messages.create(\n            model=\"claude-opus-4-5\",\n            max_tokens=3000,\n            system=\"\"\"You are a project orchestrator. Create execution plans for complex tasks.\nBreak down tasks into atomic subtasks assignable to specialist workers.\nThink carefully about dependencies — which tasks must complete before others can start.\"\"\",\n            messages=[{\n                \"role\": \"user\",\n                \"content\": f\"\"\"Create an execution plan for this task.\n\nTask: {task}\n\nAvailable workers: {available_workers}\nWorker capabilities:\n- researcher: find facts, gather information, search knowledge\n- analyst: analyze data, identify patterns, make comparisons\n- writer: create content, reports, summaries, documentation\n- coder: write code, scripts, technical implementations\n- reviewer: quality check, fact-check, improve outputs\n\nReturn a JSON array of tasks:\n[\n  {{\n    \"id\": \"T1\",\n    \"description\": \"specific instruction for this worker\",\n    \"worker_type\": \"researcher|analyst|writer|coder|reviewer\",\n    \"depends_on\": []\n  }},\n  ...\n]\n\nImportant: depends_on should list task IDs that must complete first.\"\"\"\n            }]\n        )\n\n        try:\n            text = response.content[0].text\n            start = text.find('[')\n            end = text.rfind(']') + 1\n            tasks_data = json.loads(text[start:end])\n\n            return [\n                Task(\n                    id=t[\"id\"],\n                    description=t[\"description\"],\n                    worker_type=t[\"worker_type\"],\n                    depends_on=t.get(\"depends_on\", [])\n                )\n                for t in tasks_data\n            ]\n        except Exception as e:\n            print(f\"Plan parsing failed: {e}\")\n            # Fallback: single task\n            return [Task(\"T1\", task, \"writer\", [])]\n\n    def _execute_plan(self, tasks: list[Task]) -\u0026gt; dict:\n        results = {}\n        completed_ids = set()\n\n        # Topological execution\n        max_rounds = len(tasks) * 2\n        round_num = 0\n\n        while len(completed_ids) \u0026lt; len(tasks) and round_num \u0026lt; max_rounds:\n            round_num += 1\n            progress_made = False\n\n            for task in tasks:\n                if task.id in completed_ids:\n                    continue\n\n                # Check dependencies\n                if all(dep in completed_ids for dep in task.depends_on):\n                    # Inject dependency results as context\n                    task.context = {\n                        dep_id: results[dep_id]\n                        for dep_id in task.depends_on\n                        if dep_id in results\n                    }\n\n                    print(f\"  Running {task.id} ({task.worker_type}): {task.description[:50]}...\")\n                    worker = self.workers.get(task.worker_type, self.workers[\"writer\"])\n                    task.result = worker.execute(task)\n                    results[task.id] = task.result\n                    completed_ids.add(task.id)\n                    task.status = \"completed\"\n                    progress_made = True\n\n            if not progress_made:\n                print(\"Warning: Dependency deadlock detected, breaking remaining tasks\")\n                break\n\n        return results\n\n    def _synthesize(self, original_task: str, plan: list[Task], results: dict) -\u0026gt; str:\n        results_text = \"\n\n\".join([\n            f\"=== {t.id} ({t.worker_type}): {t.description[:60]} ===\n{results.get(t.id, 'No result')}\"\n            for t in plan\n        ])\n\n        response = client.messages.create(\n            model=\"claude-opus-4-5\",\n            max_tokens=4000,\n            system=\"You are an expert synthesizer. Combine worker outputs into a coherent, high-quality final deliverable.\",\n            messages=[{\n                \"role\": \"user\",\n                \"content\": f\"\"\"Original task: {original_task}\n\nWorker outputs:\n{results_text}\n\nSynthesize all worker outputs into a comprehensive final response.\nThe final output should be self-contained and directly address the original task.\"\"\"\n            }]\n        )\n        return response.content[0].text\u003c\/code\u003e\u003c\/pre\u003e\n\n\u003ch2\u003eWorker Implementations\u003c\/h2\u003e\n\n\u003cpre\u003e\u003ccode\u003eclass BaseWorker:\n    \"\"\"Base class cho tất cả workers\"\"\"\n\n    def execute(self, task: Task) -\u0026gt; str:\n        context_text = \"\"\n        if task.context:\n            context_parts = [f\"--- {k}: {v[:300]}...\" for k, v in task.context.items()]\n            context_text = \"\nContext from previous tasks:\n\" + \"\n\".join(context_parts)\n\n        prompt = f\"{task.description}{context_text}\"\n        response = client.messages.create(\n            model=self.model,\n            max_tokens=self.max_tokens,\n            system=self.system_prompt,\n            messages=[{\"role\": \"user\", \"content\": prompt}]\n        )\n        return response.content[0].text\n\nclass ResearchWorker(BaseWorker):\n    model = \"claude-haiku-4-5\"\n    max_tokens = 2000\n    system_prompt = \"\"\"You are a meticulous researcher. Gather facts, cite sources when possible,\nidentify gaps in information, and flag uncertainty. Be thorough but concise.\"\"\"\n\nclass AnalystWorker(BaseWorker):\n    model = \"claude-haiku-4-5\"\n    max_tokens = 2000\n    system_prompt = \"\"\"You are a data analyst. Identify patterns, make comparisons,\nprovide quantitative insights where possible. Structure analysis clearly.\"\"\"\n\nclass WriterWorker(BaseWorker):\n    model = \"claude-haiku-4-5\"\n    max_tokens = 3000\n    system_prompt = \"\"\"You are a professional writer. Create clear, engaging, well-structured content.\nAdapt tone and style to context. Use formatting (headers, bullets) appropriately.\"\"\"\n\nclass CoderWorker(BaseWorker):\n    model = \"claude-opus-4-5\"  # Coder dùng model mạnh hơn\n    max_tokens = 4000\n    system_prompt = \"\"\"You are a senior software engineer. Write clean, production-ready code\nwith proper error handling, comments, and following best practices.\"\"\"\n\nclass ReviewerWorker(BaseWorker):\n    model = \"claude-haiku-4-5\"\n    max_tokens = 1000\n    system_prompt = \"\"\"You are a quality reviewer. Check for accuracy, completeness, consistency,\nand clarity. Provide specific, actionable improvement suggestions.\"\"\"\u003c\/code\u003e\u003c\/pre\u003e\n\n\u003ch2\u003eVí dụ thực tế: Research Report Generator\u003c\/h2\u003e\n\n\u003cpre\u003e\u003ccode\u003esystem = OrchestratorWorkersSystem()\n\n# Task phức tạp đòi hỏi nhiều loại chuyên môn\nresult = system.run(\"\"\"\nCreate a comprehensive market analysis report on:\n\"AI adoption in Vietnamese SMEs (small-medium enterprises) in 2024-2025\"\n\nThe report should include:\n1. Current state of AI adoption\n2. Key barriers and opportunities\n3. Success case studies\n4. Recommendations for SME owners\n5. Technology roadmap for the next 2 years\n\"\"\")\n\nprint(\"=\" * 60)\nprint(\"EXECUTION PLAN:\")\nfor t in result[\"plan\"]:\n    print(f\"  {t['id']} [{t['worker']}]: {t['desc'][:60]}\")\n\nprint(\"\n\" + \"=\" * 60)\nprint(\"FINAL REPORT:\")\nprint(result[\"final_output\"])\u003c\/code\u003e\u003c\/pre\u003e\n\n\u003cp\u003eOutput plan điển hình trông như sau:\u003c\/p\u003e\n\n\u003cpre\u003e\u003ccode\u003eEXECUTION PLAN:\n  T1 [researcher]: Research current AI adoption rates in Vietnamese SMEs\n  T2 [researcher]: Find specific AI adoption case studies from Vietnamese companies\n  T3 [analyst]: Analyze barriers to AI adoption based on T1 findings\n  T4 [analyst]: Identify top opportunities from T1 and T2 context\n  T5 [writer]: Write executive summary using T1, T3, T4\n  T6 [coder]: Create data visualization code for T1 statistics\n  T7 [writer]: Write full report combining all sections T1-T6\n  T8 [reviewer]: Review and improve final report T7\u003c\/code\u003e\u003c\/pre\u003e\n\n\u003ch2\u003eAsync Orchestration cho hiệu suất cao\u003c\/h2\u003e\n\n\u003cpre\u003e\u003ccode\u003easync def execute_parallel_tasks(tasks: list[Task], results: dict) -\u0026gt; dict:\n    \"\"\"Chạy các tasks không có dependencies song song\"\"\"\n\n    # Group tasks không phụ thuộc nhau\n    completed = set(results.keys())\n    executable = [\n        t for t in tasks\n        if t.id not in completed\n        and all(dep in completed for dep in t.depends_on)\n    ]\n\n    if not executable:\n        return results\n\n    print(f\"Running {len(executable)} tasks in parallel...\")\n\n    async def run_async_task(task: Task) -\u0026gt; tuple:\n        task.context = {dep: results[dep] for dep in task.depends_on if dep in results}\n        context_text = \"\n\".join([f\"{k}: {v[:200]}\" for k, v in task.context.items()])\n        prompt = f\"{task.description}\n{context_text}\" if context_text else task.description\n\n        response = await async_client.messages.create(\n            model=\"claude-haiku-4-5\",\n            max_tokens=2000,\n            messages=[{\"role\": \"user\", \"content\": prompt}]\n        )\n        return task.id, response.content[0].text\n\n    new_results = await asyncio.gather(*[run_async_task(t) for t in executable])\n\n    for task_id, result in new_results:\n        results[task_id] = result\n\n    return results\u003c\/code\u003e\u003c\/pre\u003e\n\n\u003ch2\u003eError Handling và Resilience\u003c\/h2\u003e\n\n\u003cpre\u003e\u003ccode\u003edef execute_with_retry(worker: BaseWorker, task: Task, max_retries: int = 2) -\u0026gt; str:\n    \"\"\"Worker execution với retry logic\"\"\"\n    last_error = None\n\n    for attempt in range(max_retries + 1):\n        try:\n            return worker.execute(task)\n        except anthropic.RateLimitError:\n            import time\n            wait_time = (2 ** attempt) * 5  # Exponential backoff\n            print(f\"Rate limit hit, waiting {wait_time}s...\")\n            time.sleep(wait_time)\n        except anthropic.APIError as e:\n            last_error = e\n            print(f\"API error on attempt {attempt + 1}: {e}\")\n\n    # Fallback: simplified version of task\n    print(f\"All retries failed for {task.id}, using fallback\")\n    response = client.messages.create(\n        model=\"claude-haiku-4-5\",\n        max_tokens=500,\n        messages=[{\"role\": \"user\", \"content\": f\"Briefly: {task.description}\"}]\n    )\n    return f\"[Fallback result] {response.content[0].text}\"\u003c\/code\u003e\u003c\/pre\u003e\n\n\u003ch2\u003eSo sánh với các Patterns khác\u003c\/h2\u003e\n\n\u003ctable\u003e\n  \u003cthead\u003e\n    \u003ctr\u003e\n\u003cth\u003ePattern\u003c\/th\u003e\n\u003cth\u003eTask Structure\u003c\/th\u003e\n\u003cth\u003eComplexity\u003c\/th\u003e\n\u003cth\u003eCost\u003c\/th\u003e\n\u003c\/tr\u003e\n  \u003c\/thead\u003e\n  \u003ctbody\u003e\n    \u003ctr\u003e\n\u003ctd\u003eSimple LLM call\u003c\/td\u003e\n\u003ctd\u003eSingle task\u003c\/td\u003e\n\u003ctd\u003eThấp\u003c\/td\u003e\n\u003ctd\u003eThấp nhất\u003c\/td\u003e\n\u003c\/tr\u003e\n    \u003ctr\u003e\n\u003ctd\u003ePrompt Chaining\u003c\/td\u003e\n\u003ctd\u003eLinear steps\u003c\/td\u003e\n\u003ctd\u003eThấp\u003c\/td\u003e\n\u003ctd\u003eThấp\u003c\/td\u003e\n\u003c\/tr\u003e\n    \u003ctr\u003e\n\u003ctd\u003eRouting\u003c\/td\u003e\n\u003ctd\u003eBranching\u003c\/td\u003e\n\u003ctd\u003eTrung bình\u003c\/td\u003e\n\u003ctd\u003eThấp\u003c\/td\u003e\n\u003c\/tr\u003e\n    \u003ctr\u003e\n\u003ctd\u003eEvaluator-Optimizer\u003c\/td\u003e\n\u003ctd\u003eIterative\u003c\/td\u003e\n\u003ctd\u003eTrung bình\u003c\/td\u003e\n\u003ctd\u003eTrung bình\u003c\/td\u003e\n\u003c\/tr\u003e\n    \u003ctr\u003e\n\u003ctd\u003e\u003cstrong\u003eOrchestrator-Workers\u003c\/strong\u003e\u003c\/td\u003e\n\u003ctd\u003eDynamic DAG\u003c\/td\u003e\n\u003ctd\u003eCao\u003c\/td\u003e\n\u003ctd\u003eCao\u003c\/td\u003e\n\u003c\/tr\u003e\n  \u003c\/tbody\u003e\n\u003c\/table\u003e\n\n\u003ch2\u003eTổng kết\u003c\/h2\u003e\n\n\u003cp\u003eOrchestrator-Workers là pattern phức tạp nhất nhưng mạnh nhất trong toolkit agentic AI. Khi implement:\u003c\/p\u003e\n\n\u003cul\u003e\n  \u003cli\u003eĐầu tư vào Orchestrator prompt — plan quality quyết định mọi thứ\u003c\/li\u003e\n  \u003cli\u003eMỗi Worker cần system prompt chuyên biệt rõ ràng\u003c\/li\u003e\n  \u003cli\u003eXử lý dependencies cẩn thận để tránh deadlocks\u003c\/li\u003e\n  \u003cli\u003eLog toàn bộ plan và kết quả để debug\u003c\/li\u003e\n  \u003cli\u003eDùng model nhỏ hơn cho simple workers để tiết kiệm chi phí\u003c\/li\u003e\n  \u003cli\u003eImplement retry và fallback cho production reliability\u003c\/li\u003e\n\u003c\/ul\u003e\n\n\u003cp\u003eTiếp theo: Kết hợp Orchestrator-Workers với \u003ca href=\"\/collections\/nang-cao\"\u003eExtended Thinking\u003c\/a\u003e để Orchestrator lập kế hoạch tốt hơn, và \u003ca href=\"\/collections\/nang-cao\"\u003eEvaluator-Optimizer\u003c\/a\u003e để mỗi Worker tự cải thiện output.\u003c\/p\u003e\n","brand":"Minh Tuấn","offers":[{"title":"Default Title","offer_id":66959028518957,"sku":null,"price":0.0,"currency_code":"VND","in_stock":true}],"thumbnail_url":"\/\/cdn.shopify.com\/s\/files\/1\/0763\/9531\/5245\/files\/orchestrator-workers-ki_n-truc-di_u-ph_i-agent-ph_c-t_p_bd5b8fe1-f709-4be0-9f20-2076de071205.jpg?v=1782891860","url":"https:\/\/claudeae.com\/products\/orchestrator-workers-ki%e1%ba%bfn-truc-di%e1%bb%81u-ph%e1%bb%91i-agent-ph%e1%bb%a9c-t%e1%ba%a1p","provider":"Claude.vn","version":"1.0","type":"link"}