Build a Video RAG Agent

Introduction

30 Second Summary

Coding tutorials can move faster than your questions. A useful line can disappear from the screen before you know what to ask about it.

In this project, you will build a terminal-based Video RAG Agent that turns a short coding tutorial into searchable evidence. Your agent grounds Gemini's answers in timestamped evidence from narration or the tutorial screen.

What You'll Build

When you ask your finished agent a question in the terminal, you'll see its retrieved timestamps followed by a grounded answer about the tutorial.

By the end of this project, you'll have:

• A meaning-aware search that uses Chroma to surface the right tutorial moment even when your question uses different words.
• A grounded terminal answer with evidence-source labels plus timestamp citations you can verify against the tutorial.
• Local-first processing that keeps your raw video, complete transcript, OCR output, embeddings, and vector database on your Mac.
• Secret Mission: Red-team the agent with narration-led, screen-only, and unanswerable questions before documenting whether every answer is supported.

Are there any prerequisites?

The project requires a Mac, internet access, plus an English coding tutorial lasting two to five minutes. Your tutorial must be public or non-sensitive, saved as tutorial.mp4, because free-tier Gemini content is used to improve Google's products.

Before We Start

This checkpoint locks in the English, non-sensitive two-to-five-minute coding tutorial you are indexing as tutorial.mp4. Your local-first privacy goal keeps the raw video, transcription, OCR, embeddings, and vector database on your Mac. Only retrieved text and selected frames may be sent to Gemini.

Set Up the Local Video AI Toolkit

Your Video RAG pipeline depends on several local tools working together. An unconfirmed setup could stop media decoding or OCR before the video becomes searchable.

Homebrew manages the local installations. Python runs the processing pipeline inside an isolated environment.

Tesseract extracts text from tutorial screenshots. Visual Studio Code gives you one workspace for the project files.

Chroma stores searchable vectors on your Mac. The Gemini API handles final answer generation after retrieval selects the relevant evidence.

In this step, get ready to:

• Install the required macOS development tools.
• Prepare a Python 3.12 environment with pinned packages.
• Authorize the toolkit with a Gemini API key.

Install the macOS toolkit

Homebrew provides the official macOS installation path for Python 3.12 and Tesseract 5.5.3. It also installs the current Visual Studio Code application.

• Check whether Homebrew is available by running this command:

brew --version

What does this command check?

The command asks Homebrew to print its installed version. A version response confirms that the package manager is available in your terminal.

✔️ I see a Homebrew version

Good. Homebrew is ready to install the local toolkit.

ⓧ Command not found

Homebrew is missing from your Mac. Its installer may request your macOS password or ask you to confirm before continuing.

• Install Homebrew by running its official installation command:

/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"

What does this installer do?

The command downloads the official Homebrew installation script over HTTPS. The script explains each change before it installs the package manager.

• Run the shell path commands printed at the end of the installer.
• Confirm that Homebrew is now available by running:

brew --version

What confirms success?

A Homebrew version confirms that your shell can now find the package manager.

Homebrew still unavailable?

Return to the final lines of the installer output. Run every shell path command shown there before checking the version again.

Help me fix my Homebrew installation.

The next check separates a complete toolkit from an outdated or missing installation. Python must report version 3.12 or higher.

• Check the installed Python, Tesseract, and Visual Studio Code versions by running:

python3.12 --version
tesseract --version
code --version

What do these checks prove?

• The Python command confirms that the dedicated python3.12 executable is available.
• The Tesseract command prints the installed OCR engine version.
• The Visual Studio Code command confirms that its terminal launcher is available.

✔️ All tools meet the required versions

Your system tools are compatible with the project. Continue to the English language-data check below.

ⓧ One tool is outdated

An older tool can produce dependency failures or different command behavior. Homebrew installs the project-compatible versions.

• Install the required versions by running:

brew install python@3.12
brew install tesseract
brew install --cask visual-studio-code

What gets installed?

• Python 3.12 provides the runtime used by the virtual environment.
• Tesseract 5.5.3 provides local optical character recognition.
• Visual Studio Code provides the editor plus the code terminal launcher.

Expect these installations to take several minutes while Homebrew downloads the packages.

• Confirm the updated installations by running:

python3.12 --version
tesseract --version
code --version

What confirms the update?

The commands should now print Python 3.12 or higher plus successful Tesseract and Visual Studio Code version details.

An old version still appears?

Use the dedicated python3.12 executable during environment creation. This avoids an older system Python executable.

Help me resolve an outdated local tool.

ⓧ A command is missing

At least one required tool is absent or unavailable through your shell path. Homebrew can install all three tools from one terminal session.

• Install the missing tools by running:

brew install python@3.12
brew install tesseract
brew install --cask visual-studio-code

What gets installed?

• Python 3.12 provides the project runtime.
• Tesseract provides local OCR with English language data.
• Visual Studio Code provides the project editor.

The installation can take several minutes while Homebrew downloads each tool.

• Confirm that every command is now available by running:

python3.12 --version
tesseract --version
code --version

What confirms the installation?

You should receive version details from all three commands without a missing-command error.

A command is still missing?

Close Terminal after Homebrew finishes. Start a fresh Terminal session so macOS reloads the updated shell path.

Help me find a missing toolkit command.

Tesseract needs English language data to read code labels and interface text from your tutorial frames. Homebrew includes the language code eng in its standard Tesseract installation.

• List the available Tesseract languages by running:

tesseract --list-langs

What does this list show?

Tesseract reports every installed language data set. The entry eng confirms that English OCR is ready.

You should see eng among the available languages.

Prepare the project workspace

A virtual environment keeps this project's package versions separate from the rest of your Mac. The pinned requirements make every later import match the complete project code.

• Move to your Desktop by running this command:

cd ~/Desktop

Why start on the Desktop?

The project folder will have a predictable location. You can find it easily from Terminal or Finder.

• Create the video-rag-agent folder and move into it by running these commands:

mkdir video-rag-agent
cd video-rag-agent

What do these commands create?

The first command creates the project folder on your Desktop. The second command makes that folder the active Terminal location.

• Create the Python 3.12 virtual environment by running:

python3.12 -m venv .venv
source .venv/bin/activate
python --version

What does this setup do?

• The first command creates an isolated environment in .venv.
• The second command activates that environment in your current shell.
• The final command confirms which Python version the active environment uses.

You should see Python 3.12 in the output. Your Terminal prompt should also begin with (.venv).

• Open the active video-rag-agent folder in Visual Studio Code by running:

code .

What does this command open?

The dot represents your current folder. Visual Studio Code opens video-rag-agent as the project workspace.

You should see video-rag-agent at the top of the Visual Studio Code Explorer sidebar.

• Press Cmd+Space to open macOS search.
• Type Finder and press Enter.
• Locate the tutorial.mp4 video you selected earlier.
• Drag tutorial.mp4 into the Visual Studio Code Explorer sidebar.
• Return to the Terminal window from earlier.
• Confirm that the video is in the project folder by running:

ls tutorial.mp4

What confirms the video location?

The command should print tutorial.mp4. That result confirms the video is in the project root.

• Click the New File button in the Visual Studio Code Explorer sidebar.
• Name the file requirements.txt.
• Paste these pinned dependencies into requirements.txt:

av==18.1.0
chromadb==1.5.9
faster-whisper==1.2.1
google-genai==2.24.0
pillow==12.3.0

What do these packages provide?

• PyAV 18.1.0 decodes the video into timestamped frames.
• Chroma 1.5.9 stores local embeddings plus evidence metadata.
• faster-whisper 1.2.1 transcribes the tutorial locally.
• Google Gen AI Python SDK 2.24.0 sends retrieved evidence to Gemini.
• Pillow 12.3.0 lets PyAV save decoded frames as JPEG images.

• Save requirements.txt.
• Install the pinned packages into the active virtual environment by running:

python -m pip install -r requirements.txt

What does this installation do?

The command reads each exact package version from requirements.txt. It installs those packages inside the active .venv environment.

The first installation can take several minutes while Python downloads the packages and their dependencies.

You should see a successful installation message after the final package finishes.

Package installation failed?

Check that your Terminal prompt begins with (.venv). Confirm that python --version reports Python 3.12.

Help me debug the pinned package installation.

✔️ Awesome, I've got everything!

Great. Save requirements.txt before moving on to the API key.

ⓧ I'd like to double check the full code

Compare your complete requirements.txt file with this reference:

av==18.1.0
chromadb==1.5.9
faster-whisper==1.2.1
google-genai==2.24.0
pillow==12.3.0

What should match?

Each package name and version must match the reference exactly. These pins define the environment used by the complete project code.

Authorize the toolkit

A Gemini API key authorizes the final generation request. The local ingestion pipeline does not need the key.

Keep the tutorial non-sensitive

The project keeps the raw video plus its full index on your Mac. Only retrieved text and selected frames are sent to Gemini.

Gemini free-tier content may be used to improve Google's products. Continue with the public or non-sensitive tutorial you selected.

API keys deserve a careful moment. Keep this key out of screenshots and project files.

• Navigate to Google AI Studio in your browser.
• Sign in with your Google account.
• Open the API Keys page.
• Create a new authorization API key.
• Copy the generated key to your clipboard.

• Open ~/.zshrc in Visual Studio Code by running:

code ~/.zshrc

Why use this file?

The ~/.zshrc file configures new zsh sessions on your Mac. Saving the export there makes the API key available after you reopen Terminal.

• Add the following export line to the bottom of ~/.zshrc:
• Replace your-api-key with the authorization key you copied.

export GEMINI_API_KEY='your-api-key'

What does this export do?

The export creates the GEMINI_API_KEY environment variable for your shell. The Google Gen AI Python SDK detects this variable automatically.

• Save ~/.zshrc.
• Return to the Terminal window from earlier.
• Load the updated shell configuration by running:

source ~/.zshrc

Why reload the shell configuration?

The command applies the saved export to your current Terminal session. Your project can now read the API key without storing it in video-rag-agent.

Before you run the final check, which command do you think is most likely to reveal a missing dependency?

• Verify the complete toolkit by running these commands:

python --version
tesseract --version
code --version
python -c "import av; import chromadb; from faster_whisper import WhisperModel; from google import genai; print('Toolkit ready')"

What does the final check cover?

• The first command verifies the Python version inside .venv.
• The next two commands verify the OCR engine and editor launcher.
• The final command imports the four Python libraries used by the pipeline.

