Amazon Bedrock과 Langchain Workshop

Generative AI on AWS

Bedrock의 기능적 특징

• 함수 하나로 여러 개의 모델을 호출하고 다양한 결과없을 얻을 수 있습니다.
• LangChain과 통합
• 고객 데이터는 사용지않고 보안을 유지 가능
• Private Link 지원 → TLS, 암호화, 보안까지 모두 지원합니다.

 

프롬프트 엔지니어링 기법

AI가 수행해야하는 작업을 설명하는 자연어 텍스트를 프롬프트라고 합니다.

 

프롬프트에 따라 성능이 달라지며, 경험적 과학입니다.

• 프롬프트를 계속 훈련하고 살을 붙여야합니다.

Prompt Engineering을 사용하여 개선 가능

• Human Assistant
• 명확하고 구체적인 작업 지시가 필요
• XML 태그를 자주 사용할 것
• 예제 사용
• 단계별로 생각하기
• 역할 할당하기
• 데이터와 지시 분리하기

 

RAG

• 학습 데이터에 의존적, 기한이 지난 정보, 환각현상같은 문제가 존재하여 검색 증강시킨 것 입니다.

모델 성능을 높히기 위한 한 가지 방법으로 도메인에 최적화된 LLM 모델을 만들 수 있습니다.

 

prompt 엔지니어링

RAG → 파운데이션 모델에 모댈 자료 입력

사전훈련된 파운데이션 모델 파인튜닝

 

데이터 추출, 청크 단위로 분할 → 벡터 임베딩 생성 → 데이터 적재 → 데이터 활용

 

 

hands-on

 

boto3

bedrock_model_id = "anthropic.claude-v2:1" #파운데이션 모델 설정

prompt = "Human:뉴햄프셔에서 가장 큰 도시가 어디인가요? Assistant:" #모델에 보낼 프롬프트 설정

body = json.dumps({
    "prompt": prompt, #claude
    "max_tokens_to_sample": 1024, 
    "temperature": 0, 
    "top_p": 0.5, 
    "stop_sequences": [], # 추론 결과와 관련되어 있습니다.
}) #요청 payload 설정

 

 

Langchain

이전 실습에서 Boto3를 통해 Amazon Bedrock을 호출했던 것과 기능은 비슷하지만, 이번 실습에서는 두 가지 접근 방식을 비교할 수 있도록 LangChain을 사용합니다. 특히 텍스트 입력 및 텍스트 출력에 집중하고 싶을 때 LangChain을 사용하면, Boto3 클라이언트 사용의 많은 세부 사항을 추상화할 수 있습니다.

 

Boto3 클라이언트가 제공하는 모든 제어 기능이 필요한 경우, 언제든지 선택하실 수 있습니다. Boto3를 사용할 경우 더 많은 코드가 필요할 수 있지만, JSON 요청 및 응답 객체에 대한 전체 액세스 권한을 제공받게 됩니다.

import os
from langchain.llms.bedrock import Bedrock

llm = Bedrock( #Bedrock llm 클라이언트 생성
    credentials_profile_name=os.environ.get("BWB_PROFILE_NAME"), #AWS 자격 증명에 사용할 프로필 이름 설정 (기본값이 아닌 경우)
    region_name=os.environ.get("BWB_REGION_NAME"), #리전 설정 (기본값이 아닌 경우)
    endpoint_url=os.environ.get("BWB_ENDPOINT_URL"), #엔드포인트 URL 설정 (필요한 경우)
    model_id="anthropic.claude-v2:1" #파운데이션 모델 설정
)

prompt = "Human:대한민국에서 가장 큰 도시가 어디인가요? 한 문장으로 말해주세요. Assistant:" 

response_text = llm.predict(prompt) #프롬프트에 응답 반환

print(response_text)

 

 

추론 매개변수

모듈마다 달라질 수 있으니 주의할 것

import sys
import os
from langchain.llms.bedrock import Bedrock

