API Reference

omni_ingest

Modules:

Name	Description
agent	Specialized agents for parsing, cleaning, chunking, and enriching knowledge items.
cli	Command-line entrypoint for running Omni Ingest pipeline YAML files.
core	Core logic and foundational data models for the ingestion framework.
parser	Resource readers for turning local or remote inputs into resolved content.
port	Interfaces and concrete implementations for external storage and data connectivity.

agent

Specialized agents for parsing, cleaning, chunking, and enriching knowledge items.

Modules:

Name	Description
chunking	Agents for segmenting large documents and transcripts into manageable chunks.
deduplication	Deduplication strategies for vector indexing.
document	Document agents.
enrichment	Agents for structured metadata extraction and enrichment.
governance	Agents for pipeline validation, audit logging, and lineage metadata.
indexing	Agents for generating embeddings.
modality	Agents for modality transformation.

chunking

Agents for segmenting large documents and transcripts into manageable chunks.

Classes:

Name	Description
PageChunkingAgent	Partitions PDF documents based on their original page layout.
PageStitchingAgent	Combines selected PDF page items into larger PDF items.
SentenceChunkingAgent	Segments text into discrete chunks while maintaining sentence boundaries.
TimedChunkingAgent	Segments audio transcription data into temporal blocks.

PageChunkingAgent pydantic-model

Bases: BaseModel, Step

Partitions PDF documents based on their original page layout.

This agent creates distinct KnowledgeItems for each page identified in a PDF source, maintaining the hierarchical structure of the input document.

Fields:

• mode (Literal['text', 'pdf'])

Source code in src/omni_ingest/agent/chunking.py

class PageChunkingAgent(BaseModel, Step):
    """
    Partitions PDF documents based on their original page layout.

    This agent creates distinct KnowledgeItems for each page identified
    in a PDF source, maintaining the hierarchical structure of the
    input document.
    """

    mode: Literal["text", "pdf"] = Field(default="text", description="Whether to emit extracted page text or single-page PDFs")

    async def run(self, ctx: IngestionContext[ResolvedResource]) -> StepResult:
        if not ctx.items:
            return StepResult(status=StepStatus.SKIPPED, error="No document")
        source = ctx.items[0]
        if (ct := await source.content_type(ctx)) != "application/pdf":
            raise ValueError(f"Expected a document, got {ct}")

        enc = source.content_encoding
        items = []
        with io.BytesIO(await source.content(ctx)) as s, PdfReader(stream=s) as reader:
            total = len(reader.pages)
            ctx.progress(0, total, "reading pages")
            for i, page in enumerate(reader.pages, 1):
                metadata = source.metadata | {"page": i, "content_type": "application/pdf" if self.mode == "pdf" else "text/plain"}
                if self.mode == "pdf":
                    items.append(KnowledgeItem(content_uri=resource_pages_uri([i]), content_encoding=enc, tenant_id=source.tenant_id, source_uri=source.source_uri, metadata=metadata))
                elif text := page.extract_text().strip():
                    items.append(KnowledgeItem(raw_content=text.encode(enc), content_encoding=enc, tenant_id=source.tenant_id, source_uri=source.source_uri, metadata=metadata))
                ctx.progress(i, total, f"page {i}/{total}")
        ctx.items = items
        return StepResult(status=StepStatus.SUCCESS, items=ctx.items, metadata={"page_count": len(ctx.items), "mode": self.mode})

mode pydantic-field

mode: Literal['text', 'pdf'] = 'text'

Whether to emit extracted page text or single-page PDFs

PageStitchingAgent pydantic-model

Bases: BaseModel, Step

Combines selected PDF page items into larger PDF items.

jq expressions define the stitch specs and selected page item ids, making this reusable for chapters, sections, ranges, or any metadata-driven slices.

Fields:

• foreach (JqExpression)
• input (JqExpression)
• metadata (JqExpression)
• replace (bool)

Source code in src/omni_ingest/agent/chunking.py

class PageStitchingAgent(BaseModel, Step):
    """
    Combines selected PDF page items into larger PDF items.

    jq expressions define the stitch specs and selected page item ids, making
    this reusable for chapters, sections, ranges, or any metadata-driven slices.
    """

    foreach: JqExpression = Field(default=".", description="jq expression producing stitch specs")
    input: JqExpression = Field(..., description="jq expression selecting page item ids for each stitch spec")
    metadata: JqExpression = Field(default=".", description="jq expression producing metadata for each stitched item")
    replace: bool = Field(default=True, description="Replace ctx.items with stitched PDFs")

    async def run(self, ctx: IngestionContext[ResolvedResource]) -> StepResult:
        item_docs = [{"id": str(item.id), "metadata": item.metadata, "source_uri": item.source_uri} for item in ctx.items]
        item_by_id = {item_doc["id"]: item for item_doc, item in zip(item_docs, ctx.items, strict=True)}
        doc = {"metadata": ctx.metadata, "items": item_docs}
        stitched = []

        specs = jq.compile(self.foreach).input_value(doc).all()
        ctx.progress(0, len(specs), "stitching pages")
        for i, spec in enumerate(specs, 1):
            ids = jq.compile(self.input).input_value(spec).first()
            if not isinstance(ids, list) or not ids or any(item_id not in item_by_id for item_id in ids):
                raise ValueError("PageStitchingAgent input must resolve to known item ids")

            items = [item_by_id[item_id] for item_id in ids]
            pages = [page for item in items if isinstance(page := item.metadata.get("page"), int)]
            uris = [item.content_uri for item in items if item.content_uri]
            metadata = jq.compile(self.metadata).input_value(spec).first()
            if len(pages) == len(items) and all(item.content_uri == resource_pages_uri([page]) for item, page in zip(items, pages, strict=True)):
                stitched.append(KnowledgeItem(raw_content=b"", content_uri=resource_pages_uri(pages), content_encoding="binary", metadata=metadata))
            elif len(uris) == len(items):
                stitched.append(KnowledgeItem(raw_content=b"", content_uri=content_chain_uri("pdf.concat", uris), content_encoding="binary", metadata=metadata))
            else:
                with io.BytesIO() as s, PdfWriter() as writer:
                    for item in items:
                        with io.BytesIO(await item.content(ctx)) as page_pdf, PdfReader(page_pdf) as reader:
                            writer.append(reader)
                    for page in writer.pages:
                        page.compress_content_streams()
                    writer.write(s)
                    stitched.append(KnowledgeItem(raw_content=s.getvalue(), content_encoding="binary", metadata=metadata))
            ctx.progress(i, len(specs), f"stitched {i}/{len(specs)}")

        if self.replace:
            ctx.items = stitched
        else:
            ctx.items.extend(stitched)
        return StepResult(status=StepStatus.SUCCESS, items=ctx.items, metadata={"count": len(stitched), "replace": self.replace})

foreach pydantic-field

foreach: JqExpression = '.'

jq expression producing stitch specs

input pydantic-field

input: JqExpression

jq expression selecting page item ids for each stitch spec

metadata pydantic-field

metadata: JqExpression = '.'

jq expression producing metadata for each stitched item

replace pydantic-field

replace: bool = True

Replace ctx.items with stitched PDFs

SentenceChunkingAgent pydantic-model

Bases: BaseModel, Step

Segments text into discrete chunks while maintaining sentence boundaries.

Large documents are partitioned into smaller segments of a specified character count. The agent prioritizes breaks at sentence endings to preserve contextual integrity for subsequent search operations.

Fields:

• chunk_size (PositiveInt)
• overlap (NonNegativeInt)
• foreach (JqExpression)
• input (JqExpression)
• separators (list[str])

Source code in src/omni_ingest/agent/chunking.py

