How to Build a Streaming RAG Pipeline with Kafka

How to Build a Streaming RAG Pipeline with Kafka

Retrieval-Augmented Generation (RAG) has become the standard pattern for grounding large language models in private or rapidly changing data. Most RAG tutorials, however, describe a static world: documents are ingested once, embedded, and stored in a vector database before queries ever arrive. Real applications rarely work this way. Documents arrive continuously, embeddings go stale, and users expect answers that reflect the latest information. A streaming RAG pipeline built on top of Apache Kafka solves this problem by treating ingestion, embedding, indexing, and retrieval as continuous, decoupled events.

What Is a Streaming RAG Pipeline?

A streaming RAG pipeline is a RAG architecture in which every stage — document ingestion, chunking, embedding, vector storage, and even query handling — is connected through an event stream rather than batch jobs or synchronous API calls. Kafka acts as the central nervous system, carrying events between producers and consumers that each handle one responsibility.

In a traditional batch RAG setup, you might run a nightly job that crawls a knowledge base, chunks documents, calls an embedding API, and writes vectors to Pinecone or pgvector. In a streaming setup, the moment a new document is published — a support ticket, a wiki edit, a Slack message — an event lands on a Kafka topic, and downstream consumers react in real time. By the time a user asks a question, the relevant context is already indexed.

Why Kafka for RAG?

Kafka provides several properties that map cleanly onto the demands of a production RAG system:

• Durability and replay: If your embedding service crashes or an API rate limit is hit, Kafka retains the events. Consumers can rewind and reprocess without data loss.
• Decoupling: Ingestion, embedding, and indexing scale independently. A burst of documents does not block user queries.
• Ordering guarantees: Within a partition, events are ordered. This matters when documents have dependencies or when you need to process updates in sequence.
• Schema evolution: With Kafka's schema registry, you can evolve your document event format without breaking consumers.
• Ecosystem integration: Kafka Connect sources can pull from databases, S3, or message queues, turning existing data streams into RAG-ready events.

Architecture Overview

The pipeline consists of five logical stages, each backed by a Kafka topic:

• raw-documents — original documents from source systems.
• chunks — text chunks produced by a chunking consumer.
• embeddings — vectors produced by an embedding consumer.
• index-events — confirmation events after vectors are written to the vector store.
• user-queries — live user questions, used for async retrieval and answer generation.

Each topic is consumed by a small, focused service. This separation lets you swap embedding models, change vector stores, or add enrichment steps without rewriting the whole system.

Setting Up Kafka Locally

The fastest way to get a Kafka broker running for development is Docker Compose. Save the following as docker-compose.yml:

version: "3.8"
services:
  kafka:
    image: bitnami/kafka:3.7
    ports:
      - "9092:9092"
    environment:
      KAFKA_CFG_NODE_ID: 0
      KAFKA_CFG_PROCESS_ROLES: controller,broker
      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 0@kafka:9093
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT

Start it with docker compose up -d. You now have a single-node Kafka cluster running in KRaft mode, without Zookeeper.

Producing Raw Documents

The first stage publishes documents to the raw-documents topic. In production this would be a Kafka Connect source connector pulling from Postgres or S3. For the tutorial, we will use a Python producer.

import json
from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)

def publish_document(doc_id: str, source: str, text: str):
    event = {
        "doc_id": doc_id,
        "source": source,
        "text": text,
        "timestamp": __import__("time").time(),
    }
    producer.send("raw-documents", value=event)
    producer.flush()

publish_document("doc-001", "wiki", "Kafka is a distributed event streaming platform...")
publish_document("doc-002", "support", "How do I configure retention for a Kafka topic?")

Notice that we key events by doc_id implicitly through topic partitioning. In a real system, pass key=doc_id.encode() to send so that all events for the same document land on the same partition and stay ordered.

Chunking Consumer

The chunking consumer reads raw documents, splits them into overlapping text chunks, and publishes each chunk to the chunks topic. Keeping chunking as a separate stage means you can re-chunk the entire corpus by rewinding the consumer offset when you change your chunking strategy.

import json
from kafka import KafkaConsumer, KafkaProducer

consumer = KafkaConsumer(
    "raw-documents",
    bootstrap_servers="localhost:9092",
    group_id="chunker",
    auto_offset_reset="earliest",
    value_deserializer=lambda m: json.loads(m.decode("utf-8")),
)

producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)

def chunk_text(text: str, size: int = 500, overlap: int = 50):
    chunks = []
    start = 0
    while start < len(text):
        end = start + size
        chunks.append(text[start:end])
        start = end - overlap
    return chunks

for msg in consumer:
    doc = msg.value
    for i, chunk in enumerate(chunk_text(doc["text"])):
        producer.send("chunks", value={
            "chunk_id": f"{doc['doc_id']}-c{i}",
            "doc_id": doc["doc_id"],
            "source": doc["source"],
            "text": chunk,
            "index": i,
        })
    producer.flush()

Embedding Consumer

