비디오 이상 탐지: 아키텍처, Twelve Labs, NVIDIA VSS

비디오 이상 탐지: 아키텍처, Twelve Labs, NVIDIA VSS (tutorials-build-essentials-video-anomaly-edge-part-1)

이 글은 엣지에서 클라우드까지 실시간 비디오 이상 탐지 시스템을 구축하는 3부작 시리즈의 파트 1이에요. 아키텍처와 통합부터 시작해 프로덕션급 탐지 파이프라인까지 다룰 거예요.

시리즈:

• 파트 1 | 아키텍처, Twelve Labs, NVIDIA VSS (현재 글)
• 파트 2 | 엣지-클라우드 파이프라인
• 파트 3 | 스코어링, 거버넌스, 배포

시간: 90분	난이도: 고급	출력: GitHub

이 튜토리얼에서는 여러 사이트의 실시간 감시 카메라를 모니터링하고, 특정 이상 유형에 대한 훈련 없이도 비정상적인 사건을 자동으로 감지하는 실시간 비디오 이상 탐지 시스템을 구축하는 방법을 배워요. Qdrant Edge가 Twelve Labs 및 NVIDIA Metropolis VSS와 통합되어 Vultr Cloud GPU에 배포되는 프로덕션급 엣지-클라우드 탐지 파이프라인을 만드는 과정을 볼 수 있어요.

---

소개 (Introduction)

감시 카메라가 특정 사건들(싸움, 사고, 침입, 장비 고장)에 대해 훈련된 적 없이도 어떤 유형의 이상*도 자동으로 감지한다면 어떨까요? 운영자가 몇 시간 분량의 영상을 수동으로 검토하는 대신, 심각도 점수와 사건 타임라인, 무슨 일이 있었는지 자연어로 된 설명이 담긴 즉각적인 알림을 받는다면 어떨까요?

이것이 우리가 Qdrant Edge 벡터 검색, Twelve Labs 비디오 인텔리전스, NVIDIA Metropolis, Vultr Cloud GPU가 전통적인 분류기로는 근본적으로 해결할 수 없는 문제를 해결하기 위해 모인 모습을 보여주는 데 만든 것이에요. 바로 한 번도 본 적 없는 이상을 감지하는 문제죠.

"이건 싸움인가?" 또는 "이건 강도인가?"라고 묻는 대신, 우리는 "이건 우리가 평소에 보는 것과 얼마나 다른가?" 라고 묻습니다. 이렇게 하면 이상 탐지를 Qdrant가 강점을 발휘하는 최근접 이웃(kNN) 탐색 문제로 재구성돼요.

이 튜토리얼에서는 단계별 가이드를 따라 애플리케이션을 배포하는 것뿐 아니라, 기술 아키텍처의 이유(why) 도 배우면서 이 모든 것이 어떻게 가능한지 알게 될 거예요.

구체적으로, 실시간 감시 스트림을 다음으로 변환하는 플랫폼을 구축해요.

• 이상 탐지: Qdrant의 일반 기준선(normal baseline)으로부터의 kNN 거리를 사용해 이상 레이블 없이 자동으로 점수를 매긴 클립.
• 사건 보고서: Twelve Labs 임베딩, VLM 캡션, NVIDIA VSS의 오디오 전사를 사용한 다중 신호 사건 형성.
• 의미적 비디오 검색: 모든 카메라와 시간대에 걸친 자연어 쿼리. "이 사건과 유사한 클립을 찾아줘" 또는 "지난주 북쪽 출입구의 비정상 활동을 보여줘".
• 대화형 Q&A: 감지된 사건에 대해 질문하고 Twelve Labs Pegasus를 통해 실제 비디오 콘텐츠에 근거한 답변을 받기.
• 엣지-클라우드 에스컬레이션: NVIDIA Jetson의 경량 엣지 트라이지가 클라우드 처리량을 약 6배 줄이면서 진짜 이상의 약 95% 를 포착.

여기서 시연하는 개념과 기술은 감시를 넘어 적용돼요. 제조 안전, 소매 분석, 교통 모니터링, 또는 이상 탐지가 필요한 어떤 영역에도 같은 아키텍처를 사용할 수 있어요. 기준선 데이터만 바꾸고 탐지 임계값을 조정하면 새 도메인에 맞출 수 있어요.

