RAG Ingestion Pipelines and Connectors

"The pipeline that runs while you sleep is worth more than the one that runs while you watch."

RAG, Battle-Tested AI Agent

1. Why Ingestion Quality Dominates Retrieval Quality

Every RAG system operates under a simple constraint: the retriever can only return documents that exist in the index, and the generator can only cite content that the retriever surfaces. If a PDF was parsed with mangled tables, the retriever will return mangled tables. If a Confluence page was never synced, the retriever cannot find it. If a 200-page document was split into 50-token fragments with no overlap, the retriever will return context-free snippets that confuse the generator. The principle is straightforward: garbage in, garbage out.

Production RAG failures are disproportionately caused by ingestion problems rather than retrieval or generation problems. A survey of deployed RAG systems by LlamaIndex (2024) found that roughly 60% of "wrong answer" failures traced back to either missing documents (never ingested), corrupted content (bad parsing), or poor chunk boundaries (relevant information split across chunks). Fixing ingestion is the highest-leverage improvement you can make to a RAG system.

Figure 20.8.1: The RAG ingestion pipeline. Data flows from diverse enterprise sources through authenticated connectors, document parsers, chunking strategies, and embedding models into the vector database. A scheduled orchestration layer triggers nightly batch syncs and incremental updates.

2. Connectors: Pulling Data from Everywhere

Enterprise knowledge lives in dozens of systems: wikis (Confluence, Notion), document stores (Google Drive, SharePoint), databases (PostgreSQL, MongoDB), ticketing systems (Jira, Linear), communication platforms (Slack, Teams), and SaaS applications (Salesforce, HubSpot). A production RAG system needs connectors that can authenticate with each source, extract content, track changes, and handle rate limits.

Airbyte is an open-source data integration platform with over 350 pre-built connectors. Originally designed for ELT (Extract, Load, Transform) workloads, Airbyte works equally well as a RAG ingestion frontend. Each connector handles authentication, pagination, rate limiting, and schema mapping for its source. Airbyte supports incremental sync, meaning it tracks a cursor (such as a last-modified timestamp) and only fetches records that changed since the last sync. This is essential for RAG systems that need nightly refreshes without re-ingesting the entire corpus.

# airbyte-connection.yaml
# Airbyte connection config: Confluence to local JSON staging
source:
 sourceDefinitionId: "confluence-source-v2"
 connectionConfiguration:
 domain: "yourcompany.atlassian.net"
 email: "${CONFLUENCE_EMAIL}"
 api_token: "${CONFLUENCE_API_TOKEN}"
 space_keys:
 - "ENG"
 - "PRODUCT"
 - "OPS"

destination:
 destinationDefinitionId: "local-json"
 connectionConfiguration:
 destination_path: "/data/staging/confluence"

sync:
 schedule:
 units: 24
 timeUnit: "hours"
 syncMode: "incremental"
 cursorField: "lastModified"

For simpler setups, LlamaIndex data connectors (formerly LlamaHub) provide lightweight Python classes for common sources. These are easier to integrate into a custom pipeline but lack Airbyte's built-in scheduling, monitoring, and incremental sync tracking. A practical middle ground is to use Airbyte for high-volume, frequently-changing sources (wikis, databases) and LlamaIndex connectors for static or infrequently-updated sources (local file directories, one-time PDF batches).

3. Document Parsing: Extracting Clean Text from Messy Formats

Raw documents arrive in diverse formats: PDF, HTML, DOCX, PPTX, XLSX, Markdown, images with embedded text, and scanned documents. Each format requires a different extraction strategy. The Unstructured library provides a unified API for parsing all of these formats. Its partition function automatically detects the file type and applies the appropriate extraction strategy, returning a list of Element objects (titles, narrative text, tables, list items, images) with metadata.

from unstructured.partition.auto import partition

# Parse a PDF with OCR fallback for scanned pages
elements = partition(
 filename="quarterly_report.pdf",
 strategy="hi_res", # Use layout detection model
 pdf_infer_table_structure=True, # Extract tables as HTML
 languages=["eng"],
 extract_image_block_types=["Image", "Table"],
)

# Each element carries type, text, and metadata
for el in elements:
 print(f"[{el.category}] {el.text[:80]}...")
 print(f" Page: {el.metadata.page_number}")
 print(f" Coordinates: {el.metadata.coordinates}")

