Elaborazione batch con async e code
Costruisca una pipeline asincrona di estrazione usando asyncio e una coda di job per elaborare migliaia di documenti in parallelo, rispettando i limiti di frequenza e monitorando l'avanzamento.
Elaborazione batch con async e code è una lezione AI Engineering Academy gratuita su CoddyKit. Questa è la lezione 3 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Engineering Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Engineering Academy include 4 lezioni in totale.
Parti di questa lezione non sono ancora state tradotte e vengono mostrate in inglese.
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.
Domande Frequenti
La lezione «Elaborazione batch con async e code» è gratuita?
Sì — il testo completo di «Elaborazione batch con async e code» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Engineering Academy, passa a CoddyKit PRO. Il corso AI Engineering Academy include 4 lezioni in totale.
Cosa imparerò in «Elaborazione batch con async e code»?
Costruisca una pipeline asincrona di estrazione usando asyncio e una coda di job per elaborare migliaia di documenti in parallelo, rispettando i limiti di frequenza e monitorando l'avanzamento. Eserciti AI Engineering Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Engineering Academy?
Non è richiesta alcuna esperienza precedente. AI Engineering Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 3 di 4.
Quanto tempo richiede la lezione «Elaborazione batch con async e code»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Engineering Academy?
Sì. Ogni lezione AI Engineering Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Instructor: estrazione tipizzata con Pydantic
- Gestire dati parziali e mancanti
- Elaborazione batch con async e code
- Evoluzione degli schemi e compatibilità con le versioni precedenti