Construiți un flux complet de viziune — proiect integrator

Un sistem de viziune pentru producție este un lanț de modele și reguli legate prin contracte de date. Componentele au fost deja prezentate în această fază; proiectul integrator le conectează capăt-la-capăt.

Tip: Construiți Limbaje: Python Cerințe prealabile: Faza 4 Lecțiile 01–15 Timp: ~120 de minute

Obiective de învățare

• Proiectați un flux de viziune pentru producție care detectează obiecte, le clasifică și emite JSON structurat — cu toate căile de eșec gestionate
• Conectați un detector (Mask R-CNN sau YOLO), un clasificator (ConvNeXt-Tiny) și un contract de date (Pydantic) într-un singur serviciu
• Evaluați performanța fluxului capăt-la-capăt și identificați primul blocaj (de obicei preprocesarea, apoi detectorul)
• Lansați un serviciu FastAPI minimal care acceptă încărcarea unei imagini, rulează fluxul și întoarce detecții cu clasificări

Problema

Modelele individuale de viziune sunt utile; produsele de viziune sunt lanțuri de astfel de modele. Un audit al rafturilor din comerțul cu amănuntul combină un detector, un clasificator de produse și un flux OCR pentru prețuri. Conducerea autonomă combină un detector 2D, un detector 3D, un segmentator, un tracker și un planificator. O preevaluare medicală combină un segmentator, un clasificator de regiuni și o interfață pentru clinician.

Conectarea acestor lanțuri este elementul care separă un prototip ML de un produs. Fiecare interfață dintre modele este un nou loc în care pot apărea erori. Fiecare transformare de coordonate, fiecare normalizare, fiecare redimensionare de mască este un posibil eșec silențios. Un flux este la fel de puternic ca cea mai slabă interfață a sa.

Acest proiect integrator configurează fluxul minim viabil: detecție + clasificare + ieșire structurată + un strat de servire. Tot ceea ce apare în Faza 4 poate fi introdus în acest schelet: înlocuiți Mask R-CNN cu YOLOv8, adăugați un cap OCR, adăugați o ramură de segmentare, adăugați un tracker. Arhitectura este stabilă; componentele sunt interschimbabile.

Conceptul

Fluxul

Șapte etape. Cele două etape de model sunt costisitoare; celelalte cinci etape sunt locul în care apar erorile.

Contracte de date cu Pydantic

Fiecare graniță dintre modele devine un obiect tipizat. Acest lucru transformă eșecurile silențioase în unele vizibile.

Detection(
    box: tuple[float, float, float, float],   # (x1, y1, x2, y2), pixeli absoluți
    score: float,                              # [0, 1]
    class_id: int,                             # din harta de etichete a detectorului
    mask: Optional[list[list[int]]],           # codificată RLE, dacă este prezentă
)

PipelineResult(
    image_id: str,
    detections: list[Detection],
    classifications: list[Classification],
    inference_ms: float,
)

Când un detector întoarce casete în (cx, cy, w, h) în loc de (x1, y1, x2, y2), validarea Pydantic eșuează la graniță și aflați imediat, în loc să depanați ulterior o decupare care întoarce în tăcere regiuni goale.

Notă tehnică a traducerii: În schema de mai sus, ambele convenții au aceeași formă — patru valori float — astfel încât Pydantic nu poate identifica singur confuzia dintre (cx, cy, w, h) și (x1, y1, x2, y2). Pentru aceasta, adăugați un validator de model sau un tip semantic care verifică relațiile stricte x2 > x1, y2 > y1 și, când este posibil, limitele imaginii.

Unde se consumă latența

Trei adevăruri se aplică în aproape fiecare flux de viziune:

1. Preprocesarea este adesea cel mai mare bloc individual. Decodarea JPEG-urilor, conversia spațiilor de culoare, redimensionarea — acestea sunt limitate de CPU și ușor de uitat.
2. Detectorul domină timpul pe GPU. 70–90% din timpul GPU este petrecut în trecerea înainte de detecție.
3. Postprocesarea (NMS, codificare/decodificare RLE) este ieftină pe GPU, costisitoare pe CPU. Profilarea trebuie făcută întotdeauna pe ținta reală.

Notă tehnică a traducerii: Procentele și ordinea acestor blocaje nu sunt garanții. Ele depind de model, rezoluție, numărul de detecții, dimensiunea lotului, operatori, runtime și hardware; pe unele ținte decodarea, transferurile de date sau postprocesarea pot domina. Măsurați fiecare etapă pe configurația de implementare reală înainte de a prioritiza optimizările.

Cunoașterea distribuției transformă optimizarea într-o listă prioritizată.

Moduri de eșec

• Detecții goale — întoarceți o listă goală, nu opriți aplicația. Înregistrați evenimentul.
• Casete în afara limitelor — restrângeți-le la dimensiunea imaginii înainte de decupare.
• Decupări foarte mici — omiteți clasificarea casetelor mai mici decât intrarea minimă a clasificatorului.
• Încărcare coruptă — răspuns 400 cu un cod de eroare specific, nu 500.
• Eșec la încărcarea modelului — eșuați la pornirea serviciului, nu la prima cerere.

Un flux pentru producție gestionează fiecare dintre acestea fără a scrie try/except generic care ascunde eșecul. Fiecare eșec primește un cod denumit și un răspuns.

Gruparea în loturi

Un serviciu pentru producție deservește mai mulți clienți. Gruparea detecțiilor și clasificărilor din cereri diferite în loturi multiplică debitul. Compromisul este latența suplimentară a așteptării până se umple un lot. O configurație tipică: colectați cereri timp de cel mult 20 ms, grupați-le, procesați-le și distribuiți răspunsurile. torchserve și triton fac acest lucru nativ; serviciile mici cu încărcare predictibilă își construiesc propriul microbatcher.

Construiți-l

Pasul 1: Contractele de date

from pydantic import BaseModel, Field
from typing import List, Optional, Tuple

class Detection(BaseModel):
    box: Tuple[float, float, float, float]
    score: float = Field(ge=0, le=1)
    class_id: int = Field(ge=0)
    mask_rle: Optional[str] = None


class Classification(BaseModel):
    detection_index: int
    class_id: int
    class_name: str
    score: float = Field(ge=0, le=1)


class PipelineResult(BaseModel):
    image_id: str
    detections: List[Detection]
    classifications: List[Classification]
    inference_ms: float

Cinci secunde de cod economisesc o oră de depanare în orice flux serios.

Pasul 2: O clasă Pipeline minimală

import time
import numpy as np
import torch
from PIL import Image

