Building a RAG Pipeline
Walk through ingestion, indexing, retrieval, prompting, and evaluation as one repeatable pipeline.
Define the data contract
Before writing ingestion code, define what a source document, version, chunk, embedding, and retrieval result mean. Specify required IDs, timestamps, permissions, content types, and status fields. Decide how updates, deletions, parsing failures, and duplicate files are represented. This contract gives every stage a shared vocabulary and makes reruns safe. It also supports provenance in citations. Without it, pipelines often become a collection of scripts that produce plausible records but cannot answer which source version created a particular vector or why a deleted file remains searchable.
Ingest and normalize
Collect documents from approved locations and assign stable identifiers before parsing. Normalize encoding, whitespace, headers, footers, and repeated navigation while preserving headings and meaningful layout. Extract text with format-specific logic and send malformed or empty files to a visible failure queue. Attach source metadata, access scope, publication dates, and an ingestion version. Keep the original document reference for audit and reprocessing. A successful ingestion job should report counts, skipped items, errors, and duration rather than only returning a generic “complete” message.
def index_document(document, embed, vector_store):
chunks = split_by_heading(document["text"], max_chars=900)
records = []
for position, chunk in enumerate(chunks):
records.append({
"id": f"{document['id']}:{document['version']}:{position}",
"document_id": document["id"],
"version": document["version"],
"position": position,
"text": chunk,
"embedding": embed(chunk),
})
vector_store.upsert(records) # retries update the same IDs
vector_store.activate_version(document["id"], document["version"])
for document in source_documents:
index_document(document, embed, vector_store)Chunk, embed, and index
Split normalized content according to structure and store each chunk with its parent document, position, metadata, and text hash. Generate embeddings using a versioned model and write them with an idempotent key so retries update the same record. Validate dimensions, empty content, and unexpected provider responses before indexing. Use batches with bounded concurrency and checkpoints for large corpora. Mark a document version active only after all expected chunks are indexed successfully; otherwise a partial rebuild can make the system answer from an incomplete source.
Retrieve candidates
At query time, validate the user request and determine the caller’s authorization scope before searching. Embed the question, apply trusted metadata filters, and retrieve enough candidates for the next ranking stage. Combine keyword search or query rewriting when exact terms, abbreviations, or conversational references matter. Preserve scores and source identifiers in a trace. If no candidate clears a tested relevance bar, return an answerability signal instead of forcing the generator to use weak context. Retrieval should be a clear, inspectable decision, not an invisible helper call.
Build a bounded prompt
The prompt builder should place the user question, system instructions, and retrieved evidence in clearly separated fields. Include source labels and concise citation identifiers, but remove unnecessary text that consumes context. Tell the model how to handle missing evidence, conflicting versions, and unsafe requests. Bound the number and size of chunks, and enforce a total token budget before calling the provider. Treat retrieved content as untrusted data: instructions found inside a document should not override application policy, permissions, or tool restrictions.
Generate and validate
Call the language model with a deadline, a controlled temperature, and a response format suited to the product. After generation, validate length, required citations, prohibited disclosures, and any structured fields. A separate grounding check can compare important claims with retrieved evidence, but it should be treated as another imperfect model or rule system. If validation fails, abstain, ask for clarification, or route to review according to the use case. Never silently convert a failed grounded answer into an unsupported confident answer.
Observe the whole trace
Log a correlation ID through ingestion, retrieval, generation, and response delivery. Record safe metadata such as model and index versions, candidate IDs, ranks, latency, token counts, error categories, and user feedback. Avoid storing raw sensitive questions or documents unless there is a documented need, access control, retention period, and redaction process. Dashboards should separate retrieval failures, provider failures, validation failures, and user dissatisfaction. A single “RAG latency” number hides the stage that needs attention and makes capacity planning unreliable.
Operate it as a pipeline
Make every stage retryable where safe, idempotent, observable, and independently testable. Use queues or scheduled jobs for ingestion, rate limits for provider calls, dead-letter handling for malformed documents, and a deployment process that can roll back prompt or index versions. Keep a golden evaluation set and run it before changing chunking, embeddings, retrieval parameters, or prompts. A RAG feature is production-ready when its evidence can be traced, its failures are understandable, its data boundaries are enforced, and its operators have a recovery path.