Ingestion API Reference¶

API reference for the anchor.ingestion module. For usage patterns and examples, see the Ingestion Guide.

DocumentIngester¶

Orchestrates document parsing, chunking, and metadata extraction. Converts raw files or text into ContextItem objects suitable for retriever.index(items).

class DocumentIngester(
    chunker: Chunker | None = None,
    tokenizer: Tokenizer | None = None,
    parsers: dict[str, DocumentParser] | None = None,
    enricher: MetadataEnricher | None = None,
    source_type: SourceType = SourceType.RETRIEVAL,
    priority: int = 5,
)

Methods¶

ingest_text(text, doc_id=None, doc_metadata=None)¶

Ingest raw text into context items.

Returns: list[ContextItem]

ingest_file(path, doc_id=None)¶

Parse and chunk a single file. The parser is auto-detected from file extension.

Returns: list[ContextItem] Raises: IngestionError if no parser found; FileNotFoundError if file missing.

ingest_directory(directory, glob_pattern="**/*", extensions=None)¶

Recursively ingest all matching files in a directory.

Returns: list[ContextItem] Raises: IngestionError if directory does not exist.

Chunkers¶

All chunkers implement the Chunker protocol:

def chunk(self, text: str, metadata: dict[str, Any] | None = None) -> list[str]

FixedSizeChunker¶

Split text into fixed-size chunks by token count with overlap.

class FixedSizeChunker(
    chunk_size: int = 512,
    overlap: int = 50,
    tokenizer: Tokenizer | None = None,
)

RecursiveCharacterChunker¶

Split text using a hierarchy of separators, falling back to finer splits.

class RecursiveCharacterChunker(
    chunk_size: int = 512,
    overlap: int = 50,
    separators: tuple[str, ...] | None = None,
    tokenizer: Tokenizer | None = None,
)

SentenceChunker¶

Split text at sentence boundaries, grouping sentences to fill chunks. Overlap is measured in sentences.

class SentenceChunker(
    chunk_size: int = 512,
    overlap: int = 1,
    tokenizer: Tokenizer | None = None,
)

SemanticChunker¶

Split text at semantic boundaries using embedding similarity.

class SemanticChunker(
    embed_fn: Callable[[list[str]], list[list[float]]],
    tokenizer: Tokenizer | None = None,
    threshold: float = 0.5,
    chunk_size: int = 512,
    min_chunk_size: int = 50,
)

CodeChunker¶

Split source code at function, class, and definition boundaries.

class CodeChunker(
    language: str | None = None,
    chunk_size: int = 512,
    overlap: int = 50,
    tokenizer: Tokenizer | None = None,
)

Supported languages: python, javascript, typescript, go, rust.

TableAwareChunker¶

Preserve tables as atomic chunks while delegating prose to an inner chunker.

class TableAwareChunker(
    inner_chunker: Any | None = None,
    chunk_size: int = 512,
    tokenizer: Tokenizer | None = None,
)

ParentChildChunker¶

Two-level hierarchical chunker producing large parent and small child chunks.

class ParentChildChunker(
    parent_chunk_size: int = 1024,
    child_chunk_size: int = 256,
    parent_overlap: int = 100,
    child_overlap: int = 25,
    tokenizer: Tokenizer | None = None,
)

Methods¶

• chunk(text, metadata=None) -- Returns child chunk texts only (list[str]).
• chunk_with_metadata(text, metadata=None) -- Returns list[tuple[str, dict[str, Any]]] with parent_id, parent_text, parent_index, child_index, and is_child_chunk in each metadata dict.

Parsers¶

All parsers implement the DocumentParser protocol:

def parse(self, source: Path | bytes) -> tuple[str, dict[str, Any]]

PlainTextParser¶

Parse plain text files. Supported extensions: .txt.

Metadata produced: filename, extension, line_count.

MarkdownParser¶

Parse Markdown files, extracting headings and detecting frontmatter. Supported extensions: .md, .markdown.

Metadata produced: filename, extension, title (first H1), headings list, has_frontmatter.

HTMLParser¶

Parse HTML files, stripping tags and extracting text. Uses Python's stdlib html.parser with zero external dependencies. Supported extensions: .html, .htm.

Metadata produced: filename, extension, title.

PDFParser¶

Parse PDF files using pypdf. Supported extensions: .pdf.

Metadata produced: filename, extension, page_count, title, author.

Metadata Functions¶

generate_doc_id(content, source_path=None)¶

Generate a deterministic 16-character hex document ID from SHA-256.

Returns: str -- 16-character hex string.

generate_chunk_id(doc_id, chunk_index)¶

Generate a chunk ID in the format "{doc_id}-chunk-{chunk_index}".

Returns: str

extract_chunk_metadata(chunk_text, chunk_index, total_chunks, doc_id, doc_metadata=None)¶

Build standard metadata for a single chunk.

Returns: dict[str, Any] with keys: parent_doc_id, chunk_index, total_chunks, word_count, char_count, plus propagated doc metadata (prefixed with doc_).

MetadataEnricher¶

Chain of user-provided metadata enrichment functions.

class MetadataEnricher(
    enrichers: list[Callable[[str, int, int, dict[str, Any]], dict[str, Any]]] | None = None,
)

Each enricher callable has signature: (text: str, chunk_index: int, total_chunks: int, metadata: dict) -> dict

Methods¶

add(fn)¶

Register an additional enricher function.

enrich(text, chunk_index, total_chunks, metadata)¶

Run all enrichers in order, threading metadata through. Returns the enriched metadata dict.

ParentExpander¶

Post-processor that expands child chunks back to parent text, deduplicating by parent_id.

class ParentExpander(
    keep_child: bool = False,
)

Methods¶

process(items, query=None)¶

Expand child chunks to parent text. Items with is_child_chunk in metadata have their content replaced with parent_text. Multiple children from the same parent are deduplicated (first occurrence wins).

Returns: list[ContextItem]