Ingestion Guide¶

from anchor.ingestion import DocumentIngester

ingester = DocumentIngester()

# Ingest raw text
items = ingester.ingest_text("Astro-context is a modular RAG framework.")
print(items[0].content)

# Ingest a single file
items = ingester.ingest_file("docs/intro.md")

# Ingest an entire directory
items = ingester.ingest_directory("docs/", extensions=[".md", ".txt"])

from anchor.ingestion import (
    DocumentIngester,
    SentenceChunker,
    MetadataEnricher,
)
from anchor.models.context import SourceType

def add_category(text, idx, total, meta):
    meta["category"] = "documentation"
    return meta

ingester = DocumentIngester(
    chunker=SentenceChunker(chunk_size=256, overlap=1),
    enricher=MetadataEnricher(enrichers=[add_category]),
    source_type=SourceType.RETRIEVAL,
    priority=5,
)

items = ingester.ingest_text(
    "First sentence. Second sentence. Third sentence.",
    doc_id="manual-id-001",
)
for item in items:
    print(item.id, item.metadata.get("category"))

from anchor.ingestion import FixedSizeChunker

chunker = FixedSizeChunker(chunk_size=128, overlap=20)
chunks = chunker.chunk("A very long document text...")

from anchor.ingestion import RecursiveCharacterChunker

chunker = RecursiveCharacterChunker(chunk_size=256, overlap=30)
chunks = chunker.chunk("Paragraph one.\n\nParagraph two.\n\nParagraph three.")

from anchor.ingestion import SentenceChunker

chunker = SentenceChunker(chunk_size=256, overlap=1)
chunks = chunker.chunk("First sentence. Second sentence. Third sentence.")

import math
from anchor.ingestion import SemanticChunker

# Deterministic embed function for demonstration
def embed_fn(texts: list[str]) -> list[list[float]]:
    return [
        [math.sin(i + c) for c in range(8)]
        for i, t in enumerate(texts)
    ]

chunker = SemanticChunker(
    embed_fn=embed_fn,
    threshold=0.5,
    chunk_size=256,
    min_chunk_size=30,
)
chunks = chunker.chunk("Sentence one. Sentence two. Totally different topic here.")

from anchor.ingestion import CodeChunker

chunker = CodeChunker(language="python", chunk_size=256)
code = '''
def hello():
    print("hello")

def world():
    print("world")

class Greeter:
    pass
'''
chunks = chunker.chunk(code)

from anchor.ingestion import TableAwareChunker

chunker = TableAwareChunker(chunk_size=256)
text = """
Some introductory text.

| Name  | Value |
|-------|-------|
| alpha | 1     |
| beta  | 2     |

More prose after the table.
"""
chunks = chunker.chunk(text)

from anchor.ingestion import ParentChildChunker

chunker = ParentChildChunker(
    parent_chunk_size=512,
    child_chunk_size=128,
    parent_overlap=50,
    child_overlap=10,
)

# Plain string chunks (Chunker protocol)
children = chunker.chunk("A long document to split hierarchically...")

# Chunks with metadata (includes parent_id, parent_text)
children_with_meta = chunker.chunk_with_metadata("A long document...")
for text, meta in children_with_meta:
    print(meta["parent_id"], len(text))

from anchor.ingestion import DocumentIngester, PlainTextParser

ingester = DocumentIngester(
    parsers={".log": PlainTextParser()},
)

from anchor.ingestion import MetadataEnricher

def tag_language(text, idx, total, meta):
    meta["language"] = "en"
    return meta

def add_summary_flag(text, idx, total, meta):
    meta["needs_summary"] = len(text.split()) > 100
    return meta

enricher = MetadataEnricher(enrichers=[tag_language, add_summary_flag])
enricher.add(lambda text, idx, total, meta: {**meta, "version": "1.0"})

from anchor.ingestion import ParentExpander
from anchor.pipeline import postprocessor_step

expander = ParentExpander(keep_child=True)
step = postprocessor_step("expand-parents", expander)

import math
from anchor.ingestion import (
    DocumentIngester,
    SemanticChunker,
    MetadataEnricher,
)

def embed_fn(texts: list[str]) -> list[list[float]]:
    return [[math.sin(i + c) for c in range(8)] for i, _ in enumerate(texts)]

def add_source(text, idx, total, meta):
    meta["source"] = "user-docs"
    return meta

ingester = DocumentIngester(
    chunker=SemanticChunker(embed_fn=embed_fn, threshold=0.5, chunk_size=256),
    enricher=MetadataEnricher(enrichers=[add_source]),
)

items = ingester.ingest_text(
    "Machine learning models learn patterns from data. "
    "They generalize to unseen examples. "
    "Transformers use self-attention mechanisms. "
    "RAG combines retrieval with generation.",
    doc_id="ml-intro",
)
for item in items:
    print(f"{item.id}: {item.content[:60]}...")
    print(f"  metadata: {item.metadata}")