# 추론 매개변수를 가져옴
def get_inference_parameters(model): #모델의 공급자에 따라 기본 매개변수 집합 반환
    bedrock_model_provider = model.split('.')[0] #모델 ID의 첫 번째 부분에서 모델 공급자 가져오기
    
    if (bedrock_model_provider == 'anthropic'): #Anthropic 모델
        return { #anthropic
            "max_tokens_to_sample": 512,
            "temperature": 0, 
            "top_k": 250, 
            "top_p": 1, 
            "stop_sequences": ["\n\nHuman:"] 
           }
    
    elif (bedrock_model_provider == 'ai21'): #AI21 모델
        return { #AI21
            "maxTokens": 512, 
            "temperature": 0, 
            "topP": 0.5, 
            "stopSequences": [], 
            "countPenalty": {"scale": 0 }, 
            "presencePenalty": {"scale": 0 }, 
            "frequencyPenalty": {"scale": 0 } 
           }
    
    elif (bedrock_model_provider == 'cohere'): #COHERE 모델
        return {
            "max_tokens": 512,
            "temperature": 0,
            "p": 0.01,
            "k": 0,
            "stop_sequences": [],
            "return_likelihoods": "NONE"
        }
    
    else: #Amazon
        #LangChain Bedrock 구현의 경우, 이러한 매개변수는 LangChain이 생성하는 textGenerationConfig 구성항목에 추가됩니다.
        return { 
            "maxTokenCount": 512, 
            "stopSequences": [], 
            "temperature": 0, 
            "topP": 0.9 
        }


def get_text_response(model, input_content): #text-to-text 클라이언트 함수
    
    model_kwargs = get_inference_parameters(model) #선택한 모델에 따른 기본 매개변수 할당
    
    llm = Bedrock( #Bedrock llm 클라이언트 생성
        credentials_profile_name=os.environ.get("BWB_PROFILE_NAME"), #AWS 자격 증명에 사용할 프로필 이름 설정 (기본값이 아닌 경우)
        region_name=os.environ.get("BWB_REGION_NAME"), #리전 설정 (기본값이 아닌 경우)
        endpoint_url=os.environ.get("BWB_ENDPOINT_URL"), #엔드포인트 URL 설정 (필요한 경우)
        model_id=model,
        model_kwargs = model_kwargs
    )
    
    return llm.predict(input_content) #프롬프트에 대한 응답 반환

response = get_text_response(sys.argv[1], sys.argv[2])

print(response)

 

응답 가변성 제어

temperature에 따른 response 차이

import sys
import os
from langchain.llms.bedrock import Bedrock

def get_text_response(input_content, temperature): #text-to-text client 함수
    
    model_kwargs = {
        "max_tokens_to_sample": 1024, 
        "temperature": temperature, 
        "top_p": 0.5, 
        "stop_sequences": [], 
    }
    
    llm = Bedrock( #Bedrock llm 클라이언트 생성
    credentials_profile_name=os.environ.get("BWB_PROFILE_NAME"), #AWS 자격 증명에 사용할 프로필 이름 설정 (기본값이 아닌 경우)
    region_name=os.environ.get("BWB_REGION_NAME"), #리전 설정 (기본값이 아닌 경우)
    endpoint_url=os.environ.get("BWB_ENDPOINT_URL"), #엔드포인트 URL 설정 (필요한 경우)
    model_id="anthropic.claude-v2:1", #파운데이션 모델 설정
    model_kwargs = model_kwargs
    )
    
    return llm.predict(input_content) #프롬프트에 대한 응답 반환

for i in range(3):
    response = get_text_response(sys.argv[1], float(sys.argv[2]))
    print(response)

 

스트리밍 데이터

유저들에게 빠른 응답을 위해 스트리밍 방식으로 사용됨.

한 번에 모든 텍스트를 전달하는게 아닌 잘라서 리턴

import os
import json
import boto3

