Sub-Agent Pattern — Dùng Haiku phân tích, Opus tổng hợp

Bạn cần phân tích một tài liệu 100 trang — mỗi trang là một ảnh scan. Nếu dùng Claude Opus cho tất cả, chi phí sẽ rất cao. Nếu dùng Haiku cho tất cả, kết quả tổng hợp sẽ thiếu chiều sâu.

Sub-Agent Pattern giải quyết điều này: dùng model nhẹ (Haiku) làm "công nhân" xử lý từng đơn vị nhỏ, sau đó model mạnh (Sonnet/Opus) đóng vai "giám đốc" tổng hợp kết quả thành báo cáo hoàn chỉnh.

Tại sao cần phân chia model?

So sánh chi phí xử lý 100 ảnh (mỗi ảnh ~1000 tokens input, 200 tokens output):

Pattern này tiết kiệm 85-90% chi phí so với dùng toàn Opus, trong khi chất lượng tổng hợp cuối vẫn đạt mức cao nhờ Sonnet/Opus có đủ thông tin chi tiết từ Haiku workers.

Kiến trúc Sub-Agent Pattern

# Luong xu ly:
#
#  [Tai lieu goc (100 trang)]
#          |
#          v
#  [Split thanh cac don vi nho]  -- Client code
#          |
#     _____|_____
#    |     |     |
#    v     v     v
# [Haiku] [Haiku] [Haiku]  -- Sub-agents (workers)
# Trang 1  Trang 2  ...
#    |     |     |
#         |    /
#      v   v   v
#  [Ket qua trung gian]   -- Structured summaries
#          |
#          v
#  [Sonnet/Opus]           -- Orchestrator (synthesis)
#          |
#          v
#  [Bao cao cuoi cung]

Implementation cơ bản

Bước 1: Worker function (Haiku)

import anthropic
import base64
import json
import time
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed

def haiku_analyze_page(
    image_path: str,
    page_number: int,
    task_description: str
) -> dict:
    """
    Worker: Haiku phan tich mot trang/anh.

    Args:
        image_path: Duong dan den anh
        page_number: So trang (de tham chieu)
        task_description: Mo ta nhiem vu (gi can trich xuat?)

    Returns:
        Dict chua ket qua phan tich
    """
    with open(image_path, "rb") as f:
        data = base64.standard_b64encode(f.read()).decode()

    ext = Path(image_path).suffix.lower()
    media_type = {
        ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
        ".png": "image/png", ".webp": "image/webp"
    }.get(ext, "image/jpeg")

    client = anthropic.Anthropic()
    message = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {"type": "base64", "media_type": media_type, "data": data}
                    },
                    {
                        "type": "text",
                        "text": f"""Trang {page_number}. Nhiem vu: {task_description}

Tra ve JSON voi cau truc nay (chi JSON, khong them gi):
{{
  "page": {page_number},
  "has_content": true_or_false,
  "content_type": "text/table/chart/image/mixed/blank",
  "key_points": ["diem chinh 1", "diem chinh 2"],
  "extracted_data": "du lieu quan trong duoc trich xuat hoac null",
  "summary": "tom tat 1-3 cau ve noi dung trang"
}}"""
                    }
                ]
            }
        ]
    )

    try:
        result = json.loads(message.content[0].text)
    except json.JSONDecodeError:
        result = {
            "page": page_number,
            "has_content": True,
            "summary": message.content[0].text,
            "parse_error": True
        }

    return result

Bước 2: Orchestrator function (Sonnet/Opus)

def synthesize_results(
    page_analyses: list,
    synthesis_task: str,
    model: str = "claude-sonnet-4-5"
) -> str:
    """
    Orchestrator: Sonnet/Opus tong hop ket qua tu cac workers.

    Args:
        page_analyses: Danh sach ket qua tu Haiku workers
        synthesis_task: Nhiem vu tong hop cu the
        model: Model dung de tong hop

    Returns:
        Bao cao tong hop cuoi cung
    """
    # Loc bo nhung trang trong/khong co noi dung
    relevant_pages = [p for p in page_analyses if p.get("has_content", True)]

    # Format cac ket qua trung gian
    analyses_text = json.dumps(relevant_pages, ensure_ascii=False, indent=2)

    client = anthropic.Anthropic()
    message = client.messages.create(
        model=model,
        max_tokens=4096,
        messages=[
            {
                "role": "user",
                "content": f"""Ban la chuyen gia phan tich tai lieu.
Duoi day la ket qua phan tich tung trang (tong {len(relevant_pages)} trang co noi dung):

{analyses_text}

Nhiem vu tong hop: {synthesis_task}

Hay tao bao cao day du va chuyen nghiep dua tren thong tin tren.
Dam bao:
- Su dung toan bo thong tin quan trong tu tat ca cac trang
- Cau truc ro rang voi heading va subheading
- Chi ra so trang nguon khi trich dan thong tin cu the
- Nhan dien pattern hoac chu de xuyen suot tai lieu"""
            }
        ]
    )
    return message.content[0].text