---

애플리케이션 데모 (Application Demo)

코딩을 시작하기 전에, 프로젝트 저장소와 라이브 데모를 확인해 우리가 무엇을 만들지 미리 알아봐요.

GitHub: qdrant/video-anomaly-edge

라이브 데모: qdrant-edge-video-anomaly.vercel.app

---

학습 목표 (Learning Objectives)

이 시리즈에서 여러분은 다음을 배워요.

• 카테고리별 훈련 없이도 새로운 이상 유형을 감지하는 Qdrant 벡터 검색을 사용한 kNN 이상 탐지기 구축.
• kNN 이상 스코어링 과 의미적 이해를 단일 모델에서 모두 수행하는 비디오 임베딩을 위한 Twelve Labs Marengo 통합.
• 임베딩, VLM 캡션, 오디오 전사, CV 파이프라인을 위한 조율된 비디오 인제스트를 위해 Vultr Cloud GPU에서 NVIDIA Metropolis VSS 배포.
• 라이브 기준선 업데이트로 서브 밀리초 kNN 조회를 위한 NVIDIA Jetson의 2-샤드 Qdrant Edge 아키텍처 구현.
• 앙상블 스코어링, 시간적 부스트, 오프라인 복원력을 갖춘 Vultr 기반 엣지-클라우드 에스컬레이션 파이프라인 구축.
• 시간이 지나도 탐지 품질을 유지하기 위한 격리(quarantine), 정리(scrubbing), 중독 방지(poisoning prevention)를 포함한 기준선 거버넌스 이해.

---

사전 준비 (Prerequisites)

• Python 3.12+: Python 다운로드
• Node.js 20+: Node.js 다운로드
• uv (Python 패키지 관리자): uv 설치
• Docker: Docker 설치
• Twelve Labs API 키: 인증
• Qdrant Cloud 계정 (또는 로컬 인스턴스): Qdrant Cloud
• Vultr 계정 (GPU 인스턴스용): Vultr Cloud GPUs
• FFmpeg: FFmpeg 다운로드 — 비디오 청킹에 필요
• Python, 벡터 데이터베이스, REST API에 대한 중급 수준 이해

---

로컬 환경 설정 (Local Environment Setup)

1. 저장소를 로컬 환경에 클론해요.

git clone https://github.com/qdrant/video-anomaly-edge.git
cd video-anomaly-edge

2. uv로 의존성을 설치해요.

uv sync

3. 환경 변수를 추가해요.

cp .env.example .env

자격 증명으로 .env를 편집해요.

QDRANT_URL=http://localhost:6333
QDRANT_API_KEY=<your-qdrant-api-key>

# Twelve Labs (클라우드 비디오 이해)
TWELVE_LABS_API_KEY=<your-twelve-labs-api-key>
TWELVE_LABS_API_URL=https://api.twelvelabs.io/v1.3
TWELVE_LABS_MARENGO_INDEX_NAME=anomaly-marengo-search
TWELVE_LABS_PEGASUS_INDEX_NAME=anomaly-pegasus-summary
TWELVE_LABS_MARENGO_MODEL=marengo3.0
TWELVE_LABS_PEGASUS_MODEL=pegasus1.2

# NVIDIA VSS
NVIDIA_VSS_BASE_URL=http://localhost:8080
VSS_ENABLED=false

# 모델
MODEL_NAME=MCG-NJU/videomae-base
MODEL_SERVER_URL=http://localhost:9877
ANOMALY_THRESHOLD=0.15

4. 참고용으로 NVIDIA VSS 프레임워크(Twelve Labs 통합 포함)를 클론해요.

git clone https://github.com/qdrant/twelvelabs-nvidia-vss

5. Docker Compose로 전체 스택을 시작해요.

# 핵심 서비스 (백엔드, Qdrant, 프론트엔드)
docker compose up

# NVIDIA VSS 포함 (GPU 필요)
docker compose -f docker-compose.yml -f docker-compose.vss.yml up

6. localhost:4321로 이동해 대시보드에 접근해요.

