Chapter 2 · Message Queues: Redis Queue and Celery Worker

Chapter goals:
Understand how Redis queue primitives (rpush/lpush/brpop) work
Master the complete configuration and operation of Celery + Redis
Learn to asyncize time-consuming tasks like LLM calls for non-blocking responses
Master Celery retry strategies: retry_backoff and max_retries
Build a complete pipeline: FastAPI submits task → Celery Worker processes → polling for results

2.1 Why Do We Need Message Queues?

In the previous chapter, we used FastAPI's BackgroundTasks to handle lightweight background tasks (see FastAPI ch16). But when tasks require:

Cross-process execution (main service crash won't affect the queue)
Persistence (tasks won't be lost after Worker restart)
Retry and failure isolation (single task failure won't affect other tasks)
Horizontal scaling (multiple Workers consume in parallel)

BackgroundTasks is no longer sufficient. Message queues are essential infrastructure for production-grade Agent services.

Client ──HTTP──▶ FastAPI Service ──enqueue──▶ Redis Queue ──dequeue──▶ Celery Worker
                                         (persistent, retry, concurrent)

2.2 Redis Queue Primitives

Redis is the most commonly used message queue backend. Its core queue operations are only three:

Command	Purpose	Blocking Version
`RPUSH queue item`	Push element from right side (producer)	—
`LPUSH queue item`	Push element from left side	—
`RPOP queue`	Pop element from right side (consumer)	`BRPOP`
`LRANGE queue 0 -1`	View all queue contents	—

2.2.1 Blocking Pop: BRPOP

RPOP returns nil immediately when the queue is empty, forcing consumers to poll continuously (busy polling wastes CPU). BRPOP is the blocking version: when the queue is empty, the client suspends until a new element is added or timeout occurs.

python

import redis

r = redis.Redis(host='localhost', port=6379, db=0)

# Producer: push tasks
r.rpush('tasks', 'task_001')
r.rpush('tasks', 'task_002')

# Consumer: block and wait (up to 5 seconds)
item = r.brpop('tasks', timeout=5)
# item looks like (b'tasks', b'task_001')
if item:
    queue_name, task_id = item
    print(f'Got task: {task_id.decode()}')
else:
    print('No tasks in queue')

2.2.2 Multi-Queue Broadcast: BRPOPLPUSH

A common requirement is "priority queue": move low-priority tasks from normal queue to degradation queue, process them after the main queue is cleared.

python

# Move task from 'high' queue to 'low' queue (atomic operation)
result = r.brpoplpush('high', 'low', timeout=5)

2.3 Celery Introduction: Configuration and Task Definition

Celery is a distributed task queue framework supporting multiple Brokers (Redis, RabbitMQ, Amazon SQS, etc.). We use Redis as the Broker.

2.3.1 Installation

bash

pip install -U celery[redis] redis

Start Redis service (using Docker is simplest):

bash

docker run -d --name redis-queue -p 6379:6379 redis:7-alpine

2.3.2 Define Celery App and Task

python

# app.py
from celery import Celery

app = Celery(
    'agent_worker',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',
)

# Set timezone and serialization
app.conf.update(
    task_serializer='json',
    result_serializer='json',
    accept_content=['json'],
    timezone='Asia/Shanghai',
    enable_utc=True,
)

2.3.3 Define a Simple Task

python

# tasks.py
from app import app
import time

@app.task(bind=True, max_retries=3)
def call_llm(self, prompt: str, model: str = "gpt-4o-mini") -> dict:
    """Simulate LLM call (replace with real API call in production)"""
    try:
        # Simulate API call delay
        time.sleep(2)
        return {
            "model": model,
            "response": f"This is a simulated reply to '{prompt}'",
            "tokens": 42,
        }
    except Exception as exc:
        # Auto-retry with exponential backoff
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

2.3.4 Start Worker

bash

# Terminal 1: Start Worker (listening to Redis queue)
celery -A tasks worker --loglevel=info --concurrency=4

2.3.5 Submit Task

python

# Synchronous submit, returns AsyncResult immediately
result = call_llm.delay("What is a large language model?", model="gpt-4o")
print(f"Task ID: {result.id}")

# Block and wait for result (not recommended in production, consumes connections)
# response = result.get(timeout=30)

# Non-blocking: check status periodically
import time
while not result.ready():
    time.sleep(1)
print(f"Done! Result: {result.get()}")

2.4 Agent Async Task Pattern: LLM Call Queueing

In Agent services, LLM calls are usually the slowest part (network latency + generation time). Queueing them allows FastAPI to return task ID immediately, and the client polls or uses WebSocket to get results.

python

# agent_tasks.py
from app import app
from openai import AsyncOpenAI
import asyncio

client = AsyncOpenAI(api_key="sk-...")

@app.task(bind=True, max_retries=5, retry_backoff=True)
def generate_agent_response(
    self,
    messages: list[dict],
    tools: list[dict] | None = None,
) -> dict:
    """
    Agent core inference task:
    - Feed multi-turn conversation to LLM
    - Parse tool_calls and execute (done internally by Worker)
    - Return final text response
    """
    try:
        # Call LLM (using OpenAI SDK here, can be replaced with any Provider)
        response = asyncio.run(
            client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                tools=tools or [],
            )
        )
        choice = response.choices[0]
        return {
            "role": choice.message.role,
            "content": choice.message.content,
            "tool_calls": [
                {
                    "id": tc.id,
                    "function": {
                        "name": tc.function.name,
                        "arguments": tc.function.arguments,
                    },
                }
                for tc in choice.message.tool_calls or []
            ],
        }
    except Exception as exc:
        # retry_backoff=True: Celery auto-exponential-backoff retry
        raise self.retry(exc=exc)

2.4.1 FastAPI Endpoint: Submit Task and Return Immediately

python

# main.py
from fastapi import FastAPI, HTTPException
from agent_tasks import generate_agent_response

app = FastAPI(title="Agent Async Service")

@app.post("/chat")
async def chat_endpoint(prompt: str, model: str = "gpt-4o"):
    """
    Submit conversation task, return task_id immediately.
    Client polls for result using task_id.
    """
    task = generate_agent_response.delay(
        messages=[{"role": "user", "content": prompt}],
        model=model,
    )
    return {"task_id": task.id, "status": "queued"}

@app.get("/chat/{task_id}")
async def get_result(task_id: str):
    """Poll task result"""
    from celery.result import AsyncResult
    result = AsyncResult(task_id)

    if result.state == "PENDING":
        return {"status": "pending"}
    if result.state == "STARTED":
        return {"status": "processing"}
    if result.state == "FAILURE":
        raise HTTPException(status_code=500, detail=str(result.result))
    if result.state == "SUCCESS":
        return {"status": "done", "result": result.result}
    return {"status": result.state}

2.5 Retry Strategy Deep Dive

In production, LLM APIs frequently encounter rate limiting (429) or timeouts. Celery provides two retry mechanisms:

2.5.1 Task-Level Retry (bind=True)

python

@app.task(bind=True, max_retries=5, retry_backoff=True)
def robust_llm_call(self, prompt: str):
    try:
        return do_api_call(prompt)
    except RateLimitError:
        # retry_backoff=True: wait 1s, 2s, 4s... on each retry
        raise self.retry(exc=self.__class__.exc, countdown=2 ** self.request.retries)
    except ConnectionError:
        # ConnectionError retries immediately, no wait
        raise self.retry(exc=self.__class__.exc, max_retries=10)

2.5.2 Global Default Configuration

python

app.conf.update(
    # All tasks default max retries: 3
    task_default_max_retries = 3,
    # Enable exponential backoff (requires bind=True)
    task_default_retry_backoff = True,
    # Backoff base (seconds)
    task_default_retry_backoff_base = 2,
    # Task timeout (hard timeout, forces kill within Worker)
    task_soft_time_limit = 120,   # Soft timeout: raises SoftTimeLimitExceeded
    task_time_limit = 180,        # Hard timeout: directly kills Worker subprocess
)

2.6 Complete Example: FastAPI + Celery + Redis

Create a runnable Agent service project:

agent-service/
├── app.py              # Celery App configuration
├── tasks.py            # All async tasks
├── main.py             # FastAPI application
├── requirements.txt
└── docker-compose.yml  # Redis + Worker one-click start

requirements.txt

txt

celery[redis]>=5.4.0
fastapi>=0.115.0
uvicorn[standard]>=0.32.0
redis>=5.0.0
openai>=1.50.0
pydantic>=2.0.0

docker-compose.yml

yaml

version: "3.9"
services:
  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]

  api:
    build: .
    command: uvicorn main:app --host 0.0.0.0 --port 8000
    ports: ["8000:8000"]
    environment:
      - REDIS_URL=redis://redis:6379/0
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    depends_on: [redis]

  worker:
    build: .
    command: celery -A tasks worker --loglevel=info --concurrency=4
    environment:
      - REDIS_URL=redis://redis:6379/0
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    depends_on: [redis]

Run

bash

# Start infrastructure
docker compose up -d redis

# Terminal 1: Start Worker
celery -A tasks worker --loglevel=info --broker=redis://localhost:6379/0

# Terminal 2: Start API
uvicorn main:app --reload

# Terminal 3: Test
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Explain quantum computing"}'
# → {"task_id": "xxx", "status": "queued"}

curl http://localhost:8000/chat/xxx
# → {"status": "done", "result": {"role": "assistant", "content": "..."}}

Chapter Summary

Redis Queue Primitives: RPUSH/LPOP implement FIFO, BRPOP implements blocking consumption, avoiding busy polling;
Celery Configuration: Both broker (task source) and backend (result storage) point to different databases in Redis;
Task Definition: Use @app.task decorator on functions, bind=True to get self for calling retry;
Deferred Submission: task.delay(...) returns AsyncResult, doesn't block FastAPI request;
Poll Results: Query AsyncResult.state through /chat/{task_id} endpoint, handle PENDING/STARTED/SUCCESS/FAILURE four states;
Retry Strategy: retry_backoff=True enables exponential backoff, max_retries limits maximum attempts.

📌 More advanced async interaction patterns (WebSocket real-time push, Server-Sent Events) are covered in LangGraph ch08, this chapter focuses on polling pattern.

🛠️ Hands-on Practice

Start Redis locally (docker run -d --name redis -p 6379:6379 redis:7-alpine), write a Celery Task that completes the full pipeline of "writing user input to Redis List, then another Task pops and returns length", verify delay() + get() flow.
Add simulated RateLimitError to generate_agent_response Task (throw exception on first 3 calls, succeed on 4th), observe auto-retry behavior in Celery Worker logs, record time intervals between retries, verify exponential backoff works.
Change the FastAPI endpoint in this chapter to return SSE streaming response (StreamingResponse), push JSON result once when Worker task completes, replacing polling endpoint. Hint: wrap result.get(timeout=...) with asyncio.to_thread.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. What is the main difference between BRPOP and RPOP?

2. When Celery retry_backoff=True is used with bind=True, how does the retry interval change?

3. What is the risk of using AsyncResult.get() to block and wait for results?

4. Which scenario is most suitable for Celery instead of FastAPI BackgroundTasks?

Chapter 2 · Message Queues: Redis Queue and Celery Worker ​

2.1 Why Do We Need Message Queues? ​

2.2 Redis Queue Primitives ​

2.2.1 Blocking Pop: BRPOP ​

2.2.2 Multi-Queue Broadcast: BRPOPLPUSH ​

2.3 Celery Introduction: Configuration and Task Definition ​

2.3.1 Installation ​

2.3.2 Define Celery App and Task ​

2.3.3 Define a Simple Task ​

2.3.4 Start Worker ​

2.3.5 Submit Task ​

2.4 Agent Async Task Pattern: LLM Call Queueing ​

2.4.1 FastAPI Endpoint: Submit Task and Return Immediately ​

2.5 Retry Strategy Deep Dive ​

2.5.1 Task-Level Retry (bind=True) ​

2.5.2 Global Default Configuration ​

2.6 Complete Example: FastAPI + Celery + Redis ​

requirements.txt ​

docker-compose.yml ​

Run ​

Chapter Summary ​

🛠️ Hands-on Practice ​

🧪 随堂测验

Chapter 2 · Message Queues: Redis Queue and Celery Worker

2.1 Why Do We Need Message Queues?

2.2 Redis Queue Primitives

2.2.1 Blocking Pop: BRPOP

2.2.2 Multi-Queue Broadcast: BRPOPLPUSH

2.3 Celery Introduction: Configuration and Task Definition

2.3.1 Installation

2.3.2 Define Celery App and Task

2.3.3 Define a Simple Task

2.3.4 Start Worker

2.3.5 Submit Task

2.4 Agent Async Task Pattern: LLM Call Queueing

2.4.1 FastAPI Endpoint: Submit Task and Return Immediately

2.5 Retry Strategy Deep Dive

2.5.1 Task-Level Retry (bind=True)

2.5.2 Global Default Configuration

2.6 Complete Example: FastAPI + Celery + Redis

requirements.txt

docker-compose.yml

Run

Chapter Summary

🛠️ Hands-on Practice