2026-05-29 강의 정리

---

목차

1. 문제와 접근
2. 비디오 메타데이터 확인
3. 프레임 샘플링
4. ffmpeg 오디오 추출
5. Whisper segment 전사
6. 프레임 캡션 생성
7. video_docs 생성
8. FAISS 기반 video RAG
9. video_rag_visual
10. 오디오 파일 일괄 전사
11. 오디오 segment 문서화와 분류
12. 핵심 개념 총정리
13. 자주 나오는 실수/주의사항
14. 답변을 원본 구간으로 되돌릴 수 있게 하기

---

1. 문제와 접근

비디오 RAG에서는 텍스트 내용만큼 그 내용이 나온 시점이 중요하다. 프레임 캡션과 음성 전사를 하나의 검색 흐름에 연결하되 출처와 시간 정보를 유지한다. 프레임 샘플링과 전사, 답변 생성을 나누어 어느 단계에서 필요한 장면을 놓쳤는지 확인할 수 있게 한다.

video
  ├─ frame sampling -> vision caption -> frame Document
  └─ audio extract  -> whisper segment -> audio Document
        ↓
     FAISS index
        ↓
     query -> relevant frames/audio -> vision LLM answer

import cv2
import base64
import subprocess
from pathlib import Path

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_core.messages import HumanMessage, SystemMessage

llm = ChatOpenAI(model="gpt-4o-mini")
vision = ChatOpenAI(model="gpt-4o-mini", max_tokens=200)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

VIDEO_PATH = Path("data/video/earnings_briefing.mp4")
WORK_DIR = Path("day53_video")
WORK_DIR.mkdir(exist_ok=True)

---

2. 비디오 메타데이터 확인

OpenCV로 FPS, 프레임 수, 길이, 해상도를 읽는다.

def video_info(path):
    cap = cv2.VideoCapture(path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    n_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    cap.release()

    return {
        "fps": round(fps, 2),
        "n_frames": n_frames,
        "duration_sec": round(n_frames / fps if fps else 0, 2),
        "width": w,
        "height": h,
    }

---

3. 프레임 샘플링

일정 간격마다 프레임을 추출한다.

def sample_frames(video_path, every_sec, out_dir=None):
    out_dir = Path(out_dir or WORK_DIR / "frames")
    out_dir.mkdir(parents=True, exist_ok=True)
    info_ = video_info(video_path)
    cap = cv2.VideoCapture(video_path)
    items = []
    sec = 0.0
    i = 0

    while sec <= info_["duration_sec"]:
        cap.set(cv2.CAP_PROP_POS_MSEC, sec * 1000)
        ok, frame = cap.read()
        if not ok:
            break

        p = out_dir / f"frame_{i:03d}_{int(sec * 1000):07d}ms.jpg"
        cv2.imwrite(str(p), frame)
        items.append({"path": str(p), "second": round(sec, 2)})

        sec += every_sec
        i += 1

    cap.release()
    return items

---

4. ffmpeg 오디오 추출

비디오에서 mp3 오디오만 추출한다.

def extract_audio(video_path, out_path):
    if Path(out_path).exists():
        return out_path

    cmd = [
        "ffmpeg", "-y",
        "-i", video_path,
        "-vn",
        "-c:a", "libmp3lame",
        "-b:a", "128k",
        out_path,
    ]
    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode != 0:
        raise RuntimeError(r.stderr[-300:])
    return out_path

---

5. Whisper segment 전사

verbose_json으로 segment 단위 시작/종료 시간을 받는다.

from openai import OpenAI
oai_audio = OpenAI()


def transcribe_with_segments(audio_path) -> dict:
    with open(audio_path, "rb") as f:
        res = oai_audio.audio.transcriptions.create(
            model="whisper-1",
            file=f,
            response_format="verbose_json",
            language="ko",
        )

    segs = [
        {"id": s.id, "start": s.start, "end": s.end, "text": s.text}
        for s in (res.segments or [])
    ]
    return {"text": res.text, "segments": segs}

---

6. 프레임 캡션 생성

프레임 이미지를 Vision 모델에 넣어 한 문장 캡션을 만든다.

def _b64(path):
    return base64.b64encode(Path(path).read_bytes()).decode()


def caption_frames(frames: list, prompt: str = "이 프레임에 무엇이 보이는지 한 문장으로 써주세요") -> list:
    out = []
    for fr in frames:
        b = _b64(fr["path"])
        msg = HumanMessage(content=[
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b}"}},
        ])
        cap = vision.invoke([msg]).content
        out.append({**fr, "caption": cap})
    return out

---

7. video_docs 생성

프레임과 음성 문서를 같은 목록에 담더라도 modality와 시간 필드는 유지한다. 검색 결과가 어떤 장면이나 발언에서 왔는지 복원할 수 있어야 같은 시점의 다른 정보와 대조하거나 원본 구간을 다시 확인할 수 있다.

프레임 캡션과 음성 segment를 같은 Document 리스트로 합친다.

def build_video_documents(frames, segments, source) -> list:
    docs = []

    for fr in frames:
        docs.append(Document(
            page_content=fr["caption"],
            metadata={
                "source": source,
                "modality": "frame",
                "second": fr["second"],
                "frame_path": fr["path"],
            },
        ))

    for seg in segments:
        docs.append(Document(
            page_content=seg["text"].strip(),
            metadata={
                "source": source,
                "modality": "audio",
                "start": round(seg["start"], 2),
                "end": round(seg["end"], 2),
            },
        ))

    return docs