---

이상 탐지에 kNN이 분류기보다 나은 이유

코드로 들어가기 전에, 이상 탐지에 전통적인 분류기 대신 kNN 벡터 검색을 사용하는 이유를 이해하는 것이 중요해요.

이진 분류기는 감지하려는 모든 이상 유형의 레이블된 예시를 요구해요. 이는 세 가지 근본적인 문제를 만듭니다.

개방 세계(open-world) 커버리지. 가능한 모든 이상을 미리 열거할 수 없어요. UCF-Crime의 13개 카테고리로 훈련된 분류기는 지게차 충돌이나 파이프 파열에는 0.0 점수를 매겨요. 잘못될 수 있는 것들의 공간은 무한하죠.

레이블 노이즈. 감시 영상은 모호해요. "달리는 사람"이 이상인가요? 전적으로 맥락에 달려 있어요. kNN은 정상 행동에 대해서만 레이블을 요구함으로써 이 문제를 우회해요.

개념 드리프트(concept drift). "정상"을 구성하는 것은 시간이 지남에 따라 변해요. 학교 복도는 수업 시간과 쉬는 시간이 다르게 보여요. kNN 기준선은 재훈련 없이도 지속적으로 업데이트될 수 있어요.

kNN 접근 방식은 단순하고 효과적이에요. 비디오 클립을 벡터 공간에 임베딩하고, Qdrant에 정상 임베딩의 기준선을 구축한 뒤, 최근접 이웃이 먼 클립에 플래그를 지정해요.

anomaly_score = 1 - mean(top_k_cosine_similarities)

유사한 정상 클립으로 둘러싸인 클립은 0에 가까운 점수를 받아요. 기준선의 어떤 것에서도 먼 클립은 1에 가까운 점수를 받아요. 훈련 루프도, 이상에 대한 클래스 레이블도, 치명적 망각(catastrophic forgetting)도 없어요.

이 설계는 디버깅도 쉽게 해줘요. 프로세스 전체에서 정확히 무슨 일이 일어나는지 따라갈 수 있어요.

이것이 왜 중요한지 숫자로 증명해 볼게요. CLIP ViT-B/32(512차원, 단일 프레임 이미지 임베딩)를 대안으로 테스트했는데 0.23 AUC-ROC로 무작위에 가까웠어요. 이 실패는 교훈적이에요. 감시 이상은 시간적 패턴으로 정의돼요. 달리는 사람, 전개되는 싸움, 펼쳐지는 자동차 충돌 같은 것들이죠. 단일 프레임 임베딩은 "서 있는 사람"과 "쓰러지는 사람"을 구분할 수 없어요. 이상이 프레임 안이 아니라 프레임 사이에 존재하기 때문이에요.

모델	AUC-ROC	비고
Twelve Labs Marengo (클라우드)	0.9696	비디오 전용, 시간적 역학 포착
EfficientNet-B0 (엣지)	~0.85	공간 특징만, 높은 재현율 트라이지
CLIP ViT-B/32	0.23	단일 프레임, 시간적 맥락 없음, 실패

이것이 클라우드에서 Twelve Labs Marengo를 사용하는 이유예요. 비디오 이해를 위해 특별히 제작되어 시간적 역학, 객체 상호작용, 장면 맥락을 통합 신호로 처리해요.

---

아키텍처 개요 (Architecture Overview)

이 시스템은 "엣지에서 저렴하고 빠른 트라이지, 클라우드에서 정확하고 풍부한 분석" 이라는 단순한 원칙에 따라 설계된 3계층 아키텍처를 사용해요.

아키텍처:

• 엣지 계층 (NVIDIA Jetson)

  • NVIDIA Metropolis:
    • DeepStream (고성능 비디오 디코드 + 추론)
    • Video Storage Toolkit
    • IoT Gateway (알림 에스컬레이션용)
  • Qdrant Edge (2-샤드)
    • Mutable(가변) 샤드: 들어오는 영상의 실시간 쓰기
    • Immutable(불변) 샤드: HNSW, 동기화된 기준선
  • 트라이지:
    • 높은 재현율: score → escalate 또는 로컬 드롭

