Queue-Driven Agents: Feeding Workloads from Kafka into a LangGraph Fleet

What happens when you have 10,000 agent tasks per minute? A single LangGraph instance can’t handle it. You need a queue — and Kafka is the gold standard for high-throughput, fault-tolerant message processing. This guide shows how to connect Kafka to a fleet of LangGraph agents for scalable, reliable task processing.

Why Queue-Driven Architecture

The Problem with Direct Invocation

User Request → LangGraph App → Process → Response

• Single point of failure
• No backpressure handling
• Can’t scale horizontally easily
• Lost requests on crashes

The Queue-Driven Solution

User Request → Kafka Topic → Consumer Group → LangGraph Workers → Kafka Response Topic

• Horizontal scaling by adding consumers
• Guaranteed message delivery
• Natural backpressure via consumer lag
• Fault tolerance with replication

Architecture Overview

Producers (API, Webhooks, Batch Jobs)
    ↓
Kafka Topic: "agent.tasks"
    ↓
Consumer Group (N workers)
    ├── Worker 1: LangGraph Agent
    ├── Worker 2: LangGraph Agent
    └── Worker N: LangGraph Agent
    ↓
Kafka Topic: "agent.results"
    ↓
Results Consumers (API responses, webhooks, storage)

Setting Up Kafka

Docker Compose

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
  
  kafka:
    image: confluentinc/cp-kafka:7.4.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"

Create Topics

kafka-topics --create --topic agent.tasks \
  --bootstrap-server localhost:9092 \
  --partitions 6 \
  --replication-factor 1

kafka-topics --create --topic agent.results \
  --bootstrap-server localhost:9092 \
  --partitions 6 \
  --replication-factor 1

Task Producer

from confluent_kafka import Producer
import json
import uuid
from datetime import datetime

class AgentTaskProducer:
    def __init__(self, bootstrap_servers: str = "localhost:9092"):
        self.producer = Producer({
            'bootstrap.servers': bootstrap_servers,
            'acks': 'all',  # Wait for all replicas
            'enable.idempotence': True,  # Exactly-once semantics
        })
    
    def submit_task(
        self,
        task_type: str,
        payload: dict,
        priority: int = 0,
        callback_url: str = None
    ) -> str:
        task_id = str(uuid.uuid4())
        
        task = {
            "task_id": task_id,
            "task_type": task_type,
            "payload": payload,
            "priority": priority,
            "callback_url": callback_url,
            "created_at": datetime.now().isoformat(),
            "retry_count": 0,
            "max_retries": 3,
        }
        
        self.producer.produce(
            topic="agent.tasks",
            key=task_id,
            value=json.dumps(task),
            callback=self._delivery_callback
        )
        self.producer.flush()
        
        return task_id
    
    def _delivery_callback(self, err, msg):
        if err:
            logger.error(f"Task delivery failed: {err}")
        else:
            logger.info(f"Task delivered to {msg.topic()} [{msg.partition()}]")

# Usage
producer = AgentTaskProducer()
task_id = producer.submit_task(
    task_type="summarize_document",
    payload={"document_url": "https://example.com/doc.pdf"},
    priority=1
)

LangGraph Worker

from confluent_kafka import Consumer, KafkaError
import json
import signal

