Building a Real-Time AI Processing Pipeline with Claude API and Apache Kafka

pip install kafka-python anthropic asyncio

# consumer_base.py
import asyncio
import json
import logging
from dataclasses import dataclass
from typing import AsyncGenerator
from kafka import KafkaConsumer
from kafka.errors import KafkaError
 
logger = logging.getLogger(__name__)
 
@dataclass
class ReviewEvent:
    event_id: str
    product_id: str
    user_id: str
    review_text: str
    rating: int
    timestamp: str
    source: str  # "web" | "mobile" | "api"
 
class ReviewEventConsumer:
    
    def __init__(
        self,
        bootstrap_servers: list[str],
        topic: str,
        group_id: str,
        max_poll_records: int = 50,
    ):
        self.consumer = KafkaConsumer(
            topic,
            bootstrap_servers=bootstrap_servers,
            group_id=group_id,
            auto_offset_reset='latest',
            enable_auto_commit=False,  # ← Critical: manual commit only
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
            max_poll_records=max_poll_records,
            # Generous timeout: Claude API calls can take several seconds
            max_poll_interval_ms=300000,
        )
        self.topic = topic
        self.running = False
    
    async def consume(self) -> AsyncGenerator[list[ReviewEvent], None]:
        """Yield event batches as an async generator"""
        self.running = True
        
        try:
            while self.running:
                loop = asyncio.get_event_loop()
                # poll() is blocking — run in executor to keep event loop free
                raw_messages = await loop.run_in_executor(
                    None,
                    lambda: self.consumer.poll(timeout_ms=1000, max_records=50)
                )
                
                if not raw_messages:
                    await asyncio.sleep(0.1)
                    continue
                
                events = []
                for tp, messages in raw_messages.items():
                    for msg in messages:
                        try:
                            event = ReviewEvent(**msg.value)
                            events.append(event)
                        except (KeyError, TypeError) as e:
                            logger.error(f"Malformed event at offset {msg.offset}: {e}")
                            # Malformed events go to DLQ, not left in Kafka
                
                if events:
                    yield events
                
                # Commit AFTER the caller has finished processing
                await loop.run_in_executor(None, self.consumer.commit)
                    
        except KafkaError as e:
            logger.error(f"Kafka consumer error: {e}")
            raise
        finally:
            self.consumer.close()
    
    def stop(self):
        self.running = False

# smart_buffer.py
import asyncio
import time
from dataclasses import dataclass
from typing import Callable
 
@dataclass
class BufferConfig:
    max_size: int = 20        # Flush when N events accumulate
    max_wait_sec: float = 3.0 # Flush after N seconds regardless of count
    max_tokens: int = 15000   # Flush when estimated token count reaches this
 
class SmartBuffer:
    """
    Three-trigger buffer: flush on count, time, or token limit.
    Each trigger serves a different traffic condition.
    """
    
    def __init__(self, config: BufferConfig, flush_callback: Callable):
        self.config = config
        self.flush_callback = flush_callback
        self._buffer: list = []
        self._estimated_tokens: int = 0
        self._last_flush: float = time.time()
        self._lock = asyncio.Lock()
    
    def _estimate_tokens(self, text: str) -> int:
        # ~1.3 tokens/char for English text, plus per-item overhead
        return int(len(text) * 1.3) + 50
    
    async def add(self, event) -> None:
        async with self._lock:
            token_estimate = self._estimate_tokens(
                getattr(event, 'review_text', str(event))
            )
            self._buffer.append(event)
            self._estimated_tokens += token_estimate
            
            should_flush = (
                len(self._buffer) >= self.config.max_size
                or self._estimated_tokens >= self.config.max_tokens
                or (time.time() - self._last_flush) >= self.config.max_wait_sec
            )
            
            if should_flush:
                await self._flush()
    
    async def _flush(self) -> None:
        if not self._buffer:
            return
        
        batch = self._buffer.copy()
        self._buffer.clear()
        self._estimated_tokens = 0
        self._last_flush = time.time()
        
        # Background task: don't block the buffer while processing
        asyncio.create_task(self.flush_callback(batch))
    
    async def force_flush(self) -> None:
        """Called during graceful shutdown — process whatever remains"""
        async with self._lock:
            await self._flush()

# model_router.py
from enum import Enum
import anthropic
import json
import asyncio
import logging
 
logger = logging.getLogger(__name__)
 
class ModelTier(Enum):
    FAST = "claude-haiku-4-5-20251001"  # $0.08 input / $0.40 output per MTok
    BALANCED = "claude-sonnet-4-6"       # $3.00 input / $15.00 output per MTok
    DEEP = "claude-opus-4-6"             # $15.00 input / $75.00 output per MTok
 
