HinterBuild logoHinterBuild
Back to blog
AI Systems · 10 min read

Event-Driven AI Pipelines with Kafka & SQS

Event-Driven AI Pipelines with Kafka & SQS guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable.

Muhammad Abdul Sami, author

Muhammad Abdul Sami

· 10 min read

  • RAG
  • Embeddings
  • Vector Databases
  • Evaluation

Production AI systems need reliable asynchronous processing - event-driven architectures with Kafka and SQS enable scalable, decoupled AI pipelines. This guide covers patterns from systems processing 100,000+ AI events daily.

What you'll learn:

  • Event-driven architecture for AI pipelines
  • Kafka vs SQS for AI workloads
  • Producer-consumer patterns for LLM processing
  • Backpressure and rate limiting strategies
  • Dead letter queues and error handling
  • Production deployment and monitoring

Reading time: 16 minutes

Key Takeaways:

  • Treat Event-Driven AI Pipelines with Kafka & SQS as a system with an explicit input and output contract.
  • Benchmark a representative baseline before choosing an optimization.
  • Bound retries, queues, concurrency, and total request deadlines.
  • Roll out through offline replay, shadow traffic, and a measurable canary.
  • Keep rollback simple and attach version identifiers to every decision.

Table of Contents:

Why Event-Driven AI Architectures

Traditional request-response APIs don't scale for AI workloads - LLM inference takes seconds, batch embeddings take minutes, and RAG indexing takes hours. Event-driven architectures decouple producers from consumers.

The Problem with Synchronous AI APIs

python
@app.post("/api/generate-report")
async def generate_report(data: dict):
    # This blocks the API server
    embeddings = await generate_embeddings(data["documents"])  # 10s
    analysis = await llm_analyze(data["query"], embeddings)  # 15s
    report = await generate_pdf(analysis)  # 5s
    
    return {"report_url": report.url}  # 30s total - terrible UX

Issues:

  • API timeout risk (most proxies timeout at 30s)
  • Server resources blocked (thread/worker tied up)
  • No retry on failure
  • No progress tracking
  • Doesn't scale beyond 10-20 concurrent requests

Event-Driven Solution

python
# Event-driven solution - returns immediately
@app.post("/api/generate-report")
async def generate_report(data: dict):
    # Publish event to queue
    job_id = str(uuid4())
    
    await publish_event(
        topic="report.requested",
        key=job_id,
        value={
            "job_id": job_id,
            "user_id": data["user_id"],
            "documents": data["documents"],
            "query": data["query"]
        }
    )
    
    return {
        "job_id": job_id,
        "status": "queued",
        "status_url": f"/api/jobs/{job_id}"
    }  # Returns in <100ms

# Separate consumer processes the job
async def process_report_job(message):
    data = message.value
    
    try:
        # Update status
        await update_job_status(data["job_id"], "processing")
        
        # Process (can take minutes)
        embeddings = await generate_embeddings(data["documents"])
        analysis = await llm_analyze(data["query"], embeddings)
        report = await generate_pdf(analysis)
        
        # Complete
        await update_job_status(data["job_id"], "completed", {"report_url": report.url})
        
        # Publish completion event
        await publish_event(
            topic="report.completed",
            key=data["job_id"],
            value={"job_id": data["job_id"], "report_url": report.url}
        )
        
    except Exception as e:
        await update_job_status(data["job_id"], "failed", {"error": str(e)})
        raise

Benefits:

  • Fast API response (<100ms vs 30s)
  • Horizontal scaling - add more consumers
  • Retry on failure - built into messaging system
  • Progress tracking - update job status at each stage
  • Decoupled services - API server separate from processing workers

For production backend API engineering, event-driven is essential for AI workloads.

When to use event-driven:

  • Long-running AI tasks (>5s)
  • Batch processing (embeddings for 1000+ docs)
  • Multi-stage pipelines (RAG: chunk → embed → index)
  • High concurrency (100+ simultaneous requests)
  • AI agent workflows with tool calls

When request-response is fine:

  • Quick inference (<2s)
  • Low concurrency (<10 requests)
  • Simple one-off tasks

Kafka for AI Pipelines

Apache Kafka is ideal for high-throughput AI pipelines with multiple consumers and replay requirements.

Kafka Architecture for AI

python
from confluent_kafka import Producer, Consumer, KafkaError
import json
from typing import Dict, Callable
import asyncio

