Amazon SageMaker, Amazon OpenSearch Service, Streamlit, LangChain을 사용하여 강력한 질문/답변 봇 구축하기

이번 게시글은 영문 게시글(Build a powerful question answering bot with Amazon SageMaker, Amazon OpenSearch Service, Streamlit, and LangChain by by Amit Arora, Navneet Tuteja, and Xin Huang)의 한글 번역글입니다.

엔터프라이즈 환경에서 생성 AI와 대규모 언어 모델(LLM; Large Language Models)의 가장 일반적인 유스케이스 중 하나는 기업의 지식 코퍼스를 기반으로 질문에 답변하는 것입니다. Amazon Lex는 AI 기반 챗봇을 구축하기 위한 프레임워크를 제공합니다. 사전 훈련된 파운데이션 모델(FM; Foundation Models)은 다양한 주제에 대한 요약, 텍스트 생성, 질문 답변과 같은 자연어 이해(NLU; Natural Language Understanding) 작업은 잘 수행하지만, 훈련 데이터의 일부로 보지 못한 콘텐츠에 대한 질문에는 정확한(오답 없이) 답변을 제공하는 데 어려움을 겪거나 완전히 실패합니다. 또한 FM은 특정 시점의 데이터 스냅샷으로 훈련하기에 추론 시점에 새로운 데이터에 액세스할 수 있는 고유한 기능이 없기에 잠재적으로 부정확하거나 부적절한 답변을 제공할 수 있습니다.

이 문제를 해결하기 위해 흔히 사용되는 접근 방식은 검색 증강 생성(RAG; Retrieval Augmented Generation)이라는 기법을 사용하는 것입니다. RAG 기반 접근 방식에서는 LLM을 사용하여 사용자 질문을 벡터 임베딩으로 변환한 다음, 엔터프라이즈 지식 코퍼스에 대한 임베딩이 미리 채워진 벡터 데이터베이스에서 이러한 임베딩에 대한 유사성 검색을 수행합니다. 소수의 유사한 문서(일반적으로 3개)가 사용자 질문과 함께 다른 LLM에 제공된 ‘프롬프트’에 컨텍스트로 추가되고, 해당 LLM은 프롬프트에 컨텍스트로 제공된 정보를 사용하여 사용자 질문에 대한 답변을 생성합니다. RAG 모델은 매개변수 메모리(parametric memory)는 사전 훈련된 seq2seq 모델이고 비매개변수 메모리(non-parametric memory)는 사전 훈련된 신경망 검색기로 액세스되는 위키백과의 고밀도 벡터 색인 모델로 2020년에 Lewis 등이 도입했습니다. RAG 기반 접근 방식의 전반적 구조를 이해하려면 Question answering using Retrieval Augmented Generation with foundation models in Amazon SageMaker JumpStart 블로그를 참조하기 바랍니다.

이 게시물에서는 질문/답변 봇과 같은 엔터프라이즈 지원 RAG 애플리케이션을 만들기 위한 모든 구성 요소가 포함된 단계별 가이드를 제공합니다. 다양한 AWS 서비스, 오픈 소스 기반 모델(텍스트 생성을 위한 FLAN-T5 XXL, 임베딩을 위한 GPT-J-6B), 모든 구성 요소와의 인터페이스를 위한 LangChain, 봇 프런트엔드 구축을 위한 Streamlit 등의 패키지를 조합하여 사용합니다.

저희는 이 솔루션을 구축하는 데 필요한 모든 리소스를 구축하는 AWS CloudFormation 템플릿을 제공합니다. 그런 다음 모든 것을 하나로 묶어주는 LangChain 툴킷을 사용하는 방법을 시연합니다.

Amazon SageMaker에서 호스팅되는 LLM과의 인터페이스
지식 베이스 문서 청킹
Amazon OpenSearch 서비스에 문서 임베딩 수집
질문 답변 작업 구현