session = boto3.Session(
    profile_name=os.environ.get("BWB_PROFILE_NAME")
) #AWS 자격 증명에 사용할 프로필 이름 설정

bedrock = session.client(
    service_name='bedrock-runtime', #Bedrock 클라이언트 생성
    region_name=os.environ.get("BWB_REGION_NAME"),
    endpoint_url=os.environ.get("BWB_ENDPOINT_URL")
)

def chunk_handler(chunk):
    print(chunk, end='')

def get_streaming_response(prompt, streaming_callback):

    bedrock_model_id = "anthropic.claude-v2" #파운데이션 모델 설정

    body = json.dumps({
        "prompt": prompt, #ANTHROPIC
        "max_tokens_to_sample": 4000,
        "temperature": 0, 
        "top_k": 250, 
        "top_p": 1, 
        "stop_sequences": ["\n\nHuman:"] 
    })

    response = bedrock.invoke_model_with_response_stream(modelId=bedrock_model_id, body=body) #스트리밍 메서드 호출
    stream = response.get('body')
    if stream:
        for event in stream: #스트림에서 반환된 각 이벤트 처리
            chunk = event.get('chunk')
            if chunk:
                chunk_json = json.loads(chunk.get('bytes').decode())
                streaming_callback(chunk_json["completion"]) #콜백 메서드에 최신 청크의 텍스트 전달

prompt = "\n\nHuman:가장 친한 친구가 된 강아지 두 마리와 새끼 고양이 두 마리에 대한 이야기를 들려주세요.\n\nAssistant:"
                
get_streaming_response(prompt, chunk_handler)

 

임베딩

텍스트 데이터를 벡터로 만들 때 사용합니다.

벡터로 만드는 이유는 문장 간 유사도를 검색하기 위함이며, 연산 자체를 빠르게 진행할 수 있습니다.

from langchain.embeddings import BedrockEmbeddings
from numpy import dot
from numpy.linalg import norm # 정규화를 위해 임포트 하였습니다.

#Bedrock Embeddings LangChain 클라이언트 생성
belc = BedrockEmbeddings()

class EmbedItem:
    def __init__(self, text):
        self.text = text
        self.embedding = belc.embed_query(text)

class ComparisonResult:
    def __init__(self, text, similarity):
        self.text = text
        self.similarity = similarity

def calculate_similarity(a, b): # 코사인 유사도를 확인하세요: https://en.wikipedia.org/wiki/Cosine_similarity
    return dot(a, b) / (norm(a) * norm(b))

#비교할 임베딩 목록을 작성합니다.
#다양한 언어로 구성된 문장에 유사도를 비교
items = []

with open("items.txt", "r") as f:
    text_items = f.read().splitlines()

for text in text_items:
    items.append(EmbedItem(text))


for e1 in items:
    print(f"Closest matches for '{e1.text}'")
    print ("----------------")
    cosine_comparisons = []
    
    for e2 in items:
        similarity_score = calculate_similarity(e1.embedding, e2.embedding)
        
        cosine_comparisons.append(ComparisonResult(e2.text, similarity_score)) # 비교 내용을 목록에 저장합니다.
        
    cosine_comparisons.sort(key=lambda x: x.similarity, reverse=True) # 가장 가까운 일치 항목을 먼저 나열합니다.
    
    for c in cosine_comparisons:
        print("%.6f" % c.similarity, "\t", c.text)
    
    print()

 

Closest matches for 'Can you please tell me how to get to the bakery?'
----------------
1.000000         Can you please tell me how to get to the bakery?
0.712236         I need directions to the bread shop
0.541959         Pouvez-vous s'il vous plaît me dire comment me rendre à la boulangerie?
0.492384         빵집으로 가는 길을 알려주세요.
0.484672         Can you please tell me how to get to the stadium?
0.455479         パン屋への行き方を教えてください
0.406388         パン屋への道順を知りたい
0.369163         Kannst du mir bitte sagen, wie ich zur Bäckerei komme?
0.238000         경기장 가는 방법을 알려주시겠어요?
0.109972         고양이, 개, 쥐
0.078357         猫、犬、ネズミ
0.022138         Cats, dogs, and mice
0.015661         Lions, tigers, and bears
0.005211         Chats, chiens et souris
-0.007595        Felines, canines, and rodents

 

