Message Queues for LLM Systems

Why Use Queues? Decouple components, handle traffic spikes, enable async processing, and improve reliability.

Queue Architecture

[API Server] --> [Message Queue] --> [LLM Workers]
                      |
                      v
              [Result Store/DB]

Common Message Brokers

Celery Example

from celery import Celery

app = Celery("llm_tasks", broker="redis://localhost:6379")

@app.task
def process_llm_request(prompt, model="gpt-4"):
    response = llm.generate(prompt, model=model)
    return response

# Producer
task = process_llm_request.delay("Explain quantum computing")
task_id = task.id

# Check result
result = process_llm_request.AsyncResult(task_id)
if result.ready():
    output = result.get()

Redis Queue (RQ)

from redis import Redis
from rq import Queue

redis_conn = Redis()
q = Queue(connection=redis_conn)

def llm_inference(prompt):
    return llm.generate(prompt)

# Enqueue
job = q.enqueue(llm_inference, "Hello, world!")

# Check status
job.refresh()
if job.is_finished:
    result = job.result

Priority Queues

high_priority = Queue("high", connection=redis_conn)
low_priority = Queue("low", connection=redis_conn)

# Premium users
high_priority.enqueue(llm_inference, prompt)

# Free users
low_priority.enqueue(llm_inference, prompt)

# Workers process high first
Worker(["high", "low"]).work()

Dead Letter Queues Handle failed messages:

@app.task(bind=True, max_retries=3)
def process_with_retry(self, prompt):
    try:
        return llm.generate(prompt)
    except Exception as e:
        if self.request.retries >= 3:
            # Move to dead letter queue
            dead_letter_queue.enqueue(prompt, error=str(e))
            raise
        raise self.retry(exc=e, countdown=2 ** self.request.retries)

Patterns

Best Practices

• Set appropriate timeouts
• Implement retry logic with backoff
• Use dead letter queues for failures
• Monitor queue depth and latency