Streaming RAG: which architecture when documents and questions arrive at the same time

A textbook RAG works like this: you gather a corpus, split it, embed it, build an index, and query. The index lives for weeks, sometimes months. Freshness never comes up as a question, and latency gets settled by tuning one approximate-search parameter.

The systems I have been meeting for the past two years look nothing like that. An event stream feeds the corpus continuously — incident tickets, application logs, news wires, sensor readings, monitoring messages — and the questions arrive at the same pace. Between the moment a document enters the system and the moment an answer has to rest on it, the margin is measured in seconds.

That shift changes the nature of the problem. A classic RAG optimises recall over a frozen corpus; a streaming RAG arbitrates continuously between index freshness, search quality and response time. The three never rise together.

What streaming really changes

Three properties of the classic architecture collapse at once.

The index stops being immutable. An HNSW graph builds well once and reads very fast; it copes badly with heavy insertion and worse still with deletion, which leaves tombstones behind until the next compaction. Writing continuously into a structure designed for reading means accepting that search latency drifts as the day goes on.

Splitting loses its global context. Splitting a complete corpus lets you compute statistics, de-duplicate, and size each chunk to its document. In a stream, every document arrives alone: splitting becomes a local decision, taken without knowing what comes next.

Embedding turns into a critical service. As long as it runs overnight, a slow embedding model bothers nobody. The moment it sits on the ingestion path, it dictates how far the index lags — and if you also embed the question, it sits on the answer path as well.

Breaking down the latency budget

Before opening any vector-database catalogue, write the budget down. Here is the one from an operations-support platform we delivered last year: on-call engineers query a corpus fed by alerts, tickets and change logs, in plain language.

Two lessons jump out. Reranking alone carries two thirds of the initial budget: it is almost always the first item to deal with. And question embedding cost a hundred and ten milliseconds because it went through the same service as ingestion, saturated by batches — separating the two paths was enough to claw back close to a hundred milliseconds.

Hold on to the rule: the ingestion path and the answer path never share the same embedding service. One chases throughput, the other chases latency. The two goals fight each other, and the batch always wins.

Three architectures, three trade-offs

Faced with a stream, three families of answers hold up. They differ less in technology than in where they place the boundary between what is already indexed and what is not yet.

A word on the middle column: it points back to the lambda architecture, which the data world has known for fifteen years. RAG is rediscovering it, and the same criticisms apply — two processing paths, twice the code, twice the opportunities to diverge.

Architecture 1 — micro-batch indexing

The principle fits in one sentence: accumulate documents over a fixed interval, then embed and write the batch in a single move.

This approach has a poor reputation, undeservedly. It offers the best ratio between operational simplicity and search quality, and a thirty-second interval satisfies the vast majority of business uses. A documentation assistant, a support knowledge base, an internal search engine: nobody notices that a document published twenty seconds ago has yet to appear.

The point that deserves care is batch size. Too small a batch wastes the embedding GPU; too large a batch stretches the freshness gap and jolts the index. The answer is to trigger the write on two competing conditions, whichever comes first.

"""Ingestion consumer: adaptive accumulation, then batch write."""
import asyncio
import time
from dataclasses import dataclass, field

import httpx
from aiokafka import AIOKafkaConsumer
from qdrant_client import AsyncQdrantClient, models

MAX_BATCH_SIZE = 64          # ceiling imposed by the embedding GPU memory
MAX_DELAY_SECONDS = 5.0     # ceiling imposed by the agreed freshness gap
EMBEDDING_URL = "http://tei-ingestion.rag.svc.cluster.local/embed"


@dataclass
class Lot:
    """Accumulator bounded in size and in time."""
    chunks: list[dict] = field(default_factory=list)
    opened_at: float = field(default_factory=time.monotonic)

    def ready(self) -> bool:
        return (len(self.chunks) >= MAX_BATCH_SIZE
                or time.monotonic() - self.opened_at >= MAX_DELAY_SECONDS)

    def drain(self) -> list[dict]:
        chunks, self.chunks = self.chunks, []
        self.opened_at = time.monotonic()
        return chunks


async def embed(client: httpx.AsyncClient, textes: list[str]) -> list[list[float]]:
    """One round trip for the whole batch: the GPU runs at full load."""
    response = await client.post(EMBEDDING_URL, json={"inputs": textes},
                                timeout=httpx.Timeout(20.0))
    response.raise_for_status()
    return response.json()


