Unstrukturierte Datenverarbeitung von der Cloud bis zum Edge
Unstrukturierte Daten machen mittlerweile etwa 80 % der Daten aus, die wir verarbeiten, von Text und Bildern bis hin zu Videos, Audio und komplexeren Formaten. Mit wachsendem Umfang steigt auch der Bedarf an besseren Möglichkeiten, sie zu verarbeiten und zu analysieren.
In einem kürzlich veranstalteten Webinar teilte Tim Spann, Principal Developer Advocate bei Zilliz, seine Erkenntnisse zum Umgang mit unstrukturierten Daten, insbesondere von Cloud-Umgebungen bis hin zu Edge-Geräten. Er hob außerdem hervor, wie Vektordatenbanken wie Milvus entscheidend dafür sind, diese Daten sinnvoll nutzbar zu machen.
In diesem Beitrag fassen wir Tims wichtigste Punkte zusammen, diskutieren die Herausforderungen beim Umgang mit unstrukturierten Daten und zeigen, wie Vektordatenbanken hier den Weg weisen. Außerdem betrachten wir, wie Edge-Geräte, die von diesen Datenbanken unterstützt werden, neue Fortschritte in der KI vorantreiben. Wenn Sie an weiteren Details interessiert sind, empfehlen wir Ihnen, die vollständige Aufzeichnung von Tims Vortrag auf YouTube anzusehen.
Unstrukturierte Daten verstehen
Unstrukturierte Daten beziehen sich auf Informationen, denen ein vordefiniertes Format oder Schema fehlt. Im Gegensatz zu strukturierten Daten, die in Datenbanken mit bestimmten Feldern und Formaten ordentlich organisiert sind, sind unstrukturierte Daten roh und ungeordnet. Sie umfassen häufig:
Text: E-Mails, Word-Dokumente, Social-Media-Beiträge
Bilder: Fotos, Screenshots, Diagramme
Videos: Aufgezeichnete Meetings, Überwachungsaufnahmen, Multimedia-Inhalte
Audio: Sprachaufnahmen, Musikdateien, Podcasts
Mit dem Anstieg digitaler Inhalte sind unstrukturierte Daten exponentiell gewachsen und bieten Unternehmen sowohl Chancen als auch Herausforderungen. Um diesen Informationsreichtum zu nutzen, können Unternehmen:
Das Kundenerlebnis verbessern durch die Analyse von Kundeninteraktionen und Feedback
Die betriebliche Effizienz steigern durch Erkenntnisse aus Sensordaten und Systemprotokollen
Innovation vorantreiben durch die Nutzung von Multimedia-Inhalten für die Produktentwicklung
Traditionelle Datenbanken, die für die Verarbeitung strukturierter Daten konzipiert sind, haben mit der Unregelmäßigkeit unstrukturierter Daten Schwierigkeiten, wodurch es schwierig wird, zu suchen, Abfragen durchzuführen oder aussagekräftige Erkenntnisse zu gewinnen. Hier werden Vektordatenbanken unverzichtbar, da sie eine Lösung für die effiziente Verwaltung und Abfrage unstrukturierter Informationen bieten.
Warum eine Vektordatenbank für die Verarbeitung unstrukturierter Daten verwenden?
Eine Vektordatenbank ist eine spezialisierte Datenbank, die darauf ausgelegt ist, Vektordarstellungen von Daten (auch bekannt als Vektoreinbettungen) zu speichern, zu indexieren und abzufragen, die häufig aus unstrukturierten Daten wie Text, Bildern und Audio abgeleitet werden. Diese Datenbanken ermöglichen effiziente Ähnlichkeitssuchen in einem hochdimensionalen Raum und sind damit ideal für semantische Suche, Verarbeitung natürlicher Sprache (NLP), Empfehlungssysteme, Bildsuchen und Retrieval Augmented Generation (RAG).
Vektordatenbanken wie Milvus und Zilliz Cloud bieten über leistungsstarke Suchfunktionen hinaus eine Reihe von Funktionen, die sie für die Verwaltung unstrukturierter Daten entscheidend machen.
Jenseits der Hochleistungssuche
Während Vektordatenbanken oft mit schnellen Ähnlichkeitssuchfunktionen in Verbindung gebracht werden, geht ihr Nutzen weit darüber hinaus. Sie bieten weitere wesentliche Funktionen, die die effektive Verwaltung und Nutzung unstrukturierter Daten sicherstellen:
CRUD-Operationen: Vektordatenbanken ermöglichen Ihnen, wie traditionelle Datenbanken, Daten zu erstellen, zu lesen, zu aktualisieren und zu löschen (CRUD). Dadurch bleibt sichergestellt, dass die grundlegenden Operationen trotz der Arbeit mit komplexen Datentypen intuitiv und zugänglich bleiben.
Datenaktualität: Eine der zentralen Stärken von Vektordatenbanken besteht darin, sicherzustellen, dass Ihre Daten auf dem neuesten Stand bleiben. In Anwendungsfällen wie Empfehlungssystemen ist die Aktualität der Daten entscheidend, um genaue Erkenntnisse zu generieren.
Persistenz: Im Gegensatz zu In-Memory-Datenstrukturen bieten Vektordatenbanken persistenten Speicher, was bedeutet, dass Ihre Daten sicher gespeichert und auch nach Systemneustarts zugänglich sind.
Verfügbarkeit: Vektordatenbanken halten Daten für Echtzeitabfragen und -abrufe jederzeit zugänglich. Dadurch wird sichergestellt, dass Sie auch bei wachsendem Datenbestand weiterhin schnelle und effiziente Suchen durchführen können, was KI-gestützte Entscheidungen ohne Verzögerung ermöglicht.
Skalierbarkeit: Wenn das Datenvolumen wächst, skalieren Vektordatenbanken effizient, um dieses Wachstum ohne Leistungseinbußen zu bewältigen. Dies ist entscheidend, wenn Milliarden unstrukturierter Datenpunkte verwaltet werden.
Vollständiges Datenmanagement
Über die oben genannten Funktionen hinaus bieten Vektordatenbanken robuste Tools zur Verwaltung Ihrer unstrukturierten Daten, einschließlich Datenaufnahme, Indizierung und Abfrage, wodurch eine effiziente Wiedergewinnung und Analyse selbst der größten Datensätze ermöglicht wird. Funktionen wie Backup und Migration stellen sicher, dass Ihre Daten sicher und wiederherstellbar bleiben, was Vektordatenbanken zu einer zuverlässigen Wahl für die Verwaltung kritischer Informationen macht.
Operative Einfachheit
Vektordatenbanken sind außerdem darauf ausgelegt, Bereitstellung und Verwaltung zu vereinfachen:
Cloud- oder On-Premise-Bereitstellung: Viele zweckgebundene Vektordatenbanken sind vielseitig und können problemlos in öffentlichen Clouds oder in On-Premise-Infrastrukturen bereitgestellt werden, wodurch Flexibilität entsprechend den Anforderungen der Organisation gewährleistet wird.
Observability: Viele Vektordatenbanken bieten Monitoring-Tools, um den Zustand und die Leistung der Datenbank zu verfolgen, sodass Benutzer ihre Systeme proaktiv optimieren können.
Mandantenfähigkeit: Vektordatenbanken wie Zilliz Cloud unterstützen mehrere Benutzer oder Anwendungen und gewährleisten sicheren Zugriff sowie isolierte Datenverarbeitung für jeden Mandanten. Dadurch sind sie ideal für groß angelegte Bereitstellungen, bei denen mehrere Teams isolierte Datenumgebungen benötigen.
Die Rolle von Milvus bei der Verwaltung unstrukturierter Daten
Milvus ist eine Open-Source-Vektordatenbank, die darauf ausgelegt ist, hochdimensionale Vektoren im Milliardenmaßstab effizient zu speichern, zu indizieren und abzurufen. Damit ist sie ideal für Anwendungsfälle mit KI und maschinellem Lernen, wie Empfehlungssysteme, Bilderkennung oder retrieval-augmented generation (RAG).
Milvus bietet mehrere Bereitstellungsoptionen, um unterschiedlichen Kundenanforderungen gerecht zu werden.
Open-Source-Milvus: Open Source, selbstverwaltet und kann mit Community-Support auf jeder Maschine gehostet werden. Milvus bietet außerdem mehrere Bereitstellungsoptionen für verschiedene Anforderungen und Umgebungen, darunter Milvus Lite, Milvus Standalone und Milvus Distributed. Weitere Details finden Sie in der Milvus-Dokumentation.
Zilliz Cloud: vollständig verwaltete Version von Milvus, für die Cloud neu konzipiert und auf führenden Public Clouds wie AWS, GCP und Azure verfügbar.
Zilliz BYOC: unternehmensreifes Milvus für private VPCs und kann in Ihrer Virtual Private Cloud bereitgestellt werden.
Fig 1- Ways of deploying Milvus .png
Abb. 1: Möglichkeiten zur Bereitstellung von Milvus
Hauptfunktionen von Milvus
Milvus ist auf hohe Leistung und Skalierbarkeit ausgelegt und bietet Funktionen wie:
Figure- Milvus key features .png
Abbildung: Hauptfunktionen von Milvus
Mandantenfähigkeit: Milvus ermöglicht es mehreren Benutzern oder Anwendungen, auf demselben System zu arbeiten, ohne sich gegenseitig zu stören, und bietet sicheren Zugriff auf isolierte Daten.
Hardwarebeschleunigte Datenverarbeitung: Optimiert, um Hardware wie GPUs zu nutzen, wodurch es für ressourcenintensive Aufgaben wie die Inferenz von KI-Modellen schneller und effizienter wird.
Sprach- und API-Unterstützung: Milvus ist äußerst vielseitig und unterstützt Python, Java, Golang, NodeJS usw., wodurch es für ein breites Spektrum von Entwicklern zugänglich ist.
Skalierbare und elastische Architektur: Milvus ist darauf ausgelegt, wachsende Datenanforderungen zu bewältigen. Es skaliert automatisch, um die Nachfrage zu erfüllen, und stellt sicher, dass die Leistung bei wachsendem Datenvolumen optimal bleibt.
Unterstützung vielfältiger Indizes: Es unterstützt mehrere Arten von Indizes, darunter HNSW, PQ, Binary und DiskANN, und ermöglicht Flexibilität dabei, wie Daten gespeichert und durchsucht werden.
Anpassbare Konsistenz: Milvus ermöglicht es Ihnen, die Konsistenzstufen für Ihre Daten anzupassen, sodass Sie je nach Anforderungen der Anwendung zwischen Leistung und Datengenauigkeit abwägen können.
Ausführlichere Informationen finden Sie in der Milvus-Dokumentation.
Technologien, die Milvus für verschiedene Anwendungsfälle antreiben
Milvus wird von einer Reihe von Technologien angetrieben, die auf verschiedene Arten von Anwendungen zugeschnitten sind und Leistung, Flexibilität und Skalierbarkeit in unterschiedlichen Umgebungen sicherstellen:
Figure- Milvus technologies for various use cases .png
Abbildung: Milvus-Technologien für verschiedene Anwendungsfälle
Compute-Typen: Milvus ist für verschiedene Hardwareumgebungen optimiert, einschließlich AVX512 (eine Reihe von CPU-Anweisungen für Hochgeschwindigkeits-Computing), Neon für SIMD (Single Instruction, Multiple Data) und GPU-Beschleunigung. Dies stellt sicher, dass Milvus Hochleistungshardware nutzen kann, um schnelle Verarbeitung und kosteneffiziente Skalierbarkeit zu liefern.
Suchtypen: Milvus unterstützt eine breite Palette von Suchmethoden, wie Top-K-ANN (Approximate Nearest Neighbors) zum Finden der ähnlichsten Datenpunkte, Bereichs-ANN, Sparse- & Dense-Suchen und gefilterte Suchen. Diese ermöglichen es uns, die Suchfunktionalität auf die spezifischen Anforderungen unserer Anwendungen zuzuschneiden, sei es die Identifizierung ähnlicher Bilder, Videos oder Texte.
Mandantenfähigkeit: Milvus unterstützt Collection- und Partitionsverwaltung, wodurch Mandantenfähigkeit ermöglicht wird. Diese Funktion erlaubt es verschiedenen Teams oder Anwendungen, dieselbe Datenbank gemeinsam zu nutzen, ohne die Daten der jeweils anderen zu beeinträchtigen, und gewährleistet eine sichere und isolierte Datenverarbeitung.
Indextypen: Milvus bietet eine breite Palette von 15 Indextypen, darunter HNSW (Hierarchical Navigable Small Worlds) für Hochgeschwindigkeitssuchen, PQ (Product Quantization) für kompakte Speicherung und DiskANN für die Verarbeitung riesiger Datensätze. Sie können den Indextyp wählen, der je nach Ihrem Anwendungsfall Leistung, Genauigkeit und Kosten am besten ausbalanciert.
Diese Technologien machen Milvus zu einer guten Wahl für eine breite Palette von Anwendungen, von groß angelegten Cloud-Bereitstellungen bis hin zu ressourcenbeschränkten Edge-Geräten.
Der Wandel hin zu Edge Computing
Da die Nachfrage nach Echtzeit-Einblicken aus unstrukturierten Daten wächst, stellen viele Unternehmen fest, dass es nicht ausreicht, sich ausschließlich auf Cloud-Umgebungen zu verlassen. Dies hat zu einer Verlagerung hin zu Edge Computing geführt, bei dem die Verarbeitung näher an dem Ort erfolgt, an dem die Daten erzeugt werden, was deutliche Vorteile für zeitkritische KI-Anwendungen bietet.
Edge Computing bezeichnet die Praxis, Daten näher an dem Ort zu verarbeiten, an dem sie erzeugt werden, häufig auf Edge-Geräten wie Sensoren, Kameras und IoT-Geräten. Anstatt Rohdaten zur Verarbeitung an zentrale Rechenzentren zurückzusenden, ermöglicht Edge Computing Berechnungen auf den Geräten selbst oder an nahegelegenen Standorten.
Mit der zunehmenden Verfügbarkeit leistungsstarker Edge-Geräte ist es möglich geworden, Machine-Learning- und KI-Modelle direkt auf diesen Geräten bereitzustellen. Dieser Ansatz ermöglicht Echtzeit-Entscheidungsfindung in Anwendungen wie autonomen Fahrzeugen, Smart Cities und industrieller Automatisierung.
Warum Daten am Edge verarbeiten?
Die Verarbeitung unstrukturierter Daten am Edge bietet mehrere Vorteile:
Geringe Latenz: Da Daten nahe an ihrer Quelle verarbeitet werden, gibt es nur minimale Verzögerungen beim Erhalt von Ergebnissen.
Reduzierte Bandbreite: Durch die lokale Verarbeitung von Daten reduzieren Sie die Notwendigkeit, große Mengen an Rohdaten zurück in die Cloud zu übertragen.
Datenschutz und Sicherheit: Sensible Daten können auf dem Edge-Gerät verbleiben, wodurch das Risiko einer Offenlegung während der Übertragung minimiert wird.
Edge-KI und Vektordatenbanken
Während Edge Computing die Datenverarbeitung näher an die Quelle auf Geräten wie Sensoren, Kameras und kleinen Geräten wie dem Raspberry Pi bringt, ist es die Vektordatenbank, die diese Geräte wirklich dazu befähigt, die wachsende Flut unstrukturierter Daten in Echtzeit zu bewältigen.
In einer industriellen Umgebung überwachen Edge-Geräte beispielsweise die Maschinenleistung und erkennen potenzielle Probleme mithilfe von Sensordaten. Eine Vektordatenbank wie Milvus befähigt das Edge-Gerät, eingehende Daten schnell mit historischen Mustern zu vergleichen und Anomalien zu identifizieren, die auf Wartungsbedarf hinweisen. Ohne die Vektordatenbank müsste das Edge-Gerät entweder alle Rohdaten zur Verarbeitung an einen Cloud-Server senden (was Verzögerungen und höhere Kosten verursachen würde) oder riskieren, kritische Erkenntnisse zu verpassen.
Milvus Lite: Edge-Geräte mit KI-Fähigkeiten ausstatten
Milvus Lite ist die leichtgewichtige Version von Milvus, die speziell für ressourcenbeschränkte Umgebungen wie Edge-Geräte entwickelt wurde. Sie bietet alle wesentlichen Funktionen einer Vektordatenbank, ist jedoch für die Bereitstellung auf kleinerer, weniger leistungsstarker Hardware optimiert. Das macht sie zu einer idealen Lösung, um Edge-Geräte in die Lage zu versetzen, komplexe KI-Aufgaben zu bewältigen, selbst in Umgebungen mit begrenzten Ressourcen.
Mit Milvus Lite, das auf einem Edge-Gerät läuft, wird das Gerät in einen KI-gestützten Datenprozessor verwandelt. Diese Kombination ermöglicht es dem Edge-Gerät, lokalisierte Bilderkennung, Video-Ähnlichkeitssuchen oder sogar Verarbeitung natürlicher Sprache durchzuführen. In einer Einzelhandelsumgebung könnte beispielsweise ein intelligentes Kassensystem, das von Milvus Lite unterstützt wird, Produkte sofort erkennen und sie mit einer gespeicherten Vektordatenbank von Artikeln abgleichen, wodurch Transaktionen beschleunigt werden, ohne dass cloudbasierte Systeme erforderlich sind.
Sehen wir uns ein praktisches Beispiel an, in dem Milvus ein Edge-Gerät antreibt, das auf einem Raspberry Pi läuft.
Aufbau eines Echtzeit-Posenschätzungssystems mit Raspberry Pi und Milvus
Raspberry Pi ist eine Reihe kleiner, erschwinglicher Einplatinencomputer, die von der Raspberry Pi Foundation entwickelt wurden. In diesem Beispiel verwenden wir Raspberry Pi als Edge-Gerät und Milvus Lite zur Verarbeitung unstrukturierter Daten.
Dieses Posenschätzungssystem wird Videostreams erfassen, sie zur Schätzung menschlicher Posen verarbeiten und die extrahierten Merkmalsvektoren in Milvus speichern, um eine effiziente Ähnlichkeitsabfrage zu ermöglichen. Zusätzlich integrieren wir Slack für Benachrichtigungen, die uns alarmieren, wenn bestimmte Posen erkannt werden.
Sie können den Quellcode dieses Beispiels in Tims Notebook auf GitHub ansehen.
Voraussetzungen
Raspberry Pi (mit installiertem GStreamer und Python)
Milvus-Einrichtung (entweder lokal oder auf einem Remote-Server)
Slack-Konto für Benachrichtigungen
Importieren der notwendigen Bibliotheken und erste Einrichtung
Beginnen Sie mit dem Import der notwendigen Bibliotheken und dem Einrichten einiger erster Konfigurationen:
import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib
import os
import argparse
import multiprocessing
import numpy as np
import setproctitle
import cv2
import time
from datetime import datetime
import uuid
import glob
import torch
from torchvision import transforms
from PIL import Image
import timm
from sklearn.preprocessing import normalize
from timm.data import resolve_data_config
from timm.data.transforms_factory import create_transform
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from pymilvus import connections
from pymilvus import utility
from pymilvus import FieldSchema, CollectionSchema, DataType, Collection
from pymilvus import MilvusClient
from pymilvus import MilvusClient
import hailo
from hailo_rpi_common import (
get_default_parser,
QUEUE,
get_caps_from_pad,
get_numpy_from_buffer,
GStreamerApp,
app_callback_class,
)
DIMENSION = 512
MILVUS_URL = "https://in05-7bd87b945683c8d.serverless.gcp-us-west1.cloud.zilliz.com"
COLLECTION_NAME = "rpipose"
TOKEN = os.environ["ZILLIZ_TOKEN"]
PATH = "/opt/demo/images"
time_list = [ 0, 5, 10, 20, 30, 40, 50, 59 ]
Der obige Code importiert verschiedene Bibliotheken für Videoverarbeitung (GStreamer), Bildverarbeitung (OpenCV), Deep Learning (PyTorch, timm), Datenbankoperationen (pymilvus) und Kommunikation (Slack SDK). Außerdem importiert er die Hailo-Bibliothek für die Arbeit mit dem KI-Beschleuniger, der entwickelt wurde, um Deep-Learning-Anwendungen auf Edge-Geräten zu ermöglichen. Wir richten globale Variablen für die Milvus-Datenbankkonfiguration ein und definieren eine Liste von Zeiten für periodische Aktionen. MILVUSURL UND TOKEN authentifizieren uns bei der Zilliz-Cloud.
Einrichtung der Milvus-Datenbank
Als Nächstes richten wir unsere Verbindung zur Milvus-Vektordatenbank ein. Wir werden Milvus verwenden, um die extrahierten Merkmale aus dem Posenschätzungsmodell zu speichern.
# Connect to Milvus
# Milvus Lite
# milvus_client = MilvusClient(uri="pipose.db")
# Cloud-Server
milvus_client = MilvusClient( uri=MILVUS_URL, token=TOKEN )
# Milvus-Collection erstellen, die die ID, den Dateipfad des Bildes und das Bild-Embedding enthält
fields = [
FieldSchema(name='id', dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name='label', dtype=DataType.VARCHAR, max_length=200),
FieldSchema(name='lefteye', dtype=DataType.VARCHAR, max_length=200),
FieldSchema(name='righteye', dtype=DataType.VARCHAR, max_length=200),
FieldSchema(name='confidence', dtype=DataType.FLOAT),
FieldSchema(name='vector', dtype=DataType.FLOAT_VECTOR, dim=DIMENSION)
]
schema = CollectionSchema(fields=fields)
milvus_client.create_collection(COLLECTION_NAME, DIMENSION, schema=schema, metric_type="COSINE", auto_id=True)
index_params = milvus_client.prepare_index_params()
index_params.add_index(field_name = "vector", metric_type="COSINE")
milvus_client.create_index(COLLECTION_NAME, index_params)
Hier richten wir unsere Milvus-Datenbank ein. Wir erstellen eine Client-Verbindung, definieren das Schema für unsere Collection (einschließlich Feldern für ID, Label, Augenpositionen, Konfidenz und den Feature-Vektor), erstellen die Collection und richten einen Index für effiziente Ähnlichkeitssuchen ein.
Slack-Einrichtung für Benachrichtigungen
Als Nächstes richten Sie eine Verbindung zu Slack ein, um Benachrichtigungen zu senden:
slack_token = os.environ["SLACK_BOT_TOKEN"]
client = WebClient(token=slack_token)
Diese Einrichtung ermöglicht es uns, Nachrichten und Dateien über ein als Umgebungsvariable gespeichertes Bot-Token an Slack zu senden.
Implementierung des Feature-Extractors
Unser Posenschätzungssystem benötigt einen Feature-Extractor, um die erkannten Posen in ein Format umzuwandeln, das für die Speicherung in Milvus geeignet ist. Definieren wir dafür eine Klasse.
class FeatureExtractor:
def __init__(self, modelname):
# Das vortrainierte Modell laden
self.model = timm.create_model(
modelname, pretrained=True, num_classes=0, global_pool="avg"
)
self.model.eval()
# Die vom Modell benötigte Eingabegröße abrufen
self.input_size = self.model.default_cfg["input_size"]
config = resolve_data_config({}, model=modelname)
# Die von TIMM für das Modell bereitgestellte Vorverarbeitungsfunktion abrufen
self.preprocess = create_transform(**config)
def __call__(self, imagepath):
# Das Eingabebild vorverarbeiten
input_image = Image.open(imagepath).convert("RGB") # Bei Bedarf in RGB konvertieren
input_image = self.preprocess(input_image)
# Das Bild in einen PyTorch-Tensor umwandeln und eine Batch-Dimension hinzufügen
input_tensor = input_image.unsqueeze(0)
# Inferenz durchführen
with torch.no_grad():
output = self.model(input_tensor)
# Den Feature-Vektor extrahieren
feature_vector = output.squeeze().numpy()
return normalize(feature_vector.reshape(1, -1), norm="l2").flatten()
extractor = FeatureExtractor("resnet34")
Diese FeatureExtractor-Klasse wandelt unsere Bilder in Feature-Vektoren um. Wir verwenden für diese Aufgabe ein vortrainiertes ResNet34-Modell. Die Methode __call__ öffnet ein Bild, verarbeitet es vor, führt es durch das Modell aus und gibt einen normalisierten Feature-Vektor zurück.
Definieren einer Benutzer-Callback-Klasse
Fahren Sie fort und definieren Sie eine einfache Benutzer-Callback-Klasse.
class user_app_callback_class(app_callback_class):
def __init__(self):
super().__init__()
Diese Klasse erbt von app_callback_class und fügt keine neue Funktionalität hinzu. Sie ist ein Platzhalter, der eine zukünftige Anpassung des Callback-Verhaltens ermöglicht.
Erstellen einer Anwendungs-Callback-Funktion
Die Funktion app_callback übernimmt den Großteil der Verarbeitung. Diese Funktion verarbeitet Videoframes, schätzt Posen und speichert die Ergebnisse in Milvus.
def app_callback(pad, info, user_data):
# Den GstBuffer aus den Probe-Informationen abrufen
lefteye = ""
righteye = ""
buffer = info.get_buffer()
# Prüfen, ob der Buffer gültig ist
if buffer is None:
return Gst.PadProbeReturn.OK
# Verwenden von user_data, um die Anzahl der Frames zu zählen
user_data.increment()
string_to_print = f"Frame-Anzahl: {user_data.get_count()}\n"
# Caps vom Pad abrufen
format, width, height = get_caps_from_pad(pad)
# Wenn user_data.use_frame auf True gesetzt ist, können wir den Video-Frame aus dem Buffer abrufen
frame = None
if user_data.use_frame and format is not None and width is not None and height is not None:
# Video-Frame abrufen
frame = get_numpy_from_buffer(buffer, format, width, height)
# Die Erkennungen aus dem Buffer abrufen
roi = hailo.get_roi_from_buffer(buffer)
detections = roi.get_objects_typed(hailo.HAILO_DETECTION)
# Die Erkennungen parsen
for detection in detections:
label = detection.get_label()
bbox = detection.get_bbox()
confidence = detection.get_confidence()
if label == "person":
string_to_print += (f"Erkennung: {label} {confidence:.2f}\n")
# Posenschätzungs-Landmarks aus der Erkennung (falls verfügbar)
landmarks = detection.get_objects_typed(hailo.HAILO_LANDMARKS)
if len(landmarks) != 0:
points = landmarks[0].get_points()
left_eye = points[1] # angenommen, 1 ist der Index für das linke Auge
right_eye = points[2] # angenommen, 2 ist der Index für das rechte Auge
# Die Landmarks sind auf die Bounding Box normalisiert, wir müssen sie auch in die Frame-Größe umrechnen
left_eye_x = int((left_eye.x() * bbox.width() + bbox.xmin()) * width)
left_eye_y = int((left_eye.y() * bbox.height() + bbox.ymin()) * height)
right_eye_x = int((right_eye.x() * bbox.width() + bbox.xmin()) * width)
right_eye_y = int((right_eye.y() * bbox.height() + bbox.ymin()) * height)
string_to_print += (f" Linkes Auge: x: {left_eye_x:.2f} y: {left_eye_y:.2f} Rechtes Auge: x: {right_eye_x:.2f} y: {right_eye_y:.2f}\n")
if user_data.use_frame:
# Marker zum Frame hinzufügen, um Augen-Landmarks anzuzeigen
cv2.circle(frame, (left_eye_x, left_eye_y), 5, (0, 255, 0), -1)
cv2.circle(frame, (right_eye_x, right_eye_y), 5, (0, 255, 0), -1)
if user_data.use_frame:
# Den Frame in BGR konvertieren
framesave = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
user_data.set_frame(framesave)
time_now = datetime.now()
current_time = int(time_now.strftime("%S"))
if current_time in time_list and len(label) > 4:
# Bild speichern
strfilename = PATH + "/personpose.jpg"
cv2.imwrite(strfilename, framesave)
lefteye = (f"x: {left_eye_x:.2f} y: {left_eye_y:.2f}")
righteye = (f"x: {right_eye_x:.2f} y: {right_eye_y:.2f}")
# Slack
try:
response = client.chat_postMessage(
channel="C06NE1FU6SE",
text=(f"Erkennung: {label} {confidence:.2f}")
)
except SlackApiError as e:
# Sie erhalten einen SlackApiError, wenn "ok" False ist
assert e.response["error"]
try:
response = client.chat_postMessage(
channel="C06NE1FU6SE",
text=(f" Linkes Auge: x: {left_eye_x:.2f} y: {left_eye_y:.2f} Rechtes Auge: x: {right_eye_x:.2f} y: {right_eye_y:.2f}\n")
)
except SlackApiError as e:
# Sie erhalten einen SlackApiError, wenn "ok" False ist
assert e.response["error"]
try:
response = client.files_upload_v2(
channel="C06NE1FU6SE",
file=strfilename,
title=label,
initial_comment="Live Camera image ",
)
except SlackApiError as e:
assert e.response["error"]
# Milvus insert
try:
imageembedding = extractor(strfilename)
milvus_client.insert( COLLECTION_NAME, {"vector": imageembedding, "lefteye": lefteye, "righteye": righteye, "label": label, "confidence": confidence})
except Exception as e:
print("An error:", e)
print(string_to_print)
return Gst.PadProbeReturn.OK
Diese Callback-Funktion verarbeitet Videoframes, um Personen zu erkennen, wobei sie sich auf das Extrahieren von Details wie Bounding Box, Konfidenz und Augen-Landmarks konzentriert. Sie markiert erkannte Augenpositionen im Frame und speichert das Frame-Bild, wenn sich eine Person im Frame befindet. Zusätzlich sendet sie Benachrichtigungen an Slack mit Erkennungsdetails und dem markierten Bild. Anschließend extrahiert die Funktion einen Einbettungsvektor aus dem gespeicherten Bild. Sie aktualisiert Milvus mit diesen Daten, einschließlich der Einbettung (vector), Augenkoordinaten (lefteye und righteye), Objektbezeichnung (label) und Erkennungskonfidenz (confidence).
Erstellen der Utility-Funktion für COCO-Keypoints
Erstellen Sie eine Utility-Funktion zum Abrufen von COCO-Keypoints. Dies sind spezifische Punkte am menschlichen Körper, die für die Posenschätzung verwendet werden.
def get_keypoints():
"""Get the COCO keypoints and their left/right flip correspondence map."""
keypoints = {
'nose': 1,
'left_eye': 2,
'right_eye': 3,
'left_ear': 4,
'right_ear': 5,
'left_shoulder': 6,
'right_shoulder': 7,
'left_elbow': 8,
'right_elbow': 9,
'left_wrist': 10,
'right_wrist': 11,
'left_hip': 12,
'right_hip': 13,
'left_knee': 14,
'right_knee': 15,
'left_ankle': 16,
'right_ankle': 17,
}
return keypoints
Diese Funktion gibt ein Wörterbuch zurück, das Körperteile ihren entsprechenden Keypoint-Indizes im COCO-Datensatzformat zuordnet.
Erstellen der GStreamer-Pipeline
Fahren Sie fort und erstellen Sie die Hauptanwendungsklasse, die eine GStreamer-Pipeline einrichtet. Dies ist eine Abfolge von Elementen, die Multimediadaten in einer bestimmten Reihenfolge verarbeiten. Sie wird Live-Videostreams verarbeiten.
# This class inherits from the hailo_rpi_common.GStreamerApp class
class GStreamerPoseEstimationApp(GStreamerApp):
def __init__(self, args, user_data):
# Call the parent class constructor
super().__init__(args, user_data)
# Additional initialization code can be added here
# Set Hailo parameters these parameters should be set based on the model used
self.batch_size = 2
self.network_width = 640
self.network_height = 640
self.network_format = "RGB"
self.default_postprocess_so = os.path.join(self.postprocess_dir, 'libyolov8pose_post.so')
self.post_function_name = "filter"
self.hef_path = os.path.join(self.current_path, '../resources/yolov8s_pose_h8l_pi.hef')
self.app_callback = app_callback
# Set the process title
setproctitle.setproctitle("Hailo Pose Estimation with Milvus")
self.create_pipeline()
def get_pipeline_string(self):
if (self.source_type == "rpi"):
source_element = f"libcamerasrc name=src_0 auto-focus-mode=2 ! "
source_element += f"video/x-raw, format={self.network_format}, width=1536, height=864 ! "
source_element += QUEUE("queue_src_scale")
source_element += f"videoscale ! "
source_element += f"video/x-raw, format={self.network_format}, width={self.network_width}, height={self.network_height}, framerate=30/1 ! "
elif (self.source_type == "usb"):
source_element = f"v4l2src device={self.video_source} name=src_0 ! "
source_element += f"video/x-raw, width=640, height=480, framerate=30/1 ! "
else:
source_element = f"filesrc location={self.video_source} name=src_0 ! "
source_element += QUEUE("queue_dec264")
source_element += f" qtdemux ! h264parse ! avdec_h264 max-threads=2 ! "
source_element += f" video/x-raw,format=I420 ! "
source_element += QUEUE("queue_scale")
source_element += f"videoscale n-threads=2 ! "
source_element += QUEUE("queue_src_convert")
source_element += f"videoconvert n-threads=3 name=src_convert qos=false ! "
source_element += f"video/x-raw, format={self.network_format}, width={self.network_width}, height={self.network_height}, pixel-aspect-ratio=1/1 ! "
pipeline_string = "hailomuxer name=hmux "
pipeline_string += source_element
pipeline_string += "tee name=t ! "
pipeline_string += QUEUE("bypass_queue", max_size_buffers=20) + "hmux.sink_0 "
pipeline_string += "t. ! " + QUEUE("queue_hailonet")
pipeline_string += "videoconvert n-threads=3 ! "
pipeline_string += f"hailonet hef-path={self.hef_path} batch-size={self.batch_size} force-writable=true ! "
pipeline_string += QUEUE("queue_hailofilter")
pipeline_string += f"hailofilter function-name={self.post_function_name} so-path={self.default_postprocess_so} qos=false ! "
pipeline_string += QUEUE("queue_hmuc") + " hmux.sink_1 "
pipeline_string += "hmux. ! " + QUEUE("queue_hailo_python")
pipeline_string += QUEUE("queue_user_callback")
pipeline_string += f"identity name=identity_callback ! "
pipeline_string += QUEUE("queue_hailooverlay")
pipeline_string += f"hailooverlay ! "
pipeline_string += QUEUE("queue_videoconvert")
pipeline_string += f"videoconvert n-threads=3 qos=false ! "
pipeline_string += QUEUE("queue_hailo_display")
pipeline_string += f"fpsdisplaysink video-sink={self.video_sink} name=hailo_display sync={self.sync} text-overlay={self.options_menu.show_fps} signal-fps-measurements=true "
print(pipeline_string)
return pipeline_string
Diese Klasse richtet unsere GStreamer-Pipeline ein. Sie führt ein Pose-Estimation-Modell auf einem Raspberry Pi mit einem Hailo-KI-Beschleuniger aus, einem Edge-KI-Prozessor, der dafür entwickelt wurde, Deep-Learning- und KI-Inferenzaufgaben auf Edge-Geräten auszuführen. Sie initialisiert Parameter wie Batch-Größe, Eingabedimensionen, Farbformat und Dateipfade für das neuronale Netzwerkmodell und die Nachverarbeitung. Abhängig von der Eingabequelle (z. B. Raspberry-Pi-Kamera, USB-Kamera oder Videodatei) erstellt sie eine GStreamer-Pipeline, um Video aufzunehmen, es durch das Hailo-Modell zu verarbeiten, Nachverarbeitung anzuwenden und die Ausgabe anzuzeigen. Die Pipeline-Konfiguration wird in der Methode get_pipeline_string angegeben, die die für diese Aufgabe erforderlichen GStreamer-Elemente zusammensetzt.
Ausführen des Programms
Der letzte Schritt ist das Ausführen des Programms.
if __name__ == "__main__":
# Create an instance of the user app callback class
user_data = user_app_callback_class()
parser = get_default_parser()
args = parser.parse_args()
app = GStreamerPoseEstimationApp(args, user_data)
app.run()
Hier verbinden wir alles miteinander. Wir erstellen eine Instanz unserer user_app_callback_class, parsen alle Befehlszeilenargumente, erstellen unsere GStreamerPoseEstimationApp und führen sie aus.
Hier ist ein Beispielergebnis dessen, was Sie erwarten können, wenn Sie das Programm ausführen.
Abb. 2- Ausgabe eines Posenschätzungsprogramms in Slack mit Augen-Landmarks, die mit grünen Punkten markiert sind.png
Abb. 2: Ausgabe eines Posenschätzungsprogramms in Slack mit Augen-Landmarks, die mit grünen Punkten markiert sind
Dies ist das Update an Slack, wenn eine Person erkannt wird. Es hat einen Konfidenzwert von 81 %, die Koordinaten des linken und rechten Auges sowie das Bild mit markierten Augen-Landmarks (grüne Kreise auf den Positionen des linken und rechten Auges).
Anwendungsfälle der Kombination von KI und Vektordatenbanken
Die Kombination von Edge-KI und Vektordatenbanken bringt zahlreiche Anwendungsfälle mit sich. Nachfolgend finden Sie einige Beispiele.
Robotik
Edge-KI und Vektordatenbanken erweitern die Fähigkeiten autonomer Roboter, indem sie ihnen ermöglichen, Sensordaten lokal zu verarbeiten und sofortige Entscheidungen zu treffen. Diese Technologie verbessert die Effizienz und Anpassungsfähigkeit in dynamischen Umgebungen erheblich.
Beispiel: In der Lagerautomatisierung verwenden autonome mobile Roboter (AMRs) Bordkameras und lokale Vektordatenbanken zur Echtzeit-Objekterkennung. Während sie durch das Lager navigieren, identifizieren und kategorisieren diese Roboter Produkte sofort, indem sie visuelle Daten mit lokal gespeicherten Vektoren vergleichen. Dadurch können sie sich an veränderte Lagerbedingungen anpassen und ihre Routen optimieren, ohne auf Cloud-Infrastruktur angewiesen zu sein.
Smart Cities
Edge-KI und Vektordatenbanken spielen eine entscheidende Rolle bei der Optimierung städtischer Abläufe, vom Verkehrsmanagement bis zur öffentlichen Sicherheit. Diese Technologien ermöglichen eine schnelle Datenverarbeitung und Entscheidungsfindung am Edge, wodurch die Effizienz der Stadt und die Lebensqualität der Bewohner verbessert werden.
Beispiel: Intelligente Verkehrsmanagementsysteme nutzen Edge-KI, um Video-Feeds von Kreuzungskameras lokal zu verarbeiten. Das System wandelt Verkehrsmuster in Vektordarstellungen um und vergleicht sie mit historischen Daten, die in einer lokalen Vektordatenbank gespeichert sind. Dies ermöglicht Echtzeit-Anpassungen der Ampelschaltungen zur Reduzierung von Staus und gewährleistet schnelle Reaktionszeiten, ohne dass eine ständige Datenübertragung an zentrale Server erforderlich ist.
Industrielle Automatisierung
Edge-KI und Vektordatenbanken fördern vorausschauende Wartung und Geräteüberwachung in Fertigungs- und Industrieumgebungen. Dieser Ansatz ermöglicht eine Echtzeitanalyse des Maschinenzustands, verhindert Ausfallzeiten und optimiert Produktionsprozesse.
Beispiel: Intelligente Fertigungsanlagen setzen Sensoren an kritischen Maschinen ein, um kontinuierlich Vibrationsdaten zu erfassen. Edge-Geräte wandeln diese Daten in Vektordarstellungen um und vergleichen sie mit bekannten „gesunden“ und „fehlerhaften“ Vibrationsmustern, die in einer lokalen Vektordatenbank gespeichert sind. Diese Echtzeitanalyse ermöglicht die sofortige Erkennung potenzieller Geräteausfälle, erleichtert proaktive Wartung und minimiert kostspielige Unterbrechungen.
Gesundheitswesen
Edge-KI und Vektordatenbanken transformieren die Patientenversorgung, indem sie eine kontinuierliche Gesundheitsüberwachung ermöglichen, ohne die Privatsphäre zu beeinträchtigen. Diese Technologien ermöglichen eine anspruchsvolle Analyse von Gesundheitsdaten direkt auf Wearables oder lokalen Systemen.
Beispiel: Tragbare EKG-Monitore nutzen Edge-KI, um Herzrhythmusdaten lokal zu verarbeiten. Das Gerät wandelt EKG-Muster in Vektoren um und vergleicht sie mit einer Datenbank normaler und abnormaler Rhythmen, die an Bord gespeichert ist. Dieser Ansatz ermöglicht die sofortige Erkennung potenzieller Herzprobleme, während sensible Gesundheitsdaten auf dem Gerät verbleiben, und verbessert so sowohl die Reaktionsfähigkeit als auch die Privatsphäre der Patienten.
Fazit
Da unstrukturierte Daten an Umfang und Bedeutung zunehmen, werden effiziente Verarbeitungslösungen immer wichtiger. Tim hat das transformative Potenzial von Vektordatenbanken wie Milvus gut hervorgehoben und fortschrittliche Funktionen sowohl für Cloud- als auch für Edge-Bereitstellungen aufgezeigt. Ob für groß angelegte KI-Anwendungen oder ressourcenbeschränkte Edge-Geräte: Milvus und sein leichtgewichtiges Pendant Milvus Lite ermöglichen es Unternehmen, unstrukturierte Daten für Entscheidungsfindung in Echtzeit, Innovation und betriebliche Effizienz zu nutzen. Mit der Flexibilität, von der Cloud bis zum Edge zu skalieren, stehen diese Technologien an der Spitze moderner KI-gesteuerter Lösungen.
Weitere Ressourcen
Tims KI-Projekte auf GitHub:
Verarbeitung unstrukturierter Daten mit einem Raspberry Pi AI Kit
Weiterlesen
Why Teams Are Migrating from Weaviate to Zilliz Cloud — and How to Do It Seamlessly
Explore how Milvus scales for large datasets and complex queries with advanced features, and discover how to migrate from Weaviate to Zilliz Cloud.
What Exactly Are AI Agents? Why OpenAI and LangChain Are Fighting Over Their Definition?
AI agents are software programs powered by AI that can perceive their environment, make decisions, and take actions to achieve a goal—often autonomously.
Building RAG Pipelines for Real-Time Data with Cloudera and Milvus
explore how Cloudera can be integrated with Milvus to effectively implement some of the key functionalities of RAG pipelines.