AI Data Pipelines: How to Build One That Doesn't Go Stale

Quick Answer: An AI data pipeline turns source content into the retrievable form a model consumes — parse, chunk, embed, index — and then keeps that index in sync as the source changes. The hard part is never the first load. It is everything after: propagating edits without leaving orphaned chunks behind, making deletions actually disappear, and re-embedding when the model version changes. Treat the vector index as a derived table rather than a database, and every one of those problems already has a known data engineering answer.

Most teams build an AI data pipeline twice. The first version is a script: read the documents, split them, embed them, write them to a vector store. It works, the demo is impressive, and it ships. The second version is written three months later, after someone notices the assistant confidently quoting a policy that was repealed in April.

That gap is where data engineering has quietly become one of the most in-demand skills in the industry. Data engineering hiring grew 23% year over year, and the global data engineering services market is estimated at $105.39 billion in 2026. Not because AI replaced the discipline, but because AI added a consumer that is far less forgiving than a dashboard.

This guide covers the parts of an AI data pipeline that only show up in production: the chunk-boundary trap that leaves deleted text retrievable, why deletion is genuinely hard in a vector index, what an embedding model upgrade really costs, and how to pick a freshness target you can defend.

What is an AI data pipeline?

An AI data pipeline is the system that converts raw source content into a form a model can retrieve, and keeps that form correct as the source changes. The classic shape has five stages:

1. Ingest — pull documents from their systems of record: a wiki, an S3 bucket, a ticketing system, a database.
2. Parse — extract clean text and structure from PDFs, HTML, slides, or transcripts.
3. Chunk — split each document into passages small enough to retrieve precisely and large enough to stay meaningful.
4. Embed — convert each chunk into a vector using an embedding model.
5. Index — write vectors plus their metadata into a vector database or a vector-capable table.

Every tutorial covers those five stages, including our own step-by-step RAG pipeline walkthrough. What tutorials rarely cover is the sixth stage, which is the one that actually decides whether the system is trustworthy six months in: synchronisation. Sources change. Documents get edited, archived, access-revoked, and deleted. A pipeline that only knows how to add is not a pipeline — it is an import.

How it differs from a traditional ETL pipeline

The engineering principles carry over almost entirely, which is exactly why data engineers are good at this work. But three things genuinely change:

That last row is the one that catches teams out. When a traditional pipeline breaks, a dashboard goes empty and someone files a ticket within the hour. When an AI pipeline breaks, the assistant keeps answering — fluently, in the right tone, citing a document that no longer exists. There is no error to alert on.

The reframe: your vector index is a derived table

Here is the idea that makes the rest of this article obvious.

A vector index is not a database. It is a materialised view over your source content. It holds no original information. Every vector in it is a lossy, derived copy of something that lives authoritatively somewhere else.

Once you see it that way, the questions stop being novel AI problems and become familiar data engineering ones:

• "How do I handle document edits?" is incremental processing.
• "How do I stop duplicate chunks when the job retries?" is idempotency.
• "How do I detect that the source changed?" is change data capture.
• "How do I rebuild after fixing a parsing bug?" is a backfill.
• "How do I know the index is still right?" is data quality testing.
• "Who is allowed to change the embedding model?" is a data contract.

None of this is new. It is the same discipline pointed at a new output format. That is why the fastest-growing role in this space is the hybrid one — a point covered in more depth in data engineer vs AI engineer.

The rest of this guide works through the four sync problems that have no exact equivalent in a row-based world.

Problem 1: The chunk-boundary trap

This is the single most common correctness bug in production RAG systems, and it hides in the write path.

Suppose a document is edited — an author removes three paragraphs from a policy page. Your pipeline detects the change, re-parses the document, re-chunks it, re-embeds it, and writes the result. Reasonable.

But the document is shorter now. Where version 1 split into 12 chunks, version 2 splits into 9. If your write is an upsert keyed by chunk ID, chunks 1 through 9 are overwritten with fresh content and chunks 10, 11, and 12 are never touched. They sit in the index holding text that has been deleted from the source.

Those orphans are the worst kind of bug because nothing about them looks broken. The vectors are well-formed. The text is grammatical. It scores high on semantic similarity because it genuinely is about the topic being asked. It is simply no longer true.

The fix: the document is the unit of replacement

The rule is short enough to put in a code review checklist:

Never upsert chunks. Delete every chunk belonging to the document, then insert the new version whole.