class AgentWorker:
    def __init__(
        self,
        worker_id: str,
        bootstrap_servers: str = "localhost:9092"
    ):
        self.worker_id = worker_id
        self.consumer = Consumer({
            'bootstrap.servers': bootstrap_servers,
            'group.id': f'agent-worker-{worker_id}',
            'auto.offset.reset': 'earliest',
            'enable.auto.commit': False,  # Manual commit for at-least-once
            'max.poll.interval.ms': 300000,  # 5 minutes for long tasks
        })
        
        self.producer = Producer({
            'bootstrap.servers': bootstrap_servers,
        })
        
        self.running = True
        self.agent_graph = self._build_agent_graph()
    
    def _build_agent_graph(self):
        """Build the LangGraph agent for this worker."""
        graph = StateGraph(AgentState)
        # ... add nodes and edges ...
        return graph.compile(checkpointer=checkpointer)
    
    def run(self):
        self.consumer.subscribe(["agent.tasks"])
        
        signal.signal(signal.SIGINT, self._shutdown)
        signal.signal(signal.SIGTERM, self._shutdown)
        
        while self.running:
            msg = self.consumer.poll(1.0)
            
            if msg is None:
                continue
            
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                logger.error(f"Consumer error: {msg.error()}")
                continue
            
            try:
                task = json.loads(msg.value().decode('utf-8'))
                self._process_task(task, msg)
            except Exception as e:
                logger.error(f"Task processing failed: {e}")
                self._handle_failure(task, str(e))
    
    def _process_task(self, task: dict, msg):
        logger.info(f"Processing task {task['task_id']}")
        
        config = {"configurable": {"thread_id": task["task_id"]}}
        
        # Run the agent
        result = self.agent_graph.invoke(
            {
                "task_type": task["task_type"],
                "payload": task["payload"],
                "status": "processing",
            },
            config
        )
        
        # Send result
        self._send_result(task, result)
        
        # Commit offset
        self.consumer.commit(msg)
    
    def _send_result(self, task: dict, result: dict):
        result_message = {
            "task_id": task["task_id"],
            "status": "completed",
            "result": result,
            "worker_id": self.worker_id,
            "completed_at": datetime.now().isoformat(),
        }
        
        self.producer.produce(
            topic="agent.results",
            key=task["task_id"],
            value=json.dumps(result_message)
        )
        self.producer.flush()
        
        # Callback if specified
        if task.get("callback_url"):
            self._send_callback(task["callback_url"], result_message)
    
    def _handle_failure(self, task: dict, error: str):
        task["retry_count"] += 1
        
        if task["retry_count"] <= task["max_retries"]:
            # Requeue with incremented retry count
            self.producer.produce(
                topic="agent.tasks",
                key=task["task_id"],
                value=json.dumps(task)
            )
        else:
            # Send failure result
            self._send_result(task, {
                "status": "failed",
                "error": error,
                "retries_exhausted": True
            })
    
    def _shutdown(self, signum, frame):
        logger.info("Shutting down worker...")
        self.running = False

Running the Fleet

Worker Pool Manager

import multiprocessing
import time

class WorkerFleet:
    def __init__(self, num_workers: int = 4):
        self.num_workers = num_workers
        self.workers = []
    
    def start(self):
        for i in range(self.num_workers):
            worker = AgentWorker(worker_id=str(i))
            process = multiprocessing.Process(target=worker.run)
            process.start()
            self.workers.append(process)
        
        logger.info(f"Started {self.num_workers} workers")
    
    def scale(self, target_workers: int):
        """Scale the fleet up or down."""
        current = len(self.workers)
        
        if target_workers > current:
            # Add workers
            for i in range(current, target_workers):
                worker = AgentWorker(worker_id=str(i))
                process = multiprocessing.Process(target=worker.run)
                process.start()
                self.workers.append(process)
        
        elif target_workers < current:
            # Remove workers
            for i in range(current - target_workers):
                worker = self.workers.pop()
                worker.terminate()
        
        logger.info(f"Scaled from {current} to {target_workers} workers")

# Start fleet
fleet = WorkerFleet(num_workers=4)
fleet.start()

# Auto-scale based on consumer lag
def auto_scale(fleet, lag_threshold=1000):
    while True:
        lag = get_consumer_lag("agent.tasks", "agent-worker")
        
        if lag > lag_threshold * 2:
            fleet.scale(min(fleet.num_workers * 2, 16))
        elif lag < lag_threshold / 2:
            fleet.scale(max(fleet.num_workers // 2, 1))
        
        time.sleep(60)

Monitoring

from prometheus_client import Counter, Gauge, start_http_server

tasks_processed = Counter("tasks_processed_total", "Total tasks processed", ["status"])
consumer_lag = Gauge("consumer_lag", "Consumer lag per partition", ["partition"])
worker_count = Gauge("worker_count", "Number of active workers")

start_http_server(8000)

# In worker loop
tasks_processed.labels(status="completed").inc()
tasks_processed.labels(status="failed").inc()

Best Practices

• Use manual commits — Commit after successful processing, not before
• Set max.poll.interval.ms — Account for long-running agent tasks
• Implement dead letter queues — For tasks that fail repeatedly
• Monitor consumer lag — Scale workers before lag grows too large
• Use idempotent producers — Prevent duplicate task submission

Conclusion

Queue-driven agent architectures with Kafka and LangGraph give you horizontal scalability, fault tolerance, and natural backpressure. The pattern is straightforward: produce tasks to Kafka, consume with a fleet of LangGraph workers, and produce results back. Start with manual commits and basic monitoring, add auto-scaling as load grows, and always implement dead letter queues for failed tasks.