Batch vs Real-Time RAG: The Architecture Choice That Makes or Breaks Your System

TL;DR

A production RAG system must handle two radically different flows: batch (document ingestion, re-indexing, analytics) and real-time (user queries, live updates). The wrong choice leads to either unacceptable latency or exploding costs. This guide presents batch, real-time, and hybrid architectures with cost comparisons, diagrams, and a decision matrix.

The fundamental dilemma

Every RAG system faces the same tradeoff:

                    BATCH                    REAL-TIME
                    ─────                    ─────────
Latency:            Minutes/hours            Milliseconds
Throughput:         High (1M+ docs/h)        Medium (100-1K req/s)
Cost:               Low per document         High per request
Consistency:        Eventually               Immediately
Complexity:         Low                      High
Use case:           Ingestion, analytics     Queries, chat

The right architecture is not one OR the other - it is a hybrid system that uses each approach in the right place.

Batch Architecture: when and how

Batch use cases

Classic batch architecture

┌──────────────────────────────────────────────────────────────┐
│                    BATCH PIPELINE                              │
├──────────────────────────────────────────────────────────────┤
│                                                               │
│  Sources              Queue              Workers              │
│  ┌─────────┐    ┌──────────────┐    ┌──────────────┐        │
│  │ S3/GCS  │───>│              │───>│  Worker 1    │        │
│  │ Ext API │───>│  Redis Queue │───>│  Worker 2    │──> Qdrant│
│  │ Webhook │───>│  or Celery   │───>│  Worker 3    │        │
│  │ Upload  │───>│              │───>│  Worker N    │        │
│  └─────────┘    └──────────────┘    └──────────────┘        │
│                                                               │
│  Steps per worker:                                            │
│  1. Download the document                                     │
│  2. Parse (PDF, HTML, DOCX)                                   │
│  3. Chunking (semantic/fixed)                                 │
│  4. Embedding (batch API)                                     │
│  5. Upsert into Qdrant                                        │
│  6. Update metadata                                           │
└──────────────────────────────────────────────────────────────┘

Implementation with Celery + Redis

DEVELOPERpython
from celery import Celery
from celery.utils.log import get_task_logger

app = Celery('rag_batch', broker='redis://localhost:6379/0')
logger = get_task_logger(__name__)

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def process_document(self, document_id: str, source_url: str):
    """Process a document in batch."""
    try:
        # 1. Download
        content = download_document(source_url)
        logger.info(f"Downloaded {document_id}: {len(content)} bytes")

        # 2. Parse
        parsed = parse_document(content, detect_format(source_url))

        # 3. Chunking
        chunks = semantic_chunking(parsed.text, max_tokens=512, overlap=50)
        logger.info(f"Created {len(chunks)} chunks for {document_id}")

        # 4. Embedding (batch for efficiency)
        embeddings = embed_batch(
            [c.text for c in chunks],
            model="text-embedding-3-large",
            batch_size=100  # 100 chunks at a time
        )

        # 5. Upsert into Qdrant
        points = [
            {
                "id": f"{document_id}_{i}",
                "vector": emb,
                "payload": {
                    "document_id": document_id,
                    "chunk_index": i,
                    "text": chunk.text,
                    "metadata": chunk.metadata
                }
            }
            for i, (chunk, emb) in enumerate(zip(chunks, embeddings))
        ]
        qdrant_client.upsert("documents", points)

        # 6. Update status
        update_document_status(document_id, "indexed",
                              chunks_count=len(chunks))

        return {"document_id": document_id, "chunks": len(chunks)}

    except Exception as exc:
        logger.error(f"Failed to process {document_id}: {exc}")
        self.retry(exc=exc)

@app.task
def batch_ingest(document_ids: list[str]):
    """Launch batch ingestion of multiple documents."""
    from celery import group

    tasks = group(
        process_document.s(doc_id, get_source_url(doc_id))
        for doc_id in document_ids
    )
    result = tasks.apply_async()
    return {"job_id": result.id, "total": len(document_ids)}

Optimization: batch embedding

The most important batch trick: use embedding APIs in batch rather than one by one.