def sync_document(index, doc_id: str, chunks: list[str], embeddings: list[list[float]]):
    """Replace a document's chunks atomically. Safe to re-run."""
    # 1. Remove ALL existing chunks for this document, however many there were.
    index.delete(filter={"doc_id": doc_id})

    # 2. Insert the new version with deterministic IDs.
    index.upsert([
        {
            "id": f"{doc_id}::{i}",
            "values": embedding,
            "metadata": {
                "doc_id": doc_id,
                "chunk_index": i,
                "text": chunk,
                "content_hash": sha256(chunk.encode()).hexdigest(),
                "embedding_model": EMBEDDING_MODEL,
                "indexed_at": datetime.now(timezone.utc).isoformat(),
            },
        }
        for i, (chunk, embedding) in enumerate(zip(chunks, embeddings))
    ])

Three details in there matter more than they look:

• doc_id in the metadata is what makes delete-by-filter possible. If you cannot cheaply select every chunk of a document, you cannot replace it correctly. Add it on day one; retrofitting it means a full reindex.
• Deterministic IDs (doc_id::chunk_index) make the insert idempotent, so a retried task cannot produce duplicates.
• content_hash lets you skip work. Hash each chunk, compare against what is indexed, and skip documents whose content is unchanged. On a corpus where 2% of documents change daily, this turns a full re-embed into a 2% re-embed.

Skip the embedding call, not just the write

That last point is where the money is. Embedding is the expensive stage, so the comparison must happen before you call the model, not after:

existing = {m["chunk_index"]: m["content_hash"] for m in index.fetch_metadata(doc_id)}
new_hashes = [sha256(c.encode()).hexdigest() for c in chunks]

if len(existing) == len(chunks) and all(
    existing.get(i) == h for i, h in enumerate(new_hashes)
):
    return  # nothing changed — no embedding calls, no writes

At SolutionGigs we run an internal assistant over our own documentation and support history. The first version re-embedded the entire corpus nightly because it was simple, and the bill was small enough that nobody looked. What eventually forced a rewrite was not cost — it was that a full nightly rebuild has a window during which the index is half-old and half-new, and answers during that window were inconsistent in ways we could not reproduce the next morning. Content hashing removed the window along with 98% of the work.

Problem 2: Deletions are harder than they look

Adding vectors is easy. Removing them is not, for two separate reasons.

Reason one: you have to notice

Most sync jobs are built around "find things that changed since the last run." Deleted documents do not appear in that list — they appear in no list at all. They are absent, and absence is invisible to an incremental query.

Two workable approaches:

• Log-based CDC. If the source is a database, the write-ahead log emits explicit DELETE events. This is the reliable option and the reason Debezium-style CDC shows up so often in AI infrastructure.
• Periodic reconciliation. List every document ID in the source, list every distinct doc_id in the index, and delete the difference. Cheap to run weekly, and it catches everything the incremental path missed — including bugs in the incremental path.

Run the reconciliation even if you have CDC. It is your only independent check that the index still matches reality.

Reason two: vector indexes resist deletion

Deleting from a graph-based vector index usually does not free anything. HNSW and similar structures store vectors as nodes in a navigable graph. Removing a node outright would sever the links that make search work, so implementations typically write a tombstone instead: the vector is filtered out of results but stays in memory and on disk.

The practical consequences:

• Deleted vectors keep consuming RAM, which is the expensive resource in vector search.
• On a high-churn corpus, tombstones accumulate and search slows as the graph fills with skipped nodes.
• Space is only reclaimed by a compaction or a full index rebuild.

If that pattern sounds familiar, it should — it is the same maintenance problem as Iceberg table upkeep, where deletes write markers and a periodic compaction does the real cleanup. Schedule vector index compaction on the same footing as any other table maintenance job.

The exception: security deletions cannot wait for a batch

There is one case where none of the above is fast enough. When someone's access is revoked, or a customer exercises a deletion right, a document must stop being retrievable now — not at 2 a.m.

Do not solve this in the indexing pipeline. Solve it at query time, with a metadata filter:

results = index.query(
    vector=query_embedding,
    top_k=10,
    filter={
        "tenant_id": user.tenant_id,
        "acl_group": {"$in": user.groups},
        "deleted": False,
    },
)

The pipeline eventually removes the vectors; the filter makes them unreachable immediately. Store the permission data as chunk metadata and enforce it on every query. An index built without tenant and ACL metadata cannot be retrofitted with it — that is a full reindex — so decide before the first load, not after the first incident.

Problem 3: Your embedding model is a schema

Changing the embedding model is a breaking schema change, and it is the most under-estimated migration in AI infrastructure.

Different models place vectors in different geometric spaces. A vector from text-embedding-3-small and a vector from a newer model are not comparable, even at the same dimensionality. Mixing them in one index does not throw an error — it produces similarity scores that are quietly meaningless, which is the worst possible failure mode.

