Event-Driven AI Agents with Kafka

Synchronous HTTP requests work well for fetching user profiles or submitting simple forms. They break down quickly when applied to AI agents.

An agentic workflow is not a simple request-response transaction. It involves multiple steps: reasoning loops, calling external APIs, searching databases, and generating final answers. This entire cycle can take anywhere from tens of seconds to several minutes. If you try to run this over a synchronous HTTP/REST connection, you will encounter client timeouts, blocked gateway threads, and resource starvation.

To build production-scale agent systems, you need to decouple the request from the execution. Transitioning to an event-driven architecture using Apache Kafka provides the asynchronous messaging fabric required to run agents reliably at scale.

Why HTTP Fails for AI Agents

Traditional web architectures assume that downstream operations complete in milliseconds. With LLMs and agentic loops, this assumption is false for several reasons:

• Inference Latency: Generating a complex response using frontier models can take over 30 seconds.
• Multi-Step Loops: A ReAct agent might invoke three different APIs sequentially, parsing and reflecting on the results between each call. The total runtime scales linearly with the number of steps.
• Network Instability: Long-lived HTTP connections are prone to drops, especially over mobile networks. If a connection drops mid-loop, the agent’s context is lost unless you have implemented complex state recovery.
• Resource Exhaustion: Keeping HTTP connections open for minutes consumes server sockets and worker processes, drastically limiting the concurrent throughput of your web servers.

Moving to Kafka resolves these bottlenecks. The web API receives the user request, writes an event to a Kafka topic, and immediately returns a 202 Accepted status with a job ID. The agent processes the task asynchronously, publishing progress updates and the final result back to other topics.

The Event-Driven Agent Architecture

An event-driven agent system separates the user-facing web services from the background agent workers. This decoupling allows you to scale and update each layer independently.

The workflow relies on four key Kafka topics:

1. agent-requests: Incoming user queries and instructions.
2. agent-status-updates: Intermediate agent outputs (e.g., status logs, tool execution steps, intermediate thoughts).
3. agent-responses: Final synthesized answers sent back to the user.
4. agent-dlq (Dead Letter Queue): Messages that caused unrecoverable processing failures.

graph TD
    Client[Client Browser/App] -->|1. HTTP Post Request| Gateway[API Gateway]
    Gateway -->|2. Accept Job| Client
    Gateway -->|3. Produce Request Event| KafkaRequests[Topic: agent-requests]
    
    subgraph Agent Cluster
        AgentWorker1[Agent Worker Instance 1]
        AgentWorker2[Agent Worker Instance 2]
    end

    KafkaRequests -->|4. Consume Request| AgentWorker1
    AgentWorker1 -->|5. Run Tool Calls & LLM Loop| LLM[LLM API / Vector DB]
    AgentWorker1 -->|6. Produce Status Update| KafkaStatus[Topic: agent-status-updates]
    AgentWorker1 -->|7. Produce Final Response| KafkaResponses[Topic: agent-responses]
    AgentWorker1 -->|Error: Produce Failed Job| KafkaDLQ[Topic: agent-dlq]
    
    KafkaResponses -->|8. Consume Response| Gateway
    Gateway -->|9. Push via WebSocket/SSE| Client

In this architecture, the client opens a WebSocket or Server-Sent Events (SSE) connection to the API gateway. The gateway subscribes to the agent-status-updates and agent-responses topics, forwarding those events to the client in real-time.

Implementing an Asynchronous Agent Consumer in Python

The worker application consumes events from the agent-requests topic, runs the agent logic, and publishes status updates and results.

The Python implementation below uses the confluent-kafka library. It demonstrates how to initialize the consumer, handle the event loop, execute agent work, and produce output events.

import json
import logging
from confluent_kafka import Consumer, Producer, KafkaError, KafkaException

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agent-worker")

# Kafka configuration
KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
CONSUMER_CONF = {
    "bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS,
    "group.id": "agent-worker-group",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False, # Manual offset commit for delivery guarantees
}
PRODUCER_CONF = {
    "bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS,
}

producer = Producer(PRODUCER_CONF)

def delivery_report(err, msg):
    """Callback to verify if message was written to Kafka successfully."""
    if err is not None:
        logger.error(f"Message delivery failed: {err}")
    else:
        logger.debug(f"Message delivered to {msg.topic()} [{msg.partition()}]")

def publish_event(topic, payload):
    """Utility to publish events to a specific topic."""
    try:
        producer.produce(
            topic=topic,
            key=payload.get("job_id"),
            value=json.dumps(payload).encode("utf-8"),
            callback=delivery_report
        )
        producer.poll(0) # Trigger delivery callbacks
    except Exception as e:
        logger.error(f"Failed to publish event to {topic}: {e}")