class AIKafkaProducer:
    """Production Kafka producer for AI events."""
    
    def __init__(self, bootstrap_servers: str):
        self.producer = Producer({
            'bootstrap.servers': bootstrap_servers,
            'client.id': 'ai-api-producer',
            'acks': 'all',  # Wait for all replicas
            'retries': 3,
            'max.in.flight.requests.per.connection': 1,  # Ordering
            'compression.type': 'gzip',  # Large payloads
            'linger.ms': 10,  # Batch for throughput
        })
    
    async def publish(
        self,
        topic: str,
        key: str,
        value: dict,
        headers: dict = None
    ):
        """Publish AI event to Kafka."""
        
        def delivery_callback(err, msg):
            if err:
                logger.error(f"Delivery failed: {err}")
            else:
                logger.info(
                    f"Message delivered to {msg.topic()} "
                    f"partition [{msg.partition()}] at offset {msg.offset()}"
                )
        
        # Serialize value
        value_bytes = json.dumps(value).encode('utf-8')
        
        # Prepare headers
        kafka_headers = []
        if headers:
            kafka_headers = [(k, v.encode('utf-8')) for k, v in headers.items()]
        
        # Produce
        self.producer.produce(
            topic=topic,
            key=key.encode('utf-8'),
            value=value_bytes,
            headers=kafka_headers,
            callback=delivery_callback
        )
        
        # Flush to ensure delivery
        self.producer.flush()
    
    def close(self):
        self.producer.flush()

class AIKafkaConsumer:
    """Production Kafka consumer for AI processing."""
    
    def __init__(
        self,
        bootstrap_servers: str,
        group_id: str,
        topics: list[str]
    ):
        self.consumer = Consumer({
            'bootstrap.servers': bootstrap_servers,
            'group.id': group_id,
            'auto.offset.reset': 'earliest',
            'enable.auto.commit': False,  # Manual commit
            'max.poll.interval.ms': 300000,  # 5 min (for slow AI processing)
            'session.timeout.ms': 45000,
            'heartbeat.interval.ms': 3000,
        })
        
        self.consumer.subscribe(topics)
        self.handlers: Dict[str, Callable] = {}
    
    def register_handler(self, topic: str, handler: Callable):
        """Register handler function for topic."""
        self.handlers[topic] = handler
    
    async def start(self):
        """Start consuming messages."""
        
        logger.info(f"Starting consumer for topics: {self.consumer.subscription()}")
        
        try:
            while True:
                msg = self.consumer.poll(timeout=1.0)
                
                if msg is None:
                    continue
                
                if msg.error():
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        continue
                    else:
                        logger.error(f"Consumer error: {msg.error()}")
                        continue
                
                # Process message
                topic = msg.topic()
                
                if topic not in self.handlers:
                    logger.warning(f"No handler for topic: {topic}")
                    continue
                
                try:
                    # Deserialize
                    value = json.loads(msg.value().decode('utf-8'))
                    headers = {k: v.decode('utf-8') for k, v in (msg.headers() or [])}
                    
                    # Call handler
                    handler = self.handlers[topic]
                    await handler(value, headers)
                    
                    # Commit offset
                    self.consumer.commit(msg)
                    
                    logger.info(
                        f"Processed message from {topic} "
                        f"partition [{msg.partition()}] offset {msg.offset()}"
                    )
                    
                except Exception as e:
                    logger.error(f"Handler error: {e}", exc_info=True)
                    # Don't commit - message will be reprocessed
                    
        except KeyboardInterrupt:
            logger.info("Consumer interrupted")
        finally:
            self.consumer.close()

AI Pipeline Example with Kafka

python
# Document processing pipeline

# Topics
TOPICS = {
    "document.uploaded": "doc-uploaded",
    "document.chunked": "doc-chunked",
    "document.embedded": "doc-embedded",
    "document.indexed": "doc-indexed"
}

# Stage 1: Document upload handler (API)
@app.post("/api/documents/upload")
async def upload_document(file: UploadFile):
    # Save file
    document_id = await save_document(file)
    
    # Publish upload event
    await kafka_producer.publish(
        topic=TOPICS["document.uploaded"],
        key=document_id,
        value={
            "document_id": document_id,
            "filename": file.filename,
            "size_bytes": file.size,
            "uploaded_at": datetime.now().isoformat()
        }
    )
    
    return {"document_id": document_id, "status": "processing"}