• 클라우드 계층 (Vultr GPUs)

  • NVIDIA Metropolis VSS:
    • Twelve Labs Marengo (비디오 임베딩 및 이해)
    • VLM 기반 캡션
    • 오디오 전사
    • 전체 컴퓨터 비전 파이프라인
  • Qdrant Cloud:
    • 벡터 + payload용 통합 인덱스
    • 풍부한 사건 관리
    • 기준선 추적 및 거버넌스

• 에스컬레이션 파이프라인

  • 이상 후보(에스컬레이션된 클립)만 엣지에서 클라우드로 전송되어 효율을 극대화해요.

• 대시보드

  • 실시간(WebSocket) 사건·점수 모니터링을 갖춘 Next.js 프론트엔드

---

이 계층적 엣지-클라우드 접근 방식은 각 카메라 사이트에서 저렴하고 고속의 트라이지를 수행하고, 흥미로운(잠재적) 이상만 더 깊은 클라우드 분석, 의미적 강화, 장기 거버넌스를 위해 보내요. 엣지는 느슨한 임계값을 갖고 있어 이상일 것 같은 것까지 전부 태그해요.

엣지에서는 오탐(거짓 양성)을 미탐(거짓 음성)보다 선호해요. 사건이 아닌 것을 사건으로 잘못 레이블하는 것은 클라우드 쿼리 한 번만 소모하지만, 이상을 놓치는 것은 치명적일 수 있기 때문이에요.

엣지 계층: 각 카메라 사이트는 Jetson의 NVIDIA Metropolis를 실행해요. 이 하드웨어 가속 장치는 DeepStream으로 비디오를 디코드하고 추론을 실행하며, Video Storage Toolkit으로 카메라 관리를 활용하고, IoT Gateway로 안전한 클라우드 전송을 확보해요. Qdrant Edge는 Metropolis 옆에 위치해 서브 밀리초 kNN 조회를 위한 2-샤드 로컬 기준선을 저장해요. 엣지는 최종 이상 판단을 내리는 대신 클라우드에 도달하는 영상의 양을 줄이는 높은 재현율 트라이지 필터로 작동해요.

클라우드 계층: 에스컬레이션된 클립은 Vultr Cloud GPU에서 실행되는 NVIDIA Metropolis VSS로 흐릅니다. Twelve Labs는 Qdrant에서 kNN 이상 스코어링을 위한 비디오 임베딩 과 사건 이해, 검색, Q&A를 위한 풍부한 의미적 신호를 생성해요. 하나의 모델이 두 작업을 모두 처리해요. 중앙 집중식 Qdrant 클러스터가 모든 표현을 인덱싱해요. 앙상블 스코어링(클라우드 70%, 엣지 30%), 시간적 평활화, 사건 형성이 여기에서 일어나요.

대시보드: 사건, 장치 상태, 이상 점수 타임라인을 보여주는 실시간 WebSocket 업데이트를 갖춘 Next.js 프론트엔드.

---

Twelve Labs 통합 구축

그렇다면 실제로 시간적 역학을 이해하는 비디오 임베딩을 어떻게 얻을까요? 여기서 Twelve Labs가 등장해요.

Twelve Labs는 우리 아키텍처를 위한 두 가지 핵심 모델을 제공해요. Marengo는 임베딩과 검색을 처리하며 비디오에서 무슨 일이 있었는지를 시각·오디오 수준에서 이해해요. Pegasus는 대화형 분석을 처리해 비디오에 대해 질문하고 상세하고 근거 있는 답변을 받을 수 있게 해요. 함께 kNN 스코어링 파이프라인과 조사용 Q&A 워크플로우를 모두 다뤄요.

이들을 백엔드에 어떻게 통합하는지 살펴볼게요.

클라이언트는 .env의 TWELVE_LABS_API_KEY에서 초기화되는 단순한 싱글턴이에요. 두 모델은 Twelve Labs에 별도의 인덱스가 필요해요.

def _ensure_index(index_name: str, models: list[dict]) -> str:
    """Twelve Labs 인덱스를 가져오거나 생성하고, 그 ID를 반환."""
    client = get_client()
    indexes = client.index.list()
    for idx in indexes:
        if idx.name == index_name:
            return idx.id
    idx = client.index.create(name=index_name, models=models)
    return idx.id

