boto3 — bedrock-runtime.converse()
Referensi parameter converse() di boto3: bentuk request, bentuk respons, dan pola pemakaian di kode Python.
Intisari
- Halaman referensi, bukan tutorial. Buka saat kamu butuh nama field yang tepat.
- Semua nama parameter memakai camelCase — beda dengan snake_case khas Python.
- Respons berupa
dictbiasa, bukan objek bertipe. Tidak ada autocomplete. - Objek client boto3 aman dipakai ulang; buat sekali, jangan per request.
- boto3 sinkron. Untuk async, pakai
aioboto3atau bungkus denganasyncio.to_thread.
Membuat client
import boto3
from botocore.config import Config
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(
retries={"max_attempts": 5, "mode": "adaptive"},
read_timeout=300, # panjang, untuk respons LLM
connect_timeout=10,
),
)
Buat client sekali, pakai berkali-kali. Client boto3 aman dipakai bersama antar thread
dan menyimpan connection pool sendiri. Membuatnya di dalam fungsi yang dipanggil per request
adalah kebocoran performa yang senyap. Mode retry adaptive menyesuaikan diri
terhadap throttling — pilihan yang tepat untuk beban kerja LLM.
Bentuk request
response = bedrock.converse(
modelId="anthropic.claude-opus-5", # wajib
messages=[ # wajib
{"role": "user", "content": [{"text": "Halo"}]},
],
system=[{"text": "Kamu asisten."}], # opsional
inferenceConfig={ # opsional
"maxTokens": 4096,
"temperature": 0.7,
"topP": 0.9,
"stopSequences": ["\n\nHuman:"],
},
toolConfig={...}, # opsional
guardrailConfig={...}, # opsional
additionalModelRequestFields={...}, # parameter khusus model
additionalModelResponseFieldPaths=[...], # ambil field khusus dari respons
)
Bentuk respons
{
"output": {
"message": {
"role": "assistant",
"content": [
{"text": "..."},
{"toolUse": {"toolUseId": "...", "name": "...", "input": {...}}},
],
}
},
"stopReason": "end_turn",
"usage": {
"inputTokens": 123,
"outputTokens": 456,
"totalTokens": 579,
"cacheReadInputTokens": 0,
"cacheWriteInputTokens": 0,
},
"metrics": {"latencyMs": 1234},
}
Mengakses respons dengan aman
# ❌ pecah kalau blok pertama adalah toolUse
teks = response["output"]["message"]["content"][0]["text"]
# ✅ telusuri dan periksa
def ambil_teks(response: dict) -> str:
return "".join(
b["text"]
for b in response["output"]["message"]["content"]
if "text" in b
)
def ambil_tool(response: dict) -> list[dict]:
return [
b["toolUse"]
for b in response["output"]["message"]["content"]
if "toolUse" in b
]
Kelemahan boto3 dibanding SDK Anthropic: responsnya dict polos.
Tidak ada autocomplete, tidak ada pengecekan mypy, dan salah ketik nama key baru ketahuan
saat runtime. Bungkus akses ke respons dalam fungsi bertipe seperti di atas — atau lebih baik,
validasi dengan model Pydantic.
Membungkus respons dengan Pydantic
from pydantic import BaseModel
from typing import Literal
class Usage(BaseModel):
inputTokens: int
outputTokens: int
totalTokens: int
class HasilConverse(BaseModel):
teks: str
stopReason: Literal["end_turn", "max_tokens", "tool_use", "stop_sequence",
"guardrail_intervened", "content_filtered"]
usage: Usage
def panggil(**kw) -> HasilConverse:
r = bedrock.converse(**kw)
return HasilConverse(
teks="".join(b["text"] for b in r["output"]["message"]["content"] if "text" in b),
stopReason=r["stopReason"],
usage=Usage(**r["usage"]),
)
Pola ini memberi kembali keamanan tipe yang hilang saat memakai boto3 — dan itu langsung memakai Fase 2.
Streaming
response = bedrock.converse_stream(
modelId=MODEL,
messages=messages,
inferenceConfig={"maxTokens": 4096},
)
for event in response["stream"]:
if "messageStart" in event:
pass
elif "contentBlockStart" in event:
pass
elif "contentBlockDelta" in event:
delta = event["contentBlockDelta"]["delta"]
if "text" in delta:
print(delta["text"], end="", flush=True)
elif "contentBlockStop" in event:
pass
elif "messageStop" in event:
stop = event["messageStop"]["stopReason"]
elif "metadata" in event:
usage = event["metadata"]["usage"]
Async
boto3 sepenuhnya sinkron. Dua pilihan:
# 1. Bungkus dengan to_thread — sederhana, cukup untuk kebanyakan kasus
hasil = await asyncio.to_thread(bedrock.converse, modelId=..., messages=...)
# 2. aioboto3 — client async sungguhan
import aioboto3
session = aioboto3.Session()
async with session.client("bedrock-runtime", region_name="us-east-1") as b:
r = await b.converse(modelId=..., messages=...)
Panggilan boto3 sinkron di dalam async def membekukan event loop.
Ini persis jebakan yang dibahas di Fase 3. Kalau aplikasimu async dan kamu memakai boto3,
asyncio.to_thread bukan opsional.
Error yang perlu ditangani
| Kode | Artinya | Ulangi? |
|---|---|---|
ThrottlingException | Rate limit | ✅ Dengan backoff |
ModelTimeoutException | Model terlalu lama | ✅ |
ServiceUnavailableException | Bedrock sedang bermasalah | ✅ |
InternalServerException | Error sisi server | ✅ |
ValidationException | Request salah bentuk | ❌ |
AccessDeniedException | IAM / model access | ❌ |
ResourceNotFoundException | modelId salah atau tidak ada di region ini | ❌ |
from botocore.exceptions import ClientError
BOLEH_DIULANG = {
"ThrottlingException", "ModelTimeoutException",
"ServiceUnavailableException", "InternalServerException",
}
try:
r = bedrock.converse(...)
except ClientError as e:
kode = e.response["Error"]["Code"]
if kode in BOLEH_DIULANG:
... # tenacity mengurus dari sini
raise
Debugging
import logging
boto3.set_stream_logger("botocore", logging.DEBUG) # request/respons HTTP mentah
# Coba cepat dari terminal
aws bedrock-runtime converse \
--model-id anthropic.claude-opus-5 \
--messages '[{"role":"user","content":[{"text":"Halo"}]}]' \
--inference-config '{"maxTokens":100}' \
--region us-east-1
Rangkuman ini sengaja dipangkas ke bagian yang dipakai di roadmap. Buka sumber aslinya saat kamu butuh detail lengkap atau referensi parameter.