数字资产是企业在日益数字化的世界中产品、服务、文化和品牌身份的重要视觉表现形式。数字资产和记录的用户行为可以通过提供互动和个性化体验来促进客户参与，使公司能够与目标受众更深入地联系。在数字资产中高效地发现和搜索特定内容对于企业优化工作流程、简化协作并向正确的受众传递相关内容至关重要。根据一项研究，到2021年，视频已经占据了所有消费者互联网流量的81%。这个观察结果并不令人惊讶，因为视频和音频是强大的视听媒体，提供更沉浸式的体验，并以更高的情感层面吸引目标受众。
随着公司积累了大量的数字资产，有效地组织和管理它们以最大化其价值变得更加具有挑战性。传统上，公司将元数据（如关键词、标题和描述）附加到这些数字资产上，以便促进相关内容的搜索和检索。但这需要一个设计良好的数字资产管理系统，并需要额外的工作来首先存储这些资产。实际上，大多数数字资产缺乏信息丰富的元数据，以实现高效的内容搜索。此外，您经常需要对整个文件的不同部分进行分析，并发现其中涵盖的概念。这是耗时且需要大量手动工作的。
生成式人工智能，特别是在自然语言处理和理解领域（NLP和NLU），已经彻底改变了我们理解和分析文本的方式，使我们能够以高效和规模化的方式获得更深入的见解。大型语言模型（LLMs）的进步使文本的表示更加丰富，为数字资产提供了更好的搜索能力。检索增强生成（RAG）是建立在LLMs和先进提示技术之上的一种流行方法，它可以基于企业数字资产存储中隐藏的信息提供更准确的答案。通过利用LLMs的嵌入模型以及强大的索引器和检索器，RAG可以理解和处理口头或书面查询，并快速找到知识库中最相关的信息。先前的研究已经显示了如何应用RAG来提供与企业私有领域知识相连接的问答解决方案。然而，在所有类型的数字资产中，视频和音频资产是最常见和重要的。
基于RAG的视频/音频问答解决方案有可能解决定位以非文本内容形式存在的培训和参考资料的业务问题。在这些资产中，标签或元数据有限，解决方案试图让用户与聊天机器人进行交互，并获取答案，这些答案可以是特定视频培训的链接（“我需要Amazon S3数据存储培训的链接”），文档的链接（“我需要学习机器学习的链接”），或者是在视频中涵盖的问题（“告诉我如何创建一个S3存储桶”）。聊天机器人的响应将能够直接回答问题，并包含指向源视频的链接，其中包含对用户请求最相关的内容的具体时间戳。
在本文中，我们演示了如何利用RAG的力量，在Amazon SageMaker上构建一个针对视频和音频资产的问答解决方案。
解决方案概述
以下图表说明了解决方案架构。
主要工作流程包括以下几个阶段：
1. 使用语音转文本模型将视频转换为文本，并进行文本与视频的对齐和组织。我们将数据存储在Amazon Simple Storage Service（Amazon S3）中。
2. 使用RAG方法和LLMs以及LangChain实现智能视频搜索。用户可以获取由LLMs生成的答案以及带有时间戳的相关来源。
3. 使用SageMaker构建一个多功能聊天机器人，其中包装和部署了上述两种解决方案。
有关详细实现，请参阅GitHub repo。
先决条件
您需要一个具有AWS Identity and Access Management（IAM）角色的AWS账户，该角色具有管理解决方案创建的资源的权限。有关详细信息，请参阅创建AWS账户。
如果这是您第一次使用Amazon SageMaker Studio，请先创建一个SageMaker域。此外，您可能需要请求相应的SageMaker处理和托管实例的服务配额增加。对于预处理视频数据，我们使用了一个ml.p3.2xlarge SageMaker处理实例。对于托管Falcon-40B，我们使用了一个ml.g5.12xlarge SageMaker托管实例。
使用语音转文本模型和句子嵌入模型将视频转换为文本
为了能够通过视频或音频数字资产进行搜索，并从视频中提供上下文信息给LLMs，我们需要将所有媒体内容转换为文本，然后按照NLP中的一般方法来处理文本数据。为了使我们的解决方案能够更灵活地处理不同的场景，我们为此任务提供了以下选项：
- Amazon Transcribe和Amazon Translate – 如果每个视频和音频文件中只包含一种语言，我们强烈推荐您选择Amazon Transcribe，这是一种由AWS管理的服务，用于转录音频和视频文件。如果您需要将它们翻译成相同的语言，Amazon Translate是另一种由AWS管理的服务，支持多语言翻译。
- Whisper – 在实际应用场景中，视频数据可能包含多种语言，例如外语学习视频。Whisper是一个多任务语音识别模型，可以执行多语言语音识别、语音翻译和语言识别。您可以使用Whisper模型来检测和转录视频数据中的不同语言，然后将所有不同语言翻译为一种语言。对于大多数RAG解决方案来说，使用相同语言的知识库非常重要。尽管OpenAI提供了Whisper API，但在本文中，我们使用了来自Hugging Face的Whisper模型。
我们使用Amazon SageMaker Processing作业在现有数据上运行此任务。您可以参考data_preparation.ipynb以获取如何运行此任务的详细信息。
将视频数据转换为音频数据
因为Amazon Transcribe可以处理视频和音频数据，而Whisper模型只能接受音频数据，为了使两个选项都能正常工作，我们需要将视频数据转换为音频数据。在下面的代码中，我们使用moviepy库中的VideoFileClip来运行此任务：
from moviepy.editor import VideoFileClip
video = VideoFileClip(video_path)
video.audio.write_audiofile(audio_path)转录音频数据
当音频数据准备好后，我们可以从两个转录选项中选择。您可以根据我们之前提到的标准，根据自己的用例选择最佳选项。
选项1：Amazon Transcribe和Amazon Translate
第一个选项是使用Amazon AI服务，如Amazon Transcribe和Amazon Translate，获取视频和音频数据集的转录结果。选择此选项时，您可以参考以下GitHub示例。
选项2：Whisper
Whisper模型可以处理长达30秒的音频数据。为了处理大型音频数据，我们使用transformers.pipeline来使用Whisper进行推理。在搜索相关视频剪辑或使用RAG生成内容时，相关剪辑的时间戳是重要的参考。因此，我们打开return_timestamps参数以获得带有时间戳的输出。通过在generate_kwargs中设置language参数，将一个视频文件中的所有不同语言转录和翻译为相同的语言。stride_length_s是每个块左右两侧的步长长度。通过此参数，我们可以使Whisper模型在每个块上进行推理时看到更多的上下文，从而得到更准确的结果。请参考以下代码：
from transformers import pipeline
import torch
target_language = "en"
whisper_model = "whisper-large-v2"
device = "cuda:0" if torch.cuda.is_available() else "cpu"
pipe = pipeline(
"automatic-speech-recognition",
model=f"openai/{whisper_model}",
device=device
)
generate_kwargs = {"task":"transcribe", "language":f"<|{target_language}|>"}
prediction = pipe(
file_path,
return_timestamps=True,
chunk_length_s=30,
stride_length_s=(5),
generate_kwargs=generate_kwargs
)pipe的输出是字典格式的数据，包含text和chunks项。其中，text包含完整的转录结果，chunks由带有时间戳和相应转录结果的块组成（请参见以下屏幕截图）。我们使用块中的数据进行进一步处理。
正如上面的截图所示，很多句子被截断并分成了不同的块。为了使这些块更有意义，我们需要在下一步中合并被截断的句子并更新时间戳。
整理句子
我们使用一个非常简单的规则来合并句子。当块以句号（.）结尾时，我们不做任何改变；否则，我们将其与下一个块连接起来。下面的代码片段解释了我们如何进行这个改变：
prev_chunk = None
new_chunks = []
for chunk in chunks:
if prev_chunk:
chunk['text'] = prev_chunk['text'] + chunk['text']
chunk['timestamp'] = (prev_chunk['timestamp'][0], chunk['timestamp'][1])
if not chunk['text'].endswith('.'):
prev_chunk = chunk
else:
new_chunks.append(chunk)
prev_chunk = None与音频转文本转换生成的原始块相比，我们可以得到最初被截断的完整句子。
分块句子
文档中的文本内容通常是按段落组织的。每个段落都专注于同一个主题。通过按段落分块，可以帮助将文本嵌入更有意义的向量中，从而提高检索准确性。
与文档中的普通文本内容不同，转录模型的转录结果没有分段。即使音频文件中有一些停顿，有时也不能用来分段句子。另一方面，langchain 提供了递归分块文本分割函数 RecursiveCharacterTextSplitter，它可以将所有语义相关的内容保留在同一个块中。因为我们需要保留时间戳与块，所以我们实现了自己的分块过程。受到帖子《如何使用Python将文本分块成段落》的启发，我们使用句子嵌入方法根据相邻句子之间的相似性来分块句子。基本思想是将与相邻句子相似性最低的句子作为分割点。我们使用 all-MiniLM-L6-v2 进行句子嵌入。您可以参考原始帖子了解该方法的解释。我们对原始源代码进行了一些小的修改，请参考我们的源代码进行实现。该过程的核心部分如下：
# 嵌入句子
model_name = "all-minilm-l6-v2"
model = SentenceTransformer(model_name)
embeddings = model.encode(sentences_all)
# 创建相似性矩阵
similarities = cosine_similarity(embeddings)
# 让我们应用我们的函数。对于较长的句子，建议使用10个或更多句子
minmimas = activate_similarities(similarities, p_size=p_size, order=order)
# 创建空字符串
split_points = [each for each in minmimas[0]]
text = ''
para_chunks = []
para_timestamp = []
start_timestamp = 0
for num, each in enumerate(sentences_all):
current_timestamp = timestamps_all[num]
if text == '' and (start_timestamp == current_timestamp[1]):
start_timestamp = current_timestamp[0]
if num in split_points:
para_chunks.append(text)
para_timestamp.append([start_timestamp, current_timestamp[1]])
text = f'{each}. '
start_timestamp = current_timestamp[1]
else:
text+=f'{each}. '
if len(text):
para_chunks.append(text)
para_timestamp.append([start_timestamp, timestamps_all[-1][1]])为了评估使用句子嵌入进行分块的效率，我们进行了不同分块机制之间的定性比较。这种比较的假设是，如果分块的文本在语义上更不同和分离，那么在问答过程中检索到的无关上下文信息将更少，从而答案将更准确和精确。同时，由于向LLM发送的上下文信息较少，推理的成本也会较低，因为计费随着令牌的大小而增加。
我们通过将高维度降至二维来可视化PCA的前两个主成分。与递归分块相比，我们可以看到使用句子嵌入表示不同块的向量之间的距离更分散，这意味着块更语义上分离。这意味着当查询的向量接近一个块的向量时，它可能较少接近其他块。在检索任务中，将会减少从多个语义相似的块中选择相关信息的机会。
当分块过程完成后，我们将时间戳附加到每个块的文件名中，将其保存为单个文件，然后上传到S3存储桶中。
使用基于RAG的LangChain实现智能视频搜索
通常有四种方法可以使用LangChain构建RAG解决方案进行Q&A：
- 使用
load_qa_chain功能，将所有信息提供给LLM。鉴于上下文窗口大小和视频音频数据的容量，这不是一种理想的方法。 - 使用
RetrievalQA工具，需要文本拆分器、文本嵌入模型和向量存储器来处理文本并检索相关信息。 - 使用
VectorstoreIndexCreator，这是第二种方法中所有逻辑的封装。文本拆分器、文本嵌入模型和向量存储器在函数内一次配置在一起。 - 使用
ConversationalRetrievalChain工具，将聊天历史的记忆进一步添加到QA解决方案中。
在本文中，我们使用第二种方法来明确自定义并选择最佳的工程实践。在接下来的几个部分中，我们详细描述每个步骤。
为了根据用户输入查询搜索相关内容，我们使用语义搜索，这可以更好地理解查询背后的意图并执行有意义的检索。我们首先使用预训练的嵌入模型将所有转录文本嵌入到一个向量空间中。在搜索时，查询也被嵌入到相同的向量空间中，并找到源语料库中最接近的嵌入。您可以部署预训练的嵌入模型，如Amazon SageMaker JumpStart中的使用基于RAG的基础模型的问答系统，以创建语义搜索的嵌入。在我们的文章中，我们采用类似的方法，使用开源的LangChain库来创建基于RAG的智能视频搜索解决方案。LangChain是一个用于开发由语言模型驱动的应用程序的开源框架。LangChain为许多不同的LLM提供了通用的接口。
我们首先部署由Amazon SageMaker JumpStart提供的GPT-J 6B嵌入模型和Hugging Face提供的Falcon-40B Instruct语言模型，为解决方案做准备。当端点准备就绪时，我们按照Amazon SageMaker JumpStart中描述的使用基于RAG的基础模型的问答系统的类似步骤来创建LLM模型和嵌入模型，以供LangChain使用。
以下代码片段展示了如何使用langchain.llms.sagemaker_endpoint.SagemakerEndpoint类创建LLM模型，并在ContentHandler中转换LLM的请求和响应负载：
from langchain.llms.sagemaker_endpoint import LLMContentHandler, SagemakerEndpoint
parameters = {
"max_new_tokens": 500,
}
class ContentHandler(LLMContentHandler):
content_type = "application/json"
accepts = "application/json"
def transform_input(self, prompt: str, model_kwargs={}) -> bytes:
self.len_prompt = len(prompt)
input_str = json.dumps({"inputs": prompt , "parameters": {**model_kwargs}})
return input_str.encode("utf-8")
def transform_output(self, output: bytes) -> str:
response_json = output.read()
res = json.loads(response_json)
print(res)
ans = res[0]['generated_text'][self.len_prompt:]
return ans
content_handler = ContentHandler()
sm_llm = SagemakerEndpoint(
endpoint_name=_MODEL_CONFIG_["huggingface-falcon-40b"]["endpoint_name"],
region_name=aws_region,
model_kwargs=parameters,
content_handler=content_handler,
) 当我们使用SageMaker JumpStart嵌入模型时，我们需要自定义LangChain SageMaker端点嵌入类，并将模型请求和响应转换为与LangChain集成。使用LangChain文档加载器加载处理过的视频转录，然后创建索引。
我们使用LangChain的DirectoryLoader包将文本文档加载到文档加载器中：
loader = DirectoryLoader("./data/demo-video-sagemaker-doc/", glob="*/.txt")
documents = loader.load()接下来，我们使用嵌入模型创建内容的嵌入，并将嵌入存储在FAISS向量存储中以创建索引。我们使用VectorstoreIndexCreator类来实现这个任务只需写几行代码：
index_creator = VectorstoreIndexCreator(
vectorstore_cls=FAISS,
embedding=embeddings,
text_splitter=CharacterTextSplitter(chunk_size=500, chunk_overlap=0),
)
index = index_creator.from_loaders([loader])现在我们可以使用索引搜索相关内容，并将其传递给LLM模型生成准确的响应：
index.query(question=question, llm=sm_llm)使用SageMaker构建多功能聊天机器人
在SageMaker上部署LLM之后，我们可以构建一个多功能智能聊天机器人，展示这些模型如何帮助您的业务构建先进的AI应用程序。在这个示例中，聊天机器人使用Streamlit构建UI，使用LangChain框架将不同组件连接在LLM周围。借助部署在SageMaker上的文本到文本和语音到文本的LLM的帮助，这个智能聊天机器人接受来自文本文件和音频文件的输入，用户可以与输入文件进行聊天（接受文本和音频文件），并在此基础上构建应用程序。以下图示展示了聊天机器人的架构。
当用户上传文本文件到聊天机器人时，聊天机器人将内容放入LangChain内存组件中，用户可以与上传的文档进行聊天。这部分的灵感来自以下GitHub示例，该示例使用SageMaker构建文档聊天机器人。我们还添加了一个选项，允许用户上传音频文件。然后，聊天机器人自动调用部署在SageMaker端点上的语音到文本模型，从上传的音频文件中提取文本内容，并将文本内容添加到LangChain内存中。最后，我们允许用户选择在回答问题时使用知识库。这是前面图示中显示的RAG功能。我们已经在前面的章节中提供的笔记本中定义了部署的SageMaker端点。请注意，在运行Streamlit应用程序时，您需要传递实际显示在您的帐户中的端点名称。您可以在SageMaker控制台的Inference和Endpoints下找到端点名称。
Falcon_endpoint_name = os.getenv("falcon_ep_name", default="falcon-40b-instruct-12xl")
whisper_endpoint_name = os.getenv('wp_ep_name', default="whisper-large-v2")
embedding_endpoint_name = os.getenv('embed_ep_name', default="huggingface-textembedding-gpt-j-6b")当未选择知识库选项时，我们使用对话链，通过使用LangChain提供的ConversationBufferMemory添加内存组件，以便机器人能够记住当前的对话历史：
def load_chain():
memory = ConversationBufferMemory(return_messages=True)
chain = ConversationChain(llm=llm, memory=memory)
return chain
chatchain = load_chain()我们使用与前面章节中显示的RAG组件相似的逻辑，并将文档检索功能添加到代码中。为了演示目的，我们将存储在SageMaker Studio本地存储中的转录文本作为文档源加载。您可以根据自己的选择使用基于向量数据库的其他RAG解决方案，例如Amazon OpenSearch Service、Amazon RDS、Amazon Kendra等。
当用户使用知识库进行提问时，以下代码片段从数据库中检索相关内容，并为LLM提供附加上下文以回答问题。在搜索相关文档时，我们使用了FAISS提供的特定方法similarity_search_with_score。这是因为它还可以提供检索到的源文件的元数据和相似度得分。返回的距离得分是L2距离。因此，得分越低越好。这为我们提供了更多选项来为用户提供更多上下文，例如提供与输入查询相关的源视频的确切时间戳。当用户从UI中选择RAG选项时，聊天机器人使用LangChain提供的load_qa_chain函数根据输入提示提供答案。
docs = docsearch.similarity_search_with_score(user_input)
contexts = []
for doc, score in docs:
print(f"内容：{doc.page_content}，元数据：{doc.metadata}，得分：{score}")
if score <= 0.9:
contexts.append(doc)
source.append(doc.metadata['source'].split('/')[-1])
print(f"\n输入上下文：{contexts}")
prompt_template = """使用以下上下文片段回答最后的问题。如果不知道答案，请直接说不知道，不要编造答案。:\n\n{context}\n\n问题：{question}\n有帮助的答案："""
PROMPT = PromptTemplate(template=prompt_template, input_variables=["context", "question"])
chain = load_qa_chain(llm=llm, prompt=PROMPT)
result = chain({"input_documents": contexts, "question": user_input},
return_only_outputs=True)["output_text"]
if len(source) != 0:
df = pd.DataFrame(source, columns=['知识源'])
st.data_editor(df)运行聊天机器人应用
现在我们准备运行Streamlit应用程序。在SageMaker Studio中打开终端并导航到克隆的GitHub存储库文件夹。您需要安装requirements.txt文件中指定的所需的Python包。运行pip install -r requirements.txt以准备Python依赖项。
然后运行以下命令，根据您在帐户中部署的端点更新环境变量中的端点名称。当您运行chatbot.py文件时，它会根据环境变量自动更新端点名称。
export falcon_ep_name=<您在帐户中部署的falcon端点名称>
export wp_ep_name=<您在帐户中部署的whisper端点名称>
export embed_ep_name=<您在帐户中部署的嵌入端点名称>
streamlit run app_chatbot/chatbot.py --server.port 6006 --server.maxUploadSize 6
要访问Streamlit UI，请复制SageMaker Studio的URL，并将lab?替换为proxy/[端口号]/。对于本文，我们将服务器端口指定为6006，因此URL应类似于https://<域ID>.studio.<区域>.sagemaker.aws/jupyter/default/proxy/6006/。
请将域ID和区域替换为帐户中的正确值以访问UI。
与音频文件交谈
在对话设置窗格中，选择浏览文件以选择要上传到聊天机器人的本地文本或音频文件。如果选择音频文件，它将自动调用语音到文本SageMaker端点处理音频文件，并将转录的文本呈现到控制台，如下图所示。您可以继续对音频文件提问，聊天机器人将能够记住音频内容并根据音频内容回答您的问题。
使用知识库进行问答
当您想回答需要特定领域知识或使用知识库的问题时，请选择使用知识库。这使得聊天机器人可以从之前构建的知识库（向量数据库）中检索相关信息，以提供额外的上下文来回答问题。例如，当我们向聊天机器人提问“首先如何自定义基础模型？”而没有使用知识库时，聊天机器人返回类似以下屏幕截图的答案。
当我们使用知识库来帮助回答这个问题时，聊天机器人返回不同的回答。在演示视频中，我们阅读了有关如何在SageMaker Jumpstart中自定义模型的SageMaker文档。
输出还提供了检索到的文本对应的原始视频文件名和时间戳。用户可以返回到原始视频文件并定位原始视频中的特定片段。
这个示例聊天机器人演示了企业如何利用各种类型的数字资产来增强他们的知识库，并为员工提供多功能的帮助，以提高生产力和效率。您可以从文档、音频和视频数据集甚至图像数据集构建知识数据库，将所有资源整合在一起。借助SageMaker作为先进的ML平台，您可以利用SageMaker服务的广度和深度，加速项目构思到投产速度，涵盖整个ML生命周期。
清理
为了节省成本，请删除您作为本文一部分部署的所有资源。您可以按照提供的笔记本的清理部分的说明来以编程方式删除资源，或者您可以通过SageMaker控制台删除可能已创建的任何SageMaker端点。
总结
由LLM支持的生成式AI模型的出现，彻底改变了企业从信息中获取和应用洞察力的方式。在这个背景下，包括视频和音频内容在内的数字资产作为产品、服务和品牌身份的视觉表达起着关键作用。在这些资产中高效地搜索和发现特定内容对于优化工作流程、增强协作能力以及为目标受众提供定制化体验至关重要。借助SageMaker上生成式AI模型的强大功能，企业可以充分发挥其视频和音频资源的潜力。生成式AI模型的集成赋予企业构建高效智能搜索解决方案的能力，使用户能够从数字资产中获取相关的上下文信息，从而最大化其价值，并在数字化领域取得业务成功。
有关在AWS上使用生成式AI的更多信息，请参阅宣布推出用于在AWS上构建生成式AI的新工具。