---

8. FAISS 기반 video RAG

프레임과 오디오 문서를 함께 임베딩한다.

video_docs = build_video_documents(captioned, audio_seg["segments"], source=VIDEO_PATH.name)
vs = FAISS.from_documents(video_docs, embeddings)

hits = vs.similarity_search("매출 188억", k=3)

for h in hits:
    m = h.metadata
    tag = (
        f"[frame @ {m.get('second')}s]"
        if m["modality"] == "frame"
        else f"[audio {m.get('start')} - {m.get('end')}]"
    )
    print(f"{tag} {h.page_content[:50]}")

---

9. video_rag_visual

검색 결과에 프레임 이미지가 있으면 Vision LLM에 다시 넣어 답변한다.

def video_rag_visual(vs, query, k=4):
    cands = vs.similarity_search(query, k=k)
    text_blocks = []
    image_paths = []

    for d in cands:
        m = d.metadata
        if m["modality"] == "frame":
            image_paths.append((m["second"], m["frame_path"], d.page_content))
        else:
            text_blocks.append(f"[audio {m['start']}-{m['end']}s] {d.page_content}")

    content = [{
        "type": "text",
        "text": f"비디오에서 발췌한 프레임 이미지와 음성 transcribe을 참고하여 한국어로 답하세요.\n\n질문 : {query}",
    }]

    if text_blocks:
        content.append({"type": "text", "text": "[음성발췌]\n" + "\n".join(text_blocks)})

    for sec, path, cap in image_paths:
        content.append({"type": "text", "text": f"\n[frame @ {sec}s] (caption : {cap})"})
        content.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{_b64(path)}"}})

    msg = HumanMessage(content=content)
    return {
        "answer": vision.invoke([msg]).content,
        "n_frames": len(image_paths),
        "n_audio": len(text_blocks),
    }

---

10. 오디오 파일 일괄 전사

LangChain의 Whisper parser로 여러 mp3를 Document로 변환한다.

from langchain_community.document_loaders.parsers.audio import OpenAIWhisperParser
from langchain_core.document_loaders.blob_loaders import Blob

AUDIO_DIR = Path("data/audio")
audio_files = sorted(AUDIO_DIR.glob("*.mp3"))

def transcribe_all(audio_paths):
    parser = OpenAIWhisperParser()
    out = []
    for p in audio_paths:
        blob = Blob.from_path(str(p))
        docs = list(parser.lazy_parse(blob))
        text = "\n".join(d.page_content for d in docs)
        out.append({"source": Path(p).name, "text": text})
    return out

---

11. 오디오 segment 문서화와 분류

이 절의 분류 스키마는 결과의 형태를 정의하는 단계로 읽는다. 오디오별 분류 호출과 결과 저장까지 연결하려면 입력 전사문, 실패 처리, source와 segment의 대응 관계를 추가로 구성해야 한다.

segment 단위로 자르면 검색 결과에서 시간 정보를 함께 제공할 수 있다.

def segments_to_documents(audio_path) -> list:
    tmp = transcribe_with_segments(audio_path)
    source = Path(audio_path).name
    docs = []

    for s in tmp["segments"]:
        if not s["text"].strip():
            continue
        docs.append(Document(
            page_content=s["text"].strip(),
            metadata={
                "source": source,
                "modality": "audio",
                "start": round(s["start"], 2),
                "end": round(s["end"], 2),
                "segment_id": s["id"],
            },
        ))

    return docs

class AudioCategory(BaseModel):
    category: Literal["revenue", "product", "hiring", "other"] = Field(..., description="주제 카테고리")
    confidence: float = Field(..., ge=0, le=1, description="확신도 0~1")
    reason: str = Field(..., description="한국어 짧은 사유")

---

12. 핵심 개념 총정리

개념	역할
OpenCV VideoCapture	비디오 메타데이터/프레임 추출
ffmpeg	비디오에서 오디오 분리
Whisper verbose_json	segment 단위 전사
frame caption	이미지 프레임을 텍스트화
video_docs	frame/audio를 같은 Document로 통합
FAISS	멀티모달 문서 검색
video_rag_visual	검색된 프레임과 음성을 함께 Vision LLM에 전달
AudioCategory	전사 문서 주제 분류

---

13. 자주 나오는 실수/주의사항

실수 1: 프레임 파일을 저장하지 않음

cv2.imwrite(str(p), frame)

실수 2: JPEG data URI 형식 오류

f"data:image/jpeg;base64,{b}"

실수 3: OpenAI 클라이언트 정의 순서

from openai import OpenAI
oai_audio = OpenAI()

transcribe_with_segments()를 호출하기 전에 oai_audio가 정의되어 있어야 한다.

실수 4: ffmpeg 설치 여부

ffmpeg가 시스템에 설치되어 있지 않으면 오디오 추출 단계에서 실패한다.

---

14. 답변을 원본 구간으로 되돌릴 수 있게 하기

답변이 부정확하면 검색된 문서의 시간 구간을 먼저 확인한다. 필요한 장면이 샘플링에서 빠졌는지, 전사가 잘못됐는지, 올바른 근거를 생성기가 놓쳤는지를 나누면 수정할 단계가 명확해진다.

샘플링 간격을 비교할 때는 처리한 프레임 수와 놓친 사건을 함께 기록한다. 모든 프레임을 처리하는 비용과 특정 장면을 회수하는 요구 사이에서 기준을 정하고, 최종 답변에는 확인할 수 있는 원본 위치를 남긴다.