The embedding consumer reads chunks, calls an embedding model, and publishes vectors to the embeddings topic. This is the stage most likely to hit rate limits or latency, which is exactly why it benefits from Kafka's buffering. If the embedding API is slow, events simply queue in the topic.

import json
from kafka import KafkaConsumer, KafkaProducer
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")

consumer = KafkaConsumer(
    "chunks",
    bootstrap_servers="localhost:9092",
    group_id="embedder",
    auto_offset_reset="earliest",
    value_deserializer=lambda m: json.loads(m.decode("utf-8")),
)

producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)

for msg in consumer:
    chunk = msg.value
    vector = model.encode(chunk["text"]).tolist()
    producer.send("embeddings", value={
        "chunk_id": chunk["chunk_id"],
        "doc_id": chunk["doc_id"],
        "source": chunk["source"],
        "text": chunk["text"],
        "vector": vector,
    })
    producer.flush()

For production workloads, replace the local model with a call to an API like OpenAI or a self-hosted inference server, and add batching — encode multiple chunks per API call to reduce cost and latency.

Indexing into the Vector Store

The final ingestion consumer writes embeddings to a vector database. The example below uses pgvector with psycopg, but the pattern is identical for Pinecone, Weaviate, or Qdrant.

import json
import psycopg
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    "embeddings",
    bootstrap_servers="localhost:9092",
    group_id="indexer",
    auto_offset_reset="earliest",
    value_deserializer=lambda m: json.loads(m.decode("utf-8")),
)

conn = psycopg.connect("dbname=rag user=postgres")
conn.autocommit = True

for msg in consumer:
    rec = msg.value
    with conn.cursor() as cur:
        cur.execute(
            """
            INSERT INTO chunks (chunk_id, doc_id, source, text, embedding)
            VALUES (%s, %s, %s, %s, %s)
            ON CONFLICT (chunk_id) DO UPDATE SET text = EXCLUDED.text,
                                                  embedding = EXCLUDED.embedding
            """,
            (rec["chunk_id"], rec["doc_id"], rec["source"],
             rec["text"], rec["vector"]),
        )

The ON CONFLICT clause makes the indexer idempotent. If you replay the topic after a crash, duplicate writes simply update the existing row. Idempotency is essential in any streaming pipeline because exactly-once semantics are expensive and often unnecessary when downstream writes are deduplicated.

Handling User Queries as Events

So far the pipeline handles ingestion. For a fully streaming system, user queries can also flow through Kafka. A query service publishes the question, a retrieval consumer performs vector search and calls the LLM, and the answer is published back to a response topic the user's session is subscribed to.

import json
import psycopg
from kafka import KafkaConsumer, KafkaProducer
from openai import OpenAI

client = OpenAI()
conn = psycopg.connect("dbname=rag user=postgres")

consumer = KafkaConsumer(
    "user-queries",
    bootstrap_servers="localhost:9092",
    group_id="rag-responder",
    value_deserializer=lambda m: json.loads(m.decode("utf-8")),
)

producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)

for msg in consumer:
    query = msg.value
    q_vec = client.embeddings.create(
        input=query["question"], model="text-embedding-3-small"
    ).data[0].embedding

    with conn.cursor() as cur:
        cur.execute(
            "SELECT text FROM chunks ORDER BY embedding <-> %s LIMIT 5",
            (q_vec,),
        )
        context = "\n\n".join(r[0] for r in cur.fetchall())

    prompt = f"Answer using this context:\n{context}\n\nQuestion: {query['question']}"
    answer = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    ).choices[0].message.content

    producer.send("query-responses", value={
        "query_id": query["query_id"],
        "answer": answer,
    })
    producer.flush()

Best Practices

• Use keys for partitioning. Key events by doc_id so updates to the same document stay ordered and land on the same partition.
• Make consumers idempotent. Replays are inevitable. Design every consumer so that reprocessing an event produces the same end state.
• Separate topics by stage, not by source. Stage-based topics let you reorder or insert pipeline steps without rewiring producers.
• Use a schema registry. Avro or Protobuf schemas prevent silent breakage when event formats evolve.
• Monitor consumer lag. Lag on the embeddings topic tells you the embedding service is the bottleneck before users notice stale answers.
• Batch embedding calls. Embedding APIs are far cheaper per token when you batch. Accumulate a small window of chunks before calling the API.
• Handle deletions explicitly. Publish tombstone events when documents are removed so the indexer can delete the corresponding vectors.
• Version your embeddings. Include a model name and version in each event so you can re-index only affected chunks when you upgrade the embedding model.

Conclusion

Building a RAG pipeline on Kafka transforms retrieval-augmented generation from a fragile batch job into a resilient, continuously running system. Each stage — ingestion, chunking, embedding, indexing, and retrieval — becomes an independent consumer that can scale, fail, and recover without affecting the others. The result is a pipeline that stays fresh as your data changes, tolerates outages gracefully through Kafka's durability and replay, and grows cleanly as you add new sources, models, or vector stores. Start with the single-node setup above, make every consumer idempotent, and you will have a foundation that scales from a prototype to a production-grade real-time knowledge service.