Building a Resilient Python Workflow Engine with Redis Streams

A resilient Python workflow engine is built by replacing synchronous HTTP calls with Redis Streams and asynchronous workers to handle long-running tasks. This event-driven architecture ensures at-least-once delivery and state persistence, allowing AI pipelines to recover from failures without losing progress.

Last Tuesday at 3:14 AM, my PagerDuty went off for the third time in a week. My AI-powered content analysis engine, which relies heavily on Gemini Pro 1.5, was hitting a wall. The logs were a mess of 504 Gateway Timeouts and "Connection Reset by Peer" errors. In my initial design, I had built a standard FastAPI endpoint that triggered a sequence of LLM calls. It worked fine for short summaries, but as soon as I started feeding it 50k-token documents, the processing time climbed to over four minutes. Cloud Run’s ingress timeout and the inherent fragility of long-lived HTTP connections were killing my success rate. Specifically, 45% of my tasks were failing mid-execution, leaving my database in an inconsistent state and wasting thousands of tokens on unfinished requests.

I realized I couldn't treat AI workflows like standard REST requests. AI agents are non-deterministic, high-latency, and prone to external API failures. I needed to move from a synchronous architecture to a resilient, event-driven Python workflow engine. I decided to build this using Redis Streams and a fleet of asynchronous Python workers. This shift didn't just stop the 504 errors; it allowed me to implement granular retries, state persistence, and backpressure management. This shift saved me roughly $400 in wasted API costs in the first week alone.

Why Synchronous FastAPI Fails for Long-Running AI Workflows

Synchronous architectures fail when processing times exceed HTTP timeouts, leading to a 45% failure rate in long-running AI tasks. My original setup was naive. A user would POST a document URL, the FastAPI worker would download the content, hit Gemini for an initial analysis, hit it again for a detailed breakdown, and then save the result. If the connection dropped at step two, the whole process died. There was no way to resume. I was essentially running a state machine inside a single volatile memory space.

When you're dealing with LLMs, you have to assume the network will fail or the provider will rate-limit you. I previously documented some of these challenges in my post on Gemini Error Handling: Building Resilient AI Pipelines with Idempotency, but handling errors at the code level isn't enough if the entire process execution environment is unstable. I needed a way to persist the "state" of the workflow so that if a worker crashed, another could pick up exactly where it left off.

How Redis Streams Provide Superior Reliability Over Pub/Sub

Redis Streams offer persistent message logs and consumer groups, which are essential for ensuring that no workflow tasks are lost during worker crashes. I chose Redis as my backbone, but specifically, I chose Redis Streams. Many developers reach for Redis Pub/Sub for event-driven systems, but that's a mistake for resilient workflows. Pub/Sub is "fire and forget." If your worker is down when a message is published, that message is gone forever. Redis Streams, however, provides a persistent log of messages. It supports Consumer Groups, which allow multiple workers to coordinate, ensuring that each message is delivered to a consumer and, crucially, acknowledging (ACKing) that the message was processed.

This "at-least-once" delivery guarantee is what I needed. If a Cloud Run instance scales down or crashes while processing a task, the message remains in the "Pending Entries List" (PEL). After a visibility timeout, another worker can claim that message and retry it. This is the foundation of a resilient system.

Implementing the Stream Producer in FastAPI for Lower Latency

Moving heavy logic out of the FastAPI request cycle and into a Redis Stream reduces API response times from 45 seconds to under 10 milliseconds. The first step was to strip the logic out of my FastAPI routes. Now, the route only validates the input and pushes a "workflow_initiated" event into the Redis stream. I used Pydantic v2 for strict schema validation, which is critical when you're passing messages between disparate processes. If you're still on the older version of Pydantic, check out my Pydantic v2 Migration Guide to fix the breaking changes that often bite you in these types of integrations.

import uuid
from fastapi import FastAPI, HTTPException
from redis.asyncio import Redis
from pydantic import BaseModel

app = FastAPI()
redis_client = Redis(host='localhost', port=6379, decode_responses=True)

class AnalysisRequest(BaseModel):
    document_url: str
    user_id: str

@app.post("/analyze")
async def trigger_analysis(request: AnalysisRequest):
    job_id = str(uuid.uuid4())
    payload = {
        "job_id": job_id,
        "document_url": request.document_url,
        "user_id": request.user_id,
        "status": "pending"
    }
    
    # Add to Redis Stream
    await redis_client.xadd("workflow_stream", payload)
    
    return {"job_id": job_id, "message": "Analysis queued"}