async def ingest(consumer: AIOKafkaConsumer, qdrant: AsyncQdrantClient,
                  http: httpx.AsyncClient) -> None:
    batch = Lot()
    async for message in consumer:
        batch.chunks.extend(split(message.value))
        if not batch.ready():
            continue

        chunks = batch.drain()
        vectors = await embed(http, [f["text"] for f in chunks])
        await qdrant.upsert(
            collection_name="corpus",
            wait=False,                       # the WAL acknowledgement is enough, the index follows
            points=[
                models.PointStruct(
                    id=f["id"], vector=v,
                    payload={"tenant": f["tenant"], "source": f["source"],
                             "timestamp": f["timestamp"], "text": f["text"]},
                )
                for f, v in zip(chunks, vectors)
            ],
        )
        await consumer.commit()           # commit after the write, never before

Three details deserve attention. The Kafka commit happens after the write to the index, which guarantees that a restart replays the lost batch rather than forgetting it. The wait=False parameter returns as soon as the log is written, without waiting for the graph to build. And the timer restarts on every drain, not on every message: without that, an irregular stream produces batches of one.

Architecture 2 — the hot and cold dual view

When the freshness gap has to drop below a second, micro-batching no longer suffices. The way round it is to keep two indexes of different natures and merge their results at query time.

The hot index holds only the last few minutes of documents — a few tens of thousands of vectors at most. At that scale, exhaustive dot-product search over a contiguous array beats any approximate structure: no graph to build, no compaction, constant-time insertion and perfect recall. A million 768-dimension vectors fit in three gigabytes and scan in about ten milliseconds with vector instructions.

The cold index receives the same documents several minutes later, in micro-batches, inside a classic and properly compacted HNSW. It carries the volume, the retention and the search quality at scale.

A promotion process removes from the hot index every document whose arrival the cold index has confirmed. The overlap window should stay generous: a duplicate removed at merge time beats a document that briefly cannot be found.

"""Merging both views at query time."""
import asyncio

import numpy as np

K_CHAUD, K_FROID, K_FINAL = 30, 50, 12


async def search(question_vector: np.ndarray, tenant: str) -> list[dict]:
    """Queries both indexes in parallel, merges, de-duplicates."""
    hot, cold = await asyncio.gather(
        hot_index.search(question_vector, k=K_CHAUD, tenant=tenant),
        cold_index.search(question_vector, k=K_FROID, tenant=tenant),
    )

    # The most recent document wins over its cold counterpart.
    by_identifier: dict[str, dict] = {}
    for result in sorted(hot + cold, key=lambda r: r["timestamp"], reverse=True):
        by_identifier.setdefault(result["id"], result)

    candidates = sorted(by_identifier.values(), key=lambda r: r["score"], reverse=True)
    return candidates[:K_FINAL]

The cost of this architecture lies in operations rather than in compute. Two indexes mean two schemas to evolve together, two sets of metrics, two failure modes, and a promotion process that becomes the most delicate component on the platform. Choose it only if the business genuinely demands sub-second freshness — and make them write that requirement down.

Architecture 3 — the continuously written index

Several engines now build continuous writing into their very design, which makes the dual view unnecessary in most cases. The principle is everywhere the same: a write-ahead log takes the insertions, a mutable segment serves them immediately, and a background task builds then seals immutable segments.

I pick Qdrant or Milvus for most of the platforms I design. The first for its operational simplicity on Kubernetes and its filtering model, the second once volume passes a hundred million vectors and separating compute from storage becomes a real subject.

The decisive setting in Qdrant is the indexing threshold. Below it, a segment is scanned in full: fast while it stays small, and it avoids rebuilding a graph for three hundred vectors. Above it, the optimiser builds the HNSW in the background.

# Collection tuned for a sustained stream, deployed by Argo CD.
apiVersion: qdrant.io/v1
kind: QdrantCollection
metadata:
  name: corpus
  namespace: rag
spec:
  vectors:
    size: 768
    distance: Cosine
    on_disk: false                  # vectors stay in memory, and so does the index
  hnsw_config:
    m: 24                           # 16 by default; 24 lifts recall by 2 to 3 points
    ef_construct: 200
    full_scan_threshold: 20000      # below this, exhaustive search: faster and exact
  optimizers_config:
    indexing_threshold: 30000       # graph construction only beyond this
    memmap_threshold: 200000
    default_segment_number: 8       # one segment per core allocated to the pod
    flush_interval_sec: 2
    max_optimization_threads: 2     # throttles merging to protect read latency
  quantization_config:
    scalar:
      type: int8
      quantile: 0.99
      always_ram: true              # cuts the footprint fourfold, costs 1 point of recall
  write_consistency_factor: 1
  replication_factor: 2
  shard_number: 6
  payload_schema:                   # index the filtered fields, otherwise filtering scans
    tenant: { type: keyword, is_tenant: true }
    timestamp: { type: integer, range: true }
    source: { type: keyword }