# Stage 2: Chunking consumer
async def handle_document_uploaded(value: dict, headers: dict):
    """Chunk document into segments."""
    
    document_id = value["document_id"]
    
    # Download and chunk
    content = await download_document(document_id)
    chunks = chunk_text(content, chunk_size=500)
    
    # Publish chunking complete
    await kafka_producer.publish(
        topic=TOPICS["document.chunked"],
        key=document_id,
        value={
            "document_id": document_id,
            "chunk_count": len(chunks),
            "chunks": [
                {"chunk_id": f"{document_id}_{i}", "text": chunk}
                for i, chunk in enumerate(chunks)
            ]
        }
    )
    
    logger.info(f"Chunked document {document_id} into {len(chunks)} chunks")

# Stage 3: Embedding consumer
async def handle_document_chunked(value: dict, headers: dict):
    """Generate embeddings for chunks."""
    
    document_id = value["document_id"]
    chunks = value["chunks"]
    
    # Generate embeddings in batch
    embeddings = await openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=[chunk["text"] for chunk in chunks]
    )
    
    # Publish embeddings
    await kafka_producer.publish(
        topic=TOPICS["document.embedded"],
        key=document_id,
        value={
            "document_id": document_id,
            "embeddings": [
                {
                    "chunk_id": chunks[i]["chunk_id"],
                    "text": chunks[i]["text"],
                    "embedding": emb.embedding
                }
                for i, emb in enumerate(embeddings.data)
            ]
        }
    )
    
    logger.info(f"Generated {len(embeddings.data)} embeddings for {document_id}")

# Stage 4: Indexing consumer
async def handle_document_embedded(value: dict, headers: dict):
    """Index embeddings in vector database."""
    
    document_id = value["document_id"]
    embeddings = value["embeddings"]
    
    # Batch upsert to Pinecone
    vectors = [
        {
            "id": emb["chunk_id"],
            "values": emb["embedding"],
            "metadata": {
                "document_id": document_id,
                "text": emb["text"]
            }
        }
        for emb in embeddings
    ]
    
    await pinecone_index.upsert(vectors=vectors)
    
    # Publish indexing complete
    await kafka_producer.publish(
        topic=TOPICS["document.indexed"],
        key=document_id,
        value={
            "document_id": document_id,
            "indexed_at": datetime.now().isoformat(),
            "vector_count": len(vectors)
        }
    )
    
    logger.info(f"Indexed {len(vectors)} vectors for {document_id}")

# Register handlers
consumer = AIKafkaConsumer(
    bootstrap_servers="localhost:9092",
    group_id="document-pipeline",
    topics=list(TOPICS.values())
)

consumer.register_handler(TOPICS["document.uploaded"], handle_document_uploaded)
consumer.register_handler(TOPICS["document.chunked"], handle_document_chunked)
consumer.register_handler(TOPICS["document.embedded"], handle_document_embedded)

# Start consumer
asyncio.run(consumer.start())

Kafka advantages for AI:

  • High throughput - handles 100K+ messages/sec
  • Message replay - reprocess from any offset
  • Multiple consumers - different services can consume same events
  • Guaranteed ordering per partition
  • Long retention - keep events for debugging

See our RAG system guide for production RAG pipelines.

SQS for AI Workloads

Amazon SQS is simpler than Kafka - fully managed, no infrastructure, perfect for async task queues.

SQS Pattern for AI Jobs

python
import boto3
import json
from typing import Optional
import asyncio

class AISQSProducer:
    """Production SQS producer for AI job queues."""
    
    def __init__(self, queue_url: str):
        self.sqs = boto3.client('sqs')
        self.queue_url = queue_url
    
    async def enqueue_job(
        self,
        job_type: str,
        job_data: dict,
        delay_seconds: int = 0
    ) -> str:
        """Enqueue AI processing job."""
        
        message_body = {
            "job_type": job_type,
            "job_data": job_data,
            "enqueued_at": datetime.now().isoformat()
        }
        
        response = self.sqs.send_message(
            QueueUrl=self.queue_url,
            MessageBody=json.dumps(message_body),
            DelaySeconds=delay_seconds,
            MessageAttributes={
                'JobType': {
                    'StringValue': job_type,
                    'DataType': 'String'
                }
            }
        )
        
        return response['MessageId']

