AWS Bedrock | KAITRUST AI 백과사전

📖 상세 설명

AWS Bedrock은 Amazon Web Services에서 제공하는 완전 관리형 생성형 AI 서비스입니다. Anthropic의 Claude, Meta의 Llama, Mistral AI, Amazon의 Titan 등 다양한 Foundation Model(FM)을 단일 API로 제공하여, 기업이 자체 인프라 구축 없이 최신 LLM을 활용한 애플리케이션을 개발할 수 있게 해줍니다.

2023년 9월 GA(정식 출시)되었으며, AWS의 기존 서비스들과 긴밀하게 통합됩니다. IAM으로 접근 제어, CloudWatch로 모니터링, S3에서 데이터 로드, Lambda/ECS에서 호출하는 등 AWS 생태계 내에서 자연스럽게 생성형 AI를 도입할 수 있습니다. 데이터가 AWS 리전 내에서 처리되어 규정 준수가 용이합니다.

Bedrock의 핵심 기능으로는 Knowledge Bases(RAG 자동화), Agents(도구 사용 및 워크플로우 자동화), Guardrails(콘텐츠 필터링 및 안전장치), Model Customization(미세 조정) 등이 있습니다. 특히 Knowledge Bases를 사용하면 S3의 문서를 벡터화하여 검색 증강 생성(RAG)을 코드 몇 줄로 구현할 수 있습니다.

실무에서 Bedrock은 기존 AWS 인프라를 사용하는 기업이 생성형 AI를 도입할 때 가장 빠른 경로입니다. 온디맨드 요금제로 사용한 토큰만큼만 비용을 지불하며, Provisioned Throughput으로 예약하면 처리량을 보장받고 비용도 절감할 수 있습니다. Claude 모델의 경우 Anthropic API 직접 사용 대비 AWS의 엔터프라이즈 지원을 받을 수 있다는 장점이 있습니다.

💻 코드 예제

import boto3
import json

# Bedrock Runtime 클라이언트 생성
bedrock = boto3.client(
    service_name='bedrock-runtime',
    region_name='us-east-1'  # Bedrock 지원 리전
)

# Claude 3.5 Sonnet 호출
def invoke_claude(prompt: str) -> str:
    response = bedrock.invoke_model(
        modelId='anthropic.claude-3-5-sonnet-20241022-v2:0',
        contentType='application/json',
        accept='application/json',
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 4096,
            "messages": [
                {
                    "role": "user",
                    "content": prompt
                }
            ]
        })
    )

    result = json.loads(response['body'].read())
    return result['content'][0]['text']

# 사용 예시
response = invoke_claude("Python으로 퀵소트 알고리즘을 구현해줘")
print(response)

# Amazon Titan Text 호출
def invoke_titan(prompt: str) -> str:
    response = bedrock.invoke_model(
        modelId='amazon.titan-text-premier-v1:0',
        contentType='application/json',
        accept='application/json',
        body=json.dumps({
            "inputText": prompt,
            "textGenerationConfig": {
                "maxTokenCount": 2048,
                "temperature": 0.7,
                "topP": 0.9
            }
        })
    )

    result = json.loads(response['body'].read())
    return result['results'][0]['outputText']

# Llama 3 호출
def invoke_llama(prompt: str) -> str:
    response = bedrock.invoke_model(
        modelId='meta.llama3-70b-instruct-v1:0',
        body=json.dumps({
            "prompt": prompt,
            "max_gen_len": 2048,
            "temperature": 0.7
        })
    )

    result = json.loads(response['body'].read())
    return result['generation']

import boto3
import json

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

def stream_claude_response(prompt: str):
    """Claude 응답을 스트리밍으로 받기"""
    response = bedrock.invoke_model_with_response_stream(
        modelId='anthropic.claude-3-5-sonnet-20241022-v2:0',
        contentType='application/json',
        accept='application/json',
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 4096,
            "messages": [
                {"role": "user", "content": prompt}
            ]
        })
    )

    # 스트리밍 응답 처리
    for event in response['body']:
        chunk = json.loads(event['chunk']['bytes'])

        if chunk['type'] == 'content_block_delta':
            # 텍스트 델타 출력
            text = chunk['delta'].get('text', '')
            print(text, end='', flush=True)

        elif chunk['type'] == 'message_stop':
            # 응답 완료
            print()  # 줄바꿈
            break

# 대화형 사용 예시
def chat_with_history(messages: list, new_message: str):
    """대화 히스토리를 유지하면서 채팅"""
    messages.append({"role": "user", "content": new_message})

    response = bedrock.invoke_model(
        modelId='anthropic.claude-3-5-sonnet-20241022-v2:0',
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 4096,
            "system": "당신은 친절한 AI 어시스턴트입니다.",
            "messages": messages
        })
    )

    result = json.loads(response['body'].read())
    assistant_message = result['content'][0]['text']
    messages.append({"role": "assistant", "content": assistant_message})

    return assistant_message

# 사용
conversation = []
print(chat_with_history(conversation, "안녕하세요!"))
print(chat_with_history(conversation, "AWS Bedrock에 대해 설명해주세요"))