class SentenceChunkingAgent(BaseModel, Step):
    """
    Segments text into discrete chunks while maintaining sentence boundaries.

    Large documents are partitioned into smaller segments of a specified
    character count. The agent prioritizes breaks at sentence endings
    to preserve contextual integrity for subsequent search operations.
    """

    chunk_size: PositiveInt = Field(default=512, description="Maximum number of characters per chunk")
    overlap: NonNegativeInt = Field(default=50, description="Character overlap between adjacent chunks")
    foreach: JqExpression = Field(default=".items[]", description="jq expression selecting items to chunk")
    input: JqExpression = Field(default=".text", description="jq expression selecting text to chunk")
    separators: list[str] = Field(default=[". ", "! ", "? ", ".\n", "!\n", "?\n"], description="Preferred sentence boundaries")

    async def run(self, ctx: IngestionContext[ResolvedResource]) -> StepResult:
        if not ctx.items:
            return StepResult(status=StepStatus.SKIPPED, error="No text")
        doc = await _doc(ctx, text=True)
        item_by_id = {str(item.id): item for item in ctx.items}
        chunks = []
        for source_idx, item_doc in enumerate(jq.compile(self.foreach).input_value(doc).all()):
            source = item_by_id[item_doc["id"]]
            text = jq.compile(self.input).input_value(item_doc).first()
            if text is None:
                continue
            if not isinstance(text, str):
                raise ValueError("Sentence chunking input must resolve to text")
            text, pos = " ".join(text.split()), 0
            while pos < len(text):
                end = min(pos + self.chunk_size, len(text))
                if end < len(text):
                    chunk = text[pos:end]
                    breaks = [chunk.rfind(separator) + len(separator) - 1 for separator in self.separators if chunk.rfind(separator) > self.chunk_size // 2]
                    if breaks:
                        end = pos + max(breaks) + 1
                if content := text[pos:end].strip():
                    chunks.append(KnowledgeItem(raw_content=content.encode(source.content_encoding), tenant_id=source.tenant_id, source_uri=source.source_uri, content_encoding=source.content_encoding, metadata=source.metadata | {"source_idx": source_idx, "char_start": pos}))
                pos = end - self.overlap if end < len(text) and end - self.overlap > pos else end
        ctx.items = chunks
        if not chunks:
            return StepResult(status=StepStatus.SKIPPED, error="No text")
        return StepResult(
            status=StepStatus.SUCCESS,
            items=ctx.items,
            metadata={"count": len(ctx.items), "chunk_size": self.chunk_size, "overlap": self.overlap},
        )

chunk_size pydantic-field

chunk_size: PositiveInt = 512

Maximum number of characters per chunk

foreach pydantic-field

foreach: JqExpression = '.items[]'

jq expression selecting items to chunk

input pydantic-field

input: JqExpression = '.text'

jq expression selecting text to chunk

overlap pydantic-field

overlap: NonNegativeInt = 50

Character overlap between adjacent chunks

separators pydantic-field

separators: list[str] = [
    ". ",
    "! ",
    "? ",
    ".\n",
    "!\n",
    "?\n",
]

Preferred sentence boundaries

TimedChunkingAgent pydantic-model

Bases: BaseModel, Step

Segments audio transcription data into temporal blocks.

This agent divides indexed audio text into intervals based on timestamps, facilitating precise time-based search and retrieval within audiovisual datasets.

Fields:

• segment_duration (timedelta)
• overlap (timedelta)
• pause_threshold (timedelta | None)
• tokens (JqExpression)
• text (JqExpression)
• start (JqExpression)
• duration (JqExpression)
• units_per_second (PositiveInt)

Validators:

• _validate_segment_duration → segment_duration
• _validate_overlap → overlap
• _validate_pause_threshold → pause_threshold

Source code in src/omni_ingest/agent/chunking.py

class TimedChunkingAgent(BaseModel, Step):
    """
    Segments audio transcription data into temporal blocks.

    This agent divides indexed audio text into intervals based on
    timestamps, facilitating precise time-based search and retrieval within
    audiovisual datasets.
    """

    segment_duration: timedelta = Field(default=timedelta(seconds=30), description="Target duration of each time block")
    overlap: timedelta = Field(default=timedelta(0), description="Overlap between adjacent time blocks")
    pause_threshold: timedelta | None = Field(default=None, description="Silence threshold to split blocks")
    tokens: JqExpression = Field(default=".metadata.word_timestamps[]?", description="jq expression selecting timed tokens")
    text: JqExpression = Field(default=".Word", description="jq expression selecting token text")
    start: JqExpression = Field(default=".Offset", description="jq expression selecting token start")
    duration: JqExpression = Field(default=".Duration", description="jq expression selecting token duration")
    units_per_second: PositiveInt = Field(default=10_000_000, description="Timestamp units per second")

    @field_validator("segment_duration")
    def _validate_segment_duration(cls, value: timedelta) -> timedelta:
        if value <= timedelta(0):
            raise ValueError("segment_duration must be positive")
        return value

    @field_validator("overlap")
    def _validate_overlap(cls, value: timedelta) -> timedelta:
        if value < timedelta(0):
            raise ValueError("overlap must be non-negative")
        return value

    @field_validator("pause_threshold")
    def _validate_pause_threshold(cls, value: timedelta | None) -> timedelta | None:
        if value is not None and value <= timedelta(0):
            raise ValueError("pause_threshold must be positive")
        return value

    @property
    def segment_duration_units(self) -> int:
        return int(self.segment_duration.total_seconds() * self.units_per_second)

    @property
    def overlap_units(self) -> int:
        return int(self.overlap.total_seconds() * self.units_per_second)

    @property
    def pause_threshold_units(self) -> int | None:
        return int(self.pause_threshold.total_seconds() * self.units_per_second) if self.pause_threshold else None

    async def run(self, ctx: IngestionContext[ResolvedResource]) -> StepResult:
        if not ctx.items:
            ctx.items = []
            return StepResult(status=StepStatus.SKIPPED, error="No timed tokens found in items")
        source = ctx.items[0]
        item_doc = {"id": str(source.id), "source_uri": source.source_uri, "metadata": source.metadata}
        words = jq.compile(self.tokens).input_value(item_doc).all()
        if not words:
            ctx.items = []
            return StepResult(status=StepStatus.SKIPPED, error="No timed tokens found in items")

        chunks: list[KnowledgeItem] = []
        current_chunk_words: list[Any] = []
        start_unit = 0
        text_expr, start_expr, duration_expr = map(jq.compile, (self.text, self.start, self.duration))

        for w in words:
            offset = start_expr.input_value(w).first()
            duration = duration_expr.input_value(w).first()
            if not current_chunk_words:
                start_unit = offset

            if (offset + duration - start_unit) > self.segment_duration_units and current_chunk_words:
                text = " ".join(text_expr.input_value(cw).first() for cw in current_chunk_words)
                last = current_chunk_words[-1]
                end = start_expr.input_value(last).first() + duration_expr.input_value(last).first()
                metadata = {"start_offset_ms": int(start_unit * 1000 / self.units_per_second), "end_offset_ms": int(end * 1000 / self.units_per_second), "word_count": len(current_chunk_words)}
                chunks.append(KnowledgeItem(raw_content=text.encode(source.content_encoding), content_encoding=source.content_encoding, tenant_id=source.tenant_id, source_uri=source.source_uri, metadata=metadata))
                current_chunk_words = [w]
                start_unit = offset
            else:
                current_chunk_words.append(w)

        if current_chunk_words:
            text = " ".join(text_expr.input_value(cw).first() for cw in current_chunk_words)
            last = current_chunk_words[-1]
            end = start_expr.input_value(last).first() + duration_expr.input_value(last).first()
            metadata = {"start_offset_ms": int(start_unit * 1000 / self.units_per_second), "end_offset_ms": int(end * 1000 / self.units_per_second), "word_count": len(current_chunk_words)}
            chunks.append(KnowledgeItem(raw_content=text.encode(source.content_encoding), content_encoding=source.content_encoding, tenant_id=source.tenant_id, source_uri=source.source_uri, metadata=metadata))

        ctx.items = chunks
        return StepResult(status=StepStatus.SUCCESS, items=chunks, metadata={"chunk_count": len(chunks)})

duration pydantic-field

duration: JqExpression = '.Duration'

jq expression selecting token duration

overlap pydantic-field

overlap: timedelta = timedelta(0)

Overlap between adjacent time blocks

pause_threshold pydantic-field

pause_threshold: timedelta | None = None

Silence threshold to split blocks

segment_duration pydantic-field

segment_duration: timedelta = timedelta(seconds=30)

Target duration of each time block

start pydantic-field

start: JqExpression = '.Offset'

jq expression selecting token start

text pydantic-field

text: JqExpression = '.Word'

jq expression selecting token text

tokens pydantic-field

tokens: JqExpression = '.metadata.word_timestamps[]?'

jq expression selecting timed tokens

units_per_second pydantic-field

units_per_second: PositiveInt = 10000000

Timestamp units per second

deduplication

Deduplication strategies for vector indexing.

Classes:

Name	Description
DeduplicationResult	Outcome of applying deduplication to a batch of items.
DeduplicationStrategy	Base class for item deduplication strategies.
DeterministicIdDeduplicationStrategy	Assigns stable IDs to items and skips IDs that already exist.
NoDeduplicationStrategy	Leaves the batch unchanged.
NormalizedTextHashDeduplicationStrategy	Deduplicates text items using a hash of normalized text.
ScopedDeduplicationStrategy	Deduplication strategy that can operate within a tenant or a source.
ShingleJaccardDeduplicationStrategy	Deduplicates text items using Jaccard similarity over token shingles.

DeduplicationResult pydantic-model

Bases: BaseModel

Outcome of applying deduplication to a batch of items.

Fields:

• kept_items (list[KnowledgeItem])
• input_count (int)
• dropped_count (int)
• strategy (str)
• scope (DeduplicationScope | None)

Source code in src/omni_ingest/agent/deduplication.py

class DeduplicationResult(BaseModel):
    """Outcome of applying deduplication to a batch of items."""

    kept_items: list[KnowledgeItem]
    input_count: int
    dropped_count: int
    strategy: str
    scope: DeduplicationScope | None = None

DeduplicationStrategy pydantic-model

Bases: BaseModel, ABC

Base class for item deduplication strategies.

Config:

• extra: forbid

Fields:

• kind (str)

Source code in src/omni_ingest/agent/deduplication.py

class DeduplicationStrategy(BaseModel, ABC):
    """Base class for item deduplication strategies."""

    model_config = ConfigDict(extra="forbid")
    kind: str

    @abstractmethod
    async def deduplicate(self, items: list[KnowledgeItem], ctx: IngestionContext[Any]) -> DeduplicationResult:
        """Returns the subset of items that should be indexed."""

deduplicate abstractmethod async

deduplicate(
    items: list[KnowledgeItem], ctx: IngestionContext[Any]
) -> DeduplicationResult

Returns the subset of items that should be indexed.

Source code in src/omni_ingest/agent/deduplication.py

@abstractmethod
async def deduplicate(self, items: list[KnowledgeItem], ctx: IngestionContext[Any]) -> DeduplicationResult:
    """Returns the subset of items that should be indexed."""

DeterministicIdDeduplicationStrategy pydantic-model

Bases: ScopedDeduplicationStrategy

Assigns stable IDs to items and skips IDs that already exist.

Fields:

• scope (DeduplicationScope)
• existing (JqExpression)
• kind (Literal['deterministic_id'])
• identity (JqExpression)

Source code in src/omni_ingest/agent/deduplication.py

class DeterministicIdDeduplicationStrategy(ScopedDeduplicationStrategy):
    """Assigns stable IDs to items and skips IDs that already exist."""

    kind: Literal["deterministic_id"] = "deterministic_id"
    identity: JqExpression = Field(default="[.tenant_id, .source_uri, .text]", description="jq expression producing stable item identity")

    async def deduplicate(self, items: list[KnowledgeItem], ctx: IngestionContext[Any]) -> DeduplicationResult:
        seen_ids = {v for item in await self._existing_items(ctx) if isinstance(v := item.metadata.get(_DETERMINISTIC_ID_KEY), str)}
        kept_items: list[KnowledgeItem] = []

        for item in items:
            identity = json.dumps(jq.compile(self.identity).input_value((await _item_document([item], ctx))["items"][0]).first(), sort_keys=True, separators=(",", ":"))
            deterministic_id = str(uuid5(NAMESPACE_URL, identity))
            item.id = uuid5(NAMESPACE_URL, identity)
            item.metadata[_DETERMINISTIC_ID_KEY] = deterministic_id
            if deterministic_id in seen_ids:
                continue
            kept_items.append(item)
            seen_ids.add(deterministic_id)

        return DeduplicationResult(
            kept_items=kept_items,
            input_count=len(items),
            dropped_count=len(items) - len(kept_items),
            strategy=self.kind,
            scope=self.scope,
        )

identity pydantic-field

identity: JqExpression = '[.tenant_id, .source_uri, .text]'

jq expression producing stable item identity

NoDeduplicationStrategy pydantic-model

Bases: DeduplicationStrategy

Leaves the batch unchanged.

Fields:

• kind (Literal['none'])

Source code in src/omni_ingest/agent/deduplication.py

class NoDeduplicationStrategy(DeduplicationStrategy):
    """Leaves the batch unchanged."""

    kind: Literal["none"] = "none"

    async def deduplicate(self, items: list[KnowledgeItem], ctx: IngestionContext[Any]) -> DeduplicationResult:
        del ctx
        return DeduplicationResult(kept_items=items, input_count=len(items), dropped_count=0, strategy=self.kind)

NormalizedTextHashDeduplicationStrategy pydantic-model

Bases: ScopedDeduplicationStrategy

Deduplicates text items using a hash of normalized text.

Fields:

• scope (DeduplicationScope)
• existing (JqExpression)
• kind (Literal['normalized_text_hash'])

Source code in src/omni_ingest/agent/deduplication.py

class NormalizedTextHashDeduplicationStrategy(ScopedDeduplicationStrategy):
    """Deduplicates text items using a hash of normalized text."""

    kind: Literal["normalized_text_hash"] = "normalized_text_hash"

    async def deduplicate(self, items: list[KnowledgeItem], ctx: IngestionContext[Any]) -> DeduplicationResult:
        seen_hashes = {
            _normalized_text_hash(await item.decode(ctx), cast(str | None, item.metadata.get(_NORMALIZED_TEXT_HASH_KEY)))
            for item in await self._existing_items(ctx)
        }
        kept_items: list[KnowledgeItem] = []

        for item in items:
            normalized_hash = _normalized_text_hash(await item.decode(ctx))
            item.metadata[_NORMALIZED_TEXT_HASH_KEY] = normalized_hash
            if normalized_hash in seen_hashes:
                continue
            kept_items.append(item)
            seen_hashes.add(normalized_hash)

        return DeduplicationResult(
            kept_items=kept_items,
            input_count=len(items),
            dropped_count=len(items) - len(kept_items),
            strategy=self.kind,
            scope=self.scope,
        )

ScopedDeduplicationStrategy pydantic-model

Bases: DeduplicationStrategy, ABC

Deduplication strategy that can operate within a tenant or a source.

Fields:

• kind (str)
• scope (DeduplicationScope)
• existing (JqExpression)

Source code in src/omni_ingest/agent/deduplication.py

class ScopedDeduplicationStrategy(DeduplicationStrategy, ABC):
    """Deduplication strategy that can operate within a tenant or a source."""

    scope: DeduplicationScope = Field(default="source", description="Scope used to compare an item against previously indexed items")
    existing: JqExpression = Field(default='.items[] | select(.metadata.vector_store != null)', description="jq expression selecting stored items for comparison")

    async def _existing_items(self, ctx: IngestionContext[ResolvedResource]) -> list[KnowledgeItem]:
        store = ctx.store
        if not isinstance(store, MetadataStore):
            return []

        if self.scope == "tenant":
            items = await store.list_knowledge_items(ctx.tenant_id)
        else:
            items = await store.list_knowledge_items(tenant_id=ctx.tenant_id, source_uri=ctx.resource.uri)
        item_by_id = {str(item.id): item for item in items}
        selected = jq.compile(self.existing).input_value(await _item_document(items, ctx)).all()
        return [item_by_id[item["id"]] for item in selected]

existing pydantic-field

existing: JqExpression = (
    ".items[] | select(.metadata.vector_store != null)"
)

jq expression selecting stored items for comparison

scope pydantic-field

scope: DeduplicationScope = 'source'

Scope used to compare an item against previously indexed items

ShingleJaccardDeduplicationStrategy pydantic-model

Bases: ScopedDeduplicationStrategy

Deduplicates text items using Jaccard similarity over token shingles.

Fields:

• scope (DeduplicationScope)
• existing (JqExpression)
• kind (Literal['shingle_jaccard'])
• threshold (float)
• shingle_size (PositiveInt)

Source code in src/omni_ingest/agent/deduplication.py

class ShingleJaccardDeduplicationStrategy(ScopedDeduplicationStrategy):
    """Deduplicates text items using Jaccard similarity over token shingles."""

    kind: Literal["shingle_jaccard"] = "shingle_jaccard"
    threshold: float = Field(default=0.9, ge=0.0, le=1.0, description="Minimum Jaccard similarity that counts as a duplicate")
    shingle_size: PositiveInt = Field(default=3, description="Token shingle size used for near-duplicate detection")

    async def deduplicate(self, items: list[KnowledgeItem], ctx: IngestionContext[Any]) -> DeduplicationResult:
        seen_shingles = [_shingles(await item.decode(ctx), self.shingle_size) for item in await self._existing_items(ctx)]
        kept_items: list[KnowledgeItem] = []

        for item in items:
            item_shingles = _shingles(await item.decode(ctx), self.shingle_size)
            item.metadata[_SHINGLE_SIZE_KEY] = self.shingle_size
            if any(_jaccard_similarity(item_shingles, existing) >= self.threshold for existing in seen_shingles):
                continue
            kept_items.append(item)
            seen_shingles.append(item_shingles)

        return DeduplicationResult(
            kept_items=kept_items,
            input_count=len(items),
            dropped_count=len(items) - len(kept_items),
            strategy=self.kind,
            scope=self.scope,
        )

shingle_size pydantic-field

shingle_size: PositiveInt = 3

Token shingle size used for near-duplicate detection

threshold pydantic-field

threshold: float = 0.9

Minimum Jaccard similarity that counts as a duplicate

document

Document agents.

Classes:

Name	Description
ImageExtractionAgent	Extract images and vector drawings from PDF items.
OcrAgent	Extracts items into Markdown using a configured OCR engine.
RasterizeAgent	Rasterizes PDF items into image-only PDFs.

ImageExtractionAgent pydantic-model

Bases: BaseModel, Step

Extract images and vector drawings from PDF items.

Fields:

• foreach (JqExpression)
• input (JqExpression)
• dpi (PositiveInt)
• padding (NonNegativeFloat)
• include_vector (bool)
• include_tables (bool)
• deduplicate (bool)
• replace (bool)
• mode (Literal['preserve', 'redact', 'caption'])
• workers (PositiveInt)

Source code in src/omni_ingest/agent/document.py

class ImageExtractionAgent(BaseModel, Step):
    """Extract images and vector drawings from PDF items."""

    foreach: JqExpression = Field(default=".items[]", description="jq expression selecting PDF items")
    input: JqExpression = Field(default="[.id]", description="jq expression selecting PDF item ids")
    dpi: PositiveInt = Field(default=240, description="DPI used to render extracted images")
    padding: NonNegativeFloat = Field(default=6, description="Padding around extracted images in PDF points")
    include_vector: bool = Field(default=True, description="Extract vector drawings")
    include_tables: bool = Field(default=False, description="Extract vector tables")
    deduplicate: bool = Field(default=True, description="Remove duplicate images from the same page")
    replace: bool = Field(default=False, description="Replace existing items with extracted images")
    mode: Literal["preserve", "redact", "caption"] = Field(default="preserve", description=(
        "How extracted images are represented in the parent PDF: preserve leaves the PDF unchanged, "
        "redact replaces each image with an `<image id=...>` marker, and caption keeps each image while "
        "adding its marker below it."
    ))
    workers: PositiveInt = Field(default=4, description="Maximum number of PDFs to process concurrently")

    async def run(self, ctx: IngestionContext[ResolvedResource]) -> StepResult:
        item_docs = [{"id": str(item.id), "metadata": item.metadata, "source_uri": item.source_uri} for item in ctx.items]
        item_by_id = {doc["id"]: item for doc, item in zip(item_docs, ctx.items, strict=True)}
        specs = jq.compile(self.foreach).input_value({"metadata": ctx.metadata, "items": item_docs}).all()
        input_expr = jq.compile(self.input)
        parents: list[KnowledgeItem] = []

        for spec in specs:
            item_ids = input_expr.input_value(spec).first()
            if not isinstance(item_ids, list) or not item_ids or any(item_id not in item_by_id for item_id in item_ids):
                raise ValueError("ImageExtractionAgent input must resolve to known item ids")
            parents.extend(item_by_id[item_id] for item_id in item_ids)

        sem, extracted = asyncio.Semaphore(self.workers), []
        ctx.progress(0, len(parents), "extracting images")
        pdf_parents = [(parent, await parent.content(ctx)) for parent in parents if await parent.content_type(ctx) == "application/pdf"]
        for i, task in enumerate(asyncio.as_completed([self._extract(parent, content, sem) for parent, content in pdf_parents]), 1):
            extracted.extend(await task)
            ctx.progress(i, len(parents), f"extracted {i}/{len(parents)}")

        ctx.items = extracted if self.replace else [*ctx.items, *extracted]
        return StepResult(status=StepStatus.SUCCESS, items=ctx.items, metadata={"count": len(extracted), "replace": self.replace})

    async def _extract(self, parent: KnowledgeItem, content: bytes, sem: asyncio.Semaphore) -> list[KnowledgeItem]:
        async with sem:
            return await asyncio.to_thread(self._extract_pdf, parent, content)

    def _extract_pdf(self, parent: KnowledgeItem, content: bytes) -> list[KnowledgeItem]:
        extracted = []
        with pymupdf.open(stream=content, filetype="pdf") as pdf:
            for page_number, page in enumerate(pdf.pages(), 1):
                assert isinstance(page, pymupdf.Page)

                regions = [pymupdf.Rect(image["bbox"]) for image in page.get_image_info()]
                tables = [] if self.include_tables or not self.include_vector else [pymupdf.Rect(table.bbox) for table in page.find_tables().tables]
                if self.include_vector:
                    regions.extend(rect for rect in page.cluster_drawings() if not any((rect & table).get_area() >= rect.get_area() * 0.8 for table in tables))

                seen = set()
                for rect in regions:
                    rect &= page.rect
                    area, page_area = rect.get_area(), page.rect.get_area()
                    if rect.is_empty or area < page_area * 0.001 or area > page_area * 0.8:
                        continue

                    rect = (rect + (-self.padding, -self.padding, self.padding, self.padding)) & page.rect
                    content = page.get_pixmap(clip=rect, dpi=self.dpi, alpha=False).tobytes("png")
                    if self.deduplicate and content in seen:
                        continue
                    seen.add(content)

                    id = uuid.uuid4()
                    marker = f'<image id="{id}"/>'
                    if self.mode == "redact":
                        page.add_redact_annot(rect, text=marker, fontsize=6, cross_out=False)
                    elif self.mode == "caption":
                        page.insert_text((rect.x0, rect.y1 + 7), marker, fontsize=6)

                    metadata = {"kind": "image", "parent_id": str(parent.id), "page": page_number, "bbox": [round(value, 3) for value in rect], "content_type": "image/png"}
                    extracted.append(KnowledgeItem(id=id, raw_content=content, content_encoding="binary", tenant_id=parent.tenant_id, source_uri=parent.source_uri, metadata=metadata))

                if self.mode == "redact":
                    page.apply_redactions()

            if self.mode != "preserve":
                parent.raw_content = pdf.tobytes(deflate=True)
                parent.content_uri = None

        return extracted

deduplicate pydantic-field

deduplicate: bool = True

Remove duplicate images from the same page

dpi pydantic-field

dpi: PositiveInt = 240

DPI used to render extracted images

foreach pydantic-field

foreach: JqExpression = '.items[]'

jq expression selecting PDF items

include_tables pydantic-field

include_tables: bool = False

Extract vector tables

include_vector pydantic-field

include_vector: bool = True

Extract vector drawings

input pydantic-field

input: JqExpression = '[.id]'

jq expression selecting PDF item ids

mode pydantic-field

mode: Literal['preserve', 'redact', 'caption'] = 'preserve'

How extracted images are represented in the parent PDF: preserve leaves the PDF unchanged, redact replaces each image with an <image id=...> marker, and caption keeps each image while adding its marker below it.

padding pydantic-field

padding: NonNegativeFloat = 6

Padding around extracted images in PDF points

replace pydantic-field

replace: bool = False

Replace existing items with extracted images

workers pydantic-field

workers: PositiveInt = 4

Maximum number of PDFs to process concurrently

OcrAgent pydantic-model

Bases: BaseModel, Step

Extracts items into Markdown using a configured OCR engine.

Fields:

• foreach (JqExpression)
• input (JqExpression)
• engine (str)
• src_lang (str | None)
• dst_lang (str | None)
• workers (PositiveInt)

Source code in src/omni_ingest/agent/document.py

class OcrAgent(BaseModel, Step):
    """Extracts items into Markdown using a configured OCR engine."""

    foreach: JqExpression = Field(default=".items[]", description="jq expression selecting items")
    input: JqExpression = Field(default="[.id]", description="jq expression selecting item ids")
    engine: str = Field(default=settings.default_ocr_engine, description="OCR engine name")
    src_lang: str | None = Field(default=None, description="Source language hint for OCR engines that support it")
    dst_lang: str | None = Field(default=None, description="Destination language hint for OCR engines that support it")
    workers: PositiveInt = Field(default=4, description="Maximum number of items OCR'd concurrently")

    async def run(self, ctx: IngestionContext[ResolvedResource]) -> StepResult:
        item_docs = [{"id": str(item.id), "metadata": item.metadata, "source_uri": item.source_uri} for item in ctx.items]
        item_by_id = {doc["id"]: item for doc, item in zip(item_docs, ctx.items, strict=True)}
        specs = jq.compile(self.foreach).input_value({"metadata": ctx.metadata, "items": item_docs}).all()
        input_expr = jq.compile(self.input)
        items: list[KnowledgeItem] = []

        for spec in specs:
            item_ids = input_expr.input_value(spec).first()
            if not isinstance(item_ids, list) or not item_ids or any(item_id not in item_by_id for item_id in item_ids):
                raise ValueError("OcrAgent input must resolve to known item ids")
            items.extend(item_by_id[item_id] for item_id in item_ids)

        ctx.progress(0, len(items), "extracting text")
        await self._extract_all(items, ctx, ctx.ocr_factory(self.engine, ctx, self.src_lang, self.dst_lang))

        return StepResult(status=StepStatus.SUCCESS, items=ctx.items, metadata={"count": len(items)})

    async def _extract_all(self, items: list[KnowledgeItem], ctx: IngestionContext[ResolvedResource], engine: Ocr) -> None:
        sem = asyncio.Semaphore(self.workers)

        async def process(item: KnowledgeItem) -> None:
            async with sem:
                content_type = await item.content_type(ctx)
                document = ByteContent(raw_content=await item.content(ctx), content_encoding=item.content_encoding, metadata={**item.metadata, "content_type": content_type})
                result = await engine(document)
            item.raw_content = await result.content(ctx)
            item.content_uri = None
            item.content_encoding = result.content_encoding
            item.metadata["content_type"] = await result.content_type(ctx)

        for i, task in enumerate(asyncio.as_completed([process(item) for item in items]), 1):
            await task
            ctx.progress(i, len(items), f"extracted {i}/{len(items)}")

dst_lang pydantic-field

dst_lang: str | None = None

Destination language hint for OCR engines that support it

engine pydantic-field

engine: str = default_ocr_engine

OCR engine name

foreach pydantic-field

foreach: JqExpression = '.items[]'

jq expression selecting items

input pydantic-field

input: JqExpression = '[.id]'

jq expression selecting item ids

src_lang pydantic-field

src_lang: str | None = None

Source language hint for OCR engines that support it

workers pydantic-field

workers: PositiveInt = 4

Maximum number of items OCR'd concurrently

RasterizeAgent pydantic-model

Bases: BaseModel, Step

Rasterizes PDF items into image-only PDFs.

Useful when a PDF has a corrupted or unreliable text layer and downstream vision models must read the visible page content instead.

Fields:

• foreach (JqExpression)
• input (JqExpression)
• dpi (PositiveInt)
• workers (PositiveInt)

Source code in src/omni_ingest/agent/document.py

class RasterizeAgent(BaseModel, Step):
    """Rasterizes PDF items into image-only PDFs.

    Useful when a PDF has a corrupted or unreliable text layer and downstream
    vision models must read the visible page content instead.
    """

    foreach: JqExpression = Field(default=".items[]", description="jq expression selecting PDF items")
    input: JqExpression = Field(default="[.id]", description="jq expression selecting PDF item ids")
    dpi: PositiveInt = Field(default=180, description="DPI used to render PDF pages")
    workers: PositiveInt = Field(default=4, description="Maximum number of PDFs to process concurrently")

    async def run(self, ctx: IngestionContext[ResolvedResource]) -> StepResult:
        item_docs = [{"id": str(item.id), "metadata": item.metadata, "source_uri": item.source_uri} for item in ctx.items]
        item_by_id = {doc["id"]: item for doc, item in zip(item_docs, ctx.items, strict=True)}
        specs = jq.compile(self.foreach).input_value({"metadata": ctx.metadata, "items": item_docs}).all()
        input_expr = jq.compile(self.input)
        items: list[KnowledgeItem] = []

        for spec in specs:
            item_ids = input_expr.input_value(spec).first()
            if not isinstance(item_ids, list) or not item_ids or any(item_id not in item_by_id for item_id in item_ids):
                raise ValueError("RasterizeAgent input must resolve to known item ids")
            items.extend(item_by_id[item_id] for item_id in item_ids)

        if any([await item.content_type(ctx) != "application/pdf" for item in items]):
            raise ValueError("RasterizeAgent only supports application/pdf items")

        sem = asyncio.Semaphore(self.workers)
        ctx.progress(0, len(items), "rasterizing PDFs")
        tasks = [self._rasterize(item, await item.content(ctx), sem) for item in items]
        for i, task in enumerate(asyncio.as_completed(tasks), 1):
            await task
            ctx.progress(i, len(items), f"rasterized {i}/{len(items)}")

        return StepResult(status=StepStatus.SUCCESS, items=ctx.items, metadata={"count": len(items), "dpi": self.dpi})

    async def _rasterize(self, item: KnowledgeItem, content: bytes, sem: asyncio.Semaphore) -> None:
        async with sem:
            await asyncio.to_thread(self._rasterize_pdf, item, content)

    def _rasterize_pdf(self, item: KnowledgeItem, content: bytes) -> None:
        with pymupdf.open(stream=content, filetype="pdf") as source, pymupdf.open() as output:
            for page in source:
                target = output.new_page(width=page.rect.width, height=page.rect.height)
                target.insert_image(target.rect, stream=page.get_pixmap(dpi=self.dpi, alpha=False).tobytes("png"))
            item.raw_content = output.tobytes(deflate=True)
            item.content_uri = None

dpi pydantic-field

dpi: PositiveInt = 180

DPI used to render PDF pages

foreach pydantic-field

foreach: JqExpression = '.items[]'

jq expression selecting PDF items

input pydantic-field

input: JqExpression = '[.id]'

jq expression selecting PDF item ids

workers pydantic-field

workers: PositiveInt = 4

Maximum number of PDFs to process concurrently

enrichment

Agents for structured metadata extraction and enrichment.

Classes:

Name	Description
ExtractAgent	Extracts structured output with an LLM and writes it to metadata.
ExtractAssertion
PromptAgent	Write model output to pipeline metadata.
TransformAgent	Applies jq transforms to root metadata.
TranslateAgent	Translates text or structured metadata in place.

ExtractAgent pydantic-model

Bases: BaseModel, Step, AgentMixin

Extracts structured output with an LLM and writes it to metadata.

Fields:

• model (KnownModelName | str | None)
• toolsets (list[str])
• retries (int | None)
• prompt (str)
• json_schema (dict[str, Any] | JqExpression)
• path (JqExpression)
• foreach (JqExpression)
• input (JqExpression)
• context (JqExpression | None)
• mode (Literal['direct', 'buffered'])
• assertions (list[ExtractAssertion])
• concurrency (PositiveInt)

Source code in src/omni_ingest/agent/enrichment.py

class ExtractAgent(BaseModel, Step, AgentMixin):
    """Extracts structured output with an LLM and writes it to metadata."""

    prompt: str = Field(..., description="Instruction prompt")
    json_schema: dict[str, Any] | JqExpression = Field(..., alias="schema", description="JSON schema or jq expression producing one")
    path: JqExpression = Field(..., description="jq path where extracted output is written")
    foreach: JqExpression = Field(default=".items[]", description="jq expression selecting documents to extract from")
    input: JqExpression = Field(default=".text", description="jq expression selecting prompt input")
    context: JqExpression | None = Field(default=None, description="jq expression selecting extra prompt context")
    mode: Literal["direct", "buffered"] = Field(default="direct", description="How selected input is supplied to the model")
    assertions: list[ExtractAssertion] = Field(default_factory=list, description="LLM retry assertions")
    concurrency: PositiveInt = Field(default=1, description="Maximum number of selected documents to extract concurrently")

    def _output_type(self, item: Any) -> type[BaseModel]:
        output = self.json_schema if isinstance(self.json_schema, dict) else jq.compile(self.json_schema).input_value(item).first()
        schema = {"type": "object", "properties": {"output": output}, "required": ["output"]}
        return generate_dynamic_models(schema, config=GenerateConfig(formatters=[]))["Model"]

    async def run(self, ctx: IngestionContext) -> StepResult:
        if not ctx.items:
            return StepResult(status=StepStatus.SKIPPED, error="No items for extraction")

        doc = await _doc(ctx, text=True)
        item_by_id = {str(item.id): item for item in ctx.items}
        selected_docs = jq.compile(self.foreach).input_value(doc).all()

        progress_total = len(selected_docs)
        if self.mode == "buffered":
            progress_total += sum(len(jq.compile(self.input).input_value(item).first()) for item in selected_docs)

        progress = 0
        def advance(message: str) -> None:
            nonlocal progress
            progress += 1
            ctx.progress(progress, progress_total, message)

        ctx.progress(0, progress_total, "extracting")
        sem = asyncio.Semaphore(self.concurrency)
        for done, task in enumerate(asyncio.as_completed([self._extract(ctx, item_doc, item_by_id, sem, advance) for item_doc in selected_docs]), 1):
            item_doc, output = await task
            if "id" in item_doc:
                item = item_by_id[item_doc["id"]]
                current_item_doc = {
                    "id": str(item.id),
                    "source_uri": item.source_uri,
                    "metadata": item.metadata,
                }
                item.metadata = _assign(current_item_doc, self.path, output)["metadata"]
            else:
                ctx.metadata = _assign({"metadata": ctx.metadata}, self.path, output)["metadata"]
            advance(f"extracted {done}/{len(selected_docs)}")

        return StepResult(status=StepStatus.SUCCESS, items=ctx.items, metadata={"items_processed": len(selected_docs), "path": self.path})

    async def _extract(self, ingestion_ctx: IngestionContext, item_doc: Any, item_by_id: dict[str, KnowledgeItem], sem: asyncio.Semaphore, advance: Callable[[str], None]) -> tuple[dict[str, Any], Any]:
        async with sem:
            item_doc = cast(dict[str, Any], item_doc)
            prompt_input = jq.compile(self.input).input_value(item_doc).first()
            item_ids: list[str] = prompt_input if isinstance(prompt_input, list) and all(item_id in item_by_id for item_id in prompt_input) else []
            prompt_context = jq.compile(self.context).input_value(item_doc).first() if self.context else None

            user_prompt: str | list[Any]
            if self.mode == "buffered":
                user_prompt = self.prompt
            elif item_ids:
                user_prompt = [
                    self.prompt,
                    *([prompt_context] if prompt_context is not None else []),
                    *[await _content(item_by_id[item_id], ingestion_ctx) for item_id in item_ids],
                ]
            else:
                user_prompt = [self.prompt, *([prompt_context] if prompt_context is not None else []), prompt_input]

            hooks = Hooks()

            @hooks.on.after_output_process
            async def validate_output(_, *, output_context, output):
                extracted = output.model_dump(mode="json")["output"]
                failures = [assertion.feedback for assertion in self.assertions if not jq.compile(assertion.expr, args={"item": item_doc}).input_value(extracted).first()]
                if failures:
                    raise ModelRetry("\n".join(failures))
                return output


            async def prepare_see_next(ctx: RunContext[ExtractAgentDeps], tool_def: ToolDefinition):
                return tool_def if self.mode == "buffered" and ctx.deps.cursor < len(ctx.deps.buffered_ids) else None


            agent = self._agent(ingestion_ctx, deps_type=ExtractAgentDeps, output_type=NativeOutput(self._output_type(item_doc), strict=True), capabilities=[hooks])

            @agent.tool(prepare=prepare_see_next, sequential=True)
            async def see_next(ctx: RunContext[ExtractAgentDeps]) -> list[Any]:
                if ctx.deps.cursor >= len(ctx.deps.buffered_ids):
                    return ["No items remain. Return the structured output now."]
                item = item_by_id[ctx.deps.buffered_ids[ctx.deps.cursor]]
                ctx.deps.cursor += 1
                advance(f"read {ctx.deps.cursor}/{len(ctx.deps.buffered_ids)}")
                content = [f"Item {ctx.deps.cursor} of {len(ctx.deps.buffered_ids)}"]
                if ctx.deps.cursor >= len(ctx.deps.buffered_ids):
                    content.append("This is the final item. You can no longer call this tool.")
                return [*content, await _content(item, ingestion_ctx)]

            async with agent:
                res = await agent.run(user_prompt, deps=ExtractAgentDeps(cursor=0, buffered_ids=tuple(item_ids)))
            return item_doc, res.output.model_dump(mode="json")["output"]

assertions pydantic-field

assertions: list[ExtractAssertion]

LLM retry assertions

concurrency pydantic-field

concurrency: PositiveInt = 1

Maximum number of selected documents to extract concurrently

context pydantic-field

context: JqExpression | None = None

jq expression selecting extra prompt context

foreach pydantic-field

foreach: JqExpression = '.items[]'

jq expression selecting documents to extract from

input pydantic-field

input: JqExpression = '.text'

jq expression selecting prompt input

json_schema pydantic-field

json_schema: dict[str, Any] | JqExpression

JSON schema or jq expression producing one

mode pydantic-field

mode: Literal['direct', 'buffered'] = 'direct'

How selected input is supplied to the model

path pydantic-field

path: JqExpression

jq path where extracted output is written

prompt pydantic-field

prompt: str

Instruction prompt

ExtractAssertion pydantic-model

Bases: BaseModel

Fields:

• expr (JqExpression)
• feedback (str)

Source code in src/omni_ingest/agent/enrichment.py

class ExtractAssertion(BaseModel):
    expr: JqExpression = Field(..., description="jq expression that must evaluate truthy against extracted output")
    feedback: str = Field(..., description="Feedback sent to the model when the assertion fails")

expr pydantic-field

expr: JqExpression

jq expression that must evaluate truthy against extracted output

feedback pydantic-field

feedback: str

Feedback sent to the model when the assertion fails

PromptAgent pydantic-model

Bases: BaseModel, Step, AgentMixin

Write model output to pipeline metadata.

Fields:

• model (KnownModelName | str | None)
• toolsets (list[str])
• retries (int | None)
• prompt (str)
• input (JqExpression)
• path (JqExpression)

Source code in src/omni_ingest/agent/enrichment.py

class PromptAgent(BaseModel, Step, AgentMixin):
    """Write model output to pipeline metadata."""

    prompt: str = Field(description="Instructions for rewriting content")
    input: JqExpression = Field(..., description="jq expression selecting prompt input")
    path: JqExpression = Field(..., description="jq path where model output is written")

    async def run(self, ctx: IngestionContext) -> StepResult:
        doc = await _doc(ctx, text=True)
        value = jq.compile(self.input).input_value(doc).first()
        async with self._agent(ctx) as agent:
            result = await agent.run([self.prompt, value])
        ctx.metadata = _assign(doc, self.path, result.output)["metadata"]
        return StepResult(status=StepStatus.SUCCESS)

input pydantic-field

input: JqExpression

jq expression selecting prompt input

path pydantic-field

path: JqExpression

jq path where model output is written

prompt pydantic-field

prompt: str

Instructions for rewriting content

TransformAgent pydantic-model

Bases: BaseModel, Step

Applies jq transforms to root metadata.

Fields:

• value (JqExpression)
• path (JqExpression)

Source code in src/omni_ingest/agent/enrichment.py

class TransformAgent(BaseModel, Step):
    """Applies jq transforms to root metadata."""

    value: JqExpression = Field(..., description="jq expression producing transformed output")
    path: JqExpression = Field(..., description="jq path where transformed output is written")

    async def run(self, ctx: IngestionContext) -> StepResult:
        doc = await _doc(ctx)
        ctx.metadata = _assign(doc, self.path, jq.compile(self.value).input_value(doc).first())["metadata"]
        return StepResult(status=StepStatus.SUCCESS, items=ctx.items, metadata={"path": self.path})

path pydantic-field

path: JqExpression

jq path where transformed output is written

value pydantic-field

value: JqExpression

jq expression producing transformed output

TranslateAgent pydantic-model

Bases: BaseModel, Step

Translates text or structured metadata in place.

Fields:

• src (str | None)
• dst (str)
• input (JqExpression)
• path (JqExpression)
• skip_fields (list[str])
• model (KnownTranslationModelName)

Source code in src/omni_ingest/agent/enrichment.py

class TranslateAgent(BaseModel, Step):
    """Translates text or structured metadata in place."""

    src: str | None = Field(..., description="Source language or null for automatic detection")
    dst: str = Field(..., description="Destination language")
    input: JqExpression = Field(..., description="jq expression selecting data to translate")
    path: JqExpression = Field(..., description="jq path where translated data is written")
    skip_fields: list[str] = Field(default_factory=list, description="Field path patterns excluded from translation")
    model: KnownTranslationModelName = Field(default=settings.default_translation_model, description="Translation model")

    async def run(self, ctx: IngestionContext) -> StepResult:
        doc = await _doc(ctx)
        async with create_translator(self.model) as translator:
            output = await translator.translate(self.src, self.dst, jq.compile(self.input).input_value(doc).first(), self.skip_fields)
        ctx.metadata = _assign(doc, self.path, output)["metadata"]
        return StepResult(status=StepStatus.SUCCESS, items=ctx.items, metadata={"path": self.path})

dst pydantic-field

dst: str

Destination language

input pydantic-field

input: JqExpression

jq expression selecting data to translate

model pydantic-field

model: KnownTranslationModelName = default_translation_model

Translation model

path pydantic-field

path: JqExpression

jq path where translated data is written

skip_fields pydantic-field

skip_fields: list[str]

Field path patterns excluded from translation

src pydantic-field

src: str | None

Source language or null for automatic detection

governance

Agents for pipeline validation, audit logging, and lineage metadata.

Classes:

Name	Description
DataLineageAgent	Maintains data provenance and lineage information for KnowledgeItems.
IngestionLoggerAgent	Records operational metadata for ingestion pipeline execution.

DataLineageAgent pydantic-model

Bases: BaseModel, Step

Maintains data provenance and lineage information for KnowledgeItems.

This agent attaches metadata documenting the origin and transformation history of each KnowledgeItem, ensuring traceability from the source data to the finalized index.

Source code in src/omni_ingest/agent/governance.py

class DataLineageAgent(BaseModel, Step):
    """
    Maintains data provenance and lineage information for KnowledgeItems.

    This agent attaches metadata documenting the origin and transformation
    history of each KnowledgeItem, ensuring traceability from the source
    data to the finalized index.
    """

    async def run(self, ctx: IngestionContext[ResolvedResource]) -> StepResult:
        lineage = {
            "run_id": str(ctx.run_id),
            "tenant_id": str(ctx.tenant_id),
            "source_uri": ctx.resource.uri,
            "items_count": len(ctx.items) if ctx.items else 0,
            "timestamp": str(datetime.now(UTC)),
        }
        return StepResult(status=StepStatus.SUCCESS, metadata=lineage)

IngestionLoggerAgent pydantic-model

Bases: BaseModel, Step

Records operational metadata for ingestion pipeline execution.

This agent logs audit information—including execution timestamps and item counts—to the metadata store, providing a record for monitoring and auditing purposes.

Source code in src/omni_ingest/agent/governance.py

class IngestionLoggerAgent(BaseModel, Step):
    """
    Records operational metadata for ingestion pipeline execution.

    This agent logs audit information—including execution timestamps
    and item counts—to the metadata store, providing a record for
    monitoring and auditing purposes.
    """

    async def run(self, ctx: IngestionContext) -> StepResult:
        if not ctx.store or not ctx.run_id:
            return StepResult(status=StepStatus.SKIPPED, error="No store/run context")

        audit_entry = {
            "event": "pipeline_checkpoint",
            "timestamp": str(datetime.now(UTC)),
            "items_in_context": len(ctx.items) if ctx.items else 0,
        }
        return StepResult(status=StepStatus.SUCCESS, metadata=audit_entry)

indexing

Agents for generating embeddings.

Classes:

Name	Description
EmbeddingAgent	Generates vector embeddings for KnowledgeItems to facilitate semantic search.

EmbeddingAgent pydantic-model

Bases: BaseModel, Step

Generates vector embeddings for KnowledgeItems to facilitate semantic search.

This agent converts textual content into numerical vector representations (embeddings). These vectors enable similarity-based retrieval operations within a vector database.

Fields:

• model (KnownEmbeddingModelName | str)
• foreach (JqExpression)
• input (JqExpression)
• path (JqExpression)

Source code in src/omni_ingest/agent/indexing.py

class EmbeddingAgent(BaseModel, Step):
    """
    Generates vector embeddings for KnowledgeItems to facilitate semantic search.

    This agent converts textual content into numerical vector representations
    (embeddings). These vectors enable similarity-based retrieval operations
    within a vector database.
    """

    model: KnownEmbeddingModelName | str = Field(default=settings.default_embedding_model, description="Model to use for embedding")
    foreach: JqExpression = Field(default=".items[]", description="jq expression selecting items to embed")
    input: JqExpression = Field(default=".text", description="jq expression selecting embedding input")
    path: JqExpression = Field(default=".metadata.embedding", description="jq path where embedding output is written")

    async def run(self, ctx: IngestionContext) -> StepResult:
        if not ctx.items:
            return StepResult(status=StepStatus.SKIPPED, error="No items to embed")

        doc = await _doc(ctx, text=True)
        item_by_id = {str(item.id): item for item in ctx.items}
        selected = jq.compile(self.foreach).input_value(doc).all()
        if not selected:
            return StepResult(status=StepStatus.SKIPPED, error="No items selected for embedding")
        values = [jq.compile(self.input).input_value(item).first() for item in selected]
        if any(not isinstance(value, str) for value in values):
            raise ValueError("Embedding input must resolve to text")
        res = await Embedder(self.model).embed_documents(values)
        for item_doc, embedding in zip(selected, res.embeddings, strict=True):
            item = item_by_id[item_doc["id"]]
            item.metadata = _assign({"metadata": item.metadata}, self.path, embedding)["metadata"]

        return StepResult(status=StepStatus.SUCCESS, items=ctx.items, metadata={"embedded_count": len(selected), "path": self.path})

foreach pydantic-field

foreach: JqExpression = '.items[]'

jq expression selecting items to embed

input pydantic-field

input: JqExpression = '.text'

jq expression selecting embedding input

model pydantic-field

model: KnownEmbeddingModelName | str = (
    default_embedding_model
)

Model to use for embedding

path pydantic-field

path: JqExpression = '.metadata.embedding'

jq path where embedding output is written

modality

Agents for modality transformation.

Classes:

Name	Description
TextAgent	Transforms incoming modality into textual content.

TextAgent pydantic-model

Bases: BaseModel, Step, AgentMixin

Transforms incoming modality into textual content.

This agent normalizes resources and existing KnowledgeItems to text. This is useful when downstream processing expects textual content. If the content is already text, it only records the encoding on the object.

Note that this agent infers content type using libmagic to check for text/ prefix. The output content may be of any valid text/ type, not necessarily text/plain.

Fields:

• model (KnownModelName | str | None)
• toolsets (list[str])
• retries (int | None)
• encoding (str)
• image_conversion_prompt (str)

Source code in src/omni_ingest/agent/modality.py

class TextAgent(BaseModel, Step, AgentMixin):
    """
    Transforms incoming modality into textual content.

    This agent normalizes resources and existing KnowledgeItems to text. This
    is useful when downstream processing expects textual content. If the content
    is already text, it only records the encoding on the object.

    Note that this agent infers content type using libmagic to check for `text/`
    prefix. The output content may be of any valid `text/` type, not necessarily
    `text/plain`.
    """

    encoding: str = Field(default="utf-8", description="Encoding used for decoded text bytes")

    image_conversion_prompt: str = Field(default=(
        "Convert the image into a complete, accessibility-focused text description for someone who cannot see it. "
        "Capture all visible text exactly, describe the layout, objects, people, actions, relationships, colors, symbols, charts, tables, UI elements, and any important visual context. "
        "Preserve reading order where possible. "
        "Be factual, avoid guessing, and clearly mark anything uncertain. "
        "The output should let a visually impaired person understand and reason about the image without needing to view it."
    ), description="The prompt to use when transforming image to text.")


    async def run(self, ctx: IngestionContext[ResolvedResource]) -> StepResult:
        for item in ctx.items:
            await self.convert(ctx, item, item.source_uri or ctx.resource.uri)
        return StepResult(status=StepStatus.SUCCESS, items=ctx.items, metadata={"content_types": [await item.content_type(ctx) for item in ctx.items]})


    async def convert(self, ctx: IngestionContext[ResolvedResource], obj: ByteContent, uri: str) -> None:
        if not (content_type := await obj.content_type(ctx)).startswith("text/"):
            text, metadata = await self.parse(ctx, uri, io.BytesIO(await obj.content(ctx)), encoding=self.encoding, content_type=content_type)
            obj.metadata.update(metadata)
            obj.metadata["content_type"] = "text/plain"
            obj.raw_content = text.encode(self.encoding)
            obj.content_uri = None
            if not (ct := await obj.content_type(ctx)).startswith("text/") and ct != "application/x-empty":
                raise ValueError(f"Could not transform content to text, resulting content-type was {ct}")
        obj.content_encoding = self.encoding


    async def parse(self, ctx: IngestionContext[ResolvedResource], path: str, stream: BinaryIO, *, content_type: str, encoding: str = "utf-8") -> tuple[str, dict[str, Any]]:
        if content_type.startswith("audio/"):
            content, metadata = await parse_audio(path, stream)
            metadata["selected_parser"] = "audio"
            return content, metadata
        if content_type.startswith("text/"):
            content, metadata = await parse_text(stream, encoding=encoding)
            metadata["selected_parser"] = "text"
            return content, metadata
        if content_type.startswith("application/"):
            content, metadata = await parse_document(stream, encoding=encoding)
            metadata["selected_parser"] = "document"
            return content, metadata
        if content_type.startswith("image/"):
            async with self._agent(ctx) as agent:
                result = await agent.run([BinaryContent(data=stream.read(), media_type=content_type)], instructions=self.image_conversion_prompt)
            return result.output, {"selected_parser": "image"}

        raise ValueError(f"Unsupported data URI MIME type: {content_type!r}")

encoding pydantic-field

encoding: str = 'utf-8'

Encoding used for decoded text bytes

image_conversion_prompt pydantic-field

image_conversion_prompt: str = (
    "Convert the image into a complete, accessibility-focused text description for someone who cannot see it. Capture all visible text exactly, describe the layout, objects, people, actions, relationships, colors, symbols, charts, tables, UI elements, and any important visual context. Preserve reading order where possible. Be factual, avoid guessing, and clearly mark anything uncertain. The output should let a visually impaired person understand and reason about the image without needing to view it."
)

The prompt to use when transforming image to text.

cli

Command-line entrypoint for running Omni Ingest pipeline YAML files.

core

Core logic and foundational data models for the ingestion framework.

Modules:

Name	Description
config	Configuration management and environment variable loading for the framework.
content	Content URI construction for virtual content.
event
model	Pydantic data models for knowledge items, contexts, and results.
ocr	OCR engine abstraction and factory wiring.
output	Output abstractions for finalized ingestion contexts.
pipeline	Orchestration logic for defining and executing ingestion pipelines.
protocol	Protocol definitions (Abstract Base Classes) for framework extensibility.

config

Configuration management and environment variable loading for the framework.

Classes:

Name	Description
Settings	Global application settings, loaded from environment variables or .env file.

Settings

Bases: BaseSettings

Global application settings, loaded from environment variables or .env file.

Attributes:

Name	Type	Description
azure_speech_key	SecretStr | None	Azure Speech SDK subscription key
azure_speech_region	str | None	Azure Speech SDK service region
azure_translation_key	SecretStr | None	Azure Translator subscription key
azure_translation_region	str | None	Azure Translator service region
default_chat_completion_model	KnownModelName | str	Default chat completion model to use during pipeline execution
default_embedding_model	KnownEmbeddingModelName | str	Default embedding completion model to use during pipeline execution
default_ocr_engine	str	Default OCR engine used by the ocr step, selected entirely through environment configuration
default_translation_model	KnownTranslationModelName	Default translation model to use during pipeline execution
documentintelligence_api_key	SecretStr | None	Azure Document Intelligence API key
documentintelligence_endpoint	AnyHttpUrl | None	Azure Document Intelligence endpoint
graph_store	GraphStoreConfig	Graph store configuration selected entirely through environment configuration
metadata_store	str	SQLAlchemy connection string
ocr_api_key	SecretStr | None	API key for OpenAI-compatible OCR
ocr_base_url	AnyHttpUrl | None	Base URL for OpenAI-compatible OCR
ocr_model	str	Model to use for OpenAI-compatible OCR
profiles_dir	Path	Directory containing reusable pipeline components
vector_store	VectorStoreConfig	Vector store configuration selected entirely through environment configuration

Source code in src/omni_ingest/core/config.py

class Settings(BaseSettings):
    """Global application settings, loaded from environment variables or .env file."""

    default_chat_completion_model: KnownModelName | str = "openai:gpt-5-mini"
    """Default chat completion model to use during pipeline execution"""

    default_embedding_model: KnownEmbeddingModelName | str = "openai:text-embedding-3-small"
    """Default embedding completion model to use during pipeline execution"""

    default_translation_model: KnownTranslationModelName = "azure"
    """Default translation model to use during pipeline execution"""

    metadata_store: str = "sqlite+aiosqlite:///./omni_ingest.db"
    """SQLAlchemy connection string"""

    profiles_dir: Path = Path("profiles")
    """Directory containing reusable pipeline components"""

    vector_store: VectorStoreConfig = Field(default_factory=InMemoryVectorStoreConfig)
    """Vector store configuration selected entirely through environment configuration"""

    graph_store: GraphStoreConfig = Field(default_factory=InMemoryGraphStoreConfig)
    """Graph store configuration selected entirely through environment configuration"""

    default_ocr_engine: str = "llm"
    """Default OCR engine used by the ocr step, selected entirely through environment configuration"""

    azure_speech_key: SecretStr | None = None
    """Azure Speech SDK subscription key"""

    azure_speech_region: str | None = None
    """Azure Speech SDK service region"""

    azure_translation_key: SecretStr | None = None
    """Azure Translator subscription key"""

    azure_translation_region: str | None = None
    """Azure Translator service region"""

    ocr_api_key: SecretStr | None = None
    """API key for OpenAI-compatible OCR"""

    ocr_base_url: AnyHttpUrl | None = None
    """Base URL for OpenAI-compatible OCR"""

    ocr_model: str = "gpt-4o-mini"
    """Model to use for OpenAI-compatible OCR"""

    documentintelligence_endpoint: AnyHttpUrl | None = None
    """Azure Document Intelligence endpoint"""

    documentintelligence_api_key: SecretStr | None = None
    """Azure Document Intelligence API key"""

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", env_nested_delimiter="__", extra="ignore")

azure_speech_key class-attribute instance-attribute

azure_speech_key: SecretStr | None = None

Azure Speech SDK subscription key

azure_speech_region class-attribute instance-attribute

azure_speech_region: str | None = None

Azure Speech SDK service region

azure_translation_key class-attribute instance-attribute

azure_translation_key: SecretStr | None = None

Azure Translator subscription key

azure_translation_region class-attribute instance-attribute

azure_translation_region: str | None = None

Azure Translator service region

default_chat_completion_model class-attribute instance-attribute

default_chat_completion_model: KnownModelName | str = (
    "openai:gpt-5-mini"
)

Default chat completion model to use during pipeline execution

default_embedding_model class-attribute instance-attribute

default_embedding_model: KnownEmbeddingModelName | str = (
    "openai:text-embedding-3-small"
)

Default embedding completion model to use during pipeline execution

default_ocr_engine class-attribute instance-attribute

default_ocr_engine: str = 'llm'

Default OCR engine used by the ocr step, selected entirely through environment configuration

default_translation_model class-attribute instance-attribute

default_translation_model: KnownTranslationModelName = (
    "azure"
)

Default translation model to use during pipeline execution

documentintelligence_api_key class-attribute instance-attribute

documentintelligence_api_key: SecretStr | None = None

Azure Document Intelligence API key

documentintelligence_endpoint class-attribute instance-attribute

documentintelligence_endpoint: AnyHttpUrl | None = None

Azure Document Intelligence endpoint

graph_store class-attribute instance-attribute

graph_store: GraphStoreConfig = Field(
    default_factory=InMemoryGraphStoreConfig
)

Graph store configuration selected entirely through environment configuration

metadata_store class-attribute instance-attribute

metadata_store: str = "sqlite+aiosqlite:///./omni_ingest.db"

SQLAlchemy connection string

ocr_api_key class-attribute instance-attribute

ocr_api_key: SecretStr | None = None

API key for OpenAI-compatible OCR

ocr_base_url class-attribute instance-attribute

ocr_base_url: AnyHttpUrl | None = None

Base URL for OpenAI-compatible OCR

ocr_model class-attribute instance-attribute

ocr_model: str = 'gpt-4o-mini'

Model to use for OpenAI-compatible OCR

profiles_dir class-attribute instance-attribute

profiles_dir: Path = Path('profiles')

Directory containing reusable pipeline components

vector_store class-attribute instance-attribute

vector_store: VectorStoreConfig = Field(
    default_factory=InMemoryVectorStoreConfig
)

Vector store configuration selected entirely through environment configuration

content

Content URI construction for virtual content.

event

Classes:

Name	Description
Event	Base class for all events.
PipelineBeginEvent	Event emitted at the start of a pipeline.
PipelineEndEvent	Event emitted at the end of a pipeline.
PipelineStepBeginEvent	Event emitted at the start of a pipeline step.
PipelineStepEndEvent	Event emitted at the end of a pipeline step.

Functions:

Name	Description
publish	Publishes an event to all subscribers.
shutdown	Shuts down the event system, notifying all subscribers.
subscribe	Subscribes to the event stream, yielding events as they are published.

Event dataclass

Base class for all events.

Attributes:

Name	Type	Description
identifier	str	Unique identifier for the event
metadata	dict[str, Any]	Additional event metadata
timestamp	datetime	Event creation timestamp

Source code in src/omni_ingest/core/event.py

@dataclass(kw_only=True)
class Event:
    """Base class for all events."""

    identifier: str
    """Unique identifier for the event"""

    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
    """Event creation timestamp"""

    metadata: dict[str, Any] = field(default_factory=dict)
    """Additional event metadata"""

identifier instance-attribute

identifier: str

Unique identifier for the event

metadata class-attribute instance-attribute

metadata: dict[str, Any] = field(default_factory=dict)

Additional event metadata

timestamp class-attribute instance-attribute

timestamp: datetime = field(
    default_factory=lambda: now(UTC)
)

Event creation timestamp

PipelineBeginEvent dataclass

Bases: Event, _PipelineMixin

Event emitted at the start of a pipeline.

Source code in src/omni_ingest/core/event.py

@dataclass(kw_only=True)
class PipelineBeginEvent(Event, _PipelineMixin):
    """Event emitted at the start of a pipeline."""
    identifier: str = field(default="pipeline_begin", init=False)
    completed_steps: list[Step] = field(default_factory=list)

PipelineEndEvent dataclass

Bases: Event, _PipelineMixin

Event emitted at the end of a pipeline.

Attributes:

Name	Type	Description
error	str | None	Human-readable error message if the pipeline failed
status	Literal['succeeded', 'failed']	Outcome of the pipeline

Source code in src/omni_ingest/core/event.py

@dataclass(kw_only=True)
class PipelineEndEvent(Event, _PipelineMixin):
    """Event emitted at the end of a pipeline."""
    identifier: str = field(default="pipeline_end", init=False)

    status: Literal["succeeded", "failed"]
    """Outcome of the pipeline"""

    error: str | None
    """Human-readable error message if the pipeline failed"""

error instance-attribute

error: str | None

Human-readable error message if the pipeline failed

status instance-attribute

status: Literal['succeeded', 'failed']

Outcome of the pipeline

PipelineStepBeginEvent dataclass

Bases: Event, _PipelineMixin, _StepMixin

Event emitted at the start of a pipeline step.

Source code in src/omni_ingest/core/event.py

@dataclass(kw_only=True)
class PipelineStepBeginEvent(Event, _PipelineMixin, _StepMixin):
    """Event emitted at the start of a pipeline step."""
    identifier: str = field(default="pipeline_step_begin", init=False)

PipelineStepEndEvent dataclass

Bases: Event, _PipelineMixin, _StepMixin

Event emitted at the end of a pipeline step.

Attributes:

Name	Type	Description
step_result	StepResult	Result of the step execution

Source code in src/omni_ingest/core/event.py

@dataclass(kw_only=True)
class PipelineStepEndEvent(Event, _PipelineMixin, _StepMixin):
    """Event emitted at the end of a pipeline step."""
    identifier: str = field(default="pipeline_step_end", init=False)

    step_result: StepResult
    """Result of the step execution"""

step_result instance-attribute

step_result: StepResult

Result of the step execution

publish

publish(event: Event)

Publishes an event to all subscribers.

Source code in src/omni_ingest/core/event.py

def publish(event: Event):
    """Publishes an event to all subscribers."""
    _buffer.append(event)
    for q in _subscribers:
        if q.full():
            q.get_nowait()
        q.put_nowait(event)

shutdown

shutdown()

Shuts down the event system, notifying all subscribers.

Source code in src/omni_ingest/core/event.py

def shutdown():
    """Shuts down the event system, notifying all subscribers."""
    for q in _subscribers:
        q.put_nowait(None)
    _subscribers.clear()
    _buffer.clear()

subscribe async

subscribe(
    populate_with_buffered: bool = True,
) -> AsyncIterator[Event]

Subscribes to the event stream, yielding events as they are published.

Parameters:

Name	Type	Description	Default
populate_with_buffered	bool	Whether to initialize this subscriber with buffered events.	True

Example

from omni_ingest.core.event import subscribe

async for event in subscribe():
    print(event)

Source code in src/omni_ingest/core/event.py

async def subscribe(populate_with_buffered: bool = True) -> AsyncIterator[Event]:
    """
    Subscribes to the event stream, yielding events as they are published.

    Args:
        populate_with_buffered: Whether to initialize this subscriber with buffered events.

    Example:
        ```python
        from omni_ingest.core.event import subscribe

        async for event in subscribe():
            print(event)
        ```
    """
    q: asyncio.Queue[Event | None] = asyncio.Queue(maxsize=32)
    if populate_with_buffered:
        for event in _buffer:
            q.put_nowait(event)

    _subscribers.add(q)
    try:
        while True:
            e = await q.get()
            if not e:
                break
            yield e
    finally:
        _subscribers.discard(q)

model

Pydantic data models for knowledge items, contexts, and results.

Classes:

Name	Description
AgentMixin
KnowledgeItem	Represents a discrete unit of information within the framework.
PipelineCheckpoint	State restored when a pipeline run resumes.
PipelineConfig	Configuration for a complete ingestion pipeline.
Step
StepConfig
StepResult	Result of an individual agentic step.
ValidationCheck	A jq validation attached to a pipeline step.

AgentMixin

Attributes:

Name	Type	Description
model	KnownModelName | str | None	Model to use for this step
retries	int | None	Per-category retry budgets for tools and output validation
toolsets	list[str]	Toolsets to enable for the agent

Source code in src/omni_ingest/core/model.py

class AgentMixin:
    model: KnownModelName | str | None = Field(default=None, description="Model to use for this step")
    toolsets: list[str] = Field(default_factory=list, description="Toolsets to enable for the agent")
    retries: int | None = Field(default=None, description="Per-category retry budgets for tools and output validation")

    @overload
    def _agent(self, ctx: IngestionContext[Any], deps_type: type[object] = object, output_type: OutputSpec[str] = str, capabilities: Sequence[AgentCapability[object]] | None = None) -> Agent[object, str]: ...

    @overload
    def _agent(self, ctx: IngestionContext[Any], deps_type: type[DepsT], output_type: OutputSpec[OutputT], capabilities: Sequence[AgentCapability[DepsT]] | None = None) -> Agent[DepsT, OutputT]: ...

    @final
    def _agent(self, ctx: IngestionContext[Any], deps_type: type[Any] = object, output_type: OutputSpec[Any] = str, capabilities: Sequence[AgentCapability[Any]] | None = None) -> Agent[Any, Any]:
        from ..core.pipeline import PIPELINE_RUNNERS
        tool_vendors = PIPELINE_RUNNERS[ctx.run_id].toolsets
        missing = [ts for ts in self.toolsets if ts not in tool_vendors]
        if missing:
            raise ValueError(f"Toolsets not configured in context: {missing}")
        model = self.model or settings.default_chat_completion_model
        toolsets = [tool_vendors[k] for k in self.toolsets]
        return Agent(model=model, toolsets=toolsets, retries=self.retries, deps_type=deps_type, output_type=output_type, capabilities=capabilities)

model pydantic-field

model: KnownModelName | str | None = None

Model to use for this step

retries pydantic-field

retries: int | None = None

Per-category retry budgets for tools and output validation

toolsets pydantic-field

toolsets: list[str]

Toolsets to enable for the agent

KnowledgeItem pydantic-model

Bases: ByteContent

Represents a discrete unit of information within the framework.

A KnowledgeItem serves as a container for processed data, such as text segments, audio transcripts, or structured metadata. It is the primary entity processed and transformed throughout the ingestion pipeline.

Config:

• from_attributes: True
• populate_by_name: True

Fields:

• content_uri (str | None)
• content_encoding (str)
• raw_content (StrictBytes)
• id (UUID)
• tenant_id (UUID | None)
• source_uri (str | None)
• metadata (dict[str, Any])

Source code in src/omni_ingest/core/model.py

class KnowledgeItem(ByteContent):
    """
    Represents a discrete unit of information within the framework.

    A KnowledgeItem serves as a container for processed data, such as
    text segments, audio transcripts, or structured metadata. It is the primary
    entity processed and transformed throughout the ingestion pipeline.
    """

    model_config = ConfigDict(from_attributes=True, populate_by_name=True)
    raw_content: StrictBytes = Field(default=b"", alias="content")
    id: UUID = Field(default_factory=uuid4, description="Unique identifier for the knowledge item")
    tenant_id: UUID | None = Field(default=None, description="Identifier for the associated tenant or organization")
    source_uri: str | None = Field(default=None, description="Origin URI of the raw data")
    metadata: dict[str, Any] = Field(default_factory=dict, alias="metadata_", description="Arbitrary key-value pairs for additional contextual information")

id pydantic-field

id: UUID

Unique identifier for the knowledge item

metadata pydantic-field

metadata: dict[str, Any]

Arbitrary key-value pairs for additional contextual information

source_uri pydantic-field

source_uri: str | None = None

Origin URI of the raw data

tenant_id pydantic-field

tenant_id: UUID | None = None

Identifier for the associated tenant or organization

PipelineCheckpoint pydantic-model

Bases: BaseModel

State restored when a pipeline run resumes.

Fields:

• stage (int)
• resource (ResolvedResource)
• tenant_id (UUID)
• domain_profile (str)
• metadata (dict[str, Any])
• items (list[KnowledgeItem])

Source code in src/omni_ingest/core/model.py

class PipelineCheckpoint(BaseModel):
    """State restored when a pipeline run resumes."""

    stage: int = Field(description="Number of completed stages")
    resource: ResolvedResource = Field(description="Resolved input resource")
    tenant_id: UUID = Field(description="Tenant that owns the pipeline run")
    domain_profile: str = Field(description="Domain profile used by the pipeline")
    metadata: dict[str, Any] = Field(default_factory=dict, description="Shared pipeline metadata")
    items: list[KnowledgeItem] = Field(default_factory=list, description="Knowledge items produced by completed stages")

domain_profile pydantic-field

domain_profile: str

Domain profile used by the pipeline

items pydantic-field

items: list[KnowledgeItem]

Knowledge items produced by completed stages

metadata pydantic-field

metadata: dict[str, Any]

Shared pipeline metadata

resource pydantic-field

resource: ResolvedResource

Resolved input resource

stage pydantic-field

stage: int

Number of completed stages

tenant_id pydantic-field

tenant_id: UUID

Tenant that owns the pipeline run

PipelineConfig pydantic-model

Bases: BaseModel

Configuration for a complete ingestion pipeline.

Fields:

• pipeline_id (str | None)
• domain_profile (str)
• input_modality (str | None)
• description (str | None)
• parameters (dict[str, Any])
• steps (list[StepConfig])
• tool_vendors (dict[str, MCPServerTypes])

Source code in src/omni_ingest/core/model.py

class PipelineConfig(BaseModel):
    """Configuration for a complete ingestion pipeline."""

    pipeline_id: str | None = Field(default=None, description="Identifier for the pipeline template")
    domain_profile: str = Field(default="default", description="Domain profile name")
    input_modality: str | None = Field(default=None, description="Expected input modality", examples=["text", "document", "audio"])
    description: str | None = Field(default=None, description="Purpose of this pipeline")
    parameters: dict[str, Any] = Field(default_factory=dict, description="JSON schema for configurable pipeline parameters")
    steps: list[StepConfig] = Field(..., description="Agentic steps partitioned into ordered execution stages")
    tool_vendors: dict[str, MCPServerTypes] = Field(default_factory=dict, description="Configured tool vendors available for agents")

description pydantic-field

description: str | None = None

Purpose of this pipeline

domain_profile pydantic-field

domain_profile: str = 'default'

Domain profile name

input_modality pydantic-field

input_modality: str | None = None

Expected input modality

parameters pydantic-field

parameters: dict[str, Any]

JSON schema for configurable pipeline parameters

pipeline_id pydantic-field

pipeline_id: str | None = None

Identifier for the pipeline template

steps pydantic-field

steps: list[StepConfig]

Agentic steps partitioned into ordered execution stages

tool_vendors pydantic-field

tool_vendors: dict[str, MCPServerTypes]

Configured tool vendors available for agents

Step

Bases: ABC

Methods:

Name	Description
run	Executes the specific logic for this ingestion step.

Source code in src/omni_ingest/core/model.py

class Step(ABC):
    @abstractmethod
    async def run(self, ctx: IngestionContext[ResolvedResource]) -> StepResult:
        """Executes the specific logic for this ingestion step."""
        ...

run abstractmethod async

run(ctx: IngestionContext[ResolvedResource]) -> StepResult

Executes the specific logic for this ingestion step.

Source code in src/omni_ingest/core/model.py

@abstractmethod
async def run(self, ctx: IngestionContext[ResolvedResource]) -> StepResult:
    """Executes the specific logic for this ingestion step."""
    ...

StepConfig pydantic-model

Bases: BaseModel

Fields:

• agent (str)
• description (str | None)
• stage (str | None)
• config (dict[str, Any])
• validation (list[ValidationCheck])

Source code in src/omni_ingest/core/model.py

class StepConfig(BaseModel):
    agent: str = Field(..., description="The registry name of the agent to execute")
    description: str | None = Field(default=None, description="Human-readable step label")
    stage: str | None = Field(default=None, description="Execution stage shared by concurrently running steps")
    config: dict[str, Any] = Field(default_factory=dict, description="Step-specific configuration parameters")
    validation: list[ValidationCheck] = Field(default_factory=list, description="Validation checks for this configured step")

agent pydantic-field

agent: str

The registry name of the agent to execute

config pydantic-field

config: dict[str, Any]

Step-specific configuration parameters

description pydantic-field

description: str | None = None

Human-readable step label

stage pydantic-field

stage: str | None = None

Execution stage shared by concurrently running steps

validation pydantic-field

validation: list[ValidationCheck]

Validation checks for this configured step

StepResult pydantic-model

Bases: BaseModel

Result of an individual agentic step.

Fields:

• status (StepStatus)
• output_paths (dict[str, str])
• error (str | None)
• metadata (dict[str, Any])
• items (list[KnowledgeItem])

Source code in src/omni_ingest/core/model.py

class StepResult(BaseModel):
    """Result of an individual agentic step."""

    status: StepStatus = Field(..., description="The completion status of the step")
    output_paths: dict[str, str] = Field(default_factory=dict, description="Paths to any artifacts generated by the step")
    error: str | None = Field(default=None, description="Error message if the step failed")
    metadata: dict[str, Any] = Field(default_factory=dict, description="Step-specific execution metadata")
    items: list[KnowledgeItem] = Field(default_factory=list, description="Knowledge items produced or modified by the step")

error pydantic-field

error: str | None = None

Error message if the step failed

items pydantic-field

items: list[KnowledgeItem]

Knowledge items produced or modified by the step

metadata pydantic-field

metadata: dict[str, Any]

Step-specific execution metadata

output_paths pydantic-field

output_paths: dict[str, str]

Paths to any artifacts generated by the step

status pydantic-field

status: StepStatus

The completion status of the step

ValidationCheck pydantic-model

Bases: BaseModel

A jq validation attached to a pipeline step.

condition checks skip the step when false. before checks run before the step. after checks run after a successful step and can inspect .result. plain checks run before the step, but when they pass, they stop the remaining validation chain for that step.

Fields:

• mode (ValidationMode)
• expr (str)
• message (str)

Source code in src/omni_ingest/core/model.py

class ValidationCheck(BaseModel):
    """
    A jq validation attached to a pipeline step.

    `condition` checks skip the step when false. `before` checks run before the
    step. `after` checks run after a successful step and can inspect `.result`.
    `plain` checks run before the step, but when they pass, they stop the remaining
    validation chain for that step.
    """

    mode: ValidationMode = Field(..., description="When the validation runs relative to the step")
    expr: str = Field(..., description="jq expression that must evaluate truthy")
    message: str = Field(..., description="Error message when the check fails")

expr pydantic-field

expr: str

jq expression that must evaluate truthy

message pydantic-field

message: str

Error message when the check fails

mode pydantic-field

mode: ValidationMode

When the validation runs relative to the step

ocr

OCR engine abstraction and factory wiring.

Functions:

Name	Description
build_simple_factory	Build an OcrFactory that dispatches by engine name to a fixed mapping.

build_simple_factory

build_simple_factory(
    mapping: dict[str, OcrEngineBuilder],
) -> OcrFactory

Build an OcrFactory that dispatches by engine name to a fixed mapping.

Source code in src/omni_ingest/core/ocr.py

def build_simple_factory(mapping: dict[str, OcrEngineBuilder]) -> OcrFactory:
    """Build an `OcrFactory` that dispatches by engine name to a fixed mapping."""

    def factory(engine: str, ctx: IngestionContext[ResolvedResource], src_lang: str | None, dst_lang: str | None) -> Ocr:
        if engine not in mapping:
            raise ValueError(f"Unknown OCR engine: {engine}")
        return mapping[engine](ctx, src_lang, dst_lang)

    return factory

output

Output abstractions for finalized ingestion contexts.

Classes:

Name	Description
GraphStoreOutput	Writes finalized graph data to a graph store.
JsonFileOutput	Writes finalized ingestion output to a JSON file.
MongoDocumentOutput	Writes KnowledgeItems as flat BSON documents to a plain MongoDB collection.
PickleFileOutput	Writes finalized ingestion output to a pickle file.
StdoutOutput	Writes finalized ingestion output to stdout as JSON.
VectorStoreOutput	Writes finalized knowledge items to a vector store.

GraphStoreOutput

Bases: Output

Writes finalized graph data to a graph store.

Attributes:

Name	Type	Description
store		The GraphStore implementation to use

Source code in src/omni_ingest/core/output.py

class GraphStoreOutput(Output):
    """Writes finalized graph data to a graph store.

    Attributes:
        store: The GraphStore implementation to use
    """

    def __init__(self, store: GraphStore) -> None:
        self.store = store


    async def write(self, ctx: IngestionContext[ResolvedResource]) -> OutputResult:
        kg_data = ctx.metadata.get("kg_data")
        if not kg_data or not ctx.items:
            return OutputResult(status=StepStatus.SKIPPED, error="No KG data found in context")

        if nodes := kg_data.get("nodes"):
            await self.store.upsert_nodes(nodes)
        if edges := kg_data.get("edges"):
            await self.store.upsert_edges(edges)
        return OutputResult(status=StepStatus.SUCCESS, metadata={"graph_nodes": len(kg_data.get("nodes", []))})

JsonFileOutput

Bases: FileOutput

Writes finalized ingestion output to a JSON file.

Source code in src/omni_ingest/core/output.py

class JsonFileOutput(FileOutput):
    """Writes finalized ingestion output to a JSON file."""

    async def dump(self, ctxs: list[IngestionContext[ResolvedResource]]) -> str:
        data = await json_dump_context(ctxs[0]) if len(ctxs) == 1 else {"runs": [await json_dump_context(ctx) for ctx in ctxs]}
        return json.dumps(data, indent=2)

    async def write(self, ctx: IngestionContext[ResolvedResource]) -> OutputResult:
        await asyncio.to_thread(self._write, await self.dump([ctx]), mode="wt")
        return OutputResult(status=StepStatus.SUCCESS, output_paths={"json": self.path})

    async def write_many(self, ctxs: list[IngestionContext[ResolvedResource]]) -> OutputResult:
        await asyncio.to_thread(self._write, await self.dump(ctxs), mode="wt")
        return OutputResult(status=StepStatus.SUCCESS, output_paths={"json": self.path})

MongoDocumentOutput

Bases: Output

Writes KnowledgeItems as flat BSON documents to a plain MongoDB collection.

URI format

mongodb://host:port/database/collection

Source code in src/omni_ingest/core/output.py

class MongoDocumentOutput(Output):
    """Writes KnowledgeItems as flat BSON documents to a plain MongoDB collection.

    URI format:
      mongodb://host:port/database/collection
    """

    def __init__(self, uri: str) -> None:
        from urllib.parse import urlparse, urlunparse
        parsed = urlparse(uri)
        path_parts = [p for p in parsed.path.split("/") if p]
        if len(path_parts) < 2:
            raise ValueError(
                f"MongoDB output URI must include database and collection in the path: "
                f"mongodb://host/database/collection — got: {uri}"
            )
        client_uri = urlunparse(parsed._replace(path=f"/{path_parts[0]}"))

        try:
            from pymongo import AsyncMongoClient, ReplaceOne
            from pymongo.uri_parser import parse_uri
            self._ReplaceOne = ReplaceOne
        except ModuleNotFoundError as exc:
            raise ImportError(
                "MongoDocumentOutput requires pymongo. Install omni-ingest[storage] or add pymongo to the environment."
            ) from exc

        if parse_uri(client_uri)["database"] is None:
            raise ValueError(f"MongoDB output URI must include a database name — got: {uri}")
        self._collection: Any = AsyncMongoClient(client_uri).get_default_database()[path_parts[1]]

    async def write(self, ctx: IngestionContext[ResolvedResource]) -> OutputResult:
        if not ctx.items:
            return OutputResult(status=StepStatus.SKIPPED, error="No items")

        ops = [
            self._ReplaceOne(
                {"_id": str(item.id)},
                {"_id": str(item.id), "source_uri": item.source_uri, "content": await item.decode(ctx), "metadata": item.metadata},
                upsert=True,
            )
            for item in ctx.items
        ]
        await self._collection.bulk_write(ops, ordered=False)
        return OutputResult(status=StepStatus.SUCCESS, metadata={"count": len(ctx.items)})

PickleFileOutput

Bases: FileOutput

Writes finalized ingestion output to a pickle file.

Source code in src/omni_ingest/core/output.py

class PickleFileOutput(FileOutput):
    """Writes finalized ingestion output to a pickle file."""

    async def write_many(self, ctxs: list[IngestionContext[ResolvedResource]]) -> OutputResult:
        runs: list[dict[str, Any]] = [{
            "resource": {**ctx.resource.model_dump(mode="python", exclude={"raw_content"}), "content": await ctx.resource.content(ctx)},
            "metadata": ctx.metadata,
            "items": [{**item.model_dump(mode="python", exclude={"raw_content"}), "content": await item.content(ctx)} for item in ctx.items],
        } for ctx in ctxs]
        data = runs[0] if len(runs) == 1 else {"runs": runs}
        await asyncio.to_thread(self._write, pickle.dumps(data), mode="wb")
        return OutputResult(status=StepStatus.SUCCESS, output_paths={"pkl": self.path})

    async def write(self, ctx: IngestionContext[ResolvedResource]) -> OutputResult:
        return await self.write_many([ctx])

StdoutOutput

Bases: Output

Writes finalized ingestion output to stdout as JSON.

Source code in src/omni_ingest/core/output.py

class StdoutOutput(Output):
    """Writes finalized ingestion output to stdout as JSON."""

    def __init__(self) -> None:
        self._inner = JsonFileOutput(os.devnull)

    async def write(self, ctx: IngestionContext[ResolvedResource]) -> OutputResult:
        print(await self._inner.dump([ctx]), file=sys.stdout)
        return OutputResult(status=StepStatus.SUCCESS)

    async def write_many(self, ctxs: list[IngestionContext[ResolvedResource]]) -> OutputResult:
        print(await self._inner.dump(ctxs), file=sys.stdout)
        return OutputResult(status=StepStatus.SUCCESS)

VectorStoreOutput

Bases: Output

Writes finalized knowledge items to a vector store.

Attributes:

Name	Type	Description
store		The physical database where data will be saved
deduplication_strategy		Strategy used to skip duplicate items before indexing

Source code in src/omni_ingest/core/output.py

class VectorStoreOutput(Output):
    """Writes finalized knowledge items to a vector store.

    Attributes:
        store: The physical database where data will be saved
        deduplication_strategy: Strategy used to skip duplicate items before indexing
    """

    def __init__(self, store: VectorStore, deduplication_strategy: DeduplicationStrategyConfig | None = None) -> None:
        self.store = store
        self.deduplication_strategy = deduplication_strategy or NormalizedTextHashDeduplicationStrategy()


    async def write(self, ctx: IngestionContext[ResolvedResource]) -> OutputResult:
        if not ctx.items:
            return OutputResult(status=StepStatus.SKIPPED, error="No items")

        deduplication_result = await self.deduplication_strategy.deduplicate(ctx.items, ctx)
        items = deduplication_result.kept_items
        if items:
            await self.store.upsert(items, ctx)
            if ctx.store and (pid := ctx.metadata.get("knowledge_item_id")):
                for item in items:
                    metadata = {
                        **item.metadata,
                        "kind": "indexed",
                        "parent_knowledge_item_id": str(pid),
                        "vector_store": {
                            "kind": self.store.kind,
                            "space": self.store.space,
                            "ref_id": str(item.id),
                        },
                    }
                    await ctx.store.create_knowledge_item(
                        tenant_id=ctx.tenant_id,
                        domain_profile=ctx.domain_profile,
                        modality="text",
                        source_uri=item.source_uri or ctx.resource.uri,
                        content=await item.content(ctx),
                        metadata=metadata,
                    )

        return OutputResult(
            status=StepStatus.SUCCESS,
            metadata={
                "count": len(items),
                "input_count": deduplication_result.input_count,
                "deduplicated_count": deduplication_result.dropped_count,
                "deduplication_strategy": deduplication_result.strategy,
                "deduplication_scope": deduplication_result.scope,
            },
        )

pipeline

Orchestration logic for defining and executing ingestion pipelines.

Classes:

Name	Description
IngestionContext	Maintains state and shared metadata across a single ingestion pipeline execution.
KnowledgeBaseAdapter	Provides a high-level interface for querying and retrieving ingested knowledge.
PipelineRunner	Orchestrates the execution of data processing steps within an ingestion pipeline.

Functions:

Name	Description
create_pipeline_from_config	Factory function to create a PipelineRunner from a YAML configuration file.
register_step	Register a Step class in the global factory.

IngestionContext dataclass

Bases: Generic[ResourceT]

Maintains state and shared metadata across a single ingestion pipeline execution.

The IngestionContext object facilitates communication between sequential processing steps by carrying operational state, configuration parameters, and the current set of KnowledgeItems.

Attributes:

Name	Type	Description
content_resolver	ContentResolver	Reference to a storage engine or backend
domain_profile	str	The configuration profile being executed
items	list[KnowledgeItem]	The working set of knowledge items
metadata	dict[str, Any]	Shared state or context across steps
ocr_factory	OcrFactory	OCR factory to vend OCR engines from
run_id	UUID	Unique identifier for the current pipeline run
store	MetadataStore	Reference to a storage engine or backend
tenant_id	UUID	Identifier for the active tenant

Source code in src/omni_ingest/core/pipeline.py

@dataclass(kw_only=True)
class IngestionContext(Generic[ResourceT]):  # noqa: UP046
    """
    Maintains state and shared metadata across a single ingestion pipeline execution.

    The IngestionContext object facilitates communication between sequential
    processing steps by carrying operational state, configuration parameters,
    and the current set of KnowledgeItems.
    """

    resource: ResourceT

    store: MetadataStore = field(default_factory=default_metadata_store)
    """Reference to a storage engine or backend"""

    content_resolver: ContentResolver = field(default_factory=default_content_resolver)
    """Reference to a storage engine or backend"""

    ocr_factory: OcrFactory = field(default_factory=default_ocr_factory)
    """OCR factory to vend OCR engines from"""

    run_id: UUID = field(default_factory=uuid4)
    """Unique identifier for the current pipeline run"""

    tenant_id: UUID = field(default_factory=uuid4)
    """Identifier for the active tenant"""

    domain_profile: str = "default"
    """The configuration profile being executed"""

    metadata: dict[str, Any] = field(default_factory=dict)
    """Shared state or context across steps"""

    items: list[KnowledgeItem] = field(default_factory=list)
    """The working set of knowledge items"""

    def progress(self, completed: float, total: float, message: str | None = None) -> None:
        if current := _CURRENT_STEP.get():
            step_name, step = current
            event.publish(event.PipelineStepProgressEvent(run_id=self.run_id, tenant_id=self.tenant_id, step_name=step_name, step=step, completed=completed, total=total, message=message))

content_resolver class-attribute instance-attribute

content_resolver: ContentResolver = field(
    default_factory=default_content_resolver
)

Reference to a storage engine or backend

domain_profile class-attribute instance-attribute

domain_profile: str = 'default'

The configuration profile being executed

items class-attribute instance-attribute

items: list[KnowledgeItem] = field(default_factory=list)

The working set of knowledge items

metadata class-attribute instance-attribute

metadata: dict[str, Any] = field(default_factory=dict)

Shared state or context across steps

ocr_factory class-attribute instance-attribute

ocr_factory: OcrFactory = field(
    default_factory=default_factory
)

OCR factory to vend OCR engines from

run_id class-attribute instance-attribute

run_id: UUID = field(default_factory=uuid4)

Unique identifier for the current pipeline run

store class-attribute instance-attribute

store: MetadataStore = field(
    default_factory=default_metadata_store
)

Reference to a storage engine or backend

tenant_id class-attribute instance-attribute

tenant_id: UUID = field(default_factory=uuid4)

Identifier for the active tenant

KnowledgeBaseAdapter

Provides a high-level interface for querying and retrieving ingested knowledge.

The adapter abstracts the underlying storage mechanisms, such as vector databases and metadata stores, allowing for unified search and retrieval operations across different data domain profiles.

Source code in src/omni_ingest/core/pipeline.py

class KnowledgeBaseAdapter:
    """
    Provides a high-level interface for querying and retrieving ingested knowledge.

    The adapter abstracts the underlying storage mechanisms, such as vector
    databases and metadata stores, allowing for unified search and retrieval
    operations across different data domain profiles.
    """

    from .protocol import MetadataStore

    def __init__(self, metadata_store: MetadataStore, vector_store: VectorStore, embedder: Embedder | None = None):
        self.metadata_store = metadata_store
        self.vector_store = vector_store
        self.embedder = embedder or Embedder(settings.default_embedding_model)

    async def list_items(self, tenant_id: UUID, domain_profile: str | None = None, **filters) -> list[KnowledgeItem]:
        return await self.metadata_store.list_knowledge_items(tenant_id=tenant_id, domain_profile=domain_profile, **filters)

    async def query(self, tenant_id: UUID, query: list[float], k: PositiveInt = 5, domain_profile: str | None = None) -> list[KnowledgeItem]:
        return await self.vector_store.search(query, k=k, tenant_id=tenant_id)

    async def get(self, item_id: UUID) -> KnowledgeItem | None:
        return await self.metadata_store.get_knowledge_item(item_id)

    async def query_text(self, tenant_id: UUID, query: str, k: PositiveInt = 5, domain_profile: str | None = None) -> list[KnowledgeItem]:
        try:
            res = await self.embedder.embed_query(query)
            emb = list(res.embeddings[0])
        except Exception as e:
            raise RuntimeError(f"Embedding failed: {e}") from e
        return await self.query(tenant_id, emb, k=k, domain_profile=domain_profile)

PipelineRunner

Orchestrates the execution of data processing steps within an ingestion pipeline.

The PipelineRunner is responsible for initializing the IngestionContext, executing ordered stages of configured steps (agents), and managing the overall lifecycle of the ingestion process, including error handling and status reporting. Steps in a stage run concurrently against the same ingestion context.

Methods:

Name	Description
resume	Resumes a pipeline run from its latest successful-stage checkpoint.
run	Executes the ingestion pipeline.

Source code in src/omni_ingest/core/pipeline.py

class PipelineRunner:
    """
    Orchestrates the execution of data processing steps within an ingestion pipeline.

    The PipelineRunner is responsible for initializing the IngestionContext,
    executing ordered stages of configured steps (agents), and managing the overall
    lifecycle of the ingestion process, including error handling and status reporting.
    Steps in a stage run concurrently against the same ingestion context.
    """

    def __init__(self, steps: list[Step] | Mapping[str, list[Step]], pipeline_version_id: UUID | None = None, toolsets: Mapping[str, AbstractToolset] = {}, logger: logging.Logger | None = None, step_descriptions: Mapping[int, str] | None = None, step_validations: Mapping[int, list[ValidationCheck]] | None = None):
        self.stages = dict(steps) if isinstance(steps, Mapping) else {f"stage-{i}-{uuid4().hex[:8]}": [step] for i, step in enumerate(steps, 1)}
        self.steps = [step for stage in self.stages.values() for step in stage]
        self.pipeline_version_id = pipeline_version_id or UUID(int=0)
        self.toolsets = toolsets
        self.output_result: OutputResult | None = None
        self.step_descriptions = dict(step_descriptions or {})
        self.step_validations = dict(step_validations or {})

        if logger is None:
            logger = logging.getLogger(__name__)
            logger.addHandler(logging.NullHandler())
        self.logger = logger

    async def run(self, ctx: IngestionContext[Any], output: Output | None = None) -> list[StepResult]:
        """
        Executes the ingestion pipeline.

        Runs stages sequentially and their steps concurrently, updating the shared
        context and reporting progress to the metadata store.

        Args:
            ctx: The shared ingestion context containing items and state.

        Returns:
            A list of StepResult objects, one for each executed step.
        """

        if not ctx.store:
            raise ValueError("Metadata store required in context")

        ctx = await self._resolve_resource_single(ctx)
        if not ctx.items:
            ctx.items = [KnowledgeItem(raw_content=ctx.resource.raw_content, content_uri=ctx.resource.content_uri, content_encoding=ctx.resource.content_encoding, metadata=ctx.resource.metadata.copy(), tenant_id=ctx.tenant_id, source_uri=ctx.resource.uri)]
        ctx.run_id = await ctx.store.create_pipeline_run(ctx.tenant_id, self.pipeline_version_id, ctx.resource.uri, ctx.run_id)
        await ctx.store.set_pipeline_checkpoint(ctx.run_id, self._checkpoint(ctx, 0))
        return await self._execute(ctx, output, 0)

    async def resume(self, run_id: UUID, store: MetadataStore, output: Output | None = None) -> tuple[IngestionContext[ResolvedResource], Awaitable[list[StepResult]]]:
        """Resumes a pipeline run from its latest successful-stage checkpoint."""

        checkpoint = await store.get_pipeline_checkpoint(run_id)
        if checkpoint is None:
            raise ValueError(f"Pipeline checkpoint not found: {run_id}")
        if checkpoint.stage > len(self.stages):
            raise ValueError(f"Pipeline has {len(self.stages)} stages, but checkpoint {run_id} completed {checkpoint.stage}; resume with the pipeline YAML used to create this run")
        ctx = IngestionContext(resource=checkpoint.resource, store=store, run_id=run_id, tenant_id=checkpoint.tenant_id, domain_profile=checkpoint.domain_profile, metadata=checkpoint.metadata, items=checkpoint.items)
        await store.update_pipeline_run(run_id, status="running")
        return ctx, self._execute(ctx, output, checkpoint.stage)

    async def _execute(self, ctx: IngestionContext[ResolvedResource], output: Output | None, stage: int) -> list[StepResult]:
        PIPELINE_RUNNERS[ctx.run_id] = self
        try:
            result = []
            async for e in self._run(ctx, stage):
                event.publish(e)
                if isinstance(e, event.PipelineStepEndEvent):
                    result.append(e.step_result)
            success = all(r.status != StepStatus.FAILURE for r in result)
            if success:
                await ctx.store.set_pipeline_checkpoint(ctx.run_id, None)
            if output is not None and success:
                self.output_result = await output.write(ctx)
            return result
        finally:
            del PIPELINE_RUNNERS[ctx.run_id]

    async def flatten(self, ctx: IngestionContext[Any]) -> AsyncIterator[IngestionContext[ResolvedResource]]:
        if isinstance(ctx.resource, ResolvedResource):
            yield ctx
            return

        async for resource in read_resources(str(ctx.resource)):
            yield replace(ctx, run_id=uuid4(), resource=resource)

    async def run_flattened(self, ctx: IngestionContext[Any]) -> AsyncIterator[tuple[IngestionContext[ResolvedResource], list[StepResult]]]:
        queue = asyncio.Queue[tuple[IngestionContext[ResolvedResource], list[StepResult]] | BaseException | None]()
        async def runner(child):
            try:
                await queue.put((child, await self.run(child)))
            except BaseException as e:
                await queue.put(e)
                raise

        async def producer():
            try:
                async with asyncio.TaskGroup() as tg:
                    async for child in self.flatten(ctx):
                        tg.create_task(runner(child))
            finally:
                await queue.put(None)

        task = asyncio.get_event_loop().create_task(producer())
        try:
            while (item := await queue.get()) is not None:
                if isinstance(item, BaseException):
                    raise item
                yield item
        finally:
            await task

    async def _run(self, ctx: IngestionContext[ResolvedResource], start: int) -> AsyncIterator[event.Event]:
        final_status: Literal["succeeded", "failed"] = "succeeded"
        error_msg = None

        yield event.PipelineBeginEvent(run_id=ctx.run_id, tenant_id=ctx.tenant_id, completed_steps=self.steps[:sum(map(len, list(self.stages.values())[:start]))])
        for stage_index, stage in enumerate(self.stages.values()):
            if stage_index < start:
                continue
            step_runs = []
            for step in stage:
                step_name = lookup_step_id(step) or "unknown"
                step_runs.append((await ctx.store.create_step_run(ctx.run_id, step_name), step_name, step, self.step_validations.get(id(step), [])))
                yield event.PipelineStepBeginEvent(run_id=ctx.run_id, tenant_id=ctx.tenant_id, step_name=step_name, step=step)
            results = await asyncio.gather(*(self._run_step(ctx, step_name, step, validation) for _, step_name, step, validation in step_runs))
            for (sid, step_name, step, _), res in zip(step_runs, results, strict=True):
                yield event.PipelineStepEndEvent(run_id=ctx.run_id, tenant_id=ctx.tenant_id, step_name=step_name, step=step, step_result=res)
                await ctx.store.update_step_run(sid, status=res.status.value, metadata=res.metadata, error=res.error)
            if failed := next((res for res in results if res.status == StepStatus.FAILURE), None):
                final_status = "failed"
                error_msg = failed.error
                break
            await ctx.store.set_pipeline_checkpoint(ctx.run_id, self._checkpoint(ctx, stage_index + 1))

        await ctx.store.update_pipeline_run(ctx.run_id, status=final_status, completed_at=datetime.now(UTC), error=error_msg)
        yield event.PipelineEndEvent(run_id=ctx.run_id, tenant_id=ctx.tenant_id, status=final_status, error=error_msg)

    async def _run_step(self, ctx: IngestionContext[ResolvedResource], step_name: str, step: Step, validation: list[ValidationCheck]) -> StepResult:
        token = _CURRENT_STEP.set((step_name, step))
        try:
            if skipped := self._validate(ctx, validation, {"condition"})[0]:
                return skipped.model_copy(update={"status": StepStatus.SKIPPED})
            failed, stop_validation = self._validate(ctx, validation, {"before", "plain"})
            if failed:
                return failed
            result = await step.run(ctx)
            if result.status == StepStatus.SUCCESS and not stop_validation and (failed := self._validate(ctx, validation, {"after"}, result)[0]):
                return failed
            return result
        except Exception as e:
            traceback.print_exception(e)
            return StepResult(status=StepStatus.FAILURE, error=str(e))
        finally:
            _CURRENT_STEP.reset(token)

    def _validate(self, ctx: IngestionContext[ResolvedResource], validation: list[ValidationCheck], modes: set[ValidationMode], result: StepResult | None = None) -> tuple[StepResult | None, bool]:
        doc: dict[str, Any] = {
            "metadata": ctx.metadata,
            "items": [{"id": str(item.id), "source_uri": item.source_uri, "metadata": item.metadata} for item in ctx.items],
        }
        if result is not None:
            doc["result"] = result.model_dump(mode="json", exclude={"items": {"__all__": {"raw_content"}}})
        for check in validation:
            if check.mode not in modes:
                continue
            if not jq.compile(check.expr).input_value(doc).first():
                return StepResult(status=StepStatus.FAILURE, error=_interpolate_value(check.message, doc)), False
            if check.mode == "plain":
                return None, True
        return None, False

    def _checkpoint(self, ctx: IngestionContext[ResolvedResource], stage: int) -> PipelineCheckpoint:
        return PipelineCheckpoint(stage=stage, resource=ctx.resource, tenant_id=ctx.tenant_id, domain_profile=ctx.domain_profile, metadata=ctx.metadata, items=ctx.items)

    async def _resolve_resource_single(self, ctx: IngestionContext[Any], **kwargs) -> IngestionContext[ResolvedResource]:
        if isinstance(ctx.resource, ResolvedResource):
            return ctx

        resolved = None
        skipped = []
        async for res in read_resources(str(ctx.resource), parse_limit=1, **kwargs):
            if resolved is None:
                resolved = res
            else:
                skipped.append(res.uri)

        if resolved is None:
            raise RuntimeError("Could not resolve resource: " + str(ctx.resource))

        if len(skipped) > 0:
            self.logger.warning("Resource %s was resolved to %s, but also returned %d other resource(s) which were ignored. "
                "Consider using %s() instead, or %s() followed by %s() for a more granularity over execution.",
                str(ctx.resource), resolved.uri, len(skipped), self.run_flattened.__qualname__, self.flatten.__qualname__, self.run.__qualname__
            )

        ctx.resource = resolved
        return ctx

resume async

resume(
    run_id: UUID,
    store: MetadataStore,
    output: Output | None = None,
) -> tuple[
    IngestionContext[ResolvedResource],
    Awaitable[list[StepResult]],
]

Resumes a pipeline run from its latest successful-stage checkpoint.

Source code in src/omni_ingest/core/pipeline.py

async def resume(self, run_id: UUID, store: MetadataStore, output: Output | None = None) -> tuple[IngestionContext[ResolvedResource], Awaitable[list[StepResult]]]:
    """Resumes a pipeline run from its latest successful-stage checkpoint."""

    checkpoint = await store.get_pipeline_checkpoint(run_id)
    if checkpoint is None:
        raise ValueError(f"Pipeline checkpoint not found: {run_id}")
    if checkpoint.stage > len(self.stages):
        raise ValueError(f"Pipeline has {len(self.stages)} stages, but checkpoint {run_id} completed {checkpoint.stage}; resume with the pipeline YAML used to create this run")
    ctx = IngestionContext(resource=checkpoint.resource, store=store, run_id=run_id, tenant_id=checkpoint.tenant_id, domain_profile=checkpoint.domain_profile, metadata=checkpoint.metadata, items=checkpoint.items)
    await store.update_pipeline_run(run_id, status="running")
    return ctx, self._execute(ctx, output, checkpoint.stage)

run async

run(
    ctx: IngestionContext[Any], output: Output | None = None
) -> list[StepResult]

Executes the ingestion pipeline.

Runs stages sequentially and their steps concurrently, updating the shared context and reporting progress to the metadata store.

Parameters:

Name	Type	Description	Default
ctx	IngestionContext[Any]	The shared ingestion context containing items and state.	required

Returns:

Type	Description
list[StepResult]	A list of StepResult objects, one for each executed step.

Source code in src/omni_ingest/core/pipeline.py

async def run(self, ctx: IngestionContext[Any], output: Output | None = None) -> list[StepResult]:
    """
    Executes the ingestion pipeline.

    Runs stages sequentially and their steps concurrently, updating the shared
    context and reporting progress to the metadata store.

    Args:
        ctx: The shared ingestion context containing items and state.

    Returns:
        A list of StepResult objects, one for each executed step.
    """

    if not ctx.store:
        raise ValueError("Metadata store required in context")

    ctx = await self._resolve_resource_single(ctx)
    if not ctx.items:
        ctx.items = [KnowledgeItem(raw_content=ctx.resource.raw_content, content_uri=ctx.resource.content_uri, content_encoding=ctx.resource.content_encoding, metadata=ctx.resource.metadata.copy(), tenant_id=ctx.tenant_id, source_uri=ctx.resource.uri)]
    ctx.run_id = await ctx.store.create_pipeline_run(ctx.tenant_id, self.pipeline_version_id, ctx.resource.uri, ctx.run_id)
    await ctx.store.set_pipeline_checkpoint(ctx.run_id, self._checkpoint(ctx, 0))
    return await self._execute(ctx, output, 0)

create_pipeline_from_config

create_pipeline_from_config(
    config_path: Path,
    factory: StepFactory | None = None,
    config_args: dict[str, Any] | None = None,
) -> PipelineRunner

Factory function to create a PipelineRunner from a YAML configuration file.

Parameters:

Name	Type	Description	Default
config_path	Path	Path to the .yaml configuration file.	required
factory	StepFactory | None	Factory to use to resolve and load steps from.	None

Source code in src/omni_ingest/core/pipeline.py

def create_pipeline_from_config(config_path: Path, factory: StepFactory | None = None, config_args: dict[str, Any] | None = None) -> PipelineRunner:
    """
    Factory function to create a PipelineRunner from a YAML configuration file.

    Args:
        config_path: Path to the .yaml configuration file.
        factory: Factory to use to resolve and load steps from.
    """
    config = _load_pipeline_config(config_path)

    stages: dict[str, list[Step]] = {}
    descriptions = {}
    validations = {}
    for s in _flatten_steps(_resolve_pipeline_steps(config, config_args or {}, config_path), config_path.parent, factory):
        stage = s.stage if s.stage is not None else f"stage-{len(stages) + 1}-{uuid4().hex[:8]}"
        if stage in stages and stage != next(reversed(stages)):
            raise ValueError(f"Stage {stage} must be contiguous")
        cls = get_step_class(s.agent, factory)
        kwargs = _resolve_step_kwargs(cls, s.config, {})
        step = cls(**kwargs)
        if s.description:
            descriptions[id(step)] = s.description
        if s.validation:
            validations[id(step)] = s.validation
        stages.setdefault(stage, []).append(step)
    toolsets = {k: MCPToolset(Client(MCPConfig(mcpServers={k: v}).to_dict())) for k, v in config.tool_vendors.items()}
    return PipelineRunner(steps=stages, toolsets=toolsets, step_descriptions=descriptions, step_validations=validations)

register_step

register_step(
    id: str, cls: type[Step], override: bool = False
)

Register a Step class in the global factory.

Parameters:

Name	Type	Description	Default
id	str	Name used to identify this step in configurations.	required
cls	type[Step]	Step class the identifier must model.	required
override	bool	Whether to override if step with the ID already exists.	False

Source code in src/omni_ingest/core/pipeline.py

@validate_call
def register_step(id: str, cls: type[Step], override: bool = False):
    """
    Register a Step class in the global factory.

    Args:
        id: Name used to identify this step in configurations.
        cls: Step class the identifier must model.
        override: Whether to override if step with the ID already exists.
    """
    if not override and cls in _STEP_FACTORY.values():
        raise ValueError(f"Step {id} already registered")
    _STEP_FACTORY[id] = cls

protocol

Protocol definitions (Abstract Base Classes) for framework extensibility.

Classes:

Name	Description
ContentResolver	Resolve virtual content URIs into bytes during ingestion.
MetadataStore	Interface for metadata management and data lineage tracking.
Translator	Translate text or string values in a nested dictionary.
VectorStore	Interface for vector-based search and retrieval systems.

ContentResolver

Bases: Protocol

Resolve virtual content URIs into bytes during ingestion.

ByteContent.content(ctx) delegates to ctx.content_resolver whenever an item has content_uri instead of inline raw_content. Implementations can use full ingestion context to resolve routes against source resources, object stores, caches, or other pipeline-scoped state.

Methods:

Name	Description
resolve	Return resolved bytes or raise ValueError when URI is unsupported.

Source code in src/omni_ingest/core/protocol.py

@runtime_checkable
class ContentResolver(Protocol):
    """Resolve virtual content URIs into bytes during ingestion.

    `ByteContent.content(ctx)` delegates to `ctx.content_resolver` whenever an
    item has `content_uri` instead of inline `raw_content`. Implementations can
    use full ingestion context to resolve routes against source resources,
    object stores, caches, or other pipeline-scoped state.
    """

    async def resolve(self, uri: str, ctx: IngestionContext[ResolvedResource]) -> bytes:
        """Return resolved bytes or raise `ValueError` when URI is unsupported."""
        ...

resolve async

resolve(
    uri: str, ctx: IngestionContext[ResolvedResource]
) -> bytes

Return resolved bytes or raise ValueError when URI is unsupported.

Source code in src/omni_ingest/core/protocol.py

async def resolve(self, uri: str, ctx: IngestionContext[ResolvedResource]) -> bytes:
    """Return resolved bytes or raise `ValueError` when URI is unsupported."""
    ...

MetadataStore

Bases: Protocol

Interface for metadata management and data lineage tracking.

The MetadataStore protocol defines the operations required for managing relational data, including multi-tenancy records, pipeline execution logs, and data provenance information.

Methods:

Name	Description
create_knowledge_item	Persists a new knowledge item.
create_pipeline_run	Records the start of a pipeline run.
create_step_run	Records the start of a pipeline step.
create_tenant	Creates a new tenant.
get_default_pipeline_version	Retrieves the default version for a pipeline.
get_knowledge_item	Retrieves a knowledge item by ID.
get_pipeline_checkpoint	Returns the latest checkpoint for a pipeline run.
get_pipeline_version	Retrieves a specific pipeline version.
get_tenant	Retrieves a tenant by slug.
list_knowledge_items	Lists knowledge items for a tenant.
set_pipeline_checkpoint	Replaces the checkpoint, or deletes it when checkpoint is None.
update_pipeline_run	Updates the status of a pipeline run.
update_step_run	Updates the status and metadata of a step run.
upsert_pipeline	Creates or updates a pipeline definition.
upsert_pipeline_version	Creates or updates a pipeline version.

Source code in src/omni_ingest/core/protocol.py

@runtime_checkable
class MetadataStore(Protocol):
    """
    Interface for metadata management and data lineage tracking.

    The MetadataStore protocol defines the operations required for
    managing relational data, including multi-tenancy records, pipeline
    execution logs, and data provenance information.
    """

    async def get_tenant(self, slug: str) -> Tenant | None:
        """Retrieves a tenant by slug."""
        ...

    async def create_tenant(self, slug: str, display_name: str) -> Tenant:
        """Creates a new tenant."""
        ...

    async def create_pipeline_run(self, tenant_id: UUID, pipeline_version_id: UUID, source_uri: str, id: UUID | None = None) -> UUID:
        """Records the start of a pipeline run."""
        ...

    async def update_pipeline_run(self, run_id: UUID, status: str, completed_at: datetime | None = None, error: str | None = None) -> None:
        """Updates the status of a pipeline run."""
        ...

    async def get_pipeline_checkpoint(self, id: UUID) -> PipelineCheckpoint | None:
        """Returns the latest checkpoint for a pipeline run."""
        ...

    async def set_pipeline_checkpoint(self, id: UUID, checkpoint: PipelineCheckpoint | None) -> None:
        """Replaces the checkpoint, or deletes it when checkpoint is None."""
        ...

    async def create_step_run(self, run_id: UUID, step_name: str) -> UUID:
        """Records the start of a pipeline step."""
        ...

    async def update_step_run(self, step_id: UUID, status: str, metadata: JsonObject | None = None, error: str | None = None) -> None:
        """Updates the status and metadata of a step run."""
        ...

    async def create_knowledge_item(
        self,
        tenant_id: UUID,
        domain_profile: str,
        modality: str,
        source_uri: str,
        content: bytes | None = None,
        pipeline_version_id: UUID | None = None,
        title: str | None = None,
        description: str | None = None,
        structured_views: JsonObject | None = None,
        lineage: JsonObject | None = None,
        metadata: JsonObject | None = None,
    ) -> UUID:
        """Persists a new knowledge item."""
        ...

    async def get_knowledge_item(self, item_id: UUID) -> KnowledgeItem | None:
        """Retrieves a knowledge item by ID."""
        ...

    async def list_knowledge_items(self, tenant_id: UUID, domain_profile: str | None = None, **filters: object) -> list[KnowledgeItem]:
        """Lists knowledge items for a tenant."""
        ...

    async def get_pipeline_version(self, pipeline_name: str, version: str) -> PipelineVersion | None:
        """Retrieves a specific pipeline version."""
        ...

    async def get_default_pipeline_version(self, pipeline_name: str) -> PipelineVersion | None:
        """Retrieves the default version for a pipeline."""
        ...

    async def upsert_pipeline(self, name: str, domain_profile: str) -> UUID:
        """Creates or updates a pipeline definition."""
        ...

    async def upsert_pipeline_version(self, pipeline_id: UUID, version: str, yaml_profile_name: str, is_default: bool = False) -> UUID:
        """Creates or updates a pipeline version."""
        ...

create_knowledge_item async

create_knowledge_item(
    tenant_id: UUID,
    domain_profile: str,
    modality: str,
    source_uri: str,
    content: bytes | None = None,
    pipeline_version_id: UUID | None = None,
    title: str | None = None,
    description: str | None = None,
    structured_views: JsonObject | None = None,
    lineage: JsonObject | None = None,
    metadata: JsonObject | None = None,
) -> UUID

Persists a new knowledge item.

Source code in src/omni_ingest/core/protocol.py

async def create_knowledge_item(
    self,
    tenant_id: UUID,
    domain_profile: str,
    modality: str,
    source_uri: str,
    content: bytes | None = None,
    pipeline_version_id: UUID | None = None,
    title: str | None = None,
    description: str | None = None,
    structured_views: JsonObject | None = None,
    lineage: JsonObject | None = None,
    metadata: JsonObject | None = None,
) -> UUID:
    """Persists a new knowledge item."""
    ...

create_pipeline_run async

create_pipeline_run(
    tenant_id: UUID,
    pipeline_version_id: UUID,
    source_uri: str,
    id: UUID | None = None,
) -> UUID

Records the start of a pipeline run.

Source code in src/omni_ingest/core/protocol.py

async def create_pipeline_run(self, tenant_id: UUID, pipeline_version_id: UUID, source_uri: str, id: UUID | None = None) -> UUID:
    """Records the start of a pipeline run."""
    ...

create_step_run async

create_step_run(run_id: UUID, step_name: str) -> UUID

Records the start of a pipeline step.

Source code in src/omni_ingest/core/protocol.py

async def create_step_run(self, run_id: UUID, step_name: str) -> UUID:
    """Records the start of a pipeline step."""
    ...

create_tenant async

create_tenant(slug: str, display_name: str) -> Tenant

Creates a new tenant.

Source code in src/omni_ingest/core/protocol.py

async def create_tenant(self, slug: str, display_name: str) -> Tenant:
    """Creates a new tenant."""
    ...

get_default_pipeline_version async

get_default_pipeline_version(
    pipeline_name: str,
) -> PipelineVersion | None

Retrieves the default version for a pipeline.

Source code in src/omni_ingest/core/protocol.py

async def get_default_pipeline_version(self, pipeline_name: str) -> PipelineVersion | None:
    """Retrieves the default version for a pipeline."""
    ...

get_knowledge_item async

get_knowledge_item(item_id: UUID) -> KnowledgeItem | None

Retrieves a knowledge item by ID.

Source code in src/omni_ingest/core/protocol.py

async def get_knowledge_item(self, item_id: UUID) -> KnowledgeItem | None:
    """Retrieves a knowledge item by ID."""
    ...

get_pipeline_checkpoint async

get_pipeline_checkpoint(
    id: UUID,
) -> PipelineCheckpoint | None

Returns the latest checkpoint for a pipeline run.

Source code in src/omni_ingest/core/protocol.py

async def get_pipeline_checkpoint(self, id: UUID) -> PipelineCheckpoint | None:
    """Returns the latest checkpoint for a pipeline run."""
    ...

get_pipeline_version async

get_pipeline_version(
    pipeline_name: str, version: str
) -> PipelineVersion | None

Retrieves a specific pipeline version.

Source code in src/omni_ingest/core/protocol.py

async def get_pipeline_version(self, pipeline_name: str, version: str) -> PipelineVersion | None:
    """Retrieves a specific pipeline version."""
    ...

get_tenant async

get_tenant(slug: str) -> Tenant | None

Retrieves a tenant by slug.

Source code in src/omni_ingest/core/protocol.py

async def get_tenant(self, slug: str) -> Tenant | None:
    """Retrieves a tenant by slug."""
    ...

list_knowledge_items async

list_knowledge_items(
    tenant_id: UUID,
    domain_profile: str | None = None,
    **filters: object
) -> list[KnowledgeItem]

Lists knowledge items for a tenant.

Source code in src/omni_ingest/core/protocol.py

async def list_knowledge_items(self, tenant_id: UUID, domain_profile: str | None = None, **filters: object) -> list[KnowledgeItem]:
    """Lists knowledge items for a tenant."""
    ...

set_pipeline_checkpoint async

set_pipeline_checkpoint(
    id: UUID, checkpoint: PipelineCheckpoint | None
) -> None

Replaces the checkpoint, or deletes it when checkpoint is None.

Source code in src/omni_ingest/core/protocol.py

async def set_pipeline_checkpoint(self, id: UUID, checkpoint: PipelineCheckpoint | None) -> None:
    """Replaces the checkpoint, or deletes it when checkpoint is None."""
    ...

update_pipeline_run async

update_pipeline_run(
    run_id: UUID,
    status: str,
    completed_at: datetime | None = None,
    error: str | None = None,
) -> None

Updates the status of a pipeline run.

Source code in src/omni_ingest/core/protocol.py

async def update_pipeline_run(self, run_id: UUID, status: str, completed_at: datetime | None = None, error: str | None = None) -> None:
    """Updates the status of a pipeline run."""
    ...

update_step_run async

update_step_run(
    step_id: UUID,
    status: str,
    metadata: JsonObject | None = None,
    error: str | None = None,
) -> None

Updates the status and metadata of a step run.

Source code in src/omni_ingest/core/protocol.py

async def update_step_run(self, step_id: UUID, status: str, metadata: JsonObject | None = None, error: str | None = None) -> None:
    """Updates the status and metadata of a step run."""
    ...

upsert_pipeline async

upsert_pipeline(name: str, domain_profile: str) -> UUID

Creates or updates a pipeline definition.

Source code in src/omni_ingest/core/protocol.py

async def upsert_pipeline(self, name: str, domain_profile: str) -> UUID:
    """Creates or updates a pipeline definition."""
    ...

upsert_pipeline_version async

upsert_pipeline_version(
    pipeline_id: UUID,
    version: str,
    yaml_profile_name: str,
    is_default: bool = False,
) -> UUID

Creates or updates a pipeline version.

Source code in src/omni_ingest/core/protocol.py

async def upsert_pipeline_version(self, pipeline_id: UUID, version: str, yaml_profile_name: str, is_default: bool = False) -> UUID:
    """Creates or updates a pipeline version."""
    ...

Translator

Bases: Protocol

Translate text or string values in a nested dictionary.

Source code in src/omni_ingest/core/protocol.py

@runtime_checkable
class Translator(Protocol):
    """Translate text or string values in a nested dictionary."""

    async def __aenter__(self) -> Translator: ...

    async def __aexit__(self, exc_type, exc, tb) -> None: ...

    @overload
    async def translate(self, src: str | None, dst: str, data: str) -> str: ...

    @overload
    async def translate(self, src: str | None, dst: str, data: dict[str, Any], skip_fields: Iterable[str] = ()) -> dict[str, Any]: ...

    @overload
    async def translate(self, src: str | None, dst: str, data: list[Any], skip_fields: Iterable[str] = ()) -> list[Any]: ...

    async def translate(self, src: str | None, dst: str, data: str | dict[str, Any] | list[Any], skip_fields: Iterable[str] = ()) -> str | dict[str, Any] | list[Any]: ...

VectorStore

Bases: Protocol

Interface for vector-based search and retrieval systems.

The VectorStore protocol specifies methods for storing and querying mathematical vector representations of data, enabling semantic search capabilities based on contextual meaning.

Methods:

Name	Description
close	Closes the connection to the store.
count	Returns the total number of items in the store.
search	Searches for items similar to the given query vector.
upsert	Persists or updates knowledge items in the store.

Attributes:

Name	Type	Description
kind	str	Stable identifier for the backing vector store implementation.
space	str	Logical namespace used by the vector store, such as a table or collection.

Source code in src/omni_ingest/core/protocol.py

@runtime_checkable
class VectorStore(Protocol):
    """
    Interface for vector-based search and retrieval systems.

    The VectorStore protocol specifies methods for storing and querying
    mathematical vector representations of data, enabling semantic
    search capabilities based on contextual meaning.
    """

    async def upsert(self, items: list[KnowledgeItem], ctx: Any) -> None:
        """Persists or updates knowledge items in the store."""
        ...

    async def search(self, query: list[float], k: PositiveInt = 5, tenant_id: UUID | None = None) -> list[KnowledgeItem]:
        """Searches for items similar to the given query vector."""
        ...

    async def count(self) -> int:
        """Returns the total number of items in the store."""
        ...

    async def __aenter__(self) -> VectorStore:
        return self

    async def __aexit__(self, exc_type, exc, tb):
        ...

    async def close(self) -> None:
        """Closes the connection to the store."""
        ...

    @property
    def kind(self) -> str:
        """Stable identifier for the backing vector store implementation."""
        ...

    @property
    def space(self) -> str:
        """Logical namespace used by the vector store, such as a table or collection."""
        ...

kind property

kind: str

Stable identifier for the backing vector store implementation.

space property

space: str

Logical namespace used by the vector store, such as a table or collection.

close async

close() -> None

Closes the connection to the store.

Source code in src/omni_ingest/core/protocol.py

async def close(self) -> None:
    """Closes the connection to the store."""
    ...

count async

count() -> int

Returns the total number of items in the store.

Source code in src/omni_ingest/core/protocol.py

async def count(self) -> int:
    """Returns the total number of items in the store."""
    ...

search async

search(
    query: list[float],
    k: PositiveInt = 5,
    tenant_id: UUID | None = None,
) -> list[KnowledgeItem]

Searches for items similar to the given query vector.

Source code in src/omni_ingest/core/protocol.py

async def search(self, query: list[float], k: PositiveInt = 5, tenant_id: UUID | None = None) -> list[KnowledgeItem]:
    """Searches for items similar to the given query vector."""
    ...

upsert async

upsert(items: list[KnowledgeItem], ctx: Any) -> None

Persists or updates knowledge items in the store.

Source code in src/omni_ingest/core/protocol.py

async def upsert(self, items: list[KnowledgeItem], ctx: Any) -> None:
    """Persists or updates knowledge items in the store."""
    ...

parser

Resource readers for turning local or remote inputs into resolved content.

Functions:

Name	Description
as_wav	Return (wav_path, is_temp). If src is already a WAV it is returned

as_wav async

as_wav(path: str, stream: BinaryIO)

Return (wav_path, is_temp). If src is already a WAV it is returned unchanged. Otherwise it is converted to a temporary WAV file; the caller owns that file and must delete it when done.

Source code in src/omni_ingest/parser.py

@asynccontextmanager
async def as_wav(path: str, stream: BinaryIO):
    """
    Return (wav_path, is_temp). If `src` is already a WAV it is returned
    unchanged. Otherwise it is converted to a temporary WAV file; the
    caller owns that file and must delete it when done.
    """
    fmt = Path(path).suffix.lstrip(".").lower() or "mp3"
    if fmt == "wav":
        yield path
    else:
        with tempfile.NamedTemporaryFile(suffix=".wav") as tmp_src, tempfile.NamedTemporaryFile(suffix=".wav") as tmp_dst:
            shutil.copyfileobj(stream, tmp_src)
            tmp_src.flush()
            tmp_src.seek(0)

            proc = await asyncio.create_subprocess_exec(
                "ffmpeg", "-y", "-i", tmp_src.name, "-ar", "16000", "-ac", "1", "-f", "wav", tmp_dst.name,
                stdout=asyncio.subprocess.DEVNULL,
                stderr=asyncio.subprocess.PIPE,
            )
            _, stderr = await proc.communicate()
            if proc.returncode != 0:
                raise RuntimeError(f"ffmpeg conversion failed:\n{stderr.decode()}")
            yield tmp_dst.name

port

Interfaces and concrete implementations for external storage and data connectivity.

Modules:

Name	Description
content	Content resolver implementations.
graph_store
metadata_store	Persistent storage implementation for pipeline metadata and tenant information.
ocr	OCR engine builders for document/page-image to Markdown extraction.
translator	Translation service implementations.
vector_store	Implementations for storing and querying high-dimensional vector embeddings.

content

Content resolver implementations.

graph_store

Classes:

Name	Description
InMemoryGraphStore	Volatile in-memory implementation of a graph store for testing and local dev.
Neo4jGraphStore	Production graph storage using Neo4j.

InMemoryGraphStore

Bases: GraphStore

Volatile in-memory implementation of a graph store for testing and local dev.

Source code in src/omni_ingest/port/graph_store.py

class InMemoryGraphStore(GraphStore):
    """
    Volatile in-memory implementation of a graph store for testing and local dev.
    """

    def __init__(self):
        self.nodes = {}
        self.edges = []

    async def upsert_nodes(self, nodes: list[dict[str, Any]]):
        for n in nodes:
            if "id" in n:
                self.nodes[n["id"]] = n

    async def upsert_edges(self, edges: list[dict[str, Any]]):
        self.edges.extend(edges)

    async def close(self) -> None:
        pass

Neo4jGraphStore

Bases: GraphStore

Production graph storage using Neo4j.

Source code in src/omni_ingest/port/graph_store.py

class Neo4jGraphStore(GraphStore):
    """
    Production graph storage using Neo4j.
    """

    def __init__(self, uri: str, user: str, password: str):
        try:
            from neo4j import AsyncGraphDatabase
        except ModuleNotFoundError as exc:
            raise ImportError("Neo4jGraphStore requires neo4j. Install omni-ingest[storage] or add neo4j to the environment.") from exc

        self._driver = AsyncGraphDatabase.driver(uri, auth=(user, password))

    async def upsert_nodes(self, nodes: list[dict[str, Any]]):
        rows = []
        for node in nodes:
            if "id" not in node:
                continue

            properties: dict[str, Any] = {}
            if "properties" in node and isinstance(node["properties"], dict):
                properties.update(node["properties"])
            for key, value in node.items():
                if key not in {"id", "label", "properties"}:
                    properties[key] = value

            label = "Entity"
            if "label" in node and isinstance(node["label"], str) and node["label"] != "":
                label = node["label"]

            rows.append({"id": str(node["id"]), "label": label, "properties": properties})

        if not rows:
            return

        query = """
        UNWIND $nodes AS node
        MERGE (n:Entity {id: node.id})
        SET n += node.properties
        SET n.label = node.label
        """
        async with self._driver.session() as session:
            result = await session.run(query, nodes=rows)
            await result.consume()

    async def upsert_edges(self, edges: list[dict[str, Any]]):
        rows = []
        for edge in edges:
            if "source" not in edge or "target" not in edge or "type" not in edge:
                continue

            properties = {}
            for key, value in edge.items():
                if key not in {"source", "target", "type"}:
                    properties[key] = value

            rows.append(
                {
                    "source": str(edge["source"]),
                    "target": str(edge["target"]),
                    "type": str(edge["type"]),
                    "properties": properties,
                }
            )

        if not rows:
            return

        query = """
        UNWIND $edges AS edge
        MERGE (source:Entity {id: edge.source})
        MERGE (target:Entity {id: edge.target})
        MERGE (source)-[rel:RELATED_TO {source: edge.source, target: edge.target, type: edge.type}]->(target)
        SET rel += edge.properties
        """
        async with self._driver.session() as session:
            result = await session.run(query, edges=rows)
            await result.consume()

    async def close(self) -> None:
        await self._driver.close()

metadata_store

Persistent storage implementation for pipeline metadata and tenant information.

Classes:

Name	Description
NullMetadataStore	Metadata store that preserves metadata records without storing item content.
SQLAlchemyMetadataStore	SQLAlchemy-based implementation of the MetadataStore protocol.

NullMetadataStore

Bases: MetadataStore

Metadata store that preserves metadata records without storing item content.

Source code in src/omni_ingest/port/metadata_store.py

class NullMetadataStore(MetadataStore):
    """
    Metadata store that preserves metadata records without storing item content.
    """

    def __init__(self):
        self._inner = SQLAlchemyMetadataStore("sqlite+aiosqlite:///:memory:")

    async def close(self) -> None:
        await self._inner.close()

    async def __aenter__(self):
        await self._inner.__aenter__()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self._inner.__aexit__(exc_type, exc, tb)

    async def get_tenant(self, slug: str) -> Tenant | None:
        return await self._inner.get_tenant(slug)

    async def create_tenant(self, slug: str, display_name: str) -> Tenant:
        return await self._inner.create_tenant(slug, display_name)

    async def create_pipeline_run(self, tenant_id: UUID, pipeline_version_id: UUID, source_uri: str, id: UUID | None = None) -> UUID:
        return await self._inner.create_pipeline_run(tenant_id, pipeline_version_id, source_uri, id=id)

    async def update_pipeline_run(self, run_id: UUID, status: str, completed_at: datetime | None = None, error: str | None = None) -> None:
        await self._inner.update_pipeline_run(run_id, status, completed_at=completed_at, error=error)

    async def get_pipeline_checkpoint(self, id: UUID) -> PipelineCheckpoint | None:
        return None

    async def set_pipeline_checkpoint(self, id: UUID, checkpoint: PipelineCheckpoint | None) -> None:
        return None

    async def create_step_run(self, run_id: UUID, step_name: str) -> UUID:
        return await self._inner.create_step_run(run_id, step_name)

    async def update_step_run(self, step_id: UUID, status: str, metadata: JsonObject | None = None, error: str | None = None) -> None:
        await self._inner.update_step_run(step_id, status, metadata=metadata, error=error)

    async def create_knowledge_item(
        self,
        tenant_id: UUID,
        domain_profile: str,
        modality: str,
        source_uri: str,
        content: bytes | None = None,
        pipeline_version_id: UUID | None = None,
        title: str | None = None,
        description: str | None = None,
        structured_views: JsonObject | None = None,
        lineage: JsonObject | None = None,
        metadata: JsonObject | None = None,
    ) -> UUID:
        return await self._inner.create_knowledge_item(
            tenant_id=tenant_id,
            domain_profile=domain_profile,
            modality=modality,
            source_uri=source_uri,
            content=b"",
            pipeline_version_id=pipeline_version_id,
            title=title,
            description=description,
            structured_views=structured_views,
            lineage=lineage,
            metadata=metadata,
        )

    async def get_knowledge_item(self, item_id: UUID) -> KnowledgeItem | None:
        return await self._inner.get_knowledge_item(item_id)

    async def list_knowledge_items(self, tenant_id: UUID, domain_profile: str | None = None, **filters: object) -> list[KnowledgeItem]:
        return await self._inner.list_knowledge_items(tenant_id, domain_profile=domain_profile, **filters)

    async def get_pipeline_version(self, pipeline_name: str, version: str) -> PipelineVersion | None:
        return await self._inner.get_pipeline_version(pipeline_name, version)

    async def get_default_pipeline_version(self, pipeline_name: str) -> PipelineVersion | None:
        return await self._inner.get_default_pipeline_version(pipeline_name)

    async def upsert_pipeline(self, name: str, domain_profile: str) -> UUID:
        return await self._inner.upsert_pipeline(name, domain_profile)

    async def upsert_pipeline_version(self, pipeline_id: UUID, version: str, yaml_profile_name: str, is_default: bool = False) -> UUID:
        return await self._inner.upsert_pipeline_version(pipeline_id, version, yaml_profile_name, is_default=is_default)

SQLAlchemyMetadataStore

Bases: MetadataStore

SQLAlchemy-based implementation of the MetadataStore protocol.

Persistent storage for tenants, pipeline runs, and knowledge base tracking using an asynchronous SQL backend.

Methods:

Name	Description
close	Disposes of the database engine and releases all connections.

Source code in src/omni_ingest/port/metadata_store.py

class SQLAlchemyMetadataStore(MetadataStore):
    """
    SQLAlchemy-based implementation of the MetadataStore protocol.

    Persistent storage for tenants, pipeline runs, and knowledge base tracking
    using an asynchronous SQL backend.
    """

    def __init__(self, url: str | None = None):
        self.engine = create_async_engine(url or settings.metadata_store)
        self.session_factory = async_sessionmaker(self.engine, expire_on_commit=False)
        self._init_lock = asyncio.Lock()
        self._init = False

    async def _init_db(self):
        if self._init:
            return
        async with self._init_lock:
            if self._init:
                return
            async with self.engine.begin() as conn:
                await conn.run_sync(SQLModel.metadata.create_all)
            self._init = True

    async def close(self):
        """Disposes of the database engine and releases all connections."""
        await self._init_db()
        await self.engine.dispose()

    async def __aenter__(self):
        await self._init_db()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.close()

    async def get_tenant(self, slug: str) -> Tenant | None:
        await self._init_db()
        async with self.session_factory() as s:
            return (await s.execute(select(Tenant).where(col(Tenant.slug) == slug))).scalar_one_or_none()

    async def create_tenant(self, slug: str, display_name: str) -> Tenant:
        await self._init_db()
        async with self.session_factory() as s:
            t = Tenant(slug=slug, display_name=display_name)
            s.add(t)
            await s.commit()
            return t

    async def create_pipeline_run(self, tenant_id: UUID, pipeline_version_id: UUID, source_uri: str, id: UUID | None = None) -> UUID:
        await self._init_db()
        async with self.session_factory() as s:
            r = PipelineRunRecord(
                id=id or uuid4(),
                tenant_id=tenant_id,
                pipeline_version_id=pipeline_version_id,
                source_uri=source_uri,
                status="running",
                started_at=datetime.now(UTC),
            )
            s.add(r)
            await s.commit()
            return r.id

    async def update_pipeline_run(self, run_id: UUID, status: str, completed_at: datetime | None = None, error: str | None = None) -> None:
        await self._init_db()
        async with self.session_factory() as s:
            r = await s.get(PipelineRunRecord, run_id)
            if r:
                r.status, r.completed_at, r.error_message = status, completed_at, error
                await s.commit()

    async def get_pipeline_checkpoint(self, id: UUID) -> PipelineCheckpoint | None:
        await self._init_db()
        async with self.session_factory() as s:
            row = await s.get(PipelineCheckpointRecord, id)
            if not row:
                return None
            try:
                checkpoint = pickle.loads(row.checkpoint)
                return checkpoint if isinstance(checkpoint, PipelineCheckpoint) else None
            except Exception:
                return None

    async def set_pipeline_checkpoint(self, id: UUID, checkpoint: PipelineCheckpoint | None) -> None:
        await self._init_db()
        async with self.session_factory() as s:
            if checkpoint is None:
                if row := await s.get(PipelineCheckpointRecord, id):
                    await s.delete(row)
            else:
                await s.merge(PipelineCheckpointRecord(pipeline_run_id=id, checkpoint=pickle.dumps(checkpoint, protocol=pickle.HIGHEST_PROTOCOL)))
            await s.commit()

    async def create_step_run(self, run_id: UUID, step_name: str) -> UUID:
        await self._init_db()
        async with self.session_factory() as s:
            st = PipelineStepRunRecord(pipeline_run_id=run_id, step_name=step_name, status="running", started_at=datetime.now(UTC))
            s.add(st)
            await s.commit()
            return st.id

    async def update_step_run(self, step_id: UUID, status: str, metadata: JsonObject | None = None, error: str | None = None) -> None:
        await self._init_db()
        async with self.session_factory() as s:
            st = await s.get(PipelineStepRunRecord, step_id)
            if st:
                st.status, st.metadata_ = status, metadata or st.metadata_
                if error:
                    st.error_message = error
                st.completed_at = datetime.now(UTC)
                await s.commit()

    async def create_knowledge_item(
        self,
        tenant_id: UUID,
        domain_profile: str,
        modality: str,
        source_uri: str,
        content: bytes | None = None,
        pipeline_version_id: UUID | None = None,
        title: str | None = None,
        description: str | None = None,
        structured_views: JsonObject | None = None,
        lineage: JsonObject | None = None,
        metadata: JsonObject | None = None,
    ) -> UUID:
        await self._init_db()
        async with self.session_factory() as s:
            i = KnowledgeItemRecord(
                tenant_id=tenant_id,
                domain_profile=domain_profile,
                modality=modality,
                source_uri=source_uri,
                content=content or b"",
                pipeline_version_id=pipeline_version_id,
                title=title,
                description=description,
                structured_views=structured_views,
                lineage=lineage,
                metadata_=metadata or {},
            )
            s.add(i)
            await s.commit()
            return i.id

    async def get_knowledge_item(self, item_id: UUID) -> KnowledgeItem | None:
        await self._init_db()
        async with self.session_factory() as s:
            if r := await s.get(KnowledgeItemRecord, item_id):
                return KnowledgeItem.model_validate(r)
            return None

    async def list_knowledge_items(self, tenant_id: UUID, domain_profile: str | None = None, **filters: object) -> list[KnowledgeItem]:
        await self._init_db()
        async with self.session_factory() as s:
            q = select(KnowledgeItemRecord).where(col(KnowledgeItemRecord.tenant_id) == tenant_id)
            if domain_profile:
                q = q.where(col(KnowledgeItemRecord.domain_profile) == domain_profile)
            for k, v in filters.items():
                q = q.where(getattr(KnowledgeItemRecord, k) == v)
            records = (await s.execute(q)).scalars().all()
            return [KnowledgeItem.model_validate(r) for r in records]

    async def get_pipeline_version(self, pipeline_name: str, version: str) -> PipelineVersion | None:
        await self._init_db()
        async with self.session_factory() as s:
            return (await s.execute(select(PipelineVersion).join(Pipeline).where(col(Pipeline.name) == pipeline_name, col(PipelineVersion.version) == version))).scalar_one_or_none()

    async def get_default_pipeline_version(self, pipeline_name: str) -> PipelineVersion | None:
        await self._init_db()
        async with self.session_factory() as s:
            return (await s.execute(select(PipelineVersion).join(Pipeline).where(col(Pipeline.name) == pipeline_name, col(PipelineVersion.is_default)))).scalar_one_or_none()

    async def upsert_pipeline(self, name: str, domain_profile: str) -> UUID:
        await self._init_db()
        async with self.session_factory() as s:
            p = (await s.execute(select(Pipeline).where(col(Pipeline.name) == name))).scalar_one_or_none()
            if not p:
                p = Pipeline(name=name, domain_profile=domain_profile)
                s.add(p)
            else:
                p.domain_profile = domain_profile
            await s.commit()
            return p.id

    async def upsert_pipeline_version(self, pipeline_id: UUID, version: str, yaml_profile_name: str, is_default: bool = False) -> UUID:
        await self._init_db()
        async with self.session_factory() as s:
            pv = (await s.execute(select(PipelineVersion).where(col(PipelineVersion.pipeline_id) == pipeline_id, col(PipelineVersion.version) == version))).scalar_one_or_none()
            if not pv:
                pv = PipelineVersion(pipeline_id=pipeline_id, version=version, yaml_profile_name=yaml_profile_name, is_default=is_default)
                s.add(pv)
            else:
                pv.yaml_profile_name, pv.is_default = yaml_profile_name, is_default
            await s.flush()
            if is_default:
                await s.execute(update(PipelineVersion).where(col(PipelineVersion.pipeline_id) == pipeline_id, col(PipelineVersion.id) != pv.id).values(is_default=False))
            await s.commit()
            return pv.id

close async

close()

Disposes of the database engine and releases all connections.

Source code in src/omni_ingest/port/metadata_store.py

async def close(self):
    """Disposes of the database engine and releases all connections."""
    await self._init_db()
    await self.engine.dispose()

ocr

OCR engine builders for document/page-image to Markdown extraction.

Functions:

Name	Description
by_page	Wrap a single-image Ocr into a full-document Ocr
llm_ocr_builder	OCR via a vision LLM. Sends the document as-is (PDF or image) in one call — any
openai_ocr_builder	OCR via any OpenAI-compatible vision/chat-completions endpoint.

by_page

by_page(
    extract_page: Ocr,
    dpi: int = 200,
    ocr_scanned: bool = True,
    text_layer_threshold: int = 20,
) -> Ocr

Wrap a single-image Ocr into a full-document Ocr

Source code in src/omni_ingest/port/ocr.py

def by_page(extract_page: Ocr, dpi: int = 200, ocr_scanned: bool = True, text_layer_threshold: int = 20) -> Ocr:
    """Wrap a single-image `Ocr` into a full-document `Ocr`"""

    def render_pdf_pages(content: bytes) -> list[tuple[str, bool, bytes | None]]:
        pages = []
        with pymupdf.open(stream=content, filetype="pdf") as pdf:
            for page in pdf.pages():
                assert isinstance(page, pymupdf.Page)
                text = page.get_text().strip()
                needs_ocr = ocr_scanned and len(text) < text_layer_threshold
                image = page.get_pixmap(dpi=dpi, alpha=False).tobytes("png") if needs_ocr else None
                pages.append((text, needs_ocr, image))
        return pages

    async def ocr(item: ByteContent) -> ByteContent:
        if item.metadata.get("content_type") != "application/pdf":
            return await extract_page(item)

        pages = await asyncio.to_thread(render_pdf_pages, item.raw_content)

        async def resolve(text: str, needs_ocr: bool, image: bytes | None) -> str:
            if not needs_ocr or image is None:
                return text
            result = await extract_page(ByteContent(raw_content=image, metadata={"content_type": "image/png"}))
            return result.raw_content.decode(result.content_encoding)

        markdown = "\n\n".join(await asyncio.gather(*(resolve(*page) for page in pages)))
        return ByteContent(raw_content=markdown.encode("utf-8"), content_encoding="utf-8", metadata={"content_type": "text/markdown"})

    return ocr

llm_ocr_builder

llm_ocr_builder(
    ctx: IngestionContext[ResolvedResource],
    src_lang: str | None,
    dst_lang: str | None,
) -> Ocr

OCR via a vision LLM. Sends the document as-is (PDF or image) in one call — any

Source code in src/omni_ingest/port/ocr.py

def llm_ocr_builder(ctx: IngestionContext[ResolvedResource], src_lang: str | None, dst_lang: str | None) -> Ocr:
    """OCR via a vision LLM. Sends the document as-is (PDF or image) in one call — any"""
    agent = _DefaultAgentMixin()._agent(ctx)

    async def ocr(item: ByteContent) -> ByteContent:
        content_type = await item.content_type(ctx)
        result = await agent.run([DEFAULT_VISION_OCR_PROMPT, BinaryContent(data=await item.content(ctx), media_type=content_type)])
        return ByteContent(raw_content=result.output.encode("utf-8"), content_encoding="utf-8", metadata={"content_type": "text/markdown"})

    return ocr

openai_ocr_builder

openai_ocr_builder(
    ctx: IngestionContext[ResolvedResource],
    src_lang: str | None,
    dst_lang: str | None,
) -> Ocr

OCR via any OpenAI-compatible vision/chat-completions endpoint.

Source code in src/omni_ingest/port/ocr.py

def openai_ocr_builder(ctx: IngestionContext[ResolvedResource], src_lang: str | None, dst_lang: str | None) -> Ocr:
    """OCR via any OpenAI-compatible vision/chat-completions endpoint."""
    if settings.ocr_api_key is None or settings.ocr_base_url is None:
        raise ValueError("OCR_API_KEY/OCR_BASE_URL are not configured")
    client = AsyncOpenAI(api_key=settings.ocr_api_key.get_secret_value(), base_url=str(settings.ocr_base_url))

    async def extract_page(image: ByteContent) -> ByteContent:
        content_type = await image.content_type(ctx)
        data = await image.content(ctx)
        response = await client.chat.completions.create(
            model=settings.ocr_model,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": DEFAULT_VISION_OCR_PROMPT},
                        {"type": "image_url", "image_url": {"url": f"data:{content_type};base64,{base64.b64encode(data).decode()}"}},
                    ],
                }
            ],
        )
        markdown = response.choices[0].message.content or ""
        return ByteContent(raw_content=markdown.encode("utf-8"), content_encoding="utf-8", metadata={"content_type": "text/markdown"})

    return by_page(extract_page)