class AISQSConsumer:
    """Production SQS consumer for AI processing."""
    
    def __init__(
        self,
        queue_url: str,
        max_concurrent: int = 10,
        visibility_timeout: int = 300  # 5 min
    ):
        self.sqs = boto3.client('sqs')
        self.queue_url = queue_url
        self.max_concurrent = max_concurrent
        self.visibility_timeout = visibility_timeout
        self.handlers = {}
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    def register_handler(self, job_type: str, handler: Callable):
        """Register handler for job type."""
        self.handlers[job_type] = handler
    
    async def process_message(self, message: dict):
        """Process single SQS message."""
        
        async with self.semaphore:
            try:
                # Parse message
                body = json.loads(message['Body'])
                job_type = body['job_type']
                job_data = body['job_data']
                
                if job_type not in self.handlers:
                    logger.warning(f"No handler for job type: {job_type}")
                    return
                
                # Process with handler
                handler = self.handlers[job_type]
                await handler(job_data)
                
                # Delete message on success
                self.sqs.delete_message(
                    QueueUrl=self.queue_url,
                    ReceiptHandle=message['ReceiptHandle']
                )
                
                logger.info(f"Processed job {job_type} - {message['MessageId']}")
                
            except Exception as e:
                logger.error(f"Job processing failed: {e}", exc_info=True)
                # Message will become visible again after visibility timeout
    
    async def start(self):
        """Start consuming messages."""
        
        logger.info(f"Starting SQS consumer for {self.queue_url}")
        
        while True:
            try:
                # Long poll for messages
                response = self.sqs.receive_message(
                    QueueUrl=self.queue_url,
                    MaxNumberOfMessages=10,
                    WaitTimeSeconds=20,  # Long polling
                    VisibilityTimeout=self.visibility_timeout,
                    MessageAttributeNames=['All']
                )
                
                messages = response.get('Messages', [])
                
                if not messages:
                    continue
                
                # Process messages concurrently
                tasks = [
                    self.process_message(msg)
                    for msg in messages
                ]
                
                await asyncio.gather(*tasks, return_exceptions=True)
                
            except Exception as e:
                logger.error(f"Consumer error: {e}", exc_info=True)
                await asyncio.sleep(5)

LLM Batch Processing with SQS

python
# Enqueue LLM inference jobs
@app.post("/api/llm/batch-inference")
async def batch_inference(requests: list[dict]):
    """Enqueue batch of LLM requests."""
    
    job_ids = []
    
    for req in requests:
        job_id = str(uuid4())
        
        await sqs_producer.enqueue_job(
            job_type="llm_inference",
            job_data={
                "job_id": job_id,
                "model": req["model"],
                "messages": req["messages"],
                "user_id": req["user_id"]
            }
        )
        
        job_ids.append(job_id)
    
    return {
        "job_ids": job_ids,
        "status": "queued",
        "check_status": "/api/jobs/batch"
    }

# Consumer: Process LLM inference
async def handle_llm_inference(job_data: dict):
    """Process LLM inference job."""
    
    job_id = job_data["job_id"]
    
    try:
        # Update status
        await update_job_status(job_id, "processing")
        
        # Call LLM
        response = await openai_client.chat.completions.create(
            model=job_data["model"],
            messages=job_data["messages"]
        )
        
        result = {
            "content": response.choices[0].message.content,
            "usage": response.usage.dict()
        }
        
        # Save result
        await save_job_result(job_id, result)
        
        # Update status
        await update_job_status(job_id, "completed", result)
        
        logger.info(f"Completed LLM inference job {job_id}")
        
    except Exception as e:
        await update_job_status(job_id, "failed", {"error": str(e)})
        raise

# Register handler
sqs_consumer.register_handler("llm_inference", handle_llm_inference)

# Start consumer
asyncio.run(sqs_consumer.start())

SQS advantages:

  • Fully managed - no infrastructure to maintain
  • Auto-scaling - handles any throughput
  • Dead letter queues - built-in error handling
  • FIFO queues - guaranteed ordering when needed
  • Cost-effective - pay per request

SQS limitations:

  • Message size limit - 256KB max (use S3 for large payloads)
  • No message replay - consumed messages are deleted
  • Limited retention - 14 days max

Producer Patterns

Best practices for publishing AI events:

Event Schema Versioning

python
from pydantic import BaseModel, Field
from enum import Enum

class EventVersion(str, Enum):
    V1 = "1.0"
    V2 = "2.0"

