Amazon Bedrock 및 Strands Agents를 이용한 롯데백화점의 AI 컨시어지 구축기

# OpenSearch 쿼리 빌더 함수 예시 
def _build_opensearch_query(
    branch_name: Optional[str] = None,
    ...
) -> Dict[str, Any]:
    must_conditions = []
    must_conditions.append({
        # 지점 관련 질문의 경우 해당 지점의 데이터만 검색하도록 제한
        "term": {"metadata.branch_name": branch_name}
    })
...

# 시스템 프롬프트 예시
<tool_choice_policy>
# 예시 도구명
<get_shopping_info>
Tool to get information related to promotions, coupons, and culture event.
Example keywords include: "세일", "쿠폰", "사은", "문화 이벤트", "샵나우" ...
Lotte Department Store refers to the contents containing shopping promotion information as "쇼핑 정보".
</get_shopping_info>
</tool_choice_policy>

from typing import Literal
from strands import tool
from strands.types.tools import ToolContext

# 예시 카테고리
CategoryLiteral = Literal[
    "레져/스포츠", "패션 액세서리", "남성패션", "여성패션", 
    "아동/유아", "가전", "주방/식기", "화장품", None
]

# 예시 도구
@tool(context=True)
async def get_facility_info(
    tool_context: ToolContext, 
    recommended_category: CategoryLiteral
    ...
) -> dict:
    """
    롯데백화점 내 브랜드 매장, F&B, 편의시설을 검색합니다.
    
    Args:
        tool_context: Strands Agents tool context
        recommended_category: 추천 카테고리 (특정 매장명이 아닌 경우 활용)
        ...
    """
    branch_code = tool_context.agent.state.get("branch_code")
    branch_name = tool_context.agent.state.get("branch_name")
    
    ...
    
    # 특정 매장명이 없고 추천 카테고리가 있는 경우
    # 해당 카테고리의 매장/브랜드를 검색하여 추천
    recommendation_keyword = recommended_category
    
    # Opensearch 검색 함수
    search_results = await find_facility_by_branch(
        query=query,
        branch_code=branch_code,
        branch_name=branch_name,
        keyword=recommendation_keyword
    )
    
    ...

# 예시 Lambda 코드
from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType, batch_processor
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord

processor = BatchProcessor(event_type=EventType.SQS)

@batch_processor(record_handler=process_record, processor=processor)
def lambda_handler(event, context):
    """Lambda 핸들러 진입점"""
    logger.info(f"Embedder function invoked with {len(event.get('Records', []))} SQS records.")
    return processor.response()

# 예시 레코드 처리 함수
def process_record(record: SQSRecord):
    """SQS 레코드 처리 - S3 이벤트 알림 파싱"""
    message_body = json.loads(record.body)

    for s3_record in message_body.get('Records', []):
        bucket = s3_record['s3']['bucket']['name']
        key = urllib.parse.unquote_plus(
            s3_record['s3']['object']['key'],
            encoding='utf-8'
        )

        # 파일명에서 데이터 유형과 버전 정보 추출
        source_parts = key.split("_")
        source_type = source_parts[1]  # 'store', 'branch', 'faq' 등
        source_date = source_parts[2] + source_parts[3].split('.')[0]  # 버전 관리

        # 데이터 처리 로직 수행
        content = get_s3_object_content(bucket, key)
        chunk_objects = chunk_content(client, INDEX_NAME, content, bucket, key)
        ...

import concurrent.futures

bedrock = boto3.client("bedrock-runtime", region_name="ap-northeast-2")

# 멀티스레딩을 활용한 병렬 임베딩 생성 예시
def get_embedding(text: str) -> Optional[list[float]]:
    """Bedrock Titan으로 텍스트 임베딩 생성"""
    if not text:
        return None

    response = bedrock.invoke_model(
        modelId=EMBEDDING_MODEL_ARN,
        body=json.dumps({"inputText": text}),
        contentType='application/json',
        accept='application/json'
    )
    response_body = json.loads(response["body"].read())
    return response_body.get("embedding")

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    embedding_futures = {
        executor.submit(get_embedding, chunk["text"]): i
        for i, chunk in enumerate(batch_chunks)
    }

    embeddings = {}
    for future in concurrent.futures.as_completed(embedding_futures):
        idx = embedding_futures[future]
        embedding = future.result()
        if embedding:
            embeddings[idx] = embedding

# 과거 버전 문서 삭제 함수 예시
def delete_old_versions(source_type: str, source_date: str):
    """이전 버전 문서 삭제 - 증분 처리"""

    search_query_body = {
        "query": {
            "bool": {
                "must": [
                    {"term": {"source_type": source_type}},
                    {"range": {"source_date": {"lt": source_date}}}
                ]
            }
        }
    }

    # search_after를 사용한 페이지네이션으로 대량 삭제 처리
    while True:
        response = client.search(index=INDEX_NAME, body=search_query_body)
        hits = response.get("hits", {}).get("hits", [])

        if not hits:
            break

        bulk_ops = [
            {"delete": {"_index": INDEX_NAME, "_id": hit["_id"]}}
            for hit in hits
        ]
        client.bulk(body=bulk_ops)

   

 

 

