Batch Processing with Async and Queues
Build an async extraction pipeline using asyncio and a job queue to process thousands of documents in parallel while respecting rate limits and tracking progress.
Batch Processing with Async and Queues is a free AI Engineering Academy lesson on CoddyKit — lesson 3 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Engineering Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.
Why Batch Processing Matters
Processing thousands of documents one at a time is too slow for production. A synchronous loop that calls the OpenAI API sequentially might process 1 document per second, meaning 10,000 documents take nearly 3 hours. Async batch processing can parallelize hundreds of requests simultaneously, reducing total wall-clock time by an order of magnitude.
The asyncio Foundation
Python's asyncio event loop lets you run many I/O-bound tasks concurrently without threads. When one API call is waiting for a network response, the event loop switches to processing another. You write code with async def and await keywords, and the runtime handles the scheduling. This is ideal for LLM calls which spend most of their time waiting for the server.
import asyncio
import instructor
from openai import AsyncOpenAI
async_client = instructor.from_openai(AsyncOpenAI())
async def extract_one(text: str) -> PersonExtract:
return await async_client.chat.completions.create(
model='gpt-4o-mini',
response_model=PersonExtract,
messages=[{'role': 'user', 'content': text}]
)Running Multiple Extractions with gather
asyncio.gather runs a list of coroutines concurrently and returns all results when the last one completes. For a small batch of documents this is sufficient. Wrap your extraction coroutines in a list comprehension and pass them to gather. The total time equals roughly the slowest single call, not the sum of all calls.
async def batch_extract(texts: list) -> list:
tasks = [extract_one(text) for text in texts]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Filter out exceptions
successes = [r for r in results if not isinstance(r, Exception)]
failures = [r for r in results if isinstance(r, Exception)]
print(f'Success: {len(successes)}, Failures: {len(failures)}')
return successes
results = asyncio.run(batch_extract(documents))Controlling Concurrency with Semaphores
Sending thousands of requests simultaneously will hit rate limits and cause 429 errors. Use asyncio.Semaphore to cap the number of concurrent API calls. A semaphore with a value of 50 means at most 50 calls are in-flight at once. Tune this number based on your OpenAI tier's rate limit for the target model.
import asyncio
sem = asyncio.Semaphore(50) # max 50 concurrent calls
async def extract_with_limit(text: str, semaphore: asyncio.Semaphore):
async with semaphore:
return await extract_one(text)
async def batch_extract_limited(texts: list):
tasks = [extract_with_limit(t, sem) for t in texts]
return await asyncio.gather(*tasks, return_exceptions=True)Exponential Backoff on Rate Limit Errors
Even with a semaphore, you may hit rate limits during traffic bursts. Implement exponential backoff: wait 1 second, then 2, then 4, then 8 seconds before retrying. Add jitter (a small random offset) to prevent all concurrent callers from retrying at exactly the same time, which would cause another burst. The tenacity library makes this easy.
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
from openai import RateLimitError
@retry(
wait=wait_exponential(multiplier=1, min=1, max=60),
stop=stop_after_attempt(5),
retry=retry_if_exception_type(RateLimitError)
)
async def extract_with_retry(text: str):
return await extract_one(text)Using a Job Queue for Large Batches
For batches larger than a few thousand items, a persistent job queue is better than asyncio.gather. Queues like Redis Queue (RQ), Celery, or Dramatiq persist jobs across restarts, allow horizontal scaling with multiple workers, and give you visibility into job status and failures. Workers pull jobs from the queue and call the API independently.
# With Redis Queue (RQ)
from rq import Queue
from redis import Redis
redis_conn = Redis()
q = Queue('extractions', connection=redis_conn)
def enqueue_documents(doc_ids: list):
for doc_id in doc_ids:
q.enqueue(
'workers.extract_document',
doc_id,
job_timeout=120,
result_ttl=3600
)
enqueue_documents(all_doc_ids)Tracking Progress with a Database
Long-running batch jobs need progress tracking so you can monitor status, identify stuck jobs, and resume after failures. Use a status table in your database with fields for document ID, status (pending, processing, completed, failed), timestamps, and error messages. Update the status atomically around each extraction call.
import asyncpg
async def process_document(pool, doc_id: str, text: str):
async with pool.acquire() as conn:
await conn.execute(
'UPDATE extractions SET status=$1, started_at=NOW() WHERE doc_id=$2',
'processing', doc_id
)
try:
result = await extract_one(text)
await conn.execute(
'UPDATE extractions SET status=$1, result=$2, completed_at=NOW() WHERE doc_id=$3',
'completed', result.model_dump_json(), doc_id
)
except Exception as e:
await conn.execute(
'UPDATE extractions SET status=$1, error=$2 WHERE doc_id=$3',
'failed', str(e), doc_id
)Resuming Failed Jobs
A batch job should be safely restartable. At startup, query the database for documents with status pending or failed and retry them. Use an idempotency key per document so if the same document is accidentally enqueued twice, the second attempt detects the completed result and skips re-processing. This prevents duplicate writes to downstream systems.
async def get_pending_docs(pool) -> list:
async with pool.acquire() as conn:
rows = await conn.fetch(
'SELECT doc_id, raw_text FROM extractions WHERE status IN ($1, $2)',
'pending', 'failed'
)
return [dict(row) for row in rows]
async def resume_batch(pool):
docs = await get_pending_docs(pool)
print(f'Resuming {len(docs)} unprocessed documents')
tasks = [process_document(pool, d['doc_id'], d['raw_text']) for d in docs]
await asyncio.gather(*tasks, return_exceptions=True)Batching with the OpenAI Batch API
OpenAI's Batch API lets you submit up to 50,000 requests in a single file and receive results within 24 hours at 50% discount. This is ideal for non-urgent extraction pipelines where cost matters more than latency. You upload a JSONL file of requests, poll for completion, and download the results file.
from openai import OpenAI
import json
client = OpenAI()
# Build JSONL batch file
with open('/tmp/batch_requests.jsonl', 'w') as f:
for i, text in enumerate(documents):
request = {
'custom_id': f'doc_{i}',
'method': 'POST',
'url': '/v1/chat/completions',
'body': {
'model': 'gpt-4o-mini',
'messages': [{'role': 'user', 'content': text}]
}
}
f.write(json.dumps(request) + '\n')
# Upload and submit
batch_file = client.files.create(file=open('/tmp/batch_requests.jsonl', 'rb'), purpose='batch')
batch = client.batches.create(input_file_id=batch_file.id, endpoint='/v1/chat/completions', completion_window='24h')
print(batch.id)Monitoring Throughput and Cost
Track extraction throughput (documents per minute) and cost per document during batch runs. Divide total API spend by documents processed to get a cost baseline. As you scale, look for linear cost growth — superlinear growth suggests you are wasting tokens on unnecessarily long prompts. A simple metrics dashboard helps you catch inefficiencies before they compound.
import time
class BatchMetrics:
def __init__(self):
self.start_time = time.time()
self.processed = 0
self.total_tokens = 0
self.cost = 0.0
def record(self, usage):
self.processed += 1
self.total_tokens += usage.total_tokens
self.cost += usage.prompt_tokens * 0.00000015 + usage.completion_tokens * 0.0000006
def report(self):
elapsed = time.time() - self.start_time
print(f'{self.processed} docs in {elapsed:.1f}s = {self.processed/elapsed:.1f} docs/sec')
print(f'Cost: ${self.cost:.4f} = ${self.cost/self.processed:.6f} per doc')Respecting Token-Per-Minute Rate Limits
OpenAI rate limits apply both to requests per minute (RPM) and tokens per minute (TPM). Sending 50 concurrent calls is fine for RPM, but if each call uses 2,000 tokens, 50 calls equal 100,000 tokens per minute — easily exceeding Tier 1 limits. Count expected tokens before submitting using tiktoken and implement a token budget alongside your concurrency semaphore.
import tiktoken
enc = tiktoken.encoding_for_model('gpt-4o-mini')
def estimate_tokens(text: str) -> int:
return len(enc.encode(text)) + 300 # +300 for schema + response
# TPM_LIMIT = 200_000 # Tier 2 limit
# Only submit a batch if estimated total tokens fits within budget
def fits_in_budget(texts: list, tpm_limit: int = 200_000) -> bool:
total = sum(estimate_tokens(t) for t in texts)
return total <= tpm_limitQuick Check
Test your understanding of async batch processing for document extraction.
Lesson Recap
In this lesson you learned: asyncio and Semaphore enable concurrent API calls while respecting rate limits, job queues and status tables make large batch jobs resumable and observable, and the OpenAI Batch API offers 50% cost savings for non-urgent workloads at the expense of 24-hour latency. Next up we handle schema evolution in long-running extraction pipelines.
Frequently asked questions
Is the “Batch Processing with Async and Queues” lesson free?
Yes — the full text of “Batch Processing with Async and Queues” is free to read here on the web, and the AI Engineering Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Engineering Academy course, upgrade to CoddyKit PRO.
What will I learn in “Batch Processing with Async and Queues”?
Build an async extraction pipeline using asyncio and a job queue to process thousands of documents in parallel while respecting rate limits and tracking progress. You practise AI Engineering Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.
Do I need any experience to start AI Engineering Academy?
No prior experience is required. AI Engineering Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 3 of 4, so you can start here or from the beginning and move at your own pace.
How long does the “Batch Processing with Async and Queues” lesson take?
Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.
Can I write and run code in this AI Engineering Academy lesson?
Yes. Every AI Engineering Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.
All lessons in this course
- Instructor: Typed Extraction with Pydantic
- Handling Partial and Missing Data
- Batch Processing with Async and Queues
- Schema Evolution and Backward Compatibility