Async Multi-Modal RAG Ingestion Pipeline

An async, production-ready document ingestion pipeline for RAG systems. Processes PDFs, images, arXiv papers, and documents from S3 through Docling extraction, GLiNER2 metadata enrichment & PII redaction, dual-vector embedding (dense + sparse), and stores everything in Qdrant with a NeonDB registry.

flowchart TD
    subgraph Upload["Document Sources"]
        U1[Local Folder] --> U2{Filter Files}
        U2 --> U3[Upload to S3 aioboto3]
        U3 --> U4[Return S3 Keys]
        AX1[ArXiv ID] --> AX2[Download PDF + Metadata]
        AX2 --> AX3[Upload to S3]
        AX3 --> U4
    end

    subgraph Ingestion["Prefect Flows"]
        U4 --> F1[ingest_batch]
        U4 --> F2[ingest_arxiv]
        F1 --> B[asyncio.gather + Semaphore concurrency=5]
        F2 --> B
    end

    subgraph PerDoc["Per-Document Pipeline"]
        B --> C1[S3 Download aioboto3]
        C1 --> D1[Docling Convert to_thread]
        D1 --> E1[Hybrid Chunker]

        E1 --> E2[Base64 Extraction Picture/Table/Formula]
        E2 --> E3{Has Text?}
        E3 -->|Yes| E4[Continue Pipeline]
        E3 -->|No (Image)| E5[NeonDB Only]

        E4 --> F1M[Metadata Extraction GLiNER2 base]
        E4 --> F2M[PII Detection GLiNER2 PII]

        F1M --> G1[INSERT documents NeonDB]
        G1 --> G2[INSERT chunks NeonDB]

        F2M --> H1[Redact Chunks]
        H1 --> I1{By Modality}

        I1 --> J1[Embed Dense jina-omni-nano 768d]
        I1 --> J2[Embed Sparse SPLADE v3 30522d]

        J1 --> K1[Upsert Qdrant point.dense_vector]
        J2 --> K1[Upsert Qdrant point.sparse_vector]

        K1 --> L1[Update chunks qdrant_point_id]
    end

    E5 --> G2
    L1 --> M[Flow Complete]

---

Features

---

Tech Stack

---

Prerequisites

• Python >= 3.12
• UV package manager (installation guide)
• Qdrant instance (local Docker or Qdrant Cloud)
• NeonDB / PostgreSQL instance (Neon or local)
• AWS S3 bucket with credentials
• Hugging Face account (for model downloads)

---

Installation

1. Clone the Repository

git clone https://github.com/your-org/prefect-dag.git
cd prefect-dag

2. Install Dependencies with UV

uv sync

This installs all 200+ dependencies including PyTorch, transformers, Docling, and Prefect.

3. Set Up Environment Variables

Copy the .env.example (create it from the table below) to .env and fill in your credentials:

cp .env.example .env

4. Initialize the Database

Run all migration scripts in order against your NeonDB / PostgreSQL instance:

psql $NEON_DATABASE_URL -f migrations/001_initial_schema.sql
psql $NEON_DATABASE_URL -f migrations/002_add_chunks_updated_at.sql
psql $NEON_DATABASE_URL -f migrations/003_add_chunk_base64.sql
# Or use the Makefile: make migrate

5. Start Local Services (Prefect + Qdrant)

make start      # Start Prefect server (4200) + Qdrant (6333) in Docker
make status     # Check service health
make stop       # Stop all services
make logs       # Follow container logs

6. Verify Setup

uv run python main.py
# Should print usage instructions

---

Environment Variables

Create a .env file in the project root with the following variables:

*AWS credentials can also be provided via AWS CLI config or IAM roles. **Required for downloading jina-omni-nano and GLiNER2 models from Hugging Face.

Example .env

# AWS S3
S3_BUCKET=my-rag-documents
S3_PREFIX=ingestion/
AWS_REGION=us-east-1
AWS_ACCESS_KEY_ID=AKIA...
AWS_SECRET_ACCESS_KEY=secret...

# Qdrant
QDRANT_URL=https://my-cluster.cloud.qdrant.io
QDRANT_API_KEY=eyJ...
QDRANT_COLLECTION=rag_documents

# NeonDB / PostgreSQL
NEON_DATABASE_URL=postgresql://user:pass@neon-host.aws.neon.tech/rag_pipeline

# Prefect (optional — for Prefect Cloud)
PREFECT_API_URL=https://api.prefect.cloud/api/...
PREFECT_API_KEY=pnu_...

# Hugging Face
HF_TOKEN=hf_...

---

Usage

Upload Documents to S3

Upload all supported files from a local folder to S3:

uv run python main.py upload ./my-documents/ s3://my-bucket/prefix/

