在设施中部署的计算机视觉模型可以发送预测结果到Kafka，这使得能够将模型直接集成到现有的Kafka事件流基础设施中。例如，如果装配线上的产品不符合质量标准，产品的元数据——装配线位置、检测到的缺陷、缺陷在产品上的位置——可以存储在集中系统中。
在本指南中，将展示如何使用Apache Kafka广播计算机视觉预测。将使用开源项目Roboflow Inference来部署计算机视觉模型，然后将预测发送给Kafka消费者。
以下是Kafka接收器接受模型预测的示例：
假设已经设置了Kafka接收器（代理）。如果没有设置Kafka接收器，请参考官方Apache Kafka快速入门指南以获取设置说明。
需要一个计算机视觉模型来构建Kafka流逻辑。在本指南中，将使用Roboflow Inference部署模型。可以在硬件上部署任何在Roboflow上训练或上传的模型。要了解在Roboflow上训练模型，请参考Roboflow入门指南。
将部署一个验证瓶盖完整性的模型。该模型可以识别瓶盖是否正确密封、松动或缺失。此模型可用于装配线以确保饮料瓶的质量。
完整的数据集和模型可在Roboflow Universe上找到。要开始，请安装Inference：
pip install inference
然后，创建一个新的Python文件并添加以下代码：
from inference import InferencePipeline
from inference.core.interfaces.stream.sinks import render_boxes
pipeline = InferencePipeline.init(
model_id="bottle-cap-integrity/7",
video_reference=0, # 视频路径，设备ID（通常是内置摄像头的0），或RTSP流URL
on_prediction=render_boxes,
)
pipeline.start()
pipeline.join()
在此代码中，使用InferencePipeline在视频流上运行模型。0值对应于摄像头的ID。0代表设备上的默认摄像头。也可以指定RTSP流URL，如果有一个RTSP流，希望在其上运行推理。
将上述代码中的bottle-cap-integrity/7替换为Roboflow模型ID。
要运行上述代码，需要将Roboflow API密钥设置在名为ROBOFLOW_API_KEY的环境变量中：
export ROBOFLOW_API_KEY=""
当运行此脚本时，视频流将显示在屏幕上。模型的检测结果将用边界框在流中注释。
可以使用Python将消息发送到Kafka主题。在可以发送消息之前，需要创建一个主题。如何做到这一点将取决于使用的Kafka代理。
创建一个描述性名称的主题。此主题将接收来自模型的消息。然后，安装kafka-python包：
pip install kafka-python
此包具有可以使用的实用程序，以Kafka协议发送消息。接下来，创建一个新的Python文件并添加以下代码：
import json
from kafka import KafkaProducer
from inference import InferencePipeline
from inference.core.interfaces.stream.sinks import render_boxes
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
TOPIC_NAME = "bottle-cap-integrity"
future = producer.send(TOPIC_NAME, b'raw_bytes')
def on_prediction(predictions, video_frame):
render_boxes(predictions=predictions, video_frame=video_frame)
predictions = json.dumps(predictions)
predictions_as_bytes = predictions.encode('utf-8')
producer.send(TOPIC_NAME, predictions_as_bytes)
pipeline = InferencePipeline.init(
model_id="bottle-cap-integrity/7",
video_reference=0,
on_prediction=on_prediction,
confidence=0.3,
)
pipeline.start()
pipeline.join()
在上述代码中，将localhost:9002替换为Kafka接收器正在运行的主机名和端口。将TOPIC_NAME的值替换为想要发送消息的主题。将bottle-cap-integrity/7替换为Roboflow模型ID。将video_reference=0替换为想要运行模型的视频流。默认摄像头应该具有ID 0。也可以传入RTSP流URL或视频文件名。
上述代码连接到Kafka消费者。然后，在传入的视频流的帧上运行指定的模型。所有预测都序列化为字节，然后发送到Kafka接收器。
当准备好时，运行上述代码。应该看到：显示模型预测的窗口打开；预测结果进入Kafka接收器。下面的视频显示了上述两个输出，表明脚本运行成功。
如果预测结果没有进入服务器，请检查是否已创建主题，是否已指定正确的主题名称，以及是否已订阅到想要看到服务器输出的窗口中的正确主题。
可以使用Apache Kafka将计算机视觉预测从设备发送到服务器。这在计算机视觉部署中有许多应用。例如，可以部署NVIDIA Jetsons在装配线上运行计算机视觉模型，然后将每个模型的所有结果收集到中央服务器进行进一步处理。可以使用Kafka发送和接收模型结果。
在本指南中，介绍了如何使用Apache Kafka框架广播计算机视觉模型预测。展示了如何使用Roboflow Inference部署模型，然后如何使用kafka-python框架将预测发送给Kafka接收器。