使用LLMs增强Amazon Lex，并通过URL摄取来改善FAQ体验
使用LLMs增强Amazon Lex，通过URL摄取改善FAQ体验
在今天的数字世界中，大多数消费者更愿意自己找到客户服务问题的答案，而不愿花时间联系企业和/或服务提供商。本博客文章探讨了一种创新的解决方案，可以在亚马逊Lex中构建一个问题和答案聊天机器人，该聊天机器人使用您网站上现有的常见问题解答。这个由人工智能驱动的工具可以提供快速、准确的回答，让客户能够快速、轻松地独立解决常见问题。
单URL摄入
许多企业在其网站上为客户提供了一套常见问题解答。在这种情况下，我们希望为客户提供一个可以根据我们发布的常见问题解答回答他们问题的聊天机器人。在题为“使用LLMs增强Amazon Lex的对话常见问题解答功能”的博客文章中，我们演示了如何使用亚马逊Lex和LlamaIndex的组合来构建一个由现有知识源（如PDF或Word文档）驱动的聊天机器人。为了支持一个简单的常见问题解答，基于网站上的常见问题解答，我们需要创建一个可以抓取网站并创建可以被LlamaIndex用于回答客户问题的嵌入的摄入过程。在这种情况下，我们将在前一篇博客文章中创建的机器人的基础上构建，该机器人使用用户的话语查询这些嵌入，并从网站常见问题解答中返回答案。
下图显示了我们的解决方案中的摄入过程和亚马逊Lex机器人如何协同工作。
在解决方案工作流程中，通过AWS Lambda对带有常见问题解答的网站进行摄入。这个Lambda函数会抓取网站，并将结果文本存储在亚马逊简单存储服务（Amazon S3）存储桶中。然后，S3存储桶触发一个Lambda函数，该函数使用LlamaIndex创建嵌入，并存储在Amazon S3中。当终端用户提出问题时，比如“你们的退货政策是什么？”，亚马逊Lex机器人使用其Lambda函数使用基于RAG的方法与LlamaIndex查询这些嵌入。有关此方法和先决条件的更多信息，请参阅博客文章“使用LLMs增强Amazon Lex的对话常见问题解答功能”。
在完成了上述博客中的先决条件之后，第一步是将常见问题解答摄入到一个可以由LlamaIndex进行向量化和索引的文档存储库中。以下代码显示了如何完成这个步骤：
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)在上面的示例中，我们使用EZWebLoader类从Zappos获取一个预定义的常见问题解答网站URL，并进行摄入。使用这个类，我们已经导航到URL并将页面中的所有问题加载到索引中。现在我们可以提出一个问题，比如“Zappos有礼品卡吗？”并直接从网站上的常见问题解答中获得答案。下面的截图显示了亚马逊Lex机器人测试控制台从常见问题解答中回答这个问题。
我们之所以能够做到这一点，是因为我们在第一步中爬取了URL并创建了LlamaIndex可以用来搜索问题答案的嵌入。我们的机器人的Lambda函数展示了每当返回回退意图时如何运行这个搜索：
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():
# 创建一个S3客户端
s3 = boto3.client('s3')
# 列出S3存储桶中的所有对象并下载每一个对象
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("下载至 /tmp/" + file_name)
s3.download_file(bucket_name, file_name, '/tmp/' + file_name)
logger.debug('所有文件从S3下载并写入本地文件系统。')
except Exception as e:
logger.error(e)
raise e
#下载本地文档存储
download_docstore()
storage_context = StorageContext.from_defaults(persist_dir="/tmp/")
# 加载索引
index = load_index_from_storage(storage_context)
query_engine = index.as_query_engine()
def lambda_handler(event, context):
"""
根据意图路由传入的请求。
请求的JSON主体在事件槽中提供。
"""
# 默认情况下，将用户请求视为来自美国/纽约时区。
os.environ['TZ'] = 'America/New_York'
time.tzset()
logger.debug("===== 开始LEX履行 =====")
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":
#根据用户提供的输入执行查询
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("===== 结束LEX履行 =====")
return final_response当一个单独的网页拥有所有答案时，这种解决方案运作得很好。然而，大多数常见问题解答网站不是建立在单个页面上的。例如，在我们的Zappos示例中，如果我们问“您是否有价格匹配政策？”，那么我们得到的答案就不够满意，如下面的截图所示。
在上述互动中，对于我们的用户来说，价格匹配政策的答案并不有用。这个答案很简短，因为所引用的常见问题解答是一个关于价格匹配政策的特定页面的链接，而我们的网络爬虫只针对单个页面进行了爬取。要获得更好的答案，意味着需要爬取这些链接。下一节将展示如何获取需要两个或更多页面深度的问题的答案。
N级爬取
当我们为常见问题解答知识爬取一个网页时，我们想要的信息可能包含在链接的页面中。例如，在我们的Zappos示例中，我们问“您是否有价格匹配政策？”答案是“是，请访问<link>了解更多信息。”如果有人问“您的价格匹配政策是什么？”我们希望给出一个完整的答案，包括政策。要实现这一点，我们需要遍历链接以获取最终用户所需的实际信息。在摄取过程中，我们可以使用我们的网络加载器来查找HTML页面中的锚链接，然后遍历它们。以下代码更改我们的网络爬虫，以便在我们爬取的页面中查找链接。它还包括一些附加逻辑，以避免循环爬取并允许按前缀进行过滤。
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)) # 去重，将集合转换为列表
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"URL数量: {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: #通过检查我们是否已经爬取过链接，防止循环
logging.info(f"爬取 {page}")
visited[page] = True #添加到已访问列表，防止重新爬取页面
response = requests.get(page, headers=headers).text
response = self._html_to_text_parser.html2text(response) #将HTML转换为文本
documents.append(Document(response))
if depth > 0:
#爬取链接的页面
ingest_urls = find_http_urls_in_parentheses(response, level_prefix)
logging.info(f"发现{len(ingest_urls)}个要爬取的页面。")
q.extend(ingest_urls)
depth -= 1 #减少深度计数器，使我们只在我们的爬取中进入num_levels深度
else:
logging.info(f"跳过{page}，因为它已经被爬取过")
logging.info(f"文档数量: {len(documents)}。")
return documents
url = "http://www.zappos.com/general-questions"
loader = EZWebLoader()
#使用深度为1和以"/c/"为前缀的客户服务根目录爬取网站
documents = loader.load_data([url]
num_levels=1, level_prefix="https://www.zappos.com/c/")
index = GPTVectorStoreIndex.from_documents(documents)在上面的代码中，我们介绍了爬取多层级的能力，并且我们提供了一个前缀，允许我们限制爬取只针对以某个URL模式开头的内容。在我们的Zappos示例中，客户服务页面都是以zappos.com/c为根目录的，所以我们将其作为前缀，限制我们的爬取范围为一个更小而且更相关的子集。代码展示了我们如何爬取两层级的内容。我们的机器人的Lambda逻辑保持不变，因为除了爬取更多的文档之外，没有发生任何变化。
现在我们已经对所有文档建立了索引，我们可以提出更详细的问题。在下面的截图中，我们的机器人对问题“您有价格匹配政策吗？”给出了正确答案。
我们现在对关于价格匹配的问题有了完整的答案。不再只是简单地告诉我们“是的，请参阅我们的政策”，而是给出了从第二层级爬取的详细信息。
清理
为了避免未来产生费用，继续删除作为本次练习的一部分部署的所有资源。我们提供了一个脚本来优雅地关闭Sagemaker终端节点。使用详细信息请参阅README。此外，您可以在与其他cdk命令相同的目录中运行cdk destroy来删除所有其他资源，以取消配置您的堆栈中的所有资源。
结论
将一组常见问题解答导入到聊天机器人中，使您的客户可以通过简单、自然语言的查询找到答案。通过将Amazon Lex中的内置支持与像LlamaIndex这样的RAG解决方案结合起来，我们可以为客户提供一个快速的路径，以获得满意、经过策划和批准的常见问题解答。通过将N级爬取应用到我们的解决方案中，我们可以允许答案跨越多个常见问题解答链接，并为客户的查询提供更深入的答案。通过按照这些步骤操作，您可以无缝地将基于LLM的问答功能和高效的URL导入功能整合到您的Amazon Lex聊天机器人中。这将实现与用户更准确、全面和具有上下文意识的交互。
