Embed

transcript_indexer.embed

Embedding pipeline.

Wraps PydanticAI's Embedder to chunk conversation turns and persist vectors into chunk_embeddings. Embeddings are cached by chunks.text_hash so re-syncing a conversation with unchanged content reuses existing vectors.

The vec0 chunk_embeddings table is keyed by chunks.id (passed as rowid).

build_embedder(cfg)

Construct an Embedder from EmbeddingConfig.

Source code in src/transcript_indexer/embed.py

def build_embedder(cfg: Config) -> Embedder:
    """Construct an Embedder from EmbeddingConfig."""
    settings: EmbeddingSettings | None = None
    if cfg.embedding.dimensions:
        settings = EmbeddingSettings(dimensions=cfg.embedding.dimensions)
    return Embedder(cfg.embedding.spec, settings=settings)

embed_conversation(conn, cfg, conversation_id, *, embedder=None, report=None, on_progress=None)

Chunk and embed a single conversation.

Assumes chunks rows for this conversation have already been cleared (sync's delete-and-reinsert handles this naturally on new/changed conversations).

on_progress fires with these events for UI feedback: - ("conversation_start", conversation_id) - ("chunks_planned", n_chunks) - ("cached", n) after each cache-hit copy - ("embedded", n) after each batch is persisted - ("conversation_done", conversation_id)

Source code in src/transcript_indexer/embed.py

def embed_conversation(
    conn: sqlite3.Connection,
    cfg: Config,
    conversation_id: int,
    *,
    embedder: Embedder | None = None,
    report: EmbedReport | None = None,
    on_progress: ProgressCallback | None = None,
) -> EmbedReport:
    """Chunk and embed a single conversation.

    Assumes `chunks` rows for this conversation have already been cleared
    (sync's delete-and-reinsert handles this naturally on new/changed
    conversations).

    `on_progress` fires with these events for UI feedback:
      - ``("conversation_start", conversation_id)``
      - ``("chunks_planned", n_chunks)``
      - ``("cached", n)`` after each cache-hit copy
      - ``("embedded", n)`` after each batch is persisted
      - ``("conversation_done", conversation_id)``
    """
    report = report or EmbedReport()
    if on_progress:
        on_progress("conversation_start", conversation_id)
    turns = _load_turns(conn, conversation_id)
    if not turns:
        report.skipped_conversations.append(conversation_id)
        if on_progress:
            on_progress("conversation_done", conversation_id)
        return report

    embedder = embedder or build_embedder(cfg)
    chunks = chunk_conversation(
        turns,
        cfg.chunking,
        count_tokens=lambda text: embedder.count_tokens_sync(text),
    )
    if on_progress:
        on_progress("chunks_planned", len(chunks))
    if not chunks:
        if on_progress:
            on_progress("conversation_done", conversation_id)
        return report

    report.chunks_total += len(chunks)

    pending: list[tuple[int, Chunk]] = []
    with conn:
        for chunk in chunks:
            cached = _find_cached_embedding(conn, chunk.text_hash)
            chunk_id = _insert_chunk(conn, conversation_id, chunk)
            if cached is not None:
                _, blob = cached
                _insert_embedding_blob(conn, chunk_id, blob)
                report.cache_hits += 1
                if on_progress:
                    on_progress("cached", 1)
            else:
                pending.append((chunk_id, chunk))
                report.cache_misses += 1

    if not pending:
        with conn:
            _record_run(conn, cfg)
        if on_progress:
            on_progress("conversation_done", conversation_id)
        return report

    batch_size = max(1, cfg.embedding.batch_size)
    for start in range(0, len(pending), batch_size):
        batch = pending[start : start + batch_size]
        texts = [c.text for _, c in batch]
        result = embedder.embed_documents_sync(texts)
        if len(result.embeddings) != len(batch):
            raise RuntimeError(
                f"embedder returned {len(result.embeddings)} vectors for {len(batch)} inputs"
            )
        with conn:
            for (chunk_id, _chunk), vector in zip(batch, result.embeddings, strict=True):
                _insert_embedding_vector(conn, chunk_id, vector)
            report.embedded += len(batch)
            if result.usage is not None and result.usage.input_tokens:
                report.input_tokens += int(result.usage.input_tokens)
        if on_progress:
            on_progress("embedded", len(batch))

    with conn:
        _record_run(conn, cfg)

    if on_progress:
        on_progress("conversation_done", conversation_id)
    return report

embed_conversations(conn, cfg, conversation_ids, *, embedder=None, on_progress=None)

Embed many conversations sharing a single Embedder instance.

Source code in src/transcript_indexer/embed.py

def embed_conversations(
    conn: sqlite3.Connection,
    cfg: Config,
    conversation_ids: Sequence[int],
    *,
    embedder: Embedder | None = None,
    on_progress: ProgressCallback | None = None,
) -> EmbedReport:
    """Embed many conversations sharing a single Embedder instance."""
    report = EmbedReport()
    if not conversation_ids:
        return report
    embedder = embedder or build_embedder(cfg)
    for conv_id in conversation_ids:
        embed_conversation(
            conn,
            cfg,
            conv_id,
            embedder=embedder,
            report=report,
            on_progress=on_progress,
        )
    return report