class VisionPipeline:
    def __init__(self, detector, classifier, class_names,
                 device="cpu", min_crop=32):
        self.detector = detector.to(device).eval()
        self.classifier = classifier.to(device).eval()
        self.class_names = class_names
        self.device = device
        self.min_crop = min_crop

    def preprocess(self, image):
        """
        image: PIL.Image or np.ndarray (H, W, 3) uint8
        returns: CHW float tensor on device
        """
        if isinstance(image, Image.Image):
            image = np.asarray(image.convert("RGB"))
        tensor = torch.from_numpy(image).permute(2, 0, 1).float() / 255.0
        return tensor.to(self.device)

    @torch.no_grad()
    def detect(self, image_tensor):
        return self.detector([image_tensor])[0]

    @torch.no_grad()
    def classify(self, crops):
        if len(crops) == 0:
            return []
        batch = torch.stack(crops).to(self.device)
        logits = self.classifier(batch)
        probs = logits.softmax(-1)
        scores, cls = probs.max(-1)
        return list(zip(cls.tolist(), scores.tolist()))

    def run(self, image, image_id="anonymous"):
        t0 = time.perf_counter()
        tensor = self.preprocess(image)
        det = self.detect(tensor)

        crops = []
        detections = []
        valid_indices = []
        for i, (box, score, cls) in enumerate(zip(det["boxes"], det["scores"], det["labels"])):
            x1, y1, x2, y2 = [max(0, int(b)) for b in box.tolist()]
            x2 = min(x2, tensor.shape[-1])
            y2 = min(y2, tensor.shape[-2])
            detections.append(Detection(
                box=(x1, y1, x2, y2),
                score=float(score),
                class_id=int(cls),
            ))
            if (x2 - x1) < self.min_crop or (y2 - y1) < self.min_crop:
                continue
            crop = tensor[:, y1:y2, x1:x2]
            crop = torch.nn.functional.interpolate(
                crop.unsqueeze(0),
                size=(224, 224),
                mode="bilinear",
                align_corners=False,
            )[0]
            crops.append(crop)
            valid_indices.append(i)

        class_preds = self.classify(crops)

        classifications = []
        for valid_idx, (cls_id, cls_score) in zip(valid_indices, class_preds):
            classifications.append(Classification(
                detection_index=valid_idx,
                class_id=int(cls_id),
                class_name=self.class_names[cls_id],
                score=float(cls_score),
            ))

        return PipelineResult(
            image_id=image_id,
            detections=detections,
            classifications=classifications,
            inference_ms=(time.perf_counter() - t0) * 1000,
        )

Fiecare interfață este tipizată. Fiecare cale de eșec are o decizie specifică de gestionare.

Notă tehnică a traducerii: Decuparea din exemplu nu restrânge x1 și y1 la limitele superioare ale imaginii și nu verifică explicit că x2 > x1 și y2 > y1. În codul de producție, restrângeți toate cele patru coordonate la limite, validați caseta nenulă înainte de decupare și definiți o politică explicită pentru coordonate NaN, inverse sau în afara imaginii.

Pasul 3: Conectați un detector și un clasificator

from torchvision.models.detection import maskrcnn_resnet50_fpn_v2
from torchvision.models import convnext_tiny

# Use ImageNet-pretrained weights for a realistic pipeline without training
detector = maskrcnn_resnet50_fpn_v2(weights="DEFAULT")
classifier = convnext_tiny(weights="DEFAULT")
class_names = [f"imagenet_class_{i}" for i in range(1000)]

pipe = VisionPipeline(detector, classifier, class_names)

# Smoke test with a synthetic image
test_image = (np.random.rand(400, 600, 3) * 255).astype(np.uint8)
result = pipe.run(test_image, image_id="demo")
print(result.model_dump_json(indent=2)[:500])

Notă tehnică a traducerii: Pentru inferență ImageNet corectă, ponderile preantrenate ConvNeXt necesită transformarea asociată lor — de exemplu ConvNeXt_Tiny_Weights.DEFAULT.transforms() — și categoriile din weights.meta["categories"]. Codul păstrează nume de clasă generice și redimensionează decupările, însă nu aplică transformarea completă (inclusiv normalizarea); prin urmare, el ilustrează forma fluxului, nu o clasificare preantrenată corectă.

Pasul 4: Serviciul FastAPI

from fastapi import FastAPI, UploadFile, HTTPException
from io import BytesIO

app = FastAPI()
pipe = None  # initialised on startup

@app.on_event("startup")
def load():
    global pipe
    detector = maskrcnn_resnet50_fpn_v2(weights="DEFAULT").eval()
    classifier = convnext_tiny(weights="DEFAULT").eval()
    pipe = VisionPipeline(detector, classifier, class_names=[f"c{i}" for i in range(1000)])

@app.post("/detect")
async def detect_endpoint(file: UploadFile):
    if file.content_type not in {"image/jpeg", "image/png", "image/webp"}:
        raise HTTPException(status_code=400, detail="unsupported image type")
    data = await file.read()
    try:
        img = Image.open(BytesIO(data)).convert("RGB")
    except Exception:
        raise HTTPException(status_code=400, detail="cannot decode image")
    result = pipe.run(img, image_id=file.filename or "upload")
    return result.model_dump()

