[Langchain] AWS Bedrock과 Langchain 활용한 LLM 어플리케이션 개발

[Langchain] AWS Bedrock과 Langchain 활용한 LLM 어플리케이션 개발

RAG 및 Agent를 활용하여 솔루션 엔지니어링에 도입하기에 앞서 AWS Bedrock과 Langchain을 활용하여 LLM(Large Language Model) 어플리케이션을 개발하는 방법에 대해 단계별로 알아보려고 합니다. 우선 개념에 대한 내용을 정리한 후  솔루션 엔지니어링 가이드 및 헬프센터 내용을 RAG를 통해 이해하고 답변을 하는 어플리케이션 개발을 진행할 예정이며, 이번 아티클에서는 LLM 개발을 처음 시작할 때 필요한 기초개념 및 기본 설계에 대한 내용을 다룹니다.

Langchain에 대한 이해

Langchain은 LLM 어플리케이션 개발을 위한 프레임워크로 다양한 컴포넌트들을 제공하여 LLM 기반 어플리케이션을 쉽게 구축할 수 있습니다.

LLM 컴포넌트

• Models: LLMs, Chat Models, Embeddings
• Prompts: 템플릿, Few-shot Examples
• Memory: 대화버퍼(이전 대화내용 저장), 요약메모리(긴 대화 내용을 요약하여 저장)
• Chains: LLMChain, 순차체인
• Agents: Zero-shot Agent, Structured Agent
• Tools: 검색(웹), 계산기

AWS Bedrock에 대한 이해

AWS Bedrock은 Amazon Web Services에서 제공하는 완전 관리형 서비스로, 다양한 기초 모델(Foundation Models)을 API를 통해 쉽게 사용할 수 있게 해주는 플랫폼입니다. Anthropic의 Claude, OpenAI의 GPT AI21 Labs의 Jurassic, Amazon의 Titan 등 최고 수준의 AI 모델들을 단일 API로 통합하여 제공합니다.

Bedrock의 특징

• 다양한 모델 선택 가능
• 보안 및 프라이버시
• 비용효율성
• AWS 솔루션과의 통합 용이

Langchain - Bedrock 연동 플로우

Langchain에서 Bedrock을 통해 LLM Application을 구축할 경우, 직접 LLM Models(OpenAI GPT, Anthropic Claude 등)의 API를 호출하지 않고, AWS Bedrock을 통해 간접적으로 모델과 통신하게 됩니다. 즉 Langchain Framework ▶ AWS SDK ▶ Bedrock Runtime ▶ LLM Model 순서로 통신을 합니다. Langchain을 통해 직접 LLM 어플리케이션 개발과 달리 AWS Bedrock을 통해 간접적으로 LLM 모델과 통신하므로 직접 OpenAI, Anthropic 등에서 API 키를 발급받지 않아도 되며, 과금 또한 AWS 사용비용으로 대체됩니다.

Lanchain - AWS Bedrock 연동 플로우

Bedrock 역할

• API 게이트웨이 역할
• 인증(Authentication)/인가(Authorization)
• 요청 및 응답
• 모델 버전 관리

Langchain 역할

• 프롬프트 엔지니어링
• Chain/Agent 로직 처리
• 응답 후 처리

API Key 연동 및 이에 따른 과금이 다름

Note. AWS Bedrock 연동시 장점
- 보안 강화: AWS IAM을 통한 접근제어
- 관리 용이성: AWS 모니터링 및 로깅 활용 가능
- 확장성: Bedrock을 통한 다양한 파운데이션 모델 사용 가능
- 비용효율성: 사용량에 비례하여 AWS 비용 과금

연동 상세

사전 준비사항

LLM 어플리케이션 개발을 위해 아래 환경을 준비합니다.

• Python 3.8 이상: Poetry를 활용한 설치 권장
• AWS 계정: Bedrock 접근권한 추가

Bedrock 기본권한 정책 추가

정책 생성

aws iam create-policy \
    --policy-name BedrockAccessPolicy \
    --policy-document '{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "BedrockFullAccess",
            "Effect": "Allow",
            "Action": [
                "bedrock:*"
            ],
            "Resource": "*"
        },
        {
            "Sid": "BedrockModelAccess",
            "Effect": "Allow",
            "Action": [
                "bedrock:InvokeModel",
                "bedrock:InvokeModelWithResponseStream"
            ],
            "Resource": "arn:aws:bedrock:*:281606047546:model/*"
        }
    ]
}'

 

IAM User에 정책추가

aws iam attach-user-policy \
    --user-name bedrock-user \
    --policy-arn arn:aws:iam::<User-ID>:policy/BedrockAccessPolicy

BedrockAccessPolicy 추가

 

정책추가 확인

aws iam list-attached-user-policies --user-name bedrock-user

 

Python 환경 설정

AWS Bedrock을 활용하여 테스트하기 위해 Langchain 공식문서의 ChatBedrock을 사용하였습니다. boto3를 활용한 AWS SDK기반 구현도 가능한 옵션입니다. 테스트시 anthropic.claude-3-5-sonnet-20240620-v1:0을 사용하였고,  Python 가상환경 및 의존성 설치는 Poetry를 활용하였습니다.

Poetry shell

poetry shell

 

Poetry 환경에서 모듈 설치

poetry add langchain_aws

 

코드작성

- ChatBedrock: 

from dotenv import load_dotenv
from langchain_aws import ChatBedrock
from langchain_teddynote import logging
import time
from botocore.exceptions import UnknownEndpointError

load_dotenv()

logging.langsmith("langchain_on_aws")

def get_bedrock_client():
    return ChatBedrock(
        model="anthropic.claude-3-5-sonnet-20240620-v1:0",
        model_kwargs={
            "temperature": 0.7,
            "max_tokens": 1024,
            "top_p": 0.9
        },
        region="ap-northeast-2"
    )

def get_bedrock_llm():
    try:
        llm = get_bedrock_client()
        return llm
    except Exception as e:
        print(f"Error creating Bedrock client: {e}")
        return None
    

def main():
    llm = get_bedrock_llm()
    if llm is None:
        print("LLM 초기화 실패")
        return
    # invocation
    message = [
        (
            "system",
            "You are a helpful assistant. Answer the user's question. Translate the answer to Korean."
        ),
        ("human", "What is the capital of Korea?")
    ]

    # 최대 재시도 횟수와 대기 시간 설정
    max_retries = 3
    retry_delay = 2  # 초단위
    
    for attempt in range(max_retries):
        try:
            response = llm.invoke(message)
            print(response.content)
            break
        except UnknownEndpointError as e:
            if attempt < max_retries - 1:
                print(f"요청이 제한되었습니다. {retry_delay}초 후 재시도합니다... ({attempt + 1}/{max_retries})")
                time.sleep(retry_delay)
                retry_delay *= 2  # 지수 백오프
            else:
                print("최대 재시도 횟수를 초과했습니다.")
                raise

if __name__ == "__main__":
    main()

 

실행결과