동일한 아키텍처를 사용하여 오픈 소스 모델을 Amazon Titan 모델로 교체할 수 있습니다. Amazon Bedrock이 출시되면 Amazon Bedrock을 사용하여 유사한 생성형 AI 애플리케이션을 구현하는 방법을 보여주는 후속 포스팅을 게시할 예정이니 기대해주세요.

솔루션 개요

이 포스트의 지식 코퍼스로 SageMaker 문서를 사용합니다. 이 사이트의 HTML 페이지를 청크 간의 컨텍스트 연속성을 유지하기 위해 겹치는 작은 정보 청크로 잘게 나눈 다음, GPT-J-6B 모델을 사용하여 이러한 청크를 임베딩으로 변환하고 임베딩을 OpenSearch 서비스에 저장합니다. 모든 요청을 Lambda로 라우팅하는 작업을 처리하기 위해 Amazon API 게이트웨이를 사용하여 AWS Lambda 함수 내부에 RAG 기능을 구현합니다. API 게이트웨이를 통해 함수를 호출하는 챗봇 애플리케이션을 Streamlit에 구현하고, 이 함수는 사용자 질문의 임베딩에 대해 OpenSearch Service 인덱스에서 유사성 검색(similarity search)을 수행합니다. 일치하는 문서(청크)는 Lambda 함수에 의해 컨텍스트로 프롬프트에 추가되고, 함수는 SageMaker 엔드포인트로 배포된 Flan-t5-XXL 모델을 사용하여 사용자 질문에 대한 답변을 생성합니다. 이 게시물의 모든 코드는 깃허브 리포지토리에 공개되어 있습니다.

다음 그림은 제안된 솔루션의 개략적인 아키텍처입니다.

그림 1. 아키텍처

단계별 설명

사용자가 Streamlit 웹 애플리케이션을 통해 쿼리를 제공합니다.
Streamlit 애플리케이션이 API 게이트웨이 엔드포인트 REST API를 호출합니다.
API 게이트웨이가 Lambda 함수를 호출합니다.
이 함수는 SageMaker 엔드포인트를 호출하여 사용자 질문을 임베딩으로 변환합니다.
이 함수는 사용자 쿼리와 유사한 문서를 찾기 위해 OpenSearch 서비스 API를 호출합니다.
이 함수는 사용자 쿼리와 “유사한 문서”를 컨텍스트로 사용하여 “프롬프트”를 생성하고 SageMaker 엔드포인트에 응답을 생성하도록 요청합니다.
응답은 함수에서 API 게이트웨이로 제공됩니다.
API 게이트웨이는 Streamlit 애플리케이션에 응답을 제공합니다.
사용자는 Streamlit 애플리케이션에서 응답 결과를 확인합니다.

아키텍처 다이어그램에서 볼 수 있듯이 다음과 같은 AWS 서비스를 사용합니다.

두 개의 LLM을 호스팅하기 위한 SageMaker 및 Amazon SageMaker JumpStart
엔터프라이즈 지식 코퍼스의 임베딩을 저장하고 사용자 질문에 대한 유사도 검색을 수행하기 위한 OpenSearch Service
RAG 기능을 구현하고 API 게이트웨이를 통해 REST 엔드포인트로 노출하기 위한 Lambda
OpenSearch로의 대규모 데이터 수집을 위한 Amazon SageMaker Processing jobs
Streamlit 애플리케이션 호스팅을 위한 Amazon SageMaker Studio.
액세스 관리를 위한 AWS Identity and Access Management
인프라를 통해 전체 솔루션 스택을 코드로 생성하기 위한 AWS CloudFormation

이 솔루션에 사용된 오픈 소스 패키지의 경우, OpenSearch 서비스 및 SageMaker와의 인터페이스에는 LangChain을 사용하고, Lambda에서 REST API 인터페이스를 구현하는 데에는 FastAPI를 사용합니다.

이 게시물에서 소개한 솔루션을 AWS 계정에서 인스턴스화하는 워크플로는 다음과 같습니다:

여러분의 AWS 계정에서 이 게시물과 함께 제공된 CloudFormation 템플릿을 실행합니다. 그러면 이 솔루션에 필요한 모든 인프라 리소스가 생성됩니다:
1. LLM을 배포하는 SageMaker 엔드포인트
2. OpenSearch 서비스 클러스터
3. API 게이트웨이
4. Lambda 함수
5. SageMaker 노트북
6. IAM 역할
SageMaker 노트북에서 data_ingestion_to_vectordb.ipynb 노트북을 실행하여 SageMaker 문서에서 OpenSearch 서비스 인덱스로 데이터를 수집합니다.
SageMaker 스튜디오의 터미널에서 Streamlit 애플리케이션을 실행하고 새 브라우저 탭에서 애플리케이션의 URL을 엽니다.
Streamlit 앱에서 제공하는 채팅 인터페이스를 통해 SageMaker 대해 질문하고 LLM에서 생성된 응답을 확인합니다.

이러한 단계는 다음 절에서 자세히 설명합니다.

사전 요구 사항

이 게시물에 제공된 솔루션을 구현하려면 AWS 계정이 있어야 하며 LLM, OpenSearch Service 및 SageMaker에 대해 숙지해야 합니다.

LLM을 호스팅하기 위해 가속화된 인스턴스(GPU)에 액세스할 수 있어야 합니다. 이 솔루션은 ml.g5.12xlarge와 ml.g5.24xlarge 인스턴스를 각각 하나씩 사용하며, 다음 스크린샷과 같이 AWS 계정에서 해당 인스턴스의 가용성을 확인하고 필요에 따라 서비스 한도(quota) 증가 요청을 통해 해당 인스턴스를 요청할 수 있습니다. (역자주- ml.g5.12xlarge, ml.g5.24xlarge의 한도(quota) 신청적용은 12~24시간 정도 소요되며, 둘 중 하나라도 증설이 안된 경우엔 설치시 CloudFormation 스택은 롤백이 됩니다.)
그림 2: 서비스 한도 증가 요청

Use AWS Cloud Formation to create the solution stack

AWS CloudFormation을 사용하여 aws-llm-apps-blog라는 SageMaker 노트북과 LLMAppsBlogIAMRole이라는 IAM 역할을 생성합니다. 리소스를 배포할 리전을 확인 후 해당 리전의 Launch Stack을 클릭합니다. CloudFormation 템플릿에 필요한 모든 매개변수에는 기본값이 이미 입력되어 있지만, 사용자가 제공해야 하는 OpenSearch 서비스 비밀번호는 예외입니다. OpenSearch Service 사용자 이름과 비밀번호는 다음 단계에서 사용하므로 메모해 두세요. 이 템플릿을 런칭하는 데에 약 15분 정도 소요됩니다.

AWS Region	Link
`us-east-1`
`us-west-2`
`eu-west-1`
`ap-northeast-1`

스택이 성공적으로 생성되면, AWS CloudFormation 콘솔에서 스택의 출력 탭으로 이동하여 OpenSearchDomainEndpoint 및 LLMAppAPIEndpoint의 값을 기록합니다. 다음 단계에서는 이 값들을 사용합니다.
그림 3: CloudFormation 스택 출력

OpenSearch 서비스에 데이터 수집

데이터를 수집하려면 다음 단계를 완료하세요.

1. SageMaker 콘솔의 탐색 창에서 노트북을 선택합니다.
2. aws-llm-apps-blog 노트북을 선택하고 Open JupyterLab 을 선택합니다.그림 4: Open JupyterLab
3. JupyterLab에서 data_ingestion_to_vectordb.ipynb 노트북을 오픈니다. 이 노트북은 llm_apps_workshop_embeddings라는 OpenSearch 서비스 인덱스에 SageMaker 문서를 수집합니다.
  그림 5: Data Ingestion 노트북 실행
