LLM으로 Amazon Lex를 개선하고 URL 수집을 사용하여 FAQ 환경 개선 | 아마존 웹 서비스

import logging
import sys
import requests
import html2text
from llama_index.readers.schema.base import Document
from llama_index import GPTVectorStoreIndex
from typing import List logging.basicConfig(stream=sys.stdout, level=logging.INFO)
logging.getLogger().addHandler(logging.StreamHandler(stream=sys.stdout)) class EZWebLoader: def __init__(self, default_header: str = None): self._html_to_text_parser = html2text() if default_header is None: self._default_header = {"User-agent":"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/47.0.2526.80 Safari/537.36"} else: self._default_header = default_header def load_data(self, urls: List[str], headers: str = None) -> List[Document]: if headers is None: headers = self._default_header documents = [] for url in urls: response = requests.get(url, headers=headers).text response = self._html2text.html2text(response) documents.append(Document(response)) return documents url = "http://www.zappos.com/general-questions"
loader = EZWebLoader()
documents = loader.load_data([url])
index = GPTVectorStoreIndex.from_documents(documents)

import time
import json
import os
import logging
import boto3
from llama_index import StorageContext, load_index_from_storage logger = logging.getLogger()
logger.setLevel(logging.DEBUG) def download_docstore(): # Create an S3 client s3 = boto3.client('s3') # List all objects in the S3 bucket and download each one
try: bucket_name = 'faq-bot-storage-001' s3_response = s3.list_objects_v2(Bucket=bucket_name) if 'Contents' in s3_response: for item in s3_response['Contents']: file_name = item['Key'] logger.debug("Downloading to /tmp/" + file_name) s3.download_file(bucket_name, file_name, '/tmp/' + file_name) logger.debug('All files downloaded from S3 and written to local filesystem.') except Exception as e: logger.error(e)
raise e #download the doc store locally
download_docstore() storage_context = StorageContext.from_defaults(persist_dir="/tmp/")
# load index
index = load_index_from_storage(storage_context)
query_engine = index.as_query_engine() def lambda_handler(event, context): """
Route the incoming request based on intent.
The JSON body of the request is provided in the event slot. """ # By default, treat the user request as coming from the America/New_York time zone. os.environ['TZ'] = 'America/New_York' time.tzset() logger.debug("===== START LEX FULFILLMENT ====") logger.debug(event) slots = {} if "currentIntent" in event and "slots" in event["currentIntent"]: slots = event["currentIntent"]["slots"] intent = event["sessionState"]["intent"] dialogaction = {"type": "Delegate"} message = [] if str.lower(intent["name"]) == "fallbackintent": #execute query from the input given by the user response = str.strip(query_engine.query(event["inputTranscript"]).response) dialogaction["type"] = "Close" message.append({'content': f'{response}', 'contentType': 'PlainText'}) final_response = { "sessionState": { "dialogAction": dialogaction, "intent": intent }, "messages": message } logger.debug(json.dumps(final_response, indent=1)) logger.debug("===== END LEX FULFILLMENT ====") return final_response

import logging
import requests
import html2text
from llama_index.readers.schema.base import Document
from typing import List
import re def find_http_urls_in_parentheses(s: str, prefix: str = None): pattern = r'((https?://[^)]+))' urls = re.findall(pattern, s) matched = [] if prefix is not None: for url in urls: if str(url).startswith(prefix): matched.append(url) else: matched = urls return list(set(matched)) # remove duplicates by converting to set, then convert back to list class EZWebLoader: def __init__(self, default_header: str = None): self._html_to_text_parser = html2text if default_header is None: self._default_header = {"User-agent":"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/47.0.2526.80 Safari/537.36"} else: self._default_header = default_header def load_data(self, urls: List[str], num_levels: int = 0, level_prefix: str = None, headers: str = None) -> List[Document]: logging.info(f"Number of urls: {len(urls)}.") if headers is None: headers = self._default_header documents = [] visited = {} for url in urls: q = [url] depth = num_levels for page in q: if page not in visited: #prevent cycles by checking to see if we already crawled a link logging.info(f"Crawling {page}") visited[page] = True #add entry to visited to prevent re-crawling pages response = requests.get(page, headers=headers).text response = self._html_to_text_parser.html2text(response) #reduce html to text documents.append(Document(response)) if depth > 0: #crawl linked pages ingest_urls = find_http_urls_in_parentheses(response, level_prefix) logging.info(f"Found {len(ingest_urls)} pages to crawl.") q.extend(ingest_urls) depth -= 1 #reduce the depth counter so we go only num_levels deep in our crawl else: logging.info(f"Skipping {page} as it has already been crawled") logging.info(f"Number of documents: {len(documents)}.") return documents url = "http://www.zappos.com/general-questions"
loader = EZWebLoader()
#crawl the site with 1 level depth and prefix of "/c/" for customer service root
documents = loader.load_data([url], num_levels=1, level_prefix="https://www.zappos.com/c/")
index = GPTVectorStoreIndex.from_documents(documents)