4 Mar 2026

Pola Multi-Agen: Orkestrator, Pekerja, dan Pipeline

Agen tunggal bagus untuk tugas dengan ruang lingkup yang jelas. Saat tugas memerlukan pengetahuan spesialis di berbagai domain, pekerjaan paralel atas input besar, atau keputusan yang harus divalidasi sebelum dieksekusi, Anda memerlukan beberapa agen.

Sistem multi-agen tidak secara inheren lebih kompleks — mereka hanya terstruktur berbeda. Kuncinya adalah memilih pola yang tepat untuk masalah. Tiga pola mencakup sebagian besar kasus penggunaan: orkestrator-pekerja, pipeline, dan fan-out paralel.

Mengapa Beberapa Agen

Argumen untuk sistem multi-agen bermuara pada tiga alasan praktis.

Spesialisasi. Satu agen dengan 50 alat dalam konteks menjadi bingung. Agen spesialis dengan 5 alat terfokus berkinerja lebih baik. Pisahkan berdasarkan domain — penelitian, penulisan, kode, verifikasi — dan setiap agen melakukan satu hal dengan baik.

Paralelisme. Beberapa tugas terurai menjadi subtugas independen. Menganalisis 20 dokumen secara berurutan itu lambat; menganalisisnya secara bersamaan dengan agen paralel itu cepat.

Verifikasi. Memiliki satu agen menghasilkan output dan agen kedua secara independen mengkritiknya menangkap kesalahan yang lewat dari tinjauan diri. Pengulas tidak memiliki kepentingan dalam mempertahankan jawaban asli.

Pola 1: Orkestrator-Pekerja

Satu agen orkestrator merencanakan dan mendelegasikan. Agen pekerja mengeksekusi tugas spesifik dan mengembalikan hasil. Orkestrator merakit output akhir.

Ini adalah pola yang paling fleksibel. Orkestrator dapat mengadaptasi rencananya berdasarkan hasil antara, mencoba ulang tugas yang gagal, atau mengeskalasinya ke pekerja yang berbeda.

import anthropic
import json

client = anthropic.Anthropic()

def run_worker(system_prompt: str, task: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system=system_prompt,
        messages=[{"role": "user", "content": task}]
    )
    return response.content[0].text

def orchestrator(user_request: str) -> str:
    # Step 1: plan the work
    plan_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="""You are a planning agent. Given a user request, break it into
2-4 specific subtasks. Return a JSON array of task descriptions only.
Example: ["Research X", "Analyze Y", "Synthesize findings"]""",
        messages=[{"role": "user", "content": user_request}]
    )

    tasks = json.loads(plan_response.content[0].text)

    # Step 2: run each task with a specialized worker
    results = []
    for task in tasks:
        result = run_worker(
            system_prompt="You are a focused execution agent. Complete the assigned task thoroughly.",
            task=task
        )
        results.append({"task": task, "result": result})

    # Step 3: synthesize
    synthesis_prompt = f"""Original request: {user_request}

Worker results:
{json.dumps(results, indent=2)}

Synthesize these results into a cohesive final response."""

    final = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": synthesis_prompt}]
    )
    return final.content[0].text

Pola orkestrator bekerja paling baik ketika struktur tugas tidak diketahui sebelumnya. Jika Anda tidak tahu berapa banyak subtugas yang Anda butuhkan sampai Anda menganalisis masalahnya, gunakan orkestrator.

Satu jebakan: orkestrator dapat menghaluskan subtugas yang tidak masuk akal. Batasi format output (array JSON, daftar bernomor) dan validasinya sebelum menjalankan pekerja. Blok try/except di sekitar parse JSON dengan langkah perencanaan ulang fallback menangani ini dengan baik.

Pola 2: Pipeline

Agen membentuk rantai sekuensial. Setiap agen mengubah input dan meneruskan outputnya ke yang berikutnya. Tidak ada agen yang mengetahui tentang yang lain — mereka menerima input dan menghasilkan output.

Ini adalah pola yang paling sederhana untuk diimplementasikan dan dipahami. Ini bekerja dengan baik untuk tugas transformasi dengan tahap yang terdefinisi dengan baik.

def run_pipeline(input_text: str) -> str:
    stages = [
        {
            "name": "Researcher",
            "system": "Extract and organize all key facts from the input. "
                      "Format as a structured list with sources noted where available.",
        },
        {
            "name": "Writer",
            "system": "Transform the research notes into clear, readable prose. "
                      "Maintain all factual content. Target a technical audience.",
        },
        {
            "name": "Editor",
            "system": "Improve clarity and concision. Remove redundancy. "
                      "Do not change facts. Return only the improved text.",
        },
        {
            "name": "Fact Checker",
            "system": "Review for internal consistency. Flag any claims that "
                      "contradict each other or seem unsupported. "
                      "If no issues, return 'VERIFIED: ' followed by the original text.",
        },
    ]

    current = input_text
    for stage in stages:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=stage["system"],
            messages=[{"role": "user", "content": current}]
        )
        current = response.content[0].text
        print(f"[{stage['name']}] complete ({len(current)} chars)")

    return current

Pipeline mengakumulasi kesalahan. Jika peneliti melewatkan sesuatu, penulis tidak bisa menambahkannya kembali. Rancang tahap Anda agar bersifat aditif daripada merusak — hindari tahap yang menghapus informasi yang mungkin dibutuhkan tahap berikutnya.

Penyesuaian praktis: lewatkan input asli bersamaan dengan output setiap tahap ketika agen hilir memerlukan konteks yang mungkin telah dikompresi oleh tahap sebelumnya.