DEVELOPERpython
from openai import OpenAI

client = OpenAI()

def embed_batch(texts: list[str], model: str = "text-embedding-3-large",
                batch_size: int = 100) -> list[list[float]]:
    """Batch embedding - 10x faster than individual."""
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = client.embeddings.create(
            model=model,
            input=batch
        )
        batch_embeddings = [item.embedding for item in response.data]
        all_embeddings.extend(batch_embeddings)

    return all_embeddings

# Performance comparison
# Individual: 1000 docs x 200ms = 200 seconds
# Batch 100: 10 requests x 2s = 20 seconds (10x faster)

Real-Time Architecture: when and how

Real-time use cases

Real-time architecture

┌──────────────────────────────────────────────────────────────┐
│                   REAL-TIME PIPELINE                          │
├──────────────────────────────────────────────────────────────┤
│                                                               │
│  Client           API Gateway        RAG Pipeline             │
│  ┌───────┐    ┌──────────────┐    ┌──────────────────┐      │
│  │Widget │───>│  FastAPI      │───>│ 1. Query embed   │      │
│  │ API  │───>│  + Rate limit │───>│ 2. Vector search │      │
│  │ Chat │───>│  + Auth       │───>│ 3. Rerank        │──> SSE│
│  └───────┘    └──────────────┘    │ 4. LLM generate  │      │
│                                    │ 5. Stream tokens │      │
│                                    └──────────────────┘      │
│                                                               │
│  Cache layers:                                                │
│  ┌─────────────────────────────────────────────┐             │
│  │ L1: In-memory (exact match)  → 1ms          │             │
│  │ L2: Redis (semantic cache)   → 5ms          │             │
│  │ L3: Qdrant (vector search)   → 10-50ms      │             │
│  └─────────────────────────────────────────────┘             │
└──────────────────────────────────────────────────────────────┘

Implementation with multi-level cache

DEVELOPERpython
import hashlib
import redis.asyncio as redis
from qdrant_client import AsyncQdrantClient

class RAGRealtimePipeline:
    def __init__(self):
        self.redis = redis.Redis()
        self.qdrant = AsyncQdrantClient()
        self.local_cache = {}  # In-memory LRU cache

    async def query(self, question: str, user_id: str) -> dict:
        # L1: Exact cache (in-memory)
        cache_key = hashlib.md5(question.lower().strip().encode()).hexdigest()
        if cache_key in self.local_cache:
            return self.local_cache[cache_key]

        # L2: Semantic cache (Redis)
        cached = await self.redis.get(f"rag:cache:{cache_key}")
        if cached:
            result = json.loads(cached)
            self.local_cache[cache_key] = result
            return result

        # L3: Full RAG pipeline
        query_embedding = await self.embed_query(question)

        # Vector search
        search_results = await self.qdrant.search(
            collection_name="documents",
            query_vector=query_embedding,
            limit=10,
            score_threshold=0.7
        )

        # Reranking
        reranked = await self.rerank(question, search_results)

        # LLM generation (streaming)
        response = await self.generate(question, reranked[:5])

        result = {
            "answer": response.text,
            "sources": [self._format_source(s) for s in reranked[:5]],
            "confidence": self._compute_confidence(reranked)
        }

        # Cache (TTL 1h)
        await self.redis.setex(
            f"rag:cache:{cache_key}", 3600, json.dumps(result)
        )
        self.local_cache[cache_key] = result

        return result

Hybrid Architecture: the best of both worlds

Hybrid architecture is the production standard. It combines batch and real-time with an event system.

Hybrid architecture diagram