업로드 및 임베딩 (Upload and Embed)

엣지에서 에스컬레이션된 클립이 도착하면, 인덱싱을 위해 Twelve Labs에 업로드해요.

def upload_video(file_path: str | Path, index_type: str = "both") -> dict:
    client = get_client()
    result = {}

    if index_type in ("marengo", "both"):
        idx_id = get_marengo_index_id()
        task = client.task.create(index_id=idx_id, file=str(file_path))
        task.wait_for_done(timeout=UPLOAD_TIMEOUT)
        if task.status == "ready":
            result["marengo_video_id"] = task.video_id

    if index_type in ("pegasus", "both"):
        idx_id = get_pegasus_index_id()
        task = client.task.create(index_id=idx_id, file=str(file_path))
        task.wait_for_done(timeout=UPLOAD_TIMEOUT)
        if task.status == "ready":
            result["pegasus_video_id"] = task.video_id

    return result

Twelve Labs에 업로드하면 단일 API 호출로 모든 무거운 작업(청킹, 임베딩, 인덱싱)을 처리해요. 클라우드 계층에 로컬 GPU가 필요 없어요.

Marengo로 의미적 검색

비디오가 인덱싱되면 의미적 검색은 간단해져요.

def search_videos(query: str, max_clips: int = 10) -> list[SearchResult]:
    client = get_client()
    idx_id = get_marengo_index_id()

    search_results = client.search.query(
        index_id=idx_id,
        search_options=["visual", "audio"],
        query_text=query,
        group_by="clip",
        threshold="medium",
        page_limit=max_clips,
        sort_option="score",
    )

    results = []
    for group in search_results.data:
        for clip in group.clips:
            results.append(SearchResult(
                video_id=clip.video_id,
                score=clip.score,
                start=clip.start,
                end=clip.end,
                confidence=clip.confidence,
            ))
    return results

이것이 조사 워크플로우를 뒷받침해요. kNN으로 이상이 감지된 후 운영자는 자연어로 유사 사건을 검색할 수 있어요. "로딩 독 근처에서 달리는 사람" 또는 "주차장에서 차량 충돌" 같은 쿼리로요. Marengo는 시각·오디오 신호를 모두 이해해요.

Pegasus로 비디오 Q&A

더 깊은 조사를 위해 Pegasus는 인덱싱된 비디오의 대화형 분석을 제공해요.

def analyze_video(video_id: str, prompt: str) -> AnalysisResult:
    client = get_client()
    t0 = time.perf_counter()

    response = client.generate.text(
        video_id=video_id,
        prompt=prompt,
        temperature=0.2,
    )

    latency_ms = (time.perf_counter() - t0) * 1000
    return AnalysisResult(
        text=response.data if hasattr(response, "data") else str(response),
        video_id=video_id,
        latency_ms=latency_ms,
    )

공장 소유주, 보안 팀, 또는 컴플라이언스 담당자는 "이 클립에서 보이는 안전 위반 사항은 무엇인가요?" 또는 "사건으로 이어지는 일련의 사건을 설명해 주세요" 같은 질문을 하고 즉시 비디오에 근거한 상세 답변을 받을 수 있어요. 이것을 비디오에 대한 RAG로 생각해 보세요.

---

NVIDIA VSS 연결하기

코드에 들어가기 전에 NVIDIA VSS가 무엇이고 어떤 도구를 제공하는지 다뤄볼게요.

VSS는 Video Search and Summarization(비디오 검색 및 요약) 의 약자예요. 이는 NVIDIA AI Blueprint로, 개발자들이 비디오 콘텐츠를 이해·검색·요약할 수 있는 강력한 AI 에이전트를 빠르게 배포할 수 있는 방법을 제공해요. 다음을 제공해요.