class DocumentEvent(BaseModel):
    """Versioned event schema."""
    
    version: EventVersion = EventVersion.V2
    event_type: str
    document_id: str
    timestamp: str
    
    # V2 additions
    tenant_id: Optional[str] = None
    metadata: dict = Field(default_factory=dict)

async def publish_document_event(
    event_type: str,
    document_id: str,
    **kwargs
):
    """Publish versioned document event."""
    
    event = DocumentEvent(
        version=EventVersion.V2,
        event_type=event_type,
        document_id=document_id,
        timestamp=datetime.now().isoformat(),
        **kwargs
    )
    
    await kafka_producer.publish(
        topic="documents",
        key=document_id,
        value=event.dict(),
        headers={"schema_version": event.version}
    )

Batching for Throughput

python
class BatchedProducer:
    """Batch events for higher throughput."""
    
    def __init__(self, producer, batch_size: int = 100, flush_interval: float = 1.0):
        self.producer = producer
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self.batch = []
        self.last_flush = time.time()
        self.lock = asyncio.Lock()
    
    async def publish(self, topic: str, key: str, value: dict):
        """Add to batch."""
        
        async with self.lock:
            self.batch.append((topic, key, value))
            
            # Flush if batch full or interval elapsed
            if len(self.batch) >= self.batch_size or \
               time.time() - self.last_flush > self.flush_interval:
                await self.flush()
    
    async def flush(self):
        """Send batch to Kafka."""
        
        if not self.batch:
            return
        
        for topic, key, value in self.batch:
            self.producer.produce(topic, key=key, value=json.dumps(value))
        
        self.producer.flush()
        self.batch.clear()
        self.last_flush = time.time()
        
        logger.info(f"Flushed batch of {len(self.batch)} messages")

Consumer Patterns

Reliable consumption patterns for AI processing:

Idempotent Processing

python
import hashlib

class IdempotentConsumer:
    """Ensure each message processed exactly once."""
    
    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 86400 * 7  # 7 days
    
    def get_message_id(self, message: dict) -> str:
        """Generate idempotency key from message."""
        key_data = json.dumps(message, sort_keys=True)
        return hashlib.sha256(key_data.encode()).hexdigest()
    
    async def is_processed(self, message_id: str) -> bool:
        """Check if message already processed."""
        return await self.redis.exists(f"processed:{message_id}")
    
    async def mark_processed(self, message_id: str):
        """Mark message as processed."""
        await self.redis.setex(
            f"processed:{message_id}",
            self.ttl,
            datetime.now().isoformat()
        )
    
    async def process_idempotently(self, message: dict, handler: Callable):
        """Process message with idempotency."""
        
        message_id = self.get_message_id(message)
        
        if await self.is_processed(message_id):
            logger.info(f"Skipping duplicate message {message_id}")
            return
        
        # Process
        await handler(message)
        
        # Mark processed
        await self.mark_processed(message_id)

Concurrent Processing with Backpressure

python
class ConcurrentConsumer:
    """Process messages concurrently with backpressure."""
    
    def __init__(
        self,
        consumer,
        handler: Callable,
        max_concurrent: int = 10
    ):
        self.consumer = consumer
        self.handler = handler
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.active_tasks = set()
    
    async def process_with_limit(self, message):
        """Process message within concurrency limit."""
        
        async with self.semaphore:
            try:
                await self.handler(message)
            except Exception as e:
                logger.error(f"Handler failed: {e}", exc_info=True)
                raise
    
    async def start(self):
        """Start consuming with concurrency control."""
        
        while True:
            # Poll for message
            msg = self.consumer.poll(timeout=1.0)
            
            if msg is None:
                # No message, wait for active tasks
                if self.active_tasks:
                    done, pending = await asyncio.wait(
                        self.active_tasks,
                        timeout=1.0,
                        return_when=asyncio.FIRST_COMPLETED
                    )
                    self.active_tasks = pending
                continue
            
            # Create task for message
            task = asyncio.create_task(self.process_with_limit(msg))
            self.active_tasks.add(task)
            
            # Commit offset
            self.consumer.commit(msg)

Check our async API guide for async patterns.

Backpressure Management

Handle consumer overload gracefully:

Adaptive Rate Limiting

