Fixed-size Chunking: размер и overlap

from dataclasses import dataclass, field
from typing import Iterator


@dataclass
class Document:
    page_content: str
    metadata: dict = field(default_factory=dict)


class CharacterSplitter:
    """
    Fixed-size chunking по символам с мягким разрывом по пробелу.

    chunk_size    — максимальная длина чанка в символах
    chunk_overlap — сколько символов из конца предыдущего чанка
                    повторяется в начале следующего
    """

    def __init__(self, chunk_size: int = 700, chunk_overlap: int = 70):
        if chunk_overlap >= chunk_size:
            raise ValueError("chunk_overlap должен быть меньше chunk_size")
        self.chunk_size    = chunk_size
        self.chunk_overlap = chunk_overlap
        self.step          = chunk_size - chunk_overlap

    def split_text(self, text: str) -> list[str]:
        """Нарезает строку на список чанков."""
        chunks = []
        start  = 0
        text_len = len(text)

        while start < text_len:
            end = min(start + self.chunk_size, text_len)

            # Откатываемся к пробелу, чтобы не резать слово
            if end < text_len:
                # Ищем последний пробел в последних 50 символах окна
                boundary = text.rfind(" ", max(start, end - 50), end)
                if boundary > start:
                    end = boundary

            chunk = text[start:end].strip()
            if chunk:
                chunks.append(chunk)

            start += self.step

        return chunks

    def split_documents(self, documents: list[Document]) -> list[Document]:
        """Разбивает список документов на чанки, сохраняя метаданные."""
        result = []
        for doc in documents:
            chunks = self.split_text(doc.page_content)
            for i, chunk_text in enumerate(chunks):
                result.append(Document(
                    page_content=chunk_text,
                    metadata={
                        **doc.metadata,           # сохраняем все исходные метаданные
                        "chunk_index": i,          # порядковый номер чанка
                        "chunk_total": len(chunks),# всего чанков в документе
                        "chunk_size":  len(chunk_text),
                    }
                ))
        return result


# Использование
splitter = CharacterSplitter(chunk_size=700, chunk_overlap=70)
docs = splitter.split_documents([
    Document(
        page_content="Длинный текст документа...",
        metadata={"source": "policy.pdf", "title": "Регламент"}
    )
])
print(f"Получено чанков: {len(docs)}")
print(f"Первый чанк ({len(docs[0].page_content)} символов):")
print(docs[0].page_content[:200])

import tiktoken


class TokenSplitter:
    """
    Fixed-size chunking по токенам.
    Использует tiktoken для подсчёта — токены совпадают с моделями OpenAI/Azure.
    Для Claude: используйте cl100k_base как приближение (≈95% точность).
    """

    def __init__(
        self,
        chunk_size: int = 256,     # токенов
        chunk_overlap: int = 32,   # токенов
        encoding_name: str = "cl100k_base",  # GPT-4 / text-embedding-3
    ):
        self.enc           = tiktoken.get_encoding(encoding_name)
        self.chunk_size    = chunk_size
        self.chunk_overlap = chunk_overlap
        self.step          = chunk_size - chunk_overlap

    def _encode(self, text: str) -> list[int]:
        return self.enc.encode(text)

    def _decode(self, tokens: list[int]) -> str:
        return self.enc.decode(tokens)

    def split_text(self, text: str) -> list[str]:
        tokens = self._encode(text)
        chunks = []
        start  = 0

        while start < len(tokens):
            end   = min(start + self.chunk_size, len(tokens))
            chunk = self._decode(tokens[start:end])
            if chunk.strip():
                chunks.append(chunk.strip())
            start += self.step

        return chunks

    def split_documents(self, documents: list[Document]) -> list[Document]:
        result = []
        for doc in documents:
            chunks = self.split_text(doc.page_content)
            for i, chunk_text in enumerate(chunks):
                token_count = len(self._encode(chunk_text))
                result.append(Document(
                    page_content=chunk_text,
                    metadata={
                        **doc.metadata,
                        "chunk_index":       i,
                        "chunk_total":       len(chunks),
                        "chunk_token_count": token_count,
                    }
                ))
        return result


# Пример: текст из 2000 слов → ~500 токенов → 2 чанка по 256
splitter = TokenSplitter(chunk_size=256, chunk_overlap=32)
docs = splitter.split_documents([
    Document(page_content="..." * 300, metadata={"source": "report.pdf"})
])
for d in docs:
    print(f"Chunk {d.metadata['chunk_index']}: {d.metadata['chunk_token_count']} tokens")

