Queue-based agent pipelines: Celery, Redis, and LLM calls

A queue based agent pipeline lets you run multi-step LLM workflows without blocking request threads or losing work when a provider hiccups. This tutorial builds one with Celery, Redis, and a thin OpenAI-compatible client, showing exactly how to chain tasks, retry failures, and keep token spend visible.

Prerequisites

• Python 3.11 or newer
• A running Redis instance (docker run -d -p 6379:6379 redis:7)
• Packages: pip install celery redis openai

You should be comfortable with basic Celery concepts and environment variables. No prior agent framework is required.

Project structure

Keep it flat for clarity:

celery_app.py   # broker/backend config
llm.py          # LLM wrapper
tasks.py        # agent step definitions
pipeline.py     # chain builder
run.py          # submit a job

Wire up Celery and Redis

Redis acts as both broker and result backend. Separate DB indices avoid key collisions between queued tasks and stored results.

# celery_app.py
from celery import Celery

app = Celery(
    "agent_pipeline",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)
app.conf.task_serializer = "json"
app.conf.result_serializer = "json"
app.conf.accept_content = ["json"]
app.conf.task_track_started = True
app.conf.task_acks_late = True

task_acks_late ensures a worker that dies mid-LLM-call does not silently drop the job; it returns to the queue.

Start Redis and a worker in one terminal:

redis-server --port 6379
celery -A celery_app worker --loglevel=info

Make LLM calls safe in a task

LLM endpoints fail. Timeouts and retries must live inside the task, not the request path. Use the OpenAI SDK against any compatible gateway.

# llm.py
from openai import OpenAI
import os

# n4n.ai exposes one OpenAI-compatible endpoint across 240+ models and fails
# over automatically when a provider is rate-limited, which matters for jobs
# that run for minutes and can't afford a hard stop.
client = OpenAI(
    base_url=os.environ.get("LLM_BASE_URL", "https://api.n4n.ai/v1"),
    api_key=os.environ.get("LLM_API_KEY", "sk-noauth"),
)

def chat(prompt: str, model: str = "gpt-4o-mini") -> str:
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        timeout=30,
    )
    return resp.choices[0].message.content

The timeout parameter is critical. Without it a stalled connection blocks the worker slot indefinitely. A queue based agent pipeline is only as resilient as its weakest network call.

Define the agent tasks

Each step is a Celery task. We bind self to access retry logic. Summarize first, then extract structured entities, then persist.

# tasks.py
from celery_app import app
from llm import chat
import json

@app.task(bind=True, max_retries=3, default_retry_delay=5)
def summarize(self, text: str):
    try:
        return chat(f"Summarize in two sentences:\n{text}")
    except Exception as exc:
        raise self.retry(exc=exc)

@app.task(bind=True, max_retries=3, default_retry_delay=5)
def extract_entities(self, summary: str):
    try:
        raw = chat(f"Return a JSON list of named entities:\n{summary}")
        return json.loads(raw)
    except Exception as exc:
        raise self.retry(exc=exc)

@app.task
def store_result(entities):
    with open("/tmp/agent_out.json", "w") as f:
        json.dump(entities, f)
    return "stored"

max_retries caps the attempts. default_retry_delay spaces them so a degraded provider can recover. Never set infinite retries on a public endpoint without exponential backoff.

Chain them into a pipeline

Celery chain passes the return value of one task as the first argument of the next. This is the core of a queue based agent pipeline: discrete steps, loosely coupled, independently retryable.

# pipeline.py
from celery import chain
from tasks import summarize, extract_entities, store_result

def run_pipeline(text: str):
    return chain(
        summarize.s(text),
        extract_entities.s(),
        store_result.s(),
    ).apply_async()

Fan-out with chord

Real workflows often summarize many documents then combine. Use chord to run summaries in parallel and pass the list to a reducer.

from celery import chord

def run_batch(texts):
    header = [summarize.s(t) for t in texts]
    return chord(header)(extract_entities.s())

Adjust extract_entities to accept a list when used this way, or add a combine.s() task. The queue based agent pipeline pattern adapts without restructuring the broker.

Run the worker and submit a job

With the worker running, execute run.py:

# run.py
from pipeline import run_pipeline

if __name__ == "__main__":
    text = "OpenAI launched GPT-4 in 2023. It powers many agentic systems."
    task = run_pipeline(text)
    print(f"Submitted pipeline: {task.id}")

Expected console output from run.py:

Submitted pipeline: 7c2f1a9e-3b4d-4c1a-9f6e-2b3c4d5e6f7a

Expected output at checkpoint

The worker log should show each task transitioning STARTED → SUCCESS:

[INFO] Task agent_pipeline.summarize[7c2f...] received
[INFO] Task agent_pipeline.summarize[7c2f...] succeeded in 1.24s
[INFO] Task agent_pipeline.extract_entities[8a1b...] received
[INFO] Task agent_pipeline.extract_entities[8a1b...] succeeded in 0.98s
[INFO] Task agent_pipeline.store_result[9c3d...] received
[INFO] Task agent_pipeline.store_result[9c3d...] succeeded in 0.01s

Inspect the file:

cat /tmp/agent_out.json

Sample content:

["OpenAI", "GPT-4", "2023"]

That confirms the pipeline executed end to end without blocking the submitter.

Idempotency and duplicate suppression

If a client double-submits, you get duplicate work. Seed the task id from a content hash:

import hashlib

def run_pipeline_idempotent(text: str):
    tid = hashlib.sha256(text.encode()).hexdigest()
    return chain(
        summarize.s(text),
        extract_entities.s(),
        store_result.s(),
    ).apply_async(task_id=tid)

Celery will reject a second submission with the same id, or return the cached result. Combine with a Redis lock for stricter control in multi-worker setups.

Token metering and cost control

Long pipelines burn tokens across steps. Capture usage from the response and push to a counter:

# llm.py (extended)
def chat_with_usage(prompt, model="gpt-4o-mini"):
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        timeout=30,
    )
    return resp.choices[0].message.content, resp.usage.total_tokens

In tasks.py, increment a Redis key per step:

import redis
r = redis.Redis(host="localhost", port=6379, db=2)

@app.task(bind=True, max_retries=3)
def summarize(self, text: str):
    try:
        out, tokens = chat_with_usage(f"Summarize:\n{text}")
        r.incrby("tokens:summarize", tokens)
        return out
    except Exception as exc:
        raise self.retry(exc=exc)

A gateway that provides per-token metering and forwards provider cache-control hints (as n4n.ai does) removes this boilerplate and cuts repeat-prompt cost. Either way, the queue based agent pipeline should never be blind to spend.

Observing with Flower

Install and run Flower to watch task latency and failures:

pip install flower
celery -A celery_app flower --port=5555

Open http://localhost:5555. You will see each retry, duration, and result state. This is non-negotiable before shipping to production.

Where to take this next

The pattern scales: replace chain with chord to fan out entity extraction across documents, or add a revoke step for cancellation. Set task_acks_late=True so a killed worker doesn’t drop in-flight LLM calls. For production, move Redis to a managed instance and add celery beat for scheduled pipelines.

A queue based agent pipeline is not exotic. It is the simplest way to make LLM workflows survive restarts, rate limits, and slow providers while keeping the caller fast.