# Filter to only narrative content and tables
content_elements = [
 el for el in elements
 if el.category in ("NarrativeText", "Title", "Table", "ListItem")
]

Unstructured offers three partition strategies. The fast strategy uses rule-based text extraction and works well for clean, text-native PDFs. The ocr_only strategy runs Tesseract OCR on every page, suitable for scanned documents. The hi_res strategy combines a layout detection model with OCR, providing the best quality for mixed documents (some pages text-native, some scanned) at the cost of higher latency. For most production workloads, hi_res is worth the extra processing time because it correctly identifies document structure (headers, paragraphs, tables, figures) rather than treating every page as a flat stream of text.

Apache Tika is an alternative for organizations that prefer a Java-based, server-mode parser. Tika supports over 1,000 file formats and runs as a REST service, making it easy to integrate with any language. It excels at metadata extraction (author, creation date, language) and handles edge cases like password-protected Office documents. However, Tika's table extraction and layout detection are less sophisticated than Unstructured's hi_res strategy. Many production systems use both: Tika for metadata extraction and format detection, Unstructured for content extraction.

import requests

# Apache Tika running as a REST service on localhost:9998
TIKA_URL = "http://localhost:9998"

def extract_with_tika(filepath: str) -> dict:
 """Extract text and metadata using Apache Tika server."""
 with open(filepath, "rb") as f:
 # Get parsed text
 text_resp = requests.put(
 f"{TIKA_URL}/tika",
 data=f,
 headers={"Accept": "text/plain"},
 )

 with open(filepath, "rb") as f:
 # Get metadata separately
 meta_resp = requests.put(
 f"{TIKA_URL}/meta",
 data=f,
 headers={"Accept": "application/json"},
 )

 return {
 "text": text_resp.text,
 "metadata": meta_resp.json(),
 "content_type": meta_resp.json().get("Content-Type", "unknown"),
 }

result = extract_with_tika("contract.docx")
print(f"Type: {result['content_type']}")
print(f"Author: {result['metadata'].get('Author', 'unknown')}")
print(f"Text length: {len(result['text'])} chars")

4. Chunking at Ingestion Time

Once documents are parsed into clean text elements, the next step is chunking: splitting content into pieces sized appropriately for embedding and retrieval. The chunking strategy directly affects retrieval precision and recall. Chunks that are too small lose context; chunks that are too large dilute relevance signals and waste context window budget. There is no universally optimal chunk size, but the choice should be deliberate and informed by the content type and the downstream use case.

4.1 Fixed-Window and Sliding-Window Chunking

The simplest approach splits text into fixed-size windows (e.g., 512 tokens) with a configurable overlap (e.g., 64 tokens). Overlap ensures that sentences near chunk boundaries appear in at least two chunks, reducing the chance that relevant information is split across a boundary. This method is fast and predictable but blind to document structure: a chunk boundary can fall in the middle of a paragraph, a code block, or a table.

4.2 Semantic Chunking

Semantic chunking uses embedding similarity to find natural breakpoints. The algorithm computes embeddings for each sentence (or small segment), then places a chunk boundary wherever the cosine similarity between consecutive segments drops below a threshold. This produces chunks that are internally coherent: each chunk covers a single topic or idea. The trade-off is that chunk sizes become variable, and the embedding computation adds latency to the ingestion pipeline.

4.3 Document-Structure-Aware Splitting

The most effective approach for structured documents (technical documentation, legal contracts, research papers) is to split along document structure boundaries. Use section headers as primary split points. Keep tables, code blocks, and figures as atomic units (never split a table across two chunks). Attach parent section titles as metadata so the retriever can use hierarchical context. Unstructured's Element types make this straightforward: group consecutive NarrativeText elements under their preceding Title element, and treat each Table as its own chunk with the parent section title prepended.

from langchain.text_splitter import RecursiveCharacterTextSplitter
from unstructured.partition.auto import partition

