Ingestion Worker CLI

Module: aibs_backend.interfaces.cli.ingestion_worker

Purpose

Run a standalone ingestion worker process. The worker claims durable IngestionJob records and advances each document through:

  1. parse

  2. chunk

  3. index

  4. semantic_profile

Use this CLI when ingestion work must be isolated from the API process, for example on a GPU/model host, a separate worker VM, or a scaled ingestion lane.

Command

pdm run python -m aibs_backend.interfaces.cli.ingestion_worker

or, from an activated virtual environment:

python -m aibs_backend.interfaces.cli.ingestion_worker

Required Environment

The worker loads .env from the repository root. It must be able to reach:

  • document repository storage

  • upload/document roots used by the API

  • OpenSearch chunk, catalog, retrieval-pyramid, and job indexes

  • embedding service/model

  • reranker service/model when reranking is enabled

  • parser dependencies

  • LLM lanes needed by enabled ingestion features, including semantic profiles

Important knobs:

Setting

Meaning

AIBS_INGESTION_WORKER_COUNT

Number of worker loops in this process. 0 disables the pool.

AIBS_INGESTION_WORKER_IDLE_SLEEP_SECONDS

Sleep when no job is claimed.

AIBS_INGESTION_WORKER_ERROR_SLEEP_SECONDS

Sleep after a loop-level exception.

AIBS_INGESTION_WORKER_LEASE_REAP_INTERVAL_SECONDS

How often expired job leases are reclaimed.

AIBS_INGESTION_WORKER_CLAIM_SCAN_SIZE

Number of claimable jobs scanned per claim attempt.

AIBS_INGESTION_MAX_CONCURRENT_PARSE_STAGES

Process-local bound for parse stages.

AIBS_INGESTION_MAX_CONCURRENT_CHUNK_STAGES

Process-local bound for chunk stages.

AIBS_INGESTION_MAX_CONCURRENT_INDEX_STAGES

Process-local bound for index stages.

AIBS_INGESTION_MAX_CONCURRENT_SEMANTIC_PROFILE_STAGES

Process-local bound for semantic-profile stages.

AIBS_INGESTION_MAX_CONCURRENT_HEAVY_MODEL_STAGES

Process-local bound for heavy model-backed ingestion stages.

Startup Flow

  1. Load .env with override enabled.

  2. Build AppSettings and configure structured logging.

  3. Build the application container.

  4. Run readiness checks:

    • document repository integrity, when available

    • chunk OpenSearch connection

    • document catalog index connection

    • retrieval pyramid index connection, when configured

    • embedding readiness

    • reranker readiness when AIBS_RERANKER_ENABLED=true

    • parser warmup

    • answer assistant readiness when abstractive summaries require LLM startup checks

  5. Reap expired ingestion leases once before the worker pool starts.

  6. Start run_ingestion_worker_pool.

Readiness failure exits before any job is claimed.

Semantic Profile Stage

The semantic profile is a persisted ingestion artifact stored on the document metadata under semantic_profile. It is separate from processing metadata such as document_family.

The stage runs after indexing. That means a document can become searchable before semantic enrichment finishes. If model-backed semantic enrichment fails during normal ingestion, the stage retries and then persists an unavailable semantic profile so the durable ingestion job can complete degraded. The maintenance backfill CLI behaves differently: it reports failures and does not silently persist unavailable profiles.

Worker Loop Behavior

Each worker loop uses an ID like:

aibs-worker-<hostname>-<pid>-<lane>

The loop repeatedly:

  1. reaps expired leases when the interval has elapsed

  2. claims one job through IngestionJobRunner.claim_and_run_once

  3. runs the next stage for that job

  4. renews/observes durable job state through the runner

  5. sleeps when no job is available

asyncio.CancelledError stops the loop cleanly. Other loop exceptions are logged and retried after AIBS_INGESTION_WORKER_ERROR_SLEEP_SECONDS.

Verification

After startup, confirm:

  • readiness logs are successful

  • worker pool started with the expected worker count

  • uploaded documents leave queued/running states

  • job stage results include parse, chunk, index, and semantic_profile

  • degraded semantic profiles are reviewed and later repaired with backfill_semantic_profiles.md

Safety Notes

  • In split deployment, set AIBS_INGESTION_WORKER_ENABLED=false on API nodes if ingestion should run only on worker nodes.

  • Multiple workers can run against the same durable job index, but model concurrency must be sized for the cluster. The process-local stage semaphores are not a replacement for global model admission.

  • Do not use production indexes from a local development .env.