我们很高兴地宣布，通过Amazon SageMaker实时推理，现在可以提供响应流式传输的功能。现在，当使用SageMaker实时推理时，您可以持续地将推理响应流式传输回客户端，以帮助您构建用于生成AI应用程序（如聊天机器人、虚拟助手和音乐生成器）的交互式体验。有了这个新功能，您可以在可用时立即开始流式传输响应，而不是等待整个响应生成。这降低了生成式AI应用程序的首字节时间。
在本文中，我们将展示如何使用SageMaker实时端点和新的响应流式传输功能来构建流式网络应用程序，以实现交互式聊天使用案例。我们在示例演示应用程序UI中使用Streamlit。
解决方案概述
要从SageMaker获取流式传输的响应，您可以使用我们的新InvokeEndpointWithResponseStream API。它通过提供更快的首字节响应时间来提高客户满意度。对于使用生成式AI模型构建的应用程序而言，减少客户感知的延迟特别重要，因为即时处理比等待整个有效负载更有价值。此外，它引入了一种粘性会话，可以实现交互的连续性，有利于聊天机器人等使用案例，以创建更自然和高效的用户体验。
SageMaker实时端点中响应流式传输的实现是通过HTTP 1.1分块编码实现的，这是一种用于发送多个响应的机制。这是一种支持二进制内容的HTTP标准，并且大多数客户端/服务器框架都支持它。HTTP分块编码支持文本和图像数据流式传输，这意味着托管在SageMaker端点上的模型可以将流式传输的响应作为文本或图像发送回来，例如Falcon、Llama 2和Stable Diffusion模型。在安全方面，输入和输出都使用TLS和AWS Sigv4 Auth进行了安全保护。还使用了同样的HTTP分块编码机制来实现其他流式传输技术，如服务器推送事件（Server-Sent Events，SSE）。要利用新的流式传输API，您需要确保模型容器将流式传输的响应返回为分块编码的数据。
下图说明了使用SageMaker推理端点进行响应流式传输的高级架构。
受益于流式传输响应的用例之一是生成式AI模型驱动的聊天机器人。传统上，用户发送一个查询，然后等待整个响应生成后才收到答案。这可能需要宝贵的几秒钟甚至更长的时间，这可能会降低应用程序的性能。使用响应流式传输，聊天机器人可以在生成过程中开始发送部分推理结果。这意味着用户几乎可以即时看到初始响应，即使AI在后台继续改进答案。这创造了一种无缝且引人入胜的对话流程，用户感觉自己在与实时理解和回应的AI进行聊天。
在本文中，我们展示了两种容器选项，用于创建具有响应流式传输的SageMaker端点：使用AWS Large Model Inference（LMI）容器和Hugging Face Text Generation Inference（TGI）容器。在接下来的章节中，我们将为您详细介绍使用LMI和TGI容器在SageMaker上部署和测试Falcon-7B-Instruct模型的实现步骤。我们选择了Falcon 7B作为示例，但任何模型都可以利用这个新的流式传输功能。
先决条件
您需要一个具有AWS Identity and Access Management（IAM）角色的AWS帐户，以便具有管理解决方案创建的资源的权限。有关详细信息，请参阅创建AWS帐户。如果您是第一次使用Amazon SageMaker Studio，请先创建一个SageMaker域。此外，您可能需要请求相应SageMaker托管实例的服务配额增加。对于Falcon-7B-Instruct模型，我们使用ml.g5.2xlarge SageMaker托管实例。对于托管Falcon-40B-Instruct模型，我们使用ml.g5.48xlarge SageMaker托管实例。您可以从服务配额UI请求配额增加。有关更多信息，请参阅请求配额增加。
选项1：使用LMI容器部署实时流式传输端点
LMI容器是SageMaker托管的用于低延迟推理用例的大型模型推理的深度学习容器之一，旨在帮助在AWS基础设施上托管大型语言模型（LLM）。LMI容器使用Deep Java Library（DJL）Serving，这是一个开源的、高级的、与引擎无关的Java框架，用于深度学习。使用这些容器，您可以使用相应的开源库，如DeepSpeed、Accelerate、Transformers-neuronx和FasterTransformer，使用模型并行技术对模型参数进行分区，以利用多个GPU或加速器的内存进行推理。有关使用LMI容器在SageMaker上部署大型模型的好处的详细信息，请参阅在Amazon SageMaker上使用FasterTransformer高性能部署大型模型和使用DJLServing和DeepSpeed模型并行推理在Amazon SageMaker上部署大型模型。您还可以在这个GitHub存储库中找到更多使用LMI容器在SageMaker上托管开源LLM的示例。
对于LMI容器，我们期望以下工件来帮助设置推理模型：
- serving.properties（必需） – 定义模型服务器设置
- model.py（可选） – 用于定义核心推理逻辑的Python文件
- requirements.txt（可选） – 需要安装的任何其他pip wheel
LMI容器可用于托管模型，而无需提供自己的推理代码。当输入数据没有自定义预处理或模型预测的后处理时，这非常有用。我们使用以下配置：
- 在此示例中，我们托管Falcon-7B-Instruct模型。我们需要创建一个包含所需托管选项的
serving.properties配置文件，并将其打包成tar.gz工件。可以通过在serving.properties文件中设置enable_streaming选项来启用DJL Serving的响应流式传输。有关所有支持的参数，请参阅Streaming Python配置。 - 在此示例中，我们使用DJL Serving中的默认处理程序来流式传输响应，因此我们只关心发送请求和解析输出响应。您还可以在
model.py文件中提供带有自定义处理程序的entrypoint代码，以自定义输入和输出处理程序。有关自定义处理程序的更多详细信息，请参阅自定义model.py处理程序。 - 由于我们将Falcon-7B-Instruct模型托管在单个GPU实例（ml.g5.2xlarge）上，因此将
option.tensor_parallel_degree设置为1。如果计划在多个GPU上运行，请使用此选项设置每个工作进程使用的GPU数量。 - 我们使用
option.output_formatter控制输出内容类型。默认的输出内容类型是application/json，因此如果您的应用程序需要不同的输出，可以覆盖此值。有关可用选项的更多信息，请参阅配置和设置和所有DJL配置选项。
%%writefile serving.properties
engine=MPI
option.model_id=tiiuae/falcon-7b-instruct
option.trust_remote_code=true
option.tensor_parallel_degree=1
option.max_rolling_batch_size=32
option.rolling_batch=auto
option.output_formatter=jsonlines
option.paged_attention=false
option.enable_streaming=true要创建SageMaker模型，请检索容器映像URI：
image_uri = image_uris.retrieve(
framework="djl-deepspeed",
region=sess.boto_session.region_name,
version="0.23.0"
)使用SageMaker Python SDK创建SageMaker模型，并使用deploy方法将其部署到SageMaker实时端点：
instance_type = "ml.g5.2xlarge"
endpoint_name = sagemaker.utils.name_from_base("lmi-model-falcon-7b")
model = Model(sagemaker_session=sess,
image_uri=image_uri,
model_data=code_artifact,
role=role)
model.deploy(
initial_instance_count=1,
instance_type=instance_type,
endpoint_name=endpoint_name,
container_startup_health_check_timeout=900
)当端点正在提供服务时，您可以使用InvokeEndpointWithResponseStream API调用来调用模型。此API允许模型作为完整响应有效负载的一部分的流式传输进行响应。这使得模型能够以更大的响应大小响应，并为响应的第一个字节和最后一个字节的生成之间存在显着差异的模型提供更快的首字节响应时间。
LMI容器的x-amzn-sagemaker-content-type中显示的响应内容类型为application/jsonlines，如模型属性配置中所指定的。因为它是推理所支持的通用数据格式之一，所以我们可以使用SageMaker Python SDK提供的默认反序列化器来反序列化JSON行数据。我们创建一个辅助的LineIterator类来解析从推理请求接收到的响应流：
class LineIterator:
"""
用于解析字节流输入的辅助类。
模型的输出将具有以下格式：
```
b'{"outputs": [" a"]}\n'
b'{"outputs": [" challenging"]}\n'
b'{"outputs": [" problem"]}\n'
...
```
虽然通常来说，来自事件流的每个PayloadPart事件将包含一个具有完整json的字节数组，
但这并不是保证的，某些json对象可能会分布在多个PayloadPart事件中。例如：
```
{'PayloadPart': {'Bytes': b'{"outputs": '}}
{'PayloadPart': {'Bytes': b'[" problem"]}\n'}}
```
此类通过连接通过'write'函数写入的字节，并通过'scan_lines'函数在缓冲区中返回以'\n'字符结尾的行，
而保持最后读取位置的位置，以确保不再次暴露先前的字节。
"""
def __init__(self, stream):
self.byte_iterator = iter(stream)
self.buffer = io.BytesIO()
self.read_pos = 0
def __iter__(self):
return self
def __next__(self):
while True:
self.buffer.seek(self.read_pos)
line = self.buffer.readline()
if line and line[-1] == ord('\n'):
self.read_pos += len(line)
return line[:-1]
try:
chunk = next(self.byte_iterator)
except StopIteration:
if self.read_pos < self.buffer.getbuffer().nbytes:
continue
raise
if 'PayloadPart' not in chunk:
print('Unknown event type:' + chunk)
continue
self.buffer.seek(0, io.SEEK_END)
self.buffer.write(chunk['PayloadPart']['Bytes'])在上述代码中的类中，每次流式响应时，它都会返回一个二进制字符串（例如，b'{"outputs": [" a"]}\n'），可以使用JSON包将其反序列化为Python字典。我们可以使用以下代码迭代每个流式文本行并返回文本响应：
body = {"inputs": "what is life", "parameters": {"max_new_tokens":400}}
resp = smr.invoke_endpoint_with_response_stream(EndpointName=endpoint_name, Body=json.dumps(body), ContentType="application/json")
event_stream = resp['Body']
for line in LineIterator(event_stream):
resp = json.loads(line)
print(resp.get("outputs")[0], end='')下面的屏幕截图显示了如果通过SageMaker笔记本使用LMI容器调用模型的情况。
选项2：使用Hugging Face TGI容器实现聊天机器人
在前一节中，您了解了如何使用LMI容器部署Falcon-7B-Instruct模型。在本节中，我们将展示如何在SageMaker上使用Hugging Face文本生成推理（TGI）容器执行相同的操作。 TGI是一个开源的、专为部署LLM的解决方案。它包含了针对更快的多GPU推理的张量并行性优化、提高整体吞吐量的动态批处理以及使用flash-attention对包括BLOOM、T5、GPT-NeoX、StarCoder和LLaMa在内的常用模型架构进行优化的transformers代码。
TGI深度学习容器支持使用Server-Sent Events（SSE）进行令牌流。使用令牌流，服务器可以在第一个prefill通行证后直接开始回答，而无需等待所有生成完成。对于非常长的查询，这意味着客户端可以在工作完成的数量级之前开始看到某些事情发生。下图显示了使用TGI容器在SageMaker端点上托管LLM的高级端到端请求/响应工作流程。
要在SageMaker端点上部署Falcon-7B-Instruct模型，我们使用SageMaker Python SDK中的HuggingFaceModel类。我们首先设置以下参数：
hf_model_id = "tiiuae/falcon-7b-instruct" # 来自huggingface.co/models的模型ID
number_of_gpus = 1 # 用于推理和张量并行性的GPU数量
health_check_timeout = 300 # 增加健康检查的超时时间为5分钟，用于下载模型
instance_type = "ml.g5.2xlarge" # 用于部署的实例类型与部署常规Hugging Face模型相比，我们首先需要检索容器URI，并将其提供给带有image_uri指向该图像的HuggingFaceModel模型类。为了在SageMaker中检索新的Hugging Face LLM DLC，我们可以使用SageMaker SDK提供的get_huggingface_llm_image_uri方法。该方法允许我们根据指定的后端、会话、区域和版本检索所需的Hugging Face LLM DLC的URI。有关可用版本的更多详细信息，请参阅HuggingFace Text Generation Inference Containers。
llm_image = get_huggingface_llm_image_uri(
"huggingface",
version="0.9.3"
)然后，我们创建HuggingFaceModel并使用deploy方法将其部署到SageMaker：
endpoint_name = sagemaker.utils.name_from_base("tgi-model-falcon-7b")
llm_model = HuggingFaceModel(
role=role,
image_uri=llm_image,
env={
'HF_MODEL_ID': hf_model_id,
# 'HF_MODEL_QUANTIZE': "bitsandbytes", # 取消注释以量化
'SM_NUM_GPUS': str(number_of_gpus),
'MAX_INPUT_LENGTH': "1900", # 输入文本的最大长度
'MAX_TOTAL_TOKENS': "2048", # 生成的最大长度（包括输入文本）
}
)
llm = llm_model.deploy(
initial_instance_count=1,
instance_type=instance_type,
container_startup_health_check_timeout=health_check_timeout,
endpoint_name=endpoint_name,
)与LMI容器相比，主要区别在于在调用端点时通过提供stream=true作为调用请求有效负载的一部分来启用响应流。以下代码是用于使用流式传输调用TGI容器的有效负载示例：
body = {
"inputs":"tell me one sentence",
"parameters":{
"max_new_tokens":400,
"return_full_text": False
},
"stream": True
}然后，您可以使用以下命令调用端点并接收流式响应：
from sagemaker.base_deserializers import StreamDeserializer
llm.deserializer=StreamDeserializer()
resp = smr.invoke_endpoint_with_response_stream(EndpointName=llm.endpoint_name, Body=json.dumps(body), ContentType='application/json')TGI容器的响应内容类型在x-amzn-sagemaker-content-type中显示为text/event-stream。我们使用StreamDeserializer将响应反序列化为EventStream类，并使用与LMI容器部分中使用的相同的LineIterator类解析响应体。
请注意，来自TGI容器的流式响应将返回一个二进制字符串（例如，b`data:{"token": {"text": " sometext"}}`），可以再次使用JSON包将其反序列化为Python字典。我们可以使用以下代码迭代每个流式文本行并返回文本响应：
event_stream = resp['Body']
start_json = b'{'
for line in LineIterator(event_stream):
if line != b'' and start_json in line:
data = json.loads(line[line.find(start_json):].decode('utf-8'))
if data['token']['text'] != stop_token:
print(data['token']['text'],end='')以下屏幕截图显示了通过SageMaker笔记本使用TGI容器调用模型的效果。
在SageMaker Studio上运行聊天机器人应用
在这个使用案例中，我们使用Streamlit在SageMaker Studio上构建一个动态聊天机器人，它调用部署在SageMaker实时端点上的Falcon-7B-Instruct模型以提供流式响应。首先，您可以在笔记本中测试流式响应是否正常工作，如前一节所示。然后，您可以在SageMaker Studio JupyterServer终端中设置Streamlit应用并通过完成以下步骤从浏览器访问聊天机器人UI：
-
在SageMaker Studio中打开系统终端。
-
在SageMaker Studio控制台的顶部菜单中，选择文件，然后选择新建，然后选择终端。
-
安装requirements.txt文件中指定的所需的Python包：
$ pip install -r requirements.txt -
使用在您的帐户中部署的Falcon-7B-Instruct端点名称设置环境变量：
$ export endpoint_name=<您的帐户中部署的Falcon-7B-Instruct端点名称> -
从
streamlit_chatbot_LMI.py或streamlit_chatbot_TGI.py文件启动Streamlit应用程序，该应用程序将根据之前设置的环境变量自动更新脚本中的端点名称：$ streamlit run streamlit_chatbot_LMI.py --server.port 6006 -
要访问Streamlit UI，请将SageMaker Studio URL复制到浏览器的另一个标签页中，并将
lab?替换为proxy/[端口号]/。因为我们将服务器端口设置为6006，所以URL应如下所示：https://<domain ID>.studio.<region>.sagemaker.aws/jupyter/default/proxy/6006/
将上述URL中的域ID和区域替换为您的帐户和区域，以访问聊天机器人UI。您可以在左窗格中找到一些建议的提示，以开始使用。
下面的演示显示了响应流如何革新用户体验。它可以使交互感觉流畅和响应，从而提高用户满意度和参与度。有关聊天机器人实现的更多详细信息，请参阅GitHub存储库。
清理
测试完模型后，作为最佳实践，如果不再需要该端点，则删除端点以节省成本：
# - 删除端点
sm_client.delete_endpoint(EndpointName=endpoint_name)结论
在本文中，我们概述了使用生成式AI构建应用程序的概述，以及面临的挑战以及SageMaker实时响应流如何帮助您解决这些挑战。我们展示了如何构建一个聊天机器人应用程序，以使用SageMaker LMI和HuggingFace TGI容器部署Falcon-7B-Instruct模型使用响应流，使用GitHub上的示例。
立即开始使用LLMs和SageMaker构建您自己的前沿流媒体应用程序！与我们联系，获取专业指导，并释放大型模型流媒体在您的项目中的潜力。