Build semantic search inside video: shot embeddings + transcript chunks, fused on the timeline

TL;DR Frame-only video search can't find "where she explains pricing." Transcript-only search can't find "the architecture diagram slide." We'll build both indexes, key them to the same timeline, and fuse at query time so results are timestamps, not video IDs. Shot detection, SigLIP-2 embeddings, Whisper chunks, pgvector. What we're building A search endpoint where "where does she explain the pricing model" returns video_id=42, t=247.3s and you can seek straight there. Two indexes over one timeline: | Index | Unit | Model | Finds | |---|---|---|---| | Visual | one shot | SigLIP-2 | "the architecture diagram", "person holding a coffee cup" | | Transcript | utterance group | text embedder | "the pricing explanation", "when they mention refunds" | Neither half works alone. In business video (talks, interviews, tutorials, meetings) most meaning is in the audio, so a frames-only index is blind to the majority of it. But nobody says "architecture diagram" out loud, they say "as you can see here." pip install torch transformers pyscenedetect faster-whisper psycopg[binary] pgvector 1. Chunk on shots, not on seconds 🎬 Sampling a frame every N seconds is wasteful in both directions: forty near-identical frames in a static screen share, and a missed shot across a fast cut. Use shot boundaries instead. # ingest/shots.py from scenedetect import detect, ContentDetector def shot_boundaries(path: str, threshold: float = 27.0) -> list[tuple[float, float]]: """Returns [(start_sec, end_sec), ...] per detected shot.""" scenes = detect(path, ContentDetector(threshold=threshold)) return [(s.get_seconds(), e.get_seconds()) for s, e in scenes] $ python -c "from ingest.shots import shot_boundaries; print(shot_boundaries('talk.mp4')[:4])" [(0.0, 8.34), (8.34, 31.07), (31.07, 33.9), (33.9, 96.21)] Grab one representative frame per shot, from the middle rather than the boundary (boundaries catch transitions and dissolves): # ingest/frames.py import subprocess, tempfile, os def midpoint_frame(video: str, start: float, end: float) -> str: t = start + (end - start) / 2 out = os.path.join(tempfile.mkdtemp(), "f.jpg") subprocess.run([ "ffmpeg", "-v", "error", "-y", "-ss", f"{t:.3f}", "-i", video, "-frames:v", "1", "-q:v", "3", out, ], check=True) return out 2. Embed the frames SigLIP-2 generally edges out CLIP at comparable size, especially on fine-grained visual detail. The SO400M variant emits 1152 dimensions (note this, it's about to matter). # ingest/visual.py import torch from PIL import Image from transformers import AutoModel, AutoProcessor MODEL_ID = "google/siglip2-so400m-patch16-512" EMBED_DIM = 1152 _device = "cuda" if torch.cuda.is_available() else "cpu" _model = AutoModel.from_pretrained(MODEL_ID).to(_device).eval() _proc = AutoProcessor.from_pretrained(MODEL_ID) @torch.inference_mode() def embed_image(path: str) -> list[float]: inputs = _proc(images=Image.open(path).convert("RGB"), return_tensors="pt").to(_device) v = _model.get_image_features(**inputs) return torch.nn.functional.normalize(v, dim=-1)[0].cpu().tolist() @torch.inference_mode() def embed_text(query: str) -> list[float]: inputs = _proc(text=[query], padding="max_length", return_tensors="pt").to(_device) v = _model.get_text_features(**inputs) return torch.nn.functional.normalize(v, dim=-1)[0].cpu().tolist() ⚠️ EMBED_DIM is not a config value you change later. Your table column isvector(1152) . Swapping to JinaCLIP-v2 (1024 dims) means re-embedding your entire library. Version your tables from day one so a model swap is a parallel build, not a migration. 3. Transcript chunks want different boundaries Speech doesn't respect cuts. A sentence runs across a shot change constantly, so chopping transcripts on visual boundaries produces fragments that embed badly. Chunk on the transcript's own logic, with overlap. # ingest/transcript.py from faster_whisper import WhisperModel _asr = WhisperModel("large-v3-turbo", device="auto", compute_type="int8") def chunks(video: str, target_chars: int = 400, overlap: int = 1) -> list[dict]: segments, _ = _asr.transcribe(video, vad_filter=True, word_timestamps=False) segs = [{"start": s.start, "end": s.end, "text": s.text.strip()} for s in segments] out, buf = [], [] for seg in segs: buf.append(seg) if sum(len(s["text"]) for s in buf) >= target_chars: out.append({ "start": buf[0]["start"], "end": buf[-1]["end"], "text": " ".join(s["text"] for s in buf), }) buf = buf[-overlap:] if overlap else [] if buf: out.append({ "start": buf[0]["start"], "end": buf[-1]["end"], "text": " ".join(s["text"] for s in buf), }) return out Two indexes, different granularities, both timestamped. That's the correct end state, and the timeline is the join key. 4. Schema Postgres + pgvector means one fewer system to run, and it lets you filter by tenancy in the same query as the similarity search. That last part is not a convenience, it's a correctness requirement (see step 6). -- migrations/001_init.sql (postgres 16 + pgvector 0.8) CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE shot_embeddings ( id bigserial PRIMARY KEY, video_id bigint NOT NULL, tenant_id bigint NOT NULL, start_sec double precision NOT NULL, end_sec double precision NOT NULL, embedding vector(1152) NOT NULL ); CREATE TABLE transcript_embeddings ( id bigserial PRIMARY KEY, video_id bigint NOT NULL, tenant_id bigint NOT NULL, start_sec double precision NOT NULL, end_sec double precision NOT NULL, text text NOT NULL, embedding vector(1152) NOT NULL ); CREATE INDEX ON shot_embeddings USING hnsw (embedding vector_cosine_ops); CREATE INDEX ON transcript_embeddings USING hnsw (embedding vector_cosine_ops); CREATE INDEX ON shot_embeddings (tenant_id, video_id); CREATE INDEX ON transcript_embeddings (tenant_id, video_id); 5. Fuse at query time, not index time It's tempting to combine visual and text vectors into one representation during ingest. Don't, at least not first. You lose explainability, you lose per-query weighting, and you've committed to a fusion strategy before you know what works on your content. Query both, merge on time overlap: # search/query.py from collections import defaultdict def search(conn, tenant_id: int, q: str, k: int = 30, w_visual: float = 0.4, w_text: float = 0.6): qv = embed_text(q) visual = _knn(conn, "shot_embeddings", tenant_id, qv, k) textual = _knn(conn, "transcript_embeddings", tenant_id, qv, k) # bucket both sides into 5s slots on the shared timeline buckets = defaultdict(lambda: {"visual": 0.0, "text": 0.0, "hit": None}) for r in visual: key = (r["video_id"], int(r["start_sec"] // 5)) b = buckets[key] b["visual"] = max(b["visual"], r["score"]) b["hit"] = b["hit"] or r for r in textual: key = (r["video_id"], int(r["start_sec"] // 5)) b = buckets[key] b["text"] = max(b["text"], r["score"]) b["hit"] = b["hit"] or r scored = [{ "video_id": vid, "t": b["hit"]["start_sec"], "score": w_visual * b["visual"] + w_text * b["text"], "why": {"visual": round(b["visual"], 3), "transcript": round(b["text"], 3)}, } for (vid, _slot), b in buckets.items()] return sorted(scored, key=lambda r: -r["score"])[:10] # search/query.py (continued) def _knn(conn, table: str, tenant_id: int, qv: list[float], k: int): with conn.cursor() as cur: cur.execute( f"""SELECT video_id, start_sec, 1 - (embedding %s::vector) AS score FROM {table} WHERE tenant_id = %s ORDER BY embedding %s::vector LIMIT %s""", (qv, tenant_id, qv, k), ) cols = [d.name for d in cur.description] return [dict(zip(cols, row)) for row in cur.fetchall()] A moment where both signals fire is a much better answer than either alone, and now you can say so: $ curl -s 'localhost:8000/search?q=where+she+explains+the+pricing+model' | jq '.[0]' { "video_id": 42, "t": 247.3, "score": 0.71, "why": { "visual": 0.31, "transcript": 0.98 } } w_visual / w_text is the knob that matters. Conference talks want transcript weighted heavily. A B-roll library wants the opposite. 6. The part that isn't in the architecture diagram 🔒 Embeddings do not carry permissions. A vector index will happily return the nearest neighbor from a video the user can't see. Filtering has to happen inside the query, not after it. Notice WHERE tenant_id = %s sits in the same statement as the ORDER BY embedding above. If you filter the results afterward, your top-10 becomes a top-4, your pagination lies, and eventually someone sees another customer's footage. ⚠️ This is the single most common way these systems ship broken. Post-filtering a KNN result is not access control. Errors you'll hit expected 1152 dimensions, not 768 : you swapped models without migrating the column. Told you. Shot detection returns one giant scene: screen recordings with no cuts. Fall back to fixed-interval sampling when len(shots) < 2 . Recall is fine, ranking is bad: your two score distributions aren't comparable. Normalize each side's scores within the result set before weighting them. Where the numbers come from One 2026 benchmark roundup puts semantic video search accuracy for this class of model in the mid-to-high 80s on general datasets, rising into the low 90s with domain-specific fine-tuning. That's someone else's data on someone else's content. Measure your own before you promise anything. On store choice: 2026 comparisons between pgvector and dedicated engines like Qdrant keep landing on "comparable speed, the bottleneck is elsewhere." If your data is already in Postgres, that's your answer. What's next - Clip generation for free. Once moments are addressable, "make me a highlight reel about X" is a query plus a concat, not a manual scrub. - Timestamped recommendations. Point at a moment instead of a video. - Rerank the top 20 with a cross-encoder or a VLM if precision at 1 matters more than latency. The reason to build this isn't that users are asking for a search box. It's that the index is the substrate for four other features. Top comments (0)