translator

Translation service implementations.

vector_store

Implementations for storing and querying high-dimensional vector embeddings.

Classes:

Name	Description
AzureVectorSearchStore	Azure AI Search vector retrieval backed by a pre-created index.
ChromaVectorStore	Persistent ChromaDB storage for local vector retrieval.
InMemoryVectorStore	Volatile, fast vector storage suitable for testing and ephemeral processing.
LanceDBVectorStore	Vector storage and semantic search powered by LanceDB.
MongoVectorStore	MongoDB Atlas vector search backed by a pre-provisioned search index.
QdrantVectorStore	Production-grade vector search using Qdrant.

AzureVectorSearchStore

Bases: VectorStore

Azure AI Search vector retrieval backed by a pre-created index.

Source code in src/omni_ingest/port/vector_store.py

class AzureVectorSearchStore(VectorStore):
    """
    Azure AI Search vector retrieval backed by a pre-created index.
    """

    def __init__(self, endpoint: str, api_key: str, index_name: str,):
        try:
            from azure.core.credentials import AzureKeyCredential
            from azure.search.documents.aio import SearchClient
        except ModuleNotFoundError as exc:
            raise ImportError(
                "AzureVectorSearchStore requires azure-search-documents. Install omni-ingest[storage] or add azure-search-documents to the environment."
            ) from exc

        self.index_name = index_name
        self._client = SearchClient(endpoint=endpoint, index_name=index_name, credential=AzureKeyCredential(api_key))

    async def __aexit__(self, exc_type, exc, tb):
        await self.close()

    async def upsert(self, items: list[KnowledgeItem], ctx: Any) -> None:
        documents = []
        for item in items:
            embedding = item.metadata.get("embedding")
            if embedding is None:
                continue
            documents.append(
                {
                    DOCUMENT_ID_FIELD: str(item.id),
                    DOCUMENT_VECTOR_FIELD: embedding,
                    DOCUMENT_CONTENT_FIELD: await item.decode(ctx),
                    DOCUMENT_TENANT_ID_FIELD: "" if item.tenant_id is None else str(item.tenant_id),
                    DOCUMENT_SOURCE_URI_FIELD: item.source_uri or "",
                    DOCUMENT_METADATA_FIELD: json.dumps(item.metadata),
                }
            )

        if documents:
            await self._client.merge_or_upload_documents(documents)

    async def search(self, query: list[float], k: PositiveInt = 5, tenant_id: UUID | None = None) -> list[KnowledgeItem]:
        try:
            from azure.search.documents.models import VectorizedQuery
        except ModuleNotFoundError as exc:
            raise ImportError(
                "AzureVectorSearchStore requires azure-search-documents. Install omni-ingest[storage] or add azure-search-documents to the environment."
            ) from exc

        vector_query = VectorizedQuery(vector=query, k_nearest_neighbors=int(k), fields=DOCUMENT_VECTOR_FIELD)
        filter_expression = None
        if tenant_id is not None:
            filter_expression = f"{DOCUMENT_TENANT_ID_FIELD} eq '{tenant_id}'"

        results = await self._client.search(
            search_text=None,
            vector_queries=[vector_query],
            filter=filter_expression,
            select=[
                DOCUMENT_ID_FIELD,
                DOCUMENT_CONTENT_FIELD,
                DOCUMENT_TENANT_ID_FIELD,
                DOCUMENT_SOURCE_URI_FIELD,
                DOCUMENT_METADATA_FIELD,
            ],
            top=int(k),
        )
        items: list[KnowledgeItem] = []
        async for document in results:
            tenant_value = document.get(DOCUMENT_TENANT_ID_FIELD)
            source_uri = document.get(DOCUMENT_SOURCE_URI_FIELD)
            metadata = document.get(DOCUMENT_METADATA_FIELD, "{}")
            items.append(
                KnowledgeItem(
                    id=UUID(str(document[DOCUMENT_ID_FIELD])),
                    raw_content=str(document.get(DOCUMENT_CONTENT_FIELD, "") or "").encode(),
                    tenant_id=None if not tenant_value else UUID(str(tenant_value)),
                    source_uri=None if not source_uri else str(source_uri),
                    metadata=json.loads(metadata) if isinstance(metadata, str) else dict(metadata),
                )
            )
        return items

    async def count(self) -> int:
        return await self._client.get_document_count()

    async def close(self):
        await self._client.close()

    @property
    def kind(self) -> str:
        return "azure_vector_search"

    @property
    def space(self) -> str:
        return self.index_name

