通过适用于 Jetson 的 NVIDIA Metropolis 微服务将生成式 AI 引入边缘

其次，由于生成式 AI 应用需要一些外部接口（如指令），因此需要应用接受 REST API 请求。

最后，应用必须容器化，以便与其他微服务无缝集成。图 5 显示了在 Redis 上进行 NanoOwl 对象检测和元数据输出的示例。

准备生成式 AI 应用

本参考示例使用的是 NanoOwl。任何可从 Python 调用加载和推理函数的模型，也同样适用这些步骤。本文通过几个 Python 代码示例，介绍将生成式 AI 与 Metropolis 微服务相结合的主要思路，但为了重点介绍通用方法，我们省略了一些代码。如需进一步了解完整的实现方式，请参阅 /NVIDIA-AI-IOT/mmj_genai GitHub 项目上的参考示例。

如要准备与 Metropolis 微服务集成的生成式 AI 模型，请按以下步骤操作：

1. 调用 predict 函数进行模型推理
2. 使用 jetson-utils 程序库添加 RTSP I/O
3. 使用 Flask 添加一个用于指令更新的 REST 端点
4. 使用 mmj_utils 生成叠加层
5. 使用 mmj_utils 与 VST 交互以获取数据流
6. 使用 mmj_utils 将元数据输出到 Redis

调用预测函数进行模型推理

NanoOwl 将生成式 AI 模型封装在一个 OwlPredictor 类中。该类实例化时，会将模型加载到内存中。如要对图像和文本输入进行推理，可调用预测函数来获取输出结果。

本例中，输出结果是检测到的对象的边界框和标签列表。

import PIL.Imageimport timeimport torchfrom nanoowl.owl_predictor import OwlPredictor image = PIL.Image.open("my_image.png")prompt = ["an owl", "a person"] #Load model predictor = OwlPredictor( "google/owlvit-base-patcp2", image_encoder_engine="../data/owlvit_image_encoder_patcp2.engine" )#Embed Texttext_encodings = predictor.encode_text(text) #Inferenceoutput = predictor.predict(image=image, text=prompt, text_encodings=text_encodings, threshold=0.1, pad_square=False)

大多数生成式 AI 模型都有类似的 Python 界面。在有图像和文本输入时，必须加载模型，模型可以从指令和图像中推理出一些输出。如要导入自己的生成式 AI 模型，可以将其封装在一个类中，并实现类似 OwlPredictor 类的接口。

使用 jetson-utils 程序库添加 RTSP 输入/输出

您可以使用 jetson-utils 程序库添加 RTSP 视频流输入。该库提供的 videoSource 和 videoOutput 类可用于从 RTSP 流中捕获帧，并在新的 RTSP 流中输出帧。

from jetson_utils import videoSource, videoOutput stream_input = "rtsp://0.0.0.0:8554/input"stream_output = "rtsp://0.0.0.0:8555/output" #Create stream I/Ov_input = videoSource(stream_input)v_output = videoOutput(stream_output) while(True): image = v_input.Capture() #get image from stream output = predictor.predict(image=image, text=prompt, ...) new_image = postprocess(output) v_output.Render(new_image) #write image to stream

该代码示例从 RTSP 流中捕获帧，然后将其传递给模型推理函数。根据模型输出创建新图像，并呈现在输出 RTSP 流中。

使用 Flask 添加一个用于指令更新的 REST 端点

许多生成式 AI 模型都能接受一些种类的指令或文本输入。为了让用户或其他服务能够动态更新指令，可使用 Flask 添加一个 REST 端点来接受指令更新并将其传递给模型。

为使 Flask 服务器更容易与您的模型集成，可创建一个调用时可在其自己的线程中启动 Flask 服务器的封装类。更多信息，参见 /NVIDIA-AI-IOT/mmj_genai GitHub 项目。

from flask_server import FlaskServer #Launch flask server and connect queue to receive prompt updates flask_queue = Queue() #hold prompts from flask input flask = FlaskServer(flask_queue)flask.start_flask() while(True): ... if not flask_queue.empty(): #get prompt update prompt = flask_queue.get() output = predictor.predict(image=image, text=prompt, ...) ...

通过一个队列连接您的主脚本和 Flask 端点，该队列可容纳任何传入的指令更新。当向 REST 端点发送 GET 请求时，Flask 服务器会将更新的指令放入队列。然后，您的主循环就可以检查队列中的新指令并将其传递给模型，以便对更新的类进行推理。

使用 mmj_utils 生成叠加层

带有可视化叠加层的模型输出对于计算机视觉任务很有帮助（图 6）。以对象检测模型为例，您可以将模型生成的边界框和标签叠加到输入图像上，以便查看模型检测到每个对象的位置。

图 6. 使用 mmj_utils 生成的检测叠加层