• 비전 언어 모델(VLM): 비디오 프레임을 VLM에 입력해 각 비디오 청크에 대한 풍부한 텍스트 설명을 생성해요.
• 대형 언어 모델(LLM): VLM의 텍스트 설명을 요약과 자연어 Q&A를 위해 LLM에 공급해요.
• 검색 증강 생성(RAG): VSS는 생성된 설명을 벡터/그래프 데이터베이스에 저장해요. 질문은 가장 관련성 높은 청크를 먼저 검색해 근거 있는 답변으로 이어져요.
• GPU 가속 인제스트: 파일 또는 실시간 RTSP 스트림에서 비디오를 가져오고, 디코딩하고, AI 모델에 준비하는 고성능 파이프라인.
• CV 파이프라인 통합: 객체 감지 모델(YOLO 또는 NVIDIA DeepStream SDK)과 함께 작동해 메타데이터를 추가해요.
• 오디오 전사: 음성-텍스트용 오디오 트랙을 처리해 또 다른 검색 가능한 계층을 추가해요.

이 NVIDIA VSS blueprint는 매우 강력하지만, 구축·배포·관리할 것도 많아요. 바로 여기서 Twelve Labs가 엄청난 가속기를 제공하며 VLM, 오디오, 추론 복잡성을 단일 API로 추상화해요.

진정한 가치는 모듈성에 있어요. 우리 아키텍처는 VSS 내에서 Twelve Labs 통합을 사용해 Twelve Labs가 지능을 처리하고 VSS가 오케스트레이션을 처리하는 하이브리드 워크플로우를 만들어요.

VSS용 비디오 청킹

VSS 파이프라인의 첫 단계는 청킹이에요. Twelve Labs x NVIDIA VSS 제조 샘플의 패턴에 따라 FFmpeg의 세그먼트 멀티플렉서로 비디오를 나눕니다.

/backend/vss.py

def chunk_video(
    input_path: str | Path,
    output_dir: str | Path,
    chunk_duration_s: float | None = None,
) -> list[Path]:
    """FFmpeg 세그먼트 멀티플렉서로 비디오를 청크로 분할."""
    input_path = Path(input_path)
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    # ffprobe로 비디오 길이 가져오기
    probe = subprocess.run(
        ["ffprobe", "-v", "quiet", "-show_entries", "format=duration",
         "-of", "default=noprint_wrappers=1:nokey=1", str(input_path)],
        capture_output=True, text=True,
    )
    duration = float(probe.stdout.strip()) if probe.stdout.strip() else 60.0

    if chunk_duration_s is None:
        if duration < 60:
            chunk_duration_s = duration  # 짧은 비디오는 분할하지 않음
        else:
            chunk_duration_s = duration / 30  # 약 30개 청크

    pattern = output_dir / f"{input_path.stem}_chunk_%04d.mp4"

    subprocess.run(
        ["ffmpeg", "-y", "-i", str(input_path),
         "-c", "copy", "-map", "0",
         "-segment_time", str(chunk_duration_s),
         "-f", "segment", "-reset_timestamps", "1",
         str(pattern)],
        capture_output=True,
    )

    return sorted(output_dir.glob(f"{input_path.stem}_chunk_*.mp4"))

왜 청킹을 하나요? 제조 샘플과 동일한 비용 문제가 여기 적용돼요. 24시간 분량의 원시 비디오를 처리하는 것은 비싸요. 우리 엣지 계층은 이미 영상의 약 85%를 걸러내고, 나머지 에스컬레이션 클립을 청킹하면 클라우드 파이프라인이 더욱 최적화돼요. 관심 있는 청크만 전체 VSS 스택을 통과해요.

VSS로 비동기 업로드

청크는 async I/O를 사용해 병렬로 VSS에 업로드돼요.

async def upload_to_vss(file_path: str | Path) -> Optional[str]:
    """단일 비디오 파일을 NVIDIA VSS에 업로드."""
    file_path = Path(file_path)

    with open(file_path, "rb") as f:
        content = f.read()

    timeout = aiohttp.ClientTimeout(total=VSS_UPLOAD_TIMEOUT)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        data = aiohttp.FormData()
        data.add_field("file", content,
                       filename=file_path.name, content_type="video/mp4")
        data.add_field("purpose", "vision")
        data.add_field("media_type", "video")

        async with session.post(f"{NVIDIA_VSS_BASE_URL}/files", data=data) as resp:
            if resp.status == 200:
                body = await resp.json()
                return body.get("id")
            return None


