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

How to architect low-latency, reactive AI workflows that eliminate rate limits, slash compute overhead, and guarantee reliable execution under load 1. THE PROBLEM IN PRODUCTION Most AI agents running in production today start as naive polling loops. A background worker periodically queries a database or third-party API: # 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) This pattern collapses under real production traffic across four distinct failure modes: - Tail Latency Bottlenecks: An urgent event arriving 100 milliseconds after a poll cycle sleeps for 29.9 seconds before being picked up. For time-sensitive workflows (such as security remediation, real-time customer routing, or automated trading), this introduces unacceptable lag. - Cascading API Rate Limits: As you scale from 10 to 1,000 agents checking independent downstream tools (e.g., Salesforce, GitHub, Slack), your infrastructure makes tens of thousands of empty GET requests every minute. The upstream systems throttle or ban your IP addresses before an actual workload is even executed. - Wasted Compute and Memory Pressure: Running thousands of blocked Python threads or event loops holding database connection pools in idle memory degrades node performance and drives up infrastructure costs. - Race Conditions and Split-Brain Execution: When horizontally scaling polling workers, two instances often grab the same record simultaneously unless complex distributed locking ( SELECT FOR UPDATE SKIP LOCKED ) is maintained. This leads to duplicate LLM calls, double payments, or corrupted state. 2. SYSTEM ARCHITECTURE An event-driven agent architecture decouples event ingestion from agent cognition. Instead of agents asking the world if work is available, the environment notifies agents through an enriched event payload. +------------------+ +------------------+ +------------------+ | 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) | +---------------------------------------------+ Core Components of the Flow - Event Producers: Ingest points (FastAPI webhooks, Change Data Capture pipelines, internal system events) publish a strictly typed schema containing the change delta and necessary contextual state. - Streaming Ingestion & Consumer Groups: Redis Streams maintain an append-only log with persistent consumer groups. If a worker process crashes mid-reasoning, the message remains unacknowledged ( XACK ) and is reassigned to a healthy worker via dead-letter / pending mechanisms. - Deterministic Idempotency Gate: Because network transports guarantee at-least-once delivery, every event must pass an atomic lock gate using a deterministic event hash before the agent initializes its context window. - Agent Execution Worker: The worker parses the pre-populated event payload, runs its reasoning chain (LangGraph, raw tool calling, or custom state machines), updates the persistent store, and acknowledges message processing. 3. CODE IMPLEMENTATION Below is a complete, runnable, production-ready Python implementation. It uses redis-py with Redis Streams and consumer groups to manage stateful, idempotent, event-driven agent invocations. 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( Top comments (0)