The max_optimization_threads parameter deserves an explanation. Left unchecked, the optimisers saturate the pod cores during a merge and search latency doubles. Capping them at two threads spreads the merge out further, but read p99 stays flat — a trade-off I revisit on every platform, never a universal value.

The ingestion pipeline, end to end

Here is the full chain, in the order data travels through it.

Intake. Kafka or Redpanda, partitioned by tenant identifier to guarantee ordering per client and allow parallelism between them. Plan a dead-letter queue from day one: a malformed document must never block a partition.

Splitting. In a stream, forget the global semantic splitting that compares every sentence with its neighbours across the whole corpus. Use a sliding window with overlap, bounded in tokens, honouring the structural boundaries of the document — headings, paragraphs, code blocks. Count tokens with the embedding model tokeniser, never in characters.

De-duplication. Operational streams repeat enormously: the same alert returns every thirty seconds. A locality-sensitive fingerprint kept in Redis with a time to live removes near-duplicates before embedding. On the on-call platform mentioned above, that single filter cut 38 % of the incoming volume and the embedding bill along with it.

Embedding. A dedicated service, separate from the query one. Server-side dynamic batching — the kind Triton or Text Embeddings Inference provides — works better than client-side batching as soon as several consumers coexist.

Writing. Into the index, without waiting for the graph to build, then the Kafka commit.

Publishing the timestamp. The most frequently forgotten point: publish, as a metric, the gap between the document original timestamp and the moment it was written. Without that measurement, nobody knows whether the freshness promise holds.

The answer path: where the milliseconds hide

The answer path follows one simple rule: whatever can happen before the question arrives must happen before.

Reranking. A cross-encoder scoring fifty candidates costs ten times more than the vector search. Three levers bring it back within budget: cut the number of candidates to rerank, take a small distilled model, and skip reranking when the score gap between the first and the tenth candidate exceeds a threshold — if search has already made the call, there is no point paying to confirm it.

Semantic caching. On a support stream, a third of the questions resemble a question already asked. A cache keyed on the question vector, with a similarity threshold at 0.97 and a short lifetime, saves the entire chain. Watch the trap, though: in a streaming system a cached answer ages along with the corpus. Invalidate on the timestamp of the most recent document that built it, not on a duration alone.

Hybrid search. Lexical and vector complement each other: the first finds identifiers, version numbers and exact error messages, the second captures intent. Merging them by reciprocal rank costs a few milliseconds and buys five to ten points of recall — the best ratio in the budget.

Progressive delivery. Streaming does not concern ingestion alone. Stream the generation model tokens as they come: users perceive the delay before the first token, not the total duration. A complete answer in three seconds whose first word arrives after two hundred milliseconds feels instant; the same answer delivered in one block after three seconds feels slow.

Sizing on Kubernetes

The platform breaks down into four families of workload with frankly divergent needs, which call for four separate node pools.

The index never enters an autoscaling loop. Adding a replica triggers a segment copy that weighs on latency at the precise moment load is climbing. Size for the peak, and resize cold.

For the rest, KEDA reads the Kafka lag and adjusts the consumers. The subtle point is the choice of metric: a count of pending messages reacts too late on an irregular stream. I prefer to drive on lag expressed in seconds, which translates directly into the freshness promise.

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: ingestion-corpus
  namespace: rag
spec:
  scaleTargetRef:
    name: ingestion-corpus
  minReplicaCount: 2               # never zero: a cold start costs 25 s
  maxReplicaCount: 24              # capped at the number of topic partitions
  pollingInterval: 10
  cooldownPeriod: 120
  advanced:
    horizontalPodAutoscalerConfig:
      behavior:
        scaleUp:
          stabilizationWindowSeconds: 0     # scale up fast
          policies: [{ type: Percent, value: 100, periodSeconds: 30 }]
        scaleDown:
          stabilizationWindowSeconds: 300   # scale down slowly
          policies: [{ type: Pods, value: 1, periodSeconds: 120 }]
  triggers:
    - type: kafka
      metadata:
        bootstrapServers: redpanda.stream.svc.cluster.local:9093
        consumerGroup: ingestion-corpus
        topic: incoming-documents
        lagThreshold: "2000"
        activationLagThreshold: "50"
        offsetResetPolicy: earliest
      authenticationRef:
        name: redpanda-mtls
    - type: prometheus                       # second trigger: freshness itself
      metadata:
        serverAddress: http://prometheus.observability.svc.cluster.local:9090
        threshold: "8"
        query: |
          histogram_quantile(0.95,
            sum by (le) (rate(rag_freshness_gap_seconds_bucket[2m])))

