Build an Idempotent Multi-Agent Pipeline: Redis Retries

Building autonomous multi-agent pipelines in production requires bulletproof idempotency to survive network retries, connection drops, and API timeouts without generating duplicate database mutations or duplicate outbound webhooks. When orchestrating multi-step LLM chains, a simple transient 502 from an upstream model provider or a dropped socket can trigger an automatic retry that replays dangerous side effects if your execution graph lacks a distributed deduplication barrier.

In our production testing at SaaSNext, we learned this lesson the hard way. During an overnight batch run processing 1,400 customer escalation tickets, an upstream OpenAI gateway reset its connection on step 4 of an invoice refund workflow. Because our worker loop was configured with basic exponential backoff without idempotency keys, the retry logic fired three times against Stripe's billing endpoint. The customer was credited $420 instead of $140 before our rate limiter caught the anomaly. We burned four hours auditing customer transaction logs and writing manual reconciliation scripts to reverse the excess credits.

To prevent duplicate execution in agent loops, you must decouple the non-deterministic LLM planning step from deterministic tool execution using a distributed Redis idempotency lock. Every tool invocation must compute a deterministic hash from the workflow ID, step index, and normalized argument payload, acquiring a distributed lease before mutating any state.

The Two-Phase Idempotency Architecture

When designing resilient agents, we separate the execution pipeline into two distinct phases: planning and committed execution. The planning phase queries the model to decide which tools to invoke and generates an argument schema. Once the tool call is emitted, the worker enters the execution phase where it checks Redis before touching any database, API, or external service.

The Redis idempotency protocol follows three rigid rules:

1. Deterministic Key Generation: The idempotency key is computed as sha256(workflow_run_id + ":" + step_name + ":" + json.dumps(payload, sort_keys=True)).
2. Atomic Lock Acquisition: The worker executes SET idempotency:{key} "IN_PROGRESS" EX 60 NX. If the key exists, another worker is either running or has completed the task.
3. Memoized Result Return: If the key contains a completed result payload, the worker immediately short-circuits execution and returns the cached JSON without contacting external endpoints.

For teams building complex agent swarms, integrating this pattern alongside durable Pydantic AI workflows with Prefect guarantees that neither process crashes nor cluster evictions can corrupt persistent application state.

Production Implementation

Here is our production-tested multi-file implementation using Python 3.12, Redis 7, and Pydantic v2.

config.py:

import os
from pydantic_settings import BaseSettings

class WorkflowConfig(BaseSettings):
    redis_url: str = os.getenv("REDIS_URL", "redis://localhost:6379/0")
    openai_api_key: str = os.getenv("OPENAI_API_KEY", "")
    idempotency_ttl_seconds: int = 86400  # 24 hours
    lock_timeout_seconds: int = 45

    class Config:
        env_file = ".env"

config = WorkflowConfig()

idempotency.py:

import hashlib
import json
import logging
from typing import Any, Callable, Dict, Optional
import redis

from config import config

logger = logging.getLogger("IdempotentAgent")
redis_client = redis.Redis.from_url(config.redis_url, decode_responses=True)

class IdempotencyManager:
    @staticmethod
    def generate_key(workflow_id: str, step_name: str, payload: Dict[str, Any]) -> str:
        serialized = json.dumps(payload, sort_keys=True)
        digest = hashlib.sha256(f"{workflow_id}:{step_name}:{serialized}".encode()).hexdigest()
        return f"idempotency:{digest}"

    @classmethod
    def execute_idempotent(
        cls,
        workflow_id: str,
        step_name: str,
        payload: Dict[str, Any],
        action: Callable[[Dict[str, Any]], Dict[str, Any]]
    ) -> Dict[str, Any]:
        key = cls.generate_key(workflow_id, step_name, payload)
        lock_key = f"lock:{key}"

        # Check if already completed
        cached_result = redis_client.get(key)
        if cached_result:
            logger.info("Cache hit for idempotency key %s. Returning memoized result.", key)
            return json.loads(cached_result)

        # Acquire distributed lock to prevent concurrent double-execution
        acquired = redis_client.set(lock_key, "LOCKED", ex=config.lock_timeout_seconds, nx=True)
        if not acquired:
            logger.warning("Concurrent execution detected for key %s. Awaiting resolution.", key)
            raise RuntimeError(f"Lock contention for task: {step_name}")

        try:
            logger.info("Executing tool action for step %s [Workflow: %s]", step_name, workflow_id)
            result = action(payload)
            
            # Store completed result with 24-hour retention
            pipeline = redis_client.pipeline()
            pipeline.set(key, json.dumps(result), ex=config.idempotency_ttl_seconds)
            pipeline.delete(lock_key)
            pipeline.execute()
            
            return result
        except Exception as e:
            # Release lock on failure so immediate retries can attempt recovery
            redis_client.delete(lock_key)
            logger.error("Step %s failed with exception: %s. Lock released.", step_name, str(e))
            raise e