You should see Python 3.12 plus successful version output from Tesseract and Visual Studio Code. The final line should read Toolkit ready.

Toolkit check failed?

If a Python import fails, confirm that (.venv) appears at the start of your Terminal prompt. Run the requirements installation again from video-rag-agent.

If a system command fails, return to the matching installation outcome above.

Help me debug the toolkit verification.

That's the setup locked in. Your local toolkit can now decode video, run OCR, store vectors, and reach Gemini when retrieved evidence is ready.

Extract Timestamped Video Frames

Your local video AI toolkit is ready. Now you can turn tutorial.mp4 into evidence that your agent can inspect.

Questions about code shown on screen cannot be answered from narration alone. PyAV exposes each decoded frame's presentation time so you can preserve where it came from.

In this step, get ready to:

• Create the script paths and timestamp helpers.
• Decode the tutorial and sample frames every 10 seconds.
• Run the frames command and inspect readable JPEG screenshots.

Create the script foundation

The project uses one Python module for video processing and retrieval. Its shared imports keep every stage connected to the same paths.

• Use the file sidebar in Visual Studio Code to create video_rag.py inside the video-rag-agent folder.
• Add the project imports by pasting this code into video_rag.py:

import argparse
import base64
import json
import os
import subprocess
import sys
from pathlib import Path

import av
import chromadb
from faster_whisper import WhisperModel
from google import genai

What Do These Imports Provide?

• The argparse module turns terminal arguments into commands that the script can handle.
• Path creates reliable paths relative to video_rag.py.
• av provides the local video decoder used in this step.
• The remaining imports support the transcription and retrieval stages that share this module.

• Save video_rag.py.
• Confirm that every imported package loads by running this command in the Terminal from earlier:

python -c "print('Imports ready' if __import__('video_rag') else '')"

What Does This Check Prove?

Python imports video_rag.py before printing the confirmation. A successful import proves that the pinned packages are available inside your active virtual environment.

You should see Imports ready in the terminal.

Imports Not Loading?

Confirm that the terminal is still using the activated .venv from the video-rag-agent folder.

Check that you copied every import exactly as shown. A misspelled module name stops the script before the confirmation prints.

Help me diagnose the failed imports in video_rag.py.

• Add the project paths and helper functions below the imports by pasting this code:

BASE_DIR = Path(__file__).resolve().parent
FRAMES_DIR = BASE_DIR / "frames"
INDEX_PATH = BASE_DIR / "video_index.json"
CHROMA_PATH = BASE_DIR / "chroma_db"
COLLECTION_NAME = "video_tutorial"
FRAME_INTERVAL_SECONDS = 10.0
CHUNK_SECONDS = 20.0
TRANSCRIPTION_MODEL = "tiny.en"
GENERATION_MODEL = "gemini-3.8-flash"
TOP_K = 3


def format_timestamp(seconds):
    total_seconds = max(0, int(seconds))
    minutes, secs = divmod(total_seconds, 60)
    hours, minutes = divmod(minutes, 60)
    if hours:
        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
    return f"{minutes:02d}:{secs:02d}"


def require_video(video_path):
    path = Path(video_path).expanduser().resolve()
    if not path.exists():
        raise FileNotFoundError(f"Video not found: {path}")
    return path

How Does the Foundation Work?

• BASE_DIR anchors every generated path to the folder containing the script.
• FRAME_INTERVAL_SECONDS sets the sampling interval to 10.0 seconds.
• format_timestamp() converts raw seconds into readable video timestamps.
• require_video() resolves the supplied path before checking that the video exists.

• Save video_rag.py.
• Test the timestamp formatter by running this command:

python -c "print(__import__('video_rag').format_timestamp(70))"

What Does This Timestamp Show?

The check passes 70 seconds into format_timestamp(). The helper converts that value into the minute-and-second format used in the frame report.

You should see 01:10 in the terminal.

Timestamp Check Failing?

Compare the indentation inside format_timestamp() with the code above. Every line after the function declaration must stay inside the function.

Help me fix the timestamp helper in video_rag.py.

Decode and sample the video

Video frame sampling keeps representative screenshots instead of saving every decoded image. The sampler also removes frames from a previous run so the folder always matches the current video.

• Add the initial extract_frames() function below require_video() by pasting this code:

def extract_frames(video_path):
    video_path = require_video(video_path)
    FRAMES_DIR.mkdir(exist_ok=True)

    for old_frame in FRAMES_DIR.glob("frame_*.jpg"):
        old_frame.unlink()

    sampled_frames = []
    next_sample_time = 0.0
    return sampled_frames

What Does the Sampling Setup Do?

• require_video() stops processing when the supplied video path does not exist.
• FRAMES_DIR.mkdir() creates the local frames directory.
• glob("frame_*.jpg") finds screenshots created by an earlier extraction.
• next_sample_time tracks the next presentation time that should produce a screenshot.

• Save video_rag.py.
• Confirm that the sampler prepares its destination by running this command:

python -c "print(len(__import__('video_rag').extract_frames('tutorial.mp4')))"

What Does This Preparation Check?

The command invokes extract_frames() directly. Its returned list reports how many sampled frame records the current function produced.

You should see 0 because this first layer prepares the destination and sampling state.

Video Path Not Found?

Confirm that tutorial.mp4 appears beside video_rag.py in the video-rag-agent folder.

Help me locate tutorial.mp4 for the frame sampler.

• In extract_frames() add the decoding block immediately before the final return sampled_frames line:

    with av.open(str(video_path)) as container:
        if not container.streams.video:
            raise RuntimeError("The input file has no video stream.")

        stream = container.streams.video[0]
        for frame in container.decode(stream):
            if frame.time is None:
                continue

            timestamp = float(frame.time)
            if timestamp + 0.001 < next_sample_time:
                continue

            frame_path = FRAMES_DIR / f"frame_{len(sampled_frames):04d}.jpg"
            frame.to_image().save(str(frame_path), quality=80)
            sampled_frames.append(
                {
                    "timestamp_seconds": timestamp,
                    "timestamp": format_timestamp(timestamp),
                    "path": str(frame_path),
                }
            )
            next_sample_time = timestamp + FRAME_INTERVAL_SECONDS

    if not sampled_frames:
        raise RuntimeError("No video frames could be decoded.")

How Does Frame Extraction Work?

• av.open() opens tutorial.mp4 for local decoding.
• container.decode(stream) yields video frames in presentation order.
• frame.time provides the real presentation time in seconds.
• frame.to_image().save() converts each selected frame into a JPEG file with image quality set to 80.
• sampled_frames records each readable timestamp beside its saved file path.

• Save video_rag.py.
• Run the extraction function directly to count its sampled frames:

python -c "print(len(__import__('video_rag').extract_frames('tutorial.mp4')))"

What Does the Frame Count Prove?

The function decodes the tutorial before returning its frame records. A positive count confirms that PyAV found the video stream and saved sampled images.

You should see a number greater than 0 in the terminal.

No Frames Extracted?

Confirm that tutorial.mp4 plays normally on your Mac. A damaged file or a video without a decodable video stream prevents screenshots from being created.

Check that the decoding block sits inside extract_frames() with the indentation shown above.

Help me debug local frame extraction with PyAV.

Add the frames command

A command-line interface gives the script named actions with their own inputs. The frames action accepts one video path and sends it to extract_frames().

• Add build_parser() below extract_frames() by pasting this code:

def build_parser():
    parser = argparse.ArgumentParser(
        description="Build and query a local-first Video RAG index."
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    frames_parser = subparsers.add_parser("frames")
    frames_parser.add_argument("video")

    return parser

How Does the Parser Read the Command?

• add_subparsers() requires the learner to choose a named action.
• add_parser("frames") registers the frame-extraction action.
• add_argument("video") captures the supplied video path as args.video.

• Save video_rag.py.
• Confirm that the parser recognises the command by running:

python -c "print(__import__('video_rag').build_parser().parse_args(['frames', 'tutorial.mp4']).command)"

What Does This Parser Check Prove?

The command passes the same arguments that the terminal command uses. Printing frames confirms that the parser selected the correct action.

You should see frames in the terminal.

Parser Check Failing?

Confirm that return parser remains inside build_parser().

Check that both uses of frames use the same lowercase spelling.

Help me fix the frames parser in video_rag.py.

• Add main() below build_parser() by pasting this code:

def main():
    args = build_parser().parse_args()

    if args.command == "frames":
        frames = extract_frames(args.video)
        print(f"Saved {len(frames)} timestamped frames:")
        for frame in frames:
            print(f"  {frame['timestamp']}  {frame['path']}")
        return


if __name__ == "__main__":
    try:
        main()
    except (FileNotFoundError, RuntimeError, subprocess.CalledProcessError) as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1)

How Does the Command Run?

• main() parses the terminal arguments before selecting the matching action.
• extract_frames(args.video) passes the supplied video path into the sampler.
• The output loop prints each frame's presentation timestamp beside its local path.
• The final error handler prints readable failures to the terminal and exits cleanly.

• Save video_rag.py.

✔️ Awesome, I've got everything!

Your frame sampler is assembled. Confirm that video_rag.py is saved before running it.

ⓧ I'd like to double check the full code

Compare your complete video_rag.py file with this cumulative version:

import argparse
import base64
import json
import os
import subprocess
import sys
from pathlib import Path

import av
import chromadb
from faster_whisper import WhisperModel
from google import genai

BASE_DIR = Path(__file__).resolve().parent
FRAMES_DIR = BASE_DIR / "frames"
INDEX_PATH = BASE_DIR / "video_index.json"
CHROMA_PATH = BASE_DIR / "chroma_db"
COLLECTION_NAME = "video_tutorial"
FRAME_INTERVAL_SECONDS = 10.0
CHUNK_SECONDS = 20.0
TRANSCRIPTION_MODEL = "tiny.en"
GENERATION_MODEL = "gemini-3.8-flash"
TOP_K = 3


def format_timestamp(seconds):
    total_seconds = max(0, int(seconds))
    minutes, secs = divmod(total_seconds, 60)
    hours, minutes = divmod(minutes, 60)
    if hours:
        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
    return f"{minutes:02d}:{secs:02d}"


def require_video(video_path):
    path = Path(video_path).expanduser().resolve()
    if not path.exists():
        raise FileNotFoundError(f"Video not found: {path}")
    return path


