Search

transcript_indexer.search

Query helpers over the indexed corpus.

Pure functions over a sqlite3.Connection. Used by the CLI today and the MCP server later. Covers FTS5 metadata search, semantic KNN search over chunk embeddings, and a hybrid RRF merge of the two.

search_conversations(conn, *, since=None, until=None, participant=None, kind=None, title_query=None, limit=None)

List conversations matching the given metadata filters.

Source code in src/transcript_indexer/search.py

def search_conversations(
    conn: sqlite3.Connection,
    *,
    since: str | None = None,
    until: str | None = None,
    participant: str | None = None,
    kind: str | None = None,
    title_query: str | None = None,
    limit: int | None = None,
) -> list[ConversationSummary]:
    """List conversations matching the given metadata filters."""
    sql = [
        "SELECT c.id, c.source_path, c.kind, c.source, c.started_at, c.title FROM conversations c"
    ]
    where: list[str] = []
    params: list[object] = []

    if participant:
        sql.append(
            "JOIN participants p ON p.conversation_id = c.id "
            "LEFT JOIN people pe ON pe.id = p.person_id "
            "LEFT JOIN person_aliases pa ON pa.person_id = p.person_id"
        )
        where.append(
            "(p.speaker = ? COLLATE NOCASE "
            "OR pe.canonical = ? COLLATE NOCASE "
            "OR pa.alias = ? COLLATE NOCASE)"
        )
        params.extend([participant, participant, participant])

    if since:
        where.append("c.started_at >= ?")
        params.append(since)
    if until:
        where.append("c.started_at <= ?")
        params.append(until)
    if kind:
        where.append("c.kind = ?")
        params.append(kind)
    if title_query:
        where.append("c.title LIKE ?")
        params.append(f"%{title_query}%")

    if where:
        sql.append("WHERE " + " AND ".join(where))
    sql.append("GROUP BY c.id ORDER BY c.started_at DESC")
    if limit is not None:
        sql.append("LIMIT ?")
        params.append(int(limit))

    rows = conn.execute(" ".join(sql), params).fetchall()
    return [_row_to_summary(conn, r) for r in rows]

get_conversation(conn, conversation_id)

Fetch a single conversation with its turns.

Source code in src/transcript_indexer/search.py

def get_conversation(conn: sqlite3.Connection, conversation_id: int) -> ConversationDetail | None:
    """Fetch a single conversation with its turns."""
    row = conn.execute(
        "SELECT id, source_path, kind, source, started_at, title FROM conversations WHERE id = ?",
        (conversation_id,),
    ).fetchone()
    if row is None:
        return None
    summary = _row_to_summary(conn, row)
    turn_rows = conn.execute(
        "SELECT turn_idx, speaker, person_id, timestamp_sec, text FROM turns "
        "WHERE conversation_id = ? ORDER BY turn_idx",
        (conversation_id,),
    ).fetchall()
    turns = [
        TurnRow(
            idx=int(r["turn_idx"]),
            speaker=str(r["speaker"]),
            person_id=int(r["person_id"]) if r["person_id"] is not None else None,
            timestamp_seconds=int(r["timestamp_sec"]) if r["timestamp_sec"] is not None else None,
            text=str(r["text"]),
        )
        for r in turn_rows
    ]
    return ConversationDetail(summary=summary, turns=turns)

search_turns_fts(conn, query, *, speaker=None, since=None, until=None, limit=20)

Run an FTS5 query against turns and return ranked hits.

Source code in src/transcript_indexer/search.py

def search_turns_fts(
    conn: sqlite3.Connection,
    query: str,
    *,
    speaker: str | None = None,
    since: str | None = None,
    until: str | None = None,
    limit: int = 20,
) -> list[TurnHit]:
    """Run an FTS5 query against turns and return ranked hits."""
    fts_q = _escape_fts_query(query)
    if not fts_q:
        return []
    sql = [
        "SELECT t.id AS turn_id, t.conversation_id, c.title AS conversation_title, "
        "  c.started_at, t.turn_idx, t.speaker, t.text, "
        f"  snippet(turns_fts, 0, '{SNIPPET_MARK_OPEN}', '{SNIPPET_MARK_CLOSE}', '...', 16) "
        "AS snippet, "
        "  bm25(turns_fts) AS rank "
        "FROM turns_fts "
        "JOIN turns t ON t.id = turns_fts.rowid "
        "JOIN conversations c ON c.id = t.conversation_id "
        "WHERE turns_fts MATCH ?"
    ]
    params: list[object] = [fts_q]
    if speaker:
        sql.append("AND t.speaker = ? COLLATE NOCASE")
        params.append(speaker)
    if since:
        sql.append("AND c.started_at >= ?")
        params.append(since)
    if until:
        sql.append("AND c.started_at <= ?")
        params.append(until)
    sql.append("ORDER BY rank LIMIT ?")
    params.append(int(limit))
    rows = conn.execute(" ".join(sql), params).fetchall()
    return [
        TurnHit(
            turn_id=int(r["turn_id"]),
            conversation_id=int(r["conversation_id"]),
            conversation_title=str(r["conversation_title"]),
            conversation_started_at=datetime.fromisoformat(str(r["started_at"])),
            turn_idx=int(r["turn_idx"]),
            speaker=str(r["speaker"]),
            text=str(r["text"]),
            snippet=str(r["snippet"]),
            rank=float(r["rank"]),
        )
        for r in rows
    ]

