`aaa.ingest`¶

Pipeline ¶

Pipeline(
    index: EmbeddingIndex,
    chunker: Chunker | None = None,
    *,
    scope: str = "default",
    embedder_id: str = "",
    batch_size: int = 64,
    concurrency: int = 4,
    max_chunks_per_document: int = 20000,
)

forget `async` ¶

forget(source: str) -> bool

Remove one source's chunks and its manifest entry.

SyncReport `dataclass` ¶

SyncReport(
    added: int = 0,
    updated: int = 0,
    unchanged: int = 0,
    removed: int = 0,
    chunks_written: int = 0,
    chunks_deleted: int = 0,
    embed_calls: int = 0,
    embedded_chunks: int = 0,
    skipped: list[tuple[str, str]] = list(),
    errors: list[tuple[str, str]] = list(),
)

DirectoryLoader ¶

DirectoryLoader(
    root: str | Path,
    *,
    glob: str = "**/*",
    extensions: Iterable[str] = TEXT_SUFFIXES,
    exclude: Sequence[str] = (),
    include_hidden: bool = False,
    follow_symlinks: bool = False,
    max_bytes: int = 5000000,
    max_files: int = 100000,
)

Every text file under root (by extension), with the path relative to root as the document's source.

Safe by default: symlinks are not followed, hidden files and directories (.git, .env) are skipped, only extensions are read, files over max_bytes and binary files are skipped (and listed in .skipped with the reason), and a file that resolves outside root is refused. max_files stops a runaway tree.

FileLoader ¶

FileLoader(
    path: str | Path,
    *,
    source: str | None = None,
    max_bytes: int = 5000000,
)

PDFLoader ¶

PDFLoader(
    path: str | Path,
    *,
    source: str | None = None,
    max_bytes: int = 50000000,
)

A PDF, one document per page (meta["page"] is 1-based). Needs pip install pypdf; text only (no OCR, so a scanned PDF yields nothing).

RecursiveChunker ¶

RecursiveChunker(
    size: int = 1000,
    overlap: int = 100,
    *,
    separators: Sequence[str] = _SEPARATORS,
    length: Callable[[str], int] = len,
)

Split on the largest natural boundary that keeps pieces under size (paragraphs, then lines, sentences, words, characters), then merge neighbours back up to size, carrying overlap of the previous chunk into the next so a sentence cut at a boundary is whole in one of them.

size and overlap are measured by length (characters by default). For token budgets use RecursiveChunker.by_tokens(256): it measures with the same estimate the context manager uses, which over-counts a little (the safe direction).

MarkdownChunker ¶

MarkdownChunker(max_chars: int = 1500, overlap: int = 100)

One chunk per section, cut at headings, each carrying meta["heading"] (the path, Guide > Install > Linux); a section longer than max_chars is cut further with RecursiveChunker. Fenced code blocks are not mistaken for headings.

Document `dataclass` ¶

Document(
    source: str, text: str, meta: dict[str, Any] = dict()
)

One source's text. source is its stable identity (a relative path, a URL): the same source on the next sync is the same document.

hash `property` ¶

hash: str

Identity of the document's content AND metadata: a changed title or page number re-syncs it (reusing vectors when the text is the same).

Chunk `dataclass` ¶

Chunk(text: str, meta: dict[str, Any] = dict())

aaa.ingest¶