import boto3
import json

# Bedrock Agent Runtime (Knowledge Base용)
bedrock_agent = boto3.client(
    'bedrock-agent-runtime',
    region_name='us-east-1'
)

def query_knowledge_base(
    knowledge_base_id: str,
    query: str,
    model_arn: str = 'arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-5-sonnet-20241022-v2:0'
):
    """Knowledge Base에서 RAG 검색 후 응답 생성"""

    response = bedrock_agent.retrieve_and_generate(
        input={'text': query},
        retrieveAndGenerateConfiguration={
            'type': 'KNOWLEDGE_BASE',
            'knowledgeBaseConfiguration': {
                'knowledgeBaseId': knowledge_base_id,
                'modelArn': model_arn,
                'retrievalConfiguration': {
                    'vectorSearchConfiguration': {
                        'numberOfResults': 5  # 검색할 문서 수
                    }
                }
            }
        }
    )

    # 생성된 응답
    generated_text = response['output']['text']

    # 인용된 소스 문서
    citations = response.get('citations', [])
    sources = []
    for citation in citations:
        for ref in citation.get('retrievedReferences', []):
            sources.append({
                'content': ref['content']['text'][:200],
                'location': ref['location']
            })

    return {
        'answer': generated_text,
        'sources': sources
    }

# 사용 예시
result = query_knowledge_base(
    knowledge_base_id='XXXXXXXXXX',  # AWS 콘솔에서 생성한 KB ID
    query='회사 휴가 정책은 어떻게 되나요?'
)

print(f"답변: {result['answer']}")
print(f"\n참조 문서: {len(result['sources'])}개")
for src in result['sources']:
    print(f"  - {src['location']}")

# Knowledge Base 없이 직접 벡터 검색만
def retrieve_only(knowledge_base_id: str, query: str):
    """생성 없이 검색만 수행"""
    response = bedrock_agent.retrieve(
        knowledgeBaseId=knowledge_base_id,
        retrievalQuery={'text': query},
        retrievalConfiguration={
            'vectorSearchConfiguration': {
                'numberOfResults': 10
            }
        }
    )

    return [
        {
            'text': r['content']['text'],
            'score': r['score'],
            'location': r['location']
        }
        for r in response['retrievalResults']
    ]

🗣️ 실무에서 이렇게 말하세요

💬 AI 도입 기획 회의에서

"Bedrock으로 가면 모델 호스팅 인프라 신경 안 써도 되고, Claude랑 Llama 같이 테스트해보면서 우리 유스케이스에 맞는 모델을 고를 수 있어요. Knowledge Base 기능 쓰면 S3에 있는 문서들로 RAG도 바로 구현됩니다. 데이터가 AWS 내에서만 처리되니까 보안팀 검토도 수월할 거예요."

💬 면접에서

"Bedrock을 활용해서 사내 문서 검색 챗봇을 구축했습니다. Knowledge Base에 Confluence 문서를 동기화하고 Claude 3.5 Sonnet으로 응답을 생성했는데, Guardrails를 적용해서 민감 정보 노출을 방지했습니다. Lambda에서 호출하고 API Gateway로 노출해서 Slack 봇과 연동했고, 월 비용은 약 $200 수준이었습니다."

💬 비용 최적화 논의에서

"Bedrock 비용이 예상보다 높은데, 로그 분석해보니까 프롬프트가 너무 길어요. 시스템 프롬프트를 캐싱하고, 대화 히스토리 요약으로 토큰을 줄여야 해요. 그리고 단순 질문은 Claude Haiku로 라우팅하고 복잡한 건 Sonnet으로 보내는 모델 라우터를 만들면 70% 정도 절감할 수 있습니다."

⚠️ 흔한 실수 & 주의사항

❌

모델 액세스 요청 누락

Bedrock 모델은 AWS 콘솔에서 명시적으로 액세스를 요청해야 사용할 수 있습니다. "Model access" 메뉴에서 원하는 모델을 활성화하지 않으면 AccessDeniedException이 발생합니다. 일부 모델은 승인까지 시간이 걸릴 수 있습니다.

❌

리전 제한 미확인

모든 Bedrock 모델이 모든 리전에서 사용 가능한 것은 아닙니다. Claude 최신 버전은 us-east-1, us-west-2 등 일부 리전에서만 지원됩니다. 서울 리전(ap-northeast-2) 지원 여부를 꼭 확인하세요.

✅

프로덕션 모범 사례

Guardrails로 입출력 필터링을 적용하고, CloudWatch Logs로 모든 호출을 로깅하세요. 비용 알람을 설정하고, 고정 트래픽이 예상되면 Provisioned Throughput을 검토하세요. 민감 데이터는 PII 필터를 활성화하거나 호출 전에 마스킹 처리하세요.

🔗 관련 용어

📚 더 배우기

📄 AWS Bedrock 공식 문서 💰 Bedrock 가격 정책 (모델별 토큰 단가) 💻 Amazon Bedrock 샘플 코드 (GitHub)