为此，请使用 mmj_utils 程序库中名为 DetectionGenerationCUDA 的实用程序类。该库依赖于 jetson_utils，后者提供用于生成叠加层的 CUDA 加速函数。

from mmj_utils.overlay_gen import DetectionOverlayCUDA overlay_gen = DetectionOverlayCUDA(draw_bbox=True, draw_text=True, text_size=45) #make overlay object while(True): ... output = predictor.predict(image=image, text=prompt, ...) #Generate overlay and output text_labels = [objects[x] for x in output.labels] bboxes = output.boxes.tolist() image = overlay_gen(image, text_labels, bboxes)#generate overlay v_output.Render(image)

您可以使用多个关键字参数实例化 DetectionGenerationCUDA 对象，以便根据需要调整文字大小、边框大小和颜色。有关使用 mmj_utils 生成叠加层的更多信息，请参阅 /NVIDIA-AI-IOT/mmj_utils GitHub 资源库。

如要生成覆盖层，请调用该对象并传递输入图像、标签列表和模型生成的边界框。然后，它会在输入图像上绘制标签和边界框，并返回带覆盖层的修改后图像。修改后的图像可呈现在 RTSP 流上。

使用 mmj_utils 与 VST 交互以获取数据流

VST 可以帮助管理 RTSP 流，并提供一个实用的 Web UI 来查看输入和输出流。如要与 VST 集成，可直接使用 VST REST API 或 mmj_utils 中的 VST 类，后者封装了 VST REST API。

请勿在 Python 脚本中对 RTSP 输入流进行硬编码，而是从 VST 抓取 RTSP 流链接。该链接可能来自 IP 摄像头或其他通过 VST 管理的视频流来源。

from mmj_utils.vst import VST vst = VST("http://0.0.0.0:81")vst_rtsp_streams = vst.get_rtsp_streams()stream_input = vst_rtsp_streams[0] v_input = videoSource(stream_input)...

这将连接到 VST 并抓取第一个有效的 RTSP 链接。可以在这里添加更复杂的逻辑，以连接到特定来源或动态更改输入。

使用 mmj_utils 向 Redis 输出元数据

生成式 AI 模型生成的元数据可被下游的其他服务用于生成分析和洞察。

在本例中，NanoOwl 会输出检测到的对象的边界框。您可以在 Redis 数据流上以 Metropolis Schema 的形式输出这些信息，分析服务可以捕获这些信息。在 mmj_utils 程序库中，有一个类可以帮助在 Redis 上生成检测元数据。

from mmj_utils.schema_gen import SchemaGenerator schema_gen = SchemaGenerator(sensor_id=1, sensor_type="camera", sensor_loc=[10,20,30])schema_gen.connect_redis(aredis_host=0.0.0.0, redis_port=6379, redis_stream="owl") while True: ... output = predictor.predict(image=image, text=prompt, ...) #Output metadata text_labels = [objects[x] for x in output.labels]schema_gen(text_labels, bboxes)

您可以使用输入摄像头数据流的相关信息实例化一个 SchemaGenerator 对象并连接到 Redis。然后，可以通过传递文本标签和模型生成的边界框来调用该对象。检测信息会被转换为 Metropolis Schema 并输出到 Redis 供其他微服务使用。

图 7. 生成式 AI 应用

应用部署

如要部署应用，可以设置 Ingress 和 Redis 等平台服务，然后通过 docker compose 将自定义生成式 AI 容器与 VST 等应用服务相结合。

在主应用准备好所有必要的 I/O 和微服务集成后（图 7），就可以部署应用并与 Metropolis 微服务连接。

1. 将生成式 AI 应用容器化。
2. 设置必要的平台服务
3. 使用 docker compose 启动应用
4. 实时查看输出结果

将生成式 AI 应用容器化

部署的第一步是使用 Docker 对生成式 AI 应用进行容器化。

一个简单的方法是使用 jetson-containers 项目。该项目提供了一种简单的方法来为 Jetson 构建 Docker 容器，以支持包括生成式 AI 模型在内的机器学习应用。使用 jetson-containers 创建一个包含必要依赖项的容器，然后进一步自定义容器，在其中加入应用代码和运行生成式 AI 模型所需的任何其他软件。

如需进一步了解如何构建适用于 NanoOwl 示例的容器，请参阅 GitHub 项目中的 /src/readme 文件。

设置必要的平台服务

接下来，设置 Metropolis 微服务提供的必要平台服务。这些平台服务提供了使用 Metropolis 微服务部署应用所需的许多功能。

这个参考生成式 AI 应用只需要 Ingress、Redis 和 Monitoring 平台服务。平台服务可以通过 APT 快速安装，并使用 systemctl 启动。

如需进一步了解如何安装和启动必要的平台服务，请参阅 适用于 Jetson 的 Metropolis 微服务快速入门指南：

https://docs.nvidia.com/moj/index.html