def run_agent_workflow(job_id, prompt):
    """
    Mock agent execution workflow.
    In a real app, this integrates with LangGraph, PydanticAI, or custom loops.
    """
    publish_event("agent-status-updates", {
        "job_id": job_id,
        "status": "processing",
        "message": "Analyzing prompt and routing to tools."
    })
    
    # Step 1: Simulated Tool Call
    publish_event("agent-status-updates", {
        "job_id": job_id,
        "status": "running_tool",
        "tool_name": "database_search",
        "message": "Querying customer records."
    })
    
    # Step 2: Simulated LLM generation
    publish_event("agent-status-updates", {
        "job_id": job_id,
        "status": "generating_response",
        "message": "Synthesizing final response."
    })
    
    # Final Result
    return {
        "job_id": job_id,
        "status": "completed",
        "response": f"Processed prompt successfully: '{prompt}'."
    }

def main():
    consumer = Consumer(CONSUMER_CONF)
    consumer.subscribe(["agent-requests"])
    logger.info("Agent worker started. Listening for requests...")

    try:
        while True:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                else:
                    raise KafkaException(msg.error())

            # Process valid message
            try:
                payload = json.loads(msg.value().decode("utf-8"))
                job_id = payload.get("job_id")
                prompt = payload.get("prompt")
                
                logger.info(f"Processing job {job_id}")
                
                # Execute agent logic
                result = run_agent_workflow(job_id, prompt)
                
                # Publish final outcome
                publish_event("agent-responses", result)
                
                # Commit offset only after successful processing
                consumer.commit(msg, asynchronous=False)
                
            except Exception as e:
                logger.error(f"Error processing message: {e}")
                # Route failed messages to the Dead Letter Queue
                publish_event("agent-dlq", {
                    "raw_message": msg.value().decode("utf-8"),
                    "error": str(e)
                })
                consumer.commit(msg, asynchronous=False)

    except KeyboardInterrupt:
        logger.info("Shutting down worker...")
    finally:
        consumer.close()
        producer.flush()

if __name__ == "__main__":
    main()

This worker pattern ensures that even if the worker process crashes mid-execution, the uncommitted Kafka offset guarantees the message is reprocessed when a consumer restarts.

Concurrency, Ordering, and Backpressure

Managing scale in an event-driven system requires understanding how Kafka handles partitions and consumer groups.

Horizontal Scaling via Consumer Groups

To scale agent throughput, run multiple instances of the worker service within the same consumer group (group.id: agent-worker-group). Kafka automatically distributes partitions among active instances. If you have 8 partitions for agent-requests, you can scale up to 8 parallel worker containers to process messages concurrently.

Preserving Message Ordering

When designing agents, keeping messages for a specific session ordered is often critical. If a user sends three messages in quick succession, the agent must process message 1 before message 2.

Kafka guarantees ordering within a partition. To maintain sequence:

• Set the Kafka message key to the session_id or user_id.
• Kafka hashes the key to route all events for that session to the same partition.
• A single worker thread will consume the partition sequentially, ensuring messages are processed in order.

Handling Backpressure

Frontier LLM APIs limit your rate of token usage (TPM) and requests per minute (RPM). Under heavy loads, workers will receive RateLimitError exceptions from LLM providers.

If you were using HTTP directly, this would cause client requests to drop. With Kafka, the consumer acts as a buffer. You can throttle the consumer rate by adding delay, pausing the consumer using consumer.pause(), or scaling down worker concurrency. The events remain safe in the Kafka queue until your rate limits reset.

Resiliency Patterns: Retries and Dead Letter Queues

In production, agents fail. The failure might be a network glitch, a model failing to return valid JSON, or an external API going offline.

Transient Failures (Retries)

For temporary issues like network timeouts or API rate limits, use an exponential backoff retry pattern. Catch the exception within the worker loop, pause consumption of that partition, and retry after a delay. Do not send these messages directly to the Dead Letter Queue (DLQ).

Poison Pill Failures (DLQ)

If a message is malformed, contains corrupted data, or triggers a persistent parsing error that code cannot resolve, it is considered a poison pill. If you attempt to process it repeatedly, the worker will crash or hang indefinitely, blocking subsequent messages in the partition.

Handle poison pills by wrapping your processing block in a general try-catch structure, logging the trace, and publishing the failed payload along with its error context to the agent-dlq topic. Once published, commit the message offset to allow the consumer to proceed to the next event. Setting up alert rules on the DLQ topic ensures your team is notified of processing errors immediately.