Rulați cu uvicorn main:app --host 0.0.0.0 --port 8000. Testați cu curl -F 'file=@dog.jpg' http://localhost:8000/detect.

Notă tehnică a traducerii: content_type provine din antetul clientului și nu este o verificare a conținutului. Într-un serviciu expus, stabiliți limite de dimensiune și de pixeli, validați imaginea decodificată și prindeți numai erorile de decodare așteptate, nu orice Exception, pentru a nu masca defecte interne drept răspunsuri 400. FastAPI recomandă în prezent lifespan pentru încărcarea și eliberarea modelelor; evenimentele startup/shutdown sunt alternativa depreciată.

Pasul 5: Evaluați performanța fluxului

import time

def benchmark(pipe, num_runs=20, image_size=(400, 600)):
    img = (np.random.rand(*image_size, 3) * 255).astype(np.uint8)
    pipe.run(img)  # warm up

    stages = {"preprocess": [], "detect": [], "classify": [], "total": []}
    for _ in range(num_runs):
        t0 = time.perf_counter()
        tensor = pipe.preprocess(img)
        t1 = time.perf_counter()
        det = pipe.detect(tensor)
        t2 = time.perf_counter()
        crops = []
        for box in det["boxes"]:
            x1, y1, x2, y2 = [max(0, int(b)) for b in box.tolist()]
            x2 = min(x2, tensor.shape[-1])
            y2 = min(y2, tensor.shape[-2])
            if (x2 - x1) >= pipe.min_crop and (y2 - y1) >= pipe.min_crop:
                crop = tensor[:, y1:y2, x1:x2]
                crop = torch.nn.functional.interpolate(
                    crop.unsqueeze(0), size=(224, 224), mode="bilinear", align_corners=False
                )[0]
                crops.append(crop)
        pipe.classify(crops)
        t3 = time.perf_counter()
        stages["preprocess"].append((t1 - t0) * 1000)
        stages["detect"].append((t2 - t1) * 1000)
        stages["classify"].append((t3 - t2) * 1000)
        stages["total"].append((t3 - t0) * 1000)

    for stage, times in stages.items():
        times.sort()
        print(f"{stage:12s}  p50={times[len(times)//2]:7.1f} ms  p95={times[int(len(times)*0.95)]:7.1f} ms")

Notă tehnică a traducerii: Pe CUDA, lansările de kernel sunt asincrone; fără sincronizare înainte de fiecare marcaj de timp, acest benchmark poate subestima etapele GPU. Pentru măsurători corecte, introduceți torch.cuda.synchronize() pe dispozitivul corespunzător înainte de a citi fiecare moment relevant și separați transferurile gazdă–dispozitiv de calcul.

Ieșire tipică pe CPU: preprocesare ~3 ms, detecție 300–500 ms, clasificare 20–40 ms, total 350–550 ms. Pe GPU, detecția este de 20–40 ms, iar preprocesarea + clasificarea încep să conteze mai mult în termeni relativi.

Utilizați-l

Șabloanele pentru producție converg către aceeași structură, plus:

• Versionarea modelului — înregistrați întotdeauna numele modelului și hash-ul ponderilor în răspuns.
• ID-uri de urmărire per cerere — înregistrați durata fiecărei etape pentru fiecare cerere, astfel încât să puteți corela răspunsurile lente cu etapele.
• Cale de rezervă — dacă clasificatorul expiră, întoarceți detecțiile fără clasificări, în loc să eșuați întreaga cerere.
• Filtre de siguranță — filtrele NSFW / PII rulează după clasificare, înainte ca răspunsul să părăsească serviciul.
• Endpoint pentru loturi — un /detect_batch care acceptă o listă de URL-uri de imagini pentru procesare în masă.

