Fase 3 · Async & HTTP

asyncio — Coroutines and Tasks

Inti asyncio: bagaimana coroutine dijalankan, cara membuat pekerjaan berjalan paralel, dan cara memasang timeout.

Sumber asli docs.python.org Resmi Rangkuman ~8 menit baca

Intisari

Memanggil async def tidak menjalankan apa pun — ia mengembalikan coroutine. Butuh await atau task.
await berurutan = tetap lambat. Paralelisme datang dari gather() atau TaskGroup.
asyncio.gather(*coros) menjalankan semuanya bersamaan dan mengembalikan hasil sesuai urutan input.
TaskGroup (3.11+) lebih aman: satu gagal → sisanya dibatalkan otomatis, tidak ada task yatim.
asyncio.timeout() memberi batas waktu pada blok kode mana pun.

Kenapa async penting untuk AI engineer

Panggilan LLM itu I/O-bound dan lambat — 5 sampai 60 detik bukan hal aneh. Selama menunggu, CPU-mu benar-benar menganggur. Tanpa async, proses Python-mu berdiri diam menunggu satu respons, sementara ia sebenarnya bisa menunggu lima puluh sekaligus.

Sequential : [10 dtk][10 dtk][10 dtk]  →  30 detik
Async      : [10 dtk]
             [10 dtk]                  →  10 detik
             [10 dtk]

Coroutine: memanggil ≠ menjalankan

async def ambil(url: str) -> str:
    await asyncio.sleep(1)
    return "data"

c = ambil("http://a")   # BELUM jalan — c adalah objek coroutine
hasil = await c          # baru sekarang jalan

Kalau kamu lupa await, Python memberi peringatan RuntimeWarning: coroutine was never awaited — dan fungsimu tidak pernah dijalankan sama sekali, tanpa error. Ini bug async yang paling sering terjadi. Ruff aturan RUF006 dan ASYNC membantu menangkapnya.

Titik masuk: `asyncio.run()`

import asyncio

async def main() -> None:
    hasil = await ambil("http://a")
    print(hasil)

asyncio.run(main())     # membuat event loop, jalankan, lalu tutup

Panggil asyncio.run() sekali saja, di titik paling luar program. Jangan memanggilnya di dalam fungsi async — itu error.

Jebakan terbesar: `await` di dalam loop

# ❌ Ini TIDAK paralel — sama lambatnya dengan kode sinkron
hasil = []
for url in urls:
    hasil.append(await ambil(url))    # menunggu satu selesai sebelum mulai berikutnya

await artinya "tunggu sampai selesai". Menaruhnya di dalam loop membuat tiap iterasi menunggu iterasi sebelumnya. Kamu memakai async tapi tidak mendapat manfaat apa pun. Paralelisme datang dari menjadwalkan semuanya lebih dulu, baru menunggu.

`gather()` — jalankan bersamaan

hasil = await asyncio.gather(
    ambil("http://a"),
    ambil("http://b"),
    ambil("http://c"),
)
# hasil = ['data-a', 'data-b', 'data-c'] — URUTAN SESUAI INPUT, bukan urutan selesai

# Dari sebuah list
hasil = await asyncio.gather(*(ambil(u) for u in urls))

Menangani kegagalan

# Default: satu gagal → gather langsung melempar, hasil lain hilang
hasil = await asyncio.gather(*coros)

# Lebih tahan banting: exception ikut masuk ke daftar hasil
hasil = await asyncio.gather(*coros, return_exceptions=True)

for url, r in zip(urls, hasil):
    if isinstance(r, Exception):
        logger.warning("gagal %s: %s", url, r)
    else:
        proses(r)

return_exceptions=True hampir selalu yang kamu mau saat memproses banyak dokumen atau memanggil LLM secara massal. Satu dokumen rusak tidak boleh menjatuhkan seluruh batch yang sudah dibayar token-nya.

`TaskGroup` — cara modern (Python 3.11+)