So the model version is not a config value. It is part of the contract that describes your index, exactly like a column type. It belongs in a data contract with an owner and a change process, not in an environment variable someone can bump on a Friday.

What the migration actually costs

The API cost is usually the smallest line item. Current OpenAI embedding pricing puts text-embedding-3-small at $0.02 per million tokens and text-embedding-3-large at $0.13 per million tokens, with the Batch API halving both.

For a corpus of one million documents averaging 500 tokens — 500 million tokens total:

Ten dollars to rebuild a million-document index. The API bill is not the reason teams delay these migrations.

The real costs are elsewhere: storage and memory double while both indexes coexist, text-embedding-3-large needs twice the vector storage of -small at 3,072 dimensions versus 1,536, rate limits stretch the backfill over hours or days, and retrieval quality can regress in ways that only show up in specific query types. A model that scores better on public benchmarks can still be worse on your particular corpus.

Run it as a blue/green migration

1. Build a fixed evaluation set first — 100 to 200 real queries with known-correct documents. Without this you cannot tell an upgrade from a regression, and you will not be able to build it honestly after you have seen the new model's output.
2. Score the current index. This is your baseline. Skipping it is how teams end up unable to answer "was it better before?"
3. Build the new index alongside the old one. Never write both models' vectors into the same index.
4. Re-score. Compare recall@k and MRR against the baseline, and segment by query type — aggregate averages hide category-specific regressions.
5. Cut over behind a flag, keeping the old index warm long enough to roll back.
6. Delete the old index only after a full business cycle has passed without complaints.

For very large corpora where a full rebuild is genuinely impractical, there is emerging research worth knowing about: Drift-Adapter (Vejendla, 2025) trains a lightweight transformation that maps new-model query embeddings back into the legacy vector space, reporting recovery of 95–99% of retrieval recall while cutting recompute costs by over 100× compared with full re-indexing. It is a research result rather than standard practice, but for teams facing a billion-vector migration it changes the shape of the problem.

Problem 4: Choosing a freshness target

"Real-time" is the default answer and almost always the wrong one — it is the most expensive option and rarely the one users notice.

Set the refresh interval by how quickly a stale answer becomes harmful, not by how often the data changes. Some content changes constantly and matters little; some changes rarely and matters enormously.

Two notes on the extremes. Streaming is worth the complexity in far fewer cases than teams assume — check whether users can even perceive the difference between 30 seconds and 10 minutes before paying for it. And the bottom row is not a slower tier, it is a different mechanism: some things must be handled at query time because no indexing latency is acceptable at all.

A practical middle path most teams land on: micro-batch CDC for content, plus query-time filtering for permissions. You can find the orchestration patterns for the batch tiers in our Airflow fundamentals guide.

Testing an AI pipeline: the eval set is your data quality gate

A traditional pipeline asserts things like "row count is within 10% of yesterday" or "no nulls in customer_id". Those checks still apply to the metadata. But they cannot tell you whether retrieval still works.

For that you need an evaluation set — a fixed collection of representative queries paired with the documents that should be retrieved. Run it after every index change and gate the deploy on the result:

def evaluate_retrieval(index, eval_set, k=10) -> dict:
    hits = 0
    reciprocal_ranks = []
    for query, expected_doc_ids in eval_set:
        results = index.query(vector=embed(query), top_k=k)
        retrieved = [r.metadata["doc_id"] for r in results]

        if set(retrieved) & set(expected_doc_ids):
            hits += 1
            rank = next(i for i, d in enumerate(retrieved, 1) if d in expected_doc_ids)
            reciprocal_ranks.append(1 / rank)
        else:
            reciprocal_ranks.append(0.0)

    return {
        "recall_at_k": hits / len(eval_set),
        "mrr": sum(reciprocal_ranks) / len(eval_set),
    }

Add three cheap structural checks alongside it, because each catches a bug the recall score can miss:

• Orphan check — every distinct doc_id in the index still exists in the source. Catches the chunk-boundary trap directly.
• Staleness check — the oldest indexed_at for documents modified recently is within your freshness SLA.
• Model consistency check — every vector in the index carries the same embedding_model value. A single mismatched value means a partial migration is live.

The broader case for why this matters more than model choice is in AI and data quality, and the general framework these checks extend is in data quality testing for pipelines.

Common mistakes to avoid