Streamlit

streamlit docs 참고

 

검색 증강 생성 (RAG)

1. 참조해야할 데이터를 임베딩 시켜 벡터 데이터베이스에 저장합니다.
2. 코사인 유사도를 통해 유사한 걸 가져옵니다.
3. Question과 유사도를 통해 가져온 컨텍스트를 파운데이션 모델에 넣어 response를 얻습니다.

 

import os
from langchain.embeddings import BedrockEmbeddings
from langchain.indexes import VectorstoreIndexCreator
from langchain.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyPDFLoader
from langchain.llms.bedrock import Bedrock
#임베딩, 인덱스, 벡터스토어, 텍스트 청크를 위한 스플리터, pdf에서 추출을 위한 pdf loader, bedrock 클래스

def get_llm():
    
    model_kwargs = { #Anthropic Cloude v2.1
        "max_tokens_to_sample": 1024, 
        "temperature": 0, 
        "top_p": 0.5, 
    }
    
    llm = Bedrock(
        credentials_profile_name=os.environ.get("BWB_PROFILE_NAME"), # AWS 자격 증명에 사용할 프로필 이름 설정(기본값이 아닌 경우)
        region_name=os.environ.get("BWB_REGION_NAME"), # 리전 이름 설정(기본값이 아닌 경우)
        endpoint_url=os.environ.get("BWB_ENDPOINT_URL"), #endpoint URL 설정(필요한 경우)
        model_id="anthropic.claude-v2:1", #파운데이션 모델 설정
        model_kwargs=model_kwargs) #Claude에 대한 속성
    
    return llm

def get_index(): #애플리케이션에서 사용할 인메모리 벡터 저장소를 생성하고 반환합니다
    
    embeddings = BedrockEmbeddings(
        credentials_profile_name=os.environ.get("BWB_PROFILE_NAME"), #기본값이 아닌 경우 AWS 자격 증명에 사용할 프로필 이름을 설정합니다
        region_name=os.environ.get("BWB_REGION_NAME"), #리전 이름 설정(기본값이 아닌 경우)
        endpoint_url=os.environ.get("BWB_ENDPOINT_URL"), #endpoint URL 설정(필요한 경우)
    ) #Titan Embeddings 클라이언트 생성
    
    pdf_path = "2022-Shareholder-Letter-ko.pdf" #로컬 PDF 파일 (2022년 주주 서한 - 한글)

    loader = PyPDFLoader(file_path=pdf_path) #pdf 파일을 로드합니다
    
    #텍스트 청크화.
    # overlap은 이전 청크와 겹쳐 만든다.
    text_splitter = RecursiveCharacterTextSplitter( #텍스트 분할기를 만듭니다
        chunk_size=1000, #구분 기호를 사용하여 1000자 청크로 분할
        chunk_overlap=100 #이전 청크와 겹칠 수 있는 문자 수
    )
    
    # 벡터 스토어(FAISS는 인메모리 벡터 DB)
    index_creator = VectorstoreIndexCreator( #벡터 스토어 팩토리 생성
        vectorstore_cls=FAISS, #데모 목적으로 인메모리 벡터 스토어 사용
        embedding=embeddings, #Titan Embeddings 사용
        text_splitter=text_splitter, #재귀적 텍스트 분할기 사용
    )
    
    # pdf에서 벡터 스토어 인덱스들이 생성되어 반환됩니다.
    index_from_loader = index_creator.from_loaders([loader]) #로드된 PDF에서 벡터 스토어 인덱스 생성
    
    return index_from_loader #클라이언트 앱에서 캐시할 인덱스 반환

