Implement event backbone with Redis Streams for AI workloads

Introduction

When your AI inference, evaluation, and audit services call one another directly, a single slow consumer stalls the whole pipeline and one crash drops events that were never persisted anywhere replayable. Teams that bolt on retries, dead-letter handling, and replay capability after the fact end up reinventing a fragile, half-finished message bus — and lose the ability to debug a misbehaving model decision later, because the events that triggered it are gone. By the end of this lesson you will be able to stand up a Redis Streams event backbone with producer and consumer-group abstractions, publish GenAI domain events from your inference service, and consume them in parallel across independently scaling worker groups.

Key Terminology

• XADD: Redis Streams command that appends a new entry to a stream, assigning it a monotonically increasing timestamp-based ID and optionally trimming the stream to a maximum length.
• XREADGROUP: Redis Streams command used by consumer groups to read events; with the special > ID it delivers only entries never seen by any consumer in the group, enabling distributed work-queue semantics.
• Consumer group: A named cursor over a stream shared by multiple consumers, where Redis distributes entries so each event is delivered to exactly one consumer in the group and tracks per-group acknowledgment state independently.
• Pending Entry List (PEL): The per-consumer-group registry of entries that have been delivered via XREADGROUP but not yet acknowledged with XACK; used to recover unprocessed events after a consumer crashes.

Concepts

Key Design Considerations for Production

• Consumer naming strategy: Use a deterministic naming scheme like {service}-{pod-id} so that pending entries from a crashed consumer can be identified and reclaimed. Random consumer names make orphaned PEL entries difficult to trace during incident response.
• Stream trimming policy: The maxlen parameter with approximate=True prevents unbounded memory growth. For GenAI audit trails where retention matters, set maxlen conservatively high (1,000,000+) and implement a separate archival process that reads older entries to cold storage before they are trimmed.

Code Walkthrough

Building on the consumer-naming and stream-trimming decisions described above, this section turns them into two concrete abstractions: an EventPublisher that writes domain events with XADD, and an AIEventBackbone that creates consumer groups and consumes with XREADGROUP. Both use the redis.asyncio client so the inference service publishes without adding latency to the model response path.

The publisher serializes a BaseAIEvent dataclass to a flat field map and appends it with maxlen plus approximate=True, exactly the conservative-but-bounded trimming policy the concepts call for:

Code snippetpython
1import json
2import time
3from dataclasses import dataclass, field, asdict
4from redis.asyncio import Redis
5
6@dataclass
7class BaseAIEvent:
8    event_type: str
9    source_service: str
10    timestamp: float = field(default_factory=time.time)
11    correlation_id: str = ""
12    payload: dict = field(default_factory=dict)
13
14class EventPublisher:
15    def __init__(self, redis_client: Redis, max_stream_length: int = 1_000_000):
16        self._redis = redis_client
17        self._max_len = max_stream_length
18
19    async def publish(self, stream: str, event: BaseAIEvent) -> str:
20        fields = {k: str(v) for k, v in asdict(event).items() if k != "payload"}
21        fields["payload"] = json.dumps(event.payload)
22        return await self._redis.xadd(
23            stream, fields, maxlen=self._max_len, approximate=True
24        )

The backbone wraps the publisher and manages the read path. It creates a consumer group at the start of the stream, then reads with the special > ID so each entry reaches exactly one consumer in the group. The consumer name follows the deterministic {group}-{pod_id} scheme so a crashed pod's pending entries can be reclaimed later from its Pending Entry List. Each handled entry is acknowledged with XACK:

Code snippetpython
1class AIEventBackbone:
2    def __init__(self, redis_client: Redis):
3        self._redis = redis_client
4        self.publisher = EventPublisher(redis_client)
5
6    async def create_consumer_group(self, stream: str, group: str) -> None:
7        try:
8            await self._redis.xgroup_create(stream, group, id="0", mkstream=True)
9        except Exception:
10            pass  # group already exists (BUSYGROUP)
11
12    async def consume(self, stream, group, pod_id, handler):
13        consumer = f"{group}-{pod_id}"  # deterministic {service}-{pod-id}
14        while True:
15            batches = await self._redis.xreadgroup(
16                group, consumer, {stream: ">"}, count=10, block=2000
17            )
18            for _name, messages in batches:
19                for entry_id, data in messages:
20                    await handler(entry_id, data)
21                    await self._redis.xack(stream, group, entry_id)

This separates the write path from the read path, letting producers and consumer groups scale independently. Verify by publishing one test event and confirming XREADGROUP delivers it to exactly one consumer per group while XACK clears it from the Pending Entry List.

Do's and Don'ts

Do's

1. ✓Do serialize BaseAIEvent.payload with json.dumps and all scalar fields with str() separately inside EventPublisher.publish() — Redis Streams stores only byte strings; applying str() to the payload dict stores a Python repr ({'k': 'v'}) rather than valid JSON, so any consumer calling json.loads(data["payload"]) raises a parse error and the event is effectively unreadable downstream.
2. ✓Do pass id="0" and mkstream=True to xgroup_create when standing up the AIEventBackbone — id="0" positions the consumer group at the head of the existing stream so events queued before the group was registered (e.g., during a rolling deploy) are not silently skipped; mkstream=True eliminates the race where XREADGROUP reaches a stream the first XADD hasn't yet created.
3. ✓Do name each consumer with the deterministic {group}-{pod_id} pattern rather than a random UUID — The AIEventBackbone.consume() loop ties the consumer name to an addressable pod so that, after a crash, XCLAIM can target that exact Pending Entry List by name and reclaim the stuck entries; a random name on each restart orphans the PEL with no live consumer to claim it.

Don'ts

1. ✗Don't pass a synchronous redis.Redis client to EventPublisher or AIEventBackbone — both classes await their Redis calls (xadd, xreadgroup, xack, xgroup_create); a blocking client turns every XADD in the inference publish path into a thread-stalling I/O call that adds latency directly to the model response, defeating the non-blocking contract redis.asyncio provides.
2. ✗Don't re-raise the exception inside create_consumer_group's except block — the method deliberately swallows BUSYGROUP because a group already existing on pod restart is the normal steady-state, not a fault; surfacing the exception would crash the backbone on every redeployment against a live stream that already has the group registered.
3. ✗Don't call XREADGROUP with a literal message ID instead of the special ">" sentinel in the consume() loop — ">" is the only value that delivers new, undelivered entries to the consumer; a numeric ID re-reads the consumer's own Pending Entry List instead, so new inference events published after startup are never received by any worker group.