4. 노트북이 열리면 실행 메뉴에서 Run All Cells을 선택해 이 노트북에서 코드를 실행합니다. 그러면 데이터 세트가 노트북에 로컬로 다운로드된 다음 OpenSearch 서비스 인덱스에 수집됩니다. 이 노트북을 실행하는 데 약 20분이 걸립니다. 또한 이 노트북은 데이터를 FAISS라는 다른 벡터 데이터베이스로 수집합니다. FAISS 인덱스 파일은 로컬에 저장되며, 대체 벡터 데이터베이스를 사용하는 예시로서 Lambda 함수에서 선택적으로 사용할 수 있도록 Amazon Simple Storage Service(S3)에 업로드됩니다.그림 6: 노트북 Run All Cells
  이제 문서를 청크로 분할할 준비가 되었으므로, 이를 임베딩으로 변환하여 OpenSearch로 수집할 수 있습니다. LangChain RecursiveCharacterTextSplitter 클래스를 사용해 문서를 청크로 분할한 다음, LangChain SagemakerEndpointEmbeddingsJumpStart 클래스를 사용해 GPT-J-6B LLM으로 이 청크를 임베딩으로 변환합니다. 임베딩은 LangChain OpenSearchVectorSearch 클래스를 통해 OpenSearch 서비스에 저장됩니다. 이 코드를 파이썬 스크립트로 패키징하여 사용자 정의 컨테이너를 통해 세이지메이커 프로세싱 잡에 제공합니다. 전체 코드는 data_ingestion_to_vectordb.ipynb 노트북을 참조하세요.
5. 사용자 정의 컨테이너를 생성한 다음, 그 안에 LangChain 및 opensearch-py 파이썬 패키지를 설치합니다.
6. 이 컨테이너 이미지를 Amazon Elastic Container Registry (ECR) 에 업로드합니다.
7. SageMaker ScriptProcessor 클래스를 사용하여 여러 노드에서 실행할 SageMaker Processing job을 생성합니다.
  - Processing job에 제공된 ProcessingInput의 일부로 s3_data_distribution_type=’ShardedByS3Key’를 설정하여 Amazon S3에서 사용 가능한 데이터 파일을 SageMaker Processing job 인스턴스 전체에 자동으로 배포합니다.
  - 각 노드는 파일의 하위 집합을 처리하므로 OpenSearch 서비스로 데이터를 수집하는 데 필요한 전체 시간이 단축됩니다.
  - 각 노드는 또한 내부적으로 파일 처리를 병렬화하기 위해 파이썬 멀티프로세싱을 사용합니다. 따라서 개별 노드가 작업(파일)을 서로 분산 처리하는 클러스터 레벨과 노드의 파일이 노드에서 실행되는 여러 프로세스로 분할 처리되는 노드 레벨에서 두 가지 레벨의 병렬화가 이뤄집니다.
```
# setup the ScriptProcessor with the above parameters
processor = ScriptProcessor(base_job_name=base_job_name,
                            image_uri=image_uri,
                            role=aws_role,
                            instance_type=instance_type,
                            instance_count=instance_count,
                            command=["python3"],
                            tags=tags)
# setup input from S3, note the ShardedByS3Key, this ensures that 
# each instance gets a random and equal subset of the files in S3.
inputs = [ProcessingInput(source=f"s3://{bucket}/{app_name}/{DOMAIN}",
                          destination='/opt/ml/processing/input_data',
                          s3_data_distribution_type='ShardedByS3Key',
                          s3_data_type='S3Prefix')]
logger.info(f"creating an opensearch index with name={opensearch_index}")
# ready to run the processing job
st = time.time()
processor.run(code="container/load_data_into_opensearch.py",
              inputs=inputs,
              outputs=[],
              arguments=["--opensearch-cluster-domain", opensearch_domain_endpoint,
                        "--opensearch-secretid", os_creds_secretid_in_secrets_manager,
                        "--opensearch-index-name", opensearch_index,
                        "--aws-region", aws_region,
                        "--embeddings-model-endpoint-name", embeddings_model_endpoint_name,
                        "--chunk-size-for-doc-split", str(CHUNK_SIZE_FOR_DOC_SPLIT),
                        "--chunk-overlap-for-doc-split", str(CHUNK_OVERLAP_FOR_DOC_SPLIT),
                        "--input-data-dir", "/opt/ml/processing/input_data",
                        "--create-index-hint-file", CREATE_OS_INDEX_HINT_FILE,
                        "--process-count", "2"])
```
8. 모든 코드 셀이 오류 없이 실행되면 노트북을 닫습니다. 이제 OpenSearch 서비스에서 데이터를 사용할 수 있습니다. 브라우저의 주소창에 다음 URL을 입력해 llm_apps_workshop_embeddings 인덱스에 있는 문서 수를 확인하세요. 아래 URL의 CloudFormation 스택 출력에서 OpenSearch 서비스 도메인 엔드포인트를 활용하세요. OpenSearch 서비스 사용자 이름과 비밀번호를 입력하라는 메시지가 표시될 것이며, 이는 CloudFormations 스택에서 확인 가능합니다.

