Search without embeddings: Postgres tsvector, LLM rerank, and 30-second clips

user query
   │
   ▼
┌──────────────────────────────────────┐
│  FTS over transcript_segments_fts    │  ← top 50, ranked by ts_rank
│  (Postgres, GIN-indexed tsvector)    │
└──────────────────────────────────────┘
   │   top 50 candidates
   ▼
┌──────────────────────────────────────┐
│  LLM rerank (Haiku via llm_client)   │  ← top 10 + one-line justification
│  tier="cheap", schema=RerankResult   │
└──────────────────────────────────────┘
   │   top 10 segment ids
   ▼
┌──────────────────────────────────────┐
│  Clip extraction                     │  ← expand each hit into a 30–60s clip,
│  (window N segments, snap to .       │     snapped to sentence boundaries
│   sentence boundaries)               │
└──────────────────────────────────────┘
   │   ranked clips
   ▼
┌──────────────────────────────────────┐
│  Cache write + response              │  ← cache (query_hash, candidate_set)
│  with YouTube deep links             │     for 7 days
└──────────────────────────────────────┘

ALTER TABLE transcript_segments
  ADD COLUMN search_tsv tsvector
  GENERATED ALWAYS AS (to_tsvector('english', coalesce(text, ''))) STORED;
 
CREATE INDEX transcript_segments_search_tsv_idx
  ON transcript_segments
  USING GIN (search_tsv);
 
-- A view that joins what the rerank prompt needs in one shot.
CREATE OR REPLACE VIEW transcript_segments_search AS
SELECT
  s.id            AS segment_id,
  s.episode_id,
  e.video_id,
  e.title         AS episode_title,
  s.start_ms,
  s.end_ms,
  s.text,
  s.search_tsv
FROM transcript_segments s
JOIN episodes e ON e.id = s.episode_id;

from dataclasses import dataclass
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncConnection
 
 
@dataclass(frozen=True)
class Candidate:
    segment_id: int
    episode_id: int
    video_id: str
    episode_title: str
    start_ms: int
    end_ms: int
    text: str
    rank: float
 
 
_FTS_SQL = text(
    """
    SELECT
      segment_id, episode_id, video_id, episode_title,
      start_ms, end_ms, text,
      ts_rank_cd(search_tsv, q) AS rank
    FROM transcript_segments_search,
         websearch_to_tsquery('english', :query) AS q
    WHERE search_tsv @@ q
    ORDER BY rank DESC, start_ms ASC
    LIMIT :limit
    """
)
 
 
async def fts_candidates(
    conn: AsyncConnection, *, query: str, limit: int = 50
) -> list[Candidate]:
    rows = await conn.execute(_FTS_SQL, {"query": query, "limit": limit})
    return [Candidate(**dict(r._mapping)) for r in rows]

[fts] 50 candidates in 38ms
  1  (rank 0.412) ep_204 @ 00:14:31  "...the early fundraising landscape in Nepal was..."
  2  (rank 0.397) ep_188 @ 00:42:08  "...we raised our seed in Kathmandu, and..."
  3  (rank 0.355) ep_172 @ 01:03:55  "...angel investors here mostly come from..."
  ...
 50  (rank 0.041) ep_119 @ 00:08:17  "...fundraise next year if the runway..."

from pydantic import BaseModel, Field
from llm_client import complete
 
 
class RerankedHit(BaseModel):
    segment_id: int
    score: float = Field(ge=0, le=1, description="0–1, higher = more relevant")
    why: str = Field(max_length=140, description="One-line justification")
 
 
class RerankResult(BaseModel):
    hits: list[RerankedHit]
 
 
_SYSTEM = """\
You are a search reranker for a podcast transcript index.
 
You receive a user's query and 50 candidate transcript segments retrieved by
full-text search. Return the 10 most relevant segments, ranked.
 
Rules:
- Prefer segments where the speaker is directly discussing the query topic.
- Demote segments where the query terms appear in passing or as off-topic asides.
- A segment is more relevant if it contains a complete thought rather than a
  partial sentence.
- Score 1.0 = a strong direct match. Score 0.5 = related but not central.
  Score below 0.3 should not appear in the output.
 
Return JSON matching the provided schema. Do not invent segment_ids.
"""
 
 
async def rerank(*, query: str, candidates: list[Candidate]) -> RerankResult:
    user = "QUERY:\n" + query + "\n\nCANDIDATES:\n" + "\n".join(
        f"[{c.segment_id}] {c.text}" for c in candidates
    )
    return await complete(
        system=_SYSTEM,
        messages=[{"role": "user", "content": user}],
        schema=RerankResult,
        tier="cheap",
        cache_system=True,
    )

{
  "hits": [
    { "segment_id": 81204, "score": 0.94, "why": "Founder describes their own Nepal fundraising round in detail." },
    { "segment_id": 79018, "score": 0.88, "why": "Guest contrasts Nepali angel scene with India in 2024." },
    { "segment_id": 76553, "score": 0.71, "why": "Quick mention of fundraising regulation changes in Nepal." },
    ...
  ]
}

from dataclasses import dataclass
 