async def upload_chunks_to_vss(chunk_paths: list[Path]) -> list[str]:
    """여러 청크를 VSS에 병렬로 업로드."""
    tasks = [upload_to_vss(p) for p in chunk_paths]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [r for r in results if isinstance(r, str)]

전체 인제스트 파이프라인은 청킹, 업로드, 결과 반환을 묶어서 처리해요.

async def ingest_video(file_path: str | Path) -> dict:
    """전체 VSS 인제스트 파이프라인: 비디오 청킹, 모든 청크 업로드."""
    if not is_enabled():
        return {"status": "disabled", "message": "VSS_ENABLED is false"}

    with tempfile.TemporaryDirectory(prefix="vss_chunks_") as tmp_dir:
        chunks = chunk_video(file_path, tmp_dir)
        file_ids = await upload_chunks_to_vss(chunks)

    return {
        "status": "ok",
        "total_chunks": len(chunks),
        "uploaded_chunks": len(file_ids),
        "vss_file_ids": file_ids,
    }

VSS용 Docker Compose

핵심 서비스와 함께 VSS를 배포하려면 docker compose 파일을 사용해요.

/docker-compose.vss.yml

## docker compose -f docker-compose.yml -f docker-compose.vss.yml up
services:
  vss-server:
    image: nvcr.io/nvidia/metropolis/vss:1.0
    ports:
      - "8080:8080"
    environment:
      TWELVE_LABS_API_KEY: ${TWELVE_LABS_API_KEY}
      VLM_MODEL_TO_USE: twelve-labs
      DISABLE_CV_PIPELINE: "true"
      DISABLE_CA_RAG: "true"
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    networks:
      - anomaly-net

  backend:
    environment:
      VSS_ENABLED: "true"
      NVIDIA_VSS_BASE_URL: http://vss-server:8080
    depends_on:
      - vss-server

참고: VLM_MODEL_TO_USE: twelve-labs를 설정하면 VSS가 로컬 VLM을 실행하는 대신 Twelve Labs 원격 배포를 사용하도록 지시해요. 이를 통해 전용 VLM GPU가 필요 없어져요. Twelve Labs가 API로 지능을 처리하고, VSS는 오케스트레이션 파이프라인을 처리해요.

---

요약 (Recap)

파트 1에서 프로젝트를 설정하고, 개방 세계 감시에서 Qdrant의 kNN 이상 탐지가 전통적인 분류기보다 나은 이유를 배우고, 비디오 임베딩과 Q&A를 위해 Twelve Labs Marengo와 Pegasus를 통합하고, GPU 가속 인제스트를 위해 NVIDIA VSS를 연결했어요. 아키텍처가 갖춰졌어요. 이제 엣지를 구축할 차례예요.

다음 단계 (What's Next)

파트 2 | 엣지-클라우드 파이프라인 에서는 2-샤드 Qdrant Edge 아키텍처, 엣지 트라이지 스코어링, 앙상블 스코어링을 갖춘 에스컬레이션 흐름, 오프라인 복원력을 구현할 거예요.

파트 3 | 스코어링, 거버넌스, 배포 에서는 사건 형성, 기준선 거버넌스, 통합 검색, UCF-Crime 결과, Vultr Cloud GPU 배포를 다룰 거예요.

---

추가 리소스:

• 프로젝트 저장소: qdrant/video-anomaly-edge
• NVIDIA VSS Twelve Labs 통합: qdrant/twelvelabs-nvidia-vss
• Twelve Labs 문서: docs.twelvelabs.io
• Qdrant 문서: qdrant.tech/documentation
• Vultr Cloud GPUs: vultr.com/products/cloud-gpu

출처: Qdrant 공식문서 - Video Anomaly Detection: Architecture, Twelve Labs, and NVIDIA VSS

더 알아보기 (Learn more)

• 비디오 이상 탐지 파트 2 | 엣지-클라우드 파이프라인
• 비디오 이상 탐지 파트 3 | 스코어링, 거버넌스, 배포
• Twelve Labs 문서
• NVIDIA VSS AI Blueprint