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 (1)
Event-driven agents are a much healthier shape than polling loops, but idempotency is the part I would inspect first. Streams help with coordination; the worker contract prevents duplicated real-world actions.