Supported formats: .pdf, .doc, .docx, .jpg, .jpeg, .png, .gif, .webp, .bmp, .tiff

Ingest a Single Document

uv run python main.py ingest s3://my-bucket/prefix/document.pdf

Ingest a Batch of Documents

uv run python main.py ingest-batch batch-001 \
  s3://my-bucket/prefix/doc1.pdf \
  s3://my-bucket/prefix/doc2.pdf \
  s3://my-bucket/prefix/image.png

Ingest an arXiv Paper

Download a paper directly from arXiv by ID and process it through the full pipeline:

# Bare arXiv ID
uv run python main.py ingest-arxiv 1706.03762

# With arxiv: prefix
uv run python main.py ingest-arxiv arxiv:1810.04805

# Full URL
uv run python main.py ingest-arxiv https://arxiv.org/abs/2109.00031

ArXiv ID formats supported:

• 1706.03762 — bare ID
• arxiv:1810.04805 — prefixed
• https://arxiv.org/abs/2109.00031 — abstract URL
• https://arxiv.org/pdf/2203.02155.pdf — PDF URL

What happens:

1. Downloads PDF + Atom metadata (title, authors, abstract, categories, published date) from arXiv
2. Uploads PDF to S3 under ARXIV_S3_PREFIX (default: arxiv/)
3. Runs full pipeline: Docling → GLiNER2 → embeddings → Qdrant → NeonDB

Flush All Data (Qdrant + NeonDB)

Clear all vectors and database records while preserving S3 data:

# Interactive mode (requires confirmation)
uv run python main.py flush

# Non-interactive mode (for scripts / CI)
uv run python main.py flush --force
# Or via Makefile:
make flush

What gets deleted:

• All Qdrant vectors in the configured collection
• All rows in NeonDB documents table
• All rows in NeonDB chunks table

What is preserved:

• S3 bucket data (all uploaded PDFs, images, arXiv papers)
• Docker volumes
• Prefect flow history

Ingest a Batch of arXiv Papers

uv run python main.py ingest-arxiv-batch batch-003 \
  1706.03762 \
  1810.04805 \
  2109.00031

Programmatic Usage

import asyncio
from src.pipeline.flows import ingest_batch

async def main():
    s3_keys = [
        "s3://my-bucket/prefix/doc1.pdf",
        "s3://my-bucket/prefix/doc2.pdf",
    ]
    doc_ids = await ingest_batch(s3_keys, batch_id="batch-001")
    print(f"Processed {len(doc_ids)} documents")

asyncio.run(main())

Multi-Modal Retrieval (with Base64 Images/Tables)

When a retrieved chunk contains a chunk_base64 field, you can pass it directly to a multi-modal LLM (GPT-4o, Claude 3, Gemini):

import asyncio
from src.pipeline.embedder import embed_query
from src.pipeline.sparse_embedder import embed_text as embed_sparse_query
from src.pipeline.qdrant_store import hybrid_search

async def search(query: str):
    dense = await embed_query(query)
    sparse = (await embed_sparse_query([query]))[0]
    results = await hybrid_search(dense, sparse, limit=10)
    return results

results = asyncio.run(search("attention mechanism diagram"))
for r in results:
    payload = r["payload"]
    text = payload.get("chunk_text", "")[:200]
    base64 = payload.get("chunk_base64")
    chunk_type = payload.get("chunk_type")
    page = payload.get("page_number")

    if base64:
        print(f"[{chunk_type}] Page {page} | Score: {r['score']:.4f}")
        print(f"  Text: {text}...")
        print(f"  Base64: {base64[:60]}... ({len(base64)//1024}KB)")
        # Pass base64 to multi-modal LLM:
        # messages = [{"role": "user", "content": [
        #     {"type": "text", "text": query},
        #     {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{base64}"}}
        # ]}]
    else:
        print(f"[{chunk_type}] Page {page} | Score: {r['score']:.4f} | {text}...")

---

Project Structure

