An index enrichment pipeline is an asynchronous processing layer that takes raw ingested documents and augments them with derived data—embeddings, summaries, text chunks, named entities, and classification labels—before or after they land in the search index. The key insight is that enrichment is decoupled from ingestion: a document becomes searchable immediately upon insertion, while enrichment tasks run in the background and update the document’s index entry once complete.
This architecture matters because enrichment operations are expensive. Computing embeddings requires GPU inference, generating summaries calls an LLM, and chunking requires parsing document structure. Running these synchronously during ingestion would add seconds of latency per document and create a bottleneck that limits throughput. By moving enrichment to an async pipeline, the system can ingest documents at full speed and process enrichments at whatever rate the available compute allows.
In a typical production search engine, the enrichment pipeline is structured as a series of stages connected by work queues. Each stage reads documents from its input queue, applies a transformation, writes the result back to the document store, and pushes the document ID to the next stage’s queue. The pipeline supports backfill (re-processing existing documents when a new enrichment is added), pause/resume (stopping a stage without losing progress), and priority ordering (enriching recently-queried documents first).
Documents are indexed immediately in their raw form, making them searchable within milliseconds of ingestion. Enrichment runs asynchronously and updates the document’s index entry in place. This means search quality improves over time as enrichments complete: a document initially found only by keyword becomes discoverable by vector similarity once its embedding is computed.
Each enrichment type (embedding, chunking, summarization, entity extraction) is a separate stage. Stages are independent and can run in parallel on different workers. A document flows through stages in dependency order: chunking must happen before per-chunk embeddings, but entity extraction can run in parallel with both. The pipeline DAG encodes these dependencies.
When a new enrichment stage is added (e.g., switching from one embedding model to another), all existing documents need re-processing. A backfill scan walks the document store, identifies documents missing the new enrichment, and enqueues them. Backfill runs at lower priority than new-document enrichment to avoid starving real-time ingestion.
Each enrichment stage must be idempotent: processing the same document twice produces the same result and doesn’t corrupt state. This is essential because queue delivery is at-least-once—worker crashes, timeouts, and retries mean a document may be processed more than once. Idempotency ensures correctness without exactly-once delivery.
Individual pipeline stages can be paused (stop dequeuing work) and resumed (continue from where they left off). This is critical for operational control: if an embedding model starts producing garbage, operators can pause the embedding stage, fix the model, and resume without losing queue position or re-processing already-enriched documents.
Each document carries an enrichment status map: a set of key-value pairs where the key is the stage name and the value is a status record (pending, in-progress, completed, failed). This map is stored alongside the document and is used to determine which stages still need to process the document. Failed enrichments include an error message and retry count, enabling dead-letter handling after a configurable number of retries.
Not all documents need enrichment equally urgently. Documents that users are actively querying benefit most from having embeddings and summaries. The pipeline supports priority levels: documents referenced in recent queries are promoted to the front of the enrichment queue, while bulk-ingested historical documents process at background priority.
The enrichment pipeline consists of a coordinator, work queues, and stage workers. The coordinator manages pipeline state (which stages exist, their dependencies, pause/resume status). Work queues connect stages and provide at-least-once delivery with visibility timeouts. Stage workers pull documents from their input queue, apply the enrichment, write results back, and acknowledge the message:
// Go: Enrichment pipeline coordinator
package enrichment
import (
"context"
"sync"
"time"
)
// Stage defines an enrichment processing stage.
type Stage struct {
Name string
DependsOn []string // stages that must complete first
Process func(ctx context.Context, doc *Document) (*EnrichmentResult, error)
Concurrency int // number of parallel workers
RetryLimit int
Paused bool
}
// Pipeline manages the enrichment workflow.
type Pipeline struct {
mu sync.RWMutex
stages map[string]*Stage
queue WorkQueue
store DocumentStore
order []string // topological order of stages
}
// Enqueue adds a document to the pipeline for enrichment.
func (p *Pipeline) Enqueue(ctx context.Context, docID string, priority int) error {
// Find the first stage(s) with no dependencies
roots := p.rootStages()
for _, stage := range roots {
err := p.queue.Push(ctx, stage, QueueItem{
DocID: docID,
Priority: priority,
EnqueuedAt: time.Now(),
})
if err != nil {
return err
}
}
return nil
}
// rootStages returns stages with no dependencies.
func (p *Pipeline) rootStages() []string {
var roots []string
for _, name := range p.order {
if len(p.stages[name].DependsOn) == 0 {
roots = append(roots, name)
}
}
return roots
}
// advance moves a document to the next stage(s) after completion.
func (p *Pipeline) advance(ctx context.Context, docID string, completedStage string, priority int) error {
for _, name := range p.order {
stage := p.stages[name]
if !containsStr(stage.DependsOn, completedStage) {
continue
}
// Check if all dependencies are satisfied
doc, _ := p.store.Get(ctx, docID)
allDone := true
for _, dep := range stage.DependsOn {
if doc.EnrichmentStatus[dep] != StatusCompleted {
allDone = false
break
}
}
if allDone {
p.queue.Push(ctx, name, QueueItem{
DocID: docID,
Priority: priority,
})
}
}
return nil
}Each stage runs a configurable number of worker goroutines. Workers dequeue items, check that the stage is not paused, load the document, call the stage’s processing function, persist results, and advance the document to downstream stages:
// Go: Stage worker implementation
package enrichment
import (
"context"
"log"
"time"
)
// RunWorker processes items for a single stage.
func (p *Pipeline) RunWorker(ctx context.Context, stageName string) {
stage := p.stages[stageName]
for {
select {
case <-ctx.Done():
return
default:
}
// Check if stage is paused
p.mu.RLock()
paused := stage.Paused
p.mu.RUnlock()
if paused {
time.Sleep(time.Second)
continue
}
// Dequeue next item (blocks until available or timeout)
item, err := p.queue.Pop(ctx, stageName, 30*time.Second)
if err != nil {
continue // timeout or context cancelled
}
// Load document
doc, err := p.store.Get(ctx, item.DocID)
if err != nil {
log.Printf("stage %s: failed to load doc %s: %v", stageName, item.DocID, err)
p.queue.Nack(ctx, stageName, item) // return to queue
continue
}
// Skip if already enriched (idempotency check)
if doc.EnrichmentStatus[stageName] == StatusCompleted {
p.queue.Ack(ctx, stageName, item)
continue
}
// Process
doc.EnrichmentStatus[stageName] = StatusInProgress
p.store.UpdateStatus(ctx, item.DocID, stageName, StatusInProgress)
result, err := stage.Process(ctx, doc)
if err != nil {
retries := doc.EnrichmentRetries[stageName] + 1
if retries >= stage.RetryLimit {
doc.EnrichmentStatus[stageName] = StatusFailed
p.store.UpdateStatus(ctx, item.DocID, stageName, StatusFailed)
log.Printf("stage %s: doc %s permanently failed after %d retries", stageName, item.DocID, retries)
} else {
doc.EnrichmentRetries[stageName] = retries
p.queue.Nack(ctx, stageName, item) // re-enqueue for retry
}
p.queue.Ack(ctx, stageName, item)
continue
}
// Persist enrichment result
p.store.SaveEnrichment(ctx, item.DocID, stageName, result)
doc.EnrichmentStatus[stageName] = StatusCompleted
p.store.UpdateStatus(ctx, item.DocID, stageName, StatusCompleted)
p.queue.Ack(ctx, stageName, item)
// Advance to downstream stages
p.advance(ctx, item.DocID, stageName, item.Priority)
}
}When a new stage is added or an existing stage’s model is updated, the backfill scanner identifies documents that need (re-)enrichment. It iterates the document store in batches, checks each document’s enrichment status map, and enqueues those missing the target stage at low priority:
// Go: Backfill scanner for re-enrichment
package enrichment
import "context"
// Backfill enqueues all documents that are missing a specific enrichment.
func (p *Pipeline) Backfill(ctx context.Context, stageName string, batchSize int) (int, error) {
var cursor string
enqueued := 0
for {
docs, nextCursor, err := p.store.ListBatch(ctx, cursor, batchSize)
if err != nil {
return enqueued, err
}
for _, doc := range docs {
if doc.EnrichmentStatus[stageName] == StatusCompleted {
continue // already enriched
}
// Check dependencies are met
stage := p.stages[stageName]
depsOK := true
for _, dep := range stage.DependsOn {
if doc.EnrichmentStatus[dep] != StatusCompleted {
depsOK = false
break
}
}
if !depsOK {
continue // dependencies not yet satisfied
}
p.queue.Push(ctx, stageName, QueueItem{
DocID: doc.ID,
Priority: PriorityBackfill, // lowest priority
})
enqueued++
}
if nextCursor == "" {
break
}
cursor = nextCursor
}
return enqueued, nil
}A production search engine using this pattern generates vector embeddings for every document via the enrichment pipeline. The embedding stage sends document text to a local inference server, receives the vector, and writes it to the document’s vector field. Because embedding is async, bulk imports of millions of documents complete quickly while embeddings are computed over the following hours. Documents are searchable by keyword immediately; vector search becomes available per-document as embeddings finish.
For retrieval-augmented generation, documents must be split into semantically coherent chunks. The chunking stage parses document structure (headings, paragraphs, code blocks), applies overlap-aware splitting with configurable chunk size, and stores chunks as child documents linked to the parent. Downstream stages then compute per-chunk embeddings, enabling fine-grained retrieval that returns the most relevant passage rather than the entire document.
An NER enrichment stage extracts person names, organizations, dates, and domain-specific entities from document text. These entities are stored as structured metadata on the document, enabling faceted search (“all documents mentioning Company X”) and graph edge creation (linking documents to entity nodes in the knowledge graph).
Search result snippets are more useful when they are intelligently summarized rather than just showing the first N characters. The summarization stage calls an LLM to produce a 2–3 sentence summary of each document, stored as metadata. The search UI displays these summaries in results, giving users a better signal for which results are relevant without requiring them to click through to each document.
Work queue items become invisible for a configured duration after being dequeued. If the worker doesn’t acknowledge completion within the timeout, the item reappears in the queue for another worker to pick up. This ensures no document is lost if a worker crashes mid-processing. The timeout must be longer than the slowest expected processing time (embedding a large document can take 30+ seconds).
Some documents consistently fail enrichment—malformed content, unsupported encodings, or content that triggers model errors. After exceeding the retry limit, these documents move to a dead letter queue (DLQ) for manual inspection. The DLQ preserves the error message from each attempt, enabling operators to diagnose and fix the issue, then re-enqueue the document:
// Zig: Dead letter queue entry
const DLQEntry = struct {
doc_id: []const u8,
stage: []const u8,
attempts: u32,
last_error: []const u8,
first_failed_at: i64,
last_failed_at: i64,
pub fn format(self: *const DLQEntry, writer: anytype) !void {
try writer.print(
"DLQ: doc={s} stage={s} attempts={d} error={s}\n",
.{ self.doc_id, self.stage, self.attempts, self.last_error },
);
}
};
const DeadLetterQueue = struct {
entries: std.ArrayList(DLQEntry),
allocator: std.mem.Allocator,
pub fn add(self: *DeadLetterQueue, entry: DLQEntry) !void {
try self.entries.append(entry);
}
/// Requeue all entries for a given stage back into the pipeline.
pub fn requeue(self: *DeadLetterQueue, stage: []const u8, queue: *WorkQueue) !u32 {
var requeued: u32 = 0;
var i: usize = 0;
while (i < self.entries.items.len) {
if (std.mem.eql(u8, self.entries.items[i].stage, stage)) {
try queue.push(self.entries.items[i].doc_id, PriorityBackfill);
_ = self.entries.orderedRemove(i);
requeued += 1;
} else {
i += 1;
}
}
return requeued;
}
};For backfill operations, a high-watermark tracks how far through the document store the scanner has progressed. If the backfill is paused and resumed, it continues from the watermark rather than restarting. The watermark is a cursor (typically the last document ID processed) stored persistently so it survives process restarts.
Enrichment stages that call external services (LLM APIs for summarization, embedding endpoints) must respect rate limits. A token bucket rate limiter per stage ensures the pipeline doesn’t overwhelm downstream services. When the bucket is empty, workers block until tokens replenish, naturally throttling throughput without losing work.
If stage B depends on stage A (e.g., per-chunk embeddings depend on chunking), enqueuing a document for stage B before stage A completes produces incomplete or incorrect results. Always check the dependency DAG before advancing a document. The pipeline coordinator should enforce this: only enqueue downstream stages when all upstream dependencies show StatusCompleted.
When a document’s content changes, its existing enrichments (embeddings, summaries) become stale. The pipeline must detect content updates and invalidate affected enrichments, resetting their status to pending and re-enqueuing them. Failing to do this means vector search returns results based on old content, creating a consistency gap between keyword search (which uses current content) and vector search (which uses stale embeddings).
Bulk-ingesting millions of documents floods the enrichment queues. If the queue implementation doesn’t handle backpressure, memory usage grows unbounded. Use persistent queues (backed by disk or a message broker) rather than in-memory queues, and apply admission control to limit how many documents are in-flight at once.
If an enrichment stage appends to a list rather than replacing it, processing a document twice creates duplicate entries. Every stage must be designed so that running it N times on the same input produces the same final state as running it once. Use upsert semantics (write-or-replace) rather than append semantics.
Running backfill at the same priority as real-time enrichment causes newly ingested documents to wait behind millions of historical documents. Always give backfill work a lower priority so that fresh documents are enriched first. Users care about recently ingested documents being searchable by vector; historical completeness is a background concern.