طراحی و پیاده‌سازی پایپ‌لاین بینایی ماشین در محیط عملیاتی

در اکثر پروژه‌های واقعی، کار شما صرفاً اجرای یک مدل مستقل در ژوپیتر نوتبوک نیست. سامانه‌های کاربردی نظیر بازرسی خودکار، تحلیل تصاویر پزشکی یا پایش دوربین‌های نظارتی از زنجیره‌ای پیوسته شامل آشکارساز اشیاء، الگوریتم‌های پس‌پردازش و دسته‌بندها تشکیل می‌شوند. بزرگ‌ترین چالش مهندسان نرم‌افزار و هوش مصنوعی، وصل کردن این قطعات مجزا به یکدیگر بدون ایجاد باگ‌های خاموش در مرز تبادل داده‌ها است.

در این درس پایانی، اسکلت یک سیستم عملیاتی بینایی ماشین (Vision Pipeline) را از صفر بنا می‌کنید. با کمک Pydantic قراردادهای داده‌ی دقیق می‌سازید تا ناهماهنگی در ابعاد باکس‌ها یا ابعاد تصویر پیش از ورود به مراحل بعدی شناسایی شود، مسیرهای شکست واقعی نظیر آپلود فایل خراب یا کادرهای فراتر از تصویر را مهار می‌کنید و در نهایت کل پایپ‌لاین را در قالب یک وب‌سرویس سریع FastAPI آماده‌ی بهره‌برداری می‌سازید.

برای اجرای تست‌های اولیه و بنچمارک این پایپ‌لاین نیازی به اجاره‌ی سرورهای ابری گران‌قیمت نخواهید داشت و کدها به‌خوبی روی CPU یا پردازنده‌های معمولی سیستم شخصی اجرا می‌شوند؛ هرچند هنگام انتقال به بستر نهایی، توزیع زمان پردازش میان CPU و کارت گرافیک را با دقت بررسی خواهیم کرد.

یک سیستم بینایی ماشین در محیط عملیاتی، زنجیره‌ای از مدل‌ها و قواعد است که با قراردادهای داده به هم وصل شده‌اند. قطعات کار در این فاز آماده شده‌اند؛ پروژه‌ی پایانی آن‌ها را به‌صورت سرتاسری به هم متصل می‌کند.

اهداف یادگیری

• طراحی یک پایپ‌لاین بینایی ماشین عملیاتی که اشیاء را شناسایی کند، آن‌ها را دسته‌بندی کند و خروجی JSON ساختاریافته بدهد — همراه با مدیریت تمام مسیرهای شکست
• اتصال یک آشکارساز (Mask R-CNN یا YOLO)، یک دسته‌بند (ConvNeXt-Tiny) و یک قرارداد داده (Pydantic) درون یک سرویس واحد
• بنچمارک پایپ‌لاین سرتاسری و شناسایی اولین گلوگاه (معمولاً پیش‌پردازش، سپس آشکارساز)
• راه‌اندازی یک سرویس مینیمال FastAPI که تصویر ارسالی را دریافت کند، پایپ‌لاین را اجرا کند و اشیاء شناسایی‌شده را همراه با دسته‌بندی بازگرداند

صورت مسئله

مدل‌های بینایی ماشین به‌صورت تکی کاربردی هستند؛ اما محصولات بینایی زنجیره‌ای از آن‌ها به‌شمار می‌روند. بررسی قفسه‌ی فروشگاه یعنی یک آشکارساز به‌علاوه‌ی یک دسته‌بند محصول و یک پایپ‌لاین OCR قیمت. رانندگی خودران یعنی یک آشکارساز 2D به‌علاوه‌ی یک آشکارساز 3D، یک قطعه‌بند، یک ردیاب و یک برنامه‌ریز حرکت. غربالگری اولیه‌ی پزشکی یعنی یک قطعه‌بند به‌همراه یک دسته‌بند ناحیه و یک رابط کاربری برای پزشک.

اتصال این زنجیره‌ها همان بخشی است که نمونه‌ی اولیه‌ی یادگیری ماشین را از یک محصول واقعی جدا می‌کند. هر رابط میان مدل‌ها، جای جدیدی برای بروز باگ است. هر تبدیل مختصات، هر نرمال‌سازی و هر تغییر اندازه‌ی ماسک می‌تواند باعث خطای پنهان شود. استحکام یک پایپ‌لاین به اندازه‌ی ضعیف‌ترین رابط آن است.

این پروژه‌ی پایانی، حداقل پایپ‌لاین کارا را می‌سازد: آشکارسازی + دسته‌بندی + خروجی ساختاریافته + لایه‌ی سرویس‌دهی. سایر بخش‌های فاز 4 درون همین اسکلت قرار می‌گیرند: جایگزینی Mask R-CNN با YOLOv8، افزودن سر OCR، افزودن یک branch قطعه‌بندی یا افزودن یک ردیاب. معماری ثابت می‌ماند و قطعات قابل تعویض هستند.