1. Upserting chunks instead of replacing documents. The chunk-boundary trap. If you fix one thing from this article, fix this.
2. No doc_id in chunk metadata. Without it, correct deletion is impossible. Retrofitting means a full reindex.
3. Treating deletion as a batch problem. Access revocation needs a query-time filter, not a nightly job.
4. The embedding model in an environment variable. It is a schema. Version it, contract it, and make changing it a deliberate migration.
5. No evaluation set. You cannot detect a quality regression you have no baseline for — and you cannot build an honest baseline after seeing the new results.
6. Re-embedding unchanged content. Hash chunks and skip. It is a one-line comparison that usually removes most of the work.
7. Ignoring index maintenance. Tombstones accumulate. Schedule compaction like any other table maintenance.
8. Defaulting to streaming. Pick the freshness tier by consequence, not ambition.
9. Retrieving more context "just in case." Larger retrievals frequently make answers worse, not better — the reasoning behind that is covered in context engineering.
10. Skipping reconciliation. The periodic full diff is what catches the bugs in your incremental path.

Frequently Asked Questions

What is an AI data pipeline?

An AI data pipeline is the system that turns source content into the retrievable form a model consumes — parsing documents, splitting them into chunks, generating vector embeddings, and writing those into an index. Unlike a one-off ingestion script, it also keeps that index in sync as the source changes: propagating edits, removing deleted content, and re-embedding on model upgrades.

Why does my RAG system return outdated or deleted content?

Almost always because the pipeline updates chunks instead of replacing documents. When a document is edited its chunk boundaries shift — 12 chunks may become 9. An upsert keyed by chunk ID overwrites the first 9 and silently leaves 3 behind, still holding deleted text. Those orphans score high on similarity because they genuinely are about the topic. Delete by document ID first, then insert.

Do I have to re-embed everything when I change embedding models?

Yes, for one consistent index. A different model produces vectors in a different geometric space, so old and new vectors are not comparable and mixing them yields meaningless similarity scores. Use a blue/green migration: build the new index alongside the old, validate against a fixed eval set, then cut over. Re-embedding a million 500-token documents with text-embedding-3-small costs roughly $10 — the engineering risk, not the API bill, is the real cost.

How often should an AI data pipeline refresh its index?

Match the interval to how fast the answer goes wrong, not to how often the data changes. Manuals and policy documents tolerate nightly rebuilds. Tickets and pricing usually need CDC at minute-level latency. Access revocations should propagate immediately via a query-time metadata filter, because a delete that is 30 minutes late is a compliance incident rather than a stale answer.

Is deleting vectors from a vector database actually hard?

Harder than inserting. Graph indexes such as HNSW usually mark deleted vectors with tombstones rather than removing them, because unlinking a node would damage the graph's connectivity. The vector stops being returned but still consumes memory and disk until a compaction or rebuild reclaims it. On high-churn corpora this accumulates, so schedule compaction as routine maintenance.

Is AI replacing data engineering?

The opposite. AI added a consumer with stricter requirements than a dashboard: it reads unstructured content, needs it chunked and embedded, and produces confident wrong answers when data is stale rather than an obviously empty chart. Data engineering hiring grew 23% year over year and the services market is estimated at $105 billion in 2026. The work shifted toward unstructured data, retrieval freshness, and governance — it did not shrink.

Should I use a dedicated vector database or add vectors to my existing database?

If your corpus is under a few million vectors and you already run PostgreSQL, pgvector avoids an entire system in your architecture — and keeping vectors next to the metadata you filter on is a genuine operational advantage. Dedicated vector databases earn their place at higher scale, with heavy filtering, or when you need distributed sharding. The trade-offs are broken down in vector databases explained.

Conclusion

The reason AI has made data engineering boom is not that models need more data. It is that models made incorrect data expensive in a way dashboards never did. A broken chart is obviously broken. A retrieval index quietly serving deleted paragraphs produces an answer that reads perfectly and is wrong, at scale, to people who trust it.

Every hard problem in this article turned out to be a data engineering problem wearing new vocabulary. Chunk orphans are a failure of idempotent replacement. Missed deletions are a change data capture gap. Embedding upgrades are a schema migration. Freshness tiers are the batch-versus-streaming decision that has existed for a decade. The vector store is new; the discipline is not.

So build it the way you would build any derived dataset. Put doc_id in your metadata before the first load. Replace documents, never chunks. Hash content and skip the work you do not need to redo. Version the embedding model like the schema it is. Keep an evaluation set and gate on it. Reconcile against the source on a schedule, because your incremental path has bugs you have not found yet.

Start with the single check that catches the most damage for the least effort: take every distinct doc_id in your index, and confirm each one still exists in the source. If the count comes back clean, you are in better shape than most production systems. If it does not, you have just found what your assistant has been quoting. And if you would rather have engineers who have run these pipelines in production build it with you, talk to us — it is free to start.