async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(ambil("http://a"))
        t2 = tg.create_task(ambil("http://b"))

    # Di luar blok: semua task DIJAMIN selesai
    print(t1.result(), t2.result())

	`gather()`	`TaskGroup`
Jumlah task	Ditentukan di muka	Bisa ditambah dinamis di dalam blok
Saat satu gagal	Task lain tetap jalan (jadi yatim)	Task lain dibatalkan otomatis
Error ganda	Hanya yang pertama	`ExceptionGroup` berisi semuanya
Versi	Semua	3.11+

try:
    async with asyncio.TaskGroup() as tg:
        for u in urls:
            tg.create_task(ambil(u))
except* httpx.HTTPError as eg:       # perhatikan except* untuk ExceptionGroup
    for e in eg.exceptions:
        logger.error("http error: %s", e)

Timeout — jangan pernah dilewatkan

async with asyncio.timeout(30):
    hasil = await panggil_llm(prompt)       # TimeoutError kalau lebih dari 30 detik

# Bentuk lama, masih umum dipakai
hasil = await asyncio.wait_for(panggil_llm(prompt), timeout=30)

Panggilan jaringan tanpa timeout adalah bug, bukan pilihan. Server yang menggantung akan membuat task-mu menunggu selamanya, memegang koneksi dan memori. Untuk LLM streaming, berikan timeout yang longgar (misalnya 300 detik) — tapi tetap ada.

Task: menjadwalkan tanpa langsung menunggu

task = asyncio.create_task(latar_belakang())   # mulai jalan SEKARANG
lakukan_hal_lain()
hasil = await task                              # baru tunggu di sini

task.cancel()          # minta dibatalkan
task.done()            # sudah selesai?
task.result()          # ambil hasil (error kalau belum selesai)

Simpan referensi ke task yang kamu buat. Event loop hanya menyimpan referensi lemah; task yang tidak dipegang siapa pun bisa dibuang garbage collector di tengah jalan. Ruff aturan RUF006 memperingatkan soal ini.

Memproses hasil sambil jalan

for coro in asyncio.as_completed(coros):
    hasil = await coro       # yang tercepat selesai keluar duluan
    tampilkan(hasil)

Berguna untuk UI yang ingin menampilkan hasil secara bertahap alih-alih menunggu semuanya beres.

Membatasi concurrency

sem = asyncio.Semaphore(5)      # maksimal 5 berjalan bersamaan

async def ambil_terbatas(url: str) -> str:
    async with sem:
        return await ambil(url)

hasil = await asyncio.gather(*(ambil_terbatas(u) for u in urls))

Tanpa semaphore, gather atas 500 URL akan menembakkan 500 request sekaligus. Hasilnya: rate limit dari penyedia API, koneksi habis, atau server tujuan yang kewalahan. Semaphore adalah rem yang wajib ada di setiap fan-out ke API berbayar.

Menjalankan kode blocking tanpa menyumbat

hasil = await asyncio.to_thread(fungsi_blocking, arg1, arg2)

Untuk fungsi sinkron yang lambat (baca file besar, library sinkron, komputasi berat), jalankan di thread terpisah supaya event loop tetap bisa melayani task lain.

Ringkasan API

Fungsi	Untuk
`asyncio.run(coro)`	Titik masuk program
`asyncio.gather(*coros)`	Jalankan paralel, kumpulkan hasil
`asyncio.TaskGroup()`	Paralel dengan pembatalan otomatis
`asyncio.create_task(coro)`	Jadwalkan sekarang, tunggu nanti
`asyncio.timeout(n)`	Batas waktu untuk satu blok
`asyncio.as_completed(coros)`	Proses hasil sesuai urutan selesai
`asyncio.Semaphore(n)`	Batasi concurrency
`asyncio.to_thread(fn)`	Jalankan fungsi blocking di thread
`asyncio.sleep(n)`	Tunda tanpa menyumbat loop

Rangkuman ini sengaja dipangkas ke bagian yang dipakai di roadmap. Buka sumber aslinya saat kamu butuh detail lengkap atau referensi parameter.