def structure_aware_chunk(filepath: str, max_chunk_tokens: int = 512):
 """Chunk a document respecting its structural boundaries."""
 elements = partition(filename=filepath, strategy="hi_res")

 chunks = []
 current_section = "Introduction"
 buffer = []
 buffer_len = 0

 for el in elements:
 if el.category == "Title":
 # Flush buffer as a chunk when we hit a new section
 if buffer:
 chunks.append({
 "text": "\n".join(buffer),
 "section": current_section,
 "page": el.metadata.page_number,
 })
 current_section = el.text
 buffer = []
 buffer_len = 0

 elif el.category == "Table":
 # Tables are always their own chunk
 if buffer:
 chunks.append({
 "text": "\n".join(buffer),
 "section": current_section,
 "page": el.metadata.page_number,
 })
 buffer = []
 buffer_len = 0
 chunks.append({
 "text": f"[Table in {current_section}]\n{el.text}",
 "section": current_section,
 "page": el.metadata.page_number,
 "is_table": True,
 })

 else:
 # Accumulate narrative text, flush at size limit
 token_est = len(el.text.split())
 if buffer_len + token_est > max_chunk_tokens:
 chunks.append({
 "text": "\n".join(buffer),
 "section": current_section,
 "page": el.metadata.page_number,
 })
 buffer = []
 buffer_len = 0
 buffer.append(el.text)
 buffer_len += token_est

 # Flush remaining buffer
 if buffer:
 chunks.append({"text": "\n".join(buffer), "section": current_section})

 return chunks

5. Incremental Refresh: Keeping the Index Current

A production RAG system must keep its vector index synchronized with the source documents. Full re-ingestion (parsing, chunking, embedding, and upserting every document on every sync) works for small corpora but becomes prohibitively expensive at scale. A corpus of 100,000 documents with an average embedding cost of $0.0001 per chunk and 10 chunks per document costs $100 per full re-index. If only 2% of documents changed since the last sync, incremental refresh reduces that cost to $2.

5.1 Change Detection

Track a content hash (SHA-256 of the raw document bytes) and a last-modified timestamp for every ingested document. On each sync run, compare the current hash against the stored hash. If the hash matches, skip the document entirely. If the hash differs, re-parse, re-chunk, and re-embed only that document. Store these hashes in a metadata table alongside the vector IDs so you can locate and replace the affected vectors.

5.2 Tombstone Handling

When a source document is deleted or archived, the corresponding vectors must be removed from the index. Otherwise, the retriever will continue surfacing content from documents that no longer exist, leading to stale or contradictory answers. Implement a tombstone pass at the end of each sync: compare the set of document IDs in the source against the set of document IDs in the vector store. Any ID present in the vector store but absent from the source is a candidate for deletion. Log tombstoned documents before deleting their vectors, and consider a grace period (e.g., 7 days) before permanent deletion to handle temporary source outages.

import hashlib
from datetime import datetime, timedelta

class IncrementalSyncTracker:
 """Track document versions for incremental RAG index updates."""

 def __init__(self, metadata_store):
 self.store = metadata_store # e.g., SQLite or PostgreSQL table

 def compute_hash(self, content: bytes) -> str:
 return hashlib.sha256(content).hexdigest()

 def needs_update(self, doc_id: str, content: bytes) -> bool:
 """Check whether a document needs re-ingestion."""
 new_hash = self.compute_hash(content)
 record = self.store.get(doc_id)
 if record is None:
 return True # New document
 return record["content_hash"] != new_hash

 def record_ingestion(self, doc_id: str, content: bytes, vector_ids: list):
 """Record successful ingestion for future change detection."""
 self.store.upsert({
 "doc_id": doc_id,
 "content_hash": self.compute_hash(content),
 "vector_ids": vector_ids,
 "ingested_at": datetime.utcnow().isoformat(),
 })

 def find_tombstones(self, active_doc_ids: set) -> list:
 """Find documents in the index that no longer exist in the source."""
 indexed_ids = set(self.store.all_doc_ids())
 tombstoned = indexed_ids - active_doc_ids
 return [
 self.store.get(doc_id)
 for doc_id in tombstoned
 ]

 def apply_tombstones(self, vector_store, grace_days: int = 7):
 """Delete vectors for documents removed more than grace_days ago."""
 cutoff = datetime.utcnow() - timedelta(days=grace_days)
 for record in self.store.get_tombstoned_before(cutoff):
 vector_store.delete(ids=record["vector_ids"])
 self.store.delete(record["doc_id"])

6. Data Quality Checks

Ingestion pipelines should validate data quality before writing vectors to the index. Without validation, the index accumulates duplicates, empty documents, malformed content, and off-topic material that degrades retrieval quality over time.

6.1 Deduplication with MinHash

