Skip to content

Chunking

transcript_indexer.chunking

Turn-aware chunking for embedding.

Chunks are derived from turns rows for a single conversation. The default strategy is turn_group: consecutive turns are grouped into overlapping windows of turns_per_chunk turns. Turns are never split. A chunk's text is the speaker-prefixed lines joined with newlines so the embedder sees turn-level structure.

fixed_tokens is a placeholder for a future strategy and currently raises.

chunk_turn_group(turns, cfg, *, count_tokens=None)

Group consecutive turns into overlapping windows.

Windows of turns_per_chunk turns advance by turns_per_chunk - overlap_turns. Windows whose token count falls below min_chunk_tokens are dropped (when a token counter is supplied). Without a counter, every window is emitted.

Source code in src/transcript_indexer/chunking.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def chunk_turn_group(
    turns: Sequence[TurnRow],
    cfg: ChunkingConfig,
    *,
    count_tokens: Callable[[str], int] | None = None,
) -> list[Chunk]:
    """Group consecutive turns into overlapping windows.

    Windows of `turns_per_chunk` turns advance by `turns_per_chunk - overlap_turns`.
    Windows whose token count falls below `min_chunk_tokens` are dropped (when
    a token counter is supplied). Without a counter, every window is emitted.
    """
    n = len(turns)
    if n == 0:
        return []

    size = max(1, cfg.turns_per_chunk)
    overlap = max(0, min(cfg.overlap_turns, size - 1))
    step = size - overlap

    chunks: list[Chunk] = []
    seen_hashes: set[str] = set()
    start = 0
    while start < n:
        end = min(start + size, n)
        window = turns[start:end]
        text = _format_chunk_text(window)
        if count_tokens is not None and cfg.max_chunk_tokens > 0:
            # Shrink window from the tail until it fits within the token ceiling.
            # Stop at 1 turn — we don't split individual turns.
            while len(window) > 1 and count_tokens(text) > cfg.max_chunk_tokens:
                window = window[:-1]
                text = _format_chunk_text(window)
        # When the window was shrunk, advance by the actual window size so no turn
        # is skipped. Use the configured step only for full-size windows.
        was_shrunk = len(window) < end - start
        advance = len(window) if was_shrunk else step
        if count_tokens is not None and count_tokens(text) < cfg.min_chunk_tokens:
            # Tail windows below threshold are skipped; mid-corpus undersized
            # windows are rare with overlap > 0.
            if end == n and not was_shrunk:
                break
            start += advance
            continue
        text_hash = sha256_text(text)
        if text_hash not in seen_hashes:
            seen_hashes.add(text_hash)
            chunks.append(
                Chunk(
                    kind="turn_group",
                    start_turn_idx=window[0].idx,
                    end_turn_idx=window[-1].idx,
                    text=text,
                    text_hash=text_hash,
                )
            )
        if end == n and not was_shrunk:
            break
        start += advance
    return chunks

chunk_conversation(turns, cfg, *, count_tokens=None)

Dispatch to the chunking strategy named in cfg.strategy.

Source code in src/transcript_indexer/chunking.py 
103
104
105
106
107
108
109
110
111
112
113
114
def chunk_conversation(
    turns: Sequence[TurnRow],
    cfg: ChunkingConfig,
    *,
    count_tokens: Callable[[str], int] | None = None,
) -> list[Chunk]:
    """Dispatch to the chunking strategy named in `cfg.strategy`."""
    if cfg.strategy == "turn_group":
        return chunk_turn_group(turns, cfg, count_tokens=count_tokens)
    if cfg.strategy == "fixed_tokens":
        raise NotImplementedError("fixed_tokens chunking is not implemented yet")
    raise ValueError(f"unknown chunking strategy: {cfg.strategy}")