How to Build an AI Video Assistant That Watches Recordings and Answers Questions

ai-video-assistant/
├── app/
│   ├── __init__.py
│   ├── keyframe_selector.py
│   └── multimodal_aligner.py
├── tests/
│   └── test_video_assistant.py
└── requirements.txt

from typing import List
import numpy as np
from pydantic import BaseModel

class ExtractedKeyframe(BaseModel):
    frame_index: int
    timestamp_sec: float
    mean_diff_score: float

def select_scene_keyframes_from_luma(
    frame_luma_arrays: List[np.ndarray],
    fps: float,
    diff_threshold: float = 18.0,
    min_interval_sec: float = 1.5,
    max_interval_sec: float = 15.0,
) -> List[ExtractedKeyframe]:
    """Selects keyframes when visual scene change exceeds diff_threshold or max_interval_sec elapses."""
    if not frame_luma_arrays or fps <= 0:
        return []

    selected: List[ExtractedKeyframe] = [
        ExtractedKeyframe(frame_index=0, timestamp_sec=0.0, mean_diff_score=100.0)
    ]
    last_selected_frame = frame_luma_arrays[0].astype(np.float32)
    last_ts = 0.0

    for idx in range(1, len(frame_luma_arrays)):
        curr_ts = round(idx / fps, 2)
        elapsed = curr_ts - last_ts
        if elapsed < min_interval_sec:
            continue

        curr_arr = frame_luma_arrays[idx].astype(np.float32)
        mean_diff = float(np.mean(np.abs(curr_arr - last_selected_frame)))

        if mean_diff >= diff_threshold or elapsed >= max_interval_sec:
            selected.append(
                ExtractedKeyframe(
                    frame_index=idx,
                    timestamp_sec=curr_ts,
                    mean_diff_score=round(mean_diff, 2),
                )
            )
            last_selected_frame = curr_arr
            last_ts = curr_ts

    return selected

from typing import List
from pydantic import BaseModel

class TranscriptSegment(BaseModel):
    start_sec: float
    end_sec: float
    text: str

class VisualFrameCaption(BaseModel):
    timestamp_sec: float
    visual_description: str
    on_screen_ocr_text: str = ""

class MultimodalVideoChunk(BaseModel):
    video_id: str
    start_sec: float
    end_sec: float
    formatted_time_range: str
    combined_search_text: str

def format_mm_ss(seconds: float) -> str:
    mins = int(seconds // 60)
    secs = int(seconds % 60)
    return f"{mins:02d}:{secs:02d}"

def build_multimodal_video_chunks(
    video_id: str,
    transcripts: List[TranscriptSegment],
    visual_captions: List[VisualFrameCaption],
    window_sec: float = 30.0,
) -> List[MultimodalVideoChunk]:
    """Merges spoken transcript segments and visual frame descriptions into unified 30s retrieval windows."""
    max_time = max(
        max((t.end_sec for t in transcripts), default=0.0),
        max((v.timestamp_sec for v in visual_captions), default=0.0),
    )
    if max_time <= 0:
        return []

    chunks: List[MultimodalVideoChunk] = []
    curr_start = 0.0

    while curr_start < max_time:
        curr_end = curr_start + window_sec
        window_speech = [
            t.text.strip()
            for t in transcripts
            if t.end_sec > curr_start and t.start_sec < curr_end
        ]
        window_visuals = [
            f"[{format_mm_ss(v.timestamp_sec)} Screen: {v.visual_description} | OCR: {v.on_screen_ocr_text}]"
            for v in visual_captions
            if curr_start <= v.timestamp_sec < curr_end
        ]

        if window_speech or window_visuals:
            time_label = f"[{format_mm_ss(curr_start)} - {format_mm_ss(curr_end)}]"
            combined = (
                f"Video Timestamp {time_label}\n"
                f"Spoken Audio: {' '.join(window_speech) or 'No speech'}\n"
                f"Visual Scene: {' ; '.join(window_visuals) or 'Static scene'}"
            )
            chunks.append(
                MultimodalVideoChunk(
                    video_id=video_id,
                    start_sec=curr_start,
                    end_sec=curr_end,
                    formatted_time_range=time_label,
                    combined_search_text=combined,
                )
            )
        curr_start = curr_end

    return chunks

import numpy as np
from app.keyframe_selector import select_scene_keyframes_from_luma
from app.multimodal_aligner import (
    TranscriptSegment,
    VisualFrameCaption,
    build_multimodal_video_chunks,
)

def test_scene_keyframe_deduplication_and_transition_detection():
    # 10 frames at 1 FPS: frames 0..4 are dark (luma=10), frames 5..9 switch to bright slide (luma=200)
    frames = [np.full((16, 16), 10, dtype=np.uint8) for _ in range(5)] + [
        np.full((16, 16), 200, dtype=np.uint8) for _ in range(5)
    ]
    keyframes = select_scene_keyframes_from_luma(frames, fps=1.0, diff_threshold=25.0, min_interval_sec=1.0)
    assert len(keyframes) == 2
    assert keyframes[0].frame_index == 0
    assert keyframes[1].frame_index == 5

def test_multimodal_window_alignment():
    speech = [TranscriptSegment(start_sec=2.0, end_sec=12.0, text="Here we configure the PostgreSQL HNSW index.")]
    visuals = [
        VisualFrameCaption(
            timestamp_sec=5.0,
            visual_description="Terminal window showing psql prompt",
            on_screen_ocr_text="CREATE INDEX ON items USING hnsw (embedding vector_cosine_ops);",
        )
    ]
    chunks = build_multimodal_video_chunks("vid_01", speech, visuals, window_sec=30.0)
    assert len(chunks) == 1
    assert chunks[0].formatted_time_range == "[00:00 - 00:30]"
    assert "CREATE INDEX ON items USING hnsw" in chunks[0].combined_search_text