Research/chunking/docs/adr/0009-ingestion-pipeline.md

ADR 0009: Ingestion Pipeline

Status

Accepted

Context

Documents arrive via API endpoints. The system must process them continuously as they're received, making new content immediately searchable. The pipeline handles multiple file formats (PDF, Word, CSV, Excel, TXT) and produces chunks with dual embeddings and sparse vectors.

Decision

Architecture

API Endpoint (/ingest)
    ↓
File Validator (type, size, format)
    ↓
Document Converter → Markdown
    ↓
Chunker (semantic / row-based by type)
    ↓
Metadata Enricher (domain, language, source, etc.)
    ↓
Dual Embedder (OpenAI + Nomic)
    ↓
Sparse Vectorizer (BM25)
    ↓
Qdrant Indexer
    ↓
Response (chunk IDs, status)

Components

1. API Endpoint

• REST API receiving multipart file uploads
• Accepts: PDF, Word (.docx), CSV, Excel (.xlsx, .xls), TXT
• Returns: ingestion ID, chunk count, status

2. File Validator

• Check file type by extension and magic bytes
• Enforce size limits (e.g., 50MB max)
• Reject unsupported formats

3. Document Converter

• PDF → Markdown: pandoc or pymupdf
• Word → Markdown: pandoc or python-docx
• CSV/Excel → preserve as structured data, convert to markdown representation
• TXT → minimal processing, detect encoding

4. Chunker

• Unstructured (PDF, Word, TXT): Semantic chunking with sentence boundaries
• Tabular (CSV, Excel): Row-by-row chunking
• Preserve column names as metadata for tabular sources

5. Metadata Enricher

• Auto-detect language (langdetect)
• Domain from API parameter or file path
• Generate content hash (SHA-256)
• Add timestamps

6. Dual Embedder

• Always generate both OpenAI and Nomic embeddings
• Queue parallel API calls to OpenAI
• Run Nomic locally (on-prem)
• Handle OpenAI failures gracefully (continue with Nomic only)

7. Sparse Vectorizer

• Use Qdrant's built-in BM25 sparse indexing
• Tokenize text, generate sparse vector

8. Qdrant Indexer

• Upsert chunks with all vectors and metadata
• Return chunk IDs for confirmation

Continuous Processing

• API endpoint queues documents for processing
• Background workers pick up documents and process asynchronously
• Status endpoint to check ingestion progress
• Webhook or callback when ingestion completes (optional)

Consequences

Positive

• Immediate availability — new docs searchable within seconds
• Scalable — background workers can scale horizontally
• Resilient — OpenAI failures don't block ingestion (fallback to Nomic)
• Observable — status tracking for each ingestion job

Negative

• Complexity — async processing requires job queue and workers
• Resource usage — continuous embedding requires compute/GPU availability
• Error handling — partial failures need clear status reporting

Neutral

• Need to implement job queue (Redis, Celery, or similar)
• Consider rate limiting to prevent overload
• May need to batch small files for efficiency

Implementation Notes

• Use FastAPI for ingestion endpoint
• Use Celery + Redis for job queue
• Store ingestion logs and status in database
• Implement retry logic for transient failures
• Consider deduplication: check content_hash before inserting