ChromaVectorStore

Bases: VectorStore

Persistent ChromaDB storage for local vector retrieval.

Source code in src/omni_ingest/port/vector_store.py

class ChromaVectorStore(VectorStore):
    """
    Persistent ChromaDB storage for local vector retrieval.
    """

    def __init__(self, collection_name: str, path: str):
        try:
            import chromadb
        except ModuleNotFoundError as exc:
            raise ImportError("ChromaVectorStore requires chromadb. Install omni-ingest[storage] or add chromadb to the environment.") from exc

        self.collection_name = collection_name
        self.path = str(Path(path).resolve())
        self._client = chromadb.PersistentClient(path=self.path)
        self._collection: Any = self._client.get_or_create_collection(name=self.collection_name, metadata={"hnsw:space": "cosine"})

    async def __aexit__(self, exc_type, exc, tb):
        await self.close()

    async def upsert(self, items: list[KnowledgeItem], ctx: Any) -> None:
        ids: list[str] = []
        documents: list[str] = []
        embeddings: list[list[float]] = []
        metadatas: list[dict[str, str]] = []

        for item in items:
            embedding = item.metadata.get("embedding")
            if embedding is None:
                continue
            ids.append(str(item.id))
            documents.append(await item.decode(ctx))
            embeddings.append(embedding)
            metadatas.append(
                {
                    DOCUMENT_TENANT_ID_FIELD: "" if item.tenant_id is None else str(item.tenant_id),
                    DOCUMENT_SOURCE_URI_FIELD: item.source_uri or "",
                    DOCUMENT_METADATA_FIELD: json.dumps(item.metadata),
                }
            )

        if ids:
            self._collection.upsert(ids=ids, documents=documents, embeddings=embeddings, metadatas=metadatas)

    async def search(self, query: list[float], k: PositiveInt = 5, tenant_id: UUID | None = None) -> list[KnowledgeItem]:
        where = None
        if tenant_id is not None:
            where = {DOCUMENT_TENANT_ID_FIELD: str(tenant_id)}

        result = self._collection.query(query_embeddings=[query], n_results=int(k), where=where)
        ids = result["ids"][0]
        documents = result["documents"][0]
        metadatas = result["metadatas"][0]

        items: list[KnowledgeItem] = []
        for item_id, content, metadata in zip(ids, documents, metadatas, strict=True):
            tenant_value = metadata.get(DOCUMENT_TENANT_ID_FIELD, "")
            source_uri = metadata.get(DOCUMENT_SOURCE_URI_FIELD, "")
            raw_metadata = metadata.get(DOCUMENT_METADATA_FIELD, "{}")
            items.append(
                KnowledgeItem(
                    id=UUID(str(item_id)),
                    raw_content=str(content or "").encode(),
                    tenant_id=None if not tenant_value else UUID(str(tenant_value)),
                    source_uri=None if not source_uri else str(source_uri),
                    metadata=json.loads(raw_metadata) if isinstance(raw_metadata, str) else dict(raw_metadata),
                )
            )
        return items

    async def count(self) -> int:
        return int(self._collection.count())

    async def close(self):
        pass

    @property
    def kind(self) -> str:
        return "chroma"

    @property
    def space(self) -> str:
        return self.collection_name