# Map 상태에서 전달된 결과 배열
branch_results = event.get('branchResults', [])

if not branch_results:
    raise ValueError("No branch results provided")

# 결과 병합 코드 예시
collected_at = datetime.now(KST).strftime("%Y-%m-%d %H:%M:%S")
aggregated_data = {
    'collected_at': collected_at,
    'branches': []
}

total_hotplaces = 0
for result in branch_results:
    branch_data = {
        'branch_code': result.get('branchData', {}).get('branchCode'),
        'branch_name': result.get('branchData', {}).get('branchName'),
        'hotplaces': result.get('validatedHotplaces', [])
    }
    aggregated_data['branches'].append(branch_data)
    total_hotplaces += len(branch_data['hotplaces'])
    logger.info(f"Branch {branch_data['branch_name']}: {len(branch_data['hotplaces'])} hotplaces")

# S3에 저장
today_kst = datetime.now(KST).strftime('%Y%m%d')
s3_key = f'hotplace_{today_kst}.json'

try:
    s3_client.put_object(
        Bucket=OUTPUT_BUCKET,
        Key=s3_key,
        Body=json.dumps(aggregated_data, ensure_ascii=False, indent=2),
        ContentType='application/json; charset=utf-8'
    )
    logger.info(f"Successfully saved to s3://{OUTPUT_BUCKET}/{s3_key}")
except Exception as e:
    logger.error(f"Failed to save to S3: {e}")
    raise

 

# 에이전트 상태 데이터 주입 함수 예시
def extract_urls(items: dict, agent_state: AgentState) -> None:
    """URL 정보를 추출하여 Agent State에 저장"""
    url_info = {}
    
    for key, item_list in items.items():
        url_info[key] = []
        url_key = 'url'
        
        for idx, item in enumerate(item_list):
            entry = {
                url_key: item.pop(url_key, '')
            }
            url_info[key].append({str(idx): entry})
    
    # Agent State에 저장 (LLM 컨텍스트에는 포함되지 않음)
    agent_state.set("url_info", url_info)

def _extract_answer_type(self, tool_result: List[Dict[str, Any]]):
    """tool_result에서 answerType만 추출하고 나머지는 truncate"""
    simplified_result = []
    answer_type = "N/A"
    
    for item in tool_result:
        if 'toolResult' in item:
            tool_result_data = item['toolResult']
            simplified_item = {
                'toolResult': {
                    'toolUseId': tool_result_data.get('toolUseId'),
                    'status': tool_result_data.get('status'),
                    'content': []
                }
            }

            for content_item in tool_result_data.get('content', []):
                if 'text' in content_item:
                    try:
                        text_data = ast.literal_eval(content_item['text'])
                        answer_type = text_data.get('answerType', 'unknown')
                        
                        # 토큰 절약을 위해 축약된 형태로 저장
                        simplified_item['toolResult']['content'].append({
                            'text': f"{{'answerType': '{answer_type}'}} Rest of the tool result has been truncated."
                        })
                    except:
                        simplified_item['toolResult']['content'].append(content_item)
                else:
                    simplified_item['toolResult']['content'].append(content_item)

            simplified_result.append(simplified_item)
        else:
            simplified_result.append(item)

    return simplified_result, answer_type

# 시스템 프롬프트 예시
<response_policy>
...
At the end of the message, append referenced indices for all keys in "results".
Format: <<'key1':[i1,i2], 'key2':[j1,j2]>>

Example:
이번주 잠실점 행사 정보 안내드립니다.
· 이달의 행사: 행사명
· 주요 기획전: 기획전 1, 기획전 2
<<'eventlist':[1,2]>>
...
</response_policy>

# 답변 생성 지침 예시
def generate_response_instruction(results: dict, query: str) -> str:
    """검색 결과 유무에 따른 응답 지침 생성"""
    
    if not is_empty_result(results):
        # 검색 결과가 있는 경우 프롬프트 예시: 결과 기반 답변 생성
        return f"""Your task is to generate appropriate answers for the user's question 
        based on the data from 'search_context_data'. If the 'search_context_data' does not have enough 
        information, inform that you cannot answer."""
    else:
        # 검색 결과가 없는 경우 프롬프트 예시: 답변 불가 안내
        return f"""No sufficient information available. Inform the user: 
        '현재 말씀해주신 {query}에 대한 정보를 제공할 수 없어요. 
        더 자세한 정보가 궁금하시다면 고객센터로 문의해주세요.'"""