def extract_frames(video_path):
    video_path = require_video(video_path)
    FRAMES_DIR.mkdir(exist_ok=True)

    for old_frame in FRAMES_DIR.glob("frame_*.jpg"):
        old_frame.unlink()

    sampled_frames = []
    next_sample_time = 0.0

    with av.open(str(video_path)) as container:
        if not container.streams.video:
            raise RuntimeError("The input file has no video stream.")

        stream = container.streams.video[0]
        for frame in container.decode(stream):
            if frame.time is None:
                continue

            timestamp = float(frame.time)
            if timestamp + 0.001 < next_sample_time:
                continue

            frame_path = FRAMES_DIR / f"frame_{len(sampled_frames):04d}.jpg"
            frame.to_image().save(str(frame_path), quality=80)
            sampled_frames.append(
                {
                    "timestamp_seconds": timestamp,
                    "timestamp": format_timestamp(timestamp),
                    "path": str(frame_path),
                }
            )
            next_sample_time = timestamp + FRAME_INTERVAL_SECONDS

    if not sampled_frames:
        raise RuntimeError("No video frames could be decoded.")

    return sampled_frames


def build_parser():
    parser = argparse.ArgumentParser(
        description="Build and query a local-first Video RAG index."
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    frames_parser = subparsers.add_parser("frames")
    frames_parser.add_argument("video")

    return parser


def main():
    args = build_parser().parse_args()

    if args.command == "frames":
        frames = extract_frames(args.video)
        print(f"Saved {len(frames)} timestamped frames:")
        for frame in frames:
            print(f"  {frame['timestamp']}  {frame['path']}")
        return


if __name__ == "__main__":
    try:
        main()
    except (FileNotFoundError, RuntimeError, subprocess.CalledProcessError) as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1)

What Should the Full File Contain?

This reference combines the shared imports and project constants with the timestamp helpers. It also includes local frame extraction plus the working frames command.

Before you run the complete command, how many screenshots do you expect a video frame sampler using 10-second intervals to save from your tutorial?

• Run the completed frame sampler from the active project terminal:

python video_rag.py frames tutorial.mp4

What Does This Command Do?

The frames subcommand decodes tutorial.mp4 locally. It recreates the sampled JPEG files before printing each real presentation timestamp and path.

You should see a saved-frame count followed by timestamped paths ending in names such as frame_0000.jpg and frame_0001.jpg.

• Return to the video-rag-agent window in Visual Studio Code.
• Expand the frames directory in the file sidebar.
• Open several generated .jpg files from different points in the tutorial.

You should see readable screenshots from the tutorial at approximately 10-second intervals. That first visual evidence layer is now working.

Frames Command Not Working?

Confirm that the terminal remains inside video-rag-agent with .venv activated.

Check that tutorial.mp4 sits beside video_rag.py. Also compare your extract_frames() indentation with the full-file reference.

Help me debug the frames command.

Your tutorial now has a timestamped visual trail. Next, you will combine those screenshots with local narration and screen text.

Build the Local Video Index

Your frame sampler now turns tutorial.mp4 into readable screenshots with real video timestamps.

Separate screenshots and speech segments are difficult to search. This step aligns both forms of evidence before the retrieval system arrives.

Local faster-whisper produces timestamped transcription. Tesseract performs optical character recognition on each sampled frame.

The script combines those results into 20-second evidence chunks. It saves the finished index as a searchable JSON file.

In this step, get ready to:

• Transcribe the tutorial narration with timestamps.
• Extract searchable text from the sampled screenshots.
• Combine both evidence sources into 20-second chunks.

Transcribe the tutorial locally

The transcription model processes the tutorial on your Mac. Each returned segment records its spoken text plus its start and end times.

• In video_rag.py, locate the final return sampled_frames line inside extract_frames.
• Add transcribe_video immediately below extract_frames by pasting this code:

def transcribe_video(video_path):
    model = WhisperModel(
        TRANSCRIPTION_MODEL,
        device="cpu",
        compute_type="int8",
    )
    segments, _ = model.transcribe(str(video_path), beam_size=5)

    transcript = []
    for segment in segments:
        text = segment.text.strip()
        if text:
            transcript.append(
                {
                    "start": float(segment.start),
                    "end": float(segment.end),
                    "text": text,
                }
            )
    return transcript

What does this code do?

• WhisperModel loads the local tiny.en transcription model.
• device="cpu" keeps transcription compatible with the learner environment.
• compute_type="int8" reduces the computation required for local inference.
• The loop converts each non-empty speech segment into reusable text with numeric start and end times.

• Save video_rag.py.
• Confirm that transcribe_video appears directly after extract_frames in your file.

Transcription function looks incomplete?

Check that return transcript aligns with the first line inside the function. It must sit outside the for loop.

Compare the opening and closing brackets around each transcript record. A missing bracket prevents the script from starting.

Help me check the transcription function.

Align OCR with 20-second chunks

OCR turns the pixels in each JPEG into searchable screen evidence. The chunking function places that text beside narration from the same part of the tutorial.

• Add ocr_frame directly below transcribe_video by pasting this code:

def ocr_frame(frame_path):
    result = subprocess.run(
        ["tesseract", str(frame_path), "-", "--psm", "6"],
        check=True,
        capture_output=True,
        text=True,
    )
    return result.stdout.strip()

How does local OCR work?

• subprocess.run invokes the local Tesseract executable for one sampled frame.
• The - output target sends recognized text back to Python.
• --psm 6 treats the screenshot as one uniform block of text.
• check=True stops indexing if OCR fails instead of silently storing missing evidence.

• Save video_rag.py.
• Confirm that the command list inside ocr_frame contains the frame path plus --psm and 6.

• Add evidence_type directly below ocr_frame by pasting this code:

def evidence_type(transcript, screen_text):
    if transcript and screen_text:
        return "both"
    if transcript:
        return "narration"
    if screen_text:
        return "screen"
    return "none"

Why label the evidence?

Each chunk records whether its evidence came from narration or the screen. The both label identifies a window where the two sources overlap.

These labels later reveal what supported a retrieved answer.

• Save video_rag.py.
• Confirm that evidence_type returns both, narration, screen, or none.

The next function carries several responsibilities inside one 20-second window. The four code sections below form one continuous build_chunks function.

• Add the complete build_chunks function below evidence_type by pasting the next four sections in order:

def build_chunks(video_path):
    frames = extract_frames(video_path)
    transcript_segments = transcribe_video(video_path)

    for frame in frames:
        frame["screen_text"] = ocr_frame(frame["path"])

    duration_candidates = [segment["end"] for segment in transcript_segments]
    duration_candidates.extend(
        frame["timestamp_seconds"] + FRAME_INTERVAL_SECONDS for frame in frames
    )
    duration = max(duration_candidates, default=0.0)

    chunks = []
    chunk_start = 0.0

    while chunk_start < duration:
        chunk_end = chunk_start + CHUNK_SECONDS
        matching_segments = [
            segment
            for segment in transcript_segments
            if segment["end"] > chunk_start and segment["start"] < chunk_end
        ]
        matching_frames = [
            frame
            for frame in frames
            if chunk_start <= frame["timestamp_seconds"] < chunk_end
        ]

How are matching records selected?

The function refreshes the sampled frames before transcribing the same video. It then uses the latest narration end time plus the frame timestamps to determine the indexed duration.

Each loop selects speech segments and screenshots that overlap the current 20-second window.

        transcript = " ".join(segment["text"] for segment in matching_segments).strip()
        screen_text = "\n\n".join(
            f"[{frame['timestamp']}] {frame['screen_text']}"
            for frame in matching_frames
            if frame["screen_text"]
        ).strip()

        if transcript or screen_text:
            midpoint = chunk_start + CHUNK_SECONDS / 2
            representative = min(
                matching_frames,
                key=lambda frame: abs(frame["timestamp_seconds"] - midpoint),
                default=None,
            )
            timestamp = (
                f"{format_timestamp(chunk_start)}-"
                f"{format_timestamp(min(chunk_end, duration))}"
            )

How is each window summarized?

Narration becomes one searchable sentence sequence. OCR records retain the timestamps of the frames that produced them.

The frame nearest the middle of the window becomes its representative image. This gives later retrieval one relevant screenshot to attach.

            chunks.append(
                {
                    "id": len(chunks),
                    "start": round(chunk_start, 3),
                    "end": round(min(chunk_end, duration), 3),
                    "timestamp": timestamp,
                    "transcript": transcript,
                    "screen_text": screen_text,
                    "evidence_type": evidence_type(transcript, screen_text),
                    "representative_frame": (
                        representative["path"] if representative else None
                    ),
                }
            )

        chunk_start = chunk_end

What does a chunk contain?

Every searchable record receives an ID plus numeric and formatted timestamps. It also stores narration or OCR text alongside an evidence label.

The representative frame path keeps the original visual evidence connected to the searchable record.

    if not chunks:
        raise RuntimeError("No searchable transcript or screen text was produced.")

    payload = {
        "source_video": str(video_path),
        "frame_interval_seconds": FRAME_INTERVAL_SECONDS,
        "chunk_seconds": CHUNK_SECONDS,
        "chunks": chunks,
    }
    INDEX_PATH.write_text(json.dumps(payload, indent=2), encoding="utf-8")
    return chunks

How is the index saved?

An empty result raises an error because a blank index cannot support retrieval. A successful result stores the video path plus both sampling intervals beside the chunk records.

json.dumps formats the payload with indentation before INDEX_PATH writes video_index.json.

• Add load_chunks directly below build_chunks by pasting this code:

def load_chunks():
    if not INDEX_PATH.exists():
        raise FileNotFoundError(
            "video_index.json is missing. Run the index command first."
        )
    payload = json.loads(INDEX_PATH.read_text(encoding="utf-8"))
    return payload["chunks"]

Why add a loader now?

load_chunks gives later search commands one reliable way to read the saved evidence. Its missing-file check directs the learner back to indexing before retrieval begins.

• Save video_rag.py.
• Confirm that load_chunks appears directly after the complete build_chunks function.

Chunking code shifted out of place?

Keep every line from transcript = through chunk_start = chunk_end indented inside the while loop.

Keep the empty-index check plus the payload at the same indentation level as the while statement. That placement finishes the loop before saving the file.

Help me fix the chunking function.

Run the index command

The new functions need an index subcommand before the terminal can reach them. The command accepts the path to the tutorial video.

• In build_parser, locate the line frames_parser.add_argument("video").
• Add the new parser immediately below that line by pasting this code:

    index_parser = subparsers.add_parser("index")
    index_parser.add_argument("video")

