AWS Bedrockの Claudeでストリーミングレスポンスを取得

bedrock-streaming.py

#!/usr/bin/env python3
import json
import os
import sys
import boto3

# モデルを定義
#modelId = "anthropic.claude-3-5-sonnet-20241022-v1" # Inference profile ARN ... NG
modelId = "apac.anthropic.claude-3-5-sonnet-20241022-v2:0" # Inference profile ID ... OK

# デバッグ用にログを有効化
boto3.set_stream_logger('', level=20)  # INFO レベル

# Bedrockクライアントの作成
bedrock_runtime = boto3.client('bedrock-runtime')

# リクエストボディを定義
body = json.dumps({
   "anthropic_version": "bedrock-2023-05-31",
   "max_tokens": 1000,
   "messages": [
       {"role": "assistant", "content": "Hi, I'm Claude. How can I help you?"},
       {
           "role": "user",
           "content": [{"type": "text", "text": "いろは歌を教えて"}]
       }
   ]
})

# レスポンスを定義
response = bedrock_runtime.invoke_model_with_response_stream(body=body, modelId=modelId)

# ストリーミング出力
for event in response.get("body"):
   chunk = json.loads(event["chunk"]["bytes"])
   if chunk['type'] == 'content_block_delta' and chunk['delta']['type'] == 'text_delta':
       print(chunk['delta']['text'], end="")

# ストリーミング終了後に改行
print()