MIN_CLIP_MS = 30_000
MAX_CLIP_MS = 60_000
SENTENCE_END = ".!?"
 
 
@dataclass(frozen=True)
class Clip:
    episode_id: int
    video_id: str
    episode_title: str
    start_ms: int
    end_ms: int
    text: str
    hit_segment_id: int
 
 
async def build_clip(conn, *, hit: Candidate) -> Clip:
    # Pull a generous window of neighbors, then trim to the target span.
    rows = (
        await conn.execute(
            text(
                """
                SELECT id, start_ms, end_ms, text
                FROM transcript_segments
                WHERE episode_id = :episode_id
                  AND start_ms BETWEEN :lo AND :hi
                ORDER BY start_ms
                """
            ),
            {
                "episode_id": hit.episode_id,
                "lo": max(0, hit.start_ms - MAX_CLIP_MS),
                "hi": hit.end_ms + MAX_CLIP_MS,
            },
        )
    ).all()
 
    # Center the window on the hit, then expand outward until we hit
    # min duration OR a sentence boundary, capped at max duration.
    center = next(i for i, r in enumerate(rows) if r.id == hit.segment_id)
    lo = hi = center
    while True:
        span = rows[hi].end_ms - rows[lo].start_ms
        boundary_ok = rows[lo].text.lstrip()[:1].isupper() or lo == 0
        boundary_ok &= rows[hi].text.rstrip()[-1:] in SENTENCE_END or hi == len(rows) - 1
        if span >= MIN_CLIP_MS and boundary_ok:
            break
        if span >= MAX_CLIP_MS:
            break
        if lo > 0:
            lo -= 1
        if hi < len(rows) - 1:
            hi += 1
        if lo == 0 and hi == len(rows) - 1:
            break
 
    return Clip(
        episode_id=hit.episode_id,
        video_id=hit.video_id,
        episode_title=hit.episode_title,
        start_ms=rows[lo].start_ms,
        end_ms=rows[hi].end_ms,
        text=" ".join(r.text for r in rows[lo : hi + 1]),
        hit_segment_id=hit.segment_id,
    )

def youtube_deep_link(video_id: str, start_ms: int) -> str:
    return f"https://youtu.be/{video_id}?t={start_ms // 1000}"

class SearchHit(BaseModel):
    video_id: str
    episode_title: str
    start_ms: int
    end_ms: int
    text: str
    youtube_url: str
    score: float
    why: str
 
 
class SearchResponse(BaseModel):
    query: str
    hits: list[SearchHit]
    cached: bool
 
 
@router.get("/search", response_model=SearchResponse)
async def search(q: str, conn: AsyncConnection = Depends(get_conn)) -> SearchResponse:
    cached = await search_cache_get(conn, query=q)
    if cached is not None:
        return SearchResponse(query=q, hits=cached, cached=True)
 
    candidates = await fts_candidates(conn, query=q, limit=50)
    if not candidates:
        return SearchResponse(query=q, hits=[], cached=False)
 
    reranked = await rerank(query=q, candidates=candidates)
    by_id = {c.segment_id: c for c in candidates}
    clips = [
        await build_clip(conn, hit=by_id[h.segment_id])
        for h in reranked.hits
        if h.segment_id in by_id
    ]
    hits = [
        SearchHit(
            video_id=c.video_id,
            episode_title=c.episode_title,
            start_ms=c.start_ms,
            end_ms=c.end_ms,
            text=c.text,
            youtube_url=youtube_deep_link(c.video_id, c.start_ms),
            score=h.score,
            why=h.why,
        )
        for c, h in zip(clips, reranked.hits)
    ]
    await search_cache_put(conn, query=q, hits=hits)
    return SearchResponse(query=q, hits=hits, cached=False)

CREATE TABLE search_cache (
  cache_key       text PRIMARY KEY,
  query           text NOT NULL,
  candidate_hash  text NOT NULL,
  hits_json       jsonb NOT NULL,
  created_at      timestamptz NOT NULL DEFAULT now()
);
 
CREATE INDEX search_cache_created_at_idx ON search_cache (created_at);

import hashlib
import json
from datetime import timedelta
 
CACHE_TTL = timedelta(days=7)
 
 
def _key(query: str, candidate_ids: list[int]) -> tuple[str, str]:
    norm = " ".join(query.lower().split())
    candidate_hash = hashlib.sha256(
        ",".join(str(i) for i in sorted(candidate_ids)).encode()
    ).hexdigest()[:16]
    return hashlib.sha256(f"{norm}|{candidate_hash}".encode()).hexdigest(), candidate_hash

HTTP/1.1 200 OK
content-type: application/json
 
{
  "query": "fundraising in Nepal",
  "cached": false,
  "hits": [
    {
      "video_id": "abc123",
      "episode_title": "Ep 204 — Building Khalti, with Binay Mainali",
      "start_ms": 854000,
      "end_ms": 902000,
      "text": "So when we tried to raise our Series A in Nepal, the angel
               base was maybe twenty people. We ended up...",
      "youtube_url": "https://youtu.be/abc123?t=854",
      "score": 0.94,
      "why": "Founder describes their own Nepal fundraising round in detail."
    },
    ...
  ]
}