add lite mode to support audio-focused youtube video

Author: Xiaoyi

README.md

@@ -9,6 +9,12 @@ This repository contains the official implementation of the paper [Deep Video Di
 ![image](https://github.com/user-attachments/assets/ac1c7f0a-3c10-4c4c-88d1-7bfe0e2010e1)
 
 
+## Update
+
+- **2025/07/16**: Add `lite_mode` to enable a lightweight version of the agent that uses only subtitles. Good for Youtube podcast analysis!
+- **2025/07/14**: Support OpenAI API and Azure OpenAI API.
+- **2025/07/08**: Initial release of the Deep Video Discovery codebase.
+
 ## Introduction
 
 **Deep Video Discovery (DVD)** is a deep-research style question answering agent designed for understanding extra-long videos. Leveraging the powerful capabilities of large language models (LLMs), DVD effectively interprets and processes extensive video content to answer complex user queries.

dvd/config.py

@@ -24,6 +24,7 @@
 AOAI_EMBEDDING_LARGE_DIM = 3072
 
 # ------------------ agent and tool setting ------------------ #
+LITE_MODE = True # if True, only leverage srt subtitle, no pixel downloaded or pixel captioning
 GLOBAL_BROWSE_TOPK = 300
 OVERWRITE_CLIP_SEARCH_TOPK = 0 # 0 means no overwrite and let agent decide
 

dvd/dvd_core.py

@@ -26,6 +26,8 @@ def finish(answer: A[str, D("Answer to the user's question.")]) -> None:
 class DVDCoreAgent:
     def __init__(self, video_db_path, video_caption_path, max_iterations):
         self.tools = [frame_inspect_tool, clip_search_tool, global_browse_tool, finish]
+        if config.LITE_MODE:
+            self.tools.remove(frame_inspect_tool)
         self.name_to_function_map = {tool.__name__: tool for tool in self.tools}
         self.function_schemas = [
             {"function": as_json_schema(func), "type": "function"}

dvd/frame_caption.py

@@ -240,6 +240,8 @@ def _caption_clip(task: Tuple[str, Dict], caption_ckpt_folder) -> Tuple[str, dic
         try:
             assert isinstance(resp, str), f"Response must be a JSON string instead of {type(resp)}:{resp}."
             parsed = json.loads(resp)
+            parsed["clip_description"] += f"\n\nTranscript during this video clip: {transcript}." # add transcript to description
+            resp = json.dumps(parsed)
             with open(os.path.join(caption_ckpt_folder, f"{timestamp}.json"), "w") as f:
                 f.write(resp)
             return timestamp, parsed
@@ -329,6 +331,25 @@ def process_video(
         json.dump(frame_captions, f, indent=4)
 
 
+def process_video_lite(
+    output_caption_folder: str,
+    subtitle_file_path: str,
+):
+    """
+    Process video in LITE_MODE using SRT subtitles.
+    """
+    captions = parse_srt_to_dict(subtitle_file_path)
+    frame_captions = {}
+    for key, text in captions.items():
+        frame_captions[key] = {
+            "caption": f"\n\nTranscript during this video clip: {text}.",
+        }
+    frame_captions["subject_registry"] = {}
+    with open(
+        os.path.join(output_caption_folder, "captions.json"), "w"
+    ) as f:
+        json.dump(frame_captions, f, indent=4)
+
 # --------------------------------------------------------------------------- #
 #                                    main                                     #
 # --------------------------------------------------------------------------- #

dvd/video_utils.py

@@ -99,10 +99,51 @@ def load_video(
             )
             shutil.copy2(subtitle_source, subtitle_destination)
 
-        return os.path.abspath(destination_path)
+def download_srt_subtitle(video_url: str, output_path: str):
+    """Downloads an SRT subtitle from a YouTube URL."""
+    if not _is_youtube_url(video_url):
+        raise ValueError("Provided URL is not a valid YouTube link.")
+
+    output_dir = os.path.dirname(output_path)
+    os.makedirs(output_dir, exist_ok=True)
+
+    ydl_opts = {
+        'writesubtitles': True,
+        'subtitleslangs': ['en'],
+        'subtitlesformat': 'srt',
+        'skip_download': True,
+        'outtmpl': os.path.join(output_dir, '%(id)s.%(ext)s'),
+    }
+
+    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+        info = ydl.extract_info(video_url, download=False)
+        video_id = info['id']
+        ydl.download([video_url])
+
+    # Locate the downloaded subtitle file (yt-dlp names them as <id>.<lang>.srt)
+    downloaded_subtitle_path = None
+    for f in os.listdir(output_dir):
+        if f.startswith(video_id) and f.endswith(".srt"):
+            downloaded_subtitle_path = os.path.join(output_dir, f)
+            break
+
+    if downloaded_subtitle_path:
+        shutil.move(downloaded_subtitle_path, output_path)
+    else:
+        # Try auto-generated subtitles
+        ydl_opts['writeautomaticsub'] = True
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl_auto:
+            ydl_auto.download([video_url])
+        
+        if os.path.exists(os.path.join(output_dir, f"{video_id}.en.vtt")):
+             # yt-dlp might download as .vtt and convert, check for final .srt
+            for f in os.listdir(output_dir):
+                if f.startswith(video_id) and f.endswith('.srt'):
+                    shutil.move(os.path.join(output_dir, f), output_path)
+                    return
+        
+        raise FileNotFoundError(f"Could not find SRT subtitle for {video_url}")
 
-    # ------------------- Not found -------------------
-    raise FileNotFoundError(f"Video source '{video_source}' not found or is not a valid URL.")
 
 def decode_video_to_frames(video_path: str) -> str:
     """

local_run.py

@@ -2,8 +2,8 @@
 import os
 import argparse
 from dvd.dvd_core import DVDCoreAgent
-from dvd.video_utils import load_video, decode_video_to_frames
-from dvd.frame_caption import process_video
+from dvd.video_utils import load_video, decode_video_to_frames, download_srt_subtitle
+from dvd.frame_caption import process_video, process_video_lite
 from dvd.utils import extract_answer
 
 def main():
@@ -27,31 +27,50 @@ def main():
     frames_dir = os.path.join(config.VIDEO_DATABASE_FOLDER, video_id, "frames")
     captions_dir = os.path.join(config.VIDEO_DATABASE_FOLDER, video_id, "captions")
     video_db_path = os.path.join(config.VIDEO_DATABASE_FOLDER, video_id, "database.json")
+    srt_path = os.path.join(config.VIDEO_DATABASE_FOLDER, video_id, "subtitles.srt")
 
-    # Download video
-    if not os.path.exists(video_path):
-        print(f"Downloading video from {video_url} to {video_path}...")
-        load_video(video_url, video_path)
-        print("Video downloaded.")
+    if config.LITE_MODE:
+        print("Running in LITE_MODE.")
+        if not os.path.exists(srt_path):
+            print(f"Downloading SRT subtitle for {video_url} to {srt_path}...")
+            try:
+                download_srt_subtitle(video_url, srt_path)
+                print("SRT subtitle downloaded.")
+            except Exception as e:
+                print(f"Error downloading subtitle: {e}")
+                print("Please turn off LITE_MODE and try again.")
+                return
+        else:
+            print(f"SRT subtitle already exists at {srt_path}.")
+        
+        # In LITE_MODE, we use srt as caption file
+        process_video_lite(captions_dir, srt_path)
+        caption_file = os.path.join(captions_dir, "captions.json")
     else:
-        print(f"Video already exists at {video_path}.")
+        # Download video
+        if not os.path.exists(video_path):
+            print(f"Downloading video from {video_url} to {video_path}...")
+            load_video(video_url, video_path)
+            print("Video downloaded.")
+        else:
+            print(f"Video already exists at {video_path}.")
 
-    # Decode video to frames
-    if not os.path.exists(frames_dir) or not os.listdir(frames_dir):
-        print(f"Decoding video to frames in {frames_dir}...")
-        decode_video_to_frames(video_path)
-        print("Video decoded.")
-    else:
-        print(f"Frames already exist in {frames_dir}.")
+        # Decode video to frames
+        if not os.path.exists(frames_dir) or not os.listdir(frames_dir):
+            print(f"Decoding video to frames in {frames_dir}...")
+            decode_video_to_frames(video_path)
+            print("Video decoded.")
+        else:
+            print(f"Frames already exist in {frames_dir}.")
 
-    # Get captions
-    caption_file = os.path.join(captions_dir, "captions.json")
-    if not os.path.exists(caption_file):
-        print("Processing video to get captions...")
-        process_video(frames_dir, captions_dir)
-        print("Captions generated.")
-    else:
-        print(f"Captions already exist at {caption_file}.")
+        # Get captions
+        caption_file = os.path.join(captions_dir, "captions.json")
+        if not os.path.exists(caption_file):
+            print("Processing video to get captions...")
+            process_video(frames_dir, captions_dir)
+            print("Captions generated.")
+        else:
+            print(f"Captions already exist at {caption_file}.")
 
     # Initialize DVDCoreAgent
     print("Initializing DVDCoreAgent...")