이전 기사에서 우리는 세 가지 오픈 소스 비전-언어 모델을 영상 인식 탐지에 대한 zero-shot 평가를 진행하여 불편한 결론에 도달했습니다: 가장 빠른 후보인 Phi-3.5-vision-instruct도 NVIDIA L4에서 프레임당 4.45초를 소요합니다. LLaVA-v1.6은 8.13초를 기록합니다. 실시간 비디오 스트림을 처리해야 하는 어떤 응용 프로그램에도 이 숫자들은 탈락 요건입니다. 하지만 VLMs가 실시간 작업 부하와 근본적으로 호환되지 않는다는 결론은 더 많은 검토를值得합니다. 그 8초의 숫자는 일반적인 zero-shot 탐지 작업에서 측정된 것이며, 모델이 임의의 장면에서 임의의 객체에 대해 추론하도록 요청했습니다. 문제를 제약하는 경우 어떻게 될까요? 모델에게 폐쇄형 어휘, 고정 해상도, 결정론적 디코딩 전략, 논블로킹 추론 파이프라인을 제공하는 경우요?
이 기사는 그 질문에 답합니다. Google의 컴팩트 비전-언어 모델인 PaliGemma를 사용하여 NVIDIA RTX A4500에서 약 0.8초에서 1.2초 사이의 프레임당 속도로 실시간 비디오 분류 시스템을 구축했습니다. 이는 유사한 전문 하드웨어에서 LLaVA보다 6배에서 8배 향상된 성능을 의미하며, 전적으로 아키텍처 결정을 통해 달성되었습니다. 이를 가능하게 한 네 가지 패턴은 다음과 같습니다.
왜 PaliGemma인가요
아키텍처에 들어가기 전에, 모델 선택 자체에 대한 설명이 필요합니다. 그 이유는 PaliGemma가 실질적 가치에 비해 개발자 문헌에서 상당히 소위로 되어 있기 때문입니다. PaliGemma는 구글에 의해 구축된 30억 파라미터 비전-언어 모델로, SigLIP 비전 인코더와 Gemma 언어 백본을 결합합니다. LLaVA-7B 또는 Phi-3.5-vision과 비교하면 크기가 약간 반으로 줄어들어, 동일한 하드웨어에서 VRAM 소비를 낮추고 추론 속도를 빠르게 합니다. 분류 작업에 더 중요한 것은, 이미지 캡션, 비전 질문 답변, 객체 위치 결정 등 다양한 비전 이해 벤치마크에서 명시적으로 미세 조정되었기 때문입니다. 이는 우리가 유도할 제한적이고 구조화된 응답에 강한 사전 정보를 가지고 있음을 의미합니다.
import torch
from transformers import PaliGemmaProcessor, PaliGemmaForConditionalGeneration
MODEL_PATH = "./paligemma_offline"
model = PaliGemmaForConditionalGeneration.from_pretrained(
MODEL_PATH,
torch_dtype=torch.bfloat16,
device_map="auto",
local_files_only=True
).eval()
processor = PaliGemmaProcessor.from_pretrained(
MODEL_PATH,
local_files_only=True
)
여기서 두 가지 사항이 주목할 가치가 있습니다. bfloat16으로 로딩하는 대신 float32로 로딩하면 분류 작업에서 정확도 저하가 거의 없이 VRAM 푸트풋을 절반으로 줄일 수 있습니다. Thelocal_files_only=True 플래그는 오프라인 환경을 위한 편의만이 아닙니다: 생산 시스템에서는 초기화 시 네트워크 라운드 트립을 제거하고, 추론 환경이 완전히 재현 가능하다는 것을 보장합니다.
패턴 1: 해상도를 지연 조절 스위치로 사용
실시간 VLM 파이프라인에서 가장 영향력 있는 결정은 입력 해상도입니다. VLM은 이미지를 패치로 나누어 각 패치를 토큰 시퀀스로 인코딩하여 이미지를 처리합니다. 1280×720 프레임은 448×448 잘라낸 이미지보다 훨씬 더 큰 토큰 시퀀스를 생성하며, 트랜스포머 주의가 시퀀스 길이에 비례하여 제곱으로 확장되기 때문에 해상도는 선형 비용이 아닙니다: 지수 비용입니다. 공간 정밀도가 중요한 zero-shot 객체 탐지에서는 다운샘플링이 실제로 트레이드오프입니다. 하지만 장면 수준 분류 작업에서는 "이 프레임에서 주된 감정은 무엇인가?"라고 묻는 대신 "모든 객체의 픽셀 좌표를 주세요"라고 묻는 경우, 448×448은 충분한 의미 정보를 보존합니다.
import cv2
import numpy as np
from PIL import Image
def preprocess_frame(frame_bgr: np.ndarray) -> Image.Image:
# Resize to inference resolution before any VLM processing
frame_small = cv2.resize(frame_bgr, (448, 448))
# Convert BGR (OpenCV) to RGB (PIL/transformers)
return Image.fromarray(cv2.cvtColor(frame_small, cv2.COLOR_BGR2RGB))
핵심 인사이트는 해상도를 실제로 작업에 필요한 정보의 튜토리티에 따라 선택해야 한다는 것입니다. 입력 스트림의 해상도에 따라 선택해서는 안 됩니다. 카메라가 1080p로 캡처하지만 분류 작업이 오직 다섯 가지 감정 상태를 구별할 뿐이라면, 결국 사용하지 않을 정보에 막대한 컴퓨팅 비용을 지불하게 됩니다.
패턴 2: 닫힌 어휘집을 사용한 결정론적 디코딩
표준 VLM 사용법은 모델을 무제한 텍스트 생성기로 취급합니다. 당신이 프롬프트를 제공하면, 모델은 확률 분포에서 샘플링하고, 당신은 자연어 응답을 받습니다. 이는 이전 기사에서 논의한 취약성 문제의 원인이며, 동시에 지연 시간의 중요한 원인이기도 합니다: 높은 확률로 샘플링max_new_tokens는 모델이 생성하는 각 토큰에 대해 전체 자기회귀 루프를 실행한다는 의미입니다. 분류 작업의 경우, 이를 완전히 중단할 수 있습니다. 모델이 무엇을 보는지 설명하도록 요청하는 대신, 그의 출력을 유효한 레이블의 고정 어휘로 제한하고, 하나를 표현하기에 필요한 최소 토큰 수로 생성을 제한합니다.
# A generic set of states tailored to your specific domain
VALID_CLASSES = ['active', 'idle', 'error', 'offline', 'unknown']
def classify_frame(model: torch.nn.Module, processor, image: Image.Image) -> str:
prompt = (
f"<image> What is the current operational state shown in this frame? "
f"You MUST choose ONLY ONE from: {VALID_CLASSES}."
)
inputs = processor(
text=prompt,
images=image,
return_tensors="pt"
).to(model.device, model.dtype)
with torch.inference_mode():
output_ids = model.generate(
**inputs,
max_new_tokens=10, # A single class label needs at most 2-3 tokens
do_sample=False # Greedy decoding: deterministic, faster, no temperature needed
)
raw_output = processor.decode(
output_ids[0][inputs.input_ids.shape[-1]:],
skip_special_tokens=True
).strip().lower()
# Sanitize to alphabetic characters only
return ''.join(filter(str.isalpha, raw_output))
설정do_sample=False는 모델을 그리디 디코딩으로 전환하며, 이는 매 스텝에서 가장 높은 확률의 토큰을 항상 선택합니다. 이는 샘플링 오버헤드를 제거하고 출력을 완전히 결정론적이게 만듭니다: 동일한 입력은 항상 동일한 출력을 생성하며, 이는 디버깅과 다음에 다루는 시간적 부드러움 패턴에 필수적입니다. Themax_new_tokens=10 cap은 레이블을 생성한 후 거의 즉시 모델이 생성을 중단한다는 의미로, 요청되지 않은 설명 텍스트를 계속 생성하는 대신입니다.
결과적으로 3B-파라미터 VLM을 고능력적이고 의미에 민감한 분류기로 사용하게 되며, 생성 모델이 아닌 전용 분류 헤드의 추론 특성에 가까운 자연어 프롬프트의 절대적 유연성을 얻게 됩니다.
패턴 3: 예측 안정성을 위한 시간적 평활화
확정된 디코딩이라도 실시간 비디오를 처리하는 VLM은 노이즈가 많은 예측을 생성합니다. 조명 변화, 움직임 블러, 일부 가리기, 일시적 시각적 아티팩트는 모델이 연속된 프레임에 걸쳐 일관되지 않은 레이블을 출력하게 합니다. 원시 각 프레임 예측을 직접 하단 시스템으로 파이프로 보내면 떨리는, 신뢰할 수 없는 출력을 얻게 됩니다. 해결책은 시간적 평활화입니다: 단일 예측을 신뢰하는 대신 최근 예측들의 롤링 윈도우를 유지하고 다수결 투표를 발송합니다.
from collections import deque, Counter
class TemporalSmoother:
def __init__(self, window_size: int = 5):
self.history = deque(maxlen=window_size)
def update(self, prediction: str) -> str:
self.history.append(prediction)
# Return the most common prediction in the window
return Counter(self.history).most_common(1)[0][0]
5프레임의 창이 우리의 추론 속도로 변환되면 대략 4에서 6초의 시간적 맥락으로 이어집니다. 이는 일시적인 소음을 흡수하면서도 진정한 상태 변화에 반응할 수 있도록 충분합니다. 창 크기는 주요 조정 파라미터입니다: 더 큰 창은 더 안정적이지만 반응이 느립니다; 더 작은 창은 더 반응적이지만 더 소음이 많습니다. 대부분의 분류 작업에는 3에서 7프레임이 실용적인 범위를 다룹니다.
패턴 4: 디커플된 공유 상태 아키텍처를 사용한 비 블로킹 추론
이전 세 가지 패턴은 추론 호출 자체를 최적화합니다. 이 패턴은 더 기본적인 시스템 문제를 해결합니다: 0.8초에서 1.2초에 걸친 VLM 추론 호출은 실행 중인 스레드를 그 전체 기간 동안 차단합니다. 비디오 캡처와 추론이 같은 스레드에서 실행되면 스트림은 추론 속도로, 그러지 않으면 카메라 속도로 지체됩니다.
쉬운 해결책은 스레드 간 프레임을 전달하기 위해 표준 Python 큐.Queue를 사용하는 것입니다. 그러나 이는 소비자 경쟁을 일으킵니다: 렌더링 스레드와 AI 스레드가 같은 큐에서 읽는 경우, 프레임을 소비하며 서로 데이터를 빼앗아 심각한 시각적 지터링과 추론 주기 건너뛰기를 유발합니다. 생산급 해결책은 비동기 공유 상태 패턴입니다.는 그래나루얼 락킹을 사용합니다. 비디오 캡처 스레드는 생산자로서 계속해서 공유되는 "가장 최근 프레임" 포인터를 덮어씁니다. 렌더링 스레드(메인 스레드에서 실행되며, macOS와 Wayland에서 OpenCV UI 작업에 필수적입니다)와 AI 백그라운드 스레드는 독립적인 소비자로서, 다음 사이클을 준비할 때마다 가장 최근 프레임을 로컬 메모리로 복사합니다.
import threading
import time
import numpy as np
import cv2
import torch
from typing import Optional, Any
class SharedState:
"""
Thread-safe state container.
The lock is strictly granular: it is only held for memory assignment/copying,
never during expensive I/O or AI inference operations.
"""
def __init__(self):
self.latest_frame: Optional[np.ndarray] = None
self.prediction: str = "WAITING"
self.lock = threading.Lock()
self.running: bool = True
shared = SharedState()
def video_capture_worker(source: int = 0) -> None:
"""Reads frames at hardware speed and updates the shared state."""
cap = cv2.VideoCapture(source)
while shared.running:
ret, frame = cap.read()
if not ret:
time.sleep(0.01)
continue
with shared.lock:
# Overwrite with the freshest data.
# Pointer assignment is fast enough to barely hold the lock.
shared.latest_frame = frame
cap.release()
def inference_worker(model: torch.nn.Module, processor: Any) -> None:
"""Consumes the latest frame at the AI's maximum throughput rate."""
smoother = TemporalSmoother(window_size=5)
while shared.running:
with shared.lock:
# Deep copy to prevent OpenCV from mutating the array during inference
frame = shared.latest_frame.copy() if shared.latest_frame is not None else None
if frame is None:
time.sleep(0.05)
continue
try:
image = preprocess_frame(frame)
raw_pred = classify_frame(model, processor, image)
smoothed_pred = smoother.update(raw_pred)
with shared.lock:
shared.prediction = smoothed_pred
except torch.cuda.OutOfMemoryError:
# Handle temporary VRAM spikes gracefully without killing the thread
with shared.lock:
shared.prediction = "OOM_ERROR"
time.sleep(1.0)
except Exception as e:
# Catch corrupt frames or tensor mismatches
with shared.lock:
shared.prediction = "ERROR"
# Initialize background workers as Daemon threads
threads = [
threading.Thread(target=video_capture_worker, args=(0,), daemon=True),
threading.Thread(target=inference_worker, args=(model, processor), daemon=True),
]
for t in threads:
t.start()
# Main Thread UI Loop
# UI libraries (cv2.imshow) must run on the main thread to prevent OS-level crashes.
while shared.running:
with shared.lock:
frame = shared.latest_frame.copy() if shared.latest_frame is not None else None
label = shared.prediction
if frame is not None:
cv2.putText(frame, label.upper(), (30, 50),
cv2.FONT_HERSHEY_SIMPLEX, 1.5, (0, 255, 0), 3)
cv2.imshow("Live Classification", frame)
# 30 FPS rendering limit (33ms) + graceful shutdown
if cv2.waitKey(33) & 0xFF == ord('q'):
shared.running = False
cv2.destroyAllWindows()
여기서 중요한 설계 원칙은 그래나러럴 락입니다: 락을 획득하고, NumPy 배열을 메모리에 복사합니다(이 작업은 마이크로초가 걸립니다), 그리고 락을 즉시 해제합니다. 한 초에 걸리는 VLM 추론 호출에 락을 유지하면 모든 세 부분이 직렬화되어 아키텍처의 전체 목적을 무효화할 것입니다. 이 구조에서는 비디오 캡처 스레드가 하드웨어 프레임레이트(예: 30 fps)로 실행되고, 렌더링 루프는 30 fps로 프레임을 표시하며, 추론 스레드는 자신만의 비동기 속도(1 fps)로 실행됩니다. 세 시스템은 시간적으로 독립적이며, 각각의 하드웨어 제약에만 제한됩니다.
벤치마크 요약
NVIDIA RTX A4500 (20GB GDDR6, 암페어 아키텍처)에서 PaliGemma를 bfloat16으로 사용하여 삼스트림 라이브 비디오 시나리오에서 완전한 파이프라인을 실행하면 매우 안정적인 성능 프로파일이 생성됩니다. 입력 해상도를 448 × 448으로 제한하고, 탐욕스러운 디코딩 전략을 통해 출력을 최대 10개의 새로운 토큰으로 제한하면 (do_sample=False), 시스템은 프레임당 0.8초에서 1.2초 사이의 추론 지연 시간을 달성합니다. 5프레임 시간적 평활화 창과 결합하여 이 구성은 신뢰할 수 있는 상태 분류를 보장하며, 분리된 아키텍처는 비디오 캡처 스레드가 추론 병목 현상과 전혀 관계없이 안정적인 25 fps를 유지할 수 있게 합니다.
비교를 위해, LLaVA-v1.6-Mistral-7B는 NVIDIA L4에서 개방형 어휘 점프샷 검출을 수행할 때 프레임당 8.13초가 걸립니다. 하드웨어가 직접적으로 동일하지는 않지만, 격차의 크기는 아키텍처 제약 조건이 원시 컴퓨팅보다 차이의 대부분을 설명한다는 것을 확인합니다.
이 아키텍처가 합리적인 상황은
이 패턴은 당신의 작업이 고정된 레이블 세트로 분류될 수 있을 때, 실시간 스트림의 지속적인 처리가 필요할 때, 데이터 프라이버시 요구 사항이 외부 API에 프레임을 전송하는 것을 방지할 때, 그리고 초 이하의 지연 시간보다 100밀리초 이하의 지연 시간을 견디는 경우에 강력한 적합성을 갖습니다. 전달带上의 진정한 실시간 응답이 필요할 때는 올바른 도구가 아닙니다, 이 경우는 50밀리초 미만의 지연 시간이 필수적입니다. 그 상황에서는 다시 YOLO 영역에 돌아가거나, 이전 기사에서 설명한 것과 같은 파이프라인을 사용합니다: VLM을 이용해 밤새 데이터 세트를 자동으로 주석화한 다음, 생산 배포를 위해 전용 가벼운 분류기를 훈련합니다.
결론
“VLMs는 비디오에 너무 느리다”와 “VLMs는 생산 비디오 파이프라인에서 작동한다” 사이의 격차는 주로 하드웨어 문제가 아닙니다. 이는 아키텍처 문제입니다. PaliGemma와 같은 복잡하지 않은 모델을 7B-파라미터 대안보다 선택하고, 해상도를 실제 작업이 요구하는 것으로 제한하고, 닫힌 어휘에서 결정론적 디코딩을 강제하고, 예측을 시간적으로 부드럽게 하고, 추론을 캡처 및 렌더링과 분리하는 것: 이러한 것들은 더 큰 GPU가 필요하지 않습니다. 이는 실제로 모델이 무엇을 요청하는지 신중하게 생각하고, 제약 조건을 반대로 파이프라인을 구축하는 것이 아니라 그 제약 조건을 기반으로 파이프라인을 구축하는 것을 요구합니다.
모델 로딩부터 멀티스레드 추론까지 전체 패턴이 파이썬 150라인 미만에 들어맞습니다. 실시간 비디오 스트림에서의 절대적 의미 분류에 대한 합리적인 가격입니다.