مفاهیم پایه

پایپ‌لاین

هفت مرحله. دو مرحله‌ی مربوط به مدل‌ها سنگین و پرهزینه‌اند؛ پنج مرحله‌ی دیگر جایی هستند که باگ‌ها جا خوش می‌کنند.

قراردادهای داده با Pydantic

هر مرز میان مدل‌ها به یک شیء دارای تایپ تبدیل می‌شود. این کار خطاهای پنهان را به خطاهایی آشکار و واضح تبدیل می‌کند.

Detection(
    box: tuple[float, float, float, float],   # (x1, y1, x2, y2), absolute pixels
    score: float,                              # [0, 1]
    class_id: int,                             # from detector's label map
    mask_rle: Optional[str],                   # RLE string if present
)

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

وقتی آشکارساز باکس‌ها را به‌جای (x1, y1, x2, y2) با فرمت (cx, cy, w, h) برمی‌گرداند، اعتبارسنجی Pydantic دقیقاً در همان مرز خطا می‌دهد و بلافاصله متوجه می‌شوید؛ به‌جای این‌که ساعت‌ها وقت صرف عیب‌یابی کراپ‌های مراحل بعد کنید که در سکوت، ناحیه‌های خالی برمی‌گردانند.

زمان پاسخ‌دهی کجا صرف می‌شود؟

سه اصل در تقریباً تمام پایپ‌لاین‌های بینایی ماشین صادق است:

1. پیش‌پردازش معمولاً بزرگ‌ترین بخش منفرد است. رمزگشایی فایل‌های JPEG، تبدیل فضاهای رنگی، تغییر اندازه — این‌ها وابسته به پردازنده (CPU-bound) هستند و خیلی راحت نادیده گرفته می‌شوند.
2. آشکارساز بیشترین زمان GPU را می‌گیرد. بین 70 تا 90 درصد زمان GPU صرف گذر رو به جلو (forward pass) مرحله‌ی آشکارسازی می‌شود.
3. پس‌پردازش (NMS، رمزگذاری و رمزگشایی RLE) روی GPU سبک است، اما روی CPU سنگین. همیشه روی سخت‌افزار واقعی هدف پروفایل بگیرید.

دانستن این توزیع زمانی به شما امکان می‌دهد بهینه‌سازی را به یک فهرست اولویت‌بندی‌شده تبدیل کنید.

حالت‌های شکست

• نبود هیچ شیء شناسایی‌شده — برگرداندن لیست خالی، بدون کرش کردن سرویس. ثبت در لاگ.
• باکس‌های خارج از کادر — محدود کردن مختصات (clamp) به ابعاد تصویر پیش از کراپ.
• کراپ‌های بسیار کوچک — صرف‌نظر از دسته‌بندی برای باکس‌های کوچک‌تر از حداقل ورودی دسته‌بند.
• فایل آپلودشده‌ی خراب — بازگرداندن خطای 400 همراه با کد خطای مشخص، نه خطای 500.
• خطا در بارگذاری مدل — بالا آمدن خطا در زمان راه‌اندازی اولیه سرویس، نه هنگام دریافت اولین درخواست.

یک پایپ‌لاین عملیاتی هر یک از این موارد را بدون نوشتن بلوک‌های عمومی try/except که خطا را مخفی می‌کنند، مدیریت می‌کند. هر خطا کد مشخص و پاسخ تعریف‌شده دارد.

بچینگ (Batching)

یک سرویس عملیاتی به چندین کلاینت پاسخ می‌دهد. تجمیع آشکارسازی‌ها و دسته‌بندی‌ها در قالب بچ میان چند درخواست، توان عملیاتی (throughput) را چند برابر می‌کند. نقطه‌ی ضعف: تاخیر اضافی ناشی از منتظر ماندن برای پر شدن یک بچ. ساختار رایج: جمع‌آوری درخواست‌ها تا سقف 20 میلی‌ثانیه، تجمیع در یک بچ، پردازش و سپس توزیع پاسخ‌ها. ابزارهای torchserve و triton این کار را به‌صورت نیتیو انجام می‌دهند؛ در سرویس‌های کوچک با بار کاری قابل پیش‌بینی، معمولاً میکروبچرهای اختصاصی نوشته می‌شود.

پیاده‌سازی از صفر

مرحله‌ی 1: قراردادهای داده

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

پنج ثانیه کدنویسی، جلوی یک ساعت دیباگ کلافه‌کننده را در هر پایپ‌لاین جدی می‌گیرد.

مرحله‌ی 2: یک کلاس Pipeline مینیمال

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,
        )

تمام رابط‌ها تایپ دارند. برای هر مسیر شکست، تصمیم مدیریتی مشخصی گرفته شده است.

مرحله‌ی 3: اتصال آشکارساز و دسته‌بند

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

# استفاده از وزن‌های پیش‌آموزش‌دیده ImageNet برای یک پایپ‌لاین واقع‌گرایانه بدون نیاز به آموزش
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)