search_semantic(conn, cfg, query, *, k=20, since=None, until=None, conversation_id=None)

KNN search against chunk embeddings, joined back to conversations.

Source code in src/transcript_indexer/search.py

def search_semantic(
    conn: sqlite3.Connection,
    cfg: Config,
    query: str,
    *,
    k: int = 20,
    since: str | None = None,
    until: str | None = None,
    conversation_id: int | None = None,
) -> list[SemanticHit]:
    """KNN search against chunk embeddings, joined back to conversations."""
    if not query.strip():
        return []
    query_blob = _embed_query(cfg, query)

    # vec0 only supports k= and MATCH on the embedding column; metadata filters
    # apply after the KNN. Over-fetch when filters are present so post-filter
    # still has hits.
    has_filters = bool(since or until or conversation_id is not None)
    fetch_k = max(k * 4, k) if has_filters else k

    sql = [
        "SELECT ce.rowid AS chunk_id, ce.distance AS distance, "
        "c.id AS c_id, c.conversation_id AS conv_id, c.start_turn_idx, c.end_turn_idx, "
        "c.text AS chunk_text, conv.title AS conv_title, conv.started_at AS conv_started "
        "FROM chunk_embeddings ce "
        "JOIN chunks c ON c.id = ce.rowid "
        "JOIN conversations conv ON conv.id = c.conversation_id "
        "WHERE ce.embedding MATCH ? AND ce.k = ?"
    ]
    params: list[object] = [query_blob, int(fetch_k)]
    if since:
        sql.append("AND conv.started_at >= ?")
        params.append(since)
    if until:
        sql.append("AND conv.started_at <= ?")
        params.append(until)
    if conversation_id is not None:
        sql.append("AND c.conversation_id = ?")
        params.append(int(conversation_id))
    sql.append("ORDER BY ce.distance LIMIT ?")
    params.append(int(fetch_k))

    rows = conn.execute(" ".join(sql), params).fetchall()
    rows = rows[:k]
    return [
        SemanticHit(
            chunk_id=int(r["chunk_id"]),
            conversation_id=int(r["conv_id"]),
            conversation_title=str(r["conv_title"]),
            conversation_started_at=datetime.fromisoformat(str(r["conv_started"])),
            start_turn_idx=(int(r["start_turn_idx"]) if r["start_turn_idx"] is not None else None),
            end_turn_idx=int(r["end_turn_idx"]) if r["end_turn_idx"] is not None else None,
            text=str(r["chunk_text"]),
            distance=float(r["distance"]),
        )
        for r in rows
    ]

search_hybrid(conn, cfg, query, *, k=20, speaker=None, since=None, until=None)

Hybrid FTS+semantic search merged with Reciprocal Rank Fusion.

Hits are keyed by conversation_id so an FTS turn match and a semantic chunk match against the same conversation reinforce each other.

Source code in src/transcript_indexer/search.py

def search_hybrid(
    conn: sqlite3.Connection,
    cfg: Config,
    query: str,
    *,
    k: int = 20,
    speaker: str | None = None,
    since: str | None = None,
    until: str | None = None,
) -> list[HybridHit]:
    """Hybrid FTS+semantic search merged with Reciprocal Rank Fusion.

    Hits are keyed by `conversation_id` so an FTS turn match and a semantic
    chunk match against the same conversation reinforce each other.
    """
    fts_hits = search_turns_fts(conn, query, speaker=speaker, since=since, until=until, limit=k)
    sem_hits = search_semantic(conn, cfg, query, k=k, since=since, until=until)

    merged: dict[int, HybridHit] = {}
    scores: dict[int, float] = {}

    for rank, hit in enumerate(fts_hits, start=1):
        s = _rrf_score(rank)
        prev = merged.get(hit.conversation_id)
        scores[hit.conversation_id] = scores.get(hit.conversation_id, 0.0) + s
        merged[hit.conversation_id] = HybridHit(
            conversation_id=hit.conversation_id,
            conversation_title=hit.conversation_title,
            conversation_started_at=hit.conversation_started_at,
            score=scores[hit.conversation_id],
            fts=hit,
            semantic=prev.semantic if prev else None,
        )

    for sem_rank, sem_hit in enumerate(sem_hits, start=1):
        s = _rrf_score(sem_rank)
        prev = merged.get(sem_hit.conversation_id)
        scores[sem_hit.conversation_id] = scores.get(sem_hit.conversation_id, 0.0) + s
        merged[sem_hit.conversation_id] = HybridHit(
            conversation_id=sem_hit.conversation_id,
            conversation_title=sem_hit.conversation_title,
            conversation_started_at=sem_hit.conversation_started_at,
            score=scores[sem_hit.conversation_id],
            fts=prev.fts if prev else None,
            semantic=sem_hit,
        )

    return sorted(merged.values(), key=lambda h: h.score, reverse=True)[:k]