Agent pipelines built entirely in-process have a critical weakness: if the orchestrating process crashes, all in-flight work is lost, and there is no record of what was attempted. Apache Kafka and Redis Streams solve this by externalizing the message bus, agent tasks become durable, replayable events that outlive any individual process. The choice between them comes down to a single axis: how long do you need messages to survive, and how much operational complexity can your team absorb?
Kafka vs Redis Streams: Choosing the Right Tool
Both systems implement a log-based message model where consumers maintain their position (offset) independently. The divergence is in durability, ecosystem, and complexity.
flowchart TD
A["Inter-Agent Messaging Need"] --> B{"Message retention\nrequirement?"}
B -->|"Hours or less\n(cache-like)"| C["Redis Streams\n- Sub-millisecond latency\n- Simple ops\n- Memory-backed"]
B -->|"Days to weeks\n(durable audit log)"| D["Apache Kafka\n- Disk-based retention\n- Replay any offset\n- Higher ops overhead"]
C --> E{"Team ops\ncapacity?"}
D --> F["Use Kafka:\nregulated industries,\nreplay, multi-day retention"]
E -->|"Small team,\nmanaged infra"| G["Use Redis (Upstash / ElastiCache)"]
E -->|"Platform team\nwith Kafka expertise"| H["Kafka also viable\nfor low-latency needs"]
style A fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style B fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style C fill:#f0fdf9,stroke:#0D9488,color:#0F172A
style D fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style F fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style G fill:#f0fdf9,stroke:#0D9488,color:#0F172AUse Kafka when: tasks must survive days, you need full replay for debugging or reprocessing, regulatory compliance requires an immutable audit log, or you’re building a multi-team platform where many services consume the same stream independently.
Use Redis Streams when: tasks are ephemeral, latency is sub-10ms, your team doesn’t have Kafka expertise, or you’re already running Redis for other purposes and want to minimize infrastructure components.
Kafka Patterns for Agent Task Distribution
from confluent_kafka import Producer, Consumer, KafkaError
import json, time
# WHY confluent-kafka over kafka-python: confluent's client wraps librdkafka,
# a C library that provides 10-20x higher throughput and more reliable
# delivery semantics than the pure-Python kafka-python client.
producer_config = {
"bootstrap.servers": KAFKA_BROKERS,
"security.protocol": "SASL_SSL",
"sasl.mechanism": "PLAIN",
"sasl.username": KAFKA_API_KEY,
"sasl.password": KAFKA_API_SECRET,
# WHY acks='all': requires all in-sync replicas to acknowledge before
# the producer considers the message committed. This prevents message
# loss if the leader broker fails immediately after a write.
"acks": "all",
# WHY enable.idempotence=True: prevents duplicate messages during
# producer retries. Without this, a network timeout could cause the
# producer to resend a message that was already committed.
"enable.idempotence": True,
}
def publish_agent_task(producer, task: dict, task_type: str):
"""
WHY key by task_type: Kafka routes messages with the same key to the
same partition. This guarantees ordering within a task type, critical
if downstream agents must process tasks for a given entity in sequence.
"""
producer.produce(
topic="agent-tasks",
key=task_type.encode(),
value=json.dumps(task).encode(),
headers={"trace_id": task["trace_id"], "priority": str(task.get("priority", 5))},
on_delivery=lambda err, msg: handle_delivery(err, msg, task),
)
producer.poll(0) # trigger delivery callbacks without blockingConsumer Groups and Dead-Letter Topics
Consumer groups allow multiple agent instances to share the load of a single topic. Each partition is assigned to exactly one consumer in the group, adding consumers scales throughput linearly up to the partition count.
flowchart TD
T["Topic: agent-tasks\n(8 partitions)"] --> P0["Partition 0"]
T --> P1["Partition 1"]
T --> P2["Partition 2-7"]
P0 --> C1["Agent Worker 1\n(group: doc-analyzers)"]
P1 --> C2["Agent Worker 2\n(group: doc-analyzers)"]
P2 --> C3["Agent Worker 3\n(group: doc-analyzers)"]
C1 -->|"processing fails\nafter 3 retries"| DLT["Dead-Letter Topic\nagent-tasks.DLT"]
DLT --> DR["DLT Consumer\n(human review / reprocess)"]
style T fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style DLT fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style DR fill:#f0fdf9,stroke:#0D9488,color:#0F172AMAX_RETRIES = 3
def consume_with_dlq(consumer, dlq_producer, topic: str):
consumer.subscribe([topic])
retry_counts = {}
while True:
msg = consumer.poll(timeout=1.0)
if msg is None or msg.error():
continue
msg_id = f"{msg.partition()}-{msg.offset()}"
try:
task = json.loads(msg.value())
process_agent_task(task)
# WHY manual commit after success: auto-commit could commit
# an offset before the task completes, causing permanent loss
# if the process dies mid-task. Manual commit ensures at-least-once
# delivery semantics: tasks may be reprocessed but never skipped.
consumer.commit(msg)
retry_counts.pop(msg_id, None)
except Exception as exc:
retries = retry_counts.get(msg_id, 0) + 1
retry_counts[msg_id] = retries
if retries >= MAX_RETRIES:
# Route to DLT with failure context for later inspection
dlq_producer.produce(
topic=f"{topic}.DLT",
value=msg.value(),
headers={
"original_topic": topic,
"original_partition": str(msg.partition()),
"original_offset": str(msg.offset()),
"exception": type(exc).__name__,
"error_message": str(exc),
},
)
dlq_producer.flush()
consumer.commit(msg) # advance past the poison pill
retry_counts.pop(msg_id, None)Redis Streams: XADD, XREAD, and Backpressure
Redis Streams use XADD to append messages and XREADGROUP for consumer-group-based consumption with acknowledgement semantics. The key operational advantage: if a consumer crashes mid-task, its unacknowledged messages remain in a “pending entries list” (PEL) and can be claimed by another consumer via XAUTOCLAIM.
import redis.asyncio as aioredis
async def consume_redis_stream(r: aioredis.Redis, stream: str, group: str, consumer: str):
"""
WHY XAUTOCLAIM before XREADGROUP: first claim any messages that have been
pending (unacknowledged) for more than 60 seconds: these belong to
consumers that likely crashed. This implements at-least-once delivery
without a separate recovery process.
"""
IDLE_THRESHOLD_MS = 60_000
while True:
# Recover stale pending messages from crashed consumers
claimed, *_ = await r.xautoclaim(stream, group, consumer, IDLE_THRESHOLD_MS, "0-0", count=10)
for msg_id, fields in claimed:
await process_and_ack(r, stream, group, msg_id, fields)
# Then consume fresh messages
results = await r.xreadgroup(group, consumer, {stream: ">"}, count=50, block=500)
if not results:
continue
for _, messages in results:
if len(messages) >= 45:
# WHY pause when batch is near-full: the consumer is keeping
# up but just barely. Pause briefly to let processing catch up
# rather than accumulating a backlog in the PEL.
await asyncio.sleep(0.1)
for msg_id, fields in messages:
await process_and_ack(r, stream, group, msg_id, fields)Monitor consumer lag in both systems: in Kafka, consumer_lag = latest_offset - committed_offset per partition; in Redis Streams, XLEN stream_name minus XINFO GROUPS delivered count. Alert when lag exceeds a threshold tied to your SLA, lag is the earliest signal of an overwhelmed agent tier.
The message bus patterns in this lesson give your agent network durability and horizontal scalability. But durable messaging only works correctly when the structure of those messages is stable and versioned, the next lesson covers how to design message contracts between agents that can evolve without breaking consumers.