Cloud

aws Lex, Lambda, Bedrock을 이용해서 RAG chatbot 만들기(1. Lambda code)

jinmc 2025. 1. 21. 15:16
반응형

aws에서 제공한 lex, lambda, bedrock을 이용한 RAG chatbot 만들기 중 Lambda  코드에 대해서 포스팅을 하려고 합니다. 시작에 앞서서 각각의 서비스들을 설명해 보자면,

 

1. AWS Lex

  • 챗봇 서비스: 사용자의 입력(텍스트/음성)을 이해하고 대화를 관리합니다.
  • 구성 요소:
    • Intent: 사용자의 의도를 정의. 예) 병원 찾기.
    • Slots: 추가로 입력받아야 할 정보. 예) 병원 이름, 위치.

2. AWS Lambda

  • 서버리스 함수: Lex가 요청한 작업(데이터 처리, 외부 API 호출 등)을 처리합니다.
  • 특징:
    • Lex에서 받은 데이터를 처리하고, Bedrock이나 DB와 연동.

3. AWS Bedrock

  • 생성형 AI 서비스: AI 모델(GPT, Claude 등)을 사용해 자연어 답변 생성.
  • Knowledge Base:
    • 데이터 검색 후 답변 생성.
    • 예) 병원 위치 데이터 검색.

 

그럼 셋의 연동은 어떤 식으로 하면 될까요?

 

Lex, Lambda, Bedrock의 연동

  1. Lex: 사용자의 요청(Intent, Slots)을 분석.
  2. Lambda: Lex에서 받은 데이터를 처리하고, Bedrock 호출.
  3. Bedrock: 데이터 검색 또는 AI 답변 생성.
  4. 결과: Lambda → Lex → 사용자.

 

라고 볼 수 있을 것 같습니다. 비유를 하자면, Lex가 frontend, bedrock이 데이터베이스, 그리고 lambda가 그 사이를 이어주는 backend 라고 볼 수 있을 것 같습니다. 

 

그럼 code를 하나하나 보도록 하겠습니다.

 

1. 주요 라이브러리와 클라이언트

import json
import boto3
from decimal import Decimal
 
  • json: JSON 데이터 처리에 사용됩니다.
  • boto3: AWS 서비스와 상호작용하기 위한 라이브러리입니다. 여기서는 Bedrock 관련 클라이언트 bedrock-runtime과 bedrock-agent-runtime을 사용합니다.
  • Decimal: AWS DynamoDB처럼 소수점 데이터를 처리할 때 주로 사용되며, 이를 JSON 직렬화하기 위한 유틸리티 함수도 있습니다.

2. Slot 관련 함수

elicit_slot

  • 특정 슬롯에 대한 값을 요청할 때 사용합니다.

confirm_intent

  • 사용자의 입력을 바탕으로 의도를 확인합니다.

close

  • 대화가 끝났을 때 호출하며 Fulfillment 상태를 반환합니다.
 

3. Bedrock 관련 함수

get_bedrock_knowledge_base_response

  • Bedrock의 Knowledge Base에서 데이터를 검색합니다.
  • 주어진 쿼리를 Knowledge Base에 전달하고, 최대 5개의 결과를 반환합니다.
  • 결과는 텍스트 기반의 간단한 구조로 변환됩니다.

get_bedrock_response

  • 사용자의 질문과 검색된 결과를 기반으로 Bedrock에서 답변을 생성합니다.
  • Claude 모델을 호출하여 결과를 생성하며, 결과는 JSON으로 반환됩니다.
  • Prompt 구성: 대화 문맥과 검색된 정보를 포함한 세부적인 프롬프트가 작성됩니다.

4. 주요 비즈니스 로직

Bedrock_intent

  • AWS Lex가 호출한 의도(Intent)에 대한 처리를 수행합니다.
  • 주요 흐름:
    1. 슬롯 값 확인: 특정 슬롯(HospitalInquiry)이 비어 있으면 값을 요청합니다.
    2. 사용자 입력 확인: 사용자가 의도를 확정(Confirmed)하거나 거부(Denied)했는지 확인합니다.
    3. Knowledge Base 검색: 사용자의 질문을 기반으로 데이터를 검색하고 답변을 생성합니다.
    4. 응답 반환: 생성된 답변을 사용자에게 전달하고, 필요한 경우 대화를 종료합니다.

lambda_handler

  • AWS Lambda가 실행될 때 진입점으로 호출되는 함수입니다.
  • Lex로부터 전달된 이벤트를 처리하며, Intent에 따라 Bedrock_intent를 호출합니다.

5. 실행 흐름

  1. 사용자가 Lex를 통해 메시지를 전송합니다.
  2. Lex는 이를 Lambda에 전달하고, Lambda는 이벤트를 기반으로 적절한 Intent를 호출합니다.
  3. Intent에서:
    • 사용자의 입력을 처리하고,
    • 필요한 데이터를 검색(Bedrock)하거나,
    • 대화 상태를 업데이트하고,
    • 최종적으로 응답을 반환합니다.

 

실제 코드:

 

import json
import boto3
from decimal import Decimal

