Belajar A2A - Advanced Task Patterns

#Pendahuluan

Di episode 16 kita membuat agent-agent bisa saling menemukan dan memilih lewat registry dan routing. Sekarang task yang mereka tukarkan bisa menjadi lebih kompleks dari sekadar tanya-jawab. Di dunia produksi, satu permintaan user sering menjadi pohon task: dipecah ke banyak agent, menunggu keputusan manusia, lalu disatukan kembali.

Protokol A2A memang dirancang untuk pola ini. Task punya lifecycle lengkap dengan state submitted, working, input-required, completed, failed, dan canceled. Episode ini memanfaatkan lifecycle tersebut untuk membangun pola-pola task lanjutan: fan-out/fan-in, human-in-the-loop, resubmission, idempotency, dan checkpoint-resume.

Roadmap episode ini: kita mulai dari orchestrasi fan-out dan fan-in, lalu human-in-the-loop dengan state input-required, resubmission, idempotency dan retry, dan menutup dengan long-running task beserta checkpoint-resume.

#Fan-Out dan Fan-In

Fan-out berarti satu task dipecah menjadi banyak sub-task yang dikirim ke beberapa agent secara paralel. Fan-in berarti mengumpulkan semua hasil dan menggabungkannya menjadi satu output. Pola ini sangat umum: cek harga dari tiga vendor sekaligus, tanya cuaca di lima kota, atau ringkas dokumen dari banyak sumber.

Implementasi orkestrator dengan fan-out lalu fan-in:

orchestrator.py — fan-out lalu fan-in

import asyncio
 
async def fan_out_and_fan_in(agent_clients, sub_tasks):
    pending = [
        client.send_task(task) for client, task in zip(agent_clients, sub_tasks)
    ]
    results = await asyncio.gather(*pending, return_exceptions=True)
    ok = [r for r in results if not isinstance(r, Exception)]
    merged = merge_results(ok)
    return await aggregator_client.send_task({"parts": [merged]})

Beberapa catatan praktis fan-out/fan-in:

• Jangan menunggu semua sub-task sebelum memulai agregasi; pertimbangkan partial result jika beberapa agent lambat.
• Setiap sub-task adalah task A2A mandiri dengan taskId sendiri — korelasi dengan task induk dilakukan lewat metadata trace seperti di episode 15.
• Jika salah satu sub-task gagal, putuskan dengan bijak: gagalkan semua, atau lanjutkan dengan hasil parsial lalu tandai peringatan.

#Human-in-the-Loop dengan State input-required

Beberapa keputusan tidak boleh ditentukan agent sendirian — transfer uang, approval dokumen hukum, atau aksi yang merusak. A2A menyediakan state input-required untuk pola ini: agent berhenti bekerja dan menunggu input dari manusia atau sistem eksternal.

Alurnya: task masuk, agent bekerja sebentar, menemukan butuh persetujuan, lalu mengembalikan task dalam state input-required dengan detail apa yang dibutuhkan. Client menampilkan pertanyaan ke manusia, lalu mengirimkan message/send baru dengan jawabannya; agent melanjutkan task.

Contoh respons saat agent membutuhkan persetujuan:

{
  "jsonrpc": "2.0",
  "id": "7",
  "result": {
    "id": "task-4821",
    "status": "input-required",
    "artifacts": [
      {
        "parts": [
          { "kind": "text", "text": "Transaksi di atas limit otomatis. Persetujuan dibutuhkan." }
        ]
      }
    ],
    "metadata": {
      "inputModal": { "type": "approval" }
    }
  }
}

#Resubmission

Kadang task gagal karena faktor sementara — LLM timeout, server overload, atau gangguan jaringan. Resubmission adalah pola mengirim ulang task yang gagal, dengan harapan percobaan berikutnya berhasil. Kuncinya adalah memutuskan: kapan boleh diulang, dan berapa kali.

Aturan praktis:

• Ulang jika kegagalan bersifat sementara (timeout, 5xx, network). Jangan ulang jika kegagalan bersifat permanen (invalid input, authorization denied).
• Batasi jumlah percobaan, misalnya maksimal 3 kali, lalu eskalasi ke manusia.
• Pertahankan taskId yang sama saat mengulang, agar server tahu ini percobaan ulang dari task yang sama.

Alur resubmission dalam kode orkestrator:

resubmit.py — ulang task yang gagal sementara

MAX_ATTEMPTS = 3
 
async def send_with_resubmit(client, task):
    for attempt in range(1, MAX_ATTEMPTS + 1):
        result = await client.send_task(task)
        if result.status == "failed" and result.retryable:
            await asyncio.sleep(2 ** attempt)
            continue
        return result
    return escalate_to_human(task)

#Idempotency dan Retry

Idempotency menjamin bahwa mengulang aksi yang sama tidak menghasilkan efek ganda. Bayangkan client timeout setelah mengirim message/send, lalu mengirim ulang. Jika server memproses dua kali, agent bisa memesan tiket dobel atau mengirim duplikat. Solusinya: client menyertakan identitas task yang konsisten, dan server mendeteksi serta menolak duplikat.

Di A2A, setiap message membawa messageId, dan setiap task memiliki taskId. Server harus menyimpan riwayat task dan: jika messageId sudah pernah diproses, kembalikan hasil yang sama tanpa menjalankan ulang.

Pola retry yang aman — dan untuk implementasi backoff di Python, pustaka seperti tenacity yang dipasang lewat pip install tenacity sangat membantu:

retry:
  max_attempts: 5
  backoff:
    type: exponential
    base_ms: 500
    multiplier: 2
  retryable_statuses:
    - failed
    - timeout
  idempotency_key: messageId

#Long-Running Task dan Checkpoint/Resume

Task A2A bisa berjalan lama — analisis data besar, pipeline multi-step, atau proses yang menunggu banyak sub-task. Risikonya: proses server bisa restart, network bisa terputus, dan task yang setengah jalan akan hilang. Di sinilah checkpoint/resume berperan.

Konsepnya: simpan snapshot state task secara berkala ke penyimpanan eksternal. Saat proses restart, agent memuat snapshot terakhir dan melanjutkan dari sana, alih-alih mengulang dari awal.

Pola implementasinya:

checkpoint.py — simpan dan pulihkan state task

import json
 
def save_checkpoint(task_id, state, step):
    snapshot = {"task_id": task_id, "state": state, "step": step}
    state_store.put(f"checkpoint:{task_id}", json.dumps(snapshot))
 
def resume_task(task_id):
    raw = state_store.get(f"checkpoint:{task_id}")
    if not raw:
        return None
    snapshot = json.loads(raw)
    return snapshot

Beberapa praktik checkpoint yang baik:

• Simpan minimal state yang diperlukan untuk melanjutkan — data mentah lengkap justru boros.
• Tulis checkpoint sebelum memulai langkah yang berisiko, bukan setelahnya, sehingga rollback selalu aman.
• Kombinasikan dengan long-running task: karena A2A memisahkan task dari proses HTTP tunggal, client bisa menutup koneksi dan mengecek ulang status task nanti via tasks/get.

#Penutup

Pada episode ini kita merakit pola task lanjutan di atas lifecycle A2A. Fan-out dan fan-in memecah task besar menjadi paralel dan menyatukannya kembali, state input-required menghadirkan manusia dalam alur kerja, resubmission menangani kegagalan sementara, idempotency dan retry menjaga konsistensi, dan checkpoint-resume membuat task panjang bertahan dari restart.

Inti yang harus dibawa pulang:

• Fan-out/fan-in memecah task paralel dan mengagregasi hasil, dengan toleransi terhadap partial result.
• State input-required menghentikan task untuk menunggu keputusan manusia, lengkap dengan timeout agar tidak menggantung.
• Resubmission hanya untuk kegagalan sementara, dengan batas percobaan dan eskalasi.
• Idempotency memakai messageId dan taskId agar retry tidak menimbulkan efek ganda.
• Checkpoint-resume menyimpan snapshot state task sehingga proses restart tidak kehilangan progress.

Dengan pola-pola ini, orchestration multi-agent kalian sudah setara praktik produksi. Namun ada pertanyaan lain yang menunggu: ketika ratusan agent memproses ribuan task, berapa biayanya dan seberapa cepat?

Di episode 18 selanjutnya kita akan membahas Performance & Scale: menangani banyak remote agent dengan connection pooling, caching agent card, load balancing, serta tradeoff streaming versus polling dan minimasi payload untuk throughput tinggi. Sampai jumpa!