추론 인프라

resp = runtime.converse(
    modelId='anthropic.claude-sonnet-4-6',
    messages=[
        {'role': 'user',
         'content': [{'text': '...'}]},
    ],
)

bedrock = boto3.client('bedrock', region_name='us-east-1')

# Cross-Region 프로파일을 래핑해 팀 태그 부여
response = bedrock.create_inference_profile(
    inferenceProfileName='team-alpha-claude',
    modelSource={'copyFrom': 'us.anthropic.claude-sonnet-4-6'},
    tags=[
        {'key': 'team', 'value': 'alpha'},
        {'key': 'environment', 'value': 'production'},
    ],
)
profile_arn = response['inferenceProfileArn']

# modelId에 프로파일 ARN 전달 — 비용이 팀 태그로 귀속
resp = runtime.converse(
    modelId=profile_arn,
    messages=[{'role': 'user', 'content': [{'text': '안녕하세요'}]}],
)

bedrock = boto3.client('bedrock', region_name='us-east-1')

response = bedrock.create_model_invocation_job(
    jobName='document-summary-batch',
    modelId='us.anthropic.claude-sonnet-4-6',
    roleArn='arn:aws:iam::123456789012:role/BedrockBatchRole',
    inputDataConfig={
        's3InputDataConfig': {
            's3Uri': 's3://my-bucket/batch-input/requests.jsonl',
        },
    },
    outputDataConfig={
        's3OutputDataConfig': {
            's3Uri': 's3://my-bucket/batch-output/',
        },
    },
)

job_arn = response['jobArn']

• 입력은 S3의 JSONL — 한 줄이 요청 1건: {"recordId": "1", "modelInput": {"messages": [...]}}
• create_model_invocation_job — 실시간 API가 아닌 비동기 Job 제출
• roleArn — 입력·출력 S3 버킷을 읽고 쓸 IAM 역할
• 출력도 S3 — 결과 JSONL이 지정 경로에 저장
• jobArn으로 상태 폴링 — Completed 후 출력 경로 확인

import boto3

bedrock = boto3.client('bedrock', region_name='us-east-1')

# 최소 1개월 커밋 — modelUnits 단위로 전용 용량 확보
# PT는 레거시 모델 전용 — Claude 3.5 Sonnet v2가 지원 상한
response = bedrock.create_provisioned_model_throughput(
    modelUnits=1,
    provisionedModelName='prod-sonnet-legacy',
    modelId='anthropic.claude-3-5-sonnet-20241022-v2:0',
    commitmentDuration='OneMonth',
)
provisioned_arn = response['provisionedModelArn']

# 호출은 동일 — modelId에 프로비저닝 ARN 전달
runtime = boto3.client('bedrock-runtime', region_name='us-east-1')
response = runtime.converse(
    modelId=provisioned_arn,
    messages=[{'role': 'user', 'content': [{'text': '안녕하세요'}]}],
)

# System Prompt Caching — cachePoint 블록으로 캐시 지점 지정
system_prompt = """당신은 고객 서비스 전담 AI입니다.
[매뉴얼: 100KB 정책 문서]
[FAQ: 50KB 자주 묻는 질문]
"""

response = bedrock_runtime.converse(
    modelId='us.anthropic.claude-sonnet-4-6',
    system=[
        {'text': system_prompt},
        {'cachePoint': {'type': 'default'}},  # ← 여기까지 캐싱
    ],
    messages=[
        {'role': 'user', 'content': [{'text': '배송 기간은?'}]}
    ]
)

# 첫 요청: 매뉴얼 + FAQ를 전부 읽고 캐시에 기록
# 두 번째 요청부터: 캐시 재사용 — 입력 토큰 비용 최대 90% 절감
# Bedrock의 Sonnet 4.6은 최소 1,024 토큰부터 캐싱 적용

• cachePoint 블록 앞까지가 캐시 대상 — 시스템 프롬프트·문서를 앞쪽에 배치
• 두 번째 요청부터 캐시 재사용 — 입력 토큰 비용 최대 90% 절감
• Sonnet 4.6은 최소 1,024 토큰부터 캐싱 적용

import boto3
from botocore.config import Config

# boto3 재시도 설정 — standard 모드는 지수 백오프 + 지터 적용
config = Config(
    retries={
        'max_attempts': 5,   # 총 시도 횟수 (standard 모드 기본 3)
        'mode': 'standard',
    }
)

bedrock_runtime = boto3.client('bedrock-runtime', config=config)

# ThrottlingException은 재시도 대상 — 설정만으로 자동 처리
response = bedrock_runtime.converse(
    modelId='us.anthropic.claude-sonnet-4-6',
    messages=[
        {'role': 'user', 'content': [{'text': '안녕하세요'}]}
    ],
)