InMemoryVectorStore

Bases: LanceDBVectorStore

Volatile, fast vector storage suitable for testing and ephemeral processing.

Source code in src/omni_ingest/port/vector_store.py

class InMemoryVectorStore(LanceDBVectorStore):
    """
    Volatile, fast vector storage suitable for testing and ephemeral processing.
    """

    def __init__(self, vector_size: PositiveInt = 1536):
        self._temp = tempfile.TemporaryDirectory()
        super().__init__(table_name="temp_vectors", db_path=self._temp.name, vector_size=vector_size)

    async def close(self):
        await super().close()
        self._temp.cleanup()

    @property
    def kind(self) -> str:
        return "in_memory"

LanceDBVectorStore

Bases: VectorStore

Vector storage and semantic search powered by LanceDB.

Source code in src/omni_ingest/port/vector_store.py

class LanceDBVectorStore(VectorStore):
    """
    Vector storage and semantic search powered by LanceDB.
    """

    def __init__(self, table_name: str, db_path: str | None = None, vector_size: PositiveInt = 1536):
        self.table_name, self.vector_size = table_name, vector_size
        self.db_path = str(Path(db_path or "./lancedb_data").resolve())
        self._db, self._table = None, None

    async def __aenter__(self):
        await self._get_table()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.close()

    async def _get_table(self) -> lancedb.Table:
        if self._table is None:
            db = lancedb.connect(self.db_path)
            self._table = db.create_table(self.table_name, schema=pa.schema([
                pa.field("id", pa.string()),
                pa.field("vector", pa.list_(pa.float32(), self.vector_size)),
                pa.field("content", pa.string()),
                pa.field("tenant_id", pa.string(), nullable=True),
                pa.field("source_uri", pa.string(), nullable=True),
                pa.field("metadata", pa.string()),
            ]), exist_ok=True)
            self._db = db
        return self._table

    async def upsert(self, items: list[KnowledgeItem], ctx: Any) -> None:
        t, data = await self._get_table(), []
        for i in items:
            if emb := i.metadata.get("embedding"):
                data.append(
                    {
                        "id": str(i.id),
                        "vector": emb,
                        "content": await i.decode(ctx),
                        "tenant_id": "" if i.tenant_id is None else str(i.tenant_id),
                        "source_uri": i.source_uri or "",
                        "metadata": json.dumps(i.metadata),
                    }
                )
        if data:
            t.add(data, mode="overwrite")

    async def search(self, query: list[float], k: PositiveInt = 5, tenant_id: UUID | None = None) -> list[KnowledgeItem]:
        t = await self._get_table()
        q = t.search(query).limit(k)
        if tenant_id:
            q = q.where(f"tenant_id = '{tenant_id}'")
        res = q.to_list()
        items: list[KnowledgeItem] = []
        for row in res:
            tenant_value = row.get("tenant_id")
            source_uri = row.get("source_uri")
            metadata = row["metadata"]
            items.append(
                KnowledgeItem(
                    id=UUID(str(row["id"])),
                    raw_content=str(row["content"] or "").encode(),
                    tenant_id=None if not tenant_value else UUID(str(tenant_value)),
                    source_uri=None if not source_uri else str(source_uri),
                    metadata=json.loads(metadata) if isinstance(metadata, str) else dict(metadata),
                )
            )
        return items

    async def count(self) -> int:
        return int(len(await self._get_table()))

    async def close(self):
        pass

    @property
    def kind(self) -> str:
        return "lancedb"

    @property
    def space(self) -> str:
        return self.table_name