Bước 3: Pipeline hoàn chỉnh với parallel processing

def process_document_multimodel(
    image_paths: list,
    extraction_task: str,
    synthesis_task: str,
    max_workers: int = 5,
    synthesis_model: str = "claude-sonnet-4-5"
) -> dict:
    """
    Pipeline day du: Haiku workers + Sonnet orchestrator.

    Args:
        image_paths: Danh sach duong dan anh (theo thu tu trang)
        extraction_task: Gi can trich xuat tu moi trang
        synthesis_task: Ket qua cuoi cung can gi
        max_workers: So luong Haiku workers chay song song
        synthesis_model: Model dung de tong hop

    Returns:
        Dict chua page_analyses va final_report
    """
    print(f"Bat dau xu ly {len(image_paths)} trang voi {max_workers} workers...")
    start_time = time.time()

    # Phase 1: Haiku workers xu ly song song
    page_analyses = [None] * len(image_paths)
    failed_pages = []

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_page = {
            executor.submit(
                haiku_analyze_page,
                path,
                i + 1,
                extraction_task
            ): i
            for i, path in enumerate(image_paths)
        }

        completed = 0
        for future in as_completed(future_to_page):
            page_idx = future_to_page[future]
            try:
                result = future.result()
                page_analyses[page_idx] = result
                completed += 1
                print(f"  [{completed}/{len(image_paths)}] Trang {page_idx+1} hoan thanh")
            except Exception as e:
                print(f"  LOI trang {page_idx+1}: {e}")
                failed_pages.append(page_idx + 1)
                page_analyses[page_idx] = {
                    "page": page_idx + 1,
                    "has_content": False,
                    "error": str(e)
                }

    phase1_time = time.time() - start_time
    print(f"
Phase 1 hoan thanh trong {phase1_time:.1f}s")
    if failed_pages:
        print(f"Cac trang loi: {failed_pages}")

    # Phase 2: Sonnet tong hop
    print(f"
Phase 2: {synthesis_model} dang tong hop...")
    final_report = synthesize_results(
        page_analyses,
        synthesis_task,
        model=synthesis_model
    )

    total_time = time.time() - start_time
    print(f"Hoan thanh trong {total_time:.1f}s tong cong")

    return {
        "total_pages": len(image_paths),
        "processed_pages": len(image_paths) - len(failed_pages),
        "failed_pages": failed_pages,
        "page_analyses": page_analyses,
        "final_report": final_report,
        "processing_time_seconds": total_time
    }

Ví dụ thực tế: Phân tích hợp đồng 100 trang

from pathlib import Path

def analyze_legal_contract(pdf_path: str) -> str:
    """Phan tich hop dong phap ly nhieu trang."""

    # Buoc 1: Chuyen PDF sang anh
    import subprocess
    output_dir = "contract_pages"
    Path(output_dir).mkdir(exist_ok=True)
    subprocess.run([
        "pdftoppm", "-r", "150", "-png",
        pdf_path, f"{output_dir}/page"
    ], check=True)

    page_paths = sorted(Path(output_dir).glob("page-*.png"))
    page_paths = [str(p) for p in page_paths]

    # Buoc 2: Chay pipeline
    result = process_document_multimodel(
        image_paths=page_paths,
        extraction_task="""Trich xuat thong tin tu trang hop dong:
        - Cac dieu khoan chinh (neu co)
        - Nghia vu cua cac ben
        - Ngay thang quan trong, deadline
        - So tien hoac gia tri (neu co)
        - Bat ky dieu khoan quan trong hoac rui ro nao""",
        synthesis_task="""Tao tom tat hop dong voi cau truc:
        1. THONG TIN CHUNG (ten hop dong, ben ky, ngay ky, gia tri)
        2. CAC DIEU KHOAN CHINH (danh sach theo thu tu quan trong)
        3. NGHIA VU TUNG BEN (Ben A / Ben B)
        4. NGAY THANG QUAN TRONG (deadline, gia han, v.v.)
        5. DIEU KHOAN RUI RO DANG CHU Y
        6. KHUYEN NGHI (cac diem can luu y hoac dam phan them)""",
        max_workers=8,
        synthesis_model="claude-opus-4-5"  # Dung Opus cho hop dong phap ly
    )

    return result["final_report"]

Ví dụ: Phân tích báo cáo tài chính hàng quý

def analyze_financial_report(pages: list) -> dict:
    """Phan tich bao cao tai chinh."""

    result = process_document_multimodel(
        image_paths=pages,
        extraction_task="""Trich xuat so lieu tai chinh:
        - Doanh thu, loi nhuan, EBITDA (neu co)
        - Bang can doi ke toan (assets, liabilities, equity)
        - Dong tien (operating, investing, financing)
        - KPI nganh cu the
        - So sanh year-over-year hoac quarter-over-quarter
        - Cac ghi chu quan trong cua ban lanh dao""",
        synthesis_task="""Tao bao cao phan tich tai chinh:
        ## Tom tat dieu hanh (Executive Summary)
        ## Ket qua kinh doanh chinh
        ## Phan tich xu huong
        ## Diem manh / Diem yeu
        ## Rui ro duoc neu trong bao cao
        ## So lieu chot (bang)""",
        max_workers=10,
        synthesis_model="claude-sonnet-4-5"
    )

    return result

Tùy chỉnh nâng cao

Adaptive worker selection — Tự động chọn model theo độ phức tạp

def smart_worker(image_path: str, page_num: int, task: str) -> dict:
    """
    Dung Haiku truoc, upgrade len Sonnet neu Haiku khong du.
    """
    # Haiku thu truoc
    result = haiku_analyze_page(image_path, page_num, task)

    # Neu Haiku bao la phuc tap hoac khong chac chan
    needs_upgrade = (
        result.get("confidence", "high") == "low" or
        result.get("complex_content", False) or
        result.get("parse_error", False)
    )

    if needs_upgrade:
        print(f"  Trang {page_num}: upgrade len Sonnet")
        # Retry bang Sonnet
        result = sonnet_analyze_page(image_path, page_num, task)
        result["upgraded"] = True

    return result

Retry logic cho pages bị lỗi

def analyze_with_retry(image_path: str, page_num: int, task: str, max_retries: int = 3) -> dict:
    """Worker voi retry logic."""
    for attempt in range(max_retries):
        try:
            return haiku_analyze_page(image_path, page_num, task)
        except anthropic.RateLimitError:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff: 1s, 2s, 4s
                print(f"  Rate limit, doi {wait_time}s...")
                time.sleep(wait_time)
            else:
                raise
        except Exception as e:
            if attempt == max_retries - 1:
                return {"page": page_num, "error": str(e), "has_content": False}
            time.sleep(1)

Benchmark chi phí thực tế

Dưới đây là chi phí ước tính cho tài liệu 100 trang (mỗi trang ~800 tokens ảnh, Haiku output ~300 tokens, Sonnet input tổng hợp ~30,000 tokens):

So sánh: Nếu dùng 100% Opus (~$15-20), pattern này tiết kiệm hơn 98% trong khi chất lượng synthesis vẫn cao vì Sonnet nhận đủ thông tin chi tiết từ Haiku workers.

Khi nào nên dùng pattern này?

Tổng kết

Sub-Agent Pattern là một trong những kỹ thuật cost-optimization mạnh nhất cho vision workloads:

• Haiku workers — Nhanh, rẻ, chạy song song — xử lý từng đơn vị nhỏ
• Sonnet/Opus orchestrator — Chất lượng cao, tổng hợp toàn bộ thông tin
• Tiết kiệm 80-98% chi phí so với dùng model mạnh nhất cho mọi tác vụ
• Parallel processing — ThreadPoolExecutor giúp xử lý nhanh hơn nhiều

Pattern này áp dụng được không chỉ cho vision mà còn cho bất kỳ task nào có thể chia nhỏ thành nhiều đơn vị độc lập — text summarization, data extraction, content moderation, v.v.

Xem thêm: Crop Tool Pattern để kết hợp với Sub-Agent khi cần phân tích chi tiết từng trang, và Claude API fundamentals để hiểu sâu hơn về pricing và rate limits.