Pentru servirea în producție, torchserve, Triton Inference Server și BentoML gestionează implicit gruparea în loturi, versionarea, metricile și verificările de sănătate. Rularea directă a FastAPI este potrivită pentru prototipuri și produse la scară mică.

Notă tehnică a traducerii: Documentația TorchServe indică întreținere limitată, fără actualizări, remedieri de erori sau patch-uri de securitate planificate; evaluați acest statut înainte de o implementare nouă. În Triton, gruparea dinamică se activează și se configurează separat pentru fiecare model fără stare, inclusiv dimensiunea maximă a lotului și întârzierea din coadă; ea nu grupează automat un flux complet cu detector, clasificator și postprocesare. Și BentoML necesită configurarea și măsurarea comportamentului de grupare pentru sarcina reală.

Livrați

Această lecție produce:

• outputs/prompt-vision-service-shape-reviewer.md — un prompt care verifică codul unui serviciu de viziune pentru încălcări ale formei contractelor/răspunsurilor și numește prima eroare care îl întrerupe.
• outputs/skill-pipeline-budget-planner.md — o competență care, date fiind latența și debitul-țintă, alocă un buget de timp fiecărei etape de flux și semnalează prima etapă care își va depăși bugetul.

Exerciții

1. (Ușor) Rulați fluxul pe 10 imagini din orice set de date deschis. Raportați timpul mediu per etapă și distribuția numărului de detecții per imagine.
2. (Mediu) Adăugați un câmp de ieșire pentru mască în Detection și codificați-l ca RLE. Verificați că JSON-ul rămâne sub 1 MB chiar și pentru o imagine cu 10 obiecte.
3. (Dificil) Adăugați un microbatcher în fața clasificatorului: colectați decupări timp de cel mult 10 ms, clasificați-le pe toate într-un singur apel GPU și întoarceți rezultatele per cerere. Măsurați câștigul de debit la 5 cereri concurente pe secundă și latența adăugată.

Termeni-cheie

Termen	Cum este numit în practică	Ce înseamnă de fapt
Flux	„Sistemul”	Un lanț ordonat de pași de preprocesare, inferență și postprocesare, cu o interfață tipizată între fiecare pereche
Contract de date	„Schema”	Definiții Pydantic / dataclass cărora le respectă fiecare intrare și ieșire de etapă; surprind erorile de integrare la graniță
Preprocesare	„Înainte de model”	Decodare, conversie de culoare, redimensionare, normalizare; de obicei cel mai mare consumator de timp CPU
Postprocesare	„După model”	NMS, redimensionare de măști, prag, codificare RLE; ieftină pe GPU, costisitoare pe CPU
Microbatcher	„Colectează, apoi propagă”	Agregator care așteaptă o fereastră fixă pentru cereri multiple și rulează o singură trecere înainte pe lot
Trace ID	„ID de cerere”	Identificator per cerere înregistrat la fiecare etapă, astfel încât cererile lente pot fi urmărite capăt-la-capăt
Cod de eșec	„Eroare denumită”	Cod de eroare specific pentru fiecare clasă de eșec, în loc de 500 generic; permite logica de reîncercare a clientului
Verificare de sănătate	„Readiness probe”	Endpoint ieftin care raportează dacă serviciul poate răspunde; load balancerele se bazează pe acesta

Lecturi suplimentare

• Full Stack Deep Learning — implementarea modelelor — prezentarea de referință a implementării ML pentru producție
• Documentația BentoML — framework de servire cu grupare în loturi, versionare și metrici
• Documentația torchserve — biblioteca oficială de servire PyTorch
• NVIDIA Triton Inference Server — servire cu debit mare, grupare în loturi și suport pentru mai multe modele

Sursă: Originalul în limba engleză

Navigare: ← Lecția 04.15 — Viziune în timp real — implementare la marginea rețelei · Faza 4 — Viziune computerizată · Catalog complet · Lecția 04.17 — Viziune auto-supervizată — SimCLR, DINO, MAE →