Pola 3: Fan-Out Paralel

Pisahkan input besar menjadi potongan independen, proses setiap potongan secara bersamaan dengan agen terpisah, kemudian agregasi hasilnya.

Ini adalah pola yang tepat ketika Anda memproses lebih banyak data daripada yang nyaman dalam satu jendela konteks, atau ketika waktu pemrosesan penting.

import asyncio

async def analyze_document(doc: str, index: int) -> dict:
    """Analyze a single document asynchronously."""
    system = """Analyze this document and return a JSON object with:
- "sentiment": positive/negative/neutral
- "key_topics": list of 3-5 main topics
- "summary": 2-3 sentence summary
- "flags": list of any concerns (empty list if none)"""

    # asyncio.to_thread lets you call synchronous code in a thread pool
    result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            system=system,
            messages=[{"role": "user", "content": doc}]
        ).content[0].text
    )
    return {"index": index, **json.loads(result)}

async def parallel_analysis(documents: list[str]) -> dict:
    # Fan out: analyze all documents concurrently
    tasks = [analyze_document(doc, i) for i, doc in enumerate(documents)]
    analyses = await asyncio.gather(*tasks)

    # Aggregate with a dedicated synthesis agent
    synthesis_input = json.dumps({
        "document_count": len(documents),
        "analyses": analyses
    })

    aggregate_result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system="Synthesize document analyses into: overall sentiment distribution, "
                   "top themes across all documents, and notable patterns in flags. "
                   "Return as JSON.",
            messages=[{"role": "user", "content": synthesis_input}]
        ).content[0].text
    )

    return {
        "individual": analyses,
        "aggregate": json.loads(aggregate_result)
    }

Langkah agregasi adalah di mana sebagian besar implementasi mengambil jalan pintas. Jangan gabungkan hasil — lewatkan ke agen yang memahami tugas agregasi. Penggabungan string dari 20 analisis tidak berguna; ringkasan yang disintesis berguna.

Perhatikan pilihan model: gunakan claude-haiku-4-5-20251001 untuk analisis per dokumen volume tinggi di mana kecepatan penting dan tugasnya sederhana, dan claude-sonnet-4-6 untuk sintesis di mana penilaian lebih penting dari throughput.

Memilih Pola yang Tepat

Situasi	Pola
Struktur tugas tidak diketahui sampai dianalisis	Orkestrator-pekerja
Tahap transformasi yang terdefinisi dengan baik	Pipeline
Input besar, potongan independen	Fan-out paralel
Perlu verifikasi independen	Orkestrator atau pipeline dengan tahap review
Minimalkan latensi pada input besar	Fan-out paralel

Pola-pola ini dapat dikomposisi. Sistem nyata mungkin menggunakan orkestrator yang mem-fan-out beberapa tugas secara paralel sambil menjalankan yang lain melalui pipeline. Mulai dengan pola tersederhana yang cocok dan tambahkan kompleksitas hanya ketika pendekatan yang lebih sederhana gagal.

Pertimbangan Praktis

Biaya. Sistem multi-agen mengalikan panggilan API. Pipeline 4 tahap mungkin menelan biaya 4× panggilan tunggal ditambah overhead sintesis. Padukan model secara strategis: gunakan Opus untuk orkestrasi dan perencanaan di mana penilaian penting, Haiku untuk tugas eksekusi volume tinggi.

Propagasi kesalahan. Putuskan terlebih dahulu bagaimana setiap agen menangani kegagalan. Opsi: propagasikan kesalahan (berhenti), kembalikan objek kesalahan (biarkan orkestrator memutuskan), atau coba ulang dengan prompt yang dimodifikasi (memulihkan dengan baik, menambah latensi). Untuk sebagian besar sistem produksi, mengembalikan objek kesalahan terstruktur dan membiarkan orkestrator memutuskan adalah default yang tepat.

Tracing. Sistem multi-agen di mana Anda tidak bisa melihat apa yang dilakukan setiap agen adalah mimpi buruk debugging. Log setiap panggilan agen dengan: input, output, model, latensi, dan jumlah token. Beri tag setiap panggilan dengan trace ID sehingga Anda dapat merekonstruksi jalur eksekusi lengkap.

Penerusan konteks. Berhati-hatilah dengan konteks yang diterima setiap agen. Meneruskan riwayat percakapan lengkap ke setiap agen itu mahal dan sering membingungkan — agen terganggu oleh konteks sebelumnya yang tidak relevan. Lewatkan hanya apa yang dibutuhkan setiap agen untuk melakukan pekerjaan spesifiknya.

Apa yang Harus Dibangun Selanjutnya

Pola di sini adalah fondasi. Apa yang Anda bangun di atasnya bergantung pada masalah Anda:

Tambahkan penggunaan alat ke pekerja — biarkan agen spesialis memanggil API, mengkueri basis data, atau menjalankan kode
Tambahkan titik pemeriksaan human-in-the-loop di mana orkestrator berhenti sejenak sebelum tindakan berisiko tinggi
Tambahkan memori dengan mempertahankan output agen ke vector store yang dapat dikueri agen masa depan
Tambahkan evaluasi dengan merutekan output melalui agen juri sebelum mengembalikannya ke pengguna

Sistem multi-agen adalah tempat terjadinya rekayasa AI yang menarik saat ini. Polanya sederhana; penilaiannya ada dalam menerapkannya dengan benar pada masalah spesifik Anda.