def get_rag_response(index, question): #rag client 함수
    
    llm = get_llm()
    
    response_text = index.query(question=question, llm=llm) #인메모리 인덱스에 대해 검색하고 결과를 프롬프트에 채워서 llm으로 전송
    
    return response_text

 

RAG를 사용한 챗봇

1. 지난 상호작용이 채팅 메모리 객체에서 추적됩니다.
2. 사용자가 새 메시지를 입력합니다.
3. 채팅 기록이 메모리 객체에서 검색되어 새 메시지 앞에 추가됩니다.
4. 타이탄 임베딩을 사용하여 질문을 벡터로 변환한 다음 벡터 데이터베이스에서 가장 가까운 벡터와 매치합니다.
5. 결합된 기록, 지식 및 새 메시지가 모델로 전송됩니다.
6. 모델의 응답이 사용자에게 표시됩니다.

 

 

문서 요약

큰 문서를 작은 청크 단위로 나누고 중간 요약을 만듭니다. 이렇게 만들어진 중간 요약들을 다시 결합하여 요약된 결과를 리턴합니다.

import os
from langchain.prompts import PromptTemplate
from langchain.llms.bedrock import Bedrock

# 문서 요약
from langchain.chains.summarize import load_summarize_chain
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyPDFLoader

def get_llm():
    
    model_kwargs =  { #Anthropic 모델
        "max_tokens_to_sample": 8000,
        "temperature": 0, 
        "top_k": 250, 
        "top_p": 0.5, 
        "stop_sequences": ["\n\nHuman:"] 
    }
    
    llm = Bedrock(
        credentials_profile_name=os.environ.get("BWB_PROFILE_NAME"), #AWS 자격 증명에 사용할 프로필 이름을 설정합니다(기본값이 아닌 경우)
        region_name=os.environ.get("BWB_REGION_NAME"), #리전 이름을 설정합니다(기본값이 아닌 경우)
        endpoint_url=os.environ.get("BWB_ENDPOINT_URL"), #엔드포인트 URL 설정(필요한 경우)
        model_id="anthropic.claude-v2:1", #파운데이션 모델 설정하기
        model_kwargs=model_kwargs) #Claude의 속성을 구성합니다.
    
    return llm


pdf_path = "uploaded_file.pdf"

def get_example_file_bytes(): #사용자가 기존 생성된 예제를 다운로드할 수 있도록 파일 바이트를 제공합니다.
    with open("2022-Shareholder-Letter-ko.pdf", "rb") as file:
        file_bytes = file.read()
    
    return file_bytes


def save_file(file_bytes): #업로드한 파일을 디스크에 저장하여 나중에 요약합니다.   
    with open(pdf_path, "wb") as f: 
        f.write(file_bytes)
    
    return f"Saved {pdf_path}"


def get_docs():
    
    loader = PyPDFLoader(file_path=pdf_path)
    documents = loader.load()
    text_splitter = RecursiveCharacterTextSplitter(
        separators=["\n\n", "\n", ".", " "], chunk_size=4000, chunk_overlap=100 
    )
    docs = text_splitter.split_documents(documents=documents)
    
    return docs

def get_summary(return_intermediate_steps=False):
    
    map_prompt_template = "{text}\n\n위의 내용을 Korean으로 bullet point 3개로 요약합니다:"
    map_prompt = PromptTemplate(template=map_prompt_template, input_variables=["text"])
    
    combine_prompt_template = "{text}\n\n위의 내용을 Korean으로 간결하게 bullet point 5개로 요약합니다:"
    combine_prompt = PromptTemplate(template=combine_prompt_template, input_variables=["text"])
    
    llm = get_llm()
    docs = get_docs()
    
    chain = load_summarize_chain(llm, chain_type="map_reduce", map_prompt=map_prompt, combine_prompt=combine_prompt, return_intermediate_steps=return_intermediate_steps,verbose=True)
    
    if return_intermediate_steps:
        return chain({"input_documents": docs}, return_only_outputs=True) #반환 구조를 chain.run(docs)와 일관성 있게 만들기
    else:
        return chain.run(docs)

 