prefect-dag/
├── pyproject.toml                    # UV dependencies & project metadata
├── main.py                           # CLI entry point (upload, ingest, ingest-batch, ingest-arxiv)
├── Makefile                          # Docker + dev commands (start, stop, migrate, lint, test)
├── docker-compose.yml                # Prefect server + Qdrant with persistent volumes
├── README.md                         # This file
├── .env.example                      # Environment variable template
├── src/
│   └── pipeline/
│       ├── __init__.py
│       ├── config.py                 # Pydantic settings (loads .env)
│       ├── logging.py                # structlog JSON logging setup
│       ├── models.py                 # Pydantic models: Document, Chunk, Metadata
│       ├── arxiv_downloader.py     # Async arXiv PDF download + metadata extraction
│       ├── s3_client.py              # aioboto3 async S3 download/upload client
│       ├── s3_uploader.py            # Folder upload utility (pre-ingestion)
│       ├── docling_processor.py      # Docling convert with enrichments (tables, formulas, figures)
│       ├── chunker.py                # Docling HybridChunker wrapper
│       ├── gliner_metadata.py        # GLiNER2: 3-level metadata extraction
│       ├── gliner_pii.py             # GLiNER2 PII: 42-type detection + redaction
│       ├── embedder.py               # jina-omni-nano dense embeddings (768d)
│       ├── sparse_embedder.py        # SPLADE v3 sparse embeddings (30522d)
│       ├── qdrant_store.py           # AsyncQdrantClient: dense+sparse upsert, hybrid RRF search
│       ├── neondb_client.py          # asyncpg: documents + chunks CRUD
│       ├── flush.py                  # flush_qdrant(), flush_neondb(), flush_all()
│       └── flows.py                  # Prefect async flows (ingest_batch, ingest_arxiv)
├── migrations/
│   ├── 001_initial_schema.sql        # NeonDB schema: documents + chunks tables
│   ├── 002_add_chunks_updated_at.sql # Fix missing updated_at + relax document_type constraint
│   └── 003_add_chunk_base64.sql      # Add base64 PNG column for LLM multi-modal use
└── tests/                            # Test directory (add your tests here)

---

How It Works

Pipeline Steps (Per Document)

1. S3 Download (aioboto3) — Async streaming download from S3 to memory
2. Docling Conversion (asyncio.to_thread) — Parse PDF/image/DOC with enrichments:
   • Table extraction with markdown structure
   • LaTeX formula extraction
   • Figure/image classification + VLM captioning
   • generate_page_images=True for table/formula/picture base64 extraction
3. Hybrid Chunking (asyncio.to_thread) — Tokenization-aware chunking preserving context
4. Base64 Extraction (asyncio.to_thread) — For each chunk, scan all Docling items:
   • Extract resized base64 PNG (max 1024px) for PictureItem, TableItem, FormulaItem
   • Images without text go to NeonDB only (no embeddings)
   • Tables and text chunks with embedded images get base64 in Qdrant payload
5. Metadata Extraction (asyncio.to_thread) — GLiNER2 extracts:
   • Document-level: domain, industry, companies, language, confidentiality
   • Chunk-level: topics, products, technologies, organizations, locations, dates, metrics
6. PII Detection & Redaction (asyncio.to_thread) — GLiNER2 scans 42 PII types:
   • Detected spans are replaced with [PII_TYPE] placeholders
   • Original text preserved in audit log (not in embeddings)
7. Dense Embedding (asyncio.to_thread) — jina-omni-nano encodes text to 768d vectors
8. Sparse Embedding (asyncio.to_thread) — SPLADE v3 encodes to 30522d sparse vectors with semantic term expansion
9. Qdrant Upsert — Both vectors stored in same point with metadata payload (including base64 for table chunks)
10. NeonDB Registry — Document + chunk records saved with qdrant_point_id linkage (all chunks including images)

Concurrency Model

---

Configuration

All configuration is managed via src/pipeline/config.py using Pydantic Settings. It automatically loads from environment variables and .env files.

Key tunables:

---

Database Schema

documents Table

chunks Table

---

Development

Running Tests

uv run pytest tests/

Code Quality

uv run ruff check src/          # Linting
uv run ruff format src/          # Formatting
uv run mypy src/                 # Type checking

Prefect UI

View flow runs and task status in the Prefect UI:

uv run prefect server start
# Open http://localhost:4200

---

Makefile Commands

make help         # Show all available commands
make start        # Start Prefect server + Qdrant in Docker
make stop         # Stop all Docker services
make logs         # Follow container logs
make status       # Check container health
make migrate      # Run NeonDB migration
make install      # Install dependencies with UV
make lint         # Run ruff linter
make format       # Auto-format code
make test         # Run syntax + import checks
make flush        # Clear Qdrant + NeonDB (keeps S3, non-interactive)
make clean        # Stop services and remove volumes
make env          # Create .env from .env.example

---

Architecture Decisions

---

Troubleshooting

Model Download Issues

If Hugging Face model downloads fail:

export HF_TOKEN=your_token_here
# Or add to .env

Qdrant Connection Issues

For local Qdrant:

docker run -p 6333:6333 qdrant/qdrant

Prefect Flow Failures

Check flow logs:

uv run prefect flow-run logs <flow-run-id>

---

Contributing

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit changes (git commit -m 'feat: add amazing feature')
4. Push to branch (git push origin feature/amazing-feature)
5. Open a Pull Request

---

License

Apache-2.0

---

Acknowledgments

• Docling for document parsing
• GLiNER2 for metadata extraction and PII detection
• Jina AI for jina-embeddings-v5-omni-nano
• Qdrant for vector database
• Prefect for workflow orchestration