def select_model(events: list) -> ModelTier:
    """
    Route to the cheapest model that can handle the batch reliably.
    
    The routing logic should be calibrated to your specific data.
    Run a sample of your events through all three models, compare accuracy,
    and adjust the thresholds until the cost/accuracy tradeoff meets your SLA.
    """
    avg_length = sum(len(e.review_text) for e in events) / len(events)
    avg_rating = sum(e.rating for e in events) / len(events)
    
    # Short, unambiguous signal → Haiku handles confidently
    if avg_length < 50 and (avg_rating <= 1.5 or avg_rating >= 4.5):
        return ModelTier.FAST
    
    # Long text or ambiguous rating requires careful reasoning
    if avg_length > 200 or (1.8 <= avg_rating <= 3.2):
        return ModelTier.DEEP
    
    return ModelTier.BALANCED
 
class ReviewModerator:
    
    def __init__(self):
        self.client = anthropic.Anthropic()
    
    async def analyze_batch(self, events: list) -> list[dict]:
        """
        Analyze a batch of reviews in a single Claude API call.
        Returns a list of analysis results, one per input event.
        """
        model_tier = select_model(events)
        
        reviews_input = [
            {"id": e.event_id, "text": e.review_text, "rating": e.rating}
            for e in events
        ]
        
        system_prompt = """You are an e-commerce content moderation AI.
 
For each review in the array, determine:
- is_spam: Unsolicited promotion, meaningless filler, or bot-generated content (true/false)
- is_abusive: Personal attacks, hate speech, harassment, or threats (true/false)
- is_fake: Unnatural rating patterns or suspiciously template-like language suggesting manipulation (true/false)
- sentiment: Overall emotional polarity ("positive" | "neutral" | "negative")
- confidence: Your confidence in the judgment, from 0.0 to 1.0
- reason: If any flag is true, a brief explanation under 60 characters. Null if no flags.
 
Return ONLY a JSON object in this exact format, no other text:
{"results": [{"id": "...", "is_spam": false, "is_abusive": false, "is_fake": false, "sentiment": "positive", "confidence": 0.95, "reason": null}]}"""
        
        try:
            response = self.client.messages.create(
                model=model_tier.value,
                max_tokens=2048,
                system=system_prompt,
                messages=[{
                    "role": "user",
                    "content": f"Moderate these reviews:\n\n{json.dumps(reviews_input, indent=2)}"
                }]
            )
            
            result_text = response.content[0].text.strip()
            result = json.loads(result_text)
            
            # Log cost for monitoring dashboard
            cost = self._calculate_cost(
                model_tier,
                response.usage.input_tokens,
                response.usage.output_tokens
            )
            logger.info(
                f"Batch completed: n={len(events)}, model={model_tier.name}, "
                f"in={response.usage.input_tokens}, out={response.usage.output_tokens}, "
                f"cost=${cost:.5f}"
            )
            
            return result.get("results", [])
            
        except json.JSONDecodeError as e:
            logger.error(f"Failed to parse Claude response: {e\!r}")
            logger.debug(f"Raw response: {response.content[0].text[:500]}")
            # Caller will route to DLQ
            raise
        except anthropic.RateLimitError:
            logger.warning("Rate limited — backing off 60s and retrying")
            await asyncio.sleep(60)
            return await self.analyze_batch(events)
        except anthropic.APITimeoutError:
            logger.error("API timeout on batch")
            raise
    
    def _calculate_cost(self, tier: ModelTier, input_t: int, output_t: int) -> float:
        costs = {
            ModelTier.FAST: (0.00008, 0.0004),
            ModelTier.BALANCED: (0.003, 0.015),
            ModelTier.DEEP: (0.015, 0.075),
        }
        in_rate, out_rate = costs[tier]
        return (input_t * in_rate + output_t * out_rate) / 1_000_000

# dlq_handler.py
from kafka import KafkaProducer
import json
import time
import asyncio
import logging
 
logger = logging.getLogger(__name__)
 
class DLQHandler:
    
    def __init__(self, bootstrap_servers: list[str]):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            acks='all',  # Wait for all in-sync replicas — no data loss on broker failure
            retries=3,
        )
    
    async def send(
        self,
        events: list,
        error_type: str,
        error_message: str,
        retry_count: int = 0
    ) -> None:
        """Route failed events to the DLQ with retry metadata."""
        
        for event in events:
            dlq_payload = {
                "original_event": {
                    "event_id": event.event_id,
                    "product_id": event.product_id,
                    "review_text": event.review_text,
                    "rating": event.rating,
                    "source": event.source,
                },
                "failure_info": {
                    "error_type": error_type,
                    "error_message": error_message[:500],  # Truncate long stack traces
                    "retry_count": retry_count,
                    "failed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                },
                "retry_policy": {
                    "max_retries": 3,
                    "exhausted": retry_count >= 3,
                    "next_retry_at": self._next_retry_time(retry_count),
                }
            }
            
            loop = asyncio.get_event_loop()
            await loop.run_in_executor(
                None,
                lambda p=dlq_payload: self.producer.send("review-dlq", value=p)
            )
        
        logger.warning(
            f"Sent {len(events)} events to DLQ "
            f"(type={error_type}, retry={retry_count})"
        )
    
    def _next_retry_time(self, retry_count: int) -> str:
        """Exponential backoff: 1 min → 5 min → 30 min → give up"""
        if retry_count >= 3:
            return "exhausted"
        backoffs = [60, 300, 1800]
        delay = backoffs[retry_count]
        return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() + delay))

