Data Pipelines for RAG & AI Agents

01 Overview

A RAG or agent system is only as good as the data it sees. The pipeline is the part of the system that decides what data exists, in what shape, when it updates, and how the model finds it.

Most production failures of RAG systems and AI agents trace back to the pipeline, not the model. "Bad answers" are usually really one of:

• The right document existed but wasn't retrieved (bad chunking, wrong embedding, missing filter).
• The retrieved chunk was stale or duplicated (bad freshness or deduplication).
• The chunk was retrieved but the metadata was wrong (bad enrichment).
• The agent had the right tool but wrong context (bad context construction).

02 Why pipelines matter

Three axes drive the design: scale (millions of documents), freshness (seconds to days), and quality (clean, deduplicated, attributed).

03 RAG pipeline architecture

The canonical RAG data pipeline is a 7-stage flow from raw source to answerable retrieval. Each stage has its own failure modes and tools.

Sections 04–10 below walk through each stage in detail.

04 Stage 1 — Ingestion

Ingestion is where data enters the pipeline. The dominant question: push or pull, batch or stream.

Common source types

Design rules

• Idempotent. Replaying the same source must not produce duplicates downstream — use deterministic IDs (e.g. sha256(source_url + version)).
• Preserve raw. Land raw bytes in object storage before any transformation. Re-derivation is the cheapest way to fix a bad parser.
• Atomic units. Each ingest unit (a document, a row, an event) carries a stable ID and a source-version stamp.
• Backpressure. A bursty source (a CDC catch-up, a re-crawl) must not knock the embedding service over — queue between stages.

05 Stage 2 — Parsing & cleaning

Raw bytes become structured text. The quality of this stage caps the quality of every downstream stage — a garbage parser produces garbage chunks.

By format

Cleaning steps

• Normalize whitespace, line endings, unicode (NFC).
• Strip boilerplate (headers, footers, navigation, cookie banners).
• Detect language; route per-language pipelines if needed.
• Detect and tag content type (prose, code, table, list).
• Redact or tokenize PII at the earliest stage where it's safe to do so.
• Hash content for deduplication keys (exact + near-dup with MinHash/SimHash).

06 Stage 3 — Chunking

Chunking splits documents into retrieval units. Too large and the LLM gets diluted context; too small and you lose meaning. Boundaries should respect semantic structure.

Strategies

Sizing rules of thumb

• Embedding model context. Stay well under the embedder's max tokens — quality often degrades long before the limit.
• Target 200–800 tokens per chunk for general prose. Smaller for QA, larger for narrative.
• Overlap 10–20% to preserve boundary context — but not so much that the index doubles in size.
• Always include a headline / breadcrumb at the top of each chunk: "<Doc Title> > <Section>\n\n<chunk text>". Hugely improves retrieval and makes the LLM cite better.

07 Stage 4 — Enrichment & metadata

Each chunk carries metadata used for filtering, ranking, attribution, and access control. Underinvested metadata is one of the most common pipeline failures.

Required metadata

08 Stage 5 — Embedding

Convert each chunk to a vector. The embedding model is the largest single quality lever in retrieval — pick deliberately and version explicitly.

Model selection

Operational concerns

• Version everything. An embedding from text-embedding-3-large is not comparable to one from 3-small. Tag every vector with embedder_id + version.
• Re-embedding is expensive. Plan for it. Keep the original chunk text addressable so you can re-embed without re-parsing.
• Batch. Most providers charge per request and per token — batch 100s of chunks per call.
• Normalize. Most cosine-similarity stores expect L2-normalized vectors. Check your store's expectations.
• Symmetric vs asymmetric. Some models distinguish between passage and query embeddings — use the right prompt template.

# Batched embedding with versioning
from openai import OpenAI
client = OpenAI()

def embed_batch(chunks):
    resp = client.embeddings.create(
        model="text-embedding-3-large",
        input=[c.text for c in chunks],
    )
    for chunk, item in zip(chunks, resp.data):
        chunk.vector = item.embedding
        chunk.meta["embedder"] = "openai/text-embedding-3-large"
        chunk.meta["embedder_dim"] = 3072
    return chunks

09 Stage 6 — Indexing

Vectors land in a vector store; sparse representations land in a keyword index; metadata lands in a filterable store. In practice you usually want all three, often in the same product.