https://your-opensearch-domain-endpoint/llm_apps_workshop_embeddings/_count

브라우저 창에 다음과 비슷한 출력이 표시되어야 합니다. 출력에 따르면 5,667개의 문서가 수집되었음을 보여줍니다.

 llm_apps_workshop_embeddings index. {"count":5667,"_shards":
{"total":5,"successful":5,"skipped":0,"failed":0}}

SageMaker Studio에서 Streamlit 애플리케이션 실행

이제 질문 답변 봇을 위해 Streamlit 웹 애플리케이션을 실행할 준비가 되었습니다. 이 애플리케이션을 사용하면 사용자가 질문을 한 다음 Lambda 함수가 제공하는/llm/rag REST API 엔드포인트를 통해 답변을 가져올 수 있습니다.
SageMaker Studio는 Streamlit 웹 애플리케이션을 호스팅할 수 있는 편리한 플랫폼을 제공합니다. 다음 단계는 Studio에서 Streamlit 앱을 실행하는 방법을 설명합니다. 또는 동일한 절차에 따라 노트북에서 앱을 실행할 수도 있습니다.

SageMaker Studio를 연 다음 새 터미널(new terminal)을 엽니다.
터미널에서 다음 커맨드를 실행하여 이 게시물의 코드 저장소를 복제하고 애플리케이션에 필요한 파이썬 패키지를 설치합니다.
```
git clone https://github.com/aws-samples/llm-apps-workshop
cd llm-apps-workshop/blogs/rag/app
pip install -r requirements.txt
```
CloudFormation 스택 출력에서 확인할 수 있는 API 게이트웨이 엔드포인트 URL을 webapp.py 파일에 설정해야 합니다. 이 작업은 다음 sed 커맨드를 실행하여 수행합니다. 셸 명령에서 replace-with-LLMAppAPIEndpoint-value-from-cloudformation-stack-outputs를 CloudFormation 스택 출력의 LLMAppAPIEndpoint 필드 값으로 바꾼 후 다음 명령을 실행하여 Studio에서 Streamlit 앱을 시작합니다.
```
EP=replace-with-LLMAppAPIEndpoint-value-from-cloudformation-stack-outputs
# replace __API_GW_ENDPOINT__ with output from the cloud formation stack
sed -i "s|__API_GW_ENDPOINT__|$EP|g" webapp.py
streamlit run webapp.py
```
애플리케이션이 성공적으로 실행되면 아래와 유사한 출력이 표시됩니다. (표시되는 IP 주소는 이 예제에 표시된 것과 다릅니다.) 출력에서 포트 번호(일반적으로 8501)를 기록하여 다음 단계의 앱 URL에 활용하세요.
```
sagemaker-user@studio$ streamlit run webapp.py

Collecting usage statistics. To deactivate, set browser.gatherUsageStats to False.

You can now view your Streamlit app in your browser.

Network URL: http://169.255.255.2:8501
External URL: http://52.4.240.77:8501
```
Studio 도메인 URL과 유사한 URL을 사용하여 새 브라우저 탭에서 앱에 액세스할 수 있습니다. 예를 들어 Studio URL이 https://d-randomidentifier.studio.us-east-1.sagemaker.aws/jupyter/default/lab? 인 경우 Streamlit 앱의 URL은 https://d-randomidentifier.studio.us-east-1.sagemaker.aws/jupyter/default/proxy/8501/webapp 입니다. (lab이 proxy/8501/webapp로 대체됩니다.) 이전 단계에서 확인된 포트 번호가 8501과 다른 경우 Streamlit 앱의 URL에 8501 대신 해당 포트 번호를 사용하세요.