What does this parser add?

The parser recognizes index as a valid command. Its required video argument becomes args.video inside main.

• In main, locate the frames command block ending with return.
• Add the new index command block immediately below it by pasting this code:

    if args.command == "index":
        video_path = require_video(args.video)
        chunks = build_chunks(video_path)
        print(f"Saved {len(chunks)} chunks to {INDEX_PATH}")
        print("Sample chunk:")
        print(json.dumps(chunks[0], indent=2)[:1200])
        return

What does the command print?

The command validates the video path before building the index. It then prints the number of saved chunks plus a sample record from the generated file.

That sample exposes the timestamped narration and screen evidence without requiring a separate inspection tool.

• Save video_rag.py.

✔️ Awesome, I've got everything!

Your indexing functions are connected to the command-line parser. Double check that you saved video_rag.py before running the index.

ⓧ I'd like to double check the full code

Compare your cumulative video_rag.py file with this version:

import argparse
import base64
import json
import os
import subprocess
import sys
from pathlib import Path

import av
import chromadb
from faster_whisper import WhisperModel
from google import genai

BASE_DIR = Path(__file__).resolve().parent
FRAMES_DIR = BASE_DIR / "frames"
INDEX_PATH = BASE_DIR / "video_index.json"
CHROMA_PATH = BASE_DIR / "chroma_db"
COLLECTION_NAME = "video_tutorial"
FRAME_INTERVAL_SECONDS = 10.0
CHUNK_SECONDS = 20.0
TRANSCRIPTION_MODEL = "tiny.en"
GENERATION_MODEL = "gemini-3.8-flash"
TOP_K = 3


def format_timestamp(seconds):
    total_seconds = max(0, int(seconds))
    minutes, secs = divmod(total_seconds, 60)
    hours, minutes = divmod(minutes, 60)
    if hours:
        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
    return f"{minutes:02d}:{secs:02d}"


def require_video(video_path):
    path = Path(video_path).expanduser().resolve()
    if not path.exists():
        raise FileNotFoundError(f"Video not found: {path}")
    return path


def extract_frames(video_path):
    video_path = require_video(video_path)
    FRAMES_DIR.mkdir(exist_ok=True)

    for old_frame in FRAMES_DIR.glob("frame_*.jpg"):
        old_frame.unlink()

    sampled_frames = []
    next_sample_time = 0.0

    with av.open(str(video_path)) as container:
        if not container.streams.video:
            raise RuntimeError("The input file has no video stream.")

        stream = container.streams.video[0]
        for frame in container.decode(stream):
            if frame.time is None:
                continue

            timestamp = float(frame.time)
            if timestamp + 0.001 < next_sample_time:
                continue

            frame_path = FRAMES_DIR / f"frame_{len(sampled_frames):04d}.jpg"
            frame.to_image().save(str(frame_path), quality=80)
            sampled_frames.append(
                {
                    "timestamp_seconds": timestamp,
                    "timestamp": format_timestamp(timestamp),
                    "path": str(frame_path),
                }
            )
            next_sample_time = timestamp + FRAME_INTERVAL_SECONDS

    if not sampled_frames:
        raise RuntimeError("No video frames could be decoded.")

    return sampled_frames


def transcribe_video(video_path):
    model = WhisperModel(
        TRANSCRIPTION_MODEL,
        device="cpu",
        compute_type="int8",
    )
    segments, _ = model.transcribe(str(video_path), beam_size=5)

    transcript = []
    for segment in segments:
        text = segment.text.strip()
        if text:
            transcript.append(
                {
                    "start": float(segment.start),
                    "end": float(segment.end),
                    "text": text,
                }
            )
    return transcript


def ocr_frame(frame_path):
    result = subprocess.run(
        ["tesseract", str(frame_path), "-", "--psm", "6"],
        check=True,
        capture_output=True,
        text=True,
    )
    return result.stdout.strip()


def evidence_type(transcript, screen_text):
    if transcript and screen_text:
        return "both"
    if transcript:
        return "narration"
    if screen_text:
        return "screen"
    return "none"


def build_chunks(video_path):
    frames = extract_frames(video_path)
    transcript_segments = transcribe_video(video_path)

    for frame in frames:
        frame["screen_text"] = ocr_frame(frame["path"])

    duration_candidates = [segment["end"] for segment in transcript_segments]
    duration_candidates.extend(
        frame["timestamp_seconds"] + FRAME_INTERVAL_SECONDS for frame in frames
    )
    duration = max(duration_candidates, default=0.0)

    chunks = []
    chunk_start = 0.0

    while chunk_start < duration:
        chunk_end = chunk_start + CHUNK_SECONDS
        matching_segments = [
            segment
            for segment in transcript_segments
            if segment["end"] > chunk_start and segment["start"] < chunk_end
        ]
        matching_frames = [
            frame
            for frame in frames
            if chunk_start <= frame["timestamp_seconds"] < chunk_end
        ]

        transcript = " ".join(segment["text"] for segment in matching_segments).strip()
        screen_text = "\n\n".join(
            f"[{frame['timestamp']}] {frame['screen_text']}"
            for frame in matching_frames
            if frame["screen_text"]
        ).strip()

        if transcript or screen_text:
            midpoint = chunk_start + CHUNK_SECONDS / 2
            representative = min(
                matching_frames,
                key=lambda frame: abs(frame["timestamp_seconds"] - midpoint),
                default=None,
            )
            timestamp = (
                f"{format_timestamp(chunk_start)}-"
                f"{format_timestamp(min(chunk_end, duration))}"
            )
            chunks.append(
                {
                    "id": len(chunks),
                    "start": round(chunk_start, 3),
                    "end": round(min(chunk_end, duration), 3),
                    "timestamp": timestamp,
                    "transcript": transcript,
                    "screen_text": screen_text,
                    "evidence_type": evidence_type(transcript, screen_text),
                    "representative_frame": (
                        representative["path"] if representative else None
                    ),
                }
            )

        chunk_start = chunk_end

    if not chunks:
        raise RuntimeError("No searchable transcript or screen text was produced.")

    payload = {
        "source_video": str(video_path),
        "frame_interval_seconds": FRAME_INTERVAL_SECONDS,
        "chunk_seconds": CHUNK_SECONDS,
        "chunks": chunks,
    }
    INDEX_PATH.write_text(json.dumps(payload, indent=2), encoding="utf-8")
    return chunks


def load_chunks():
    if not INDEX_PATH.exists():
        raise FileNotFoundError(
            "video_index.json is missing. Run the index command first."
        )
    payload = json.loads(INDEX_PATH.read_text(encoding="utf-8"))
    return payload["chunks"]


def build_parser():
    parser = argparse.ArgumentParser(
        description="Build and query a local-first Video RAG index."
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    frames_parser = subparsers.add_parser("frames")
    frames_parser.add_argument("video")

    index_parser = subparsers.add_parser("index")
    index_parser.add_argument("video")
    return parser


def main():
    args = build_parser().parse_args()

    if args.command == "frames":
        frames = extract_frames(args.video)
        print(f"Saved {len(frames)} timestamped frames:")
        for frame in frames:
            print(f"  {frame['timestamp']}  {frame['path']}")
        return

    if args.command == "index":
        video_path = require_video(args.video)
        chunks = build_chunks(video_path)
        print(f"Saved {len(chunks)} chunks to {INDEX_PATH}")
        print("Sample chunk:")
        print(json.dumps(chunks[0], indent=2)[:1200])
        return


if __name__ == "__main__":
    try:
        main()
    except (FileNotFoundError, RuntimeError, subprocess.CalledProcessError) as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1)

The first indexing run can take several minutes while the transcription model downloads and processes the tutorial. OCR also runs once for every sampled screenshot.

Before you run the command, which evidence fields do you expect one 20-second chunk to contain?

• Build the local video index by running this command:

python video_rag.py index tutorial.mp4

What happens during indexing?

• The script samples fresh frames from tutorial.mp4.
• The local transcription model returns timestamped speech segments.
• Tesseract reads visible text from each JPEG.
• The script aligns both sources before writing video_index.json.

You should see a saved chunk count followed by a sample chunk. The sample includes timestamp, transcript, screen_text, evidence_type, and representative_frame.

• Switch to video_index.json in VS Code.
• Confirm that the top-level object contains source_video, frame_interval_seconds, chunk_seconds, and chunks.
• Open several chunk records to confirm that their timestamps advance in 20-second windows.

That is the evidence layer complete. Your narration and screen text now share one timestamped local index.

Index command stopped early?

Confirm that tutorial.mp4 remains inside the video-rag-agent folder. The command resolves that path before transcription begins.

If the first model download cannot finish, check your internet connection before running the command again. Later transcription runs reuse the downloaded model.

Help me debug the index command.

Your tutorial is now searchable evidence instead of disconnected media. Next, you will expose why literal phrase matching cannot reliably find paraphrased questions.

Expose Exact-Phrase Search Limits

Your local JSON index now holds timestamped narration plus OCR evidence. A retriever still needs a rule for deciding which chunk answers a question.

A basic retriever can search for the complete question inside each chunk after standardizing capitalization. The question in this step describes the same dictionary operation as the evidence with different wording.

In this step, get ready to:

• Build a complete-phrase retrieval baseline.
• Create a repeatable paraphrase demonstration.
• Run the demonstration to expose the missed match.

Build the literal retriever

An exact-phrase search looks for the complete query as one continuous sequence. Case folding makes capitalization consistent before that comparison.

• Return to video_rag.py in Visual Studio Code.
• Locate the end of load_chunks().
• Add this function directly below load_chunks():

def combined_text(chunk):
    return "\n".join(
        part
        for part in (chunk.get("transcript", ""), chunk.get("screen_text", ""))
        if part
    )

What does this function do?

• The chunk parameter represents one timestamped record from video_index.json.
• The function joins available narration with available screen text.
• The if part condition leaves empty evidence fields out of the searchable text.

• Save video_rag.py.
• Confirm that combined_text() appears directly below load_chunks().

Function in the wrong place?

Check that combined_text() starts at the left edge of the file. An indented definition would place it inside load_chunks().

Ask for help if the function structure looks different: help me place combined_text correctly in video_rag.py

The next function applies the strict comparison. It returns a chunk only when the complete normalized question appears inside that chunk's combined evidence.

• Add this function directly below combined_text():