bedrock_runtime_client = boto3.client('bedrock-runtime')
bedrock_agent_runtime_client = boto3.client('bedrock-agent-runtime')

def decimal_to_int(obj):
    if isinstance(obj, Decimal):
        return int(obj)

def elicit_slot(slot_to_elicit, intent_name, slots, session_attributes):
    return {
        'sessionState': {
            'dialogAction': {
                'type': 'ElicitSlot',
                'slotToElicit': slot_to_elicit,
            },
            'intent': {
                'name': intent_name,
                'slots': slots,
                'state': 'InProgress'
            },
            'sessionAttributes': session_attributes
        }
    }

def confirm_intent(message_content, intent_name, slots, session_attributes):
    return {
        'messages': [{'contentType': 'PlainText', 'content': message_content}],
        'sessionState': {
            'dialogAction': {
                'type': 'ConfirmIntent',
            },
            'intent': {
                'name': intent_name,
                'slots': slots,
                'state': 'Fulfilled'
            },
            'sessionAttributes': session_attributes
        }
    }

def close(fulfillment_state, message_content, intent_name, slots, session_attributes):
    return {
        'messages': [{'contentType': 'PlainText', 'content': message_content}],
        "sessionState": {
            'dialogAction': {
                'type': 'Close',
            },
            'intent': {
                'name': intent_name,
                'slots': slots,
                'state': fulfillment_state
            },
            'sessionAttributes': session_attributes
        }
    }

def get_bedrock_knowledge_base_response(query_text, knowledgeBaseId):
    print("GET BEDROCK KNOWLEDGE BASE RESPONSE")
    
    try:
        response = bedrock_agent_runtime_client.retrieve(
            knowledgeBaseId=knowledgeBaseId,
            retrievalQuery={
                'text': query_text
            },
            retrievalConfiguration={
                'vectorSearchConfiguration': {
                    'numberOfResults': 5
                }
            }
        )
        
        results = response['retrievalResults'][:5] if response['retrievalResults'] else []
        
        retrieved_results = []
        
        for item in results:
            content = item.get('content', {}).get('text', '')
            retrieved_results.append({
                'Content': content
            })
        
        return retrieved_results
        
    except Exception as e:
        print(f"Error retrieving from Knowledge Base: {str(e)}")
        return []

def get_bedrock_response(user_prompt, retrieved_results, conversation_context):
    prompt = f"""\n\nHuman: 아래 [참고] 정보를 바탕으로 [질문]에 적절히 답해주세요. 
    200자 이내로 대답해주세요. 대답할 때 200자 이내라고 하지 않으셔도 됩니다. 완전한 문장으로 답변해주세요.
    
    이전 대화 내용:
    {json.dumps(conversation_context, ensure_ascii=False)}
    
    [질문]
    {user_prompt}
    [참고]
    {json.dumps(retrieved_results, ensure_ascii=False)}
    Assistant: """

    modelId = 'anthropic.claude-3-sonnet-20240229-v1:0'
    accept = 'application/json'
    contentType = 'application/json'

    body = json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 300,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.0,
    })

    response = bedrock_runtime_client.invoke_model(
        modelId=modelId,
        accept=accept,
        contentType=contentType,
        body=body
    )

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

def Bedrock_intent(event):
    intent_name = event['sessionState']['intent']['name']
    slots = event['sessionState']['intent']['slots']
    session_attributes = event['sessionState'].get('sessionAttributes', {})
    user_prompt = event['inputTranscript']

    if slots['HospitalInquiry'] is None:
        return elicit_slot('HospitalInquiry', intent_name, slots, session_attributes)

    confirmation_status = event['sessionState']['intent']['confirmationState']

    if confirmation_status == "Confirmed":
        return close("Fulfilled", '그럼 통화를 종료하겠습니다.', intent_name, slots, session_attributes)
    elif confirmation_status == "Denied":
        return close("Failed", '올바른 정보를 못 찾아드려서 죄송합니다.', intent_name, slots, session_attributes)

    knowledgeBaseId = '생성한 Bedrock의 Knowledge Bases 입력'
    
    conversation_context = {
        "intent_name": intent_name,
        "slots": slots,
        "session_attributes": session_attributes
    }
    
    retrieved_results = get_bedrock_knowledge_base_response(user_prompt, knowledgeBaseId)
    response_text = get_bedrock_response(user_prompt, retrieved_results, conversation_context)
    
    # Update session attributes with the latest response
    session_attributes['last_response'] = response_text
    
    return confirm_intent(
        f'{response_text}',
        intent_name, slots, session_attributes)

def lambda_handler(event, context):
    print("Received event:" + json.dumps(event, default=decimal_to_int, ensure_ascii=False))

    intent_name = event['sessionState']['intent']['name']

    if intent_name == 'selfAction':
        return Bedrock_intent(event)
반응형

'Cloud' 카테고리의 다른 글

gcp cloudsql 과 mysqlbench 연결하기  (1) 2025.01.02

'Cloud'의 다른글

  • 현재글aws Lex, Lambda, Bedrock을 이용해서 RAG chatbot 만들기(1. Lambda code)

관련글

댓글

티스토리툴바