다음 스크린샷은 앱을 통해 실행한 결과로 몇 개의 사용자 질문과 응답을 보여줍니다.

Lambda 함수의 RAG 구현 자세히 살펴보기

이제 애플리케이션이 엔드-투-엔드로 작동하도록 만들었으니 Lambda 함수에 대해 자세히 살펴보겠습니다. Lambda 함수는 FastAPI를 사용하여 RAG용 REST API를 구현하고 Mangum 패키지를 사용하여 함수에 패키징 및 배포하는 핸들러로 API를 감쌉니다. API 게이트웨이를 사용하여 함수를 호출하기 위해 들어오는 모든 요청을 라우팅하고 애플리케이션 내에서 내부적으로 라우팅을 처리합니다.

다음 코드 스니펫은 OpenSearch 인덱스에서 사용자 질문과 유사한 문서를 찾은 다음 질문과 유사한 문서를 결합하여 프롬프트를 생성하는 방법을 보여줍니다. 그런 다음 이 프롬프트는 사용자 질문에 대한 답변을 생성하기 위해 LLM에 전달됩니다.

@router.post("/rag")
async def rag_handler(req: Request) -> Dict[str, Any]:
    # dump the received request for debugging purposes
    logger.info(f"req={req}")

    # initialize vector db and SageMaker Endpoint
    _init(req)

    # Use the vector db to find similar documents to the query
    # the vector db call would automatically convert the query text
    # into embeddings
    docs = _vector_db.similarity_search(req.q, k=req.max_matching_docs)
    logger.info(f"here are the {req.max_matching_docs} closest matching docs to the query=\"{req.q}\"")
    for d in docs:
        logger.info(f"---------")
        logger.info(d)
        logger.info(f"---------")

    # now that we have the matching docs, lets pack them as a context
    # into the prompt and ask the LLM to generate a response
    prompt_template = """Answer based on context:\n\n{context}\n\n{question}"""

    prompt = PromptTemplate(
        template=prompt_template, input_variables=["context", "question"]
    )
    logger.info(f"prompt sent to llm = \"{prompt}\"")
    chain = load_qa_chain(llm=_sm_llm, prompt=prompt)
    answer = chain({"input_documents": docs, "question": req.q}, return_only_outputs=True)['output_text']
    logger.info(f"answer received from llm,\nquestion: \"{req.q}\"\nanswer: \"{answer}\"")
    resp = {'question': req.q, 'answer': answer}
    if req.verbose is True:
        resp['docs'] = docs

    return resp

리소스 정리

향후 요금이 부과되지 않으려면 리소스를 삭제하세요. 다음 스크린샷과 같이 CloudFormation 스택을 삭제하면 됩니다.
그림 7: 리소스 장리

결론

이 포스팅에서는 AWS 서비스, 오픈 소스 LLM 및 오픈 소스 파이썬 패키지의 조합을 사용하여 엔터프라이즈급 RAG 솔루션을 만드는 방법을 살펴보았습니다.

이 포스팅에서 제공된 샘플 구현과 비즈니스와 관련된 데이터 세트를 사용하여 JumpStart, Amazon Titan 모델, Amazon Bedrock 및 OpenSearch 서비스를 살펴보고 직접 솔루션을 구축하여 더 많은 것을 알아보시기 바랍니다.

AWS 기술 블로그