Event-Driven AI Pipelines with FastAPI

Key facts, with sources

• McKinsey's State of AI 2025 found nearly nine in ten organizations now use AI in at least one business function, yet only about 6 percent attribute 5 percent or more of EBIT to their AI use. (McKinsey)
• McKinsey found AI high performers are 2.8x more likely than others to have fundamentally redesigned workflows (55 percent versus 20 percent), and workflow redesign has the biggest effect on realizing EBIT impact from gen AI. (McKinsey)
• Zapier's survey of 525 enterprise executives found human-in-the-loop is the most common agent management approach at 38 percent, while 20 percent say their AI systems now operate autonomously with minimal oversight. (Zapier)
• 84 percent of enterprise leaders say they will likely or certainly increase AI agent investment over the next 12 months, with customer support (49 percent) and operations (47 percent) leading deployment. (Yahoo Finance)
• Menlo Ventures found coding and developer tools were the largest enterprise AI workflow category at $7.3 billion in 2025 spend, with half of developers now using AI tools daily. (Menlo Ventures)

Why request-response breaks for AI work

LLM calls routinely take several seconds, sometimes tens of seconds for long documents or multi-step chains. Webhook providers typically expect a response within a few seconds and retry on timeout, so a slow synchronous handler causes the sender to fire the same event again while your first call is still running. Now you are paying for duplicate inference and possibly writing duplicate results downstream.

The fix is structural, not a bigger timeout. The endpoint's only jobs are validate, deduplicate, persist to a queue, and return 202. Inference happens in workers that can take as long as they need, retry independently, and scale horizontally without touching the ingestion tier.

The ingestion endpoint: validate, dedupe, enqueue

The endpoint stays deliberately dumb. Pydantic validates the payload shape, a Redis SET with NX gives cheap idempotency against sender retries, and XADD appends the event to a stream. Returning 202 tells the sender the event is accepted but not yet processed, which is the truthful status.

Note what is absent: no model client, no business logic, no downstream writes. Everything that can fail slowly lives behind the stream.

from fastapi import FastAPI
from pydantic import BaseModel
import redis.asyncio as redis

app = FastAPI()
r = redis.from_url("redis://localhost:6379", decode_responses=True)

class DocumentEvent(BaseModel):
    doc_id: str
    tenant_id: str
    source: str

@app.post("/events/documents", status_code=202)
async def ingest(event: DocumentEvent):
    # Idempotency: sender retries with the same doc_id become no-ops
    is_new = await r.set(f"seen:{event.doc_id}", 1, nx=True, ex=86400)
    if not is_new:
        return {"status": "duplicate"}
    await r.xadd("ai-events", event.model_dump())
    return {"status": "queued"}

Workers with consumer groups

Redis Streams consumer groups give you the semantics AI workloads need: each event is delivered to exactly one worker in the group, unacknowledged events remain pending and can be reclaimed if a worker dies mid-inference, and adding throughput is just starting more worker processes with distinct consumer names.

The key discipline is acknowledging only after the full step succeeds — model call, validation, and downstream write. Ack early and a crash loses the event; ack late and the worst case is a retry, which your idempotency keys already absorb.

import asyncio
import redis.asyncio as redis
from redis.exceptions import ResponseError

STREAM, GROUP = "ai-events", "ai-workers"

async def main():
    r = redis.from_url("redis://localhost:6379", decode_responses=True)
    try:
        await r.xgroup_create(STREAM, GROUP, id="0", mkstream=True)
    except ResponseError:
        pass  # group already exists

    while True:
        batches = await r.xreadgroup(
            GROUP, "worker-1", {STREAM: ">"}, count=10, block=5000
        )
        for _stream, messages in batches or []:
            for msg_id, fields in messages:
                try:
                    await handle_event(fields)  # the LLM call lives in here
                    await r.xack(STREAM, GROUP, msg_id)
                except Exception:
                    # leave unacked; a periodic reclaim job (XAUTOCLAIM)
                    # retries or dead-letters stalled events
                    continue

asyncio.run(main())

Idempotency all the way down

Deduplicating at ingestion is not enough, because workers also retry. Every side effect a worker performs needs its own idempotency: cache the model output keyed by a hash of the input so a retried step reuses the previous result instead of paying for inference twice, and make downstream writes upserts keyed by the event id rather than blind inserts.

This discipline is what makes the whole system boring to operate. Any component can crash at any point, the event replays, and the world converges to the same state. Skip it and every incident becomes an archaeology session through duplicated records and doubled API bills.

Backpressure, priorities, and results

Two operational patterns worth adding from day one. First, monitor stream depth and pending-entry counts — those are your backpressure signals, and they tell you to scale workers or shed load long before users notice. Rate limits from the model provider surface naturally here: workers slow down, the stream absorbs the burst, and nothing is lost.

Second, decide how results get back to callers. For machine consumers, publish a completion event to another stream or fire an outbound webhook. For humans waiting in a UI, write results to a table the frontend polls, or push over a WebSocket. Keeping responses out of the ingestion request is what lets the rest of the design hold.

When to hire senior help

Bring in senior help when workflows cross system boundaries such as CRM, billing, or anything touching customer PII, or when a no-code prototype hits reliability and cost limits. The redesign work itself, mapping the process, defining checkpoints, and instrumenting metrics, benefits most from someone who has shipped production AI workflows before. If your stack includes React Native + Python + AI, a senior engineer who owns the full product beats coordinating multiple juniors.

Bottom line

Dhairya Senjaliya ships AI — AI Workflows projects worldwide — book a scoping call to discuss your specific situation.

Common pitfalls to avoid

• ✕Bolting AI onto an existing process instead of redesigning it, when McKinsey data shows redesign, not adoption, separates the roughly 6 percent of companies seeing real EBIT impact
• ✕Automating a workflow nobody measured first, leaving no baseline to prove time or cost savings when budget review comes
• ✕Using an expensive frontier model for every step instead of routing simple steps to cheap models and reserving reasoning models for the hard ones
• ✕Jumping to full autonomy on day one and skipping the human-in-the-loop stage most enterprises use to build trust and surface failure modes

Frequently asked questions

Sources

• McKinsey
• McKinsey
• Zapier
• Yahoo Finance
• Menlo Ventures