# pipeline.py
import asyncio
import logging
import signal
import os
from consumer_base import ReviewEventConsumer
from smart_buffer import SmartBuffer, BufferConfig
from model_router import ReviewModerator
from dlq_handler import DLQHandler
 
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(name)s: %(message)s'
)
logger = logging.getLogger(__name__)
 
KAFKA_SERVERS = os.environ.get(
    "KAFKA_BOOTSTRAP_SERVERS", "localhost:9092"
).split(",")
 
class ReviewModerationPipeline:
    
    def __init__(self):
        self.consumer = ReviewEventConsumer(
            bootstrap_servers=KAFKA_SERVERS,
            topic="review-events",
            group_id="review-moderator",
        )
        self.moderator = ReviewModerator()
        self.dlq = DLQHandler(bootstrap_servers=KAFKA_SERVERS)
        self.buffer = SmartBuffer(
            config=BufferConfig(max_size=20, max_wait_sec=3.0, max_tokens=15000),
            flush_callback=self._process_batch,
        )
        self._stats = {"processed": 0, "errors": 0, "dlq_sent": 0}
    
    async def _process_batch(self, events: list) -> None:
        """Called by SmartBuffer when a batch is ready."""
        try:
            results = await self.moderator.analyze_batch(events)
            
            flagged = [
                r for r in results
                if r.get("is_spam") or r.get("is_abusive") or r.get("is_fake")
            ]
            
            if flagged:
                logger.warning(f"Flagged {len(flagged)}/{len(events)} in batch")
                # → Write to moderation queue, send alert webhook
                # (implementation depends on your stack)
            
            self._stats["processed"] += len(events)
            
        except json.JSONDecodeError as e:
            logger.error(f"Model response parse error: {e}")
            self._stats["errors"] += len(events)
            await self.dlq.send(events, "parse_error", str(e))
            self._stats["dlq_sent"] += len(events)
            
        except Exception as e:
            logger.error(f"Unexpected error in batch: {e}")
            self._stats["errors"] += len(events)
            await self.dlq.send(events, "processing_error", str(e))
            self._stats["dlq_sent"] += len(events)
    
    async def run(self) -> None:
        logger.info("🚀 Review Moderation Pipeline starting")
        logger.info(f"Kafka: {KAFKA_SERVERS}, Buffer: {self.buffer.config}")
        
        loop = asyncio.get_event_loop()
        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(
                sig, lambda: asyncio.create_task(self.shutdown())
            )
        
        try:
            async for events in self.consumer.consume():
                for event in events:
                    await self.buffer.add(event)
        except asyncio.CancelledError:
            logger.info("Pipeline run cancelled")
    
    async def shutdown(self) -> None:
        """Graceful shutdown: drain the buffer before exiting."""
        logger.info("Shutdown signal received — draining buffer...")
        self.consumer.stop()
        await self.buffer.force_flush()
        # Give background tasks a moment to complete
        await asyncio.sleep(2.0)
        logger.info(f"Shutdown complete. Final stats: {self._stats}")
 
if __name__ == "__main__":
    asyncio.run(ReviewModerationPipeline().run())

{
  "period": "2026-04-18T05:00:00Z",
  "duration_hours": 1,
  "total_events": 8420,
  "model_breakdown": {
    "haiku": {
      "events": 5100,
      "pct": "60.6%",
      "cost_usd": 0.024
    },
    "sonnet": {
      "events": 2800,
      "pct": "33.3%",
      "cost_usd": 0.41
    },
    "opus": {
      "events": 520,
      "pct": "6.2%",
      "cost_usd": 0.39
    }
  },
  "total_cost_usd": 0.824,
  "cost_per_1000_events": 0.098,
  "dlq_events": 3,
  "avg_batch_size": 16.8,
  "avg_latency_ms": 1240
}

# Start Kafka in Docker (single-node for development)
docker run -d -p 9092:9092 \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  apache/kafka:latest
 
# Create the topics
docker exec <container-id> kafka-topics.sh \
  --create --topic review-events --partitions 6 \
  --bootstrap-server localhost:9092