Skip to content

ingest.md

Latest commit

 

History

History
209 lines (163 loc) · 7.04 KB

ingest.md

File metadata and controls

209 lines (163 loc) · 7.04 KB
Raw

Ingest pipeline — reference

End-to-end ingest for AgentContextOS — the CLI demo milestone of Phase 1. Step 1.10 wires every Phase 1 stage into a single orchestrator (rag_ingest.IngestPipeline) and exposes it via the FastAPI gateway (POST /v1/ingest/document) and the ragctl ingest smoke command.

Overview

The ingest pipeline runs each document through:

Connector → Parser → Chunker → Enricher → PII processor → Embedder → Index
              ↑         ↑          ↑           ↑               ↑         ↑
            1.3       1.5        1.6         1.7             1.8     1.1 / 2.x

The pipeline returns IngestResult per document and IngestRunSummary when driven from a connector. Both are Pydantic v2 frozen models in rag_core.types; both are exported to JSON Schema under dist/schemas/.

Python API

from rag_ingest import IngestPipeline
from rag_chunker import HeadingAwareChunker
from rag_enricher import DefaultEnricher
from rag_pii import PiiProcessor, RegexPIIDetector
from rag_embedders import OpenAIEmbedder
from rag_backends import PgVectorStore
from rag_parsers import default_registry

pipeline = IngestPipeline(
    chunker=HeadingAwareChunker(),
    enricher=DefaultEnricher(),
    pii_processor=PiiProcessor(detector=RegexPIIDetector()),
    embedder=OpenAIEmbedder(),
    index_backend=PgVectorStore(...),
    parser_registry=default_registry().select,
    policy_engine=None,  # NoopPolicyEngine() in dev; production PDP in prod
)

# Single document (e.g., from an HTTP upload)
result = await pipeline.ingest_document(ctx, document, content_bytes)
# IngestResult(document_id=..., status=ingested, chunk_count=12, ...)

# Connector-driven (filesystem crawl, S3 bucket scan, CDC stream, …)
summary = await pipeline.ingest_connector(ctx, connector)
# IngestRunSummary(total=42, ingested=40, blocked=1, failed=1, ...)

Constructor

Argument Type Required Notes
chunker Chunker ✅ typically HeadingAwareChunker
enricher Enricher ✅ typically DefaultEnricher
embedder Embedder ✅ OpenAI / Cohere / BGE / E5 / Noop
index_backend VectorIndexBackend ✅ PgVectorStore / Qdrant / Noop
parser_registry ParserSelector ✅ usually default_registry().select
pii_processor PiiProcessor | None ❌ None skips PII filtering entirely
policy_engine PolicyEngine | None ❌ None skips the ingest_doc PDP check

ingest_document

async def ingest_document(
    self,
    ctx: RequestContext,
    document: Document,
    content: bytes,
    *,
    mime_type: str | None = None,
) -> IngestResult

MIME-type fallback chain: explicit mime_type= → document.mime_type → "application/octet-stream". Returns an IngestResult with one of four statuses:

Status When
ingested Pipeline completed; chunks were embedded and indexed
blocked_by_policy PolicyEngine.evaluate(...) returned deny
skipped_empty Parser produced no chunks (empty file, unparseable content)
failed An unhandled error escaped a stage; error carries the message

Pipeline failures are caught per-document — one bad document never aborts the rest of a connector run.

ingest_connector

async def ingest_connector(
    self,
    ctx: RequestContext,
    connector: Connector,
    *,
    on_result: Callable[[IngestResult], Awaitable[None]] | None = None,
) -> IngestRunSummary

Iterates every document the connector yields. The optional on_result callback fires after each document so callers can stream progress. Callback errors are logged and swallowed — they never abort the run.

REST API

POST /v1/ingest/document

Multipart upload. Form fields + a single file part.

Field Type Required Notes
tenant_id str (form) ✅ propagated into RequestContext.tenant_id
corpus_id str (form) ✅ target corpus / partition
principal_id str (form) ❌ defaults to "gateway-anon"
source_uri str (form) ❌ defaults to upload:///<filename>
file binary (file) ✅ document bytes; Content-Type taken as MIME hint

Response (200): IngestResult JSON. Status failed is still 200 — the upload was received successfully; the result is the pipeline's outcome.

Error responses:

  • 400 empty upload — file part was zero bytes.
  • 422 — missing required form field (handled by FastAPI form validation).

GET /healthz / GET /v1/info

Lightweight liveness + capability advertisement. /v1/info lists the endpoints the gateway currently exposes (the surface will grow in Phase 3 with query / retrieve / corpora / SSE).

CLI

# Crawl a directory; default tenant + corpus; one-shot summary
ragctl ingest path/to/docs/

# Per-document progress line + custom tenant + corpus
ragctl ingest path/to/docs/ --per-doc --tenant acme --corpus kb

# Single file
ragctl ingest path/to/single.md

Uses the same IngestPipeline as the gateway, wired with the default parser registry + heading-aware chunker + default enricher + regex-PII processor + noop embedder + noop vector store. No infra required.

Output:

path:     docs/
corpus:   ragctl-local
tenant:   ragctl-local
total:    7
ingested: 6
blocked:  0
skipped:  1
failed:   0

--per-doc streams a one-liner per document as they complete:

          ingested  chunks=12   emb=12  bytes=2415  doc-1
          ingested  chunks=4    emb=4   bytes=489   doc-2
    skipped_empty  chunks=0    emb=0   bytes=0     empty.txt

Domain types

IngestResult

Field Type Notes
document_id DocumentId unchanged from the input Document
status IngestStatus enum (see above)
chunk_count int chunks indexed (post-PII filter)
embedding_count int typically equals chunk_count
blocked_chunk_count int chunks dropped by PII processor block action
bytes_in int size of the source bytes
error str | None populated when status == failed or blocked_by_policy
metadata dict[str, Any] parser name, MIME type, PII summary, etc.

IngestRunSummary

Field Type Notes
results list[IngestResult] full per-document detail
total / ingested / blocked / skipped / failed int per-status counters

Both ship in dist/schemas/IngestResult.json and dist/schemas/IngestRunSummary.json.

Extension points

The pipeline is call-site composable — every dependency is a constructor argument. Custom chunker / enricher / embedder / vector store implementations plug in without subclassing IngestPipeline itself.

To inject a custom stage between two existing ones, wrap the relevant SPI with a decorator that calls through to the wrapped instance. The upcoming Step 1.10 follow-ups (PolicyEngine wiring, rag.yaml backend selection, multi-stage observability spans) all land via the same shape.