┌──────────────────────────────────────────────────────────────────┐
│                    HYBRID RAG ARCHITECTURE                        │
├──────────────────────────────────────────────────────────────────┤
│                                                                   │
│  SOURCES                    EVENT BUS              CONSUMERS      │
│  ┌──────────┐          ┌──────────────┐                          │
│  │ Upload   │─────────>│              │──> [Batch Worker]        │
│  │ API sync │─────────>│    Kafka /   │      → Ingestion         │
│  │ Webhook  │─────────>│    Redis     │      → Re-indexing       │
│  │ Cron job │─────────>│    Streams   │      → Analytics         │
│  └──────────┘          │              │                          │
│                         │              │──> [Realtime Worker]     │
│  USERS                 │              │      → Query processing  │
│  ┌──────────┐          │              │      → Live updates      │
│  │ Widget   │─────────>│              │      → Streaming         │
│  │ API      │─────────>│              │                          │
│  │ Chat     │─────────>│              │                          │
│  └──────────┘          └──────────────┘                          │
│                                                                   │
│  STORAGE                                                          │
│  ┌───────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐       │
│  │ PostgreSQL│  │  Qdrant  │  │  Redis   │  │   S3     │       │
│  │ (metadata)│  │ (vectors)│  │ (cache)  │  │ (files)  │       │
│  └───────────┘  └──────────┘  └──────────┘  └──────────┘       │
└──────────────────────────────────────────────────────────────────┘

Event-driven document updates

DEVELOPERpython
import redis.asyncio as redis
import json
from datetime import datetime

class EventBus:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)

    async def publish(self, event_type: str, data: dict):
        """Publish an event to the bus."""
        event = {
            "type": event_type,
            "data": data,
            "timestamp": datetime.utcnow().isoformat(),
            "id": str(uuid4())
        }
        await self.redis.xadd(
            f"events:{event_type}",
            {"payload": json.dumps(event)}
        )

    async def subscribe(self, event_type: str, consumer_group: str,
                        consumer_name: str):
        """Subscribe to an event type."""
        try:
            await self.redis.xgroup_create(
                f"events:{event_type}",
                consumer_group, id="0", mkstream=True
            )
        except redis.ResponseError:
            pass  # Group already exists

        while True:
            messages = await self.redis.xreadgroup(
                consumer_group, consumer_name,
                {f"events:{event_type}": ">"},
                count=10, block=5000
            )

            for stream, msgs in messages:
                for msg_id, msg_data in msgs:
                    event = json.loads(msg_data[b"payload"])
                    yield event
                    await self.redis.xack(
                        f"events:{event_type}",
                        consumer_group, msg_id
                    )

# Publishing events
event_bus = EventBus()

# When a document is uploaded
await event_bus.publish("document.uploaded", {
    "document_id": "doc_123",
    "source": "api_upload",
    "size_bytes": 45000
})

# When a document is updated
await event_bus.publish("document.updated", {
    "document_id": "doc_123",
    "changed_sections": ["section_2", "section_5"],
    "update_type": "partial"
})

Consumer: incremental updates

DEVELOPERpython
async def incremental_update_consumer():
    """Consumer that updates vectors incrementally."""
    async for event in event_bus.subscribe(
        "document.updated", "indexing_group", "worker_1"
    ):
        doc_id = event["data"]["document_id"]
        update_type = event["data"]["update_type"]

        if update_type == "partial":
            # Partial update: re-index only changed sections
            changed_sections = event["data"]["changed_sections"]
            doc = await fetch_document(doc_id)

            for section_id in changed_sections:
                section_content = doc.get_section(section_id)
                chunks = semantic_chunking(section_content)
                embeddings = await embed_batch([c.text for c in chunks])

                # Delete old chunks for this section
                await qdrant_client.delete(
                    collection_name="documents",
                    points_selector=FilterSelector(
                        filter=Filter(
                            must=[
                                FieldCondition(
                                    key="document_id",
                                    match=MatchValue(value=doc_id)
                                ),
                                FieldCondition(
                                    key="section_id",
                                    match=MatchValue(value=section_id)
                                )
                            ]
                        )
                    )
                )

                # Insert new chunks
                await qdrant_client.upsert("documents", [
                    PointStruct(
                        id=uuid4().int >> 64,
                        vector=emb,
                        payload={
                            "document_id": doc_id,
                            "section_id": section_id,
                            "text": chunk.text
                        }
                    )
                    for chunk, emb in zip(chunks, embeddings)
                ])

        elif update_type == "full":
            await process_document.delay(doc_id)

Queue system comparison

Why Redis Streams for most RAG systems