agent_pipeline.py:

import logging
import time
from typing import Dict, Any
from idempotency import IdempotencyManager

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("InvoiceAgent")

def issue_refund_tool(payload: Dict[str, Any]) -> Dict[str, Any]:
    invoice_id = payload["invoice_id"]
    amount_cents = payload["amount_cents"]
    # Simulated external payment provider call
    time.sleep(0.12)
    return {
        "status": "success",
        "invoice_id": invoice_id,
        "refund_id": f"ref_{invoice_id}_9981",
        "credited_amount": amount_cents,
        "processed_at": time.time()
    }

def run_workflow_step(workflow_id: str, invoice_id: str, amount: int):
    payload = {"invoice_id": invoice_id, "amount_cents": amount}
    
    # First execution: executes external API call and caches
    result1 = IdempotencyManager.execute_idempotent(
        workflow_id=workflow_id,
        step_name="stripe_refund",
        payload=payload,
        action=issue_refund_tool
    )
    logger.info("First attempt response: %s", result1["refund_id"])

    # Simulated worker retry after dropped network packet
    logger.info("Simulating duplicate retry due to gateway timeout...")
    result2 = IdempotencyManager.execute_idempotent(
        workflow_id=workflow_id,
        step_name="stripe_refund",
        payload=payload,
        action=issue_refund_tool
    )
    logger.info("Retry attempt response (Memoized): %s", result2["refund_id"])
    assert result1["processed_at"] == result2["processed_at"], "Idempotency violation!"

if __name__ == "__main__":
    run_workflow_step("wf_run_98412", "inv_2026_094", 14000)

requirements.txt:

redis>=5.0.4
pydantic>=2.8.2
pydantic-settings>=2.3.4

When NOT to Use This Pattern

While distributed idempotency protects critical mutations, it introduces latency and operational overhead that makes it unsuitable for every agent action:

1. Pure Information Retrieval Queries: Do not wrap read-only vector searches, documentation lookups, or web search scrapes in Redis idempotency locks. The lock acquisition overhead (1.5ms to 3ms) adds unnecessary latency when duplicate reads are completely harmless. You can review how low-latency in-process analytics bypasses Redis entirely in our guide on sub-12ms SQL analytics with FastMCP and DuckDB.
2. Non-Deterministic Multi-Modal Streaming: Avoid caching raw token streams or real-time voice outputs where users expect fluid token delivery rather than atomic block returns.
3. High-Velocity Stateless Counters: For telemetry or click-tracking nodes, use atomic Redis increments (INCRBY) rather than heavy SHA-256 payload locking.

Production Bottlenecks and Failure Modes

During high-concurrency spikes, Redis connection pool exhaustion is the primary vulnerability. If an upstream LLM step stalls for 40 seconds while holding an open connection, worker threads pile up and starve the pool. To eliminate this issue:

• Always enforce strict socket timeouts (socket_timeout=5.0, socket_connect_timeout=2.0).
• Never run LLM inference inside the critical section of the Redis distributed lock. Generate the payload first, acquire the lock, invoke the deterministic tool, and release immediately.
• Couple this approach with the principles covered in our breakdown of event-driven agents with zero DAG bottlenecks to maintain high throughput across asynchronous fan-out topologies.

For deeper architectural discussions on agentic reliability and distributed execution, explore our comprehensive collection of production AI workflows.

By Deepak Bagada, Founder & Editor-in-Chief at Daily AI World.