VideoDB에 연결하기

VideoDB를 LlamaIndex 리트리버(VideoDBRetriever)로 사용해 영상의 음성(spoken) 콘텐츠와 시각(Scene) 콘텐츠를 검색하고, 두 모달리티를 결합한 멀티모달 RAG를 구축하는 방법을 보여드려요. 멀티 비디오 컬렉션에서 RAG를 만드는 방법과 리트리버 설정 옵션까지 살펴볼게요.

출처: 문서

본문

Step 1: VideoDB에 연결하고 비디오 업로드

connect()로 연결한 뒤 컬렉션을 만들고 비디오를 업로드합니다.

conn = connect()
coll = conn.create_collection(
    name="VideoDB Retrievers",
    description="VideoDB Retrievers",
)

# upload videos to default collection in VideoDB

print("Uploading Video")
video = coll.upload(url="https://www.youtube.com/watch?v=aRgP3n0XiMc")
print(f"Video uploaded with ID: {video.id}")

# video = coll.get_video("m-b6230808-307d-468a-af84-863b2c321f05")

coll = conn.get_collection() : 기본 컬렉션 객체를 반환합니다.

coll.get_videos() : 컬렉션의 모든 비디오 목록을 반환합니다.

coll.get_video(video_id): 주어진 video_id에서 Video 객체를 반환합니다.

🗣️ Step 2: 음성 콘텐츠의 인덱싱과 검색

비디오는 서로 다른 모달리티를 가진 데이터로 볼 수 있습니다. 먼저 spoken content(음성 콘텐츠)를 다뤄 보겠습니다.

🗣️ 음성 콘텐츠 인덱싱

print("Indexing spoken content in Video...")
video.index_spoken_words()

🗣️ 음성 인덱스에서 관련 노드 검색

VideoDBRetriever를 사용해 인덱싱된 콘텐츠에서 관련 노드를 검색합니다. 비디오 ID를 파라미터로 전달하고, index_type을 IndexType.spoken_word로 설정해야 합니다.

score_threshold와 result_threshold는 실험 후에 구성할 수 있습니다.

from llama_index.retrievers.videodb import VideoDBRetriever
from videodb import SearchType, IndexType


spoken_retriever = VideoDBRetriever(
    collection=coll.id,
    video=video.id,
    search_type=SearchType.semantic,
    index_type=IndexType.spoken_word,
    score_threshold=0.1,
)


spoken_query = "Nationwide exams"
nodes_spoken_index = spoken_retriever.retrieve(spoken_query)

🗣️ 결과 보기 : 💬 텍스트

관련 노드를 사용해 llamaindex로 응답을 합성합니다.

from llama_index.core import get_response_synthesizer


response_synthesizer = get_response_synthesizer()


response = response_synthesizer.synthesize(
    spoken_query, nodes=nodes_spoken_index
)
print(response)

🗣️ 결과 보기 : 🎥 비디오 클립

쿼리와 관련된 각 검색 노드에 대해, 노드의 start와 end 필드가 노드가 다루는 시간 구간을 나타냅니다.

VideoDB의 Programmable Stream을 사용해 이 노드들의 타임스탬프에 기반한 관련 비디오 클립 스트림을 생성합니다.

from videodb import play_stream


results = [
    (node.metadata["start"], node.metadata["end"])
    for node in nodes_spoken_index
]


stream_link = video.generate_stream(results)
play_stream(stream_link)

📸 Step 3: 시각 콘텐츠의 인덱싱과 검색

Scene Index에 대해 더 알아보려면 다음 가이드를 참고하세요:

Quickstart Guide는 Scene Index를 단계별로 소개합니다. 빠르게 시작하고 주요 기능을 이해하기에 이상적입니다.
Scene Extraction Options Guide는 Scene Index 내에서 씬 추출에 사용할 수 있는 다양한 옵션을 심층적으로 다룹니다. 고급 설정, 커스터마이즈 기능, 그리고 다양한 요구·선호도에 따라 씬 추출을 최적화하는 팁을 다룹니다.

from videodb import SceneExtractionType