MongoVectorStore

Bases: VectorStore

MongoDB Atlas vector search backed by a pre-provisioned search index.

Source code in src/omni_ingest/port/vector_store.py

class MongoVectorStore(VectorStore):
    """
    MongoDB Atlas vector search backed by a pre-provisioned search index.
    """

    def __init__(self, uri: str, database_name: str, collection_name: str, index_name: str, num_candidates: PositiveInt = 100):
        try:
            from pymongo import AsyncMongoClient
        except ModuleNotFoundError as exc:
            raise ImportError("MongoVectorStore requires pymongo. Install omni-ingest[storage] or add pymongo to the environment.") from exc

        self.database_name = database_name
        self.collection_name = collection_name
        self.index_name = index_name
        self.num_candidates = num_candidates
        self._client: Any = AsyncMongoClient(uri)
        self._collection = self._client[self.database_name][self.collection_name]

    async def __aexit__(self, exc_type, exc, tb):
        await self.close()

    async def upsert(self, items: list[KnowledgeItem], ctx: Any) -> None:
        try:
            from pymongo import ReplaceOne
        except ModuleNotFoundError as exc:
            raise ImportError("MongoVectorStore requires pymongo. Install omni-ingest[storage] or add pymongo to the environment.") from exc

        operations = []
        for item in items:
            embedding = item.metadata.get("embedding")
            if embedding is None:
                continue
            operations.append(
                ReplaceOne(
                    {DOCUMENT_ID_FIELD: str(item.id)},
                    {
                        DOCUMENT_ID_FIELD: str(item.id),
                        DOCUMENT_VECTOR_FIELD: embedding,
                        DOCUMENT_CONTENT_FIELD: await item.decode(ctx),
                        DOCUMENT_TENANT_ID_FIELD: "" if item.tenant_id is None else str(item.tenant_id),
                        DOCUMENT_SOURCE_URI_FIELD: item.source_uri or "",
                        DOCUMENT_METADATA_FIELD: json.dumps(item.metadata),
                    },
                    upsert=True,
                )
            )

        if operations:
            await self._collection.bulk_write(operations, ordered=False)

    async def search(self, query: list[float], k: PositiveInt = 5, tenant_id: UUID | None = None) -> list[KnowledgeItem]:
        vector_stage: dict[str, Any] = {
            "index": self.index_name,
            "path": DOCUMENT_VECTOR_FIELD,
            "queryVector": query,
            "numCandidates": max(int(self.num_candidates), int(k)),
            "limit": int(k),
        }
        if tenant_id is not None:
            vector_stage["filter"] = {DOCUMENT_TENANT_ID_FIELD: str(tenant_id)}

        cursor = await self._collection.aggregate([
            {"$vectorSearch": vector_stage},
            {
                "$project": {
                    DOCUMENT_ID_FIELD: 1,
                    DOCUMENT_CONTENT_FIELD: 1,
                    DOCUMENT_TENANT_ID_FIELD: 1,
                    DOCUMENT_SOURCE_URI_FIELD: 1,
                    DOCUMENT_METADATA_FIELD: 1,
                }
            },
        ])
        results: list[KnowledgeItem] = []
        async for document in cursor:
            tenant_value = document.get(DOCUMENT_TENANT_ID_FIELD)
            source_uri = document.get(DOCUMENT_SOURCE_URI_FIELD)
            metadata = document.get(DOCUMENT_METADATA_FIELD, "{}")
            results.append(
                KnowledgeItem(
                    id=UUID(str(document[DOCUMENT_ID_FIELD])),
                    raw_content=str(document.get(DOCUMENT_CONTENT_FIELD, "") or "").encode(),
                    tenant_id=None if not tenant_value else UUID(str(tenant_value)),
                    source_uri=None if not source_uri else str(source_uri),
                    metadata=json.loads(metadata) if isinstance(metadata, str) else dict(metadata),
                )
            )
        return results

    async def count(self) -> int:
        return await self._collection.count_documents({})

    async def close(self):
        await self._client.close()

    @property
    def kind(self) -> str:
        return "mongo"

    @property
    def space(self) -> str:
        return f"{self.database_name}.{self.collection_name}"