Vector stores compared

Index design

• Hybrid by default. Combine dense (semantic) + sparse (keyword) — neither alone handles both rare terms and paraphrases well.
• Filterable metadata. Tenancy, ACL, doc type, date — first-class filters on the vector store, not post-filtering.
• Tiered storage. Hot tier in memory, cold tier on disk/S3 — reflect access patterns.
• Sharding strategy. Per-tenant shards prevent noisy neighbors but raise operational cost; pick based on traffic shape.
• Reindex playbook. Document the steps to rebuild the index from raw — you will need it.

10 Stage 7 — Retrieval & rerank

At query time: rewrite the query, retrieve candidates with hybrid search, rerank, then assemble the LLM context.

The retrieval flow

1. Query understanding. Rewrite, expand, decompose. Multi-query generates 3–5 paraphrases. HyDE generates a hypothetical answer and embeds that.
2. Filter. Apply ACLs, tenant, freshness window — never skip this.
3. Hybrid retrieve. Top-K dense + top-K sparse, then merge (Reciprocal Rank Fusion is the standard).
4. Rerank. Cross-encoder rerank the top 30–100 candidates down to top 5–10. Cohere Rerank, Voyage Rerank, BGE Reranker are the default options.
5. Assemble context. Add breadcrumbs and citations; deduplicate near-identical chunks; budget tokens.

Retrieval techniques worth knowing

11 Agent data pipelines

Agents need more than a vector store. They consume four distinct data substrates — knowledge, tool data, memory, and traces — and each needs its own pipeline.

12 Agent context engineering

"Context engineering" is the practice of deciding what enters the model's context window on each step. It is the agent equivalent of feature engineering.

Context inputs (per agent step)

13 Agent memory pipelines

Memory is data that persists across sessions. Treat it as a write-heavy pipeline with its own ingestion, scoring, and eviction.

Memory types

The memory write pipeline

1. Extract. After each turn, an LLM call extracts candidate memories from the conversation.
2. Score. Confidence + utility + novelty. Drop low-value candidates.
3. Deduplicate. Embed and check against existing memories — update if similar exists, insert otherwise.
4. Tag. Attach user_id, session_id, source, expiry.
5. Persist. Write to memory store + vector index.

The memory read pipeline

1. Embed the current context.
2. Hybrid retrieve from memory store, filter by user_id and validity.
3. Rerank by relevance + recency + confidence.
4. Inject top-N into the system or user prompt with clear provenance ("Recalled from your prior session: …").

14 Batch vs streaming

The biggest architectural decision after "what's in the corpus" is "how fresh does it need to be." Batch is cheaper and simpler; streaming is fresher and operationally heavier.

Hybrid: micro-batch

The pragmatic middle ground: a streaming consumer batches events into 10–60 second windows, then runs the batch pipeline on each window. You get near-real-time freshness with batch's testability.

15 Freshness & incremental updates

Production pipelines must be incremental — full reindexes are an anti-pattern past a few thousand docs. The hard problem is doing inserts, updates, and deletes safely.

Update patterns

Soft delete vs hard delete

Tombstone (soft delete) is usually safer in production — set deleted_at, exclude in retrieval, GC asynchronously. Hard delete is required for GDPR / right-to-be-forgotten requests; build the GC path to actually remove vectors and audit-log the action.

16 Data quality & lineage

Quality controls and lineage tracking turn a research-grade pipeline into a production-grade one.

Quality controls to add

• Deduplication. Exact (hash) + near-dup (MinHash, SimHash) at chunk level. Index size shrinks 20–50% on real corpora.
• PII scanning. Detect and redact (or block) sensitive data before embedding. Tools: Presidio, AWS Macie, Nightfall.
• Toxic/harmful content filters at ingest — don't let bad data into retrieval.
• Language detection. Route per language or filter to supported languages.
• Freshness SLA. Track p50 / p95 ingestion-to-queryable latency; alert on regression.
• Chunk health metrics. Distribution of chunk sizes, % empty, % code, duplicate rate.

Lineage

For every chunk in the index, you should be able to answer:

• What raw document did this come from?
• What parser / chunker / embedder version produced it?
• When was it ingested? When last updated?
• Who has permission to retrieve it?

Lineage metadata enables incident response, compliance, and intelligent reprocessing. Tools: OpenLineage, DataHub, Marquez.

