Services

The composed flows: ingest runs the whole pipeline and persists every stage; retrieve answers questions with cited chunks. Sync wrappers and async a-forms, same convention as the operations.

ingest

ingestlib.services.ingest.ingestor.aingest async

aingest(
    path: Path | str,
    *,
    store: VectorStore | None = None,
    namespace: str = "",
    skip_existing: bool = True,
    max_chunk_tokens: int = DEFAULT_MAX_CHUNK_TOKENS,
    on_stage: StageCallback | None = None,
) -> IngestResult

Run a document through the full pipeline (async).

The classify and split stages honor rules.yaml's presets when they exist (classify: closed-set rules + page settings; split: user section vocabulary + unmatched mode) — see rules.example.yaml.

path — PDF/DOCX/PPTX to ingest 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 max_chunk_tokens — split's chunk-size ceiling on_stage — optional progress callback, called as on_stage(stage, event) with stage parse|classify|split|embed|upsert 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,
    max_chunk_tokens: int = DEFAULT_MAX_CHUNK_TOKENS,
    on_stage: StageCallback | None = None,
) -> IngestResult

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

ingestlib.services.ingest.ingestor.IngestResult

Bases: BaseModel

Outcome of one document's journey through the full pipeline.

status — "ingested" (fresh run) or "skipped" (this checksum already completed the full pipeline and skip_existing was True) doc_id — the document's content checksum; keys every artifact and vector durations — per-stage wall-clock seconds (parse/classify/split/embed/upsert)

total_seconds property

total_seconds: float

Wall-clock total across all stages.

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,
) -> RetrievalResult

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

question — natural-language query top_k — hits to return filters — payload constraints, e.g. {"category": "research_paper"} rerank — rerank candidates with the reranker selected by config.yaml's reranker key (recommended; jina needs JINA_API_KEY) store — vector store connector; defaults to the one selected by config.yaml's vector_store key

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,
) -> RetrievalResult

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

ingestlib.services.retrieve.retriever.RetrievalResult

Bases: BaseModel

Ranked hits for one question, ready for prompt building.

context property

context: str

Numbered, cited chunks — paste-ready as LLM context.

ingestlib.services.retrieve.retriever.Hit

Bases: BaseModel

One retrieved chunk with both scoring signals.

vector_score — the store's retrieval score: cosine similarity on dense queries, an RRF rank score on fused hybrid queries rerank_score — reranker relevance (None when reranking was off)

citation property

citation: str

Human-readable source pointer, e.g. 'doc 7b6b95d79149 · p.4 · methods'.