Services API¶

The composed flows. Sync and async forms of each — use the async form inside a running event loop.

from ingestlib.services import ingest, retrieve
from ingestlib.services import aingest, aretrieve

ingest¶

ingestlib.services.ingest.ingestor.aingest `async` ¶

aingest(
    path: Path | str,
    *,
    store: VectorStore | None = None,
    namespace: str = "",
    skip_existing: bool = True,
    replaces: str | None = None,
    max_chunk_tokens: int = DEFAULT_MAX_CHUNK_TOKENS,
    categories: dict[str, str] | None = None,
    target_pages: str | None = None,
    max_pages: int | None = None,
    vocabulary: dict[str, str] | None = None,
    unmatched: str | None = None,
    on_stage: StageCallback | None = None,
) -> IngestResult

Run a document through the full pipeline (async).

The content-rule arguments pass straight to classify and split with their exact semantics: None resolves from rules.yaml's preset, an explicit {} forces open-ended/discovery, explicit values always win.

Lifecycle: a document's logical identity is (namespace, source path). When this path already holds an OLDER version (different checksum), the new version replaces it — the old version's vectors and artifacts are deleted AFTER the new one is fully live, so retrieval never has a gap (status="replaced"). Same checksum arriving from a new path is a move: only the registry re-points, nothing runs (status="moved").

path — PDF/DOCX/PPTX document, or a PNG/JPEG/WebP image store — vector store connector; defaults to the one selected by config.yaml's vector_store key namespace — vector-store namespace for multi-corpus setups skip_existing — return status="skipped" when this exact file (by checksum) already completed the FULL pipeline; a run that failed partway is retried. Dedup keys on file CONTENT only — re-ingesting with different rules still skips; pass skip_existing=False to re-run replaces — explicit doc_id of the version this file supersedes; normally unnecessary (path matching detects it), for when the new version lives at a different path max_chunk_tokens — split's chunk-size ceiling categories — classify rules {label: description}, max 20 target_pages — classify page selection like "1,3,5-7" (1-based) max_pages — classify page cap applied after selection vocabulary — split section categories {name: description}, max 50 unmatched — split's policy for pages fitting no category: "other" | "require" | "skip" on_stage — optional progress callback, called as on_stage(stage, event) with stage parse|classify|split|embed|upsert — plus replace when an old version is deleted — and event start|done; a stage that raises leaves its "start" unmatched. Exceptions from the callback are logged and ignored. Never called on the skip_existing fast path — nothing runs there.

ingestlib.services.ingest.ingestor.ingest ¶

ingest(
    path: Path | str,
    *,
    store: VectorStore | None = None,
    namespace: str = "",
    skip_existing: bool = True,
    replaces: str | None = None,
    max_chunk_tokens: int = DEFAULT_MAX_CHUNK_TOKENS,
    categories: dict[str, str] | None = None,
    target_pages: str | None = None,
    max_pages: int | None = None,
    vocabulary: dict[str, str] | None = None,
    unmatched: str | None = None,
    on_stage: StageCallback | None = None,
) -> IngestResult

Run a document through the full pipeline. Sync wrapper — use aingest() inside an event loop.

retrieve¶

ingestlib.services.retrieve.retriever.aretrieve `async` ¶

aretrieve(
    question: str,
    *,
    top_k: int = 5,
    filters: dict[str, Any] | None = None,
    namespace: str = "",
    rerank: bool = True,
    store: VectorStore | None = None,
    sources: list[str] | None = None,
) -> RetrievalResult

Retrieve the most relevant results for a question (async).

question — natural-language query top_k — results to return per source filters — payload constraints for document search, e.g. {"category": "x"} rerank — rerank document candidates with config.yaml's reranker store — vector store connector; defaults to config.yaml's vector_store sources — names from sources.yaml to query (documents and/or SQL databases). When given, retrieve fans out over them and returns a normalized envelope (result.results); omit it for plain document search (result.hits) — the exact prior behavior.

ingestlib.services.retrieve.retriever.retrieve ¶

retrieve(
    question: str,
    *,
    top_k: int = 5,
    filters: dict[str, Any] | None = None,
    namespace: str = "",
    rerank: bool = True,
    store: VectorStore | None = None,
    sources: list[str] | None = None,
) -> RetrievalResult

Retrieve the most relevant results for a question. Sync wrapper — use aretrieve() inside an event loop.

With sources=[...] (names from sources.yaml), retrieve fans out over the document corpus and/or SQL databases and returns normalized SourceResults in result.results — see Query databases (SQL).

Lifecycle¶

Manage the corpus as files change — replace, remove, sync, backfill. See Manage a corpus for the guide.

from ingestlib.services import remove, sync, backfill

ingestlib.services.lifecycle.remover.remove ¶

remove(
    target: Path | str,
    *,
    namespace: str = "",
    store: VectorStore | None = None,
) -> RemoveResult

Erase one document — vectors AND artifacts. Sync wrapper — use aremove() inside an event loop.

ingestlib.services.lifecycle.syncer.sync ¶

sync(
    directory: Path | str,
    *,
    namespace: str = "",
    prune: bool = False,
    dry_run: bool = False,
    glob: str = "**/*",
    store: VectorStore | None = None,
    on_stage: StageCallback | None = None,
) -> SyncResult

Make the corpus match a folder. Sync wrapper — use async_sync() inside an event loop (async is a keyword, hence the one naming exception to the a-prefix convention).

ingestlib.services.lifecycle.backfiller.backfill ¶

backfill(
    *, store: VectorStore | None = None, namespace: str = ""
) -> BackfillResult

Re-embed every stored document's chunks into a vector store. Sync wrapper — use abackfill() inside an event loop.