python
class AdaptiveConsumer:
    """Adjust consumption rate based on processing speed."""
    
    def __init__(self, base_concurrency: int = 10):
        self.concurrency = base_concurrency
        self.success_count = 0
        self.failure_count = 0
        self.recent_durations = []
    
    def adjust_concurrency(self):
        """Adjust based on recent performance."""
        
        if not self.recent_durations:
            return
        
        avg_duration = sum(self.recent_durations) / len(self.recent_durations)
        
        # If processing fast, increase concurrency
        if avg_duration < 5.0 and self.success_count > 10:
            self.concurrency = min(50, int(self.concurrency * 1.2))
            logger.info(f"Increased concurrency to {self.concurrency}")
        
        # If processing slow or errors, decrease
        elif avg_duration > 30.0 or self.failure_count > 5:
            self.concurrency = max(1, int(self.concurrency * 0.8))
            logger.warning(f"Decreased concurrency to {self.concurrency}")
        
        # Reset counters
        self.success_count = 0
        self.failure_count = 0
        self.recent_durations.clear()

Error Handling & DLQ

Dead letter queues for failed messages:

python
class DLQHandler:
    """Handle failed messages with DLQ."""
    
    def __init__(self, dlq_producer, max_retries: int = 3):
        self.dlq_producer = dlq_producer
        self.max_retries = max_retries
    
    async def process_with_dlq(self, message: dict, handler: Callable):
        """Process with automatic DLQ on failure."""
        
        retry_count = message.get("_retry_count", 0)
        
        try:
            await handler(message)
            
        except Exception as e:
            logger.error(f"Processing failed (attempt {retry_count + 1}): {e}")
            
            if retry_count >= self.max_retries:
                # Send to DLQ
                await self.send_to_dlq(message, e)
            else:
                # Retry
                message["_retry_count"] = retry_count + 1
                message["_last_error"] = str(e)
                
                # Re-enqueue with delay
                await self.retry_with_backoff(message, retry_count)
    
    async def send_to_dlq(self, message: dict, error: Exception):
        """Send failed message to DLQ."""
        
        dlq_message = {
            "original_message": message,
            "error": str(error),
            "error_type": type(error).__name__,
            "failed_at": datetime.now().isoformat(),
            "retry_count": message.get("_retry_count", 0)
        }
        
        await self.dlq_producer.publish(
            topic="dlq",
            key=message.get("job_id"),
            value=dlq_message
        )
        
        logger.warning(f"Sent message to DLQ: {message.get('job_id')}")
    
    async def retry_with_backoff(self, message: dict, retry_count: int):
        """Retry with exponential backoff."""
        
        delay = min(300, 2 ** retry_count)  # Max 5 min
        
        await asyncio.sleep(delay)
        # Re-publish to queue

Monitoring & Observability

Essential metrics for event-driven AI:

python
from prometheus_client import Counter, Histogram, Gauge

# Producer metrics
events_published = Counter(
    'events_published_total',
    'Total events published',
    ['topic', 'event_type']
)

publish_duration = Histogram(
    'publish_duration_seconds',
    'Event publish duration',
    ['topic']
)

# Consumer metrics
events_processed = Counter(
    'events_processed_total',
    'Total events processed',
    ['topic', 'status']
)

processing_duration = Histogram(
    'processing_duration_seconds',
    'Event processing duration',
    ['topic', 'handler']
)

queue_lag = Gauge(
    'queue_lag_seconds',
    'Consumer lag in seconds',
    ['topic', 'consumer_group']
)

# Usage
with publish_duration.labels(topic="documents").time():
    await kafka_producer.publish(...)

events_published.labels(
    topic="documents",
    event_type="uploaded"
).inc()

Primary references: official documentation, official documentation, official documentation, official documentation.

Event-Driven AI Pipelines with Kafka & SQS Decision Table

DecisionPrefer the simpler path whenAdd operational complexity when
ArchitectureOne component can own the contract and stateIndependent scaling or fault isolation is required
RolloutOffline replay covers the meaningful casesLive behavior requires shadow traffic and a canary
RecoveryA failed operation is safe to repeatPartial effects require idempotency or reconciliation
MeasurementOne service objective represents user impactQuality, latency, and cost need separate gates

Operating Event-Driven AI Pipelines with Kafka & SQS as a System

The implementation is only one part of Event-Driven AI Pipelines with Kafka & SQS. A production design also needs an explicit contract for inputs, outputs, ownership, and failure behavior. Write that contract before selecting a library. It should identify which component validates input, where state lives, what may be retried, and which result is authoritative when two components disagree. This prevents a convenient prototype boundary from silently becoming the long-term architecture.

