Fase 4 · Bicara dengan LLM

Streaming Messages

Streaming wajib untuk UX chat dan untuk max_tokens besar. Ini cara memakainya dengan benar.

Sumber asli platform.claude.com Resmi Rangkuman ~6 menit baca

Intisari

Pakai helper client.messages.stream() — ia mengakumulasi state dan menyediakan text_stream.
stream.get_final_message() memberi objek respons lengkap setelah stream selesai, termasuk usage.
Wajib streaming saat max_tokens besar — tanpa itu koneksi HTTP bisa timeout.
Enam jenis event; untuk teks biasa kamu cukup peduli pada content_block_delta.
Versi async tinggal ganti jadi async with + async for.

Cara paling sederhana

with client.messages.stream(
    model="claude-opus-5",
    max_tokens=64000,
    messages=[{"role": "user", "content": "Tulis artikel panjang tentang RAG"}],
) as stream:
    for teks in stream.text_stream:
        print(teks, end="", flush=True)

    final = stream.get_final_message()
    print(f"\n\nToken keluar: {final.usage.output_tokens}")

Kenapa max_tokens besar wajib streaming: tanpa streaming, koneksi HTTP berdiam tanpa data sampai seluruh respons selesai dibuat. Untuk output puluhan ribu token, itu bisa lewat dari batas waktu koneksi dan gagal. SDK bahkan menolak request non-streaming yang diperkirakan lebih dari ~10 menit.

Versi async

async with async_client.messages.stream(
    model="claude-opus-5",
    max_tokens=64000,
    messages=[{"role": "user", "content": "..."}],
) as stream:
    async for teks in stream.text_stream:
        print(teks, end="", flush=True)

    final = await stream.get_final_message()

Jenis event

Event	Kapan	Isinya
`message_start`	Sekali di awal	Metadata pesan, model, usage awal
`content_block_start`	Tiap blok baru dimulai	Tipe blok: text / thinking / tool_use
`content_block_delta`	Tiap potongan	Isi yang mengalir
`content_block_stop`	Blok selesai	—
`message_delta`	Menjelang akhir	`stop_reason`, usage akhir
`message_stop`	Sekali di akhir	—

Menangani teks, thinking, dan tool sekaligus

with client.messages.stream(
    model="claude-opus-5",
    max_tokens=64000,
    thinking={"type": "adaptive", "display": "summarized"},
    tools=tools,
    messages=messages,
) as stream:
    for event in stream:
        if event.type == "content_block_start":
            if event.content_block.type == "thinking":
                print("\n[berpikir...]")
            elif event.content_block.type == "text":
                print("\n[jawaban]")
            elif event.content_block.type == "tool_use":
                print(f"\n[memanggil {event.content_block.name}]")

        elif event.type == "content_block_delta":
            if event.delta.type == "thinking_delta":
                print(event.delta.thinking, end="", flush=True)
            elif event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)
            elif event.delta.type == "input_json_delta":
                pass      # argumen tool mengalir sebagai JSON parsial

    final = stream.get_final_message()

thinking.display default-nya "omitted" pada Claude Opus 5. Artinya blok thinking tetap muncul di stream, tapi isinya string kosong. Kalau kamu memang ingin menampilkan progres berpikir ke pengguna, set display: "summarized" secara eksplisit — tanpa itu, UI-mu akan terlihat menggantung lama sebelum teks pertama muncul.

Menampilkan progres dan biaya

total_keluar = 0

with client.messages.stream(...) as stream:
    for event in stream:
        if event.type == "content_block_delta" and event.delta.type == "text_delta":
            print(event.delta.text, end="", flush=True)
        elif event.type == "message_delta" and event.usage:
            total_keluar = event.usage.output_tokens

    final = stream.get_final_message()

print(f"\nmasuk={final.usage.input_tokens} keluar={final.usage.output_tokens}")

Menangani error di tengah stream

import anthropic

try:
    with client.messages.stream(...) as stream:
        for teks in stream.text_stream:
            print(teks, end="", flush=True)
except anthropic.APIConnectionError:
    print("\n[koneksi terputus]")
except anthropic.RateLimitError:
    print("\n[kena rate limit]")
except anthropic.APIStatusError as e:
    print(f"\n[error {e.status_code}]")

Stream yang terputus meninggalkan teks setengah jadi. Kumpulkan potongan yang sudah diterima ke buffer supaya kamu bisa menampilkan yang sudah ada, atau menyimpannya untuk dilanjutkan. Jangan asumsikan stream selalu selesai.

Bentuk tingkat rendah

for event in client.messages.create(
    model="claude-opus-5",
    max_tokens=64000,
    messages=[...],
    stream=True,
):
    print(event.type)

Bentuk ini memberi iterator event mentah tanpa akumulasi apa pun — tidak ada text_stream dan tidak ada get_final_message(). Pakai hanya kalau kamu memang butuh kendali penuh atau ingin menghemat memori seminimal mungkin.

Meneruskan stream ke browser

from collections.abc import AsyncIterable
from fastapi.sse import EventSourceResponse

@app.post("/chat/stream", response_class=EventSourceResponse)
async def chat_stream(req: ChatRequest) -> AsyncIterable[str]:
    async with async_client.messages.stream(
        model="claude-opus-5",
        max_tokens=16000,
        messages=[{"role": "user", "content": req.pesan}],
    ) as stream:
        async for teks in stream.text_stream:
            yield teks

Ini persis jembatan ke Fase 5 — SSE FastAPI dibahas lengkap di sana.

Kapan tidak perlu streaming

Klasifikasi atau ekstraksi — outputnya pendek, tidak ada yang menunggu
Pekerjaan batch — tidak ada pengguna yang melihat
Panggilan di dalam pipeline yang hasilnya diproses lagi

Kalau max_tokens di bawah ~16.000 dan tidak ada yang menunggu di layar, non-streaming lebih sederhana.

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