Tip: Say "fan out subagents" and Claude sends a team. Each one digs deep so nothing gets missed.

Membangun sistem berbasis AI untuk menangani tugas berskala besar sering kali menghadapi kendala batas konteks (context window limits) dan waktu pemrosesan yang lambat jika dijalankan secara sekuensial.

Pola arsitektur Fan Out / Fan In Subagents mengatasi masalah ini dengan memecah satu masalah besar ke beberapa AI sub-agen spesialis secara paralel, lalu menyatukan kembali hasilnya melalui satu orkestrator terpusat.

---

Konsep Dasar: Fan Out vs. Fan In

• Fan Out (Distribusi Paralel): Orkestrator membagi tugas menjadi sub-masalah spesifik dan mengirimkannya ke beberapa sub-agen secara bersamaan.
• Eksekusi Mandiri: Setiap sub-agen fokus hanya pada domain tugasnya tanpa terbebani konteks dari domain lain.
• Fan In (Sintesis Hasil): Orkestrator mengumpulkan semua output, menyaring redundansi, dan menyusun laporan akhir yang terpadu.

---

Implementasi dengan Python & Claude API

Berikut adalah contoh implementasi pola Fan Out untuk melakukan Code Review Otomatis menggunakan 3 sub-agen: Keamanan, Performa, dan Kualitas Kode (Clean Code).

import asyncio
import os
from anthropic import AsyncAnthropic

# Inisialisasi klien Anthropic Asinkron
client = AsyncAnthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

# 1. Definisi Sub-agen Individual
async def run_subagent(role: str, focus_area: str, code_snippet: str) -> dict:
    prompt = f"""
    Anda adalah sub-agen spesialis: {role}.
    Tugas: Analisis mendalam hanya pada fokus: {focus_area}.
    Jangan bahas hal di luar domain Anda.
    
    Kode untuk dianalisis:
    ```python
    {code_snippet}
    ```
    
    Berikan temuan spesifik dan rekomendasi singkat.
    """
    
    response = await client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        temperature=0.2,
        messages=[{"role": "user", "content": prompt}]
    )
    
    return {
        "role": role,
        "findings": response.content[0].text
    }

# 2. Orchestrator: Fan Out & Fan In
async def orchestrate_audit(code_snippet: str):
    subagent_tasks = [
        {"role": "Security Auditor", "focus": "Celah keamanan, SQL Injection, sanitasi input"},
        {"role": "Performance Engineer", "focus": "Kompleksitas memori/waktu, resource leak, bottleneck"},
        {"role": "Clean Code Reviewer", "focus": "Readability, standar PEP8, error handling"}
    ]
    
    # Fan Out: Jalankan semua sub-agen secara paralel
    tasks = [
        run_subagent(agent["role"], agent["focus"], code_snippet)
        for agent in subagent_tasks
    ]
    subagent_results = await asyncio.gather(*tasks)
    
    # Agregasi hasil temuan
    aggregated_context = "\n\n".join([
        f"--- Temuan {res['role']} ---\n{res['findings']}"
        for res in subagent_results
    ])
    
    # Fan In: Model utama mensintesis temuan akhir
    synthesis_prompt = f"""
    Berikut adalah temuan dari tim sub-agen spesialis mengenai sebuah kode:

    {aggregated_context}

    Tugas Anda sebagai Lead Architect:
    1. Gabungkan temuan di atas ke dalam satu laporan komprehensif.
    2. Hilangkan redundansi jika ada overlap antar sub-agen.
    3. Buat action items prioritas (Tinggi, Sedang, Rendah).
    """
    
    final_report = await client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1500,
        temperature=0.3,
        messages=[{"role": "user", "content": synthesis_prompt}]
    )
    
    return final_report.content[0].text

# Contoh Eksekusi
if __name__ == "__main__":
    sample_code = """
    import sqlite3

    def get_user_data(user_id):
        conn = sqlite3.connect('users.db')
        cursor = conn.cursor()
        query = f"SELECT * FROM users WHERE id = '{user_id}'"
        cursor.execute(query)
        result = cursor.fetchall()
        return result
    """
    
    report = asyncio.run(orchestrate_audit(sample_code))
    print(report)

---

Keuntungan Menggunakan Pola Ini

• Latensi Jauh Lebih Rendah: Pemrosesan 3–5 tugas berlangsung dalam waktu yang hampir sama dengan 1 tugas karena dijalankan secara paralel via asyncio.gather.
• Kualitas Analisis Lebih Tajam: Sub-agen terhindar dari bias konteks berlebih (lost-in-the-middle), menghasilkan evaluasi yang fokus dan mendalam.
• Modular & Fleksibel: Penambahan peran baru (misal: Documentation Reviewer) cukup dilakukan pada konfigurasi tanpa mengubah alur logika utama.

Berbagi :