The second trigger closes the loop: even when Kafka lag stays low, a slowdown in embedding degrades freshness, and the platform reacts to the consequence rather than to the cause.

Multi-tenancy and the right to be forgotten

Two subjects demos ignore and production imposes.

Filtering must precede approximate search, not follow it. Filtering afterwards looks equivalent and is not: if the fifty nearest neighbours all belong to another tenant, the final answer is empty although relevant documents existed. Worse, latency explodes as the engine widens the search to compensate. Qdrant handles this with a payload index marked is_tenant, which physically groups one tenant vectors into the same segments; Milvus offers partition keys. Either way, declare the partitioning when you create the collection — retrofitting it means rewriting everything.

Deleting a document is not enough. Removing a vector from an index makes it unfindable, but it survives in the write-ahead log, in uncompacted segments, in backup snapshots and — the most common blind spot — in the semantic cache of answers already produced. A complete erasure procedure chains four moves: deletion in the index, forced compaction of the segment concerned, purge of cache entries whose provenance cites the document, and verification through a control query. Automate it, document it, and test it regularly: the day the request arrives, the legal clock is already running.

In the same spirit, the ingestion stream carries data nobody has proof-read. A personal-data detector placed before embedding, which masks or discards, costs a few milliseconds and stops a social security number found in an application log from ending up in a generated answer.

Measuring: the six metrics that matter

A streaming RAG is watched like any distributed platform: metrics, thresholds, alerts. Six are enough to know whether the system holds.

The fourth deserves a word. Recall cannot be measured in production without a reference: put a probe set in place — fifty questions whose expected documents you know — replayed every ten minutes by a periodic job. It is the only guard against silent degradation, the kind where the system answers fast and badly without a single technical alert firing.

groups:
  - name: rag-stream
    rules:
      - record: rag:freshness:p95
        expr: |
          histogram_quantile(0.95,
            sum by (le, collection) (rate(rag_freshness_gap_seconds_bucket[5m])))

      - alert: IndexFreshnessDegraded
        expr: rag:freshness:p95 > 10
        for: 5m
        labels: { severity: major, team: ai-platform }
        annotations:
          summary: "Freshness gap of {{ $value | humanizeDuration }} on {{ $labels.collection }}"
          action: "Check the Kafka lag, then the embedding service queue."

      - alert: RecallDropping
        expr: rag_probe_recall < 0.90
        for: 15m
        labels: { severity: major, team: ai-platform }
        annotations:
          summary: "Recall measured at {{ $value }} — the system answers, but badly."
          action: "Compare with unoptimised segments; suspect compaction falling behind."

      - alert: CompactionFallingBehind
        expr: increase(qdrant_collection_unoptimized_segments[30m]) > 0
                and qdrant_collection_unoptimized_segments > 12
        for: 30m
        labels: { severity: minor, team: ai-platform }
        annotations:
          summary: "The optimisers no longer keep up with the write rate."
          action: "Raise max_optimization_threads or slow ingestion down."

The mistakes I have seen

Confusing freshness with latency. They are two distinct budgets carried by two distinct paths. One team kept optimising search latency while its real complaint was documents invisible for four minutes. Two weeks of work on the wrong problem.

Sharing the embedding service between ingestion and queries. The ingestion batch holds the GPU while the question waits. It is the number one cause of erratic latency I come across, and the fix takes half a day.

Leaving the optimisers unchecked. An unthrottled compaction turns a twenty-millisecond p99 into eighty for ten minutes, several times an hour. The curve looks like a comb, and nobody understands why until the two series get overlaid.

Filtering after the approximate search. It works in a demo with three tenants and breaks the moment one of them holds 80 % of the corpus. Partitioning is declared when the collection is created, never afterwards.

Forgetting the cache in the erasure procedure. The document disappears from the index, and the system keeps citing it for hours because a cached answer contains it. Audit the whole chain, not just the index.

Skipping de-duplication upstream. Operational streams repeat enormously. Embedding the same alert ten times costs ten times the price and degrades search, because the ten copies take the first ten places in every result.

Believing a vector database is enough. Hybrid search — lexical and vector — buys five to ten points of recall for a few milliseconds. No amount of HNSW tuning makes up for a missing BM25 on a corpus full of exact references.