Building Event-Driven AI Agents: Replacing Polling Loops with Redis Streams and Idempotent Workers

# The naive anti-pattern running in hundreds of production services
while True:
    records = db.query("SELECT * FROM invoices WHERE status = 'pending_review'")
    for record in records:
        agent.process(record)
    time.sleep(30)

+------------------+      +------------------+      +------------------+
|  Event Producer  |      |  Event Producer  |      |  Event Producer  |
| (API / Webhooks) |      | (CDC / Postgres) |      | (IoT / Sensors)  |
+--------+---------+      +--------+---------+      +--------+---------+
         |                         |                         |
         +-------------------------+-------------------------+
                                   |
                                   v
             +---------------------------------------------+
             |    Message Broker (Redis Streams / Kafka)   |
             |       Stream: `agent:events:v1`             |
             +---------------------+-----------------------+
                                   |
                    +--------------+--------------+
                    |                             |
                    v                             v
       +-------------------------+   +-------------------------+
       | Consumer Group: Worker 1|   | Consumer Group: Worker 2|
       +------------+------------+   +------------+------------+
                    |                             |
                    v                             v
       +-------------------------+   +-------------------------+
       |   Idempotency Check     |   |   Idempotency Check     |
       |   (Redis SET key EX NX) |   |   (Redis SET key EX NX) |
       +------------+------------+   +------------+------------+
                    |                             |
                    v                             v
       +-------------------------+   +-------------------------+
       | Agent Execution Engine  |   | Agent Execution Engine  |
       | (LangGraph / DSPy / LLM)|   | (LangGraph / DSPy / LLM)|
       +------------+------------+   +------------+------------+
                    |                             |
                    +--------------+--------------+
                                   |
                                   v
             +---------------------------------------------+
             | State Checkpoints / Write-Back (Postgres)   |
             +---------------------------------------------+

python
import asyncio
import hashlib
import json
import logging
import os
import sys
from typing import Any, Callable, Dict, Optional
from pydantic import BaseModel, Field
import redis.asyncio as redis

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] (%(name)s) %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger("EventDrivenAgent")

# ---------------------------------------------------------------------------
# 1. SCHEMAS
# ---------------------------------------------------------------------------
class AgentEvent(BaseModel):
    event_id: str
    event_type: str
    source: str
    payload: Dict[str, Any]
    timestamp: int

    def generate_idempotency_key(self) -> str:
        """Generates a deterministic hash based on event payload and type."""
        raw_signature = f"{self.event_type}:{self.event_id}:{json.dumps(self.payload, sort_keys=True)}"
        return f"idemp:{hashlib.sha256(raw_signature.encode()).hexdigest()}"


# ---------------------------------------------------------------------------
# 2. CORE EVENT CONSUMER & AGENT RUNNER
# ---------------------------------------------------------------------------
class EventDrivenAgentWorker:
    def __init__(
        self,
        redis_url: str,
        stream_name: str,
        group_name: str,
        consumer_name: str,
        lock_ttl_seconds: int = 300,
    ):
        self.redis_url = redis_url
        self.stream_name = stream_name
        self.group_name = group_name
        self.consumer_name = consumer_name
        self.lock_ttl = lock_ttl_seconds
        self.redis_client: Optional[redis.Redis] = None
        self._running = False

    async def connect(self):
        """Initializes the Redis connection and sets up consumer groups."""
        self.redis_client = redis.from_url(self.redis_url, decode_responses=True)
        try:
            # Create the stream consumer group if it doesn't already exist
            await self.redis_client.xgroup_create(
                name=self.stream_name,
                groupname=self.group_name,
                id="0",
                mkstream=True,
            )
            logger.info(f"Consumer group '{self.group_name}' initialized on stream '{self.stream_name}'.")
        except redis.exceptions.ResponseError as e:
            if "BUSYGROUP" in str(e):
                logger.info(f"Consumer group '{self.group_name}' already exists.")
            else:
                raise e

    async def acquire_idempotency_lock(self, key: str) -> bool:
        """
        Uses Redis SET key NX to prevent duplicate execution of the same event.
        Returns True if lock was acquired, False if the event was already processed.
        """
        is_new = await self.redis_client.set(key, "PROCESSING", ex=self.lock_ttl, nx=True)
        return bool(is_new)

    async def process_stream(self, agent_handler: Callable[[AgentEvent], Any]):
        """Continuous event consumption loop."""
        self._running = True
        logger.info(f"Worker {self.consumer_name} listening for events...")

        while self._running:
            try:
                # Read new messages assigned to this consumer group
                # Block for 2000ms if no messages exist
                response = await self.redis_client.xreadgroup(
                    groupname=self.group_name,
                    consumername=self.consumer_name,
                    streams={self.stream_name: ">"},
                    count=10,
                    block=2000,
                )

                if not response:
                    await asyncio.sleep(0.01)
                    continue

                for stream, messages in response:
                    for message_id, raw_data in messages:
                        await self._handle_single_message(message_id, raw_data, agent_handler)

            except asyncio.CancelledError:
                logger.info("Worker shutdown initiated.")
                self._running = False
            except Exception as exc:
                logger.error(f"Unexpected error in consumer loop: {str(exc)}", exc_info=True)
                await asyncio.sleep(1)

    async def _handle_single_message(
        self,
        message_id: str,
        raw_data: Dict[str, str],
        agent_handler: Callable[[AgentEvent], Any],
    ):
        try:
            # Reconstruct domain event from message payload
            event_obj = AgentEvent(
                event_id=raw_data["event_id"],
                event_type=raw_data["event_type"],
                source=raw_data["source"],
                payload=json.loads(raw_data["payload"]),
                timestamp=int(raw_data["timestamp"]),
            )
        except Exception as err:
            logger.error(f"Failed to parse event {message_id}: {err}. Poison pill discarded.")
            # Acknowledge unparseable messages to prevent infinite poison pill loops
            await self.redis_client.xack(self.stream_name, self.group_name, message_id)
            return

        # Check idempotency
        idempotency_key = event_obj.generate_idempotency_key()
        lock_acquired = await self.acquire_idempotency_lock(idempotency_key)

        if not lock_acquired:
            logger.warning(f"Duplicate event detected ({idempotency_key}). Skipping execution.")
            await self.redis_client.xack(self.stream_name, self.group_name, message_id)
            return

        try:
            logger.info(