print("Indexing Visual content in Video...")


# Index scene content
index_id = video.index_scenes(
    extraction_type=SceneExtractionType.shot_based,
    extraction_config={"frame_count": 3},
    prompt="Describe the scene in detail",
)
video.get_scene_index(index_id)


print(f"Scene Index successful with ID: {index_id}")

📸 Scene Index에서 관련 노드 검색

음성 인덱스에 VideoDBRetriever를 사용했던 것처럼 씬 인덱스에도 동일하게 사용합니다. 여기서는 index_type을 IndexType.scene으로 설정하고 scene_index_id를 전달해야 합니다.

from llama_index.retrievers.videodb import VideoDBRetriever
from videodb import SearchType, IndexType




scene_retriever = VideoDBRetriever(
    collection=coll.id,
    video=video.id,
    search_type=SearchType.semantic,
    index_type=IndexType.scene,
    scene_index_id=index_id,
    score_threshold=0.1,
)


scene_query = "accident scenes"
nodes_scene_index = scene_retriever.retrieve(scene_query)

📸 결과 보기 : 💬 텍스트

from llama_index.core import get_response_synthesizer


response_synthesizer = get_response_synthesizer()


response = response_synthesizer.synthesize(
    scene_query, nodes=nodes_scene_index
)
print(response)

📸 결과 보기 : 🎥 비디오 클립

from videodb import play_stream


results = [
    (node.metadata["start"], node.metadata["end"])
    for node in nodes_scene_index
]


stream_link = video.generate_stream(results)
play_stream(stream_link)

🛠️ Step 4: 단순 멀티모달 RAG — 두 모달리티의 결과 결합

비디오 라이브러리에서 다음과 같은 멀티모달 쿼리를 열고 싶습니다:

📸🗣️ "Show me 1.Accident Scene 2.Discussion about nationwide exams"

멀티모달 RAG를 만드는 방법은 여러 가지가 있지만, 단순함을 위해 간단한 접근 방식을 선택합니다:

🧩 쿼리 변환(Query Transformation): 쿼리를 씬 인덱스와 음성 인덱스 각각에 사용할 수 있는 두 부분으로 나눕니다.
🔎 각 모달리티에 대한 관련 노드 찾기: VideoDBRetriever를 사용해 음성 인덱스와 씬 인덱스에서 관련 노드를 찾습니다.
✏️ 결과 보기 : 텍스트: 정확한 비디오 세그먼트 식별을 위해 두 인덱스의 결과를 통합해 관련 노드로 텍스트 응답을 합성합니다.
🎥 결과 보기 : 비디오 클립: 정확한 비디오 세그먼트 식별을 위해 두 인덱스의 결과를 통합합니다.

더 고급 멀티모달 기법을 확인하려면 고급 멀티모달 가이드를 참고하세요.

🧩 쿼리 변환

from llama_index.llms.openai import OpenAI




def split_spoken_visual_query(query):
    transformation_prompt = """
    Divide the following query into two distinct parts: one for spoken content and one for visual content. The spoken content should refer to any narration, dialogue, or verbal explanations and The visual content should refer to any images, videos, or graphical representations. Format the response strictly as:\nSpoken: <spoken_query>\nVisual: <visual_query>\n\nQuery: {query}
    """
    prompt = transformation_prompt.format(query=query)
    response = OpenAI(model="gpt-4").complete(prompt)
    divided_query = response.text.strip().split("\n")
    spoken_query = divided_query[0].replace("Spoken:", "").strip()
    scene_query = divided_query[1].replace("Visual:", "").strip()
    return spoken_query, scene_query




query = "Show me 1.Accident Scene 2.Discussion about nationwide exams "
spoken_query, scene_query = split_spoken_visual_query(query)
print("Query for Spoken retriever : ", spoken_query)
print("Query for Scene retriever : ", scene_query)

🔎 각 모달리티에 대한 관련 노드 찾기

from videodb import SearchType, IndexType


# Retriever for Spoken Index
spoken_retriever = VideoDBRetriever(
    collection=coll.id,
    video=video.id,
    search_type=SearchType.semantic,
    index_type=IndexType.spoken_word,
    score_threshold=0.1,
)