import streamlit as st
import summarization_lib_kr as glib

st.set_page_config(layout="wide", page_title="문서 요약")
st.title("문서 요약")

uploaded_file = st.file_uploader("Select a PDF", type=['pdf'])

upload_button = st.button("Upload", type="primary")

if upload_button:
    with st.spinner("Uploading..."):

        upload_response = glib.save_file(file_bytes=uploaded_file.getvalue())

        st.success(upload_response)
        
        st.session_state.has_document = True

if 'has_document' in st.session_state: #문서가 업로드되었는지 확인하기
    
    return_intermediate_steps = st.checkbox("중간 단계 요약 보기", value=True)
    summarize_button = st.button("요약하기", type="primary")
    
    
    if summarize_button:
        st.subheader("통합 요약")

        with st.spinner("Running..."):
            response_content = glib.get_summary(return_intermediate_steps=return_intermediate_steps)


        if return_intermediate_steps:

            st.write(response_content["output_text"])

            st.subheader("중간 단계 요약")

            for step in response_content["intermediate_steps"]:
                st.write(step)
                st.markdown("---")

        else:
            st.write(response_content)

응답 스트리밍

모든 내용을 한 번에 반환하는 게 아닌 중간 중간 계속해서 응답을 반환합니다.

 

임베딩 검색

실제 시나리오에서는 OpenSearch Serverless를 사용해서 배포합니다.

1. 문서가 텍스트 청크로 분할됩니다. 이 청크는 Titan Embedding으로 전달되어 벡터로 변환됩니다. 그런 다음 벡터는 벡터 데이터베이스에 저장됩니다.
2. 사용자가 질문을 제출합니다.
3. 질문은 Amazon Titan Embeddin을 사용하여 벡터로 변환된 다음 벡터 데이터베이스에서 가장 가까운 벡터와 매칭됩니다.
4. 일치하는 벡터에서 결합된 콘텐츠가 사용자에게 반환됩니다.

 

개인 맞춤형 추천

일반적인 컨텐츠를 임베딩시켜 두었다가 컨텐츠에 대한 요청이 들어왔을 때 BedRock을 통해 개인화된 추천을 해주는 방식입니다.

 

Text to JSON 데이터 추출하기

고객과 대화하면서 감정 분석을 위해 사용합니다.

또한 콜센터에서 상담을 하며 요약할 때 고객의 정보와 CS 내용 등을 분석하고 정리할 수 있습니다.

 

 

Text to CSV 데이터 추출하기

데이터 전처리 혹은 비정혀 데이터를 처리하는데 사용합니다.

프롬프트 엔지니어링

프롬프트 엔지니어링 이란 사용자의 요청에 대한 파운데이션 모델의 응답 품질과 성능을 최적화하는 작업입니다.

• 단어 선택
• 문구
• 예제 제공(few-shot learning)
• 줄 바꿈 및 콘텐츠 구분 기호 사용
• 모델이 학습했던 방식과 일치하는 정해진 형식 따르기
• 모델이 텍스트 생성을 중단해야 하는 시점을 알 수 있도록 stop sequences 사용

 

프롬프트 엔지니어링 사용 사례

• 개요 작성
• 초안 작성
• 수정 및 재작성
• 축약된 콘텐츠
• 콘텐츠에 대한 간략한 설명
• 구조화 된 요약
• 사용자 관심사에 기반한 맞춤형 요약
• 분류
• 감정 분석
• 이메일 회신
• 권장 조치
• 키워드 추출
• 몇 가지 예제를 제공(Few-shot prompting)
• Human: / Assistant:
• 서문 제거하기
• 서문 건너뛰기
• XML 태그에 출력