# تست دود با یک تصویر ساختگی
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])

مرحله‌ی 4: سرویس 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()

با دستور uvicorn main:app --host 0.0.0.0 --port 8000 اجرا کنید. با دستور curl -F 'file=@dog.jpg' http://localhost:8000/detect تست کنید.

مرحله‌ی 5: بنچمارک پایپ‌لاین

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")

خروجی معمول روی CPU: پیش‌پردازش حدود 3 میلی‌ثانیه، آشکارسازی 300 تا 500 میلی‌ثانیه، دسته‌بندی 20 تا 40 میلی‌ثانیه، کل زمان 350 تا 550 میلی‌ثانیه. روی GPU، آشکارسازی به 20 تا 40 میلی‌ثانیه می‌رسد و سهم نسبی پیش‌پردازش و دسته‌بندی پررنگ‌تر می‌شود.

استفاده از ابزارهای واقعی

قالب‌های آماده‌ی عملیاتی همگی به همین ساختار همگرا می‌شوند، به‌علاوه‌ی موارد زیر:

• نسخه‌بندی مدل — همیشه نام مدل و هش وزن‌ها را در پاسخ لاگ کنید.
• شناسه‌ی ردگیری (Trace ID) به ازای هر درخواست — زمان‌بندی تمام مراحل را برای هر درخواست ثبت کنید تا بتوانید پاسخ‌های کند را به مراحل مشخص ربط دهید.
• مسیر جایگزین (Fallback) — اگر دسته‌بند با تایم‌اوت مواجه شد، به‌جای رد کردن کل درخواست، آشکارسازی‌ها را بدون دسته‌بندی برگردانید.
• فیلترهای ایمنی — فیلترهای محتوای نامناسب (NSFW) و داده‌های حساس شخصی (PII) بعد از دسته‌بندی و پیش از خروج پاسخ از سرویس اجرا شوند.
• اندپوینت پردازش دسته‌ای — یک اندپوینت /detect_batch که فهرستی از URLهای تصاویر را برای پردازش حجیم دریافت کند.

در محیط‌های عملیاتی، ابزارهای torchserve، Triton Inference Server و BentoML مواردی مثل بچینگ، نسخه‌بندی، متریک‌ها و بررسی سلامت (health check) را آماده و بدون نیاز به پیکربندی خاص (out of the box) پوشش می‌دهند. اجرای مستقیم FastAPI برای پروتوتایپ‌ها و محصولات با مقیاس کوچک کاملاً مناسب است.

خروجی این درس

این درس دو خروجی تولید می‌کند:

• outputs/prompt-vision-service-shape-reviewer.md — یک پرامپت که کد سرویس بینایی را بررسی می‌کند تا مغایرت با قرارداد داده یا ساختار پاسخ را پیدا کند و اولین باگ شکننده را گزارش دهد.
• outputs/skill-pipeline-budget-planner.md — مهارتی که با دریافت تاخیر هدف و نرخ پاسخ‌دهی، بودجه‌ی زمانی هر مرحله از پایپ‌لاین را تعیین کرده و مرحله‌ای را که زودتر از بودجه فراتر می‌رود مشخص می‌کند.

تمرین‌ها

1. (ساده) پایپ‌لاین را روی 10 تصویر از یک دیتاست متن‌باز دلخواه اجرا کنید. میانگین زمان هر مرحله و توزیع تعداد اشیاء شناسایی‌شده در هر تصویر را گزارش دهید.
2. (متوسط) یک فیلد ماسک خروجی به Detection اضافه کنید و آن را با RLE کدگذاری کنید. بررسی کنید که حجم JSON خروجی حتی برای تصویری با 10 شیء زیر 1 مگابایت بماند.
3. (سخت) یک میکروبچر جلوی دسته‌بند اضافه کنید: کراپ‌ها را تا حداکثر 10 میلی‌ثانیه جمع‌آوری کرده، همه‌ی آن‌ها را در یک فراخوانی GPU دسته‌بندی کنید و نتایج را به تفکیک درخواست بازگردانید. افزایش توان عملیاتی را در نرخ 5 درخواست هم‌زمان در ثانیه و میزان تاخیر اضافه‌شده را اندازه بگیرید.

اصطلاحات کلیدی

برای مطالعه‌ی بیشتر

• دوره Full Stack Deep Learning — دیپلوی مدل‌ها — منبع مرجع درباره‌ی راه‌اندازی و دیپلوی عملیاتی مدل‌های یادگیری عمیق
• مستندات BentoML — فریم‌ورک سرویس‌دهی همراه با بچینگ، نسخه‌بندی و متریک‌ها
• مستندات torchserve — کتابخانه‌ی رسمی سرویس‌دهی PyTorch
• NVIDIA Triton Inference Server — سرور استنتاج با توان کاری بالا، همراه با قابلیت بچینگ و پشتیبانی هم‌زمان از چند مدل