Start with a representative baseline. Capture request shape, traffic distribution, dependency latency, error classes, and the quality signal users actually care about. Averages hide the cases that cause incidents, so keep percentiles and segment measurements by workload type. Record the configuration and dataset version beside every result. Without that context, a faster or more accurate run cannot be reproduced and should not be used to approve a rollout.

Define the failure model

List failures by where they originate: invalid input, capacity exhaustion, dependency timeout, partial state change, malformed output, and semantically wrong output. Each class needs a different response. Validation errors should fail immediately. Transient dependency failures may be retried with a budget and jitter. An operation that may have committed must use an idempotency key or reconciliation step before retrying. A syntactically valid but incorrect result belongs in evaluation and review, not a blind retry loop.

Set a deadline for the complete operation and derive smaller budgets for each dependency. Local timeouts that add up to more than the caller's deadline merely create abandoned work. Propagate cancellation where the protocol supports it. Bound every queue, retry loop, context buffer, and concurrency pool; an unbounded safety mechanism becomes a second outage during overload.

Design a degraded mode before it is needed. Depending on the workload, that can mean returning a cached answer, selecting a simpler path, placing work in a durable queue, or asking for human review. The degraded response must be visible in telemetry and, where it changes meaning, visible to the caller. Silent fallback makes quality regressions almost impossible to diagnose.

Measure the decision, not just the component

Use three layers of signals. System metrics cover latency, throughput, saturation, and errors. Correctness metrics measure whether the result satisfies its contract. Business or user metrics show whether the system solved the intended problem. Improving only one layer can move the others backward, so release criteria should name acceptable movement for all three.

Attach a reason code to every route, rejection, fallback, and retry. Include version identifiers for configuration, code, model, schema, and data when relevant. Logs should let an engineer reconstruct a decision without storing secrets or raw personal data. Traces should cross process boundaries, while metrics should remain low-cardinality enough to operate reliably.

Alert on symptoms that require action, not every internal anomaly. A useful alert names the affected service objective, links to a runbook, and distinguishes a customer-visible incident from exhausted headroom. Dashboards serve a different purpose: they support diagnosis and capacity planning. Treating a dashboard as an alerting strategy leaves failures undiscovered until someone happens to look.

Roll out with reversible steps

Ship Event-Driven AI Pipelines with Kafka & SQS behind a versioned interface and a kill switch. Begin with offline replay using production-shaped, privacy-safe samples. Then use shadow execution when duplicate work has acceptable cost and side effects can be suppressed. A small canary should exercise the real dependency graph before traffic expands. Compare the canary with the baseline by cohort rather than mixing both populations into one aggregate.

Promotion gates should be written before the rollout. Include a minimum sample size or observation window, maximum regression in tail latency and error rate, and a correctness threshold. Roll back automatically when a hard safety boundary is crossed; use manual review for ambiguous quality movement. Preserve enough evidence from both paths to explain why the gate passed or failed.

Configuration deserves the same discipline as code. Review changes, validate them before activation, keep an immutable history, and make rollback a single operation. If a deployment changes code and configuration together, record both versions. Otherwise an incident responder may roll back the binary while leaving the triggering configuration active.

Capacity and cost controls

Model capacity in units the bottleneck understands: concurrent connections, tokens, queue jobs, database transactions, GPU memory, or bytes in flight. Convert the expected traffic distribution into those units and include burst behavior. Then load-test the first constrained dependency, not merely the public endpoint. A system that accepts more work than it can finish within its deadline is overloaded even if CPU utilization looks comfortable.

Cost is also a reliability limit. Add per-request attribution, tenant or workflow budgets, and a global circuit breaker for unexpectedly expensive paths. Review unit economics at the same granularity as performance; a cheap median can conceal a small class of requests responsible for most spend. Optimize only after measuring, because reducing context, replicas, validation, or redundancy can trade visible cost for less visible risk.

Production readiness review

Before launch, ask an engineer who did not build the feature to follow the runbook through one simulated failure. Verify backups or checkpoints by restoring them, not by checking that a job reported success. Exercise credential rotation, dependency unavailability, bad configuration, and rollback. Assign an owner for each alarm and a date for reviewing thresholds after real traffic arrives.