import asyncio
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path


def _split_one(args: tuple) -> list[dict]:
    """Запускается в процессе — нет GIL, чистый CPU."""
    text, metadata, chunk_size, overlap = args
    splitter = CharacterSplitter(chunk_size=chunk_size, chunk_overlap=overlap)
    chunks = splitter.split_text(text)
    return [
        {"page_content": c, "metadata": {**metadata, "chunk_index": i, "chunk_total": len(chunks)}}
        for i, c in enumerate(chunks)
    ]


async def chunk_documents_async(
    documents: list[Document],
    chunk_size: int = 700,
    chunk_overlap: int = 70,
    max_workers: int = 4,
) -> list[Document]:
    """
    Параллельный чанкинг через ProcessPoolExecutor.
    1000 документов × 5000 символов ≈ 3–5 сек вместо 15–20 сек синхронно.
    """
    loop = asyncio.get_event_loop()
    args_list = [
        (doc.page_content, doc.metadata, chunk_size, chunk_overlap)
        for doc in documents
    ]

    with ProcessPoolExecutor(max_workers=max_workers) as pool:
        results = await asyncio.gather(*[
            loop.run_in_executor(pool, _split_one, args)
            for args in args_list
        ])

    return [
        Document(page_content=r["page_content"], metadata=r["metadata"])
        for batch in results
        for r in batch
    ]


# Бенчмарк
async def demo():
    import time

    # Генерируем 500 документов по ~3000 символов
    docs = [Document(
        page_content="Слово " * 500,
        metadata={"source": f"doc_{i}.pdf"}
    ) for i in range(500)]

    t = time.perf_counter()
    chunks = await chunk_documents_async(docs, chunk_size=700, chunk_overlap=70)
    elapsed = time.perf_counter() - t

    print(f"Входных документов: {len(docs)}")
    print(f"Получено чанков:    {len(chunks)}")
    print(f"Время:              {elapsed:.2f}с")


if __name__ == "__main__":
    asyncio.run(demo())

import statistics
from collections import Counter


def analyze_chunks(chunks: list[Document]) -> dict:
    """
    Анализирует распределение чанков:
    - Средний/медианный размер
    - Количество слишком маленьких (< 50 символов) — признак плохого текста
    - Количество слишком больших (> chunk_size × 1.1) — не должно быть
    """
    sizes = [len(c.page_content) for c in chunks]
    if not sizes:
        return {}

    tiny_threshold = 50
    tiny_count = sum(1 for s in sizes if s < tiny_threshold)

    return {
        "total_chunks":     len(chunks),
        "mean_size":        round(statistics.mean(sizes)),
        "median_size":      round(statistics.median(sizes)),
        "min_size":         min(sizes),
        "max_size":         max(sizes),
        "stdev":            round(statistics.stdev(sizes)) if len(sizes) > 1 else 0,
        "tiny_chunks":      tiny_count,           # подозрительно маленькие
        "tiny_pct":         round(tiny_count / len(sizes) * 100, 1),
    }


def print_chunk_report(chunks: list[Document], chunk_size: int) -> None:
    stats = analyze_chunks(chunks)
    print(f"{'─' * 40}")
    print(f"Всего чанков:       {stats['total_chunks']}")
    print(f"Средний размер:     {stats['mean_size']} символов")
    print(f"Медиана:            {stats['median_size']} символов")
    print(f"Мин / Макс:         {stats['min_size']} / {stats['max_size']}")
    print(f"Σ отклонение:       {stats['stdev']}")
    print(f"Tiny (< 50 симв):   {stats['tiny_chunks']} ({stats['tiny_pct']}%)")

    # Гистограмма размеров
    buckets = [0] * 5
    for c in chunks:
        size = len(c.page_content)
        idx  = min(int(size / chunk_size * 4), 4)
        buckets[idx] += 1

    print(f"\nРаспределение:")
    labels = ["0–25%", "25–50%", "50–75%", "75–100%", "100%+"]
    for label, count in zip(labels, buckets):
        bar = "█" * (count * 20 // max(buckets, default=1))
        print(f"  {label:8s} {bar} {count}")
    print(f"{'─' * 40}")


# Использование
splitter = CharacterSplitter(chunk_size=700, chunk_overlap=70)
chunks   = splitter.split_documents(my_documents)
print_chunk_report(chunks, chunk_size=700)