QdrantVectorStore

Bases: VectorStore

Production-grade vector search using Qdrant.

Source code in src/omni_ingest/port/vector_store.py

class QdrantVectorStore(VectorStore):
    """
    Production-grade vector search using Qdrant.
    """

    def __init__(self, col: str, url: str | None = None, key: str | None = None, size: PositiveInt = 1536):
        self.col, self.size = col, size
        self._client = AsyncQdrantClient(
            url=url or "http://localhost:6333",
            api_key=key,
        )
        self._init = False

    async def __aenter__(self):
        await self._ensure()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.close()

    async def _ensure(self):
        if not self._init:
            if self.col not in [c.name for c in (await self._client.get_collections()).collections]:
                await self._client.create_collection(self.col, VectorParams(size=self.size, distance=Distance.COSINE))
            self._init = True

    async def upsert(self, items: list[KnowledgeItem], ctx: Any) -> None:
        await self._ensure()
        pts = [
            PointStruct(
                id=str(i.id),
                vector=i.metadata["embedding"],
                payload={
                    DOCUMENT_CONTENT_FIELD: await i.decode(ctx),
                    DOCUMENT_TENANT_ID_FIELD: "" if i.tenant_id is None else str(i.tenant_id),
                    DOCUMENT_SOURCE_URI_FIELD: i.source_uri or "",
                    DOCUMENT_METADATA_FIELD: json.dumps(i.metadata),
                },
            )
            for i in items
            if i.metadata.get("embedding")
        ]
        if pts:
            await self._client.upsert(self.col, pts)

    async def search(self, query: list[float], k: PositiveInt = 5, tenant_id: UUID | None = None) -> list[KnowledgeItem]:
        await self._ensure()
        filt = Filter(must=[FieldCondition(key="tenant_id", match=MatchValue(value=str(tenant_id)))]) if tenant_id else None
        res = await self._client.search(self.col, query, query_filter=filt, limit=k)  # type: ignore
        items: list[KnowledgeItem] = []
        for row in res:
            tenant_value = row.payload.get(DOCUMENT_TENANT_ID_FIELD)
            source_uri = row.payload.get(DOCUMENT_SOURCE_URI_FIELD)
            metadata = row.payload.get(DOCUMENT_METADATA_FIELD, "{}")
            items.append(
                KnowledgeItem(
                    id=UUID(str(row.id)),
                    raw_content=str(row.payload.get(DOCUMENT_CONTENT_FIELD, "") or "").encode(),
                    tenant_id=None if not tenant_value else UUID(str(tenant_value)),
                    source_uri=None if not source_uri else str(source_uri),
                    metadata=json.loads(metadata) if isinstance(metadata, str) else dict(metadata),
                )
            )
        return items

    async def count(self) -> int:
        await self._ensure()
        return (await self._client.get_collection(self.col)).points_count or 0

    async def close(self):
        await self._client.close()

    @property
    def kind(self) -> str:
        return "qdrant"

    @property
    def space(self) -> str:
        return self.col