Synchronous LLM calls block your server thread for 500ms–5s per request. Async and batch patterns let you serve orders of magnitude more requests with the same infrastructure — and reduce cost by consolidating many small calls into efficient batches.
Select a pattern to see its use case and implementation:
Multiple independent calls in parallel — agent tool calls, fan-out, batch classification.
import asyncio
import anthropic
client = anthropic.AsyncAnthropic() # Note: Async client
async def classify(text: str) -> str:
response = await client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=50,
messages=[{"role": "user", "content": f"Sentiment: {text}"}],
)
return response.content[0].text.strip()
async def classify_many(texts: list[str]) -> list[str]:
# Cap concurrency to avoid rate limits
semaphore = asyncio.Semaphore(20)
async def bounded(text):
async with semaphore:
return await classify(text)
return await asyncio.gather(*[bounded(t) for t in texts])
# 100 texts: sequential = 40s → concurrent = ~0.8s| Pattern | When to Use | Throughput |
|---|---|---|
| Synchronous | User-facing: response needed in <5s (chat, Q&A) | Low — one at a time |
| Async (concurrent) | Multiple independent calls in parallel (agent tools, fan-out) | High — N concurrent |
| Queue + Worker | Background tasks: doc processing, email gen, report gen | Highest — decoupled from API |
| Batch API | Offline: classification, eval, data enrichment (24hr budget) | Maximum — provider-optimized, 50% cheaper |
import asyncio
import anthropic
# Use the async client for concurrent calls
client = anthropic.AsyncAnthropic()
async def classify_single(text: str) -> str:
response = await client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=50,
messages=[{"role": "user", "content": f"Classify sentiment: {text}"}],
)
return response.content[0].text.strip()
async def classify_batch_concurrent(texts: list[str]) -> list[str]:
"""Process N texts concurrently — wall time = slowest single call."""
tasks = [classify_single(text) for text in texts]
return await asyncio.gather(*tasks)
# 100 texts: sequential ≈ 100 × 400ms = 40s
# Concurrent: ≈ 1 × 600ms = 0.6s (but respect rate limits!)
texts = ["The product was great!", "Terrible service.", "Average experience."] * 33
results = asyncio.run(classify_batch_concurrent(texts))
# Rate limit awareness — use a semaphore to cap concurrency
async def classify_with_rate_limit(texts: list[str], max_concurrent: int = 20) -> list[str]:
semaphore = asyncio.Semaphore(max_concurrent)
async def bounded_classify(text: str) -> str:
async with semaphore:
return await classify_single(text)
return await asyncio.gather(*[bounded_classify(t) for t in texts])# pip install celery redis anthropic
# tasks.py
from celery import Celery
import anthropic
app_celery = Celery("ai_tasks", broker="redis://localhost:6379/0")
client = anthropic.Anthropic()
@app_celery.task(bind=True, max_retries=3)
def process_document(self, doc_id: str, content: str) -> dict:
"""Background task — runs outside the request/response cycle."""
try:
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1000,
messages=[{"role": "user", "content": f"Summarize this document:\n\n{content}"}],
)
return {"doc_id": doc_id, "summary": response.content[0].text}
except anthropic.RateLimitError as exc:
raise self.retry(exc=exc, countdown=60) # Retry after 60s
# api.py — FastAPI endpoint
from fastapi import FastAPI, BackgroundTasks
api = FastAPI()
@api.post("/documents/{doc_id}/process")
async def start_processing(doc_id: str, content: str):
"""Returns immediately — user polls /status endpoint."""
task = process_document.delay(doc_id, content)
return {"task_id": task.id, "status": "queued"}
@api.get("/tasks/{task_id}/status")
async def task_status(task_id: str):
task = process_document.AsyncResult(task_id)
return {"status": task.status, "result": task.result if task.ready() else None}import anthropic
client = anthropic.Anthropic()
# Create a batch (up to 10,000 requests per batch)
batch = client.messages.batches.create(
requests=[
{
"custom_id": f"doc-{i}",
"params": {
"model": "claude-haiku-4-5-20251001",
"max_tokens": 200,
"messages": [{"role": "user", "content": f"Classify: {text}"}],
},
}
for i, text in enumerate(texts)
]
)
print(f"Batch ID: {batch.id}")
print(f"Processing time: up to 24 hours")
# Poll for completion (or use webhook)
import time
while True:
status = client.messages.batches.retrieve(batch.id)
if status.processing_status == "ended":
break
time.sleep(60)
# Retrieve results
for result in client.messages.batches.results(batch.id):
print(f"{result.custom_id}: {result.result.message.content[0].text}")Next: even with async and batching, the API will sometimes fail. Lesson 5 covers circuit breakers and fallback patterns that keep your system serving users even when the LLM provider is down.