The final architecture document should be short enough to remain current. Keep the decision, rejected alternatives, invariants, dependency contracts, dashboards, and rollback procedure. Link detailed experiments rather than pasting them into the document. Teams that need help turning this review into an operable service can use our Event-Driven AI Pipelines with Kafka & SQS engineering support.

Frequently Asked Questions

Should I use Kafka or SQS for AI workloads?

Use SQS for simple task queues, fire-and-forget jobs, and AWS-native deployments. Use Kafka when you need: (1) Multiple consumers on same events, (2) Message replay for reprocessing, (3) Strict ordering guarantees, (4) Very high throughput (>10K msg/sec), or (5) Event sourcing patterns. Most AI systems start with SQS and migrate to Kafka when scaling beyond 100K events/day.

How do I handle messages larger than 256KB in SQS?

Use S3 as extended storage - upload large payload to S3, send S3 reference in SQS message. Consumer downloads from S3 when processing. AWS has official Extended Client Library that handles this automatically. For Kafka, increase max.message.bytes (default 1MB) or use similar S3 pattern for very large payloads.

What's the right visibility timeout for LLM processing?

5-10 minutes for typical LLM inference (allowing for retries). For batch embedding jobs, use 30-60 minutes. Set based on your 99th percentile processing time × 2. Too short = duplicate processing, too long = delayed retries on failure. Monitor actual processing times and adjust.

How do I prevent duplicate processing?

Implement idempotency using Redis/database to track processed message IDs. Generate deterministic message ID from content hash, check before processing, mark after completion. Keep processed IDs for 7-14 days (longer than message retention). This handles network duplicates, consumer restarts, and manual retries.

Should consumers commit offsets before or after processing?

After successful processing - this ensures at-least-once delivery. Committing before processing risks message loss on crash. Trade-off: potential duplicates on consumer crash (solved with idempotency). Never use auto-commit for AI workloads - processing time varies too much.

How many consumers should I run?

Start with 1 consumer per CPU core on your worker nodes. Scale horizontally by adding more worker nodes with consumers. Maximum useful consumers = number of partitions (Kafka) or unlimited (SQS). Monitor queue lag - if lag > 1 minute, add more consumers.

How do I handle rate limits from LLM providers?

Implement token bucket in consumer to enforce provider limits. Track API calls per minute, pause consumption when nearing limit, resume when quota refreshes. Alternative: multiple consumer groups with different provider API keys for higher total throughput. See our rate limiting guide.

What's the best pattern for multi-stage AI pipelines?

Chain topics - each stage publishes to next stage's topic. Example: doc.uploaded → doc.chunked → doc.embedded → doc.indexed. Each stage has dedicated consumers. Benefits: independent scaling per stage, replay individual stages, parallel processing where possible. Use correlation IDs to track documents across stages.

Conclusion

Event-driven architectures with Kafka and SQS enable scalable, reliable AI systems that handle long-running workloads without blocking API servers. The patterns in this guide power production systems processing 100,000+ AI events daily.

Key takeaways:

  • Decouple API from processing - return immediately, process async
  • Choose based on needs - SQS for simplicity, Kafka for scale
  • Idempotency is mandatory - AI processing is expensive to duplicate
  • Handle backpressure - adapt consumption rate to processing capacity
  • DLQ for failures - don't lose failed messages, debug and retry
  • Monitor everything - track lag, processing times, error rates

Start with SQS for simple job queues, migrate to Kafka when you need replay, multiple consumers, or >100K events/day. Add idempotency from day one, implement DLQ early, and scale horizontally by adding consumers.

Need help building event-driven AI systems? Our backend API engineering team specializes in scalable message-driven architectures for AI agents, RAG pipelines, and multi-stage AI workflows.

Related guides:

Free consultation

Book a free consultation call on event-driven AI architectures

30-minute call with the HinterBuild team. Discuss your project, architecture questions, or next steps — no obligation.

Book a meeting

Keep reading

Related articles

Backend Systems · 9 min read

Event-Driven Architecture: When to Use It and How to

Learn event-driven architecture through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post

AI Systems · 11 min read

Reranking in RAG Pipelines: Cross-Encoders, ColBERT, and

Learn reranking in rag pipelines through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post

AI Systems · 9 min read

When to Self-Host LLMs: Cost Analysis & Decision Framework

Learn when to self-host llms through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post

AI Systems · 12 min read

When Fine-Tuning Makes Things Worse

Learn when fine-tuning makes things worse through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post