queue
**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**
| Broker | Best For |
|--------|----------|
| Redis | Simple queues, low latency |
| RabbitMQ | Complex routing, reliability |
| Kafka | High throughput, streaming |
| AWS SQS | Managed, serverless |
| Celery | Python task queue |
**Celery Example**
```python
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)**
```python
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**
```python
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:
```python
@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**
| Pattern | Use Case |
|---------|----------|
| Request-response | Synchronous-like with polling |
| Fire-and-forget | Background processing |
| Fan-out | Multiple consumers |
| Priority | Tiered service levels |
**Best Practices**
- Set appropriate timeouts
- Implement retry logic with backoff
- Use dead letter queues for failures
- Monitor queue depth and latency