This change alone dropped my API response times significantly. The user gets an immediate confirmation, and the heavy lifting happens in the background via the Python workflow engine.

How to Build a Resilient Worker Loop with Consumer Groups

Using the XREADGROUP command allows multiple Python workers to coordinate and process messages with a guaranteed visibility timeout for recovery. The worker is where the real complexity lies. I built a Python script that runs an infinite loop, listening for new messages in the stream. I'm using the XREADGROUP command to ensure that multiple workers don't process the same job simultaneously.

One of the hardest parts of building this was handling the "Visibility Timeout." If a worker takes 10 minutes to process a Gemini request, Redis needs to know that the worker is still alive and hasn't crashed. I implemented a "heartbeat" mechanism where the worker updates a status key in Redis every 30 seconds.

import asyncio
from redis.asyncio import Redis

async def process_worker():
    r = Redis(host='localhost', port=6379, decode_responses=True)
    group_name = "analysis_group"
    consumer_name = "worker_1"

    # Create group if it doesn't exist
    try:
        await r.xgroup_create("workflow_stream", group_name, id="0", mkstream=True)
    except Exception:
        pass # Group already exists

    while True:
        # Read from stream, wait up to 5 seconds for new data
        messages = await r.xreadgroup(group_name, consumer_name, {"workflow_stream": ">"}, count=1, block=5000)
        
        for stream, payload_list in messages:
            for message_id, payload in payload_list:
                try:
                    print(f"Processing job: {payload['job_id']}")
                    
                    # Simulate complex AI workflow
                    await perform_ai_analysis(payload)
                    
                    # Acknowledge successful completion
                    await r.xack("workflow_stream", group_name, message_id)
                    await r.xdel("workflow_stream", message_id)
                except Exception as e:
                    print(f"Error processing {message_id}: {e}")
                    # Logic for retries or Dead Letter Queue goes here

async def perform_ai_analysis(data):
    # This is where the Gemini API calls live
    # We use idempotency keys here to prevent double-billing
    await asyncio.sleep(10) # Simulating latency
    return True

if __name__ == "__main__":
    asyncio.run(process_worker())

I learned the hard way that you must use XACK only after your processing is complete. If you ACK immediately after receiving the message, you lose the "at-least-once" guarantee. If your worker dies during perform_ai_analysis, the message is gone.

Managing Workflow State with High-Performance Redis Hashes

Storing the current step of a multi-stage process in Redis Hashes reduces orchestration overhead by 70% compared to traditional SQL databases. A workflow isn't just one task; it's a sequence. I needed to track whether Step 1 (Extraction) was done before starting Step 2 (Summarization). I used Redis Hashes to store the state of each job_id. This allows the system to be truly resumable. If the worker crashes during Step 2, the next worker to pick up the job checks the Hash, sees Step 1 is complete, and skips it.

This is significantly more efficient than checking a relational database like PostgreSQL every few seconds. Using Redis for state tracking reduced my workflow orchestration overhead by 70%. The sub-millisecond latency of Redis is perfect for these high-frequency state checks.

Solving the Backpressure Problem on Google Cloud Run

Throttling worker concurrency through Cloud Run's max-instances setting prevents overwhelming downstream AI APIs and avoids 429 rate limit errors. One of the biggest risks with an event-driven architecture is overwhelming your downstream services. If I have 1,000 messages in my Redis stream and I scale my Cloud Run workers to 100 instances, I might hit the Gemini API rate limits instantly. I had to implement backpressure.

I used a simple semaphore pattern within my workers, but more importantly, I configured Cloud Run's max-instances setting to act as a global throttle. By calculating my Gemini TPM (Tokens Per Minute) limits and mapping them to the number of concurrent workers, I was able to maintain a steady flow of requests without ever seeing a 429 Rate Limit error. For more details on managing high-volume scraping or API tasks on GCP, I've written about Building a Scalable Web Scraper with Python Playwright and Cloud Run, which uses a similar scaling philosophy.

Implementing a Dead Letter Queue Strategy for Persistent Failures

A Dead Letter Queue (DLQ) prevents "retry storms" by isolating poison pill messages that fail after a maximum number of retry attempts. Not all failures are transient. If a user provides a malformed document URL, no amount of retrying will fix it. I implemented a "Max Retries" counter in the Redis Hash. Once a job fails three times, the worker moves the message to a separate stream called workflow_dlq and sends a notification to Slack.

Before I had the DLQ, these "poison pill" messages would stay at the top of my queue, causing workers to