Data pipelines for AI applications
An AI data pipeline pulls content from source systems, detects what changed, parses and cleans it, masks sensitive data, chunks and embeds it, updates the indexes (including deletes and permission changes), and reports how fresh everything is. Run it incrementally, on a schedule or on change events, instead of rebuilding everything each time. Deletes and permission changes are the parts teams forget, and both cause real incidents.
Why "index once" fails
The policy changed last Tuesday, and the assistant still quotes the old one, because the index was built once at launch. Or an employee left, their access was removed from the drive, and the assistant still answers from their private folder. A pipeline that runs once is a snapshot, and snapshots go stale.
Stages
- Connect: read from the sources (drives, wikis, ticketing, databases, APIs) with incremental sync, only what changed since the last run.
- Detect change: compare content hashes, so unchanged documents are skipped.
- Parse: extract text, tables and structure. See document parsing.
- Clean: normalise whitespace and encoding, remove boilerplate, mask personal data.
- Enrich: add metadata (source, version, date, owner, permissions, document type).
- Chunk and embed: by structure where possible, with the chunk's metadata attached.
- Upsert: write to the indexes, keyed by document ID and version, so re-running is safe.
- Delete: remove chunks for documents deleted at the source, from every index and cache.
- Report: freshness, failures, counts.
Incremental sync with content hashes
import hashlib
index: dict[str, dict] = {}
def content_hash(text: str) -> str:
return hashlib.sha256(" ".join(text.split()).encode()).hexdigest()
def sync(source_docs: dict[str, str]) -> dict[str, int]:
stats = {"added": 0, "updated": 0, "unchanged": 0, "deleted": 0}
for doc_id, text in source_docs.items():
h = content_hash(text)
if doc_id not in index:
stats["added"] += 1
elif index[doc_id]["hash"] == h:
stats["unchanged"] += 1
continue
else:
stats["updated"] += 1
index[doc_id] = {"hash": h, "chunks": [text[i:i + 200] for i in range(0, len(text), 200)]}
for doc_id in set(index) - set(source_docs):
del index[doc_id]
stats["deleted"] += 1
return stats
print(sync({"refund-policy": "Refunds within 30 days.", "shipping": "Ships in 2 days."}))
print(sync({"refund-policy": "Refunds within 14 days in the EU.", "shipping": "Ships in 2 days."}))
print(sync({"refund-policy": "Refunds within 14 days in the EU."}))
The second run updates only the changed policy, and the third removes the deleted document, which is the step most pipelines forget.
Orchestration
For small setups, a scheduled job (cron, a cloud scheduler) is enough. As the number of sources and steps grows, use an orchestrator that handles dependencies, retries and history: Apache Airflow, Dagster or Prefect.
Freshness
Freshness is the time from a change at the source to the change being reflected in answers. Set a target per source (policies within an hour, archives weekly), measure it, and alert when it's missed.
A simple test: change one document in the source and measure how long until the assistant's answer reflects it.
Measuring it
- Freshness lag at p95, per source.
- Failed sync runs per week.
- Documents deleted at the source but still in the index (should be zero).
From the floor
For the financial report RAG, the production phases after building locally included orchestration, with Apache Airflow installed for it, because once the pipeline leaves your laptop, something has to run ingestion, save artifacts and model weights, and re-run it reliably.
In the legal LLM lab, the ingest step pulled data from public sources into a raw folder, and the normalise step removed extra whitespace and split text by article, section and clause before anything was chunked, which is the parse and clean stage of this pipeline.