Near-duplicate documents are common in enterprise corpora: the same policy document saved in three locations, multiple drafts of the same report, wiki pages copied across spaces. Exact deduplication (hash matching) catches identical copies, but near-duplicates require fuzzy matching. MinHash with Locality-Sensitive Hashing (LSH) estimates Jaccard similarity between document shingle sets in sublinear time. Documents with similarity above a threshold (commonly 0.8) are flagged as duplicates; only the most recent version is indexed.

6.2 Format Validation and Schema Enforcement

Define a minimum quality schema for ingested chunks. At minimum, validate that: (1) the chunk text is not empty or whitespace-only, (2) the chunk text contains at least 20 tokens (very short chunks are usually parsing artifacts), (3) required metadata fields (source URL, document ID, ingestion timestamp) are present, and (4) the text encoding is valid UTF-8 without control characters. Chunks that fail validation are logged for review rather than silently dropped, because a spike in validation failures often signals a connector or parser bug.

7. Orchestration: Scheduling and Monitoring Pipelines

A complete ingestion pipeline involves multiple steps: connector sync, parsing, chunking, validation, embedding, upserting to the vector store, and tombstone cleanup. Each step can fail independently. A robust pipeline orchestrator handles scheduling, retries, dependency management, and lineage tracking. Two popular choices for Python-based pipelines are Dagster and Prefect.

Dagster models pipelines as directed acyclic graphs of "assets." Each asset represents a materialized data product (e.g., the set of parsed elements, the set of validated chunks, the updated vector index). Dagster tracks dependencies between assets, supports incremental materialization (only re-run assets whose upstream inputs changed), and provides a web UI for monitoring and debugging. Its "software-defined assets" paradigm maps naturally onto ingestion pipeline stages.

Prefect takes a more imperative approach: decorate Python functions with @task and @flow to get automatic retries, logging, and scheduling. Prefect is often easier to adopt for teams already writing ingestion logic as Python scripts, because the migration path is to add decorators to existing functions. Both tools support cron-based scheduling, webhook triggers, and integration with alerting systems (Slack, PagerDuty).

from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta

@task(retries=3, retry_delay_seconds=60,
 cache_key_fn=task_input_hash,
 cache_expiration=timedelta(hours=24))
def sync_confluence(space_key: str) -> list[dict]:
 """Pull changed pages from Confluence since last sync."""
 connector = ConfluenceConnector(space_key=space_key)
 return connector.fetch_incremental()

@task(retries=2, retry_delay_seconds=30)
def parse_documents(raw_docs: list[dict]) -> list[dict]:
 """Parse raw documents into structured elements."""
 parsed = []
 for doc in raw_docs:
 elements = partition(file=doc["content"], strategy="hi_res")
 parsed.append({"doc_id": doc["id"], "elements": elements})
 return parsed

@task
def chunk_and_validate(parsed: list[dict]) -> list[dict]:
 """Chunk parsed elements and validate quality."""
 chunks = []
 for doc in parsed:
 doc_chunks = structure_aware_chunk(doc["elements"])
 for chunk in doc_chunks:
 if len(chunk["text"].split()) >= 20:
 chunk["doc_id"] = doc["doc_id"]
 chunks.append(chunk)
 return chunks

@task(retries=2, retry_delay_seconds=10)
def embed_and_upsert(chunks: list[dict]) -> int:
 """Embed chunks and upsert to the vector store."""
 embeddings = embed_batch([c["text"] for c in chunks])
 vector_store.upsert(vectors=embeddings, metadata=chunks)
 return len(chunks)

@flow(name="nightly-rag-ingestion", log_prints=True)
def nightly_ingestion():
 """Orchestrate the full nightly RAG ingestion pipeline."""
 spaces = ["ENG", "PRODUCT", "OPS"]
 total = 0

 for space in spaces:
 raw = sync_confluence(space)
 parsed = parse_documents(raw)
 chunks = chunk_and_validate(parsed)
 count = embed_and_upsert(chunks)
 total += count
 print(f"[{space}] Ingested {count} chunks")

 print(f"Total chunks ingested: {total}")

# Schedule: run every night at 2 AM UTC
# prefect deployment build nightly_ingestion.py:nightly_ingestion \
# --name "nightly-rag" --cron "0 2 * * *"

Lab: Building a Living Knowledge Base

Exercises