def exact_phrase_search(question, chunks):
    phrase = question.casefold().strip()
    return [
        chunk
        for chunk in chunks
        if phrase and phrase in combined_text(chunk).casefold()
    ]

How does the comparison work?

• The casefold() calls normalize capitalization in both strings.
• The strip() call removes whitespace from the ends of the question.
• The membership check requires the complete question to occur unchanged inside the evidence.
• The returned list contains every chunk that passes that strict test.

• Save video_rag.py.
• Confirm that exact_phrase_search() appears directly below combined_text().

Seeing an indentation warning?

Align phrase = question.casefold().strip() with the return statement. Keep the list-comprehension lines indented inside the return value.

Ask for a line-by-line indentation check: help me fix exact_phrase_search indentation

Create the paraphrase demonstration

A fixed demonstration keeps the comparison repeatable. Every run uses one evidence sentence plus one question about the same dictionary behavior.

• Add this function directly below exact_phrase_search():

def run_keyword_demo():
    question = "How does the function keep dictionary values?"
    chunks = [
        {
            "transcript": "The function stores values in a dictionary.",
            "screen_text": "",
        }
    ]
    matches = exact_phrase_search(question, chunks)
    print(f"Question: {question}")
    print("Exact-phrase result:", "Match" if matches else "No match")

What does the demonstration control?

• The question asks how the function keeps dictionary values.
• The transcript says that the function stores values in a dictionary.
• The empty screen_text value keeps the demonstration focused on one evidence sentence.
• The final conditional prints the outcome of the strict phrase comparison.

• Save video_rag.py.
• Locate the existing build_parser() function.
• Replace the complete build_parser() function with this version:

def build_parser():
    parser = argparse.ArgumentParser(
        description="Build and query a local-first Video RAG index."
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    frames_parser = subparsers.add_parser("frames")
    frames_parser.add_argument("video")

    index_parser = subparsers.add_parser("index")
    index_parser.add_argument("video")

    subparsers.add_parser("demo-keyword")
    return parser

How does the new command become available?

The new parser entry registers demo-keyword as a command that needs no extra arguments. The existing frame and index commands keep their video argument.

• Save video_rag.py.
• Check the parser's available commands by running:

python video_rag.py --help

What should you see?

You should see demo-keyword listed with frames and index in the generated command help. This confirms that the parser recognizes the new command.

Command missing from the help output?

Confirm that subparsers.add_parser("demo-keyword") sits before return parser. Save the file before running the help command again.

Ask for help comparing the parser definitions: help me register the demo-keyword command

Run the strict search test

The parser can now recognize the demonstration. The final branch connects that command to run_keyword_demo().

• Locate the existing main() function.
• Replace the complete main() function with this version:

def main():
    args = build_parser().parse_args()

    if args.command == "frames":
        frames = extract_frames(args.video)
        print(f"Saved {len(frames)} timestamped frames:")
        for frame in frames:
            print(f"  {frame['timestamp']}  {frame['path']}")
        return

    if args.command == "index":
        video_path = require_video(args.video)
        chunks = build_chunks(video_path)
        print(f"Saved {len(chunks)} chunks to {INDEX_PATH}")
        print("Sample chunk:")
        print(json.dumps(chunks[0], indent=2)[:1200])
        return

    if args.command == "demo-keyword":
        run_keyword_demo()
        return

How does the command run?

The new conditional checks whether demo-keyword was selected. It calls the deterministic demonstration before returning from main().

• Save video_rag.py.

✔️ Awesome, I've got everything!

Great. Confirm that video_rag.py is saved before you run the demonstration.

ⓧ I'd like to double check the full code

Compare your cumulative video_rag.py file with this version:

import argparse
import json
import subprocess
import sys
from pathlib import Path

import av
from faster_whisper import WhisperModel

BASE_DIR = Path(__file__).resolve().parent
FRAMES_DIR = BASE_DIR / "frames"
INDEX_PATH = BASE_DIR / "video_index.json"
FRAME_INTERVAL_SECONDS = 10.0
CHUNK_SECONDS = 20.0
TRANSCRIPTION_MODEL = "tiny.en"


def format_timestamp(seconds):
    total_seconds = max(0, int(seconds))
    minutes, secs = divmod(total_seconds, 60)
    hours, minutes = divmod(minutes, 60)
    if hours:
        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
    return f"{minutes:02d}:{secs:02d}"


def require_video(video_path):
    path = Path(video_path).expanduser().resolve()
    if not path.exists():
        raise FileNotFoundError(f"Video not found: {path}")
    return path


def extract_frames(video_path):
    video_path = require_video(video_path)
    FRAMES_DIR.mkdir(exist_ok=True)

    for old_frame in FRAMES_DIR.glob("frame_*.jpg"):
        old_frame.unlink()

    sampled_frames = []
    next_sample_time = 0.0

    with av.open(str(video_path)) as container:
        if not container.streams.video:
            raise RuntimeError("The input file has no video stream.")

        stream = container.streams.video[0]
        for frame in container.decode(stream):
            if frame.time is None:
                continue

            timestamp = float(frame.time)
            if timestamp + 0.001 < next_sample_time:
                continue

            frame_path = FRAMES_DIR / f"frame_{len(sampled_frames):04d}.jpg"
            frame.to_image().save(str(frame_path), quality=80)
            sampled_frames.append(
                {
                    "timestamp_seconds": timestamp,
                    "timestamp": format_timestamp(timestamp),
                    "path": str(frame_path),
                }
            )
            next_sample_time = timestamp + FRAME_INTERVAL_SECONDS

    if not sampled_frames:
        raise RuntimeError("No video frames could be decoded.")

    return sampled_frames


def transcribe_video(video_path):
    model = WhisperModel(
        TRANSCRIPTION_MODEL,
        device="cpu",
        compute_type="int8",
    )
    segments, _ = model.transcribe(str(video_path), beam_size=5)

    transcript = []
    for segment in segments:
        text = segment.text.strip()
        if text:
            transcript.append(
                {
                    "start": float(segment.start),
                    "end": float(segment.end),
                    "text": text,
                }
            )
    return transcript


def ocr_frame(frame_path):
    result = subprocess.run(
        ["tesseract", str(frame_path), "-", "--psm", "6"],
        check=True,
        capture_output=True,
        text=True,
    )
    return result.stdout.strip()


def evidence_type(transcript, screen_text):
    if transcript and screen_text:
        return "both"
    if transcript:
        return "narration"
    if screen_text:
        return "screen"
    return "none"


def build_chunks(video_path):
    frames = extract_frames(video_path)
    transcript_segments = transcribe_video(video_path)

    for frame in frames:
        frame["screen_text"] = ocr_frame(frame["path"])

    duration_candidates = [segment["end"] for segment in transcript_segments]
    duration_candidates.extend(
        frame["timestamp_seconds"] + FRAME_INTERVAL_SECONDS for frame in frames
    )
    duration = max(duration_candidates, default=0.0)

    chunks = []
    chunk_start = 0.0

    while chunk_start < duration:
        chunk_end = chunk_start + CHUNK_SECONDS
        matching_segments = [
            segment
            for segment in transcript_segments
            if segment["end"] > chunk_start and segment["start"] < chunk_end
        ]
        matching_frames = [
            frame
            for frame in frames
            if chunk_start <= frame["timestamp_seconds"] < chunk_end
        ]

        transcript = " ".join(segment["text"] for segment in matching_segments).strip()
        screen_text = "\n\n".join(
            f"[{frame['timestamp']}] {frame['screen_text']}"
            for frame in matching_frames
            if frame["screen_text"]
        ).strip()

        if transcript or screen_text:
            midpoint = chunk_start + CHUNK_SECONDS / 2
            representative = min(
                matching_frames,
                key=lambda frame: abs(frame["timestamp_seconds"] - midpoint),
                default=None,
            )
            timestamp = (
                f"{format_timestamp(chunk_start)}-"
                f"{format_timestamp(min(chunk_end, duration))}"
            )
            chunks.append(
                {
                    "id": len(chunks),
                    "start": round(chunk_start, 3),
                    "end": round(min(chunk_end, duration), 3),
                    "timestamp": timestamp,
                    "transcript": transcript,
                    "screen_text": screen_text,
                    "evidence_type": evidence_type(transcript, screen_text),
                    "representative_frame": (
                        representative["path"] if representative else None
                    ),
                }
            )

        chunk_start = chunk_end

    if not chunks:
        raise RuntimeError("No searchable transcript or screen text was produced.")

    payload = {
        "source_video": str(video_path),
        "frame_interval_seconds": FRAME_INTERVAL_SECONDS,
        "chunk_seconds": CHUNK_SECONDS,
        "chunks": chunks,
    }
    INDEX_PATH.write_text(json.dumps(payload, indent=2), encoding="utf-8")
    return chunks


def load_chunks():
    if not INDEX_PATH.exists():
        raise FileNotFoundError(
            "video_index.json is missing. Run the index command first."
        )
    payload = json.loads(INDEX_PATH.read_text(encoding="utf-8"))
    return payload["chunks"]


def combined_text(chunk):
    return "\n".join(
        part
        for part in (chunk.get("transcript", ""), chunk.get("screen_text", ""))
        if part
    )


def exact_phrase_search(question, chunks):
    phrase = question.casefold().strip()
    return [
        chunk
        for chunk in chunks
        if phrase and phrase in combined_text(chunk).casefold()
    ]


def run_keyword_demo():
    question = "How does the function keep dictionary values?"
    chunks = [
        {
            "transcript": "The function stores values in a dictionary.",
            "screen_text": "",
        }
    ]
    matches = exact_phrase_search(question, chunks)
    print(f"Question: {question}")
    print("Exact-phrase result:", "Match" if matches else "No match")


def build_parser():
    parser = argparse.ArgumentParser(
        description="Build and query a local-first Video RAG index."
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    frames_parser = subparsers.add_parser("frames")
    frames_parser.add_argument("video")

    index_parser = subparsers.add_parser("index")
    index_parser.add_argument("video")

    subparsers.add_parser("demo-keyword")
    return parser


def main():
    args = build_parser().parse_args()

    if args.command == "frames":
        frames = extract_frames(args.video)
        print(f"Saved {len(frames)} timestamped frames:")
        for frame in frames:
            print(f"  {frame['timestamp']}  {frame['path']}")
        return

    if args.command == "index":
        video_path = require_video(args.video)
        chunks = build_chunks(video_path)
        print(f"Saved {len(chunks)} chunks to {INDEX_PATH}")
        print("Sample chunk:")
        print(json.dumps(chunks[0], indent=2)[:1200])
        return

    if args.command == "demo-keyword":
        run_keyword_demo()
        return


if __name__ == "__main__":
    try:
        main()
    except (FileNotFoundError, RuntimeError, subprocess.CalledProcessError) as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1)

Your file should end with the existing error handler directly below main().

Before you run the demonstration, do you think lowercasing will bridge the wording difference between keep and stores?

• Run the exact-phrase demonstration with this command:

python video_rag.py demo-keyword

What should you see?

You should see the paraphrased question followed by Exact-phrase result: No match.

The question uses keep. The evidence uses stores.

Lowercasing standardizes capitalization. Literal matching still cannot connect equivalent meaning expressed with different words.

You found the intended limitation. The retriever misses relevant evidence even though both sentences describe the same dictionary operation.

Demo command not running?

Confirm that your activated terminal is still inside the video-rag-agent folder. Check that run_keyword_demo() appears above build_parser().

If the command produces no result, confirm that main() contains the demo-keyword branch.

Ask for help tracing the command path: help me debug the demo-keyword command

Your literal retrieval baseline has exposed its blind spot. Next, Chroma will compare meaning through local embeddings before the agent answers from retrieved evidence.

Retrieve Evidence and Generate Answers

Your exact-phrase demo proved that literal matching can miss evidence with the same meaning. Now you can replace that limitation with local semantic retrieval.

In this step, Chroma stores searchable embeddings for each video chunk. Gemini receives only the retrieved text plus selected frames for a grounded answer.

In this step, get ready to:

• Prove that semantic retrieval finds evidence missed by exact-phrase search.
• Persist the video chunks and their metadata in a local Chroma collection.
• Retrieve timestamped evidence before asking Gemini to answer.

Prove semantic retrieval works

Semantic retrieval compares the meaning represented by dense vectors. This allows a paraphrased question to find relevant evidence even when the wording differs.

• In video_rag.py, find run_keyword_demo().
• Add run_vector_demo() directly below it by pasting this code:

def run_vector_demo():
    question = "How does the function keep dictionary values?"
    client = chromadb.Client()
    collection = client.get_or_create_collection(name="retrieval_demo")
    collection.upsert(
        ids=["relevant", "unrelated"],
        documents=[
            "The function stores values in a dictionary.",
            "The loop prints each result to the terminal.",
        ],
    )
    results = collection.query(query_texts=[question], n_results=1)
    print(f"Question: {question}")
    print("Semantic result:", results["documents"][0][0])

What does this demo prove?

• The temporary collection embeds two evidence sentences in memory.
• The query uses the same paraphrased question that defeated exact-phrase matching.
• The nearest vector identifies the dictionary sentence by meaning.

• In build_parser(), find the line that adds demo-keyword.
• Add the new command directly below that line:

    subparsers.add_parser("demo-vector")

Why add a subcommand?

This line makes demo-vector a valid command in the argument parser. The parser can now route that command into the semantic demonstration.

• In main(), find the demo-keyword command branch.
• Add the vector demonstration branch directly below it:

    if args.command == "demo-vector":
        run_vector_demo()
        return

How is the demo dispatched?

The branch calls run_vector_demo() when the parsed command is demo-vector. The immediate return prevents another command branch from running.

• Save video_rag.py.

Before you run the demo, which sentence do you think semantic retrieval will return?

• Run the semantic comparison with this command:

python video_rag.py demo-vector

What should you see?

You should see the paraphrased question followed by Semantic result: The function stores values in a dictionary. The vector search has recovered the evidence that literal matching missed.

Semantic demo not running?

Check that run_vector_demo() sits outside run_keyword_demo() with no leading indentation. Confirm that both new command lines use demo-vector exactly.

If the command still fails, help me debug the semantic retrieval demo.

Persist the video evidence in Chroma

The in-memory demonstration disappears when the process ends. A persistent vector database keeps the video embeddings plus their timestamps and evidence labels inside chroma_db.

• In video_rag.py, replace the existing constants section with this complete constants section:

BASE_DIR = Path(__file__).resolve().parent
FRAMES_DIR = BASE_DIR / "frames"
INDEX_PATH = BASE_DIR / "video_index.json"
CHROMA_PATH = BASE_DIR / "chroma_db"
COLLECTION_NAME = "video_tutorial"
FRAME_INTERVAL_SECONDS = 10.0
CHUNK_SECONDS = 20.0
TRANSCRIPTION_MODEL = "tiny.en"
GENERATION_MODEL = "gemini-3.8-flash"
TOP_K = 3

What do these settings control?

• CHROMA_PATH points to the on-disk vector database.
• COLLECTION_NAME gives the tutorial records a stable collection name.
• TOP_K limits each question to the three closest evidence chunks.
• GENERATION_MODEL identifies the model used for final answer generation.

• Find exact_phrase_search() in video_rag.py.
• Add open_collection() below it by pasting this code:

def open_collection():
    client = chromadb.PersistentClient(path=str(CHROMA_PATH))
    collection = client.get_or_create_collection(name=COLLECTION_NAME)
    return client, collection

How does persistence work?

PersistentClient stores collection data at CHROMA_PATH. Calling get_or_create_collection() reuses video_tutorial after the script restarts.

• Add index_chunks() directly below open_collection() by pasting this code:

def index_chunks(chunks):
    client, collection = open_collection()
    existing = collection.get()
    if existing["ids"]:
        collection.delete(ids=existing["ids"])

    collection.upsert(
        ids=[f"chunk-{chunk['id']}" for chunk in chunks],
        documents=[combined_text(chunk) for chunk in chunks],
        metadatas=[
            {
                "chunk_id": chunk["id"],
                "timestamp": chunk["timestamp"],
                "evidence_type": chunk["evidence_type"],
                "representative_frame": chunk["representative_frame"] or "",
            }
            for chunk in chunks
        ],
    )
    return client, collection

What gets stored?

• Existing record IDs are deleted so a rebuilt index cannot leave stale chunks behind.
• Each record receives an ID based on its chunk number.
• The searchable document combines narration with OCR text.
• The metadata preserves the chunk ID, timestamp, evidence type, and representative frame path.

• In build_parser(), add the persistent indexing command below demo-vector.

    subparsers.add_parser("embed")

What does this parser line enable?

The parser now accepts embed as a command. This gives the indexing operation its own terminal entry point.

• In main(), add the embedding branch below the demo-vector branch:

    if args.command == "embed":
        chunks = load_chunks()
        _, collection = index_chunks(chunks)
        print(f"Stored {collection.count()} embedded chunks in {CHROMA_PATH}")
        return

How does the embed command work?

The branch loads the chunks from video_index.json before passing them to index_chunks(). It prints the stored record count after the upsert completes.

• Save video_rag.py.

The first embedding run can take a few minutes while Chroma downloads its local embedding model. Later runs reuse the cached model.

• Build the persistent collection by running:

python video_rag.py embed

What should you see?

You should see a stored chunk count followed by the local chroma_db path. That message confirms that Chroma embedded the combined evidence and persisted the collection.

Embedding did not finish?

Confirm that video_index.json remains inside video-rag-agent. Keep the internet connection active during the first local model download.

If the collection still fails to build, help me troubleshoot the Chroma embedding step.

Retrieve evidence and ask Gemini

Retrieval-augmented generation searches the local index before generation begins. The answer request contains only the strongest timestamped chunks plus their unique representative frames.

• In video_rag.py, replace the imports at the top of the file with this complete import section:

import argparse
import base64
import json
import os
import subprocess
import sys
from pathlib import Path

import av
import chromadb
from faster_whisper import WhisperModel
from google import genai

Why are the new imports needed?

• base64 converts selected JPEG bytes into inline image data.
• os checks whether the API key is available in the shell environment.
• genai creates the Gemini client used for the grounded request.

• Add vector_search() directly below index_chunks() by pasting this code:

def vector_search(question, chunks, top_k=TOP_K):
    client, collection = open_collection()
    record_count = collection.count()
    if record_count == 0:
        raise FileNotFoundError(
            "The Chroma collection is empty. Run the embed command first."
        )

    results = collection.query(
        query_texts=[question],
        n_results=min(top_k, record_count),
    )
    metadatas = results["metadatas"][0] if results["metadatas"] else []
    distances = results["distances"][0] if results["distances"] else []
    chunks_by_id = {chunk["id"]: chunk for chunk in chunks}

    retrieved = []
    for metadata, distance in zip(metadatas, distances):
        if metadata is None or distance is None:
            continue
        chunk = chunks_by_id.get(int(metadata["chunk_id"]))
        if chunk is not None:
            retrieved.append((float(distance), chunk))
    return retrieved

How does vector retrieval work?

• The collection embeds the question with the same local embedding function used for the documents.
• The query returns up to TOP_K records with their metadata and distances.
• The metadata reconnects each result to its complete chunk from video_index.json.
• The returned tuples preserve each distance alongside its evidence chunk.

• Add require_api_key() directly below vector_search() by pasting this code:

def require_api_key():
    if not (os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY")):
        raise RuntimeError(
            "Set GEMINI_API_KEY in your shell before using the ask command."
        )

Why check the environment first?

The check stops the request before client creation when no supported API key variable is available. Your key remains in the shell environment instead of entering video_rag.py.

• Add the first part of answer_question() below require_api_key() by pasting this code:

def answer_question(client, question, retrieved):
    evidence_blocks = []
    for rank, (_, chunk) in enumerate(retrieved, start=1):
        evidence_blocks.append(
            "\n".join(
                [
                    f"Evidence {rank}",
                    f"Timestamp: {chunk['timestamp']}",
                    f"Evidence source: {chunk['evidence_type']}",
                    f"Narration: {chunk['transcript'] or '[none]'}",
                    f"Screen OCR: {chunk['screen_text'] or '[none]'}",
                ]
            )
        )

How is the evidence organized?

Each retrieved chunk becomes a numbered evidence block. The block identifies its timestamp and source before presenting narration plus OCR text.

• Continue answer_question() directly below the evidence loop with this prompt:

    prompt = f"""
You answer questions about one coding tutorial.

Rules:
1. Use only the evidence below and the attached video frames.
2. If the evidence does not support the answer, reply exactly:
   Insufficient evidence in the indexed video.
3. When supported, provide these sections: Answer, Code, Concept, Sources.
4. Include code only when it is visible in a frame or explicitly present in the evidence.
5. Do not invent missing code lines.
6. In Sources, cite timestamps and state whether each claim came from narration,
   the screen, or both.

Question:
{question}

Retrieved evidence:
{chr(10).join(evidence_blocks)}
""".strip()

How does the prompt enforce grounding?

The rules restrict the answer to retrieved evidence and attached frames. They also define a precise fallback when the indexed video cannot support the answer.

Supported answers separate the explanation into Answer, Code, Concept, and Sources sections. The Sources section must identify timestamps plus evidence types.

• Continue answer_question() below the prompt with the multimodal input assembly:

    interaction_input = [{"type": "text", "text": prompt}]
    attached_paths = set()

    for _, chunk in retrieved:
        frame_value = chunk.get("representative_frame")
        if not frame_value:
            continue

        frame_path = Path(frame_value)
        if not frame_path.exists() or frame_path in attached_paths:
            continue

        interaction_input.append(
            {
                "type": "text",
                "text": f"Representative frame for {chunk['timestamp']}:",
            }
        )
        interaction_input.append(
            {
                "type": "image",
                "data": base64.b64encode(frame_path.read_bytes()).decode("utf-8"),
                "mime_type": "image/jpeg",
            }
        )
        attached_paths.add(frame_path)

Which frames leave your Mac?

Only representative frames attached to retrieved chunks enter the request. The attached_paths set prevents the same JPEG from being uploaded twice.

The raw video and the complete frame collection stay local. Retrieval controls the evidence boundary.

• Finish answer_question() with the Gemini interaction and returned text:

    interaction = client.interactions.create(
        model=GENERATION_MODEL,
        input=interaction_input,
    )
    return interaction.output_text

What completes the answer request?

The interaction sends the prepared text and selected images to GENERATION_MODEL. The function returns only the generated response text to the command handler.

• In build_parser(), add the question command after the embed parser line:

    ask_parser = subparsers.add_parser("ask")
    ask_parser.add_argument("question")

How does the question reach the script?

The ask subcommand requires one positional question. The parser stores that value on args.question for retrieval and generation.

• In main(), add the question branch below the embed branch:

    if args.command == "ask":
        require_api_key()
        chunks = load_chunks()
        retrieved = vector_search(args.question, chunks)

        if not retrieved:
            print("Insufficient evidence in the indexed video.")
            return

        print("Retrieved evidence, lowest distance first:")
        for distance, chunk in retrieved:
            print(
                f"  {chunk['timestamp']} | {chunk['evidence_type']} | "
                f"distance={distance:.3f}"
            )

        with genai.Client() as client:
            answer = answer_question(client, args.question, retrieved)

        print("\nAnswer:\n")
        print(answer)

What happens when you ask a question?

• The branch verifies the API key before loading local chunks.
• Vector search returns the closest evidence in lowest-distance order.
• The terminal prints each timestamp and evidence label before generation.
• The Gemini client receives the retrieved evidence before printing the grounded answer.

• At the bottom of video_rag.py, find the current direct call to main().
• Replace that final block with this error-handling block:

if __name__ == "__main__":
    try:
        main()
    except (FileNotFoundError, RuntimeError, subprocess.CalledProcessError) as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1)

Why handle these errors here?

The final block converts missing files, local runtime failures, and OCR command failures into concise terminal errors. It exits with a failure status instead of printing a full traceback for expected project problems.

✔️ Awesome, I've got everything!

Save video_rag.py. Your local retrieval and grounded generation pipeline is ready to test.

ⓧ I'd like to double check the full code

Compare your complete video_rag.py file with this reference:

import argparse
import base64
import json
import os
import subprocess
import sys
from pathlib import Path

import av
import chromadb
from faster_whisper import WhisperModel
from google import genai

BASE_DIR = Path(__file__).resolve().parent
FRAMES_DIR = BASE_DIR / "frames"
INDEX_PATH = BASE_DIR / "video_index.json"
CHROMA_PATH = BASE_DIR / "chroma_db"
COLLECTION_NAME = "video_tutorial"
FRAME_INTERVAL_SECONDS = 10.0
CHUNK_SECONDS = 20.0
TRANSCRIPTION_MODEL = "tiny.en"
GENERATION_MODEL = "gemini-3.8-flash"
TOP_K = 3


def format_timestamp(seconds):
    total_seconds = max(0, int(seconds))
    minutes, secs = divmod(total_seconds, 60)
    hours, minutes = divmod(minutes, 60)
    if hours:
        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
    return f"{minutes:02d}:{secs:02d}"


def require_video(video_path):
    path = Path(video_path).expanduser().resolve()
    if not path.exists():
        raise FileNotFoundError(f"Video not found: {path}")
    return path


def extract_frames(video_path):
    video_path = require_video(video_path)
    FRAMES_DIR.mkdir(exist_ok=True)

    for old_frame in FRAMES_DIR.glob("frame_*.jpg"):
        old_frame.unlink()

    sampled_frames = []
    next_sample_time = 0.0

    with av.open(str(video_path)) as container:
        if not container.streams.video:
            raise RuntimeError("The input file has no video stream.")

        stream = container.streams.video[0]
        for frame in container.decode(stream):
            if frame.time is None:
                continue

            timestamp = float(frame.time)
            if timestamp + 0.001 < next_sample_time:
                continue

            frame_path = FRAMES_DIR / f"frame_{len(sampled_frames):04d}.jpg"
            frame.to_image().save(str(frame_path), quality=80)
            sampled_frames.append(
                {
                    "timestamp_seconds": timestamp,
                    "timestamp": format_timestamp(timestamp),
                    "path": str(frame_path),
                }
            )
            next_sample_time = timestamp + FRAME_INTERVAL_SECONDS

    if not sampled_frames:
        raise RuntimeError("No video frames could be decoded.")

    return sampled_frames


def transcribe_video(video_path):
    model = WhisperModel(
        TRANSCRIPTION_MODEL,
        device="cpu",
        compute_type="int8",
    )
    segments, _ = model.transcribe(str(video_path), beam_size=5)

    transcript = []
    for segment in segments:
        text = segment.text.strip()
        if text:
            transcript.append(
                {
                    "start": float(segment.start),
                    "end": float(segment.end),
                    "text": text,
                }
            )
    return transcript


def ocr_frame(frame_path):
    result = subprocess.run(
        ["tesseract", str(frame_path), "-", "--psm", "6"],
        check=True,
        capture_output=True,
        text=True,
    )
    return result.stdout.strip()


def evidence_type(transcript, screen_text):
    if transcript and screen_text:
        return "both"
    if transcript:
        return "narration"
    if screen_text:
        return "screen"
    return "none"


def build_chunks(video_path):
    frames = extract_frames(video_path)
    transcript_segments = transcribe_video(video_path)

    for frame in frames:
        frame["screen_text"] = ocr_frame(frame["path"])

    duration_candidates = [segment["end"] for segment in transcript_segments]
    duration_candidates.extend(
        frame["timestamp_seconds"] + FRAME_INTERVAL_SECONDS for frame in frames
    )
    duration = max(duration_candidates, default=0.0)

    chunks = []
    chunk_start = 0.0

    while chunk_start < duration:
        chunk_end = chunk_start + CHUNK_SECONDS
        matching_segments = [
            segment
            for segment in transcript_segments
            if segment["end"] > chunk_start and segment["start"] < chunk_end
        ]
        matching_frames = [
            frame
            for frame in frames
            if chunk_start <= frame["timestamp_seconds"] < chunk_end
        ]

        transcript = " ".join(segment["text"] for segment in matching_segments).strip()
        screen_text = "\n\n".join(
            f"[{frame['timestamp']}] {frame['screen_text']}"
            for frame in matching_frames
            if frame["screen_text"]
        ).strip()

        if transcript or screen_text:
            midpoint = chunk_start + CHUNK_SECONDS / 2
            representative = min(
                matching_frames,
                key=lambda frame: abs(frame["timestamp_seconds"] - midpoint),
                default=None,
            )
            timestamp = (
                f"{format_timestamp(chunk_start)}-"
                f"{format_timestamp(min(chunk_end, duration))}"
            )
            chunks.append(
                {
                    "id": len(chunks),
                    "start": round(chunk_start, 3),
                    "end": round(min(chunk_end, duration), 3),
                    "timestamp": timestamp,
                    "transcript": transcript,
                    "screen_text": screen_text,
                    "evidence_type": evidence_type(transcript, screen_text),
                    "representative_frame": (
                        representative["path"] if representative else None
                    ),
                }
            )

        chunk_start = chunk_end

    if not chunks:
        raise RuntimeError("No searchable transcript or screen text was produced.")

    payload = {
        "source_video": str(video_path),
        "frame_interval_seconds": FRAME_INTERVAL_SECONDS,
        "chunk_seconds": CHUNK_SECONDS,
        "chunks": chunks,
    }
    INDEX_PATH.write_text(json.dumps(payload, indent=2), encoding="utf-8")
    return chunks


def load_chunks():
    if not INDEX_PATH.exists():
        raise FileNotFoundError(
            "video_index.json is missing. Run the index command first."
        )
    payload = json.loads(INDEX_PATH.read_text(encoding="utf-8"))
    return payload["chunks"]


def combined_text(chunk):
    return "\n".join(
        part
        for part in (chunk.get("transcript", ""), chunk.get("screen_text", ""))
        if part
    )


def exact_phrase_search(question, chunks):
    phrase = question.casefold().strip()
    return [
        chunk
        for chunk in chunks
        if phrase and phrase in combined_text(chunk).casefold()
    ]


def open_collection():
    client = chromadb.PersistentClient(path=str(CHROMA_PATH))
    collection = client.get_or_create_collection(name=COLLECTION_NAME)
    return client, collection


def index_chunks(chunks):
    client, collection = open_collection()
    existing = collection.get()
    if existing["ids"]:
        collection.delete(ids=existing["ids"])

    collection.upsert(
        ids=[f"chunk-{chunk['id']}" for chunk in chunks],
        documents=[combined_text(chunk) for chunk in chunks],
        metadatas=[
            {
                "chunk_id": chunk["id"],
                "timestamp": chunk["timestamp"],
                "evidence_type": chunk["evidence_type"],
                "representative_frame": chunk["representative_frame"] or "",
            }
            for chunk in chunks
        ],
    )
    return client, collection


def vector_search(question, chunks, top_k=TOP_K):
    client, collection = open_collection()
    record_count = collection.count()
    if record_count == 0:
        raise FileNotFoundError(
            "The Chroma collection is empty. Run the embed command first."
        )

    results = collection.query(
        query_texts=[question],
        n_results=min(top_k, record_count),
    )
    metadatas = results["metadatas"][0] if results["metadatas"] else []
    distances = results["distances"][0] if results["distances"] else []
    chunks_by_id = {chunk["id"]: chunk for chunk in chunks}

    retrieved = []
    for metadata, distance in zip(metadatas, distances):
        if metadata is None or distance is None:
            continue
        chunk = chunks_by_id.get(int(metadata["chunk_id"]))
        if chunk is not None:
            retrieved.append((float(distance), chunk))
    return retrieved


def require_api_key():
    if not (os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY")):
        raise RuntimeError(
            "Set GEMINI_API_KEY in your shell before using the ask command."
        )


def answer_question(client, question, retrieved):
    evidence_blocks = []
    for rank, (_, chunk) in enumerate(retrieved, start=1):
        evidence_blocks.append(
            "\n".join(
                [
                    f"Evidence {rank}",
                    f"Timestamp: {chunk['timestamp']}",
                    f"Evidence source: {chunk['evidence_type']}",
                    f"Narration: {chunk['transcript'] or '[none]'}",
                    f"Screen OCR: {chunk['screen_text'] or '[none]'}",
                ]
            )
        )

    prompt = f"""
You answer questions about one coding tutorial.

Rules:
1. Use only the evidence below and the attached video frames.
2. If the evidence does not support the answer, reply exactly:
   Insufficient evidence in the indexed video.
3. When supported, provide these sections: Answer, Code, Concept, Sources.
4. Include code only when it is visible in a frame or explicitly present in the evidence.
5. Do not invent missing code lines.
6. In Sources, cite timestamps and state whether each claim came from narration,
   the screen, or both.

Question:
{question}

Retrieved evidence:
{chr(10).join(evidence_blocks)}
""".strip()

    interaction_input = [{"type": "text", "text": prompt}]
    attached_paths = set()

    for _, chunk in retrieved:
        frame_value = chunk.get("representative_frame")
        if not frame_value:
            continue

        frame_path = Path(frame_value)
        if not frame_path.exists() or frame_path in attached_paths:
            continue

        interaction_input.append(
            {
                "type": "text",
                "text": f"Representative frame for {chunk['timestamp']}:",
            }
        )
        interaction_input.append(
            {
                "type": "image",
                "data": base64.b64encode(frame_path.read_bytes()).decode("utf-8"),
                "mime_type": "image/jpeg",
            }
        )
        attached_paths.add(frame_path)

    interaction = client.interactions.create(
        model=GENERATION_MODEL,
        input=interaction_input,
    )
    return interaction.output_text


def run_keyword_demo():
    question = "How does the function keep dictionary values?"
    chunks = [
        {
            "transcript": "The function stores values in a dictionary.",
            "screen_text": "",
        }
    ]
    matches = exact_phrase_search(question, chunks)
    print(f"Question: {question}")
    print("Exact-phrase result:", "Match" if matches else "No match")


def run_vector_demo():
    question = "How does the function keep dictionary values?"
    client = chromadb.Client()
    collection = client.get_or_create_collection(name="retrieval_demo")
    collection.upsert(
        ids=["relevant", "unrelated"],
        documents=[
            "The function stores values in a dictionary.",
            "The loop prints each result to the terminal.",
        ],
    )
    results = collection.query(query_texts=[question], n_results=1)
    print(f"Question: {question}")
    print("Semantic result:", results["documents"][0][0])


def build_parser():
    parser = argparse.ArgumentParser(
        description="Build and query a local-first Video RAG index."
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    frames_parser = subparsers.add_parser("frames")
    frames_parser.add_argument("video")

    index_parser = subparsers.add_parser("index")
    index_parser.add_argument("video")

    subparsers.add_parser("demo-keyword")
    subparsers.add_parser("demo-vector")
    subparsers.add_parser("embed")

    ask_parser = subparsers.add_parser("ask")
    ask_parser.add_argument("question")
    return parser


def main():
    args = build_parser().parse_args()

    if args.command == "frames":
        frames = extract_frames(args.video)
        print(f"Saved {len(frames)} timestamped frames:")
        for frame in frames:
            print(f"  {frame['timestamp']}  {frame['path']}")
        return

    if args.command == "index":
        video_path = require_video(args.video)
        chunks = build_chunks(video_path)
        print(f"Saved {len(chunks)} chunks to {INDEX_PATH}")
        print("Sample chunk:")
        print(json.dumps(chunks[0], indent=2)[:1200])
        return

    if args.command == "demo-keyword":
        run_keyword_demo()
        return

    if args.command == "demo-vector":
        run_vector_demo()
        return

    if args.command == "embed":
        chunks = load_chunks()
        _, collection = index_chunks(chunks)
        print(f"Stored {collection.count()} embedded chunks in {CHROMA_PATH}")
        return

    if args.command == "ask":
        require_api_key()
        chunks = load_chunks()
        retrieved = vector_search(args.question, chunks)

        if not retrieved:
            print("Insufficient evidence in the indexed video.")
            return

        print("Retrieved evidence, lowest distance first:")
        for distance, chunk in retrieved:
            print(
                f"  {chunk['timestamp']} | {chunk['evidence_type']} | "
                f"distance={distance:.3f}"
            )

        with genai.Client() as client:
            answer = answer_question(client, args.question, retrieved)

        print("\nAnswer:\n")
        print(answer)


if __name__ == "__main__":
    try:
        main()
    except (FileNotFoundError, RuntimeError, subprocess.CalledProcessError) as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1)

How should you use this reference?

Compare the imports, constants, function order, parser commands, command branches, and final error handling with your saved file. The complete file should match this reference exactly.

• Save video_rag.py.

Before you ask the question, which timestamps do you expect the local vector search to retrieve?

• Run the complete retrieval and answer pipeline with this command:

python video_rag.py ask "What code is being written and why?"

What should you see?

The terminal first lists retrieved timestamps in lowest-distance order. Each line includes a narration, screen, or both evidence label.

A supported response follows with Answer, Code, Concept, and Sources sections. Unsupported evidence produces Insufficient evidence in the indexed video..

Question command not working?

Confirm that your terminal still shows the activated .venv environment. Confirm that the earlier embedding command created the chroma_db directory.

If authentication fails, return to the shell where GEMINI_API_KEY is available. Avoid printing or sharing the key while troubleshooting.

For help with another failure, help me debug the grounded answer command.

You now have the complete Video RAG loop. Your agent searches local evidence before generating an answer grounded in timestamps, evidence labels, and selected frames.

Secret mission

Red-Team the Grounding

Test whether your Video RAG Agent stays faithful to its evidence across three different question types. You will inspect every retrieved timestamp before recording where the agent is trustworthy and where the next version needs stronger retrieval.

Clean Up Your Resources

Clean Up Your Resources

Keeping local files costs nothing because Gemini API requests only happen when you run the agent under its free-tier limits. Choose whether to keep the project ready, pause your shell, or delete its local resources.

Resources you used:

• The local video-rag-agent folder containing .venv, video_rag.py, requirements.txt, tutorial.mp4, frames, video_index.json, chroma_db, and evaluation.md.
• The Chroma embedding model cached at ~/.cache/chroma/onnx_models/all-MiniLM-L6-v2.
• The GEMINI_API_KEY export stored in ~/.zshrc.
• The authorization key created in Google AI Studio.

Keep everything running

No action is needed. Choose this if you want to ask more questions or index another tutorial.

• Keep the video-rag-agent folder to preserve your frames, searchable index, vector database, and evaluation.
• Keep GEMINI_API_KEY in ~/.zshrc to continue using grounded answer generation.
• Keep the cached all-MiniLM-L6-v2 model to avoid downloading it again during a future embedding run.

Pause - I'll come back to this later

Exit the virtual environment to reset your shell. Your project files remain ready for your next session.

• Exit the activated .venv by running:

deactivate

Your shell prompt stops showing .venv. The local project files remain unchanged.

• Leave the video-rag-agent folder on your Mac for later.
• Leave GEMINI_API_KEY in ~/.zshrc if you plan to return to the agent.

Chroma does not use a separate local server in this project. Your persistent collection stays inside chroma_db.

Delete - I don't want to use this again

Choose this option for a complete reset. Deleting the folder can feel drastic because every local artifact inside it disappears together.

• Copy tutorial.mp4 elsewhere if you want to keep the original video.
• Copy evaluation.md elsewhere if you want to keep your grounding assessment.
• Exit the activated .venv by running:

deactivate

Your shell leaves the project environment. This releases the active virtual environment before its files are removed.

• Move one level above video-rag-agent and delete the folder by running:

cd ..
rm -rf video-rag-agent

The video-rag-agent folder is removed from its parent directory. This also removes its virtual environment, frames, JSON index, persistent collection, and evaluation.

• Skip the cache deletion if another Chroma project uses the same embedding model.
• Remove the cached all-MiniLM-L6-v2 model by running:

rm -rf ~/.cache/chroma/onnx_models/all-MiniLM-L6-v2

The cached embedding model is removed from your home directory. A future Chroma project can download the model again when required.

• Open ~/.zshrc in a text editor.
• Find the line that starts with export GEMINI_API_KEY=.
• Delete the export line from ~/.zshrc.
• Save ~/.zshrc.
• Close the current Terminal window to clear the key from the active shell.
• Return to the API Keys page in Google AI Studio.
• Revoke the authorization key you created for this project.

Nice Work!

Nice Work!

You did it! You built a local-first Video RAG agent that turns a coding tutorial into searchable evidence. Your terminal now retrieves timestamped proof before generating an answer.

You've learned how to:

• Create a multimodal local index from real video timestamps. Your pipeline transcribes narration locally. It extracts screen text with OCR. It combines both sources into 20-second evidence chunks.
• Expose the weakness of exact-phrase search with a paraphrased question. Replace it with local dense embeddings in a persistent Chroma vector database.
• Generate grounded multimodal answers with Gemini from retrieved text plus selected frames. Display timestamp citations for the answer. Label each source as narration, screen, or both.
• Secret Mission: Red-team the agent with narration-led questions, screen-only questions, and unanswerable questions. Record each verdict in evaluation.md with supporting observations. Finish with an evidence-based recommendation for the next version.

Ready to quiz yourself?