17 Observability & eval

A pipeline you can't measure is a pipeline you can't improve. Instrument every stage and every retrieval.

Pipeline metrics

Eval frameworks

What to evaluate

• Retrieval quality — labeled query/passage set; recall@k, MRR, nDCG.
• Answer faithfulness — does the answer follow from the retrieved context?
• Answer relevance — does it actually answer the question?
• Context precision — were the retrieved chunks actually used?
• Agent task success — did the multi-step workflow reach the right outcome?

18 Orchestration tools

You can hand-roll pipelines, but past a certain scale you'll want orchestration. The choice shapes how easy retries, backfills, and observability are.

19 Reference tech stack

A representative production stack as of 2026. Substitute components freely — every layer has multiple credible options.

20 End-to-end example

A minimal but production-shaped pipeline for a customer-support RAG system. Adapt the details — the structure generalizes.

# pipeline.py — simplified end-to-end
from dataclasses import dataclass
import hashlib

@dataclass
class Chunk:
    id: str
    text: str
    source_id: str
    section_path: str
    meta: dict
    vector: list | None = None

def stable_id(source_id: str, section: str, idx: int) -> str:
    return hashlib.sha256(f"{source_id}:{section}:{idx}".encode()).hexdigest()[:16]

def run(source_uri: str):
    # 1. Ingest — fetch raw bytes, persist immutable copy
    raw = ingest(source_uri)                          # bytes
    raw_uri = persist_raw(raw, source_uri)            # s3://raw/...

    # 2. Parse — extract structured text
    doc = parse(raw, source_uri)                      # Document(sections=[...])

    # 3. Chunk — structural + recursive
    chunks = []
    for section in doc.sections:
        for i, piece in enumerate(recursive_split(section.text, target=600, overlap=80)):
            chunks.append(Chunk(
                id=stable_id(doc.source_id, section.path, i),
                text=f"{doc.title} > {section.path}\n\n{piece}",
                source_id=doc.source_id,
                section_path=section.path,
                meta={
                    "source_uri": source_uri,
                    "updated_at": doc.updated_at,
                    "acl": doc.acl,
                    "doc_type": doc.doc_type,
                    "parser_version": "v3",
                    "chunker_version": "v2",
                },
            ))

    # 4. Embed — batch
    for batch in batched(chunks, 100):
        embed_batch(batch)                            # sets .vector + meta["embedder"]

    # 5. Index — upsert keyed by stable id (idempotent)
    vector_store.upsert(chunks)
    keyword_index.upsert(chunks)

    # 6. Tombstone removed chunks
    existing_ids = vector_store.ids_for(source_id=doc.source_id)
    new_ids = {c.id for c in chunks}
    for dead in existing_ids - new_ids:
        vector_store.tombstone(dead)

    # 7. Emit lineage event
    emit_lineage(doc.source_id, len(chunks), embedder="voyage-3")

Note the design: deterministic IDs make every step idempotent; raw bytes are preserved so you can re-derive without re-fetching; tombstones handle deletes; lineage events feed observability.

21 Anti-patterns

Mistakes that look reasonable but cause grief in production. Recognize and avoid.

22 Production checklist

Before calling a pipeline production-ready, you should be able to check every item below.

Correctness

• Ingest is idempotent — replay produces no duplicates.
• Updates and deletes propagate to all indexes.
• Every chunk has provenance metadata (source, version, timestamps).
• ACLs and tenant filters are applied at query time, not after.
• Embedder and chunker versions are tracked per chunk.

Quality

• Eval set with labeled queries exists; nDCG / recall reported on every change.
• Faithfulness / groundedness measured on a regression set.
• Dedup pass runs at ingest (exact + near-dup).
• PII redaction or block list active on all sources.

Operability

• Raw bytes preserved → reprocessable from scratch.
• Tracing on every prompt, tool call, retrieval (OpenTelemetry).
• Dashboards for ingestion lag, retrieval latency, eval scores.
• Reindex playbook documented and rehearsed.
• Cost dashboards per stage (embedding $, vector store $, LLM $).

Safety

• Right-to-be-forgotten (hard delete) path implemented and tested.
• Memory writes are audit-logged.
• Per-tenant isolation verified by automated test.
• Rate limits and cost caps on agent loops.