# Retriever for Scene Index
scene_retriever = VideoDBRetriever(
    collection=coll.id,
    video=video.id,
    search_type=SearchType.semantic,
    index_type=IndexType.scene,
    scene_index_id=index_id,
    score_threshold=0.1,
)


# Fetch relevant nodes for Spoken index
nodes_spoken_index = spoken_retriever.retrieve(spoken_query)


# Fetch relevant nodes for Scene index
nodes_scene_index = scene_retriever.retrieve(scene_query)

💬 결과 보기 : 텍스트

response_synthesizer = get_response_synthesizer()


response = response_synthesizer.synthesize(
    query, nodes=nodes_scene_index + nodes_spoken_index
)
print(response)

🎥 결과 보기 : 비디오 클립

각 모달리티에서 해당 모달리티(이 경우 의미/씬·시각) 내에서 쿼리와 관련된 결과를 가져왔습니다.

각 노드는 메타데이터에 start와 end 필드를 가지며, 이는 노드가 다루는 시간 구간을 나타냅니다.

결과를 합성하는 방법은 여러 가지가 있으며, 여기서는 단순한 방법을 사용합니다:

Union: 모든 노드의 모든 타임스탬프를 가져와, 한 모달리티에만 나타나는 타임스탬프라도 모든 관련 시간을 포함하는 종합적인 목록을 만듭니다.

다른 방법으로는 Intersection이 있습니다:

Intersection: 모든 노드에 존재하는 타임스탬프만 포함해, 모든 모달리티에 걸쳐 보편적으로 관련된 시간으로 구성된 더 작은 목록을 만듭니다.

from videodb import play_stream




def merge_intervals(intervals):
    if not intervals:
        return []
    intervals.sort(key=lambda x: x[0])
    merged = [intervals[0]]
    for interval in intervals[1:]:
        if interval[0] <= merged[-1][1]:
            merged[-1][1] = max(merged[-1][1], interval[1])
        else:
            merged.append(interval)
    return merged




# Extract timestamps from both relevant nodes
results = [
    [node.metadata["start"], node.metadata["end"]]
    for node in nodes_spoken_index + nodes_scene_index
]
merged_results = merge_intervals(results)


# Use Videodb to create a stream of relevant clips
stream_link = video.generate_stream(merged_results)
play_stream(stream_link)

🛠 VideoDBRetriever로 비디오 컬렉션용 RAG 구축

컬렉션에 비디오 더 추가하기

video_2 = coll.upload(url="https://www.youtube.com/watch?v=kMRX3EA68g4")

🗣️ 음성 콘텐츠 인덱싱

video_2.index_spoken_words()

📸 씬 인덱싱

from videodb import SceneExtractionType


print("Indexing Visual content in Video...")


# Index scene content
index_id = video_2.index_scenes(
    extraction_type=SceneExtractionType.shot_based,
    extraction_config={"frame_count": 3},
    prompt="Describe the scene in detail",
)
video_2.get_scene_index(index_id)


print(f"Scene Index successful with ID: {index_id}")

🧩 쿼리 변환

query = "Show me 1.Accident Scene 2.Kiara is speaking "
spoken_query, scene_query = split_spoken_visual_query(query)
print("Query for Spoken retriever : ", spoken_query)
print("Query for Scene retriever : ", scene_query)

🔎 관련 노드 찾기

여러 비디오 컬렉션에서 검색할 때는 video 파라미터 없이 collection만 지정해 컬렉션 전체를 대상으로 검색할 수 있습니다.

from videodb import SearchType, IndexType


# Retriever for Spoken Index
spoken_retriever = VideoDBRetriever(
    collection=coll.id,
    search_type=SearchType.semantic,
    index_type=IndexType.spoken_word,
    score_threshold=0.2,
)


# Retriever for Scene Index
scene_retriever = VideoDBRetriever(
    collection=coll.id,
    search_type=SearchType.semantic,
    index_type=IndexType.scene,
    score_threshold=0.2,
)