Redis Streams offers the best balance for a typical RAG system:

1. Already there for caching: Redis serves as L2 cache, no additional component
2. Sub-millisecond latency: perfect for live updates
3. Consumer groups: load distribution between workers
4. Replay: reprocessing possible on error
5. Low complexity: no Kafka cluster (KRaft controllers, brokers) to manage

Decision matrix

When to use what?

Decision flowchart

Is latency critical (<1s)?
├── YES → Does volume exceed 100 docs/min?
│         ├── YES → HYBRID architecture
│         │         (queue + real-time workers)
│         └── NO  → Pure REAL-TIME architecture
│                   (synchronous API + cache)
└── NO  → Does volume exceed 10K docs?
          ├── YES → Pure BATCH architecture
          │         (Celery + distributed workers)
          └── NO  → Simple BATCH architecture
                    (cron job + Python script)

Cost comparison

Monthly infrastructure (10M documents, 100K queries/day)

Cost per query

The hybrid model reduces per-query cost by 68% compared to pure real-time thanks to semantic caching.

Advanced patterns

Pattern: Write-behind cache

The write-behind cache allows responding immediately while updating the index in the background.

DEVELOPERpython
class WriteBehindRAG:
    """Respond from cache, update index in background."""

    async def update_document(self, doc_id: str, new_content: str):
        # 1. Update cache immediately
        await self.redis.hset(f"doc:{doc_id}", "content", new_content)
        await self.redis.hset(f"doc:{doc_id}", "status", "pending_index")

        # 2. Publish event for re-indexing
        await self.event_bus.publish("document.updated", {
            "document_id": doc_id,
            "update_type": "full"
        })

        # 3. Queries use cache in the meantime
        return {"status": "updated", "indexing": "in_progress"}

    async def query(self, question: str):
        # Normal vector search
        results = await self.qdrant.search(question_embedding)

        # Enrich with cache (recently modified documents)
        for result in results:
            cached = await self.redis.hgetall(f"doc:{result.id}")
            if cached and cached.get("status") == "pending_index":
                result.text = cached["content"]

        return results

Pattern: Backpressure

Prevent system overload when volume spikes.

DEVELOPERpython
class BackpressureController:
    def __init__(self, max_queue_size: int = 10000):
        self.max_queue_size = max_queue_size

    async def should_accept(self, queue_name: str) -> bool:
        """Check if the queue can accept more messages."""
        queue_size = await self.redis.xlen(f"events:{queue_name}")

        if queue_size > self.max_queue_size:
            return False

        if queue_size > self.max_queue_size * 0.8:
            await asyncio.sleep(0.1)

        return True

    async def ingest_with_backpressure(self, documents: list[dict]):
        """Ingestion with backpressure control."""
        accepted, rejected = [], []

        for doc in documents:
            if await self.should_accept("document.uploaded"):
                await self.event_bus.publish("document.uploaded", doc)
                accepted.append(doc["id"])
            else:
                rejected.append(doc["id"])

        return {
            "accepted": len(accepted),
            "rejected": len(rejected),
            "retry_after": 30 if rejected else None
        }

Our architecture at Ailog

At Ailog, we use a hybrid architecture:

• Batch: client document ingestion (Celery + Redis), nightly re-indexing
• Real-time: widget queries with semantic cache (Redis) + Qdrant
• Event bus: Redis Streams for batch/realtime coordination
• Multi-level cache: L1 in-memory, L2 Redis, L3 Qdrant

The result: p50 latency of 800ms for a complete query (retrieval + LLM), with documents indexed in less than 30 seconds after upload.

Discover our guide on RAG response streaming and caching strategies.

Conclusion

The choice between batch and real-time is not binary. The best RAG systems use a hybrid architecture:

1. Batch for bulk ingestion and re-indexing
2. Real-time for user queries and critical updates
3. Event bus (Redis Streams) to coordinate both
4. Multi-level cache to reduce cost and latency

Start simple (batch + synchronous API), then add complexity when volume demands it.

Want a hybrid RAG system without managing the infrastructure? Try Ailog - we handle batch, real-time, cache, and event bus for you.