# Fetch relevant nodes for Spoken index
nodes_spoken_index = spoken_retriever.retrieve(spoken_query)


# Fetch relevant nodes for Scene index
nodes_scene_index = scene_retriever.retrieve(scene_query)

💬 결과 보기 : 텍스트

response_synthesizer = get_response_synthesizer()


response = response_synthesizer.synthesize(
    "What is kaira speaking. And tell me about accident scene",
    nodes=nodes_scene_index + nodes_spoken_index,
)
print(response)

🎥 결과 보기 : 비디오 클립

여러 비디오를 다루는 편집 워크플로우에서는 VideoAsset의 Timeline을 만들고 이를 컴파일해야 합니다.

from videodb import connect, play_stream
from videodb.timeline import Timeline
from videodb.asset import VideoAsset


# Create a new timeline Object
timeline = Timeline(conn)


for node_obj in nodes_scene_index + nodes_spoken_index:
    node = node_obj.node


    # Create a Video asset for each node
    node_asset = VideoAsset(
        asset_id=node.metadata["video_id"],
        start=node.metadata["start"],
        end=node.metadata["end"],
    )


    # Add the asset to timeline
    timeline.add_inline(node_asset)


# Generate stream for the compiled timeline
stream_url = timeline.generate_stream()
play_stream(stream_url)

Configuring `VideoDBRetriever`

⚙️ 단일 비디오용 리트리버

비디오 객체의 id를 전달하면 해당 비디오에서만 검색합니다.

VideoDBRetriever(video="my_video.id")

⚙️ 비디오 집합/컬렉션용 리트리버

컬렉션의 id를 전달하면 해당 컬렉션에서만 검색합니다.

VideoDBRetriever(collection="my_coll.id")

⚙️ 다양한 유형의 인덱스용 리트리버

from videodb import IndexType
spoken_word = VideoDBRetriever(index_type=IndexType.spoken_word)


scene_retriever = VideoDBRetriever(index_type=IndexType.scene, scene_index_id="my_index_id")

⚙️ 리트리버의 검색 유형 구성

search_type은 주어진 쿼리에 대해 노드를 검색할 때 사용하는 검색 방법을 결정합니다.

from videodb import SearchType, IndexType


keyword_spoken_search = VideoDBRetriever(
    search_type=SearchType.keyword,
    index_type=IndexType.spoken_word
)


semantic_scene_search = VideoDBRetriever(
    search_type=SearchType.semantic,
    index_type=IndexType.spoken_word
)

⚙️ 임계값 파라미터 구성

result_threshold: 리트리버가 반환하는 결과 수의 임계값; 기본값은 5
score_threshold: score_threshold보다 높은 점수를 가진 노드만 리트리버가 반환; 기본값은 0.2

custom_retriever = VideoDBRetriever(result_threshold=2, score_threshold=0.5)

✨ 인덱싱과 청킹 구성

이 예시에서는 비디오 검색에 VideoDB의 인덱싱을 활용했습니다. 하지만 Transcript와 Scene 데이터를 모두 로드하고 llamaindex를 사용해 자체 인덱싱 기법을 적용할 수도 있습니다.

더 자세한 안내는 이 가이드를 참고하세요.

🏃‍♂️ 다음 단계

이 가이드에서는 VideoDB, LlamaIndex, OpenAI를 사용해 영상을 위한 단순 멀티모달 RAG를 구축했습니다.

더 고급 기법을 통합해 파이프라인을 최적화할 수 있습니다:

쿼리 변환 최적화
서로 다른 모달리티에서 검색한 노드를 결합하는 더 많은 방법
Knowledge Graph 같은 다양한 RAG 파이프라인 실험

관련 클립을 만드는 데 사용한 Programmable Stream 기능에 대해 더 알아보려면 Dynamic Video Stream Guide를 참고하세요.

Scene Index에 대해 더 알아보려면 다음 가이드를 참고하세요:

👨‍👩‍👧‍👦 지원 및 커뮤니티

질문이나 피드백이 있다면 언제든 연락 주세요 🙌🏼

VideoDB에 연결하기